Add real-time market truth layer
This commit is contained in:
Codex
@@ -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
|
||||
|
||||
+2
-1
@@ -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
|
||||
|
||||
@@ -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 叙事能力
|
||||
- 真实自动下单
|
||||
|
||||
@@ -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);
|
||||
@@ -9,6 +9,7 @@ dependencies = [
|
||||
"pydantic>=2.7",
|
||||
"pydantic-settings>=2.2",
|
||||
"httpx>=0.27",
|
||||
"websocket-client>=1.8",
|
||||
]
|
||||
|
||||
[build-system]
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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 []),
|
||||
}
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
+57
-11
@@ -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"}]
|
||||
|
||||
@@ -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"
|
||||
|
||||
+3
-1
@@ -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(),
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
@@ -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]
|
||||
|
||||
@@ -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",
|
||||
]
|
||||
|
||||
@@ -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)
|
||||
@@ -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}")
|
||||
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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)
|
||||
|
||||
@@ -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",
|
||||
},
|
||||
|
||||
@@ -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={
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user