diff --git a/.env.example b/.env.example index 59afcfa..49b91db 100644 --- a/.env.example +++ b/.env.example @@ -29,6 +29,15 @@ NEWS_PROVIDER_MAX_AGE_SECONDS=900 NEWS_EVENT_PRE_WINDOW_MINUTES=15 NEWS_EVENT_POST_WINDOW_MINUTES=15 NEWS_EVENT_HIGH_IMPACT_ONLY=1 +TRADING_RUNTIME_ENABLED=1 +TRADING_RUNTIME_SYMBOLS=BTC-USDT,ETH-USDT +TRADING_RUNTIME_BIAS_TIMEFRAME=1m +TRADING_RUNTIME_SETUP_TIMEFRAME=1m +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_PLAN_INTERVAL_SECONDS=900 DEFAULT_TIMEZONE=UTC SESSION_TIMEZONE=America/New_York LOG_LEVEL=INFO diff --git a/SESSION.md b/SESSION.md index a5a47b0..3c9e44c 100644 --- a/SESSION.md +++ b/SESSION.md @@ -2,7 +2,7 @@ ## Current Goal -Continue the trading-module mainline in autonomous mode. `P0-2` implementation is in place, deployed, and locally validated; the current stop point is the only allowed external blocker: invalid OKX read-only credentials on the remote runtime. +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. ## Current Branch @@ -12,50 +12,48 @@ Continue the trading-module mainline in autonomous mode. `P0-2` implementation i 2026-05-08 UTC -## Completed In Last Session +## Completed In This Run -- Implemented and deployed `P0-2` groundwork: - - `BrokerReadModelService` - - `/trading/broker-state` - - `RiskContext.broker_state` - - execution risk-context integration with broker read model - - workbench broker account / position / order / alert summary - - `run_until_blocked.py` plus tests +- Completed `P0-2` Real Account And Order Read Sync acceptance: + - corrected the remote OKX read-only credential mapping + - verified `/trading/broker-state` returns `broker_state_ok` + - verified `/trading/market-state` carries `broker_state_ok` +- Completed `P0-3` Runtime Scheduling And Supervision implementation: + - `TradingRuntimeScheduler` + - `TradingRuntimeSupervisor` + - runtime task state with last success, next run, blockers, degraded mode, and incidents + - `/trading/runtime-status` + - `/trading/runtime/run-once` + - `scripts/run_trading_runtime.py` + - workbench Runtime Supervisor panel - Local quality gates passed: - `npm run lint` - `npm run build` - `npm run visual:snapshots` -- Updated `scripts/deploy_server.exp` to preserve the remote `.env` file during deploys. ## Current Working State -- Remote `/trading/broker-state` exists and returns `200`. -- Current remote broker-state summary: - - `status = broker_state_unavailable` - - `error = okx_read_http_401:50111:Invalid OK-ACCESS-KEY` -- This means: - - the code path is live - - the API path is live - - deploy and migration completed - - the blocker is not an implementation gap - - the blocker is invalid remote OKX read-only credentials -- Remote market truth can still degrade if the local proxy host `192.168.3.73:10808` is unavailable. +- Autonomous execution protocol is anchored at: + - `docs/AI_ICT_全自动开发执行流程_v1.md` +- Product roadmap is anchored at: + - `docs/AI_ICT_实盘辅助决策系统开发方案_v2.md` +- Runtime market truth depends on the local LAN proxy remaining available at: + - `192.168.3.73:10808` +- Real broker read sync depends on the remote OKX read-only credentials in: + - `/root/deploy/ai-ict/.env` ## Exact Next Step -Wait for valid remote OKX read-only credentials, then continue: +Proceed directly to: -### `P0-2` final acceptance close +### `P0-4` Decision Packet And Proactive Alerting -Targets: +Goals: -- make `/trading/broker-state` return real account / positions / orders / fills -- make `RiskGovernor` consume real broker state -- make the workbench show real `open position / pending order` - -When `P0-2` is green, automatically advance to: - -### `P0-3` Runtime Scheduling And Supervision +- 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 ## Validation Commands @@ -67,6 +65,7 @@ npm run visual:snapshots ## Risks Or Open Questions -- Sole external blocker: `okx_read_http_401:50111:Invalid OK-ACCESS-KEY` -- Runtime still depends on local proxy reachability at `192.168.3.73:10808` - +- Do not enable default live trading. +- Do not let AI bypass Risk Governor decisions. +- If `192.168.3.73:10808` is unavailable, market truth will degrade. +- If OKX read-only credentials are changed or revoked, broker state will degrade. diff --git a/TASKS.md b/TASKS.md index 85b630b..fea1fda 100644 --- a/TASKS.md +++ b/TASKS.md @@ -8,29 +8,10 @@ ## In Progress -- [ ] `P0-2` Real Account And Order Read Sync final acceptance -Files: -`src/services/broker/` -`src/api/broker_state.py` -`src/services/execution/execution_service.py` -`src/services/risk/risk_context.py` -`src/web/app.js` -`tests/unit/test_broker_read_model_service.py` -Acceptance: -- `/trading/broker-state` returns real account / position / order / fill data -- `RiskContext` carries real broker state -- workbench shows real `open position / pending order` -- unknown positions / orders produce alerts -Current State: -- code implemented -- local quality gates passed -- remote deploy complete -- remote blocker: `okx_read_http_401:50111:Invalid OK-ACCESS-KEY` +- [ ] `P0-4` Decision Packet And Proactive Alerting ## Next Up -- [ ] `P0-3` Runtime Scheduling And Supervision -- [ ] `P0-4` Decision Packet And Proactive Alerting - [ ] `P0-5` Real Trade Mapping And Post-Trade Loop - [ ] `P1-1` Trust Scoring And Model Whitelists - [ ] `P1-2` Operator Experience And Workflow Tools @@ -42,11 +23,12 @@ Current State: ## Blocked -- [ ] `P0-2`: `okx_read_http_401:50111:Invalid OK-ACCESS-KEY` +- [ ] No active blocker ## Done - [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] 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/scripts/run_trading_runtime.py b/scripts/run_trading_runtime.py new file mode 100644 index 0000000..efbd74e --- /dev/null +++ b/scripts/run_trading_runtime.py @@ -0,0 +1,37 @@ +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.runtime import get_trading_runtime_service + + +def main(env=None) -> int: + env = os.environ if env is None else env + output_json = _get_bool(env, "TRADING_RUNTIME_OUTPUT_JSON", False) + run_once = _get_bool(env, "TRADING_RUNTIME_RUN_ONCE", True) + service = get_trading_runtime_service() + payload = service.run_once() if run_once else service.status() + if output_json: + print(json.dumps(payload, ensure_ascii=False, sort_keys=True)) + else: + print(f"Trading runtime: {payload.get('scheduler_state')} / {payload.get('supervisor_state', 'stopped')}") + print(f"Blockers: {', '.join(payload.get('current_blockers') or []) or 'none'}") + return 0 if not payload.get("current_blockers") else 2 + + +def _get_bool(env, key: str, default: bool) -> 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 2cdcae6..bea33ba 100644 --- a/src/api/__init__.py +++ b/src/api/__init__.py @@ -24,10 +24,12 @@ from src.api.trading_runtime import ( get_trading_candidate_by_id, get_trading_market_state, get_trading_plan_by_id, + get_trading_runtime_status, get_trading_ticket_by_id, get_weekly_review, rerun_ticket_risk_check, run_candidate_confirm_check, + run_trading_runtime_once, run_trading_watch, sync_execution, get_trading_executions, @@ -48,6 +50,7 @@ __all__ = [ "get_latest_trading_plan", "get_trading_market_state", "get_trading_plan_by_id", + "get_trading_runtime_status", "get_trading_ticket_by_id", "get_trading_executions", "get_trading_journal_summary", @@ -60,6 +63,7 @@ __all__ = [ "run_trading_watch", "get_trading_candidate_by_id", "run_candidate_confirm_check", + "run_trading_runtime_once", "build_ticket_from_candidate", "rerun_ticket_risk_check", "confirm_ticket_only", diff --git a/src/api/server.py b/src/api/server.py index c8f1d5d..10b8b65 100644 --- a/src/api/server.py +++ b/src/api/server.py @@ -25,11 +25,13 @@ from src.api.trading_runtime import ( get_trading_candidate_by_id, get_trading_market_state, get_trading_plan_by_id, + get_trading_runtime_status, get_trading_ticket_by_id, get_trading_executions, get_weekly_review, rerun_ticket_risk_check, run_candidate_confirm_check, + run_trading_runtime_once, run_trading_watch, sync_execution, ) @@ -50,6 +52,7 @@ from src.api.training import ( ) from src.services.execution import ExecutionService from src.services.market_state import start_market_truth_runtime +from src.services.runtime import start_trading_runtime_service def build_api_dependencies() -> dict: @@ -66,6 +69,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_runtime_status": get_trading_runtime_status, + "trading_runtime_run_once": run_trading_runtime_once, "trading_watch_run": run_trading_watch, "trading_candidate_by_id": get_trading_candidate_by_id, "trading_candidate_confirm_check": run_candidate_confirm_check, @@ -166,6 +171,10 @@ def route_request( instrument_id = _require_int(query, "instrument_id") timeframe = _get_single(query, "timeframe", "1m") return 200, deps["trading_market_state"](instrument_id=instrument_id, timeframe=timeframe) + if method == "GET" and path == "/trading/runtime-status": + return 200, deps["trading_runtime_status"]() + if method == "POST" and path == "/trading/runtime/run-once": + return 200, deps["trading_runtime_run_once"]() if method == "POST" and path == "/trading/watch/run": instrument_id = _require_json_int(body or {}, "instrument_id") candidate = deps["trading_watch_run"]( @@ -416,6 +425,7 @@ class APIRequestHandler(BaseHTTPRequestHandler): def serve_api(host: str = "127.0.0.1", port: int = 8000) -> None: start_market_truth_runtime() + start_trading_runtime_service() server = ThreadingHTTPServer((host, port), APIRequestHandler) print(f"AI ICT API listening on http://{host}:{port}") server.serve_forever() diff --git a/src/api/trading_runtime.py b/src/api/trading_runtime.py index 2bac6d6..fe77e59 100644 --- a/src/api/trading_runtime.py +++ b/src/api/trading_runtime.py @@ -11,6 +11,7 @@ from src.services.broker import BrokerReadModelService, BROKER_STATE_OK from src.services.execution import ExecutionService from src.services.journal import TradingJournalService, WeeklyReviewService from src.services.market_state import MarketStateService, get_market_truth_supervisor +from src.services.runtime import get_trading_runtime_service from src.services.trade_plan_candidates import SetupWatcherService from src.services.trading_plans import ACTIVE_PLAN_STATUS, TradingPlanService @@ -397,6 +398,14 @@ def get_trading_broker_state(instrument_id: int) -> dict: return snapshot.to_dict() +def get_trading_runtime_status() -> dict: + return get_trading_runtime_service().status() + + +def run_trading_runtime_once() -> dict: + return get_trading_runtime_service().run_once() + + 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 8661b0c..628fa73 100644 --- a/src/config/settings.py +++ b/src/config/settings.py @@ -37,6 +37,15 @@ class Settings(BaseSettings): news_event_pre_window_minutes: int = 15 news_event_post_window_minutes: int = 15 news_event_high_impact_only: bool = True + trading_runtime_enabled: bool = True + trading_runtime_symbols: str = "BTC-USDT,ETH-USDT" + trading_runtime_bias_timeframe: str = "1m" + trading_runtime_setup_timeframe: str = "1m" + trading_runtime_trigger_timeframe: str = "1m" + 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_plan_interval_seconds: int = 900 default_timezone: str = "UTC" session_timezone: str = "America/New_York" log_level: str = "INFO" diff --git a/src/services/runtime/__init__.py b/src/services/runtime/__init__.py new file mode 100644 index 0000000..b9637bf --- /dev/null +++ b/src/services/runtime/__init__.py @@ -0,0 +1,13 @@ +from src.services.runtime.trading_runtime_service import ( + TradingRuntimeScheduler, + TradingRuntimeSupervisor, + get_trading_runtime_service, + start_trading_runtime_service, +) + +__all__ = [ + "TradingRuntimeScheduler", + "TradingRuntimeSupervisor", + "get_trading_runtime_service", + "start_trading_runtime_service", +] diff --git a/src/services/runtime/trading_runtime_service.py b/src/services/runtime/trading_runtime_service.py new file mode 100644 index 0000000..c1daa6d --- /dev/null +++ b/src/services/runtime/trading_runtime_service.py @@ -0,0 +1,369 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from datetime import datetime, timezone +import threading +import time +from typing import Optional + +from src.config.settings import settings +from src.services.broker import BROKER_STATE_OK, BrokerReadModelService +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 + + +@dataclass +class RuntimeTaskState: + name: str + interval_seconds: int + last_started_ts: Optional[str] = None + last_completed_ts: Optional[str] = None + last_success_ts: Optional[str] = None + next_run_ts: Optional[str] = None + last_status: str = "idle" + last_error: Optional[str] = None + run_count: int = 0 + success_count: int = 0 + failure_count: int = 0 + details: dict = field(default_factory=dict) + + def to_dict(self) -> dict: + return { + "name": self.name, + "interval_seconds": self.interval_seconds, + "last_started_ts": self.last_started_ts, + "last_completed_ts": self.last_completed_ts, + "last_success_ts": self.last_success_ts, + "next_run_ts": self.next_run_ts, + "last_status": self.last_status, + "last_error": self.last_error, + "run_count": self.run_count, + "success_count": self.success_count, + "failure_count": self.failure_count, + "details": dict(self.details or {}), + } + + +class TradingRuntimeScheduler: + def __init__( + self, + *, + market_state_service: Optional[MarketStateService] = None, + trading_plan_service: Optional[TradingPlanService] = None, + setup_watcher_service: Optional[SetupWatcherService] = None, + broker_read_model_service: Optional[BrokerReadModelService] = 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.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 + self.trigger_tf = settings.trading_runtime_trigger_timeframe + self.tasks = { + "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), + "plan": RuntimeTaskState("plan", settings.trading_runtime_plan_interval_seconds), + } + self.current_blockers: list[str] = [] + self.degraded_mode_state: dict = {} + self.incidents: list[dict] = [] + self.latest_market_state_by_symbol: dict[str, dict] = {} + self.latest_broker_state_by_symbol: dict[str, dict] = {} + + def run_cycle(self) -> dict: + if not settings.trading_runtime_enabled: + self.current_blockers = ["trading_runtime_disabled"] + self.degraded_mode_state = { + "is_degraded": True, + "codes": list(self.current_blockers), + } + return self.status_snapshot() + blockers: list[str] = [] + now = datetime.now(timezone.utc) + for symbol in self.symbols: + instrument = self.trading_plan_service.repository.fetch_instrument_by_symbol(symbol) + if not instrument: + blockers.append(f"missing_instrument:{symbol}") + continue + instrument_id = int(instrument["id"]) + self._run_plan_task(symbol=symbol, instrument_id=instrument_id, now=now, blockers=blockers) + 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.current_blockers = _dedupe(blockers) + self.degraded_mode_state = { + "is_degraded": bool(self.current_blockers), + "codes": list(self.current_blockers), + } + return self.status_snapshot() + + def status_snapshot(self) -> dict: + return { + "enabled": bool(settings.trading_runtime_enabled), + "scheduler_state": "running" if settings.trading_runtime_enabled else "disabled", + "symbols": list(self.symbols), + "timeframe_set": { + "bias_tf": self.bias_tf, + "setup_tf": self.setup_tf, + "trigger_tf": self.trigger_tf, + }, + "tasks": {name: task.to_dict() for name, task in self.tasks.items()}, + "current_blockers": list(self.current_blockers), + "degraded_mode_state": dict(self.degraded_mode_state), + "incidents": list(self.incidents[-50:]), + "latest_market_state_by_symbol": dict(self.latest_market_state_by_symbol), + "latest_broker_state_by_symbol": dict(self.latest_broker_state_by_symbol), + } + + def _run_plan_task(self, *, symbol: str, instrument_id: int, now: datetime, blockers: list[str]) -> None: + task = self.tasks["plan"] + if not _task_due(task, now): + return + _task_start(task, now) + try: + candidate = self.trading_plan_service.build_plan( + instrument_id=instrument_id, + bias_tf=self.bias_tf, + setup_tf=self.setup_tf, + trigger_tf=self.trigger_tf, + plan_date=now.date(), + ) + plan_id = self.trading_plan_service.persist_plan(candidate) + task.details = {"symbol": symbol, "instrument_id": instrument_id, "plan_id": plan_id, "status": candidate.status} + _task_success(task, now) + except Exception as exc: + _task_failure(task, now, str(exc)) + code = f"plan_task_failed:{symbol}" + blockers.append(code) + self._record_incident(task_name=task.name, symbol=symbol, code=code, error=str(exc), observed_ts=now) + + def _run_market_state_task(self, *, symbol: str, instrument_id: int, now: datetime, blockers: list[str]) -> dict: + task = self.tasks["market_state"] + if not _task_due(task, now): + return self.latest_market_state_by_symbol.get(symbol, {}) + _task_start(task, now) + try: + market_state = self.market_state_service.build_market_state( + instrument_id=instrument_id, + timeframe=self.trigger_tf, + ) + snapshot_id = self.market_state_service.persist_market_state_snapshot(market_state=market_state) + payload = market_state.to_dict() + payload["snapshot_id"] = snapshot_id + self.latest_market_state_by_symbol[symbol] = payload + task.details = {"symbol": symbol, "instrument_id": instrument_id, "snapshot_id": snapshot_id} + _task_success(task, now) + if payload.get("freshness", {}).get("code") != "market_data_fresh": + code = f"market_state_degraded:{symbol}:{payload.get('freshness', {}).get('code')}" + blockers.append(code) + self._record_incident(task_name=task.name, symbol=symbol, code=code, observed_ts=now) + if payload.get("stream_state", {}).get("code") != "market_stream_connected": + code = f"market_stream_degraded:{symbol}:{payload.get('stream_state', {}).get('code')}" + blockers.append(code) + self._record_incident(task_name=task.name, symbol=symbol, code=code, observed_ts=now) + if payload.get("data_gap_state", {}).get("code") != "market_data_gap_clear": + code = f"market_gap_degraded:{symbol}:{payload.get('data_gap_state', {}).get('code')}" + blockers.append(code) + self._record_incident(task_name=task.name, symbol=symbol, code=code, observed_ts=now) + return payload + except Exception as exc: + _task_failure(task, now, str(exc)) + code = f"market_state_task_failed:{symbol}" + blockers.append(code) + self._record_incident(task_name=task.name, symbol=symbol, code=code, error=str(exc), observed_ts=now) + return self.latest_market_state_by_symbol.get(symbol, {}) + + def _run_broker_sync_task(self, *, symbol: str, instrument_id: int, now: datetime, blockers: list[str]) -> None: + task = self.tasks["broker_sync"] + if not _task_due(task, now): + return + _task_start(task, now) + try: + snapshot = self.broker_read_model_service.fetch_snapshot( + instrument_id=instrument_id, + known_execution_refs={"broker_order_ids": [], "open_execution_count": 0}, + ) + self.latest_broker_state_by_symbol[symbol] = snapshot.to_dict() + task.details = {"symbol": symbol, "instrument_id": instrument_id, "status": snapshot.status} + if snapshot.status != BROKER_STATE_OK: + code = f"broker_sync_degraded:{symbol}:{snapshot.error or snapshot.status}" + blockers.append(code) + self._record_incident(task_name=task.name, symbol=symbol, code=code, error=snapshot.error, observed_ts=now) + _task_success(task, now) + except Exception as exc: + _task_failure(task, now, str(exc)) + code = f"broker_sync_task_failed:{symbol}" + blockers.append(code) + self._record_incident(task_name=task.name, symbol=symbol, code=code, error=str(exc), observed_ts=now) + + def _run_watch_task(self, *, symbol: str, instrument_id: int, now: datetime, market_payload: dict, blockers: list[str]) -> None: + task = self.tasks["watch"] + if not _task_due(task, now): + return + _task_start(task, now) + freshness_code = ((market_payload or {}).get("freshness") or {}).get("code") + stream_code = ((market_payload or {}).get("stream_state") or {}).get("code") + gap_code = ((market_payload or {}).get("data_gap_state") or {}).get("code") + if freshness_code != "market_data_fresh" or stream_code != "market_stream_connected" or gap_code != "market_data_gap_clear": + task.details = {"symbol": symbol, "status": "skipped_market_degraded"} + _task_success(task, now) + code = f"watch_skipped_market_degraded:{symbol}" + blockers.append(code) + self._record_incident(task_name=task.name, symbol=symbol, code=code, observed_ts=now) + return + try: + candidate = self.setup_watcher_service.build_candidate( + instrument_id=instrument_id, + setup_tf=self.setup_tf, + trigger_tf=self.trigger_tf, + plan_date=now.date(), + ) + candidate_id = self.setup_watcher_service.persist_candidate(candidate) if candidate is not None else None + task.details = { + "symbol": symbol, + "instrument_id": instrument_id, + "candidate_id": candidate_id, + "candidate_generated": bool(candidate), + } + _task_success(task, now) + except Exception as exc: + _task_failure(task, now, str(exc)) + code = f"watch_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( + { + "task_name": task_name, + "symbol": symbol, + "code": code, + "error": error, + "observed_ts": observed_ts.isoformat(), + } + ) + if len(self.incidents) > 200: + self.incidents = self.incidents[-200:] + + +class TradingRuntimeSupervisor: + def __init__(self, scheduler: Optional[TradingRuntimeScheduler] = None) -> None: + self.scheduler = scheduler or TradingRuntimeScheduler() + self._thread: Optional[threading.Thread] = None + self._stop_event = threading.Event() + self._lock = threading.RLock() + self._last_status: dict = self.scheduler.status_snapshot() + self._started_ts: Optional[str] = None + + def start(self) -> None: + if not settings.trading_runtime_enabled: + return + with self._lock: + if self._thread and self._thread.is_alive(): + return + self._stop_event.clear() + self._started_ts = datetime.now(timezone.utc).isoformat() + self._thread = threading.Thread(target=self._run_loop, name="trading-runtime-supervisor", daemon=True) + self._thread.start() + + def stop(self) -> None: + self._stop_event.set() + with self._lock: + if self._thread and self._thread.is_alive(): + self._thread.join(timeout=2) + + def status(self) -> dict: + with self._lock: + status = dict(self._last_status) + status["supervisor_state"] = "running" if self._thread and self._thread.is_alive() else "stopped" + status["started_ts"] = self._started_ts + return status + + def run_once(self) -> dict: + status = self.scheduler.run_cycle() + with self._lock: + self._last_status = status + current = dict(status) + current["supervisor_state"] = "running" if self._thread and self._thread.is_alive() else "stopped" + current["started_ts"] = self._started_ts + return current + + def _run_loop(self) -> None: + while not self._stop_event.is_set(): + self.run_once() + for _ in range(5): + if self._stop_event.is_set(): + return + time.sleep(1) + + +_runtime_supervisor: Optional[TradingRuntimeSupervisor] = None + + +def get_trading_runtime_service() -> TradingRuntimeSupervisor: + global _runtime_supervisor + if _runtime_supervisor is None: + _runtime_supervisor = TradingRuntimeSupervisor() + return _runtime_supervisor + + +def start_trading_runtime_service() -> TradingRuntimeSupervisor: + service = get_trading_runtime_service() + service.start() + return service + + +def _task_due(task: RuntimeTaskState, now: datetime) -> bool: + if task.last_started_ts is None: + return True + last_started = _parse_iso(task.last_started_ts) + return (now - last_started).total_seconds() >= max(task.interval_seconds, 1) + + +def _task_start(task: RuntimeTaskState, now: datetime) -> None: + task.last_started_ts = now.isoformat() + task.last_status = "running" + task.last_error = None + task.run_count += 1 + + +def _task_success(task: RuntimeTaskState, now: datetime) -> None: + iso = now.isoformat() + task.last_completed_ts = iso + task.last_success_ts = iso + task.next_run_ts = _next_run_iso(task, now) + task.last_status = "ok" + task.last_error = None + task.success_count += 1 + + +def _task_failure(task: RuntimeTaskState, now: datetime, error: str) -> None: + task.last_completed_ts = now.isoformat() + task.next_run_ts = _next_run_iso(task, now) + task.last_status = "error" + task.last_error = error + task.failure_count += 1 + + +def _parse_iso(value: str) -> datetime: + return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc) + + +def _next_run_iso(task: RuntimeTaskState, now: datetime) -> str: + return datetime.fromtimestamp(now.timestamp() + max(task.interval_seconds, 1), tz=timezone.utc).isoformat() + + +def _dedupe(values: list[str]) -> list[str]: + result = [] + seen = set() + for item in values: + text = str(item or "").strip() + if not text or text in seen: + continue + seen.add(text) + result.append(text) + return result diff --git a/src/web/app.js b/src/web/app.js index eaaa6cf..6c11636 100644 --- a/src/web/app.js +++ b/src/web/app.js @@ -10,6 +10,7 @@ const lastUpdated = document.getElementById('lastUpdated'); const activeFilters = document.getElementById('activeFilters'); const overviewCards = document.getElementById('overviewCards'); const marketStatePanel = document.getElementById('marketStatePanel'); +const runtimeStatusPanel = document.getElementById('runtimeStatusPanel'); const preMarketPlanPanel = document.getElementById('preMarketPlanPanel'); const tradeCandidatesPanel = document.getElementById('tradeCandidatesPanel'); const executionTicketsPanel = document.getElementById('executionTicketsPanel'); @@ -260,6 +261,42 @@ function renderMarketState(payload, executions) { marketStatePanel.appendChild(panel); } +function renderRuntimeStatus(runtimeStatus) { + clearNode(runtimeStatusPanel); + const tasks = runtimeStatus?.tasks || {}; + const blockers = runtimeStatus?.current_blockers || []; + const incidents = runtimeStatus?.incidents || []; + runtimeStatusPanel.appendChild( + buildInfoGrid([ + ['scheduler state', runtimeStatus?.scheduler_state || 'unknown'], + ['supervisor state', runtimeStatus?.supervisor_state || 'unknown'], + ['enabled', String(Boolean(runtimeStatus?.enabled))], + ['degraded', String(Boolean(runtimeStatus?.degraded_mode_state?.is_degraded))], + ['symbols', formatList(runtimeStatus?.symbols || [])], + ['blockers', formatList(blockers)], + ['last incident', incidents.length ? `${incidents[incidents.length - 1].code} @ ${formatTimestamp(incidents[incidents.length - 1].observed_ts)}` : 'none'], + ]) + ); + const grid = createElement('div', 'cards'); + Object.entries(tasks).forEach(([name, task]) => { + grid.appendChild( + buildItemCard( + name, + task.last_status || 'idle', + [ + ['last success', formatTimestamp(task.last_success_ts)], + ['next run', formatTimestamp(task.next_run_ts)], + ['run count', String(task.run_count ?? 0)], + ['success / failure', `${task.success_count ?? 0} / ${task.failure_count ?? 0}`], + ['last error', task.last_error || 'none'], + ['details', formatJsonInline(task.details || {})], + ] + ) + ); + }); + runtimeStatusPanel.appendChild(grid); +} + function renderPreMarketPlan(plan) { clearNode(preMarketPlanPanel); preMarketPlanPanel.appendChild( @@ -608,6 +645,7 @@ function drawCandles(items) { function showErrorPanel(message) { [ marketStatePanel, + runtimeStatusPanel, preMarketPlanPanel, tradeCandidatesPanel, executionTicketsPanel, @@ -628,6 +666,7 @@ function renderDashboardData({ candles, tradingPlan, marketState, + runtimeStatus, tradeCandidates, executionTickets, tradingExecutions, @@ -638,6 +677,7 @@ function renderDashboardData({ renderHealth(health); renderOverview(weeklyReview, tradeCandidates.items || [], executionTickets.items || [], tradingExecutions.items || []); renderMarketState(marketState, tradingExecutions.items || []); + renderRuntimeStatus(runtimeStatus); renderPreMarketPlan(tradingPlan); renderCandidates(tradeCandidates.items || []); renderTickets(executionTickets.items || []); @@ -654,11 +694,12 @@ async function refreshDashboard() { const limit = limitInput.value || '20'; try { - const [health, candles, tradingPlan, marketState, tradeCandidates, executionTickets, tradingExecutions, journalSummary, journalItems, weeklyReview] = await Promise.all([ + const [health, candles, tradingPlan, marketState, runtimeStatus, 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/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: [] })), @@ -672,6 +713,7 @@ async function refreshDashboard() { candles, tradingPlan, marketState, + runtimeStatus, tradeCandidates, executionTickets, tradingExecutions, diff --git a/src/web/index.html b/src/web/index.html index 1733a30..25a36a1 100644 --- a/src/web/index.html +++ b/src/web/index.html @@ -59,6 +59,11 @@
+
+

