From bbed0581ad33b828caaf37769f655d9d51dfb65e Mon Sep 17 00:00:00 2001 From: Codex Date: Fri, 8 May 2026 14:39:44 +0800 Subject: [PATCH] Add decision packets and alert records --- .env.example | 1 + SESSION.md | 27 +- TASKS.md | 4 +- migrations/030_create_decision_packets.sql | 35 ++ scripts/run_decision_packet.py | 36 ++ src/api/__init__.py | 4 + src/api/server.py | 11 + src/api/trading_runtime.py | 9 + src/config/settings.py | 1 + .../decision_packet_repository.py | 165 +++++++++ src/services/decision/__init__.py | 3 + .../decision/decision_packet_service.py | 346 ++++++++++++++++++ .../runtime/trading_runtime_service.py | 27 ++ src/web/app.js | 65 +++- src/web/index.html | 5 + tests/unit/test_api_server.py | 17 + tests/unit/test_decision_packet_service.py | 105 ++++++ tests/unit/test_runtime_script_json.py | 22 ++ tests/unit/test_trading_runtime_service.py | 14 + 19 files changed, 888 insertions(+), 9 deletions(-) create mode 100644 migrations/030_create_decision_packets.sql create mode 100644 scripts/run_decision_packet.py create mode 100644 src/repositories/decision_packet_repository.py create mode 100644 src/services/decision/__init__.py create mode 100644 src/services/decision/decision_packet_service.py create mode 100644 tests/unit/test_decision_packet_service.py diff --git a/.env.example b/.env.example index 077a180..abc8e91 100644 --- a/.env.example +++ b/.env.example @@ -43,6 +43,7 @@ TRADING_RUNTIME_TRIGGER_TIMEFRAME=1m TRADING_RUNTIME_MARKET_STATE_INTERVAL_SECONDS=15 TRADING_RUNTIME_BROKER_SYNC_INTERVAL_SECONDS=30 TRADING_RUNTIME_WATCH_INTERVAL_SECONDS=20 +TRADING_RUNTIME_DECISION_PACKET_INTERVAL_SECONDS=20 TRADING_RUNTIME_PLAN_INTERVAL_SECONDS=900 DEFAULT_TIMEZONE=UTC SESSION_TIMEZONE=America/New_York diff --git a/SESSION.md b/SESSION.md index 3c9e44c..be6403a 100644 --- a/SESSION.md +++ b/SESSION.md @@ -2,7 +2,7 @@ ## Current Goal -Continue the trading-module mainline in autonomous mode until `Real-time Assisted Trading V1` is complete. The next active stage is `P0-4` Decision Packet And Proactive Alerting. +Continue the trading-module mainline in autonomous mode until `Real-time Assisted Trading V1` is complete. The next active stage is `P0-5` Real Trade Mapping And Post-Trade Loop. ## Current Branch @@ -26,10 +26,26 @@ Continue the trading-module mainline in autonomous mode until `Real-time Assiste - `/trading/runtime/run-once` - `scripts/run_trading_runtime.py` - workbench Runtime Supervisor panel +- Completed `P0-4` Decision Packet And Proactive Alerting implementation: + - `DecisionPacketService` + - `decision_packets` and `decision_alerts` persistence + - `/trading/decision-packet` + - `/trading/alerts` + - runtime decision-packet generation task + - workbench Decision Packet panel with recent alerts + - `scripts/run_decision_packet.py` - Local quality gates passed: - `npm run lint` - `npm run build` - `npm run visual:snapshots` +- Remote deploy and smoke passed: + - `/health` + - `/trading/broker-state` + - `/trading/runtime-status` + - `/trading/runtime/run-once` + - `/trading/market-state` + - `/trading/decision-packet` + - `/trading/alerts` ## Current Working State @@ -46,14 +62,13 @@ Continue the trading-module mainline in autonomous mode until `Real-time Assiste Proceed directly to: -### `P0-4` Decision Packet And Proactive Alerting +### `P0-5` Real Trade Mapping And Post-Trade Loop Goals: -- build `DecisionPacketService` -- combine plan, candidate, target, invalidation, and Risk Governor into one operator decision packet -- add status-change alert plumbing -- show the decision packet in the workbench +- map real orders and fills back to framework executions +- let real position changes trigger post-trade journal updates +- strengthen weekly review with real-trade attribution and next-step actions ## Validation Commands diff --git a/TASKS.md b/TASKS.md index fea1fda..5207c2e 100644 --- a/TASKS.md +++ b/TASKS.md @@ -8,11 +8,10 @@ ## In Progress -- [ ] `P0-4` Decision Packet And Proactive Alerting +- [ ] `P0-5` Real Trade Mapping And Post-Trade Loop ## Next Up -- [ ] `P0-5` Real Trade Mapping And Post-Trade Loop - [ ] `P1-1` Trust Scoring And Model Whitelists - [ ] `P1-2` Operator Experience And Workflow Tools - [ ] `P1-3` Live Adapter Readiness Prep @@ -30,5 +29,6 @@ - [x] `P0-1` Real-time market truth layer - [x] `P0-2` Real Account And Order Read Sync - [x] `P0-3` Runtime Scheduling And Supervision +- [x] `P0-4` Decision Packet And Proactive Alerting - [x] Autonomous execution protocol codified in repo docs - [x] Market-truth proxy switched to direct v2rayN LAN proxy `192.168.3.73:10808` diff --git a/migrations/030_create_decision_packets.sql b/migrations/030_create_decision_packets.sql new file mode 100644 index 0000000..51e7398 --- /dev/null +++ b/migrations/030_create_decision_packets.sql @@ -0,0 +1,35 @@ +create table if not exists decision_packets ( + id bigserial primary key, + instrument_id bigint not null references instruments(id), + trading_plan_id bigint references trading_plans(id), + candidate_id bigint references trade_plan_candidates(id), + signal_id bigint references trade_signals(id), + execution_ticket_id bigint references execution_tickets(id), + packet_ts timestamptz not null, + status text not null, + action_recommendation text not null, + quality_score numeric, + ai_explanation text, + payload jsonb not null default '{}'::jsonb, + created_at timestamptz not null default now() +); + +create index if not exists idx_decision_packets_instrument_ts_desc +on decision_packets (instrument_id, packet_ts desc, id desc); + +create table if not exists decision_alerts ( + id bigserial primary key, + instrument_id bigint not null references instruments(id), + decision_packet_id bigint references decision_packets(id), + alert_key text not null unique, + alert_type text not null, + status text not null, + channel text not null default 'workbench', + message text not null, + payload jsonb not null default '{}'::jsonb, + created_ts timestamptz not null, + delivered_ts timestamptz +); + +create index if not exists idx_decision_alerts_instrument_ts_desc +on decision_alerts (instrument_id, created_ts desc, id desc); diff --git a/scripts/run_decision_packet.py b/scripts/run_decision_packet.py new file mode 100644 index 0000000..0f69d82 --- /dev/null +++ b/scripts/run_decision_packet.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +from pathlib import Path +import json +import os +import sys + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from src.services.decision import DecisionPacketService + + +def main(env=None) -> int: + env = os.environ if env is None else env + instrument_id = int(env.get("DECISION_PACKET_INSTRUMENT_ID", "1")) + output_json = _get_bool(env, "DECISION_PACKET_OUTPUT_JSON", False) + payload = DecisionPacketService().build_persist_and_alert(instrument_id=instrument_id) + if output_json: + print(json.dumps(payload, ensure_ascii=False, sort_keys=True)) + else: + print(f"Decision packet: {payload.get('status')} / {payload.get('action_recommendation')}") + print(f"Why blocked: {', '.join(payload.get('why_blocked') or []) or 'none'}") + return 0 + + +def _get_bool(env, key: str, default: bool = False) -> bool: + value = env.get(key) + if value is None or value == "": + return default + return str(value).strip().lower() in {"1", "true", "yes", "on"} + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/api/__init__.py b/src/api/__init__.py index bea33ba..2ac252f 100644 --- a/src/api/__init__.py +++ b/src/api/__init__.py @@ -20,6 +20,8 @@ from src.api.trading_runtime import ( execute_ticket, generate_trading_journal, generate_trading_plan, + get_trading_decision_alerts, + get_trading_decision_packet, get_latest_trading_plan, get_trading_candidate_by_id, get_trading_market_state, @@ -58,6 +60,8 @@ __all__ = [ "get_weekly_review", "generate_trading_journal", "generate_trading_plan", + "get_trading_decision_packet", + "get_trading_decision_alerts", "activate_trading_plan", "cancel_trading_plan", "run_trading_watch", diff --git a/src/api/server.py b/src/api/server.py index 10b8b65..48719bf 100644 --- a/src/api/server.py +++ b/src/api/server.py @@ -21,6 +21,8 @@ from src.api.trading_runtime import ( execute_ticket, generate_trading_journal, generate_trading_plan, + get_trading_decision_alerts, + get_trading_decision_packet, get_latest_trading_plan, get_trading_candidate_by_id, get_trading_market_state, @@ -69,6 +71,8 @@ def build_api_dependencies() -> dict: "trading_plan_activate": activate_trading_plan, "trading_plan_cancel": cancel_trading_plan, "trading_market_state": get_trading_market_state, + "trading_decision_packet": get_trading_decision_packet, + "trading_decision_alerts": get_trading_decision_alerts, "trading_runtime_status": get_trading_runtime_status, "trading_runtime_run_once": run_trading_runtime_once, "trading_watch_run": run_trading_watch, @@ -175,6 +179,13 @@ def route_request( return 200, deps["trading_runtime_status"]() if method == "POST" and path == "/trading/runtime/run-once": return 200, deps["trading_runtime_run_once"]() + if method == "GET" and path == "/trading/decision-packet": + instrument_id = _require_int(query, "instrument_id") + return 200, deps["trading_decision_packet"](instrument_id=instrument_id) + if method == "GET" and path == "/trading/alerts": + instrument_id = _require_int(query, "instrument_id") + limit = _get_int(query, "limit", 20) + return 200, {"items": deps["trading_decision_alerts"](instrument_id=instrument_id, limit=limit)} if method == "POST" and path == "/trading/watch/run": instrument_id = _require_json_int(body or {}, "instrument_id") candidate = deps["trading_watch_run"]( diff --git a/src/api/trading_runtime.py b/src/api/trading_runtime.py index fe77e59..8ad428b 100644 --- a/src/api/trading_runtime.py +++ b/src/api/trading_runtime.py @@ -8,6 +8,7 @@ from src.repositories.signal_repository import SignalRepository from src.repositories.trade_plan_candidate_repository import TradePlanCandidateRepository from src.repositories.trading_plan_repository import TradingPlanRepository from src.services.broker import BrokerReadModelService, BROKER_STATE_OK +from src.services.decision import DecisionPacketService from src.services.execution import ExecutionService from src.services.journal import TradingJournalService, WeeklyReviewService from src.services.market_state import MarketStateService, get_market_truth_supervisor @@ -406,6 +407,14 @@ def run_trading_runtime_once() -> dict: return get_trading_runtime_service().run_once() +def get_trading_decision_packet(instrument_id: int) -> dict: + return DecisionPacketService().build_preview(instrument_id=instrument_id) + + +def get_trading_decision_alerts(instrument_id: int, limit: int = 20) -> list[dict]: + return DecisionPacketService().fetch_recent_alerts(instrument_id=instrument_id, limit=limit) + + def market_truth_degraded(market_state) -> bool: payload = market_state.to_dict() if hasattr(market_state, "to_dict") else dict(market_state or {}) freshness_code = str(((payload.get("freshness") or {}).get("code")) or "") diff --git a/src/config/settings.py b/src/config/settings.py index 889afdf..8bcaeda 100644 --- a/src/config/settings.py +++ b/src/config/settings.py @@ -47,6 +47,7 @@ class Settings(BaseSettings): trading_runtime_market_state_interval_seconds: int = 15 trading_runtime_broker_sync_interval_seconds: int = 30 trading_runtime_watch_interval_seconds: int = 20 + trading_runtime_decision_packet_interval_seconds: int = 20 trading_runtime_plan_interval_seconds: int = 900 default_timezone: str = "UTC" session_timezone: str = "America/New_York" diff --git a/src/repositories/decision_packet_repository.py b/src/repositories/decision_packet_repository.py new file mode 100644 index 0000000..13434cd --- /dev/null +++ b/src/repositories/decision_packet_repository.py @@ -0,0 +1,165 @@ +import json +from typing import Optional + +from sqlalchemy import text + +from src.db.session import SessionLocal + + +class DecisionPacketRepository: + def insert_decision_packet( + self, + *, + instrument_id: int, + trading_plan_id: Optional[int], + candidate_id: Optional[int], + signal_id: Optional[int], + execution_ticket_id: Optional[int], + packet_ts, + status: str, + action_recommendation: str, + quality_score: Optional[float], + ai_explanation: str, + payload: dict, + ) -> int: + sql = text( + """ + insert into decision_packets ( + instrument_id, + trading_plan_id, + candidate_id, + signal_id, + execution_ticket_id, + packet_ts, + status, + action_recommendation, + quality_score, + ai_explanation, + payload + ) values ( + :instrument_id, + :trading_plan_id, + :candidate_id, + :signal_id, + :execution_ticket_id, + :packet_ts, + :status, + :action_recommendation, + :quality_score, + :ai_explanation, + cast(:payload as jsonb) + ) + returning id + """ + ) + with SessionLocal.begin() as session: + result = session.execute( + sql, + { + "instrument_id": instrument_id, + "trading_plan_id": trading_plan_id, + "candidate_id": candidate_id, + "signal_id": signal_id, + "execution_ticket_id": execution_ticket_id, + "packet_ts": packet_ts, + "status": status, + "action_recommendation": action_recommendation, + "quality_score": quality_score, + "ai_explanation": ai_explanation, + "payload": json.dumps(payload, default=str), + }, + ) + return int(result.scalar_one()) + + def insert_decision_alert( + self, + *, + instrument_id: int, + decision_packet_id: Optional[int], + alert_key: str, + alert_type: str, + status: str, + channel: str, + message: str, + payload: dict, + created_ts, + delivered_ts=None, + ) -> Optional[int]: + sql = text( + """ + insert into decision_alerts ( + instrument_id, + decision_packet_id, + alert_key, + alert_type, + status, + channel, + message, + payload, + created_ts, + delivered_ts + ) values ( + :instrument_id, + :decision_packet_id, + :alert_key, + :alert_type, + :status, + :channel, + :message, + cast(:payload as jsonb), + :created_ts, + :delivered_ts + ) + on conflict (alert_key) do nothing + returning id + """ + ) + with SessionLocal.begin() as session: + result = session.execute( + sql, + { + "instrument_id": instrument_id, + "decision_packet_id": decision_packet_id, + "alert_key": alert_key, + "alert_type": alert_type, + "status": status, + "channel": channel, + "message": message, + "payload": json.dumps(payload, default=str), + "created_ts": created_ts, + "delivered_ts": delivered_ts, + }, + ) + value = result.scalar_one_or_none() + return int(value) if value is not None else None + + def fetch_latest_decision_packet(self, instrument_id: int) -> Optional[dict]: + sql = text( + """ + select id, instrument_id, trading_plan_id, candidate_id, signal_id, + execution_ticket_id, packet_ts, status, action_recommendation, + quality_score, ai_explanation, payload + from decision_packets + where instrument_id = :instrument_id + order by packet_ts desc, id desc + limit 1 + """ + ) + with SessionLocal() as session: + row = session.execute(sql, {"instrument_id": instrument_id}).mappings().first() + return dict(row) if row else None + + def fetch_recent_decision_alerts(self, instrument_id: int, limit: int = 20) -> list[dict]: + sql = text( + """ + select id, instrument_id, decision_packet_id, alert_key, alert_type, + status, channel, message, payload, created_ts, delivered_ts + from decision_alerts + where instrument_id = :instrument_id + order by created_ts desc, id desc + limit :limit + """ + ) + with SessionLocal() as session: + rows = session.execute(sql, {"instrument_id": instrument_id, "limit": limit}).mappings().all() + return [dict(row) for row in rows] diff --git a/src/services/decision/__init__.py b/src/services/decision/__init__.py new file mode 100644 index 0000000..c91418d --- /dev/null +++ b/src/services/decision/__init__.py @@ -0,0 +1,3 @@ +from src.services.decision.decision_packet_service import DecisionPacketService + +__all__ = ["DecisionPacketService"] diff --git a/src/services/decision/decision_packet_service.py b/src/services/decision/decision_packet_service.py new file mode 100644 index 0000000..0347fa6 --- /dev/null +++ b/src/services/decision/decision_packet_service.py @@ -0,0 +1,346 @@ +from __future__ import annotations + +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from typing import Optional + +from src.repositories.decision_packet_repository import DecisionPacketRepository +from src.repositories.execution_repository import ExecutionRepository +from src.repositories.signal_repository import SignalRepository +from src.repositories.trade_plan_candidate_repository import TradePlanCandidateRepository +from src.repositories.trading_plan_repository import TradingPlanRepository +from src.services.execution import ExecutionService + + +DEFAULT_MODEL_CODE = "OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE" + + +@dataclass(frozen=True) +class DecisionPacket: + instrument_id: int + packet_ts: object + status: str + action_recommendation: str + symbol: str + trading_plan_id: Optional[int] + candidate_id: Optional[int] + signal_id: Optional[int] + execution_ticket_id: Optional[int] + narrative: str + primary_draw: dict + bias: str + setup_type: str + confirmation_chain: dict + entry_zone: dict + stop_loss: Optional[float] + targets: dict + invalidation: dict + expiry_ts: Optional[object] + risk_result: dict + why_allowed: list[str] + why_blocked: list[str] + quality_score: Optional[float] + ai_explanation: str + alerts: list[dict] + raw: dict + + def to_dict(self) -> dict: + return _serialize(asdict(self)) + + +class DecisionPacketService: + def __init__( + self, + *, + decision_repository: Optional[DecisionPacketRepository] = None, + plan_repository: Optional[TradingPlanRepository] = None, + candidate_repository: Optional[TradePlanCandidateRepository] = None, + signal_repository: Optional[SignalRepository] = None, + execution_repository: Optional[ExecutionRepository] = None, + execution_service: Optional[ExecutionService] = None, + ) -> None: + self.decision_repository = decision_repository or DecisionPacketRepository() + self.plan_repository = plan_repository or TradingPlanRepository() + self.candidate_repository = candidate_repository or TradePlanCandidateRepository() + self.signal_repository = signal_repository or SignalRepository() + self.execution_repository = execution_repository or ExecutionRepository() + self.execution_service = execution_service or ExecutionService() + + def build_packet(self, *, instrument_id: int, model_code: str = DEFAULT_MODEL_CODE, now: Optional[datetime] = None) -> DecisionPacket: + observed_ts = _ensure_aware(now or datetime.now(timezone.utc)) + plan = self.plan_repository.fetch_latest_trading_plan(instrument_id) or {} + candidates = self.candidate_repository.fetch_recent_candidates(instrument_id=instrument_id, limit=10) + candidate = _select_candidate(candidates, plan_id=plan.get("id"), model_code=model_code) + signal = self.signal_repository.fetch_latest_trade_signal_for_candidate(int(candidate["id"])) if candidate and candidate.get("id") else None + ticket = _latest_ticket(self.execution_repository, signal_id=int(signal["id"])) if signal and signal.get("id") else None + risk_result = _build_risk_result(self.execution_service, signal) + packet = build_decision_packet_from_rows( + instrument_id=instrument_id, + observed_ts=observed_ts, + plan=plan, + candidate=candidate or {}, + signal=signal or {}, + ticket=ticket or {}, + risk_result=risk_result, + ) + return packet + + def persist_packet(self, packet: DecisionPacket) -> int: + return self.decision_repository.insert_decision_packet( + instrument_id=packet.instrument_id, + trading_plan_id=packet.trading_plan_id, + candidate_id=packet.candidate_id, + signal_id=packet.signal_id, + execution_ticket_id=packet.execution_ticket_id, + packet_ts=packet.packet_ts, + status=packet.status, + action_recommendation=packet.action_recommendation, + quality_score=packet.quality_score, + ai_explanation=packet.ai_explanation, + payload=packet.to_dict(), + ) + + def build_preview(self, *, instrument_id: int, model_code: str = DEFAULT_MODEL_CODE) -> dict: + packet = self.build_packet(instrument_id=instrument_id, model_code=model_code) + return { + **packet.to_dict(), + "alerts": build_alert_candidates(packet), + } + + def build_persist_and_alert(self, *, instrument_id: int, model_code: str = DEFAULT_MODEL_CODE) -> dict: + packet = self.build_packet(instrument_id=instrument_id, model_code=model_code) + packet_id = self.persist_packet(packet) + alerts = [] + for alert in build_alert_candidates(packet): + alert_id = self.decision_repository.insert_decision_alert( + instrument_id=instrument_id, + decision_packet_id=packet_id, + alert_key=alert["alert_key"], + alert_type=alert["alert_type"], + status="queued", + channel="workbench", + message=alert["message"], + payload=alert["payload"], + created_ts=packet.packet_ts, + ) + alerts.append({**alert, "id": alert_id, "status": "queued"}) + return { + "id": packet_id, + **packet.to_dict(), + "alerts": alerts, + } + + def fetch_recent_alerts(self, *, instrument_id: int, limit: int = 20) -> list[dict]: + return self.decision_repository.fetch_recent_decision_alerts(instrument_id=instrument_id, limit=limit) + + +def build_decision_packet_from_rows( + *, + instrument_id: int, + observed_ts: datetime, + plan: dict, + candidate: dict, + signal: dict, + ticket: dict, + risk_result: dict, +) -> DecisionPacket: + confirmation_state = dict(candidate.get("confirmation_state") or signal.get("confirmation_quality") or {}) + target_plan = dict(signal.get("target_plan") or (candidate.get("entry") or {}).get("target_plan") or {}) + risk_reasons = list(risk_result.get("reasons") or []) + missing = list(candidate.get("missing_confirmations") or confirmation_state.get("missing_conditions") or []) + rejected = list(confirmation_state.get("rejected_conditions") or []) + status, action = resolve_action(plan=plan, candidate=candidate, ticket=ticket, risk_result=risk_result, missing=missing) + why_blocked = _dedupe([*missing, *rejected, *risk_reasons]) + why_allowed = [] + if action in {"build_execution_ticket", "confirm_ticket", "manage_ticket"}: + why_allowed = _dedupe(["active_plan_bound", "candidate_confirmed", "risk_governor_passed"]) + ai_explanation = build_ai_explanation(action=action, plan=plan, candidate=candidate, risk_reasons=risk_reasons, missing=missing) + return DecisionPacket( + instrument_id=instrument_id, + packet_ts=observed_ts, + status=status, + action_recommendation=action, + symbol=str((candidate.get("evidence") or {}).get("symbol") or ""), + trading_plan_id=_int_or_none(candidate.get("trading_plan_id") or plan.get("id")), + candidate_id=_int_or_none(candidate.get("id")), + signal_id=_int_or_none(signal.get("id")), + execution_ticket_id=_int_or_none(ticket.get("id")), + narrative=str(plan.get("narrative") or candidate.get("narrative") or ""), + primary_draw=dict(plan.get("primary_draw") or signal.get("primary_draw") or {}), + bias=str(plan.get("bias") or "unknown"), + setup_type=str(candidate.get("model_code") or signal.get("model_code") or DEFAULT_MODEL_CODE), + confirmation_chain={ + "present": list(confirmation_state.get("present_conditions") or confirmation_state.get("present") or []), + "missing": missing, + "rejected": rejected, + "reason_codes": list(confirmation_state.get("reason_codes") or []), + }, + entry_zone={ + "low": _float_or_none(signal.get("entry_low") or candidate.get("entry_low")), + "high": _float_or_none(signal.get("entry_high") or candidate.get("entry_high")), + "type": signal.get("entry_type") or (candidate.get("entry") or {}).get("entry_type") or "limit", + }, + stop_loss=_float_or_none(signal.get("stop_loss") or candidate.get("stop_loss")), + targets={ + "tp1": _target_item("tp1", signal=signal, candidate=candidate, target_plan=target_plan), + "tp2": _target_item("tp2", signal=signal, candidate=candidate, target_plan=target_plan), + "management_rules": list(signal.get("management_rules") or candidate.get("management_rules") or target_plan.get("management_rules") or []), + }, + invalidation={ + "plan": list(plan.get("invalidations") or []), + "candidate": list(candidate.get("invalidations") or []), + "signal": list(signal.get("invalidations") or []), + }, + expiry_ts=signal.get("expires_ts") or candidate.get("expiry_ts") or candidate.get("expires_ts"), + risk_result=risk_result, + why_allowed=why_allowed, + why_blocked=why_blocked, + quality_score=_float_or_none(candidate.get("quality_score") or signal.get("quality_score")), + ai_explanation=ai_explanation, + alerts=[], + raw={ + "plan": plan, + "candidate": candidate, + "signal": signal, + "ticket": ticket, + }, + ) + + +def resolve_action(*, plan: dict, candidate: dict, ticket: dict, risk_result: dict, missing: list[str]) -> tuple[str, str]: + if not plan or plan.get("status") != "active": + return "blocked", "wait_for_active_plan" + if not candidate: + return "monitoring", "wait_for_candidate" + if missing or candidate.get("status") in {"blocked", "draft"}: + return "blocked", "wait_for_confirmations" + if risk_result.get("allowed") is False: + return "blocked", "do_not_execute" + if ticket.get("status") == "awaiting_confirmation": + return "actionable", "confirm_ticket" + if ticket.get("status") in {"confirmed", "pending"}: + return "actionable", "manage_ticket" + return "actionable", "build_execution_ticket" + + +def build_alert_candidates(packet: DecisionPacket) -> list[dict]: + alerts = [] + base = f"{packet.instrument_id}:{packet.candidate_id or 'none'}:{packet.action_recommendation}:{packet.status}" + if packet.action_recommendation in {"confirm_ticket", "build_execution_ticket"}: + alerts.append(_alert(base, "candidate_actionable", "Candidate is actionable inside the current plan.", packet)) + if packet.action_recommendation in {"do_not_execute", "wait_for_confirmations", "wait_for_active_plan"}: + alerts.append(_alert(base, "candidate_blocked", f"Candidate blocked: {', '.join(packet.why_blocked) or packet.action_recommendation}", packet)) + if packet.action_recommendation == "manage_ticket": + alerts.append(_alert(base, "ticket_ready", "Execution ticket is confirmed or pending management.", packet)) + return alerts + + +def build_ai_explanation(*, action: str, plan: dict, candidate: dict, risk_reasons: list[str], missing: list[str]) -> str: + if not plan or plan.get("status") != "active": + return "No active trading plan is available, so the system should only observe." + if not candidate: + return "Active plan exists, but no current candidate is available." + if missing: + return f"Candidate is not ready because confirmations are missing: {', '.join(missing)}." + if risk_reasons: + return f"Risk Governor blocks execution: {', '.join(risk_reasons)}." + return f"Candidate is inside the active plan and current recommendation is {action}." + + +def _build_risk_result(execution_service: ExecutionService, signal: Optional[dict]) -> dict: + if not signal or not signal.get("id"): + return {"allowed": None, "status": "signal_missing", "reasons": ["signal_missing"]} + decision = execution_service.prepare_execution(signal_id=int(signal["id"]), execution_mode="semi_auto") + if decision is None: + return {"allowed": False, "status": "signal_not_found", "reasons": ["signal_not_found"]} + return { + "allowed": decision.allowed, + "status": decision.status, + "reasons": list(decision.reasons), + "requires_confirmation": decision.requires_confirmation, + "risk_pct": decision.risk_pct, + "size": decision.size, + "entry_price": decision.entry_price, + "meta": dict(decision.meta), + } + + +def _latest_ticket(repository: ExecutionRepository, *, signal_id: int) -> Optional[dict]: + rows = repository.fetch_recent_execution_ticket_rows(signal_id=signal_id, limit=1) + return rows[0] if rows else None + + +def _select_candidate(candidates: list[dict], *, plan_id, model_code: str) -> Optional[dict]: + if plan_id: + scoped = [candidate for candidate in candidates if candidate.get("trading_plan_id") == plan_id] + for candidate in scoped: + if candidate.get("model_code") == model_code: + return candidate + return scoped[0] if scoped else None + for candidate in candidates: + if candidate.get("model_code") == model_code: + return candidate + return candidates[0] if candidates else None + + +def _target_item(name: str, *, signal: dict, candidate: dict, target_plan: dict) -> dict: + target = dict(target_plan.get(name) or {}) + return { + "price": _float_or_none(signal.get(name) or candidate.get(name) or target.get("price")), + "source": target.get("source") or target.get("source_type") or "unknown", + "raw": target, + } + + +def _alert(key: str, alert_type: str, message: str, packet: DecisionPacket) -> dict: + return { + "alert_key": f"{key}:{alert_type}", + "alert_type": alert_type, + "message": message, + "payload": { + "action_recommendation": packet.action_recommendation, + "status": packet.status, + "why_blocked": list(packet.why_blocked), + "why_allowed": list(packet.why_allowed), + }, + } + + +def _float_or_none(value) -> Optional[float]: + if value in (None, ""): + return None + return float(value) + + +def _int_or_none(value) -> Optional[int]: + if value in (None, ""): + return None + return int(value) + + +def _ensure_aware(value: datetime) -> datetime: + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +def _serialize(value): + if isinstance(value, datetime): + return value.isoformat() + if isinstance(value, dict): + return {key: _serialize(item) for key, item in value.items()} + if isinstance(value, list): + return [_serialize(item) for item in value] + return value + + +def _dedupe(values: list[str]) -> list[str]: + result = [] + seen = set() + for value in values: + text = str(value or "").strip() + if not text or text in seen: + continue + seen.add(text) + result.append(text) + return result diff --git a/src/services/runtime/trading_runtime_service.py b/src/services/runtime/trading_runtime_service.py index c1daa6d..4497387 100644 --- a/src/services/runtime/trading_runtime_service.py +++ b/src/services/runtime/trading_runtime_service.py @@ -8,6 +8,7 @@ from typing import Optional from src.config.settings import settings from src.services.broker import BROKER_STATE_OK, BrokerReadModelService +from src.services.decision import DecisionPacketService from src.services.market_state import MarketStateService, get_market_truth_supervisor from src.services.trade_plan_candidates import SetupWatcherService from src.services.trading_plans import TradingPlanService @@ -53,11 +54,13 @@ class TradingRuntimeScheduler: trading_plan_service: Optional[TradingPlanService] = None, setup_watcher_service: Optional[SetupWatcherService] = None, broker_read_model_service: Optional[BrokerReadModelService] = None, + decision_packet_service: Optional[DecisionPacketService] = None, ) -> None: self.market_state_service = market_state_service or MarketStateService(market_truth_supervisor=get_market_truth_supervisor()) self.trading_plan_service = trading_plan_service or TradingPlanService() self.setup_watcher_service = setup_watcher_service or SetupWatcherService() self.broker_read_model_service = broker_read_model_service or BrokerReadModelService() + self.decision_packet_service = decision_packet_service or DecisionPacketService() self.symbols = [item.strip() for item in settings.trading_runtime_symbols.split(",") if item.strip()] self.bias_tf = settings.trading_runtime_bias_timeframe self.setup_tf = settings.trading_runtime_setup_timeframe @@ -66,6 +69,7 @@ class TradingRuntimeScheduler: "market_state": RuntimeTaskState("market_state", settings.trading_runtime_market_state_interval_seconds), "broker_sync": RuntimeTaskState("broker_sync", settings.trading_runtime_broker_sync_interval_seconds), "watch": RuntimeTaskState("watch", settings.trading_runtime_watch_interval_seconds), + "decision_packet": RuntimeTaskState("decision_packet", settings.trading_runtime_decision_packet_interval_seconds), "plan": RuntimeTaskState("plan", settings.trading_runtime_plan_interval_seconds), } self.current_blockers: list[str] = [] @@ -94,6 +98,7 @@ class TradingRuntimeScheduler: market_payload = self._run_market_state_task(symbol=symbol, instrument_id=instrument_id, now=now, blockers=blockers) self._run_broker_sync_task(symbol=symbol, instrument_id=instrument_id, now=now, blockers=blockers) self._run_watch_task(symbol=symbol, instrument_id=instrument_id, now=now, market_payload=market_payload, blockers=blockers) + self._run_decision_packet_task(symbol=symbol, instrument_id=instrument_id, now=now, blockers=blockers) self.current_blockers = _dedupe(blockers) self.degraded_mode_state = { "is_degraded": bool(self.current_blockers), @@ -236,6 +241,28 @@ class TradingRuntimeScheduler: blockers.append(code) self._record_incident(task_name=task.name, symbol=symbol, code=code, error=str(exc), observed_ts=now) + def _run_decision_packet_task(self, *, symbol: str, instrument_id: int, now: datetime, blockers: list[str]) -> None: + task = self.tasks["decision_packet"] + if not _task_due(task, now): + return + _task_start(task, now) + try: + packet = self.decision_packet_service.build_persist_and_alert(instrument_id=instrument_id) + task.details = { + "symbol": symbol, + "instrument_id": instrument_id, + "packet_id": packet.get("id"), + "status": packet.get("status"), + "action_recommendation": packet.get("action_recommendation"), + "alerts": len(packet.get("alerts") or []), + } + _task_success(task, now) + except Exception as exc: + _task_failure(task, now, str(exc)) + code = f"decision_packet_task_failed:{symbol}" + blockers.append(code) + self._record_incident(task_name=task.name, symbol=symbol, code=code, error=str(exc), observed_ts=now) + def _record_incident(self, *, task_name: str, symbol: str, code: str, observed_ts: datetime, error: Optional[str] = None) -> None: self.incidents.append( { diff --git a/src/web/app.js b/src/web/app.js index 6c11636..70598ae 100644 --- a/src/web/app.js +++ b/src/web/app.js @@ -11,6 +11,7 @@ const activeFilters = document.getElementById('activeFilters'); const overviewCards = document.getElementById('overviewCards'); const marketStatePanel = document.getElementById('marketStatePanel'); const runtimeStatusPanel = document.getElementById('runtimeStatusPanel'); +const decisionPacketPanel = document.getElementById('decisionPacketPanel'); const preMarketPlanPanel = document.getElementById('preMarketPlanPanel'); const tradeCandidatesPanel = document.getElementById('tradeCandidatesPanel'); const executionTicketsPanel = document.getElementById('executionTicketsPanel'); @@ -297,6 +298,60 @@ function renderRuntimeStatus(runtimeStatus) { runtimeStatusPanel.appendChild(grid); } +function renderDecisionPacket(packet, alertRows) { + clearNode(decisionPacketPanel); + if (!packet || packet.status === 'unavailable') { + decisionPacketPanel.appendChild(createElement('div', 'empty-state', 'Decision packet unavailable.')); + return; + } + decisionPacketPanel.appendChild( + buildInfoGrid([ + ['status', packet.status || 'unknown'], + ['recommendation', packet.action_recommendation || 'monitor'], + ['bias', packet.bias || 'unknown'], + ['setup', packet.setup_type || 'n/a'], + ['candidate', packet.candidate_id || 'n/a'], + ['ticket', packet.execution_ticket_id || 'n/a'], + ['risk result', formatJsonInline(packet.risk_result?.reasons || [])], + ['why allowed', formatList(packet.why_allowed || [])], + ['why blocked', formatList(packet.why_blocked || [])], + ['ai explanation', packet.ai_explanation || 'n/a'], + ]) + ); + const grid = createElement('div', 'cards'); + grid.appendChild( + buildItemCard( + 'Plan Context', + packet.action_recommendation || 'monitor', + [ + ['narrative', packet.narrative || 'n/a'], + ['primary draw', formatJsonInline(packet.primary_draw || {})], + ['confirmation', formatJsonInline(packet.confirmation_chain || {})], + ['entry zone', `${formatNumber(packet.entry_zone?.low, 3)} - ${formatNumber(packet.entry_zone?.high, 3)}`], + ['stop', formatNumber(packet.stop_loss, 3)], + ['targets', formatJsonInline(packet.targets || {})], + ['invalidation', formatJsonInline(packet.invalidation || {})], + ['expiry', formatTimestamp(packet.expiry_ts)], + ] + ) + ); + const recentAlerts = alertRows?.items || packet.alerts || []; + recentAlerts.slice(0, 3).forEach((alert) => { + grid.appendChild( + buildItemCard( + alert.alert_type || 'alert', + alert.status || 'queued', + [ + ['channel', alert.channel || 'workbench'], + ['message', alert.message || 'n/a'], + ['payload', formatJsonInline(alert.payload || {})], + ] + ) + ); + }); + decisionPacketPanel.appendChild(grid); +} + function renderPreMarketPlan(plan) { clearNode(preMarketPlanPanel); preMarketPlanPanel.appendChild( @@ -646,6 +701,7 @@ function showErrorPanel(message) { [ marketStatePanel, runtimeStatusPanel, + decisionPacketPanel, preMarketPlanPanel, tradeCandidatesPanel, executionTicketsPanel, @@ -667,6 +723,8 @@ function renderDashboardData({ tradingPlan, marketState, runtimeStatus, + decisionPacket, + decisionAlerts, tradeCandidates, executionTickets, tradingExecutions, @@ -678,6 +736,7 @@ function renderDashboardData({ renderOverview(weeklyReview, tradeCandidates.items || [], executionTickets.items || [], tradingExecutions.items || []); renderMarketState(marketState, tradingExecutions.items || []); renderRuntimeStatus(runtimeStatus); + renderDecisionPacket(decisionPacket, decisionAlerts); renderPreMarketPlan(tradingPlan); renderCandidates(tradeCandidates.items || []); renderTickets(executionTickets.items || []); @@ -694,12 +753,14 @@ async function refreshDashboard() { const limit = limitInput.value || '20'; try { - const [health, candles, tradingPlan, marketState, runtimeStatus, tradeCandidates, executionTickets, tradingExecutions, journalSummary, journalItems, weeklyReview] = await Promise.all([ + const [health, candles, tradingPlan, marketState, runtimeStatus, decisionPacket, decisionAlerts, tradeCandidates, executionTickets, tradingExecutions, journalSummary, journalItems, weeklyReview] = await Promise.all([ fetchJson('/health'), fetchJson(`/candles?${queryString({ instrument_id: instrumentId, timeframe, limit })}`), fetchJson(`/trading/plans/latest?${queryString({ instrument_id: instrumentId })}`).catch(() => null), fetchJson(`/trading/market-state?${queryString({ instrument_id: instrumentId, timeframe })}`).catch(() => ({ market_state: {} })), fetchJson('/trading/runtime-status').catch(() => ({ scheduler_state: 'unavailable', tasks: {}, current_blockers: ['runtime_status_unavailable'] })), + fetchJson(`/trading/decision-packet?${queryString({ instrument_id: instrumentId })}`).catch(() => ({ status: 'unavailable', action_recommendation: 'monitor' })), + fetchJson(`/trading/alerts?${queryString({ instrument_id: instrumentId, limit: 10 })}`).catch(() => ({ items: [] })), fetchJson(`/trading/candidates?${queryString({ instrument_id: instrumentId, limit })}`).catch(() => ({ items: [] })), fetchJson(`/trading/tickets?${queryString({ instrument_id: instrumentId, limit })}`).catch(() => ({ items: [] })), fetchJson(`/trading/executions?${queryString({ instrument_id: instrumentId, limit })}`).catch(() => ({ items: [] })), @@ -714,6 +775,8 @@ async function refreshDashboard() { tradingPlan, marketState, runtimeStatus, + decisionPacket, + decisionAlerts, tradeCandidates, executionTickets, tradingExecutions, diff --git a/src/web/index.html b/src/web/index.html index 25a36a1..2025fdd 100644 --- a/src/web/index.html +++ b/src/web/index.html @@ -64,6 +64,11 @@
+
+

