From 364761edeea3dda3dcdfb0c3cf26b6cecd06f468 Mon Sep 17 00:00:00 2001 From: Ted Boudros Date: Mon, 16 Feb 2026 11:21:46 +0200 Subject: [PATCH] feat: Add signal tools and portfolio tracker --- README.md | 13 +- core/duration.py | 26 ++++ core/models/signals.py | 4 + docs/ARCHITECTURE.md | 25 +++- docs/DATA_MODELS.md | 16 ++- docs/FLOWS.md | 46 +++++- docs/LEARNING_LOOP.md | 12 +- engine/interface.py | 221 +++++++++++++++++++++++------ engine/pending_confirmation.py | 89 ++++++++++++ engine/signal_delivery.py | 100 +++++++++++++ engine/tools.py | 79 +++++++++-- main.py | 26 ++++ plugins/integrations/discord.py | 10 ++ plugins/integrations/telegram.py | 10 ++ plugins/task_handlers/ai_runner.py | 14 +- risk/engine.py | 6 + simulator/engine.py | 1 + 17 files changed, 616 insertions(+), 82 deletions(-) create mode 100644 core/duration.py create mode 100644 engine/pending_confirmation.py create mode 100644 engine/signal_delivery.py diff --git a/README.md b/README.md index 91e1b26..802cfcd 100644 --- a/README.md +++ b/README.md @@ -19,6 +19,8 @@ curl -fsSL https://raw.githubusercontent.com/tedboudros/ClawQuant/main/install.s - **Conversational AI via Telegram + Discord**: one AI interface handles chat, trade confirmations, portfolio queries, and task management. - **Multi-step tool loops**: the model can chain multiple tool calls in one turn before replying. +- **Live signal lifecycle for AI-proposed positions**: `open_potential_position` runs synchronous risk gating and then delivery (`signal.proposed -> signal.approved/rejected -> signal.delivered`). +- **Signal-ID confirmation tracking**: delivered signals require `confirm_signal(signal_id, entry_price, quantity)` or `skip_signal(signal_id, reason?)`, with one timeout reminder and pending state preserved. - **Task scheduling with AI self-invocation**: `ai.run_prompt` lets scheduled jobs run through the same central AI stack. - **Built-in schedulers/handlers**: `ai.run_prompt`, `news.briefing`, `notifications.send`, `comparison.weekly`. - **Plugin-defined AI tools**: plugins can register tools dynamically (`get_tools` / `call_tool`). @@ -31,7 +33,7 @@ curl -fsSL https://raw.githubusercontent.com/tedboudros/ClawQuant/main/install.s ## Coming Soon (Documented Target-State, Not Fully Wired Yet) -- Full autonomous **orchestrator -> risk gate -> signal delivery** production pipeline. +- Full autonomous **orchestrator-driven** production pipeline from ambient live inputs (without explicit AI tool invocation). - Additional integrations described in docs/examples (email/webhook/custom scrapers). - Additional market-data providers described in docs/examples (e.g., CoinGecko). - Auto-created default recurring learning tasks from config at startup. @@ -41,10 +43,11 @@ curl -fsSL https://raw.githubusercontent.com/tedboudros/ClawQuant/main/install.s 1. **Message arrives** via Telegram/Discord plugin. 2. **AI interface runs** with tool-calling (including plugin tools). -3. **Tools execute** actions (record trades, manage tasks, fetch prices/news/web results). -4. **Response is published** on `integration.output`. -5. **Output dispatcher delivers** via the right adapter/channel. -6. **Scheduler runs tasks** and can invoke the same AI (`ai.run_prompt`). +3. **Tools execute** actions (including proposing signals, recording confirmations, managing tasks, and fetching prices/news/web results). +4. **Signal proposals** (when used) pass through deterministic risk gating and adapter delivery. +5. **Responses/notifications are published** on `integration.output`. +6. **Output dispatcher delivers** via the right adapter/channel. +7. **Scheduler runs tasks** and can invoke the same AI (`ai.run_prompt`). ## Design diff --git a/core/duration.py b/core/duration.py new file mode 100644 index 0000000..67d6bef --- /dev/null +++ b/core/duration.py @@ -0,0 +1,26 @@ +"""Duration parsing helpers for configuration values.""" + +from __future__ import annotations + +import re +from datetime import timedelta + +_DURATION_RE = re.compile(r"^\s*(\d+)\s*([smhd])\s*$", re.IGNORECASE) +_UNIT_SECONDS = { + "s": 1, + "m": 60, + "h": 3600, + "d": 86400, +} + + +def parse_duration(value: str) -> timedelta: + """Parse compact duration strings like '60s', '4h', '7d'.""" + match = _DURATION_RE.match(str(value or "")) + if not match: + raise ValueError(f"Invalid duration: {value!r}. Expected ''.") + + amount = int(match.group(1)) + unit = match.group(2).lower() + seconds = amount * _UNIT_SECONDS[unit] + return timedelta(seconds=seconds) diff --git a/core/models/signals.py b/core/models/signals.py index a015ed8..27fc0bf 100644 --- a/core/models/signals.py +++ b/core/models/signals.py @@ -30,6 +30,10 @@ class Signal(BaseModel): risk_result: RiskResult | None = None delivered_at: datetime | None = None delivered_via: str | None = None + confirmation_status: Literal["pending", "confirmed", "skipped"] | None = None + confirmation_due_at: datetime | None = None + confirmation_reminder_sent_at: datetime | None = None + delivery_errors: list[str] | None = None class Position(BaseModel): diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index bb81a0d..f025da2 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -9,6 +9,9 @@ This file separates: Current runtime is **chat-first**: `integration message -> AI tool loop -> integration.output -> output dispatcher`. +Signal lifecycle is also live for AI-proposed positions: +`open_potential_position -> signal.proposed -> risk gate -> signal.approved/rejected -> signal.delivered`. + --- ## Core Design Principles @@ -48,6 +51,9 @@ graph TB subgraph Core AI[AI Interface] BUS[AsyncIOBus] + RISK[RiskEngine] + DEL[SignalDeliveryService] + REM[PendingConfirmationWatcher] SCH[Scheduler] OUT[OutputDispatcher] REG[PluginRegistry] @@ -65,6 +71,9 @@ graph TB AI --> BT AI --> PT AI --> BUS + BUS --> RISK + BUS --> DEL + REM --> BUS SCH --> TH TH --> BUS BUS --> OUT @@ -84,6 +93,9 @@ graph TB - `Scheduler` - `PortfolioTracker` - `AIInterface` +- `RiskEngine` subscriber (`signal.proposed`) +- `SignalDeliveryService` subscriber (`signal.approved`) +- `PendingConfirmationWatcher` loop (overdue pending reminders) - `OutputDispatcher` subscribed to `integration.output` - `aiohttp` server (`server.py`) @@ -100,7 +112,7 @@ graph TB - `web.search` - `browser.selenium` (optional, loaded only when `selenium_browser` is enabled) -Risk rules are loaded and registered, but risk-gating events are part of the target-state pipeline (see below). +Risk rules are loaded and actively evaluated in runtime through `RiskEngine`. --- @@ -110,8 +122,9 @@ Risk rules are loaded and registered, but risk-gating events are part of the tar ### Built-in tools -- `confirm_trade` -- `skip_trade` +- `open_potential_position` +- `confirm_signal` +- `skip_signal` - `close_position` - `user_initiated_trade` - `get_portfolio` @@ -177,6 +190,7 @@ Scheduled run context order: 3. task prompt (`params.prompt`) Result is published as `integration.output` and delivered through the same output dispatcher path as chat. +If the scheduled AI responds with exactly `[NO_REPLY]`, delivery is skipped for that run. --- @@ -191,6 +205,8 @@ All outbound user-visible messages are published as `integration.output` events. This keeps integrations modular and transport-focused. +For approved trade signals, `SignalDeliveryService` uses output adapters' `send(signal, memo=None)` path and then emits `signal.delivered` when at least one adapter succeeds. + --- ## HTTP Server (Current Routes) @@ -238,8 +254,7 @@ This keeps integrations modular and transport-focused. The following modules and flows exist conceptually (and partly in code) but are **not the default live runtime path** today: - `engine/orchestrator.py` as always-on subscriber for live input events -- `risk/engine.py` active gating of `signal.proposed` events -- Full signal lifecycle chain in production (`signal.proposed` -> `signal.approved/rejected` -> `signal.delivered`) +- Fully autonomous ambient orchestrator triggering (without explicit tool invocation) - Auto-creation of recurring tasks from config (`scheduler.default_tasks`, `learning.comparison_schedule`) - Simulator exposure through CLI/server with validated workflow coverage - Additional integrations/providers shown in legacy examples (email/webhook/custom scraper/CoinGecko) diff --git a/docs/DATA_MODELS.md b/docs/DATA_MODELS.md index 9f4a327..9e260d6 100644 --- a/docs/DATA_MODELS.md +++ b/docs/DATA_MODELS.md @@ -53,11 +53,11 @@ class Event(BaseModel): | `memory.created` | Yes | Yes (when comparison creates memory) | | `context.assembled` | Yes | Target-state path | | `memo.created` | Yes | Target-state path | -| `signal.proposed` | Yes | Target-state path | -| `signal.approved` | Yes | Target-state path | -| `signal.rejected` | Yes | Target-state path | -| `signal.delivered` | Yes | Target-state path | -| `alert.triggered` | Yes | Target-state path | +| `signal.proposed` | Yes | Yes (via `open_potential_position`) | +| `signal.approved` | Yes | Yes | +| `signal.rejected` | Yes | Yes | +| `signal.delivered` | Yes | Yes | +| `alert.triggered` | Yes | Yes (delivery failures) | | `simulation.started/completed` | Yes | Not emitted by current simulation module | --- @@ -116,9 +116,13 @@ class Signal(BaseModel): risk_result: RiskResult | None delivered_at: datetime | None delivered_via: str | None + confirmation_status: Literal["pending", "confirmed", "skipped"] | None + confirmation_due_at: datetime | None + confirmation_reminder_sent_at: datetime | None + delivery_errors: list[str] | None ``` -Status fields are valid and persisted; however, in default live runtime the full orchestrator/risk delivery lifecycle is not the primary wired path. +Signal lifecycle fields are actively used in runtime for tool-driven proposals and confirmation tracking. --- diff --git a/docs/FLOWS.md b/docs/FLOWS.md index b2095c0..ed34908 100644 --- a/docs/FLOWS.md +++ b/docs/FLOWS.md @@ -46,6 +46,49 @@ Key behaviors: --- +## Current Runtime Flow: AI-Proposed Signal (`open_potential_position`) + +1. AI calls `open_potential_position` with ticker/direction/catalyst/confidence/entry target/horizon. +2. Interface persists a proposed signal and publishes `signal.proposed`. +3. `RiskEngine` evaluates all registered risk rules synchronously. +4. If rejected: + - signal persisted as `rejected` + - tool result includes failed rule details so AI can retry with adjusted parameters. +5. If approved: + - AI portfolio position is opened + - signal persisted as `approved` + - `signal.approved` published. +6. `SignalDeliveryService` broadcasts the signal via all configured output adapters. +7. If any adapter succeeds: + - signal persisted as `delivered` + - confirmation status set to `pending` + - confirmation due timestamp set from `position_tracking.confirmation_timeout` + - `signal.delivered` published. +8. If all adapters fail: + - signal remains `approved` + - delivery errors are persisted + - `alert.triggered` published. + +--- + +## Current Runtime Flow: Signal Confirmation (`confirm_signal` / `skip_signal`) + +1. AI references signal IDs from `get_signals` context and calls `confirm_signal` or `skip_signal`. +2. Interface validates: + - signal exists + - signal is `delivered` + - signal confirmation is `pending` + - for confirm: `entry_price > 0` and `quantity > 0`. +3. On confirm: + - human portfolio position is recorded with `signal_id` + - signal confirmation status becomes `confirmed`. +4. On skip: + - skipped human position is recorded with `signal_id` + - signal confirmation status becomes `skipped`. +5. Pending confirmations past due receive one reminder message and remain pending. + +--- + ## Current Runtime Flow: Create Scheduled AI Task (`ai.run_prompt`) 1. User asks to schedule recurring monitoring. @@ -57,6 +100,7 @@ Key behaviors: 5. Handler `ai.run_prompt` runs same central AI stack. 6. Scheduled run injects: `system prompt -> last 10 channel messages -> task prompt`. 7. Result is published to `integration.output` and delivered back to same channel. + If the AI returns exactly `[NO_REPLY]`, delivery is skipped for that run. --- @@ -96,7 +140,7 @@ Key behaviors: The following are documented architecture goals but are **not fully wired as the default live runtime path**: -- Live input -> orchestrator agent chain -> `signal.proposed` -> risk gate -> delivery. +- Live input -> orchestrator agent chain -> autonomous signal proposal without explicit tool invocation. - Email/webhook/custom scraper as first-class live integrations. - Automatic recurring learning/comparison task creation from config on startup. - `run_analysis` event path consumed by an active orchestrator subscriber in production runtime. diff --git a/docs/LEARNING_LOOP.md b/docs/LEARNING_LOOP.md index 6098a8a..514778d 100644 --- a/docs/LEARNING_LOOP.md +++ b/docs/LEARNING_LOOP.md @@ -11,6 +11,10 @@ This doc describes what is implemented today and what is still target-state. - Dual portfolio files: - `positions/ai/*.json` - `positions/human/*.json` +- Signal confirmation lifecycle: + - `confirm_signal(signal_id, entry_price, quantity)` + - `skip_signal(signal_id, reason?)` + - one-time pending reminder after `position_tracking.confirmation_timeout` - Comparison task handler: `comparison.weekly` - Memory generation model + storage: - JSON files in `memories/` @@ -20,7 +24,7 @@ This doc describes what is implemented today and what is still target-state. ### Not fully wired / coming soon - Automatic scheduling from `learning.comparison_schedule` -- Automatic assumed-execution timeout flow from `position_tracking.confirmation_timeout` +- Automatic assumed-execution timeout flow (timeout currently reminders-only, remains pending) - Broker sync path for human portfolio - Full live orchestrator context-pack loop that routinely feeds memories into every production decision @@ -32,14 +36,14 @@ This doc describes what is implemented today and what is still target-state. - Stored in `~/.clawquant/positions/ai/` - Intended to represent system-side execution tracking -- In the default chat-first runtime, this portfolio is mainly updated where risk/orchestrator flows are explicitly invoked (target-state path), not by everyday chat tasking alone +- In the default chat-first runtime, this portfolio is updated when AI uses `open_potential_position` and signals pass risk approval ### Human Portfolio (actual) - Stored in `~/.clawquant/positions/human/` - Updated directly by AI tools such as: - - `confirm_trade` - - `skip_trade` + - `confirm_signal` + - `skip_signal` - `close_position` - `user_initiated_trade` diff --git a/engine/interface.py b/engine/interface.py index 4826e70..d2ba83d 100644 --- a/engine/interface.py +++ b/engine/interface.py @@ -20,7 +20,7 @@ from core.data.store import Store from core.models.events import Event, EventTypes from core.models.memories import Memory -from core.models.signals import Position, Signal +from core.models.signals import Signal from core.models.tasks import Task from core.protocols import LLMProvider from core.registry import PluginRegistry @@ -33,8 +33,9 @@ SYSTEM_PROMPT = """You are the AI assistant for ClawQuant, a trading advisory system. You help the user manage their trading activity. You can: -- Record trades they've made (confirm_trade, close_position, user_initiated_trade) -- Record trades they've skipped (skip_trade) +- Propose new trades for deterministic risk gating (open_potential_position) +- Record signal outcomes (confirm_signal, skip_signal) +- Record user-managed positions (close_position, user_initiated_trade) - Show portfolio state (get_portfolio) - Look up prices (get_price) - Manage scheduled tasks (list_tasks, create_task, delete_task, delete_task_by_name) @@ -45,11 +46,14 @@ IMPORTANT RULES: - When the user tells you about a trade they made, use the appropriate tool to record it. +- Use open_potential_position to propose a trade signal. If risk rejects it, adjust and retry. +- To confirm a delivered signal, use confirm_signal with signal_id, entry_price, and quantity. +- To skip a delivered signal, use skip_signal with signal_id. - When they ask about their portfolio or positions, use get_portfolio. -- When they want to skip a signal, use skip_trade and record their reason. - Before creating a scheduled task, use list_task_handlers and choose a valid handler name. - For recurring monitoring/updates (news checks, periodic watch, alerts), prefer handler `ai.run_prompt` unless the user explicitly asks for another handler. - For `ai.run_prompt` tasks, set `params.prompt` to the execution instruction (what to do each run), not to a meta-instruction about creating tasks. +- In scheduled `ai.run_prompt` executions, you may return exactly `[NO_REPLY]` when no user-facing update is needed. - If the user asks for news/research/web lookups, call available tools first; do not claim inability before attempting relevant tool calls. - Act-first rule: when the user requests an action that tools can perform, execute the tool calls in the same turn, then report the result. - Never send intent-only replies like "Let me check", "I'll do it", or "I can do that" when tools can run now. @@ -71,6 +75,7 @@ - Do not create/modify/delete tasks unless the prompt explicitly asks you to manage schedules. - For news/research requests, call relevant tools before saying data is unavailable. - Follow plugin-specific runtime instructions appended below this prompt when available (enabled plugins only). +- If this run should not notify the user, respond with exactly `[NO_REPLY]` and no extra text. - Respond with only the current run update for the user.""" FIRST_CHAT_ONBOARDING_DIRECTIVE = """[INTERNAL ONBOARDING DIRECTIVE] @@ -96,6 +101,7 @@ [/INTERNAL ONBOARDING DIRECTIVE]""" ONBOARDING_DIRECTIVE_MARKER = "[INTERNAL ONBOARDING DIRECTIVE]" +NO_REPLY_SENTINEL = "[NO_REPLY]" class AIInterface: @@ -183,6 +189,10 @@ def _coerce_instruction_text(value: Any) -> str: return "\n".join(parts).strip() return str(value).strip() + @staticmethod + def _is_no_reply_response(value: str) -> bool: + return str(value or "").strip().upper() == NO_REPLY_SENTINEL + def _get_plugin_prompt_instructions(self, plugin: Any, scheduled: bool) -> str: """Read optional prompt instructions from plugin hooks.""" context = "scheduled" if scheduled else "chat" @@ -544,6 +554,9 @@ async def handle_scheduled_prompt( logger.exception("Scheduled LLM call failed") return "Sorry, I couldn't process that scheduled run right now." + if self._is_no_reply_response(final_response): + return NO_REPLY_SENTINEL + if persist_output and final_response: self._append_message(channel_id, "assistant", final_response) @@ -553,10 +566,12 @@ async def _execute_tool(self, name: str, args: dict, source: str, channel_id: st """Execute a tool call and return a result string.""" try: match name: - case "confirm_trade": - return await self._tool_confirm_trade(args, source) - case "skip_trade": - return await self._tool_skip_trade(args, source) + case "open_potential_position": + return await self._tool_open_potential_position(args, source) + case "confirm_signal": + return await self._tool_confirm_signal(args, source) + case "skip_signal": + return await self._tool_skip_signal(args, source) case "close_position": return await self._tool_close_position(args, source) case "user_initiated_trade": @@ -700,51 +715,158 @@ async def _execute_plugin_tool( # Tool implementations # ------------------------------------------------------------------ - async def _tool_confirm_trade(self, args: dict, source: str) -> str: - ticker = args["ticker"].upper() - price = args["entry_price"] - size = args.get("size") + async def _tool_open_potential_position(self, args: dict, source: str) -> str: + ticker = str(args["ticker"]).strip().upper() + direction = str(args["direction"]).strip().lower() + if direction not in ("buy", "sell"): + return "direction must be 'buy' or 'sell'." + + confidence = float(args["confidence"]) + entry_target = float(args["entry_target"]) + if not 0 <= confidence <= 1: + return "confidence must be between 0.0 and 1.0." + if entry_target <= 0: + return "entry_target must be positive." + + signal = Signal( + ticker=ticker, + direction=direction, + catalyst=str(args["catalyst"]).strip(), + confidence=confidence, + entry_target=entry_target, + stop_loss=args.get("stop_loss"), + take_profit=args.get("take_profit"), + horizon=str(args["horizon"]).strip(), + status="proposed", + ) + event = Event( + type=EventTypes.SIGNAL_PROPOSED, + source="interface", + payload={}, + ) + signal.correlation_id = event.correlation_id + event.payload = signal.model_dump(mode="json") + self._store.write_json("signals", f"{signal.id}.json", signal) + await self._bus.publish(event) - # Find the most recent signal for this ticker - signals = self._store.list_json("signals", Signal) - matching = [s for s in signals if s.ticker == ticker and s.status in ("approved", "delivered")] + resolved = self._store.read_json("signals", f"{signal.id}.json", Signal) or signal + + if resolved.status == "rejected": + failed = resolved.risk_result.failed_rules if resolved.risk_result else [] + if failed: + details = "; ".join(f"{rule.rule_name}: {rule.reason}" for rule in failed) + else: + details = "Risk rules rejected the signal." + return ( + f"Signal rejected [{resolved.id}] {resolved.direction.upper()} {resolved.ticker}: " + f"{details}" + ) - if matching: - signal = matching[-1] # most recent - pos = self._portfolio.human_confirm_position( - signal=signal, entry_price=price, size=size, via=source, + if resolved.status == "delivered": + due_text = resolved.confirmation_due_at.isoformat() if resolved.confirmation_due_at else "unknown" + return ( + f"Signal delivered [{resolved.id}] {resolved.direction.upper()} {resolved.ticker} " + f"entry_target=${resolved.entry_target or 0:,.2f}, confidence={resolved.confidence:.0%}. " + f"Confirmation status: pending (due: {due_text})." ) - await self._bus.publish(Event( - type=EventTypes.POSITION_CONFIRMED, - source="interface", - payload={"ticker": ticker, "price": price, "portfolio": "human"}, - )) - return f"Confirmed: {ticker} position opened at ${price:,.2f}" + (f" ({size} units)" if size else "") - else: - # No matching signal -- treat as user-initiated - return await self._tool_user_initiated( - {"ticker": ticker, "direction": "long", "entry_price": price, "size": size}, - source, + + if resolved.status == "approved": + errors = "; ".join(resolved.delivery_errors or []) + detail = f" Delivery errors: {errors}" if errors else "" + return ( + f"Signal approved but not delivered [{resolved.id}] {resolved.direction.upper()} " + f"{resolved.ticker}.{detail}" ) - async def _tool_skip_trade(self, args: dict, source: str) -> str: - ticker = args["ticker"].upper() - reason = args.get("reason", "") + return ( + f"Signal created [{resolved.id}] with status={resolved.status}. " + "Risk evaluation did not complete as expected." + ) - signals = self._store.list_json("signals", Signal) - matching = [s for s in signals if s.ticker == ticker and s.status in ("approved", "delivered")] + async def _tool_confirm_signal(self, args: dict, source: str) -> str: + signal_id = str(args["signal_id"]).strip() + signal = self._store.read_json("signals", f"{signal_id}.json", Signal) + if signal is None: + return f"Signal not found: {signal_id}" + + if signal.status != "delivered": + return f"Signal {signal_id} is not delivered (current status: {signal.status})." + if signal.confirmation_status != "pending": + return ( + f"Signal {signal_id} is not pending confirmation " + f"(current confirmation_status: {signal.confirmation_status})." + ) - if matching: - signal = matching[-1] - self._portfolio.human_skip_position(signal=signal, via=source, notes=reason) - await self._bus.publish(Event( - type=EventTypes.POSITION_SKIPPED, - source="interface", - payload={"ticker": ticker, "reason": reason}, - )) - return f"Skipped: {ticker} signal." + (f" Reason: {reason}" if reason else "") - else: - return f"No pending signal found for {ticker}." + entry_price = float(args["entry_price"]) + quantity = float(args["quantity"]) + if entry_price <= 0: + return "entry_price must be positive." + if quantity <= 0: + return "quantity must be positive." + + self._portfolio.human_confirm_position( + signal=signal, + entry_price=entry_price, + size=quantity, + via=source, + ) + signal.confirmation_status = "confirmed" + self._store.write_json("signals", f"{signal.id}.json", signal) + + await self._bus.publish(Event( + type=EventTypes.POSITION_CONFIRMED, + source="interface", + payload={ + "signal_id": signal.id, + "ticker": signal.ticker, + "entry_price": entry_price, + "quantity": quantity, + "portfolio": "human", + }, + )) + + return ( + f"Confirmed signal {signal.id}: {signal.ticker} at ${entry_price:,.2f} " + f"for quantity {quantity}." + ) + + async def _tool_skip_signal(self, args: dict, source: str) -> str: + signal_id = str(args["signal_id"]).strip() + reason = str(args.get("reason", "")).strip() + + signal = self._store.read_json("signals", f"{signal_id}.json", Signal) + if signal is None: + return f"Signal not found: {signal_id}" + + if signal.status != "delivered": + return f"Signal {signal_id} is not delivered (current status: {signal.status})." + if signal.confirmation_status != "pending": + return ( + f"Signal {signal_id} is not pending confirmation " + f"(current confirmation_status: {signal.confirmation_status})." + ) + + self._portfolio.human_skip_position( + signal=signal, + via=source, + notes=reason or None, + ) + signal.confirmation_status = "skipped" + self._store.write_json("signals", f"{signal.id}.json", signal) + + await self._bus.publish(Event( + type=EventTypes.POSITION_SKIPPED, + source="interface", + payload={ + "signal_id": signal.id, + "ticker": signal.ticker, + "reason": reason, + "portfolio": "human", + }, + )) + if reason: + return f"Skipped signal {signal.id} ({signal.ticker}). Reason: {reason}" + return f"Skipped signal {signal.id} ({signal.ticker})." async def _tool_close_position(self, args: dict, source: str) -> str: ticker = args["ticker"].upper() @@ -967,7 +1089,12 @@ def _tool_get_signals(self, args: dict) -> str: parts = [] for s in signals: - parts.append(f" [{s.status}] {s.direction.upper()} {s.ticker} conf={s.confidence:.0%} ({s.created_at.strftime('%Y-%m-%d')})") + confirmation = s.confirmation_status or "n/a" + parts.append( + f" [{s.status}] id={s.id} {s.direction.upper()} {s.ticker} " + f"conf={s.confidence:.0%} confirmation={confirmation} " + f"({s.created_at.strftime('%Y-%m-%d')})" + ) return f"{len(parts)} signals:\n" + "\n".join(parts) async def _tool_run_analysis(self, args: dict) -> str: diff --git a/engine/pending_confirmation.py b/engine/pending_confirmation.py new file mode 100644 index 0000000..46acab7 --- /dev/null +++ b/engine/pending_confirmation.py @@ -0,0 +1,89 @@ +"""Pending confirmation reminder watcher.""" + +from __future__ import annotations + +import asyncio +import logging +from datetime import datetime, timezone + +from core.bus import AsyncIOBus +from core.data.store import Store +from core.models.events import Event, EventTypes +from core.models.signals import Signal + +logger = logging.getLogger(__name__) + + +class PendingConfirmationWatcher: + """Send one reminder when pending signal confirmations become overdue.""" + + def __init__( + self, + bus: AsyncIOBus, + store: Store, + check_interval_seconds: int = 60, + ) -> None: + self._bus = bus + self._store = store + self._check_interval_seconds = max(1, int(check_interval_seconds)) + self._running = False + self._task: asyncio.Task | None = None + + async def start(self) -> None: + if self._running: + return + self._running = True + self._task = asyncio.create_task(self._loop()) + logger.info( + "Pending confirmation watcher started (check every %ds)", + self._check_interval_seconds, + ) + + async def stop(self) -> None: + self._running = False + if self._task: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + logger.info("Pending confirmation watcher stopped") + + async def _loop(self) -> None: + while self._running: + try: + await self._scan_once() + except Exception: + logger.exception("Pending confirmation scan failed") + await asyncio.sleep(self._check_interval_seconds) + + async def _scan_once(self) -> None: + now = datetime.now(timezone.utc) + signals = self._store.list_json("signals", Signal) + + for signal in signals: + if signal.status != "delivered": + continue + if signal.confirmation_status != "pending": + continue + if signal.confirmation_reminder_sent_at is not None: + continue + if signal.confirmation_due_at is None or signal.confirmation_due_at > now: + continue + + reminder = ( + f"Signal confirmation pending: {signal.direction.upper()} {signal.ticker} " + f"(signal_id={signal.id}). To confirm, provide entry price and quantity. " + f"Example: confirm signal {signal.id} entry_price 123.45 quantity 10. " + f"To skip: skip signal {signal.id} reason ." + ) + + await self._bus.publish(Event( + type=EventTypes.INTEGRATION_OUTPUT, + source="pending_confirmation", + payload={"text": reminder}, + )) + + signal.confirmation_reminder_sent_at = now + self._store.write_json("signals", f"{signal.id}.json", signal) diff --git a/engine/signal_delivery.py b/engine/signal_delivery.py new file mode 100644 index 0000000..d035f75 --- /dev/null +++ b/engine/signal_delivery.py @@ -0,0 +1,100 @@ +"""Signal delivery service. + +Subscribes to approved signals, delivers via output adapters, and updates +signal lifecycle state to delivered when any adapter succeeds. +""" + +from __future__ import annotations + +import logging +from datetime import datetime, timedelta, timezone + +from core.bus import AsyncIOBus +from core.data.store import Store +from core.models.events import Event, EventTypes +from core.models.signals import Signal +from core.registry import PluginRegistry + +logger = logging.getLogger(__name__) + + +class SignalDeliveryService: + """Deliver approved signals through all configured output adapters.""" + + def __init__( + self, + bus: AsyncIOBus, + store: Store, + registry: PluginRegistry, + confirmation_timeout: timedelta, + ) -> None: + self._bus = bus + self._store = store + self._registry = registry + self._confirmation_timeout = confirmation_timeout + self._bus.subscribe(EventTypes.SIGNAL_APPROVED, self._handle_signal_approved) + + async def _handle_signal_approved(self, event: Event) -> None: + try: + signal = Signal(**event.payload) + except Exception: + logger.exception("Failed to parse approved signal event payload") + return + + adapters = self._registry.get_all("output") + successes: list[str] = [] + errors: list[str] = [] + + for adapter in adapters: + try: + result = await adapter.send(signal, memo=None) + except Exception as exc: + logger.exception( + "Signal delivery failed via %s for %s", + getattr(adapter, "name", "unknown"), + signal.id, + ) + errors.append(f"{getattr(adapter, 'name', 'unknown')}: {exc}") + continue + + if result.success: + successes.append(result.adapter or getattr(adapter, "name", "unknown")) + else: + message = result.message or "delivery failed" + errors.append(f"{result.adapter or getattr(adapter, 'name', 'unknown')}: {message}") + + if successes: + delivered_at = datetime.now(timezone.utc) + signal.status = "delivered" + signal.delivered_at = delivered_at + signal.delivered_via = ", ".join(sorted(set(successes))) + signal.confirmation_status = "pending" + signal.confirmation_due_at = delivered_at + self._confirmation_timeout + signal.confirmation_reminder_sent_at = None + signal.delivery_errors = errors or None + self._store.write_json("signals", f"{signal.id}.json", signal) + + delivered_event = event.derive( + type=EventTypes.SIGNAL_DELIVERED, + source="signal_delivery", + payload=signal.model_dump(mode="json"), + ) + await self._bus.publish(delivered_event) + return + + signal.status = "approved" + signal.delivery_errors = errors or ["No output adapters configured"] + self._store.write_json("signals", f"{signal.id}.json", signal) + + alert_event = event.derive( + type=EventTypes.ALERT_TRIGGERED, + source="signal_delivery", + payload={ + "level": "error", + "signal_id": signal.id, + "ticker": signal.ticker, + "message": "Signal approved but delivery failed on all output adapters", + "errors": signal.delivery_errors, + }, + ) + await self._bus.publish(alert_event) diff --git a/engine/tools.py b/engine/tools.py index bb72484..15160cb 100644 --- a/engine/tools.py +++ b/engine/tools.py @@ -8,8 +8,8 @@ { "type": "function", "function": { - "name": "confirm_trade", - "description": "User confirms they executed a trade that was signaled. Records the position in the human portfolio.", + "name": "open_potential_position", + "description": "Propose a new trade signal for synchronous risk evaluation and delivery lifecycle handling.", "parameters": { "type": "object", "properties": { @@ -17,37 +17,90 @@ "type": "string", "description": "The ticker symbol (e.g., NVDA, BTC-USD, AAPL)", }, + "direction": { + "type": "string", + "enum": ["buy", "sell"], + "description": "Proposed signal direction", + }, + "catalyst": { + "type": "string", + "description": "Why this position is being proposed", + }, + "confidence": { + "type": "number", + "description": "Signal confidence from 0.0 to 1.0", + }, + "entry_target": { + "type": "number", + "description": "Target entry price for the signal", + }, + "stop_loss": { + "type": "number", + "description": "Optional stop-loss price", + }, + "take_profit": { + "type": "number", + "description": "Optional take-profit price", + }, + "horizon": { + "type": "string", + "description": "Expected holding period (e.g., 1-3 months)", + }, + }, + "required": [ + "ticker", + "direction", + "catalyst", + "confidence", + "entry_target", + "horizon", + ], + }, + }, + }, + { + "type": "function", + "function": { + "name": "confirm_signal", + "description": "Confirm a delivered signal using explicit signal_id, entry price, and quantity.", + "parameters": { + "type": "object", + "properties": { + "signal_id": { + "type": "string", + "description": "Signal identifier (e.g., sig_ab12cd34ef56)", + }, "entry_price": { "type": "number", - "description": "The price at which the user entered the trade", + "description": "Actual executed entry price", }, - "size": { + "quantity": { "type": "number", - "description": "Number of units/shares/coins bought (optional)", + "description": "Executed position quantity (required)", }, }, - "required": ["ticker", "entry_price"], + "required": ["signal_id", "entry_price", "quantity"], }, }, }, { "type": "function", "function": { - "name": "skip_trade", - "description": "User decides to skip/reject a signaled trade. Records the skip with the user's reason.", + "name": "skip_signal", + "description": "Skip a delivered signal using its signal_id.", "parameters": { "type": "object", "properties": { - "ticker": { + "signal_id": { "type": "string", - "description": "The ticker symbol of the signal being skipped", + "description": "Signal identifier (e.g., sig_ab12cd34ef56)", }, "reason": { "type": "string", - "description": "Why the user is skipping this trade", + "description": "Optional reason for skipping", }, }, - "required": ["ticker"], + "required": ["signal_id"], }, }, }, @@ -193,7 +246,7 @@ }, "params": { "type": "object", - "description": "Parameters to pass to the handler. For ai.run_prompt include params.prompt with the per-run execution instruction.", + "description": "Parameters to pass to the handler. For ai.run_prompt include params.prompt with the per-run execution instruction. A run can return exactly [NO_REPLY] to skip user notification.", }, }, "required": ["name", "handler"], diff --git a/main.py b/main.py index 8cf1803..d72a5d0 100644 --- a/main.py +++ b/main.py @@ -18,10 +18,14 @@ from core.bus import AsyncIOBus from core.config import load_config from core.data.store import Store +from core.duration import parse_duration from core.models.events import Event, EventTypes from core.output_dispatcher import OutputDispatcher from core.registry import PluginRegistry from engine.interface import AIInterface +from engine.pending_confirmation import PendingConfirmationWatcher +from engine.signal_delivery import SignalDeliveryService +from risk.engine import RiskEngine from risk.portfolio import PortfolioTracker from scheduler.runner import Scheduler from server import create_app @@ -438,6 +442,26 @@ async def run(config_path: str | None = None, env_path: str | None = None) -> No await _load_plugins(config, bus, store, registry, ai_interface) logger.info("Plugin registry: %s", registry.summary()) + # Wire risk, signal delivery, and pending-confirmation lifecycle services. + _risk_engine = RiskEngine( + bus=bus, + store=store, + registry=registry, + portfolio=portfolio, + ) + confirmation_timeout = parse_duration(config.position_tracking.confirmation_timeout) + _signal_delivery = SignalDeliveryService( + bus=bus, + store=store, + registry=registry, + confirmation_timeout=confirmation_timeout, + ) + pending_confirmation = PendingConfirmationWatcher( + bus=bus, + store=store, + check_interval_seconds=60, + ) + # Generic integration.output delivery pipeline (adapter-agnostic) output_dispatcher = OutputDispatcher(registry=registry) bus.subscribe(EventTypes.INTEGRATION_OUTPUT, output_dispatcher.handle_integration_output) @@ -453,6 +477,7 @@ async def run(config_path: str | None = None, env_path: str | None = None) -> No # Start scheduler await scheduler.start() + await pending_confirmation.start() # Start server runner = web.AppRunner(app) @@ -474,6 +499,7 @@ async def run(config_path: str | None = None, env_path: str | None = None) -> No pass finally: logger.info("Shutting down...") + await pending_confirmation.stop() await scheduler.stop() # Stop all input integrations (dedupe because adapters can implement input+output) diff --git a/plugins/integrations/discord.py b/plugins/integrations/discord.py index fb30ef4..9dffd6c 100644 --- a/plugins/integrations/discord.py +++ b/plugins/integrations/discord.py @@ -310,6 +310,7 @@ def _format_signal_message(self, signal: Signal, memo: InvestmentMemo | None = N lines = [ f"{icon} **{signal.direction.upper()} {signal.ticker}**", + f"Signal ID: `{signal.id}`", f"Confidence: {signal.confidence:.0%}", ] @@ -329,4 +330,13 @@ def _format_signal_message(self, signal: Signal, memo: InvestmentMemo | None = N summary = memo.executive_summary[:500] lines.append(f"\n{summary}") + lines.append( + "\nTo confirm: `confirm signal " + f"{signal.id} entry_price quantity `" + ) + lines.append( + "To skip: `skip signal " + f"{signal.id} reason `" + ) + return "\n".join(lines) diff --git a/plugins/integrations/telegram.py b/plugins/integrations/telegram.py index bd78252..807c004 100644 --- a/plugins/integrations/telegram.py +++ b/plugins/integrations/telegram.py @@ -295,6 +295,7 @@ def _format_signal_message(self, signal: Signal, memo: InvestmentMemo | None = N lines = [ f"{icon} *{signal.direction.upper()} {signal.ticker}*", + f"Signal ID: `{signal.id}`", f"Confidence: {signal.confidence:.0%}", ] @@ -314,4 +315,13 @@ def _format_signal_message(self, signal: Signal, memo: InvestmentMemo | None = N summary = memo.executive_summary[:500] lines.append(f"\n{summary}") + lines.append( + "\nTo confirm: `confirm signal " + f"{signal.id} entry_price quantity `" + ) + lines.append( + "To skip: `skip signal " + f"{signal.id} reason `" + ) + return "\n".join(lines) diff --git a/plugins/task_handlers/ai_runner.py b/plugins/task_handlers/ai_runner.py index 0a7c924..f6fa864 100644 --- a/plugins/task_handlers/ai_runner.py +++ b/plugins/task_handlers/ai_runner.py @@ -10,6 +10,7 @@ from core.models.tasks import TaskResult logger = logging.getLogger(__name__) +NO_REPLY_SENTINEL = "[NO_REPLY]" PLUGIN_META = { "name": "ai_runner", @@ -19,7 +20,7 @@ "protocols": ["task_handler"], "class_name": "AIRunnerHandler", "pip_dependencies": [], - "setup_instructions": "Use handler 'ai.run_prompt' with params.prompt and optional params.channel_id.", + "setup_instructions": "Use handler 'ai.run_prompt' with params.prompt and optional params.channel_id. Prompt may return [NO_REPLY] to suppress delivery.", "config_fields": [], } @@ -35,6 +36,10 @@ def __init__(self, ai_interface: Any, bus: AsyncIOBus) -> None: def name(self) -> str: return "ai.run_prompt" + @staticmethod + def _is_no_reply_response(value: str) -> bool: + return str(value or "").strip().upper() == NO_REPLY_SENTINEL + async def run(self, params: dict) -> TaskResult: prompt = str(params.get("prompt", "")).strip() if not prompt: @@ -54,6 +59,13 @@ async def run(self, params: dict) -> TaskResult: persist_output=True, ) + if self._is_no_reply_response(response): + logger.info("ai.run_prompt returned %s; skipping integration.output publish", NO_REPLY_SENTINEL) + return TaskResult( + status="no_action", + message="AI run completed with NO_REPLY; no user notification sent.", + ) + await self._bus.publish(Event( type=EventTypes.INTEGRATION_OUTPUT, source=self.name, diff --git a/risk/engine.py b/risk/engine.py index 85907dd..e8212d1 100644 --- a/risk/engine.py +++ b/risk/engine.py @@ -10,6 +10,7 @@ import logging from core.bus import AsyncIOBus +from core.data.store import Store from core.models.events import Event, EventTypes from core.models.signals import RiskResult, RuleEvaluation, Signal from core.registry import PluginRegistry @@ -28,10 +29,12 @@ class RiskEngine: def __init__( self, bus: AsyncIOBus, + store: Store, registry: PluginRegistry, portfolio: PortfolioTracker, ) -> None: self._bus = bus + self._store = store self._registry = registry self._portfolio = portfolio @@ -73,8 +76,10 @@ async def _handle_signal(self, event: Event) -> None: # APPROVED -- open position in AI portfolio, publish approved event signal.status = "approved" signal.risk_result = result + signal.delivery_errors = None self._portfolio.ai_open_position(signal) + self._store.write_json("signals", f"{signal.id}.json", signal) approved_event = event.derive( type=EventTypes.SIGNAL_APPROVED, @@ -94,6 +99,7 @@ async def _handle_signal(self, event: Event) -> None: # REJECTED signal.status = "rejected" signal.risk_result = result + self._store.write_json("signals", f"{signal.id}.json", signal) rejected_event = event.derive( type=EventTypes.SIGNAL_REJECTED, diff --git a/simulator/engine.py b/simulator/engine.py index 6e7eb39..be7aa99 100644 --- a/simulator/engine.py +++ b/simulator/engine.py @@ -84,6 +84,7 @@ async def run_simulation(self, config: SimulationConfig) -> SimulationRun: # Set up risk engine in sandbox risk_engine = RiskEngine( bus=sim_bus, + store=sim_store, registry=self._registry, portfolio=sim_portfolio, )