Complete runtime scheduling supervision

.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

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:
+- Completed `P0-2` Real Account And Order Read Sync acceptance:
-  - `BrokerReadModelService`
+  - corrected the remote OKX read-only credential mapping
-  - `/trading/broker-state`
+  - verified `/trading/broker-state` returns `broker_state_ok`
-  - `RiskContext.broker_state`
+  - verified `/trading/market-state` carries `broker_state_ok`
-  - execution risk-context integration with broker read model
+- Completed `P0-3` Runtime Scheduling And Supervision implementation:
-  - workbench broker account / position / order / alert summary
+  - `TradingRuntimeScheduler`
-  - `run_until_blocked.py` plus tests
+  - `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`.
+- Autonomous execution protocol is anchored at:
-- Current remote broker-state summary:
+  - `docs/AI_ICT_全自动开发执行流程_v1.md`
-  - `status = broker_state_unavailable`
+- Product roadmap is anchored at:
-  - `error = okx_read_http_401:50111:Invalid OK-ACCESS-KEY`
+  - `docs/AI_ICT_实盘辅助决策系统开发方案_v2.md`
-- This means:
+- Runtime market truth depends on the local LAN proxy remaining available at:
-  - the code path is live
+  - `192.168.3.73:10808`
-  - the API path is live
+- Real broker read sync depends on the remote OKX read-only credentials in:
-  - deploy and migration completed
+  - `/root/deploy/ai-ict/.env`
-  - 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.
 
 ## 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
+- build `DecisionPacketService`
-- make `RiskGovernor` consume real broker state
+- combine plan, candidate, target, invalidation, and Risk Governor into one operator decision packet
-- make the workbench show real `open position / pending order`
+- add status-change alert plumbing
-
+- show the decision packet in the workbench
-When `P0-2` is green, automatically advance to:
-
-### `P0-3` Runtime Scheduling And Supervision
 
 ## 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`
+- Do not enable default live trading.
-- Runtime still depends on local proxy reachability at `192.168.3.73:10808`
+- 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.

TASKS.md

@@ -8,29 +8,10 @@
 
 ## In Progress
 
-- [ ] `P0-2` Real Account And Order Read Sync final acceptance
+- [ ] `P0-4` Decision Packet And Proactive Alerting
-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`
 
 ## 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`
-

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())

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",

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()

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 "")

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"

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",
+]

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

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,

src/web/index.html

@@ -59,6 +59,11 @@
         <div id="marketStatePanel" class="detail-panel"></div>
       </section>
 
+      <section class="panel panel-wide">
+        <h2>Runtime Supervisor</h2>
+        <div id="runtimeStatusPanel" class="detail-panel"></div>
+      </section>
+
       <section class="panel panel-wide">
         <h2>Pre-Market Plan</h2>
         <div id="preMarketPlanPanel" class="detail-panel"></div>

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()