diff --git a/.env.example b/.env.example index 630e4f8..36d3619 100644 --- a/.env.example +++ b/.env.example @@ -1,11 +1,30 @@ DATABASE_URL=postgresql+psycopg://postgres:postgres@localhost:5432/ai_ict OKX_BASE_URL=https://www.okx.com OKX_TIMEOUT=30 +OKX_WS_PUBLIC_URL=wss://ws.okx.com:8443/ws/v5/public LIVE_TRADING_ENABLED=0 OKX_ORDER_BASE_URL=https://www.okx.com OKX_API_KEY= OKX_API_SECRET= OKX_API_PASSPHRASE= +MARKET_DATA_STREAM_ENABLED=1 +MARKET_DATA_STREAM_SYMBOLS=BTC-USDT,ETH-USDT +MARKET_DATA_STREAM_TIMEFRAMES=1m +MARKET_DATA_STREAM_HEARTBEAT_SECONDS=30 +MARKET_DATA_STREAM_MAX_LAG_SECONDS=90 +MARKET_DATA_STREAM_RECONNECT_BACKOFF_SECONDS=5 +MARKET_TRUTH_SUPERVISOR_ENABLED=1 +MARKET_TRUTH_SUPERVISOR_INTERVAL_SECONDS=15 +MARKET_TRUTH_FALLBACK_LIMIT=20 +MARKET_TRUTH_GAP_MULTIPLIER=2 +NEWS_PROVIDER_ENABLED=1 +NEWS_PROVIDER_URL=https://nfs.faireconomy.media/ff_calendar_thisweek.json +NEWS_PROVIDER_TIMEOUT=15 +NEWS_PROVIDER_REFRESH_SECONDS=300 +NEWS_PROVIDER_MAX_AGE_SECONDS=900 +NEWS_EVENT_PRE_WINDOW_MINUTES=15 +NEWS_EVENT_POST_WINDOW_MINUTES=15 +NEWS_EVENT_HIGH_IMPACT_ONLY=1 DEFAULT_TIMEZONE=UTC SESSION_TIMEZONE=America/New_York LOG_LEVEL=INFO diff --git a/Dockerfile b/Dockerfile index bcc9c8a..f448ee6 100644 --- a/Dockerfile +++ b/Dockerfile @@ -12,7 +12,8 @@ RUN pip install --no-cache-dir \ "psycopg[binary]>=3.1" \ "pydantic>=2.7" \ "pydantic-settings>=2.2" \ - "httpx>=0.27" + "httpx>=0.27" \ + "websocket-client>=1.8" COPY src /app/src COPY scripts /app/scripts diff --git a/docs/AI_ICT_实盘辅助决策系统开发方案_v2.md b/docs/AI_ICT_实盘辅助决策系统开发方案_v2.md new file mode 100644 index 0000000..f7a9d9a --- /dev/null +++ b/docs/AI_ICT_实盘辅助决策系统开发方案_v2.md @@ -0,0 +1,520 @@ +# AI ICT 实盘辅助决策系统开发方案 v2 + +## 1. 文档目标 + +本方案用于定义下一轮开发目标： + +> 在不默认开启真实自动下单的前提下，尽快把系统升级为可上线使用的实盘辅助分析与决策系统。 + +这里的“可上线”不是指全自动实盘交易，而是指： + +- 系统能够持续获取真实市场与账户事实 +- 系统能够自动生成盘前计划与盘中决策包 +- 系统能够在关键状态变化时主动提醒人工 +- 系统能够把真实交易行为纳入复盘与研究闭环 +- 系统具备足够的可解释性、可审计性与操作安全性 + +## 2. 当前状态判断 + +### 2.1 当前已经具备的能力 + +系统当前已经具备以下基础能力： + +- 结构识别链路： + - session + - swing + - dealing range + - liquidity pool + - sweep + - displacement + - MSS + - FVG +- 盘前计划与盘中候选： + - trading plan + - trade plan candidate + - confirmation chain + - target selection +- 执行与风控骨架： + - market state + - risk governor + - execution ticket + - semi-auto gating +- 研究与复盘： + - plan replay + - journal + - weekly review +- 产品界面： + - trading workbench at `/` + - training console at `/training` + +### 2.2 当前还不能支撑“实盘辅助上线”的原因 + +当前系统仍然不能视为可上线的实盘辅助系统，主要原因如下： + +1. 缺少连续实时行情真值层 + 当前主要依赖数据库中的最新 candle 状态，能判断 stale，但不能保证实时连续供给。 + +2. 缺少真实新闻/事件窗口接入 + news window 和 opening volatility 仍是占位接口，不是稳定外部事件源。 + +3. 缺少真实账户与订单只读同步闭环 + 系统还没有稳定同步真实持仓、挂单、成交、撤单状态。 + +4. 缺少持续运行的盘中调度层 + 现在更多是脚本式能力，而不是全天自动运行的实盘助手。 + +5. 缺少告警/推送层 + 当前更多依赖人工刷新页面，不能在盘中状态变化时主动提醒。 + +6. 缺少面向人工决策的“决策包”聚合层 + 已有 candidate/ticket/risk 数据，但缺少把它们压缩成一张可快速判断的辅助决策卡。 + +因此，下一轮重点不应放在新增更多 ICT 识别器，而应放在： + +```text +实时真值 -> 持续调度 -> 决策包 -> 提醒 -> 真实账户映射 -> 研究闭环 +``` + +## 3. 下一轮总体目标 + +下一轮开发目标定义为： + +```text +发布 Real-time Assisted Trading V1 +``` + +即： + +- 系统不自动替人下真实单 +- 系统可以稳定辅助人工进行实盘分析与决策 +- 系统可以在盘前、盘中、盘后形成完整运营闭环 + +## 4. 范围与非目标 + +### 4.1 本轮范围 + +- 接入真实市场真值与账户真值 +- 实现盘中持续运行 +- 生成可操作的决策包 +- 增加主动提醒 +- 接入真实交易行为映射与复盘 +- 建立上线验收指标 + +### 4.2 非目标 + +- 默认开启真实自动下单 +- 让 LLM 决定是否绕过风控 +- 一轮内支持大量新模型策略 +- 一轮内做完全自动实盘执行 + +## 5. 核心设计原则 + +1. **事实优先** + 行情、账户、持仓、订单、新闻、session 必须来自可验证数据源。 + +2. **规则先于智能体** + 风控、执行许可、失效条件、持仓限制必须由规则引擎控制。 + +3. **AI 负责解释与监督，不负责越权执行** + AI 可以总结、排序、提醒、复盘，但不能跳过硬闸门。 + +4. **只读实盘优先** + 在真实下单之前，必须先把真实市场和真实账户只读同步做稳。 + +5. **先可用，再扩模型** + 先做一个能稳定帮助人决策的系统，再扩更多 setup 和自动化程度。 + +## 6. P0 开发计划 + +## P0-1：实时市场真值层 + +### 目标 + +让系统持续拥有可信、及时的市场状态，而不是仅靠离线 candle 快照。 + +### 任务 + +1. 新增 `MarketDataStreamService` + - 接入 OKX WebSocket + - 维护订阅状态 + - 自动重连 + - 心跳检测 + - lag 监控 + +2. 新增实时 candle / ticker ingest + - 最新价格 + - 最新 candle close/open + - stream delay + - data gap detection + +3. 新增 `MarketTruthSupervisor` + - 定期检测 stream health + - 自动 fallback 到 REST 补数 + - 自动记录 stream incidents + +4. 增强 `MarketStateService` + - freshness 不再只依赖已有数据库最新值 + - 接入 stream health 状态 + - 增加 `stream_state` + - 增加 `data_gap_state` + +5. 接入真实 news/event provider + - 宏观事件窗口 + - 高影响新闻窗口 + - event freshness + +### 验收 + +- BTC / ETH 1m 数据在正常运行时 freshness 长期保持可用 +- stream 断开后系统能自动重连 +- data stale / gap / disconnect 时 candidate 和 ticket 自动降级 +- `/trading/market-state` 能返回： + - session + - freshness + - stream state + - news state + - data gap state + +### 预计工作量 + +- 6-8 人日 + +## P0-2：真实账户与订单只读同步 + +### 目标 + +让系统知道“账户当前真实发生了什么”，从而真正辅助实盘决策。 + +### 任务 + +1. 新增 `BrokerReadModelService` + - 拉取真实账户余额 + - 拉取真实持仓 + - 拉取真实挂单 + - 拉取真实成交 + +2. 定义只读 broker snapshot schema + - account snapshot + - position snapshot + - order snapshot + - fill snapshot + +3. 将真实账户状态接入 `RiskContext` + - open positions + - same-side exposure + - unrealized pnl + - daily pnl + - available balance + +4. 新增 reconciliation 读模型 + - 本地 execution 与真实订单的映射 + - 孤儿订单识别 + - 孤儿持仓识别 + +### 验收 + +- 系统可读取真实账户当前持仓与挂单 +- Risk Governor 可基于真实账户状态工作 +- 页面可显示当前真实 open position / pending order +- 出现未知真实持仓或未知挂单时系统给出告警 + +### 预计工作量 + +- 5-7 人日 + +## P0-3：盘中持续调度与运行监督 + +### 目标 + +把当前“脚本能力”升级为“全天连续运行的实盘辅助流程”。 + +### 任务 + +1. 新增 `TradingRuntimeScheduler` + - 盘前定时生成 plan + - 盘中定时运行 watch + - 定时刷新 market state + - 定时同步账户状态 + +2. 新增 `TradingRuntimeSupervisor` + - 记录任务成功/失败 + - 记录最后成功时间 + - 记录下一次计划执行时间 + - 记录运行 incidents + +3. 新增 runtime health API + - scheduler state + - last successful runs + - current blockers + - degraded mode state + +4. 前端新增 runtime status 区 + - 当前是否正在运行 + - 当前是否 degraded + - 最新成功时间 + - 最近错误 + +### 验收 + +- 不需要人工手动循环跑脚本 +- 盘前计划能按时自动生成 +- 盘中 watch 能持续运行 +- 运行异常时页面能明确显示 + +### 预计工作量 + +- 4-6 人日 + +## P0-4：决策包与主动提醒 + +### 目标 + +让系统给出可操作的决策辅助，而不是只给零散状态。 + +### 任务 + +1. 新增 `DecisionPacketService` + - 汇总 trading plan + - 汇总 candidate + - 汇总 confirmation chain + - 汇总 target plan + - 汇总 invalidation + - 汇总 risk governor decision + +2. 输出统一决策包字段 + - symbol + - narrative + - primary draw + - bias + - setup type + - confirmation chain + - entry zone + - stop + - TP1 / TP2 and source + - invalidation + - expiry + - risk result + - action recommendation + - why allowed / why blocked + +3. 新增提醒通道 + - Telegram 或企业微信或飞书 + - 状态变化提醒 + - candidate -> executable + - executable -> blocked + - ticket confirmed + - market degraded + +4. 新增 workbench 决策卡 UI + - 30 秒内可读 + - 不展示原始噪声为主 + - 重点展示执行与不执行的理由 + +### 验收 + +- candidate 状态变化能自动推送 +- 推送内容能解释： + - 为什么可以看 + - 为什么不能做 + - 目标来自哪里 + - 失效条件是什么 +- workbench 上能直接看到同一张决策卡 + +### 预计工作量 + +- 5-7 人日 + +## P0-5：真实交易映射与盘后闭环 + +### 目标 + +把真实交易行为纳入系统，形成真正的实盘研究闭环。 + +### 任务 + +1. 建立 execution <-> real order/fill 映射 +2. 真实持仓变化自动触发 journal entry +3. 新增真实交易归因字段 + - framework execution + - out-of-model execution + - plan violation + - risk violation + - execution violation + +4. 周报增强 + - 基于真实交易与真实持仓行为 + - 标记哪些辅助是有效的 + - 标记哪些 session 值得继续盯 + +### 验收 + +- 每笔真实交易都能进入 journal +- 系统能区分模型问题还是执行问题 +- 周报能输出明确的下周动作 + +### 预计工作量 + +- 4-6 人日 + +## 7. P1 开发计划 + +## P1-1：辅助可信度与模型白名单 + +### 目标 + +让系统明确知道“什么时候值得信，什么时候只观察”。 + +### 任务 + +1. 按 symbol / session / setup 统计长期表现 +2. 建立辅助可信度评分 +3. 增加 model whitelist / blacklist +4. 给出不同 session 的策略建议 + +### 验收 + +- 系统能明确区分： + - 值得操作 + - 仅观察 + - 禁止执行 + +### 预计工作量 + +- 4-5 人日 + +## P1-2：操作体验与运营工具 + +### 目标 + +让人工使用过程更顺滑，降低盯盘负担。 + +### 任务 + +1. workbench 快捷筛选 +2. session 视图 +3. alert history +4. operator notes +5. 一键生成盘后 review packet + +### 验收 + +- 人工操作链更短 +- 关键状态变化可追溯 + +### 预计工作量 + +- 3-4 人日 + +## P1-3：真实 live adapter 前的准入准备 + +### 目标 + +在不默认启用 live 的前提下，为后续真实下单做好只读和安全准备。 + +### 任务 + +1. 校准真实 order schema +2. 补全 bracket / OCO reconciliation 语义 +3. 建立 live readiness checklist +4. 新增 operator confirmation protocol + +### 验收 + +- 系统具备 live 前评估能力 +- 但默认仍不允许自动真实下单 + +### 预计工作量 + +- 3-4 人日 + +## 8. 上线验收标准 + +只有满足以下条件，才可认为达到“实盘辅助上线”： + +1. 市场数据连续稳定 + - freshness 正常 + - stream 自动重连有效 + - gap 告警有效 + +2. 账户只读同步稳定 + - 真实持仓 + - 真实挂单 + - 真实成交 + - 真实 pnl + +3. 盘前盘中流程自动运行 + - 自动生成 plan + - 自动更新 market state + - 自动运行 watch + +4. 决策包可解释 + - 能说明允许原因 + - 能说明拒绝原因 + - TP1/TP2 可追溯 + - invalidation 明确 + +5. 提醒系统有效 + - 关键状态变化能主动通知 + - 不依赖人工频繁刷新 + +6. 真实交易闭环成立 + - 每笔真实交易都能进 journal + - 周报能输出改进建议 + +7. 安全边界成立 + - 默认无 live 自动下单 + - 风控不能被 LLM 绕过 + - 所有执行动作可审计 + +## 9. 推荐实施顺序 + +建议严格按下面顺序执行： + +1. P0-1 实时市场真值层 +2. P0-2 真实账户与订单只读同步 +3. P0-3 盘中持续调度与运行监督 +4. P0-4 决策包与主动提醒 +5. P0-5 真实交易映射与盘后闭环 +6. P1-1 辅助可信度与模型白名单 +7. P1-2 操作体验与运营工具 +8. P1-3 真实 live adapter 前准备 + +## 10. 预计总工作量 + +### P0 总工作量 + +- 24-34 人日 + +### P1 总工作量 + +- 10-13 人日 + +### 总计 + +- 34-47 人日 + +如果以单人全职推进，建议按 6-8 周排期； +如果 2 人并行推进，建议按 4-5 周排期。 + +## 11. 下一步建议 + +下一步不要同时铺太多线，建议直接启动： + +### 第一阶段启动包 + +- P0-1 实时市场真值层 +- P0-2 真实账户与订单只读同步 +- P0-3 盘中持续调度与运行监督 + +原因： + +- 这是最短路径 +- 这是所有实盘辅助能力的基础 +- 这三块做完，系统才第一次具备“真实盘中连续辅助”的雏形 + +在这三块完成之前，不建议优先投入： + +- 新增更多 setup 模型 +- 新增更多 LLM 叙事能力 +- 真实自动下单 + diff --git a/migrations/029_add_market_truth_runtime_fields.sql b/migrations/029_add_market_truth_runtime_fields.sql new file mode 100644 index 0000000..ba0be7a --- /dev/null +++ b/migrations/029_add_market_truth_runtime_fields.sql @@ -0,0 +1,21 @@ +alter table if exists market_states + add column if not exists stream_state jsonb not null default '{}'::jsonb, + add column if not exists data_gap_state jsonb not null default '{}'::jsonb; + +create table if not exists market_truth_incidents ( + id bigserial primary key, + instrument_id bigint references instruments(id), + timeframe text, + incident_type text not null, + incident_code text not null, + status text not null, + observed_ts timestamptz not null, + details jsonb not null default '{}'::jsonb, + created_at timestamptz not null default now() +); + +create index if not exists idx_market_truth_incidents_instr_observed_desc +on market_truth_incidents (instrument_id, observed_ts desc); + +create index if not exists idx_market_truth_incidents_code_observed_desc +on market_truth_incidents (incident_code, observed_ts desc); diff --git a/pyproject.toml b/pyproject.toml index 6d68720..02b4df5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,6 +9,7 @@ dependencies = [ "pydantic>=2.7", "pydantic-settings>=2.2", "httpx>=0.27", + "websocket-client>=1.8", ] [build-system] diff --git a/scripts/run_execution.py b/scripts/run_execution.py index e0fc65e..1c7c0dc 100644 --- a/scripts/run_execution.py +++ b/scripts/run_execution.py @@ -11,6 +11,7 @@ if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from src.services.execution import AccountRiskState, ExecutionService +from src.services.market_state import MarketStateService, get_market_truth_supervisor def build_account_risk_state_from_env(env=None) -> AccountRiskState: @@ -31,7 +32,9 @@ def build_account_risk_state_from_env(env=None) -> AccountRiskState: def main(env=None) -> int: env = os.environ if env is None else env output_json = _get_bool(env, "EXECUTION_OUTPUT_JSON", False) - service = ExecutionService() + service = ExecutionService( + market_state_service=MarketStateService(market_truth_supervisor=get_market_truth_supervisor()) + ) signal_id = 1 execution_mode = "paper" account_state = build_account_risk_state_from_env(env=env) diff --git a/scripts/run_execution_ticket.py b/scripts/run_execution_ticket.py index c885f8e..cdad022 100644 --- a/scripts/run_execution_ticket.py +++ b/scripts/run_execution_ticket.py @@ -12,6 +12,7 @@ if str(ROOT) not in sys.path: from scripts.run_execution import build_account_risk_state_from_env from src.services.execution import ExecutionService +from src.services.market_state import MarketStateService, get_market_truth_supervisor def build_execution_mode_from_env(env=None) -> str: @@ -28,7 +29,9 @@ def main(env=None) -> int: risk_pct = _get_float(env, "EXECUTION_RISK_PCT", 0.01) confirm_ticket = _get_bool(env, "EXECUTION_CONFIRM_TICKET", False) confirmed_by = str(env.get("EXECUTION_CONFIRMED_BY") or "local_operator") - service = ExecutionService() + service = ExecutionService( + market_state_service=MarketStateService(market_truth_supervisor=get_market_truth_supervisor()) + ) account_state = build_account_risk_state_from_env(env=env) try: ticket = service.create_execution_ticket( diff --git a/scripts/run_risk_governor.py b/scripts/run_risk_governor.py index 573ccb0..e34b7e5 100644 --- a/scripts/run_risk_governor.py +++ b/scripts/run_risk_governor.py @@ -12,7 +12,11 @@ if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from scripts.run_execution import build_account_risk_state_from_env -from src.services.market_state import DEFAULT_MAX_CANDLE_AGE_SECONDS, MarketStateService +from src.services.market_state import ( + DEFAULT_MAX_CANDLE_AGE_SECONDS, + MarketStateService, + get_market_truth_supervisor, +) from src.services.risk import RiskContext, RiskGovernorService @@ -22,7 +26,7 @@ def main(env=None) -> int: instrument_id = _get_int(env, "RISK_INSTRUMENT_ID", 1) timeframe = str(env.get("RISK_TIMEFRAME") or "1m") execution_mode = str(env.get("RISK_EXECUTION_MODE") or "paper") - service = MarketStateService() + service = MarketStateService(market_truth_supervisor=get_market_truth_supervisor()) try: market_state = service.build_market_state( instrument_id=instrument_id, diff --git a/scripts/run_trade_plan_candidates.py b/scripts/run_trade_plan_candidates.py index dd5dab7..c2ced33 100644 --- a/scripts/run_trade_plan_candidates.py +++ b/scripts/run_trade_plan_candidates.py @@ -11,6 +11,7 @@ if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from scripts.run_trading_plans import DEFAULT_TRADING_PLAN_SYMBOLS +from src.api.trading_runtime import run_trading_watch from src.services.trade_plan_candidates import SetupWatcherService @@ -49,17 +50,16 @@ def main(env=None) -> int: if not instrument: candidates.append({"symbol": symbol, "passed": False, "error": "instrument_not_found"}) continue - candidate = service.build_candidate( + candidate = run_trading_watch( instrument_id=int(instrument["id"]), setup_tf=timeframe_set["setup_tf"], trigger_tf=timeframe_set["trigger_tf"], ) - candidate_id = service.persist_candidate(candidate) candidates.append( build_candidate_result( symbol=symbol, instrument_id=int(instrument["id"]), - candidate_id=candidate_id, + candidate_id=int((candidate or {}).get("id") or 0), candidate=candidate, ) ) @@ -102,7 +102,18 @@ def build_candidate_result(symbol: str, instrument_id: int, candidate_id: int, c "candidate_id": 0, "candidate_rows_written": 0, } - confirmation_state = dict(candidate.confirmation_state or {}) + if isinstance(candidate, dict) and candidate.get("candidate_generated") is False: + return { + "symbol": symbol, + "instrument_id": instrument_id, + "passed": True, + "candidate_generated": False, + "reason": candidate.get("status") or "market_truth_blocked", + "reason_codes": list(candidate.get("reason_codes") or []), + "candidate_id": 0, + "candidate_rows_written": 0, + } + confirmation_state = dict((candidate or {}).get("confirmation_state") or {}) return { "symbol": symbol, "instrument_id": instrument_id, @@ -110,11 +121,11 @@ def build_candidate_result(symbol: str, instrument_id: int, candidate_id: int, c "candidate_generated": True, "candidate_id": candidate_id, "candidate_rows_written": 1 if candidate_id else 0, - "trading_plan_id": candidate.trading_plan_id, - "model_code": candidate.model_code, - "status": candidate.status, + "trading_plan_id": candidate.get("trading_plan_id"), + "model_code": candidate.get("model_code"), + "status": candidate.get("status"), "executable": bool(confirmation_state.get("executable")), - "missing_confirmations": list(candidate.missing_confirmations), + "missing_confirmations": list(candidate.get("missing_conditions") or candidate.get("missing_confirmations") or []), "reason_codes": list(confirmation_state.get("reason_codes") or []), } diff --git a/src/api/server.py b/src/api/server.py index 441b198..d334872 100644 --- a/src/api/server.py +++ b/src/api/server.py @@ -48,6 +48,7 @@ from src.api.training import ( submit_training_attempt, ) from src.services.execution import ExecutionService +from src.services.market_state import start_market_truth_runtime def build_api_dependencies() -> dict: @@ -409,6 +410,7 @@ class APIRequestHandler(BaseHTTPRequestHandler): def serve_api(host: str = "127.0.0.1", port: int = 8000) -> None: + start_market_truth_runtime() 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 067fa9e..c4914f3 100644 --- a/src/api/trading_runtime.py +++ b/src/api/trading_runtime.py @@ -9,7 +9,7 @@ from src.repositories.trade_plan_candidate_repository import TradePlanCandidateR from src.repositories.trading_plan_repository import TradingPlanRepository from src.services.execution import ExecutionService from src.services.journal import TradingJournalService, WeeklyReviewService -from src.services.market_state import MarketStateService +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 ACTIVE_PLAN_STATUS, TradingPlanService @@ -68,14 +68,20 @@ def cancel_trading_plan(plan_id: int) -> Optional[dict]: def get_trading_market_state(instrument_id: int, timeframe: str = "1m") -> dict: - service = MarketStateService() + service = build_market_state_service() market_state = service.build_market_state(instrument_id=instrument_id, timeframe=timeframe) snapshot_id = service.persist_market_state_snapshot(market_state=market_state) quality_gate = service.repository.fetch_latest_backtest_quality_gate(instrument_id=instrument_id) + incidents = service.repository.fetch_recent_market_truth_incidents( + instrument_id=instrument_id, + timeframe=timeframe, + limit=10, + ) return { "snapshot_id": snapshot_id, "market_state": market_state.to_dict(), "backtest_quality_gate": quality_gate, + "recent_incidents": incidents, } @@ -85,6 +91,17 @@ def run_trading_watch( setup_tf: str = "1m", trigger_tf: str = "1m", ) -> Optional[dict]: + market_state = build_market_state_service().build_market_state( + instrument_id=instrument_id, + timeframe=trigger_tf, + ) + if market_truth_degraded(market_state): + return { + "candidate_generated": False, + "status": "market_truth_blocked", + "market_state": market_state.to_dict(), + "reason_codes": market_truth_reject_codes(market_state), + } service = SetupWatcherService() candidate = service.build_candidate( instrument_id=instrument_id, @@ -135,7 +152,7 @@ def build_ticket_from_candidate( if not candidate.get("executable"): raise ValueError("candidate_not_executable") signal_id = ensure_trade_signal_for_candidate(candidate_row) - ticket = ExecutionService().create_execution_ticket( + ticket = build_execution_service().create_execution_ticket( signal_id=signal_id, execution_mode=execution_mode, account_equity=account_equity, @@ -145,12 +162,12 @@ def build_ticket_from_candidate( def get_trading_ticket_by_id(ticket_id: int) -> Optional[dict]: - row = ExecutionService().repository.fetch_execution_ticket_by_id(ticket_id) + row = build_execution_service().repository.fetch_execution_ticket_by_id(ticket_id) return dict(row) if row else None def get_trading_executions(instrument_id: int, limit: int = 20) -> list[dict]: - return ExecutionService().build_execution_dashboard_items(instrument_id=instrument_id, limit=limit) + return build_execution_service().build_execution_dashboard_items(instrument_id=instrument_id, limit=limit) def rerun_ticket_risk_check( @@ -159,7 +176,7 @@ def rerun_ticket_risk_check( account_equity: float = 10000.0, risk_pct: float = 0.01, ) -> Optional[dict]: - service = ExecutionService() + service = build_execution_service() ticket = service.repository.fetch_execution_ticket_by_id(ticket_id) if not ticket: return None @@ -183,7 +200,7 @@ def rerun_ticket_risk_check( def confirm_ticket_only(ticket_id: int, confirmed_by: str) -> Optional[dict]: - service = ExecutionService() + service = build_execution_service() ticket = service.repository.fetch_execution_ticket_by_id(ticket_id) if not ticket: return None @@ -239,13 +256,13 @@ def execute_ticket(ticket_id: int, *, confirmed_by: str, execution_mode_expected raise ValueError("ticket_execution_mode_mismatch") if ticket.get("status") != "confirmed": raise ValueError("execution_ticket_not_confirmed") - service = ExecutionService() + service = build_execution_service() full = service.submit_execution_ticket(ticket_id) return full.to_dict() if full else None def cancel_execution(execution_id: int) -> Optional[dict]: - service = ExecutionService() + service = build_execution_service() return service.persist_execution_observation( trade_execution=service.repository.fetch_execution_by_id(execution_id), observation={"source": "manual", "status": "cancelled", "close_reason": "cancelled"}, @@ -253,11 +270,11 @@ def cancel_execution(execution_id: int) -> Optional[dict]: def close_execution(execution_id: int) -> Optional[dict]: - return ExecutionService().close_execution_by_id(execution_id) + return build_execution_service().close_execution_by_id(execution_id) def sync_execution(execution_id: int) -> Optional[dict]: - return ExecutionService().settle_execution_by_id_from_latest_backtest(execution_id) + return build_execution_service().settle_execution_by_id_from_latest_backtest(execution_id) def generate_trading_journal(execution_id: Optional[int] = None) -> dict: @@ -348,3 +365,32 @@ def ensure_trade_signal_for_candidate(candidate_row: dict) -> int: if not created: raise ValueError("signal_not_created_from_candidate") return int(created["id"]) + + +def build_market_state_service() -> MarketStateService: + return MarketStateService(market_truth_supervisor=get_market_truth_supervisor()) + + +def build_execution_service() -> ExecutionService: + return ExecutionService(market_state_service=build_market_state_service()) + + +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 "") + stream_code = str(((payload.get("stream_state") or {}).get("code")) or "") + gap_code = str(((payload.get("data_gap_state") or {}).get("code")) or "") + return freshness_code in {"market_data_missing", "market_data_stale"} or stream_code in { + "market_stream_disconnected", + "market_stream_stale", + } or gap_code == "market_data_gap_detected" + + +def market_truth_reject_codes(market_state) -> list[str]: + payload = market_state.to_dict() if hasattr(market_state, "to_dict") else dict(market_state or {}) + codes = [ + str(((payload.get("freshness") or {}).get("code")) or ""), + str(((payload.get("stream_state") or {}).get("code")) or ""), + str(((payload.get("data_gap_state") or {}).get("code")) or ""), + ] + return [code for code in codes if code and code not in {"market_data_fresh", "market_stream_connected", "market_data_gap_clear"}] diff --git a/src/config/settings.py b/src/config/settings.py index 5f10158..08614c2 100644 --- a/src/config/settings.py +++ b/src/config/settings.py @@ -6,11 +6,30 @@ class Settings(BaseSettings): database_connect_timeout: int = 5 okx_base_url: str = "https://www.okx.com" okx_timeout: int = 30 + okx_ws_public_url: str = "wss://ws.okx.com:8443/ws/v5/public" live_trading_enabled: bool = False okx_order_base_url: str = "https://www.okx.com" okx_api_key: str = "" okx_api_secret: str = "" okx_api_passphrase: str = "" + market_data_stream_enabled: bool = True + market_data_stream_symbols: str = "BTC-USDT,ETH-USDT" + market_data_stream_timeframes: str = "1m" + market_data_stream_heartbeat_seconds: int = 30 + market_data_stream_max_lag_seconds: int = 90 + market_data_stream_reconnect_backoff_seconds: int = 5 + market_truth_supervisor_enabled: bool = True + market_truth_supervisor_interval_seconds: int = 15 + market_truth_fallback_limit: int = 20 + market_truth_gap_multiplier: int = 2 + news_provider_enabled: bool = True + news_provider_url: str = "https://nfs.faireconomy.media/ff_calendar_thisweek.json" + news_provider_timeout: int = 15 + news_provider_refresh_seconds: int = 300 + news_provider_max_age_seconds: int = 900 + news_event_pre_window_minutes: int = 15 + news_event_post_window_minutes: int = 15 + news_event_high_impact_only: bool = True default_timezone: str = "UTC" session_timezone: str = "America/New_York" log_level: str = "INFO" diff --git a/src/main.py b/src/main.py index 0e9daa0..5570018 100644 --- a/src/main.py +++ b/src/main.py @@ -19,6 +19,7 @@ from src.services.displacement import DisplacementService from src.services.execution import AccountRiskState, ExecutionService from src.services.fvg import FVGService from src.services.liquidity import LiquidityPoolService +from src.services.market_state import MarketStateService, get_market_truth_supervisor from src.services.ranges import DealingRangeService from src.services.reports import IntradayReportService, PostTradeReviewService, PreMarketReportService from src.services.sessions import SessionGenerator @@ -30,6 +31,7 @@ from src.services.trade_plan_candidates import SetupWatcherService def build_default_runtime_dependencies(signal_required_evidence_keys: tuple[str, ...] = STRICT_SIGNAL_EVIDENCE_KEYS) -> dict: + market_state_service = MarketStateService(market_truth_supervisor=get_market_truth_supervisor()) return { "session_generator": SessionGenerator(), "swing_service": SwingService(), @@ -42,7 +44,7 @@ def build_default_runtime_dependencies(signal_required_evidence_keys: tuple[str, "bias_service": BiasService(), "signal_service": SignalService(required_evidence_keys=signal_required_evidence_keys), "setup_watcher_service": SetupWatcherService(), - "execution_service": ExecutionService(), + "execution_service": ExecutionService(market_state_service=market_state_service), "backtest_service": BacktestService(), "pre_market_report_service": PreMarketReportService(), "intraday_report_service": IntradayReportService(), diff --git a/src/models/__init__.py b/src/models/__init__.py index 65a6ed9..30af6a0 100644 --- a/src/models/__init__.py +++ b/src/models/__init__.py @@ -13,6 +13,7 @@ from src.models.live_operation_audit import LiveOperationAudit from src.models.liquidity_pool import LiquidityPool from src.models.market_session import MarketSession from src.models.market_state_snapshot import MarketStateSnapshot +from src.models.market_truth_incident import MarketTruthIncident from src.models.structure_event import StructureEvent from src.models.sweep_event import SweepEvent from src.models.swing import Swing @@ -45,6 +46,7 @@ __all__ = [ "LiquidityPool", "MarketSession", "MarketStateSnapshot", + "MarketTruthIncident", "StructureEvent", "SweepEvent", "Swing", diff --git a/src/models/market_state_snapshot.py b/src/models/market_state_snapshot.py index 49e1951..9a02740 100644 --- a/src/models/market_state_snapshot.py +++ b/src/models/market_state_snapshot.py @@ -17,6 +17,8 @@ class MarketStateSnapshot(Base): candle_age_seconds: Mapped[float] = mapped_column(Numeric(18, 4)) news_window_state: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) opening_volatility_state: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) + stream_state: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) + data_gap_state: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) account_state: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) position_state: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) meta: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) diff --git a/src/models/market_truth_incident.py b/src/models/market_truth_incident.py new file mode 100644 index 0000000..d335db7 --- /dev/null +++ b/src/models/market_truth_incident.py @@ -0,0 +1,18 @@ +from sqlalchemy import ForeignKey, Text +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from src.db.base import Base + + +class MarketTruthIncident(Base): + __tablename__ = "market_truth_incidents" + + id: Mapped[int] = mapped_column(primary_key=True) + instrument_id: Mapped[int | None] = mapped_column(ForeignKey("instruments.id")) + timeframe: Mapped[str | None] = mapped_column(Text) + incident_type: Mapped[str] = mapped_column(Text, nullable=False) + incident_code: Mapped[str] = mapped_column(Text, nullable=False) + status: Mapped[str] = mapped_column(Text, nullable=False) + observed_ts: Mapped[object] = mapped_column(nullable=False) + details: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict) diff --git a/src/repositories/market_state_repository.py b/src/repositories/market_state_repository.py index 6e75b43..edb6ca4 100644 --- a/src/repositories/market_state_repository.py +++ b/src/repositories/market_state_repository.py @@ -7,6 +7,35 @@ from src.db.session import SessionLocal class MarketStateRepository: + def fetch_instrument_by_id(self, instrument_id: int) -> Optional[dict]: + sql = text( + """ + select id, symbol, venue_id, active + from instruments + where id = :instrument_id + 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_active_instruments_by_symbols(self, symbols: list[str]) -> list[dict]: + if not symbols: + return [] + sql = text( + """ + select id, symbol, venue_id, active + from instruments + where active = true + and symbol = any(:symbols) + order by id asc + """ + ) + with SessionLocal() as session: + rows = session.execute(sql, {"symbols": symbols}).mappings().all() + return [dict(row) for row in rows] + def fetch_latest_candle(self, instrument_id: int, timeframe: str) -> Optional[dict]: sql = text( """ @@ -73,6 +102,8 @@ class MarketStateRepository: candle_age_seconds, news_window_state: dict, opening_volatility_state: dict, + stream_state: dict, + data_gap_state: dict, account_state: dict, position_state: dict, meta: dict, @@ -88,6 +119,8 @@ class MarketStateRepository: candle_age_seconds, news_window_state, opening_volatility_state, + stream_state, + data_gap_state, account_state, position_state, meta @@ -100,6 +133,8 @@ class MarketStateRepository: :candle_age_seconds, cast(:news_window_state as jsonb), cast(:opening_volatility_state as jsonb), + cast(:stream_state as jsonb), + cast(:data_gap_state as jsonb), cast(:account_state as jsonb), cast(:position_state as jsonb), cast(:meta as jsonb) @@ -119,6 +154,8 @@ class MarketStateRepository: "candle_age_seconds": candle_age_seconds, "news_window_state": json.dumps(news_window_state, default=str), "opening_volatility_state": json.dumps(opening_volatility_state, default=str), + "stream_state": json.dumps(stream_state, default=str), + "data_gap_state": json.dumps(data_gap_state, default=str), "account_state": json.dumps(account_state, default=str), "position_state": json.dumps(position_state, default=str), "meta": json.dumps(meta, default=str), @@ -131,6 +168,7 @@ class MarketStateRepository: """ select id, instrument_id, snapshot_ts, timeframe, session_code, data_freshness, candle_age_seconds, news_window_state, opening_volatility_state, + stream_state, data_gap_state, account_state, position_state, meta from market_states where instrument_id = :instrument_id @@ -141,3 +179,80 @@ class MarketStateRepository: with SessionLocal() as session: rows = session.execute(sql, {"instrument_id": instrument_id, "limit": limit}).mappings().all() return [dict(row) for row in rows] + + def insert_market_truth_incident( + self, + *, + instrument_id: Optional[int], + timeframe: Optional[str], + incident_type: str, + incident_code: str, + status: str, + observed_ts, + details: dict, + ) -> int: + sql = text( + """ + insert into market_truth_incidents ( + instrument_id, + timeframe, + incident_type, + incident_code, + status, + observed_ts, + details + ) values ( + :instrument_id, + :timeframe, + :incident_type, + :incident_code, + :status, + :observed_ts, + cast(:details as jsonb) + ) + returning id + """ + ) + with SessionLocal.begin() as session: + result = session.execute( + sql, + { + "instrument_id": instrument_id, + "timeframe": timeframe, + "incident_type": incident_type, + "incident_code": incident_code, + "status": status, + "observed_ts": observed_ts, + "details": json.dumps(details, default=str), + }, + ) + return int(result.scalar_one()) + + def fetch_recent_market_truth_incidents( + self, + *, + instrument_id: Optional[int] = None, + timeframe: Optional[str] = None, + limit: int = 20, + ) -> list[dict]: + filters = [] + params = {"limit": limit} + if instrument_id is not None: + filters.append("instrument_id = :instrument_id") + params["instrument_id"] = instrument_id + if timeframe is not None: + filters.append("timeframe = :timeframe") + params["timeframe"] = timeframe + where_sql = f"where {' and '.join(filters)}" if filters else "" + sql = text( + f""" + select id, instrument_id, timeframe, incident_type, incident_code, status, observed_ts, details + from market_truth_incidents + {where_sql} + order by observed_ts desc, id desc + limit :limit + """ + ) + with SessionLocal() as session: + rows = session.execute(sql, params).mappings().all() + return [dict(row) for row in rows] diff --git a/src/services/market_state/__init__.py b/src/services/market_state/__init__.py index a8a9efc..b05afeb 100644 --- a/src/services/market_state/__init__.py +++ b/src/services/market_state/__init__.py @@ -1,6 +1,7 @@ """Market state service package.""" from src.services.market_state.market_state_service import ( + DataGapState, DEFAULT_ALLOWED_EXECUTION_SESSIONS, DEFAULT_MAX_CANDLE_AGE_SECONDS, FreshnessState, @@ -9,23 +10,69 @@ from src.services.market_state.market_state_service import ( NewsWindowState, OpeningVolatilityState, SessionState, + StreamState, + build_gap_state, + build_news_window_state_from_provider, build_session_state, + build_stream_state, evaluate_freshness, evaluate_news_window, evaluate_opening_volatility, ) +from src.services.market_state.market_data_stream_service import ( + DATA_GAP_CODE_CLEAR, + DATA_GAP_CODE_DETECTED, + DATA_GAP_CODE_UNAVAILABLE, + STREAM_CODE_CONNECTED, + STREAM_CODE_DISABLED, + STREAM_CODE_DISCONNECTED, + STREAM_CODE_STALE, + MarketDataStreamService, + build_data_gap_state, + timeframe_to_seconds, +) +from src.services.market_state.market_truth_supervisor import ( + MarketTruthRuntimeState, + MarketTruthSupervisor, + build_okx_inst_id, +) +from src.services.market_state.news_event_provider import EconomicCalendarProvider, NewsEvent, NewsProviderSnapshot +from src.services.market_state.runtime import get_market_truth_supervisor, start_market_truth_runtime __all__ = [ + "DATA_GAP_CODE_CLEAR", + "DATA_GAP_CODE_DETECTED", + "DATA_GAP_CODE_UNAVAILABLE", "DEFAULT_ALLOWED_EXECUTION_SESSIONS", "DEFAULT_MAX_CANDLE_AGE_SECONDS", + "DataGapState", + "EconomicCalendarProvider", "FreshnessState", "MarketState", + "MarketDataStreamService", "MarketStateService", + "MarketTruthSupervisor", + "MarketTruthRuntimeState", + "NewsEvent", "NewsWindowState", + "NewsProviderSnapshot", "OpeningVolatilityState", "SessionState", + "STREAM_CODE_CONNECTED", + "STREAM_CODE_DISABLED", + "STREAM_CODE_DISCONNECTED", + "STREAM_CODE_STALE", + "StreamState", + "build_data_gap_state", + "build_gap_state", + "build_news_window_state_from_provider", + "build_okx_inst_id", "build_session_state", + "build_stream_state", "evaluate_freshness", "evaluate_news_window", "evaluate_opening_volatility", + "get_market_truth_supervisor", + "start_market_truth_runtime", + "timeframe_to_seconds", ] diff --git a/src/services/market_state/market_data_stream_service.py b/src/services/market_state/market_data_stream_service.py new file mode 100644 index 0000000..068547f --- /dev/null +++ b/src/services/market_state/market_data_stream_service.py @@ -0,0 +1,438 @@ +from dataclasses import dataclass, field +from datetime import datetime, timezone +import json +import threading +import time +from typing import Optional + +from websocket import WebSocketApp + +from src.config.settings import settings + + +STREAM_CODE_CONNECTING = "market_stream_connecting" +STREAM_CODE_CONNECTED = "market_stream_connected" +STREAM_CODE_STALE = "market_stream_stale" +STREAM_CODE_DISCONNECTED = "market_stream_disconnected" +STREAM_CODE_DISABLED = "market_stream_disabled" +DATA_GAP_CODE_CLEAR = "market_data_gap_clear" +DATA_GAP_CODE_DETECTED = "market_data_gap_detected" +DATA_GAP_CODE_UNAVAILABLE = "market_data_gap_unavailable" + + +@dataclass +class StreamSubscription: + instrument_id: int + symbol: str + inst_id: str + timeframe: str + + @property + def ticker_channel(self) -> str: + return "tickers" + + @property + def candle_channel(self) -> str: + return f"candle{self.timeframe}" + + @property + def key(self) -> tuple[int, str]: + return (self.instrument_id, self.timeframe) + + +@dataclass +class StreamSnapshot: + instrument_id: int + symbol: str + inst_id: str + timeframe: str + stream_code: str = STREAM_CODE_CONNECTING + connected: bool = False + subscribed: bool = False + reconnect_count: int = 0 + last_message_ts: Optional[datetime] = None + last_market_ts: Optional[datetime] = None + last_ticker_ts: Optional[datetime] = None + last_ticker_price: Optional[float] = None + latest_candle_open_price: Optional[float] = None + latest_candle_close_price: Optional[float] = None + latest_candle_volume: Optional[float] = None + latest_candle_open_ts: Optional[datetime] = None + latest_candle_close_ts: Optional[datetime] = None + last_confirmed_candle_close_ts: Optional[datetime] = None + gap_detected: bool = False + gap_seconds: Optional[float] = None + expected_interval_seconds: Optional[int] = None + last_disconnect_reason: Optional[str] = None + last_error: Optional[str] = None + + def to_dict(self, *, now: Optional[datetime] = None, max_lag_seconds: int = 90) -> dict: + now = _ensure_aware(now or datetime.now(timezone.utc)) + heartbeat_age_seconds = None + if self.last_message_ts is not None: + heartbeat_age_seconds = max((now - self.last_message_ts).total_seconds(), 0.0) + lag_seconds = None + if self.last_market_ts is not None: + lag_seconds = max((now - self.last_market_ts).total_seconds(), 0.0) + code = self.stream_code + if code not in {STREAM_CODE_DISABLED, STREAM_CODE_DISCONNECTED} and heartbeat_age_seconds is not None and heartbeat_age_seconds > max_lag_seconds: + code = STREAM_CODE_STALE + elif self.connected and self.subscribed: + code = STREAM_CODE_CONNECTED + return { + "instrument_id": self.instrument_id, + "symbol": self.symbol, + "inst_id": self.inst_id, + "timeframe": self.timeframe, + "code": code, + "connected": self.connected, + "subscribed": self.subscribed, + "reconnect_count": self.reconnect_count, + "last_message_ts": _serialize_dt(self.last_message_ts), + "heartbeat_age_seconds": heartbeat_age_seconds, + "lag_seconds": lag_seconds, + "last_market_ts": _serialize_dt(self.last_market_ts), + "last_ticker_ts": _serialize_dt(self.last_ticker_ts), + "latest_price": self.last_ticker_price, + "latest_candle_open": self.latest_candle_open_price, + "latest_candle_close": self.latest_candle_close_price, + "latest_candle_volume": self.latest_candle_volume, + "latest_candle_open_ts": _serialize_dt(self.latest_candle_open_ts), + "latest_candle_close_ts": _serialize_dt(self.latest_candle_close_ts), + "last_confirmed_candle_close_ts": _serialize_dt(self.last_confirmed_candle_close_ts), + "last_disconnect_reason": self.last_disconnect_reason, + "last_error": self.last_error, + } + + +class MarketDataStreamService: + def __init__( + self, + *, + url: Optional[str] = None, + heartbeat_seconds: Optional[int] = None, + max_lag_seconds: Optional[int] = None, + reconnect_backoff_seconds: Optional[int] = None, + ) -> None: + self.url = url or settings.okx_ws_public_url + self.heartbeat_seconds = heartbeat_seconds or settings.market_data_stream_heartbeat_seconds + self.max_lag_seconds = max_lag_seconds or settings.market_data_stream_max_lag_seconds + self.reconnect_backoff_seconds = reconnect_backoff_seconds or settings.market_data_stream_reconnect_backoff_seconds + self._lock = threading.RLock() + self._subscriptions: dict[tuple[int, str], StreamSubscription] = {} + self._snapshots: dict[tuple[int, str], StreamSnapshot] = {} + self._thread: Optional[threading.Thread] = None + self._stop_event = threading.Event() + self._ws: Optional[WebSocketApp] = None + + def subscribe(self, *, instrument_id: int, symbol: str, inst_id: str, timeframe: str) -> None: + subscription = StreamSubscription( + instrument_id=instrument_id, + symbol=symbol, + inst_id=inst_id, + timeframe=timeframe, + ) + with self._lock: + self._subscriptions[subscription.key] = subscription + self._snapshots.setdefault( + subscription.key, + StreamSnapshot( + instrument_id=instrument_id, + symbol=symbol, + inst_id=inst_id, + timeframe=timeframe, + expected_interval_seconds=timeframe_to_seconds(timeframe), + ), + ) + ws = self._ws + if ws is not None: + self._send_current_subscriptions(ws) + + def start(self) -> None: + with self._lock: + if self._thread and self._thread.is_alive(): + return + self._stop_event.clear() + self._thread = threading.Thread(target=self._run_loop, name="market-data-stream", daemon=True) + self._thread.start() + + def stop(self) -> None: + self._stop_event.set() + with self._lock: + ws = self._ws + if ws is not None: + try: + ws.close() + except Exception: + pass + with self._lock: + if self._thread and self._thread.is_alive(): + self._thread.join(timeout=2) + + def get_snapshot(self, *, instrument_id: int, timeframe: str, now: Optional[datetime] = None) -> dict: + with self._lock: + snapshot = self._snapshots.get((instrument_id, timeframe)) + if snapshot is None: + return { + "instrument_id": instrument_id, + "timeframe": timeframe, + "code": STREAM_CODE_DISCONNECTED if settings.market_data_stream_enabled else STREAM_CODE_DISABLED, + "connected": False, + "subscribed": False, + "reconnect_count": 0, + "last_message_ts": None, + "heartbeat_age_seconds": None, + "lag_seconds": None, + "latest_price": None, + "latest_candle_open": None, + "latest_candle_close": None, + "latest_candle_open_ts": None, + "latest_candle_close_ts": None, + "last_confirmed_candle_close_ts": None, + "last_disconnect_reason": "subscription_missing", + "last_error": None, + } + return snapshot.to_dict(now=now, max_lag_seconds=self.max_lag_seconds) + + def iter_subscriptions(self) -> list[StreamSubscription]: + with self._lock: + return list(self._subscriptions.values()) + + def _run_loop(self) -> None: + while not self._stop_event.is_set(): + self._mark_connecting() + ws = WebSocketApp( + self.url, + on_open=self._on_open, + on_message=self._on_message, + on_error=self._on_error, + on_close=self._on_close, + ) + with self._lock: + self._ws = ws + try: + ws.run_forever(ping_interval=max(1, self.heartbeat_seconds), ping_timeout=max(1, self.heartbeat_seconds // 2)) + except Exception as exc: + self._mark_disconnected(reason=str(exc), mark_error=True) + finally: + with self._lock: + self._ws = None + if self._stop_event.is_set(): + break + time.sleep(max(self.reconnect_backoff_seconds, 1)) + self._increment_reconnects() + + def _on_open(self, ws: WebSocketApp) -> None: + self._send_current_subscriptions(ws) + with self._lock: + for snapshot in self._snapshots.values(): + snapshot.connected = True + snapshot.stream_code = STREAM_CODE_CONNECTED + snapshot.last_disconnect_reason = None + snapshot.last_error = None + + def _on_message(self, _ws: WebSocketApp, message: str) -> None: + payload = json.loads(message) + if payload.get("event") == "subscribe": + arg = dict(payload.get("arg") or {}) + inst_id = str(arg.get("instId") or "") + channel = str(arg.get("channel") or "") + with self._lock: + for subscription in self._subscriptions.values(): + if subscription.inst_id == inst_id and channel in {subscription.ticker_channel, subscription.candle_channel}: + snapshot = self._snapshots[subscription.key] + snapshot.subscribed = True + snapshot.connected = True + return + arg = dict(payload.get("arg") or {}) + data = list(payload.get("data") or []) + channel = str(arg.get("channel") or "") + inst_id = str(arg.get("instId") or "") + if not channel or not inst_id or not data: + return + with self._lock: + subscriptions = [item for item in self._subscriptions.values() if item.inst_id == inst_id] + received_ts = datetime.now(timezone.utc) + for subscription in subscriptions: + if channel not in {subscription.ticker_channel, subscription.candle_channel}: + continue + snapshot = self._snapshots[subscription.key] + snapshot.connected = True + snapshot.subscribed = True + snapshot.last_message_ts = received_ts + snapshot.stream_code = STREAM_CODE_CONNECTED + if channel == subscription.ticker_channel: + self._apply_ticker_message(snapshot, data) + elif channel == subscription.candle_channel: + self._apply_candle_message(snapshot, data) + + def _apply_ticker_message(self, snapshot: StreamSnapshot, data: list[dict]) -> None: + ticker = dict(data[0] or {}) + snapshot.last_ticker_price = _to_optional_float(ticker.get("last")) + snapshot.last_ticker_ts = _parse_market_datetime(ticker.get("ts")) + if snapshot.last_ticker_ts is not None: + snapshot.last_market_ts = snapshot.last_ticker_ts + + def _apply_candle_message(self, snapshot: StreamSnapshot, data: list[list]) -> None: + row = list(data[0] or []) + if len(row) < 6: + return + open_ts = _parse_market_datetime(row[0]) + open_price = _to_optional_float(row[1]) + close_price = _to_optional_float(row[4]) + volume = _to_optional_float(row[5]) + confirm = str(row[8]) if len(row) > 8 else "0" + interval_seconds = snapshot.expected_interval_seconds or timeframe_to_seconds(snapshot.timeframe) + if snapshot.latest_candle_open_ts and open_ts and interval_seconds: + expected_next = snapshot.latest_candle_open_ts.timestamp() + interval_seconds + actual = open_ts.timestamp() + if actual - expected_next > 1: + snapshot.gap_detected = True + snapshot.gap_seconds = max(actual - expected_next, 0.0) + snapshot.latest_candle_open_ts = open_ts + snapshot.latest_candle_close_ts = ( + datetime.fromtimestamp(open_ts.timestamp() + interval_seconds - 1, tz=timezone.utc) + if open_ts and interval_seconds + else None + ) + snapshot.latest_candle_open_price = open_price + snapshot.latest_candle_close_price = close_price + snapshot.latest_candle_volume = volume + snapshot.last_market_ts = snapshot.latest_candle_close_ts or open_ts + if confirm == "1": + snapshot.last_confirmed_candle_close_ts = snapshot.latest_candle_close_ts + + def _on_error(self, _ws: WebSocketApp, error) -> None: + self._mark_disconnected(reason=str(error), mark_error=True) + + def _on_close(self, _ws: WebSocketApp, status_code, message) -> None: + reason = f"{status_code}:{message}" if status_code is not None or message else "closed" + self._mark_disconnected(reason=reason, mark_error=False) + + def _send_current_subscriptions(self, ws: WebSocketApp) -> None: + args: list[dict] = [] + with self._lock: + subscriptions = list(self._subscriptions.values()) + for subscription in subscriptions: + args.append({"channel": subscription.ticker_channel, "instId": subscription.inst_id}) + args.append({"channel": subscription.candle_channel, "instId": subscription.inst_id}) + if not args: + return + try: + ws.send(json.dumps({"op": "subscribe", "args": args})) + except Exception as exc: + self._mark_disconnected(reason=str(exc), mark_error=True) + + def _mark_connecting(self) -> None: + with self._lock: + for snapshot in self._snapshots.values(): + snapshot.connected = False + snapshot.subscribed = False + snapshot.stream_code = STREAM_CODE_CONNECTING + + def _mark_disconnected(self, *, reason: str, mark_error: bool) -> None: + with self._lock: + for snapshot in self._snapshots.values(): + snapshot.connected = False + snapshot.subscribed = False + snapshot.stream_code = STREAM_CODE_DISCONNECTED + snapshot.last_disconnect_reason = reason + if mark_error: + snapshot.last_error = reason + + def _increment_reconnects(self) -> None: + with self._lock: + for snapshot in self._snapshots.values(): + snapshot.reconnect_count += 1 + + +def build_data_gap_state( + *, + stream_snapshot: dict, + latest_candle_ts, + timeframe: str, + now: Optional[datetime] = None, + gap_multiplier: int = 2, +) -> dict: + now = _ensure_aware(now or datetime.now(timezone.utc)) + expected_interval_seconds = timeframe_to_seconds(timeframe) + if stream_snapshot.get("code") in {STREAM_CODE_DISABLED, STREAM_CODE_DISCONNECTED}: + return { + "is_gap": True, + "code": DATA_GAP_CODE_UNAVAILABLE, + "expected_interval_seconds": expected_interval_seconds, + "gap_seconds": None, + "last_candle_ts": _serialize_dt(latest_candle_ts), + "source": "stream", + } + last_seen_ts = _ensure_aware_or_none(latest_candle_ts) or _ensure_aware_or_none(stream_snapshot.get("last_confirmed_candle_close_ts")) + if last_seen_ts is None: + return { + "is_gap": True, + "code": DATA_GAP_CODE_UNAVAILABLE, + "expected_interval_seconds": expected_interval_seconds, + "gap_seconds": None, + "last_candle_ts": None, + "source": "stream", + } + age_seconds = max((now - last_seen_ts).total_seconds(), 0.0) + is_gap = age_seconds > (expected_interval_seconds * max(gap_multiplier, 1)) + return { + "is_gap": is_gap or bool(stream_snapshot.get("gap_detected")), + "code": DATA_GAP_CODE_DETECTED if is_gap or bool(stream_snapshot.get("gap_detected")) else DATA_GAP_CODE_CLEAR, + "expected_interval_seconds": expected_interval_seconds, + "gap_seconds": max(stream_snapshot.get("gap_seconds") or 0.0, age_seconds - expected_interval_seconds) if (is_gap or stream_snapshot.get("gap_detected")) else 0.0, + "last_candle_ts": _serialize_dt(last_seen_ts), + "source": "stream", + } + + +def timeframe_to_seconds(timeframe: str) -> int: + text = str(timeframe or "1m").strip().lower() + if text.endswith("m"): + return int(text[:-1]) * 60 + if text.endswith("h"): + return int(text[:-1]) * 60 * 60 + if text.endswith("d"): + return int(text[:-1]) * 24 * 60 * 60 + raise ValueError(f"unsupported_timeframe:{timeframe}") + + +def _parse_market_datetime(value) -> Optional[datetime]: + if value is None or value == "": + return None + text = str(value) + if text.isdigit(): + return datetime.fromtimestamp(int(text) / 1000, tz=timezone.utc) + return _ensure_aware_or_none(text) + + +def _ensure_aware(value) -> datetime: + parsed = _ensure_aware_or_none(value) + if parsed is None: + raise ValueError("datetime_required") + return parsed + + +def _ensure_aware_or_none(value) -> Optional[datetime]: + if value is None or value == "": + return None + if isinstance(value, datetime): + parsed = value + else: + text = str(value) + parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) + if parsed.tzinfo is None: + return parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def _serialize_dt(value) -> Optional[str]: + dt = _ensure_aware_or_none(value) + return dt.isoformat() if dt is not None else None + + +def _to_optional_float(value) -> Optional[float]: + if value is None or value == "": + return None + return float(value) diff --git a/src/services/market_state/market_state_service.py b/src/services/market_state/market_state_service.py index cdca2d5..c73ae38 100644 --- a/src/services/market_state/market_state_service.py +++ b/src/services/market_state/market_state_service.py @@ -3,7 +3,14 @@ from datetime import datetime, timedelta, timezone from decimal import Decimal from typing import Optional +from src.config.settings import settings from src.repositories.market_state_repository import MarketStateRepository +from src.services.market_state.market_data_stream_service import ( + DATA_GAP_CODE_UNAVAILABLE, + STREAM_CODE_DISABLED, + STREAM_CODE_DISCONNECTED, +) +from src.services.market_state.market_truth_supervisor import MarketTruthSupervisor DEFAULT_MAX_CANDLE_AGE_SECONDS = 300 @@ -42,6 +49,47 @@ class NewsWindowState: window_id: Optional[str] start_ts: Optional[object] end_ts: Optional[object] + provider_name: Optional[str] = None + fetched_ts: Optional[object] = None + freshness_code: Optional[str] = None + event_count: int = 0 + active_events: Optional[list[dict]] = None + error: Optional[str] = None + + def to_dict(self) -> dict: + return _serialize(asdict(self)) + + +@dataclass(frozen=True) +class StreamState: + code: str + connected: bool + subscribed: bool + reconnect_count: int + last_message_ts: Optional[object] + heartbeat_age_seconds: Optional[float] + lag_seconds: Optional[float] + latest_price: Optional[float] + latest_candle_open: Optional[float] + latest_candle_close: Optional[float] + latest_candle_open_ts: Optional[object] + latest_candle_close_ts: Optional[object] + last_confirmed_candle_close_ts: Optional[object] + last_disconnect_reason: Optional[str] + last_error: Optional[str] + + def to_dict(self) -> dict: + return _serialize(asdict(self)) + + +@dataclass(frozen=True) +class DataGapState: + is_gap: bool + code: str + expected_interval_seconds: Optional[int] + gap_seconds: Optional[float] + last_candle_ts: Optional[object] + source: str def to_dict(self) -> dict: return _serialize(asdict(self)) @@ -68,26 +116,35 @@ class MarketState: freshness: FreshnessState session: SessionState news_window: NewsWindowState + stream_state: StreamState + data_gap_state: DataGapState opening_volatility: OpeningVolatilityState def to_dict(self) -> dict: - return _serialize( - { - "instrument_id": self.instrument_id, - "timeframe": self.timeframe, - "observed_ts": self.observed_ts, - "latest_candle": dict(self.latest_candle or {}), - "freshness": self.freshness.to_dict(), - "session": self.session.to_dict(), - "news_window": self.news_window.to_dict(), - "opening_volatility": self.opening_volatility.to_dict(), - } - ) + payload = { + "instrument_id": self.instrument_id, + "timeframe": self.timeframe, + "observed_ts": self.observed_ts, + "latest_candle": dict(self.latest_candle or {}), + "freshness": self.freshness.to_dict(), + "session": self.session.to_dict(), + "news_window": self.news_window.to_dict(), + "news_state": self.news_window.to_dict(), + "stream_state": self.stream_state.to_dict(), + "data_gap_state": self.data_gap_state.to_dict(), + "opening_volatility": self.opening_volatility.to_dict(), + } + return _serialize(payload) class MarketStateService: - def __init__(self, repository: Optional[MarketStateRepository] = None) -> None: + def __init__( + self, + repository: Optional[MarketStateRepository] = None, + market_truth_supervisor: Optional[MarketTruthSupervisor] = None, + ) -> None: self.repository = repository or MarketStateRepository() + self.market_truth_supervisor = market_truth_supervisor def build_market_state( self, @@ -100,8 +157,28 @@ class MarketStateService: opening_volatility_minutes: int = 0, ) -> MarketState: observed_ts = _ensure_aware(now or datetime.now(timezone.utc)) + runtime_state = None + if self.market_truth_supervisor is not None: + runtime_state = self.market_truth_supervisor.current_state( + instrument_id=instrument_id, + timeframe=timeframe, + now=observed_ts, + ) + latest_candle = self.repository.fetch_latest_candle(instrument_id=instrument_id, timeframe=timeframe) latest_ts = latest_candle.get("ts_close") if latest_candle else None + stream_state = build_stream_state(runtime_state.stream_state if runtime_state else None) + data_gap_state = build_gap_state(runtime_state.data_gap_state if runtime_state else None, timeframe=timeframe) + if latest_candle and stream_state.latest_candle_open is not None: + latest_candle = { + **dict(latest_candle), + "open": stream_state.latest_candle_open, + "close": stream_state.latest_candle_close, + "latest_price": stream_state.latest_price, + "ts_close": stream_state.last_confirmed_candle_close_ts or latest_candle.get("ts_close"), + } + latest_ts = latest_candle.get("ts_close") + freshness = evaluate_freshness( latest_ts=latest_ts, now=observed_ts, @@ -109,7 +186,10 @@ class MarketStateService: ) session_row = self.repository.fetch_session_at(instrument_id=instrument_id, ts=observed_ts) session = build_session_state(session_row) - news_window = evaluate_news_window(now=observed_ts, windows=news_windows or []) + if runtime_state is not None: + news_window = build_news_window_state_from_provider(runtime_state.news_snapshot) + else: + news_window = evaluate_news_window(now=observed_ts, windows=news_windows or []) opening_volatility = evaluate_opening_volatility( now=observed_ts, session=session, @@ -123,6 +203,8 @@ class MarketStateService: freshness=freshness, session=session, news_window=news_window, + stream_state=stream_state, + data_gap_state=data_gap_state, opening_volatility=opening_volatility, ) @@ -143,6 +225,8 @@ class MarketStateService: candle_age_seconds=market_state.freshness.age_seconds, news_window_state=market_state.news_window.to_dict(), opening_volatility_state=market_state.opening_volatility.to_dict(), + stream_state=market_state.stream_state.to_dict(), + data_gap_state=market_state.data_gap_state.to_dict(), account_state=dict(account_state or {}), position_state=dict(position_state or {}), meta={ @@ -207,6 +291,7 @@ def evaluate_news_window(*, now: datetime, windows: list[dict]) -> NewsWindowSta window_id=str(window.get("id") or f"window:{index}"), start_ts=start_ts, end_ts=end_ts, + active_events=[], ) return NewsWindowState( is_active=False, @@ -214,6 +299,40 @@ def evaluate_news_window(*, now: datetime, windows: list[dict]) -> NewsWindowSta window_id=None, start_ts=None, end_ts=None, + active_events=[], + ) + + +def build_news_window_state_from_provider(snapshot) -> NewsWindowState: + if snapshot is None: + return NewsWindowState( + is_active=False, + code="news_provider_stale", + window_id=None, + start_ts=None, + end_ts=None, + provider_name=None, + fetched_ts=None, + freshness_code="news_provider_stale", + event_count=0, + active_events=[], + error="snapshot_missing", + ) + active_events = list(snapshot.active_events or []) + first_event = active_events[0] if active_events else {} + event_ts = first_event.get("event_ts") + return NewsWindowState( + is_active=bool(snapshot.is_active), + code=str(snapshot.code), + window_id=str(first_event.get("event_id")) if first_event else None, + start_ts=event_ts, + end_ts=event_ts, + provider_name=snapshot.provider_name, + fetched_ts=snapshot.fetched_ts, + freshness_code=snapshot.freshness_code, + event_count=int(snapshot.event_count), + active_events=active_events, + error=snapshot.error, ) @@ -242,6 +361,43 @@ def evaluate_opening_volatility( ) +def build_stream_state(payload: Optional[dict]) -> StreamState: + payload = dict(payload or {}) + code = str(payload.get("code") or (STREAM_CODE_DISABLED if not settings.market_data_stream_enabled else STREAM_CODE_DISCONNECTED)) + return StreamState( + code=code, + connected=bool(payload.get("connected")), + subscribed=bool(payload.get("subscribed")), + reconnect_count=int(payload.get("reconnect_count") or 0), + last_message_ts=payload.get("last_message_ts"), + heartbeat_age_seconds=_float_or_none(payload.get("heartbeat_age_seconds")), + lag_seconds=_float_or_none(payload.get("lag_seconds")), + latest_price=_float_or_none(payload.get("latest_price")), + latest_candle_open=_float_or_none(payload.get("latest_candle_open")), + latest_candle_close=_float_or_none(payload.get("latest_candle_close")), + latest_candle_open_ts=payload.get("latest_candle_open_ts"), + latest_candle_close_ts=payload.get("latest_candle_close_ts"), + last_confirmed_candle_close_ts=payload.get("last_confirmed_candle_close_ts"), + last_disconnect_reason=payload.get("last_disconnect_reason"), + last_error=payload.get("last_error"), + ) + + +def build_gap_state(payload: Optional[dict], *, timeframe: Optional[str] = None) -> DataGapState: + payload = dict(payload or {}) + expected_interval_seconds = payload.get("expected_interval_seconds") + if expected_interval_seconds is None and timeframe: + expected_interval_seconds = _timeframe_to_seconds(timeframe) + return DataGapState( + is_gap=bool(payload.get("is_gap", True if not payload else False)), + code=str(payload.get("code") or DATA_GAP_CODE_UNAVAILABLE), + expected_interval_seconds=int(expected_interval_seconds) if expected_interval_seconds is not None else None, + gap_seconds=_float_or_none(payload.get("gap_seconds")), + last_candle_ts=payload.get("last_candle_ts"), + source=str(payload.get("source") or "stream"), + ) + + def _ensure_aware(value) -> datetime: if isinstance(value, datetime): parsed = value @@ -268,3 +424,20 @@ def _serialize(value): if isinstance(value, timedelta): return value.total_seconds() return value + + +def _float_or_none(value) -> Optional[float]: + if value is None or value == "": + return None + return float(value) + + +def _timeframe_to_seconds(timeframe: str) -> int: + text = str(timeframe or "1m").strip().lower() + if text.endswith("m"): + return int(text[:-1]) * 60 + if text.endswith("h"): + return int(text[:-1]) * 60 * 60 + if text.endswith("d"): + return int(text[:-1]) * 24 * 60 * 60 + raise ValueError(f"unsupported_timeframe:{timeframe}") diff --git a/src/services/market_state/market_truth_supervisor.py b/src/services/market_state/market_truth_supervisor.py new file mode 100644 index 0000000..1c0e6c8 --- /dev/null +++ b/src/services/market_state/market_truth_supervisor.py @@ -0,0 +1,331 @@ +from dataclasses import dataclass +from datetime import datetime, timezone +import threading +import time +from typing import Optional + +from src.config.settings import settings +from src.repositories.candle_repository import CandleRepository +from src.repositories.market_state_repository import MarketStateRepository +from src.services.market_data import OKXClient, normalize_okx_candles +from src.services.market_data.okx_client import OKXCandle +from src.services.market_state.market_data_stream_service import ( + DATA_GAP_CODE_CLEAR, + DATA_GAP_CODE_DETECTED, + STREAM_CODE_CONNECTED, + STREAM_CODE_DISCONNECTED, + STREAM_CODE_STALE, + MarketDataStreamService, + build_data_gap_state, +) +from src.services.market_state.news_event_provider import EconomicCalendarProvider, NewsProviderSnapshot + + +INCIDENT_TYPE_STREAM = "stream" +INCIDENT_TYPE_FALLBACK = "fallback" +INCIDENT_TYPE_NEWS = "news" +INCIDENT_STATUS_OPEN = "open" +INCIDENT_STATUS_RESOLVED = "resolved" + + +@dataclass(frozen=True) +class MarketTruthRuntimeState: + stream_state: dict + data_gap_state: dict + news_snapshot: NewsProviderSnapshot + + +class MarketTruthSupervisor: + def __init__( + self, + *, + repository: Optional[MarketStateRepository] = None, + candle_repository: Optional[CandleRepository] = None, + stream_service: Optional[MarketDataStreamService] = None, + rest_client: Optional[OKXClient] = None, + news_provider: Optional[EconomicCalendarProvider] = None, + enabled: Optional[bool] = None, + interval_seconds: Optional[int] = None, + fallback_limit: Optional[int] = None, + gap_multiplier: Optional[int] = None, + ) -> None: + self.repository = repository or MarketStateRepository() + self.candle_repository = candle_repository or CandleRepository() + self.stream_service = stream_service or MarketDataStreamService() + self.rest_client = rest_client or OKXClient() + self.news_provider = news_provider or EconomicCalendarProvider() + self.enabled = settings.market_truth_supervisor_enabled if enabled is None else enabled + self.interval_seconds = interval_seconds or settings.market_truth_supervisor_interval_seconds + self.fallback_limit = fallback_limit or settings.market_truth_fallback_limit + self.gap_multiplier = gap_multiplier or settings.market_truth_gap_multiplier + self._lock = threading.RLock() + self._thread: Optional[threading.Thread] = None + self._stop_event = threading.Event() + self._news_snapshot: Optional[NewsProviderSnapshot] = None + self._last_news_refresh_ts: Optional[datetime] = None + self._last_incident_status: dict[tuple[Optional[int], Optional[str], str], str] = {} + self._ensure_bootstrap_subscriptions() + + def start(self) -> None: + if not self.enabled: + return + with self._lock: + if self._thread and self._thread.is_alive(): + return + self._stop_event.clear() + self.stream_service.start() + self._thread = threading.Thread(target=self._run_loop, name="market-truth-supervisor", daemon=True) + self._thread.start() + + def stop(self) -> None: + self._stop_event.set() + self.stream_service.stop() + with self._lock: + if self._thread and self._thread.is_alive(): + self._thread.join(timeout=2) + + def ensure_subscription(self, *, instrument_id: int, timeframe: str) -> None: + instrument = self.repository.fetch_instrument_by_id(instrument_id) + if not instrument: + return + symbol = str(instrument.get("symbol") or "") + if not symbol: + return + self.stream_service.subscribe( + instrument_id=instrument_id, + symbol=symbol, + inst_id=build_okx_inst_id(symbol), + timeframe=timeframe, + ) + if self.enabled: + self.start() + + def current_state(self, *, instrument_id: int, timeframe: str, now: Optional[datetime] = None) -> MarketTruthRuntimeState: + now = _ensure_aware(now or datetime.now(timezone.utc)) + self.ensure_subscription(instrument_id=instrument_id, timeframe=timeframe) + latest_candle = self.repository.fetch_latest_candle(instrument_id=instrument_id, timeframe=timeframe) + latest_candle_ts = (latest_candle or {}).get("ts_close") + stream_state = self.stream_service.get_snapshot(instrument_id=instrument_id, timeframe=timeframe, now=now) + data_gap_state = build_data_gap_state( + stream_snapshot=stream_state, + latest_candle_ts=latest_candle_ts, + timeframe=timeframe, + now=now, + gap_multiplier=self.gap_multiplier, + ) + news_snapshot = self._get_news_snapshot(now=now) + if self._should_fallback(stream_state=stream_state, data_gap_state=data_gap_state): + self._run_rest_fallback(instrument_id=instrument_id, timeframe=timeframe, now=now) + latest_candle = self.repository.fetch_latest_candle(instrument_id=instrument_id, timeframe=timeframe) + latest_candle_ts = (latest_candle or {}).get("ts_close") + data_gap_state = build_data_gap_state( + stream_snapshot=stream_state, + latest_candle_ts=latest_candle_ts, + timeframe=timeframe, + now=now, + gap_multiplier=self.gap_multiplier, + ) + self._record_runtime_incidents( + instrument_id=instrument_id, + timeframe=timeframe, + stream_state=stream_state, + data_gap_state=data_gap_state, + news_snapshot=news_snapshot, + observed_ts=now, + ) + return MarketTruthRuntimeState( + stream_state=stream_state, + data_gap_state=data_gap_state, + news_snapshot=news_snapshot, + ) + + def _run_loop(self) -> None: + while not self._stop_event.is_set(): + now = datetime.now(timezone.utc) + for subscription in self.stream_service.iter_subscriptions(): + self.current_state( + instrument_id=subscription.instrument_id, + timeframe=subscription.timeframe, + now=now, + ) + self._get_news_snapshot(now=now, force_refresh=False) + for _ in range(max(self.interval_seconds, 1)): + if self._stop_event.is_set(): + return + time.sleep(1) + + def _ensure_bootstrap_subscriptions(self) -> None: + symbols = [item.strip() for item in settings.market_data_stream_symbols.split(",") if item.strip()] + timeframes = [item.strip() for item in settings.market_data_stream_timeframes.split(",") if item.strip()] + if not symbols or not timeframes: + return + try: + instruments = self.repository.fetch_active_instruments_by_symbols(symbols) + except Exception: + instruments = [] + for instrument in instruments: + symbol = str(instrument.get("symbol") or "") + inst_id = build_okx_inst_id(symbol) + for timeframe in timeframes: + self.stream_service.subscribe( + instrument_id=int(instrument["id"]), + symbol=symbol, + inst_id=inst_id, + timeframe=timeframe, + ) + + def _should_fallback(self, *, stream_state: dict, data_gap_state: dict) -> bool: + return ( + stream_state.get("code") in {STREAM_CODE_DISCONNECTED, STREAM_CODE_STALE} + or data_gap_state.get("code") == DATA_GAP_CODE_DETECTED + ) + + def _run_rest_fallback(self, *, instrument_id: int, timeframe: str, now: datetime) -> None: + instrument = self.repository.fetch_instrument_by_id(instrument_id) + if not instrument: + return + symbol = str(instrument.get("symbol") or "") + if not symbol: + return + inst_id = build_okx_inst_id(symbol) + rows = self.rest_client.get_candles(inst_id=inst_id, bar=timeframe, limit=self.fallback_limit) + normalized = normalize_okx_candles(symbol=symbol, venue="OKX", timeframe=timeframe, candles=rows) + confirmed = [item for item in normalized if item.confirm] + written = self.candle_repository.upsert_many(confirmed, instrument_id=instrument_id) + self._record_incident( + instrument_id=instrument_id, + timeframe=timeframe, + incident_type=INCIDENT_TYPE_FALLBACK, + incident_code="rest_backfill_applied", + status=INCIDENT_STATUS_RESOLVED if written else INCIDENT_STATUS_OPEN, + observed_ts=now, + details={"symbol": symbol, "inst_id": inst_id, "timeframe": timeframe, "fetched": len(rows), "written": written}, + ) + + def _get_news_snapshot(self, *, now: datetime, force_refresh: bool = False) -> NewsProviderSnapshot: + if not settings.news_provider_enabled: + return self.news_provider.evaluate_cached_snapshot(snapshot=None, now=now) + refresh_interval = max(settings.news_provider_refresh_seconds, 1) + should_refresh = force_refresh or self._news_snapshot is None or self._last_news_refresh_ts is None + if not should_refresh and self._last_news_refresh_ts is not None: + should_refresh = (now - self._last_news_refresh_ts).total_seconds() >= refresh_interval + if should_refresh: + self._news_snapshot = self.news_provider.fetch_snapshot(now=now) + self._last_news_refresh_ts = now + return self.news_provider.evaluate_cached_snapshot(snapshot=self._news_snapshot, now=now) + + def _record_runtime_incidents( + self, + *, + instrument_id: int, + timeframe: str, + stream_state: dict, + data_gap_state: dict, + news_snapshot: NewsProviderSnapshot, + observed_ts: datetime, + ) -> None: + stream_code = str(stream_state.get("code") or "") + if stream_code in {STREAM_CODE_DISCONNECTED, STREAM_CODE_STALE}: + self._record_incident( + instrument_id=instrument_id, + timeframe=timeframe, + incident_type=INCIDENT_TYPE_STREAM, + incident_code=stream_code, + status=INCIDENT_STATUS_OPEN, + observed_ts=observed_ts, + details=stream_state, + ) + else: + self._record_incident( + instrument_id=instrument_id, + timeframe=timeframe, + incident_type=INCIDENT_TYPE_STREAM, + incident_code="market_stream_connected", + status=INCIDENT_STATUS_RESOLVED, + observed_ts=observed_ts, + details=stream_state, + ) + gap_code = str(data_gap_state.get("code") or "") + if gap_code == DATA_GAP_CODE_DETECTED: + self._record_incident( + instrument_id=instrument_id, + timeframe=timeframe, + incident_type=INCIDENT_TYPE_STREAM, + incident_code=gap_code, + status=INCIDENT_STATUS_OPEN, + observed_ts=observed_ts, + details=data_gap_state, + ) + else: + self._record_incident( + instrument_id=instrument_id, + timeframe=timeframe, + incident_type=INCIDENT_TYPE_STREAM, + incident_code=DATA_GAP_CODE_CLEAR, + status=INCIDENT_STATUS_RESOLVED, + observed_ts=observed_ts, + details=data_gap_state, + ) + if news_snapshot.code in {"news_provider_stale", "news_provider_error"}: + self._record_incident( + instrument_id=None, + timeframe=None, + incident_type=INCIDENT_TYPE_NEWS, + incident_code=news_snapshot.code, + status=INCIDENT_STATUS_OPEN, + observed_ts=observed_ts, + details=news_snapshot.to_dict(), + ) + else: + self._record_incident( + instrument_id=None, + timeframe=None, + incident_type=INCIDENT_TYPE_NEWS, + incident_code="news_provider_fresh", + status=INCIDENT_STATUS_RESOLVED, + observed_ts=observed_ts, + details=news_snapshot.to_dict(), + ) + + def _record_incident( + self, + *, + instrument_id: Optional[int], + timeframe: Optional[str], + incident_type: str, + incident_code: str, + status: str, + observed_ts: datetime, + details: dict, + ) -> None: + key = (instrument_id, timeframe, incident_code) + previous = self._last_incident_status.get(key) + if previous == status: + return + self._last_incident_status[key] = status + try: + self.repository.insert_market_truth_incident( + instrument_id=instrument_id, + timeframe=timeframe, + incident_type=incident_type, + incident_code=incident_code, + status=status, + observed_ts=observed_ts, + details=details, + ) + except Exception: + return + + +def build_okx_inst_id(symbol: str) -> str: + return f"{symbol}-SWAP" + + +def _ensure_aware(value) -> datetime: + if isinstance(value, datetime): + parsed = value + else: + parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00")) + if parsed.tzinfo is None: + return parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) diff --git a/src/services/market_state/news_event_provider.py b/src/services/market_state/news_event_provider.py new file mode 100644 index 0000000..f57e0e9 --- /dev/null +++ b/src/services/market_state/news_event_provider.py @@ -0,0 +1,205 @@ +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Optional + +import httpx + +from src.config.settings import settings + + +NEWS_CODE_ACTIVE = "news_window_active" +NEWS_CODE_INACTIVE = "news_window_inactive" +NEWS_CODE_PROVIDER_STALE = "news_provider_stale" +NEWS_CODE_PROVIDER_ERROR = "news_provider_error" + + +@dataclass(frozen=True) +class NewsEvent: + event_id: str + title: str + country: str + impact: str + event_ts: datetime + + def to_dict(self) -> dict: + return { + "event_id": self.event_id, + "title": self.title, + "country": self.country, + "impact": self.impact, + "event_ts": self.event_ts.isoformat(), + } + + +@dataclass(frozen=True) +class NewsProviderSnapshot: + provider_name: str + fetched_ts: Optional[datetime] + freshness_code: str + active_events: list[dict] + event_count: int + is_active: bool + code: str + error: Optional[str] + + def to_dict(self) -> dict: + return { + "provider_name": self.provider_name, + "fetched_ts": self.fetched_ts.isoformat() if self.fetched_ts else None, + "freshness_code": self.freshness_code, + "active_events": list(self.active_events), + "event_count": self.event_count, + "is_active": self.is_active, + "code": self.code, + "error": self.error, + } + + +class EconomicCalendarProvider: + provider_name = "forex_factory_public_feed" + + def __init__( + self, + *, + url: Optional[str] = None, + timeout: Optional[int] = None, + pre_window_minutes: Optional[int] = None, + post_window_minutes: Optional[int] = None, + max_age_seconds: Optional[int] = None, + high_impact_only: Optional[bool] = None, + ) -> None: + self.url = url or settings.news_provider_url + self.timeout = timeout or settings.news_provider_timeout + self.pre_window_minutes = pre_window_minutes if pre_window_minutes is not None else settings.news_event_pre_window_minutes + self.post_window_minutes = post_window_minutes if post_window_minutes is not None else settings.news_event_post_window_minutes + self.max_age_seconds = max_age_seconds if max_age_seconds is not None else settings.news_provider_max_age_seconds + self.high_impact_only = settings.news_event_high_impact_only if high_impact_only is None else high_impact_only + + def fetch_snapshot(self, *, now: Optional[datetime] = None, rows: Optional[list[dict]] = None) -> NewsProviderSnapshot: + now = _ensure_aware(now or datetime.now(timezone.utc)) + try: + source_rows = list(rows) if rows is not None else self._fetch_rows() + except Exception as exc: + return NewsProviderSnapshot( + provider_name=self.provider_name, + fetched_ts=None, + freshness_code=NEWS_CODE_PROVIDER_ERROR, + active_events=[], + event_count=0, + is_active=False, + code=NEWS_CODE_PROVIDER_ERROR, + error=str(exc), + ) + fetched_ts = now + events = self._parse_events(source_rows) + active_events = [event for event in events if self._event_is_active(event, now=now)] + freshness_code = NEWS_CODE_PROVIDER_STALE if self.max_age_seconds <= 0 else "news_provider_fresh" + return NewsProviderSnapshot( + provider_name=self.provider_name, + fetched_ts=fetched_ts, + freshness_code=freshness_code, + active_events=[event.to_dict() for event in active_events], + event_count=len(events), + is_active=bool(active_events), + code=NEWS_CODE_ACTIVE if active_events else NEWS_CODE_INACTIVE, + error=None, + ) + + def evaluate_cached_snapshot( + self, + *, + snapshot: Optional[NewsProviderSnapshot], + now: Optional[datetime] = None, + ) -> NewsProviderSnapshot: + now = _ensure_aware(now or datetime.now(timezone.utc)) + if snapshot is None: + return NewsProviderSnapshot( + provider_name=self.provider_name, + fetched_ts=None, + freshness_code=NEWS_CODE_PROVIDER_STALE, + active_events=[], + event_count=0, + is_active=False, + code=NEWS_CODE_PROVIDER_STALE, + error="snapshot_missing", + ) + fetched_ts = snapshot.fetched_ts + age_seconds = (now - fetched_ts).total_seconds() if fetched_ts else None + if age_seconds is None or age_seconds > self.max_age_seconds: + return NewsProviderSnapshot( + provider_name=snapshot.provider_name, + fetched_ts=fetched_ts, + freshness_code=NEWS_CODE_PROVIDER_STALE, + active_events=list(snapshot.active_events), + event_count=snapshot.event_count, + is_active=snapshot.is_active, + code=NEWS_CODE_PROVIDER_STALE, + error=snapshot.error, + ) + return snapshot + + def _fetch_rows(self) -> list[dict]: + with httpx.Client(timeout=self.timeout, follow_redirects=True) as client: + response = client.get(self.url, headers={"User-Agent": "ai-ict/market-truth"}) + response.raise_for_status() + payload = response.json() + if isinstance(payload, list): + return [dict(item) for item in payload if isinstance(item, dict)] + if isinstance(payload, dict): + items = payload.get("data") or payload.get("events") or [] + return [dict(item) for item in items if isinstance(item, dict)] + raise ValueError("unsupported_news_provider_payload") + + def _parse_events(self, rows: list[dict]) -> list[NewsEvent]: + events: list[NewsEvent] = [] + for index, row in enumerate(rows): + impact = str(row.get("impact") or row.get("importance") or row.get("impactTitle") or "").strip() + if self.high_impact_only and impact.lower() not in {"high", "high impact", "red"}: + continue + event_ts = _parse_event_datetime(row) + if event_ts is None: + continue + event_id = str(row.get("id") or row.get("eventId") or f"event:{index}") + title = str(row.get("title") or row.get("event") or row.get("name") or "macro_event") + country = str(row.get("country") or row.get("currency") or "unknown") + events.append( + NewsEvent( + event_id=event_id, + title=title, + country=country, + impact=impact or "unknown", + event_ts=event_ts, + ) + ) + return events + + def _event_is_active(self, event: NewsEvent, *, now: datetime) -> bool: + window_start = event.event_ts - timedelta(minutes=max(self.pre_window_minutes, 0)) + window_end = event.event_ts + timedelta(minutes=max(self.post_window_minutes, 0)) + return window_start <= now <= window_end + + +def _parse_event_datetime(row: dict) -> Optional[datetime]: + for key in ("timestamp", "dateUtc", "date", "datetime", "eventTs"): + value = row.get(key) + if value in (None, ""): + continue + try: + return _ensure_aware(value) + except ValueError: + continue + return None + + +def _ensure_aware(value) -> datetime: + if isinstance(value, datetime): + parsed = value + else: + text = str(value) + if text.isdigit(): + parsed = datetime.fromtimestamp(int(text) / 1000, tz=timezone.utc) + else: + parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) + if parsed.tzinfo is None: + return parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) diff --git a/src/services/market_state/runtime.py b/src/services/market_state/runtime.py new file mode 100644 index 0000000..4a1f60a --- /dev/null +++ b/src/services/market_state/runtime.py @@ -0,0 +1,19 @@ +from typing import Optional + +from src.services.market_state.market_truth_supervisor import MarketTruthSupervisor + + +_market_truth_supervisor: Optional[MarketTruthSupervisor] = None + + +def get_market_truth_supervisor() -> MarketTruthSupervisor: + global _market_truth_supervisor + if _market_truth_supervisor is None: + _market_truth_supervisor = MarketTruthSupervisor() + return _market_truth_supervisor + + +def start_market_truth_runtime() -> MarketTruthSupervisor: + supervisor = get_market_truth_supervisor() + supervisor.start() + return supervisor diff --git a/src/services/risk/risk_governor_service.py b/src/services/risk/risk_governor_service.py index 76f8217..16de581 100644 --- a/src/services/risk/risk_governor_service.py +++ b/src/services/risk/risk_governor_service.py @@ -11,6 +11,9 @@ RISK_CODE_MAX_OPEN_POSITIONS_REACHED = "max_open_positions_reached" RISK_CODE_SAME_SIDE_EXPOSURE_LIMIT_REACHED = "same_side_exposure_limit_reached" RISK_CODE_MARKET_DATA_MISSING = "market_data_missing" RISK_CODE_MARKET_DATA_STALE = "market_data_stale" +RISK_CODE_MARKET_STREAM_DISCONNECTED = "market_stream_disconnected" +RISK_CODE_MARKET_STREAM_STALE = "market_stream_stale" +RISK_CODE_MARKET_DATA_GAP_DETECTED = "market_data_gap_detected" RISK_CODE_NEWS_WINDOW_ACTIVE = "news_window_active" RISK_CODE_OPENING_VOLATILITY_ACTIVE = "opening_volatility_active" RISK_CODE_BACKTEST_GATE_FAILED_SEMI_AUTO = "backtest_quality_gate_failed_semi_auto" @@ -61,6 +64,16 @@ class RiskGovernorService: freshness_code = freshness.get("code") if freshness_code in {RISK_CODE_MARKET_DATA_MISSING, RISK_CODE_MARKET_DATA_STALE}: reject_codes.append(str(freshness_code)) + stream_state = dict(market_state.get("stream_state") or {}) + stream_code = str(stream_state.get("code") or "") + if stream_code in { + RISK_CODE_MARKET_STREAM_DISCONNECTED, + RISK_CODE_MARKET_STREAM_STALE, + }: + reject_codes.append(stream_code) + data_gap_state = dict(market_state.get("data_gap_state") or {}) + if data_gap_state.get("code") == RISK_CODE_MARKET_DATA_GAP_DETECTED: + reject_codes.append(RISK_CODE_MARKET_DATA_GAP_DETECTED) session = dict(market_state.get("session") or {}) if new_position and session and session.get("allows_new_positions") is False: reject_codes.append(RISK_CODE_SESSION_NOT_ALLOWED) diff --git a/tests/unit/test_documentation_consistency.py b/tests/unit/test_documentation_consistency.py index 2256a23..f70a927 100644 --- a/tests/unit/test_documentation_consistency.py +++ b/tests/unit/test_documentation_consistency.py @@ -30,6 +30,7 @@ class DocumentationConsistencyTests(unittest.TestCase): top_level_docs, { "AI_ICT_实盘交易分析系统开发计划_v1.md", + "AI_ICT_实盘辅助决策系统开发方案_v2.md", "ICT交易全书_出版终校版.md", "training_console_public_surface_v1.md", }, diff --git a/tests/unit/test_market_state_and_risk_governor.py b/tests/unit/test_market_state_and_risk_governor.py index 992d495..fe34111 100644 --- a/tests/unit/test_market_state_and_risk_governor.py +++ b/tests/unit/test_market_state_and_risk_governor.py @@ -8,6 +8,8 @@ from src.services.execution import evaluate_execution_decision from src.services.market_state import ( MarketStateService, build_session_state, + build_gap_state, + build_stream_state, evaluate_freshness, evaluate_news_window, evaluate_opening_volatility, @@ -98,6 +100,8 @@ class MarketStateServiceTests(unittest.TestCase): self.assertEqual(state.session.session_code, "LONDON") self.assertTrue(state.session.allows_new_positions) self.assertFalse(state.news_window.is_active) + self.assertEqual(state.stream_state.code, "market_stream_disconnected") + self.assertEqual(state.data_gap_state.code, "market_data_gap_unavailable") class RiskGovernorServiceTests(unittest.TestCase): @@ -141,6 +145,23 @@ class RiskGovernorServiceTests(unittest.TestCase): self.assertFalse(decision.allowed) self.assertIn("news_window_active", decision.reject_codes) + def test_risk_governor_rejects_stream_disconnect_and_gap(self) -> None: + decision = RiskGovernorService().evaluate( + context=RiskContext( + market_state={ + "freshness": {"code": "market_data_fresh"}, + "session": {"allows_new_positions": True}, + "stream_state": build_stream_state({"code": "market_stream_disconnected"}).to_dict(), + "data_gap_state": build_gap_state({"code": "market_data_gap_detected", "is_gap": True, "source": "stream"}).to_dict(), + } + ), + execution_mode="paper", + ) + + self.assertFalse(decision.allowed) + self.assertIn("market_stream_disconnected", decision.reject_codes) + self.assertIn("market_data_gap_detected", decision.reject_codes) + def test_risk_governor_rejects_low_quality_backtest_gate_for_semi_auto_only(self) -> None: context = RiskContext( backtest_quality_gate={ diff --git a/tests/unit/test_market_truth_runtime.py b/tests/unit/test_market_truth_runtime.py new file mode 100644 index 0000000..5f575df --- /dev/null +++ b/tests/unit/test_market_truth_runtime.py @@ -0,0 +1,322 @@ +from datetime import datetime, timedelta, timezone +import json +import os +import unittest + +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test") + +from src.services.market_data.okx_client import OKXCandle +from src.services.market_state import ( + DATA_GAP_CODE_DETECTED, + EconomicCalendarProvider, + MarketDataStreamService, + MarketStateService, + MarketTruthRuntimeState, + MarketTruthSupervisor, + STREAM_CODE_CONNECTED, + STREAM_CODE_DISCONNECTED, + build_data_gap_state, +) +from src.services.market_state.news_event_provider import NewsProviderSnapshot + + +class MarketDataStreamServiceTests(unittest.TestCase): + def test_stream_service_parses_ticker_and_candle_messages(self) -> None: + service = MarketDataStreamService(url="wss://example.test") + service.subscribe( + instrument_id=1, + symbol="BTC-USDT", + inst_id="BTC-USDT-SWAP", + timeframe="1m", + ) + + service._on_message(None, json.dumps({"event": "subscribe", "arg": {"channel": "tickers", "instId": "BTC-USDT-SWAP"}})) + service._on_message( + None, + json.dumps( + { + "arg": {"channel": "tickers", "instId": "BTC-USDT-SWAP"}, + "data": [{"last": "65000.5", "ts": "1778032800000"}], + } + ), + ) + service._on_message( + None, + json.dumps( + { + "arg": {"channel": "candle1m", "instId": "BTC-USDT-SWAP"}, + "data": [["1778032800000", "64990.1", "65010.0", "64980.0", "65000.5", "12.3", "0", "0", "1"]], + } + ), + ) + + snapshot = service.get_snapshot(instrument_id=1, timeframe="1m") + + self.assertEqual(snapshot["code"], STREAM_CODE_CONNECTED) + self.assertTrue(snapshot["connected"]) + self.assertTrue(snapshot["subscribed"]) + self.assertEqual(snapshot["latest_price"], 65000.5) + self.assertEqual(snapshot["latest_candle_open"], 64990.1) + self.assertEqual(snapshot["latest_candle_close"], 65000.5) + self.assertIsNotNone(snapshot["last_confirmed_candle_close_ts"]) + + def test_build_data_gap_state_detects_missing_candle_interval(self) -> None: + now = datetime(2026, 5, 7, 12, 5, tzinfo=timezone.utc) + gap_state = build_data_gap_state( + stream_snapshot={"code": STREAM_CODE_CONNECTED, "gap_detected": False}, + latest_candle_ts=now - timedelta(minutes=4), + timeframe="1m", + now=now, + gap_multiplier=2, + ) + + self.assertTrue(gap_state["is_gap"]) + self.assertEqual(gap_state["code"], DATA_GAP_CODE_DETECTED) + + +class EconomicCalendarProviderTests(unittest.TestCase): + def test_provider_marks_active_high_impact_event(self) -> None: + now = datetime(2026, 5, 7, 12, 0, tzinfo=timezone.utc) + provider = EconomicCalendarProvider( + pre_window_minutes=15, + post_window_minutes=15, + high_impact_only=True, + ) + + snapshot = provider.fetch_snapshot( + now=now, + rows=[ + { + "id": "cpi", + "title": "US CPI", + "country": "USD", + "impact": "High", + "dateUtc": now.isoformat(), + } + ], + ) + + self.assertTrue(snapshot.is_active) + self.assertEqual(snapshot.code, "news_window_active") + self.assertEqual(snapshot.event_count, 1) + + +class MarketTruthSupervisorTests(unittest.TestCase): + def test_supervisor_runs_rest_fallback_and_records_incident(self) -> None: + now = datetime(2026, 5, 7, 12, 0, tzinfo=timezone.utc) + repository = _SupervisorRepository(now=now) + candle_repository = _FallbackCandleRepository(repository) + supervisor = MarketTruthSupervisor( + repository=repository, + candle_repository=candle_repository, + stream_service=_DisconnectedStreamService(now=now), + rest_client=_FallbackOKXClient(now=now), + news_provider=_InactiveNewsProvider(now=now), + enabled=False, + ) + + state = supervisor.current_state(instrument_id=1, timeframe="1m", now=now) + + self.assertEqual(state.stream_state["code"], STREAM_CODE_DISCONNECTED) + self.assertEqual(candle_repository.write_calls, 1) + self.assertTrue(any(item["incident_code"] == "rest_backfill_applied" for item in repository.incidents)) + self.assertTrue(any(item["incident_code"] == STREAM_CODE_DISCONNECTED for item in repository.incidents)) + + def test_market_state_service_uses_supervisor_runtime_state(self) -> None: + now = datetime(2026, 5, 7, 12, 0, tzinfo=timezone.utc) + repository = _SupervisorRepository(now=now) + supervisor = _StaticSupervisor(now=now) + service = MarketStateService(repository=repository, market_truth_supervisor=supervisor) + + state = service.build_market_state(instrument_id=1, timeframe="1m", now=now) + + payload = state.to_dict() + self.assertIn("stream_state", payload) + self.assertIn("data_gap_state", payload) + self.assertIn("news_state", payload) + self.assertEqual(payload["stream_state"]["code"], STREAM_CODE_CONNECTED) + self.assertEqual(payload["latest_candle"]["latest_price"], 65111.0) + + +class _SupervisorRepository: + def __init__(self, now: datetime) -> None: + self.now = now + self.latest_candle = { + "id": 1, + "instrument_id": 1, + "timeframe": "1m", + "ts_close": now - timedelta(minutes=10), + "open": 64000.0, + "close": 64010.0, + } + self.incidents: list[dict] = [] + + def fetch_active_instruments_by_symbols(self, symbols: list[str]) -> list[dict]: + return [{"id": 1, "symbol": "BTC-USDT", "active": True}] if "BTC-USDT" in symbols else [] + + def fetch_instrument_by_id(self, instrument_id: int): + return {"id": instrument_id, "symbol": "BTC-USDT", "active": True} + + def fetch_latest_candle(self, instrument_id: int, timeframe: str): + return dict(self.latest_candle) + + def fetch_session_at(self, instrument_id: int, ts): + return { + "id": 5, + "session_code": "LONDON", + "status": "open", + "start_ts": self.now - timedelta(minutes=20), + "end_ts": self.now + timedelta(hours=2), + } + + def insert_market_truth_incident(self, **kwargs) -> int: + self.incidents.append(dict(kwargs)) + return len(self.incidents) + + +class _FallbackCandleRepository: + def __init__(self, repository: _SupervisorRepository) -> None: + self.repository = repository + self.write_calls = 0 + + def upsert_many(self, candles, instrument_id: int) -> int: + self.write_calls += 1 + latest = candles[-1] + self.repository.latest_candle = { + "id": 99, + "instrument_id": instrument_id, + "timeframe": latest.timeframe, + "ts_close": datetime.fromtimestamp((latest.ts_open_ms + 60_000 - 1) / 1000, tz=timezone.utc), + "open": latest.open, + "close": latest.close, + } + return len(candles) + + +class _DisconnectedStreamService: + def __init__(self, now: datetime) -> None: + self.now = now + self.subscriptions = [] + + def subscribe(self, **kwargs) -> None: + self.subscriptions.append(dict(kwargs)) + + def start(self) -> None: + return + + def stop(self) -> None: + return + + def iter_subscriptions(self): + return [] + + def get_snapshot(self, *, instrument_id: int, timeframe: str, now=None) -> dict: + return { + "instrument_id": instrument_id, + "timeframe": timeframe, + "code": STREAM_CODE_DISCONNECTED, + "connected": False, + "subscribed": False, + "reconnect_count": 1, + "last_message_ts": None, + "heartbeat_age_seconds": None, + "lag_seconds": None, + "latest_price": None, + "latest_candle_open": None, + "latest_candle_close": None, + "latest_candle_open_ts": None, + "latest_candle_close_ts": None, + "last_confirmed_candle_close_ts": None, + "last_disconnect_reason": "socket_closed", + "last_error": "socket_closed", + "gap_detected": False, + "gap_seconds": None, + } + + +class _FallbackOKXClient: + def __init__(self, now: datetime) -> None: + self.now = now + + def get_candles(self, inst_id: str, bar: str, limit: int): + return [ + OKXCandle( + inst_id=inst_id, + bar=bar, + ts=str(int((self.now - timedelta(minutes=1)).timestamp() * 1000)), + open="65000.0", + high="65120.0", + low="64980.0", + close="65111.0", + volume="5.2", + confirm="1", + ) + ] + + +class _InactiveNewsProvider: + def __init__(self, now: datetime) -> None: + self.now = now + + def fetch_snapshot(self, *, now: datetime): + return NewsProviderSnapshot( + provider_name="test", + fetched_ts=now, + freshness_code="news_provider_fresh", + active_events=[], + event_count=0, + is_active=False, + code="news_window_inactive", + error=None, + ) + + def evaluate_cached_snapshot(self, *, snapshot, now: datetime): + return snapshot + + +class _StaticSupervisor: + def __init__(self, now: datetime) -> None: + self.now = now + + def current_state(self, *, instrument_id: int, timeframe: str, now: datetime): + return MarketTruthRuntimeState( + stream_state={ + "code": STREAM_CODE_CONNECTED, + "connected": True, + "subscribed": True, + "reconnect_count": 0, + "last_message_ts": now.isoformat(), + "heartbeat_age_seconds": 1.0, + "lag_seconds": 1.0, + "latest_price": 65111.0, + "latest_candle_open": 65000.0, + "latest_candle_close": 65111.0, + "latest_candle_open_ts": (now - timedelta(minutes=1)).isoformat(), + "latest_candle_close_ts": now.isoformat(), + "last_confirmed_candle_close_ts": now.isoformat(), + "last_disconnect_reason": None, + "last_error": None, + }, + data_gap_state={ + "is_gap": False, + "code": "market_data_gap_clear", + "expected_interval_seconds": 60, + "gap_seconds": 0.0, + "last_candle_ts": now.isoformat(), + "source": "stream", + }, + news_snapshot=NewsProviderSnapshot( + provider_name="test", + fetched_ts=now, + freshness_code="news_provider_fresh", + active_events=[], + event_count=0, + is_active=False, + code="news_window_inactive", + error=None, + ), + ) + + +if __name__ == "__main__": + unittest.main()