Runtime Supervisor

+
+
+

Pre-Market Plan

diff --git a/tests/unit/test_trading_runtime_service.py b/tests/unit/test_trading_runtime_service.py new file mode 100644 index 0000000..58b743d --- /dev/null +++ b/tests/unit/test_trading_runtime_service.py @@ -0,0 +1,170 @@ +from datetime import datetime, timezone +import contextlib +import io +import json +import os +import unittest +from unittest.mock import patch + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test") + +from scripts import run_trading_runtime +from src.api.server import route_request +from src.services.runtime import TradingRuntimeScheduler, TradingRuntimeSupervisor + + +class FakeInstrumentRepository: + def fetch_instrument_by_symbol(self, symbol): + return {"id": 1 if symbol == "BTC-USDT" else 2, "symbol": symbol} + + +class FakeTradingPlanService: + def __init__(self): + self.repository = FakeInstrumentRepository() + self.build_calls = 0 + + def build_plan(self, **kwargs): + self.build_calls += 1 + return type("Plan", (), {"status": "active"})() + + def persist_plan(self, candidate): + return 101 + + +class FakeMarketState: + def __init__(self, *, freshness="market_data_fresh", stream="market_stream_connected", gap="market_data_gap_clear"): + self.freshness = freshness + self.stream = stream + self.gap = gap + + def to_dict(self): + return { + "freshness": {"code": self.freshness}, + "stream_state": {"code": self.stream}, + "data_gap_state": {"code": self.gap}, + } + + +class FakeMarketStateService: + def __init__(self, market_state=None): + self.market_state = market_state or FakeMarketState() + + def build_market_state(self, **kwargs): + return self.market_state + + def persist_market_state_snapshot(self, **kwargs): + return 301 + + +class FakeBrokerSnapshot: + status = "broker_state_ok" + error = None + + def to_dict(self): + return { + "status": self.status, + "error": self.error, + "account": {"equity": 1000.0}, + "positions": [], + "orders": [], + "fills": [], + "alerts": [], + } + + +class FakeBrokerReadModelService: + def fetch_snapshot(self, **kwargs): + return FakeBrokerSnapshot() + + +class FakeSetupWatcherService: + def build_candidate(self, **kwargs): + return type("Candidate", (), {"status": "blocked"})() + + def persist_candidate(self, candidate): + return 401 if candidate is not None else None + + +class TradingRuntimeSchedulerTests(unittest.TestCase): + def test_run_cycle_records_successful_task_state(self) -> None: + scheduler = TradingRuntimeScheduler( + market_state_service=FakeMarketStateService(), + trading_plan_service=FakeTradingPlanService(), + setup_watcher_service=FakeSetupWatcherService(), + broker_read_model_service=FakeBrokerReadModelService(), + ) + with patch("src.services.runtime.trading_runtime_service.settings.trading_runtime_symbols", "BTC-USDT"): + payload = scheduler.run_cycle() + + self.assertEqual(payload["scheduler_state"], "running") + 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["latest_broker_state_by_symbol"]["BTC-USDT"]["account"]["equity"], 1000.0) + self.assertTrue(payload["tasks"]["watch"]["next_run_ts"]) + + def test_run_cycle_enters_degraded_mode_when_market_truth_is_bad(self) -> None: + scheduler = TradingRuntimeScheduler( + market_state_service=FakeMarketStateService(FakeMarketState(stream="market_stream_disconnected")), + trading_plan_service=FakeTradingPlanService(), + setup_watcher_service=FakeSetupWatcherService(), + broker_read_model_service=FakeBrokerReadModelService(), + ) + with patch("src.services.runtime.trading_runtime_service.settings.trading_runtime_symbols", "BTC-USDT"): + payload = scheduler.run_cycle() + + self.assertTrue(payload["degraded_mode_state"]["is_degraded"]) + self.assertIn("market_stream_degraded:BTC-USDT:market_stream_disconnected", payload["current_blockers"]) + self.assertIn("watch_skipped_market_degraded:BTC-USDT", payload["current_blockers"]) + self.assertGreaterEqual(len(payload["incidents"]), 1) + + def test_supervisor_run_once_exposes_supervisor_state(self) -> None: + scheduler = TradingRuntimeScheduler( + market_state_service=FakeMarketStateService(), + trading_plan_service=FakeTradingPlanService(), + setup_watcher_service=FakeSetupWatcherService(), + broker_read_model_service=FakeBrokerReadModelService(), + ) + supervisor = TradingRuntimeSupervisor(scheduler=scheduler) + with patch("src.services.runtime.trading_runtime_service.settings.trading_runtime_symbols", "BTC-USDT"): + payload = supervisor.run_once() + + self.assertEqual(payload["supervisor_state"], "stopped") + self.assertEqual(payload["current_blockers"], []) + + +class TradingRuntimeApiTests(unittest.TestCase): + def test_routes_runtime_status_and_run_once(self) -> None: + status, payload = route_request( + "/trading/runtime-status", + {}, + dependencies={"trading_runtime_status": lambda: {"scheduler_state": "running", "current_blockers": []}}, + ) + run_status, run_payload = route_request( + "/trading/runtime/run-once", + {}, + method="POST", + body={}, + dependencies={"trading_runtime_run_once": lambda: {"scheduler_state": "running", "tasks": {}}}, + ) + + self.assertEqual(status, 200) + self.assertEqual(payload["scheduler_state"], "running") + self.assertEqual(run_status, 200) + self.assertIn("tasks", run_payload) + + +class TradingRuntimeScriptTests(unittest.TestCase): + def test_script_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"}) + + self.assertEqual(exit_code, 0) + payload = json.loads(output.getvalue()) + self.assertEqual(payload["scheduler_state"], "running") + + +if __name__ == "__main__": + unittest.main()