feat(replay): M9-6 Bar Replay 线性重放(as_of 逐日仅用当时数据)

- domain/entities/replay.py:ReplayDay{top/events/counts}/ReplayResult 时间线
- ReplayService:universe.symbols 白名单必填(≤40)且重放交易日 ≤90(防全市场长任务);
  每个交易日以 <=as_of 数据经同一 signal/score 引擎生成帧
- POST /api/replays(边界校验)→ 时间线;供前端 Bar Replay 控件(v3 §20.6 阶段二)
- tests/test_replays.py:重放帧 == 回测 selection_history 逐调仓日一致;范围约束;
  后段暴涨股不泄漏进早段帧(未来函数);API 400/200;全量 pytest 通过
This commit is contained in:
Simon
2026-09-09 07:18:36 +08:00
parent 5bde8f9f5f
commit 37510c1b89
6 changed files with 327 additions and 0 deletions
+34
View File
@@ -0,0 +1,34 @@
"""Bar Replay(v3 §20.6,第二阶段 MVP)领域实体。
线性重放:对给定选股查询与信号规则,在交易日序列上逐日以「当日为止的数据」执行
(as_of 语义),输出每日时间线(意图 Top + 信号 + 计数),用于核对:
- 未来函数:每日计算只用 <= as_of 数据(与静态研究同一引擎)
- 一致性:重放某日结果 == 该日独立 select/signal 结果(亦 == 回测该调仓日意图)
"""
from __future__ import annotations
from datetime import date
from pydantic import BaseModel, Field
from app.domain.entities.signal import SignalEvent
class ReplayTop(BaseModel):
symbol: str
score: float
class ReplayDay(BaseModel):
as_of: date
top: list[ReplayTop] = Field(default_factory=list, description="意图排名前 N(score 降序)")
events: list[SignalEvent] = Field(default_factory=list, description="当日信号(<=max_output_rank)")
counts: dict[str, int] = Field(default_factory=dict, description="BUY/WATCH/SELL 计数")
class ReplayResult(BaseModel):
start: date
end: date
days: list[ReplayDay] = Field(default_factory=list)
top_n: int = 5