Decision Packet

+
+
+

Pre-Market Plan

diff --git a/tests/unit/test_api_server.py b/tests/unit/test_api_server.py index 13c193b..88f8196 100644 --- a/tests/unit/test_api_server.py +++ b/tests/unit/test_api_server.py @@ -51,6 +51,23 @@ class ApiServerRoutingTests(unittest.TestCase): self.assertEqual(payload["snapshot_id"], 1) self.assertEqual(payload["market_state"]["timeframe"], "5m") + def test_routes_trading_decision_packet_and_alerts(self) -> None: + status_packet, payload_packet = route_request( + "/trading/decision-packet", + {"instrument_id": ["1"]}, + dependencies={"trading_decision_packet": lambda **kwargs: {"status": "actionable", "action_recommendation": "confirm_ticket"}}, + ) + status_alerts, payload_alerts = route_request( + "/trading/alerts", + {"instrument_id": ["1"], "limit": ["2"]}, + dependencies={"trading_decision_alerts": lambda **kwargs: [{"id": 1, "alert_type": "candidate_actionable"}]}, + ) + + self.assertEqual(status_packet, 200) + self.assertEqual(payload_packet["action_recommendation"], "confirm_ticket") + self.assertEqual(status_alerts, 200) + self.assertEqual(payload_alerts["items"], [{"id": 1, "alert_type": "candidate_actionable"}]) + def test_routes_trading_broker_state(self) -> None: status, payload = route_request( "/trading/broker-state", diff --git a/tests/unit/test_decision_packet_service.py b/tests/unit/test_decision_packet_service.py new file mode 100644 index 0000000..d001be8 --- /dev/null +++ b/tests/unit/test_decision_packet_service.py @@ -0,0 +1,105 @@ +from datetime import datetime, timezone +import os +import unittest + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test") + +from src.services.decision.decision_packet_service import ( + _select_candidate, + build_alert_candidates, + build_decision_packet_from_rows, +) + + +class DecisionPacketServiceTests(unittest.TestCase): + def test_build_packet_recommends_confirm_ticket_when_risk_passes(self) -> None: + packet = build_decision_packet_from_rows( + instrument_id=1, + observed_ts=datetime(2026, 5, 8, 1, 0, tzinfo=timezone.utc), + plan={ + "id": 10, + "status": "active", + "narrative": "active:bullish", + "primary_draw": {"source": "liquidity_pool", "price_high": 101}, + "bias": "bullish", + "invalidations": [{"code": "range_low_lost"}], + }, + candidate={ + "id": 20, + "trading_plan_id": 10, + "model_code": "OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE", + "status": "executable", + "entry_low": 100, + "entry_high": 101, + "stop_loss": 99, + "tp1": 102, + "tp2": 104, + "quality_score": 0.8, + "confirmation_state": {"present_conditions": ["sweep", "mss", "fvg"], "reason_codes": ["confirmed"]}, + "missing_confirmations": [], + "management_rules": [{"rule": "tp1_partial"}], + "entry": {"target_plan": {"tp1": {"source": "irl"}, "tp2": {"source": "erl"}}}, + }, + signal={"id": 30, "entry_type": "limit", "target_plan": {"tp1": {"source": "irl"}, "tp2": {"source": "erl"}}}, + ticket={"id": 40, "status": "awaiting_confirmation"}, + risk_result={"allowed": True, "status": "allowed", "reasons": []}, + ) + + self.assertEqual(packet.status, "actionable") + self.assertEqual(packet.action_recommendation, "confirm_ticket") + self.assertEqual(packet.targets["tp1"]["source"], "irl") + self.assertIn("risk_governor_passed", packet.why_allowed) + + def test_build_packet_explains_missing_confirmations(self) -> None: + packet = build_decision_packet_from_rows( + instrument_id=1, + observed_ts=datetime(2026, 5, 8, 1, 0, tzinfo=timezone.utc), + plan={"id": 10, "status": "active", "bias": "bearish"}, + candidate={ + "id": 20, + "trading_plan_id": 10, + "status": "blocked", + "missing_confirmations": ["mss_confirmed"], + "confirmation_state": {"rejected_conditions": ["fvg_missing"]}, + }, + signal={}, + ticket={}, + risk_result={"allowed": None, "status": "signal_missing", "reasons": ["signal_missing"]}, + ) + + self.assertEqual(packet.status, "blocked") + self.assertEqual(packet.action_recommendation, "wait_for_confirmations") + self.assertIn("mss_confirmed", packet.why_blocked) + self.assertIn("fvg_missing", packet.why_blocked) + self.assertIn("signal_missing", packet.why_blocked) + + def test_alert_candidates_are_stable_for_actionable_packet(self) -> None: + packet = build_decision_packet_from_rows( + instrument_id=1, + observed_ts=datetime(2026, 5, 8, 1, 0, tzinfo=timezone.utc), + plan={"id": 10, "status": "active"}, + candidate={"id": 20, "trading_plan_id": 10, "status": "executable", "missing_confirmations": []}, + signal={"id": 30}, + ticket={"id": 40, "status": "awaiting_confirmation"}, + risk_result={"allowed": True, "status": "allowed", "reasons": []}, + ) + alerts = build_alert_candidates(packet) + + self.assertEqual(alerts[0]["alert_type"], "candidate_actionable") + self.assertIn("confirm_ticket", alerts[0]["alert_key"]) + + def test_select_candidate_prefers_active_plan_scope(self) -> None: + selected = _select_candidate( + [ + {"id": 1, "trading_plan_id": 9, "model_code": "OTHER"}, + {"id": 2, "trading_plan_id": 10, "model_code": "OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE"}, + ], + plan_id=10, + model_code="OSOK_DAILY_RANGE_SWEEP_MSS_FVG_RETRACE", + ) + + self.assertEqual(selected["id"], 2) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_runtime_script_json.py b/tests/unit/test_runtime_script_json.py index dda66a4..a708f50 100644 --- a/tests/unit/test_runtime_script_json.py +++ b/tests/unit/test_runtime_script_json.py @@ -21,7 +21,9 @@ from scripts import ( run_liquidity, run_mss, run_ranges, + run_decision_packet, run_session_generator, + run_trading_runtime, run_sweeps, run_swings, show_models, @@ -113,6 +115,26 @@ class RuntimeScriptJsonTests(unittest.TestCase): self.assertEqual(exit_code, 0) serve_api.assert_called_once_with(host="127.0.0.1", port=8123) + def test_run_trading_runtime_outputs_json(self) -> None: + service = type("Service", (), {"run_once": lambda self: {"scheduler_state": "running", "current_blockers": []}})() + output = io.StringIO() + with patch("scripts.run_trading_runtime.get_trading_runtime_service", return_value=service), contextlib.redirect_stdout(output): + exit_code = run_trading_runtime.main(env={"TRADING_RUNTIME_OUTPUT_JSON": "1"}) + + payload = json.loads(output.getvalue()) + self.assertEqual(exit_code, 0) + self.assertEqual(payload["scheduler_state"], "running") + + def test_run_decision_packet_outputs_json(self) -> None: + service = type("Service", (), {"build_persist_and_alert": lambda self, instrument_id: {"id": 1, "status": "actionable", "instrument_id": instrument_id}})() + output = io.StringIO() + with patch("scripts.run_decision_packet.DecisionPacketService", return_value=service), contextlib.redirect_stdout(output): + exit_code = run_decision_packet.main(env={"DECISION_PACKET_OUTPUT_JSON": "1", "DECISION_PACKET_INSTRUMENT_ID": "2"}) + + payload = json.loads(output.getvalue()) + self.assertEqual(exit_code, 0) + self.assertEqual(payload["instrument_id"], 2) + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_trading_runtime_service.py b/tests/unit/test_trading_runtime_service.py index 58b743d..71cc794 100644 --- a/tests/unit/test_trading_runtime_service.py +++ b/tests/unit/test_trading_runtime_service.py @@ -85,6 +85,16 @@ class FakeSetupWatcherService: return 401 if candidate is not None else None +class FakeDecisionPacketService: + def build_persist_and_alert(self, **kwargs): + return { + "id": 501, + "status": "actionable", + "action_recommendation": "confirm_ticket", + "alerts": [{"alert_type": "candidate_actionable"}], + } + + class TradingRuntimeSchedulerTests(unittest.TestCase): def test_run_cycle_records_successful_task_state(self) -> None: scheduler = TradingRuntimeScheduler( @@ -92,6 +102,7 @@ class TradingRuntimeSchedulerTests(unittest.TestCase): trading_plan_service=FakeTradingPlanService(), setup_watcher_service=FakeSetupWatcherService(), broker_read_model_service=FakeBrokerReadModelService(), + decision_packet_service=FakeDecisionPacketService(), ) with patch("src.services.runtime.trading_runtime_service.settings.trading_runtime_symbols", "BTC-USDT"): payload = scheduler.run_cycle() @@ -100,6 +111,7 @@ class TradingRuntimeSchedulerTests(unittest.TestCase): self.assertEqual(payload["current_blockers"], []) self.assertEqual(payload["tasks"]["plan"]["last_status"], "ok") self.assertEqual(payload["tasks"]["broker_sync"]["details"]["status"], "broker_state_ok") + self.assertEqual(payload["tasks"]["decision_packet"]["details"]["packet_id"], 501) self.assertEqual(payload["latest_broker_state_by_symbol"]["BTC-USDT"]["account"]["equity"], 1000.0) self.assertTrue(payload["tasks"]["watch"]["next_run_ts"]) @@ -109,6 +121,7 @@ class TradingRuntimeSchedulerTests(unittest.TestCase): trading_plan_service=FakeTradingPlanService(), setup_watcher_service=FakeSetupWatcherService(), broker_read_model_service=FakeBrokerReadModelService(), + decision_packet_service=FakeDecisionPacketService(), ) with patch("src.services.runtime.trading_runtime_service.settings.trading_runtime_symbols", "BTC-USDT"): payload = scheduler.run_cycle() @@ -124,6 +137,7 @@ class TradingRuntimeSchedulerTests(unittest.TestCase): trading_plan_service=FakeTradingPlanService(), setup_watcher_service=FakeSetupWatcherService(), broker_read_model_service=FakeBrokerReadModelService(), + decision_packet_service=FakeDecisionPacketService(), ) supervisor = TradingRuntimeSupervisor(scheduler=scheduler) with patch("src.services.runtime.trading_runtime_service.settings.trading_runtime_symbols", "BTC-USDT"):