Simon
23972e7063
汇总三轮未提交的开发（每轮均在本机 MariaDB + 真实浏览器上验证）：
1) 股息率案例（全市场股息率最高 n 只，默认 20，每 m 月择股）
- 新增日频估值表 daily_basic + 迁移；股息率因子（dv_ratio / dividend_yield / TTM）
- 名称历史表 stock_name_history：剔除 ST 按**择股日当时名称**判定，消除
「曾高股息后 ST」的股息陷阱（实测 3.70pp 偏差）
- 区间择股/调仓双周期（m 择股 / y 调仓）、指数成分与白名单、停牌近似剔除
- 复权因子口径核对（4,164,742 行、缺失 0.0%）、收盘价成交与涨跌停拦单
- 案例实测：2020-01-01~2026-09-04 总收益 +24.86%（年化 3.52%、回撤 -28.58%）
2) 策略库与前端统一
- strategy 表 + CRUD/PUT 原地更新 + `describe_strategy` 按 spec 真实推导
「一句话说明 + 计算公式 + 执行步骤 + 注意事项」（与引擎实执行规则同源）
- 任何出现股票代码处都成对显示名称且可点击进个股页
- 全站图表基座统一 TradingView Lightweight Charts（ECharts 依赖、
锁文件、组件与文档标注一并清除），买卖点标记只落在真实交易日上
3) 回测存档完整化（可往复查看）
- 同步端点（POST /api/backtests、/api/factor-tests）此前完全不落库 → 现在同样归档，
归档 id 经响应头 X-Experiment-Id 返回（不破坏 response_model）
- data_version 首次真实写入（数据快照指纹：最新交易日 + 各表规模）
- 个股收益曲线默认**全量保存**（此前硬截断 60 只）；超出体积预算才裁剪，
并写 archive_meta（机器可读）+ unimplemented（人可读）如实标注
- 列表 kind/q 过滤 + X-Total-Count（此前 limit=50 静默截断）、DELETE 归档
- 只读归档页 /experiments/{id}（Server Component，SSR 直出**选股条件**与
**交易执行依据**）；结果视图按 kind 分发（backtest/factor_test/selection），
非回测归档不套用回测口径
- 新增 CLI：prune_experiments（保留策略，默认 dry-run）、
restore_experiment_from_job（从 Job 副本按原 id 重建被删的历史归档，默认 dry-run）
门禁：pytest 388 passed、ruff All checks passed、tsc 0 错误、图表单测 7 passed、
next build 成功、契约脚本 verify_strategy_workspace 59/59（含按 kind 逐类验证归档页）。
565 lines
24 KiB
Python
565 lines
24 KiB
Python
"""研究服务：把 Research Specification 编排为数据获取 + 引擎执行。
|
||
|
||
本层是业务入口：API / Agent 只能调用这里的用例（AGENT.md §16/§17），
|
||
禁止直接拼接引擎配置。数据一律经 Repository 获取（防未来函数由查询层保证）。
|
||
|
||
内存优化：大数据面板优先走 Repository 的流式列裁剪查询
|
||
（stream_range_many_columns，SQL 侧转 REAL、分批拉取），避免 ORM 对象 /
|
||
Decimal 全量物化；老实现回退到 get_range_many 逐实体路径。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from collections.abc import Iterable
|
||
from datetime import date, timedelta
|
||
from typing import Any
|
||
|
||
import pandas as pd
|
||
|
||
from app.domain.entities.market import (
|
||
DAILY_BAR_NUMERIC_FIELDS,
|
||
DAILY_BASIC_NUMERIC_FIELDS,
|
||
FinancialIndicator,
|
||
)
|
||
from app.domain.entities.research import (
|
||
BacktestResult,
|
||
FactorCorrelationReport,
|
||
FactorTestReport,
|
||
ResearchSpec,
|
||
)
|
||
from app.domain.repositories.market import (
|
||
DailyBarRepository,
|
||
StockRepository,
|
||
)
|
||
from app.quant.composite import build_factor_panels
|
||
from app.quant.engine import QuantEngine
|
||
from app.quant.evaluation import factor_correlation_report
|
||
from app.quant.selection import build_condition_fields, eligible_symbols
|
||
from app.quant.universe import ( # noqa: F401 —— 范围过滤
|
||
filter_stocks,
|
||
names_as_of,
|
||
resolve_members,
|
||
)
|
||
|
||
# 流式路径每攒多少行落一个 DataFrame 分片（控制 concat 峰值）
|
||
_FRAME_CHUNK_ROWS = 50_000
|
||
|
||
|
||
def bars_to_daily_df(bars) -> pd.DataFrame:
|
||
"""DailyBar 列表 → 引擎长表 DataFrame（symbol/trade_date/ohlc/volume/amount）。
|
||
|
||
领域实体中的 Decimal 在此转 float，供 pandas 数值运算（保持 DataFrame 全数值列）。
|
||
"""
|
||
df = pd.DataFrame([b.model_dump() for b in bars])
|
||
if not df.empty:
|
||
for col in ("open", "high", "low", "close", "volume", "amount"):
|
||
if col in df.columns:
|
||
df[col] = df[col].astype(float)
|
||
return df
|
||
|
||
|
||
def _frame_from_stream(rows: Iterable[tuple], columns: list[str]) -> pd.DataFrame:
|
||
"""把流式 (symbol, trade_date_iso, *float_cols) 分批拼成 float 长表。
|
||
|
||
全程只保留分批 DataFrame + 最终一份结果，避免整批 tuple/Decimal 同时驻留。
|
||
返回列：symbol, trade_date(datetime64), *columns(float64)。
|
||
"""
|
||
cols = ["symbol", "trade_date", *columns]
|
||
buf: list[tuple] = []
|
||
pieces: list[pd.DataFrame] = []
|
||
for row in rows:
|
||
buf.append(row)
|
||
if len(buf) >= _FRAME_CHUNK_ROWS:
|
||
pieces.append(pd.DataFrame(buf, columns=cols))
|
||
buf = []
|
||
if buf:
|
||
pieces.append(pd.DataFrame(buf, columns=cols))
|
||
if not pieces:
|
||
return pd.DataFrame()
|
||
df = pd.concat(pieces, ignore_index=True)
|
||
df["trade_date"] = pd.to_datetime(df["trade_date"])
|
||
for col in columns: # NULL → NaN，统一 float64
|
||
df[col] = pd.to_numeric(df[col], errors="coerce")
|
||
return df
|
||
|
||
|
||
def load_daily_df(
|
||
daily_repo,
|
||
symbols: list[str],
|
||
start: date,
|
||
end: date,
|
||
columns: list[str],
|
||
adjust: str = "none",
|
||
price_adjust: str = "none",
|
||
) -> pd.DataFrame:
|
||
"""从 Repository 装配行情长表（供研究/选股共用）。
|
||
|
||
两个 adjust 语义不同（v3 §20.5）：
|
||
- `adjust`：行集过滤（stock_daily.adjust 主口径 none / 新浪兜底 qfq 行）
|
||
- `price_adjust`：复权折算（none/qfq/hfq，按 adjust_factor 在 SQL 侧折算价格列）
|
||
|
||
优先走流式列裁剪（stream_range_many_columns，SQL 侧转 REAL、分批），
|
||
失败或实现缺失时回退 get_range_many / 逐只 get_range（回退路径不支持 price_adjust，
|
||
此时由调用方保证 price_adjust=none，避免静默混入未折算价格）。
|
||
"""
|
||
if not symbols:
|
||
return pd.DataFrame()
|
||
streamer = getattr(daily_repo, "stream_range_many_columns", None)
|
||
if streamer is not None:
|
||
try:
|
||
return _frame_from_stream(
|
||
streamer(
|
||
symbols, start, end, sorted(columns),
|
||
adjust=adjust, price_adjust=price_adjust,
|
||
),
|
||
sorted(columns),
|
||
)
|
||
except TypeError: # 实现未升级（无 price_adjust 形参）→ 回退
|
||
if price_adjust != "none":
|
||
raise
|
||
return _frame_from_stream(
|
||
streamer(symbols, start, end, sorted(columns), adjust=adjust), sorted(columns)
|
||
)
|
||
except Exception: # noqa: BLE001 —— 流式路径失败回退旧路径（兼容非 SQL 实现）
|
||
if price_adjust != "none":
|
||
raise # 复权路径失败必须显式报错，禁止静默退回不复权
|
||
pass
|
||
elif price_adjust != "none":
|
||
raise ValueError(
|
||
f"行情仓储不支持流式列裁剪，无法按 {price_adjust} 复权装配（拒绝静默退回不复权）"
|
||
)
|
||
get_many = getattr(daily_repo, "get_range_many", None)
|
||
if get_many is not None:
|
||
bars = list(get_many(symbols, start, end, adjust=adjust))
|
||
else: # 兜底：逐只查询
|
||
bars = []
|
||
for sym in symbols:
|
||
bars.extend(daily_repo.get_range(sym, start, end))
|
||
return bars_to_daily_df(bars)
|
||
|
||
|
||
def split_factor_columns(columns: Iterable[str]) -> tuple[set[str], set[str]]:
|
||
"""把因子所需列拆成 (行情列, 每日指标列)。
|
||
|
||
行情列来自 stock_daily（含 close 等）；每日指标列来自 daily_basic（dv_ratio 等）。
|
||
未识别的列按行情列处理（由仓储的列白名单兜底报错，错误信息更贴近调用点）。
|
||
"""
|
||
bars: set[str] = set()
|
||
basics: set[str] = set()
|
||
for col in columns:
|
||
if col in DAILY_BASIC_NUMERIC_FIELDS and col not in DAILY_BAR_NUMERIC_FIELDS:
|
||
basics.add(col)
|
||
else:
|
||
bars.add(col)
|
||
return bars, basics
|
||
|
||
|
||
def load_basic_df(
|
||
basic_repo,
|
||
symbols: list[str],
|
||
start: date,
|
||
end: date,
|
||
columns: list[str],
|
||
) -> pd.DataFrame:
|
||
"""装配每日指标长表（daily_basic）—— 仅取所需列。
|
||
|
||
与 load_daily_df 同构（流式优先、回退批量/逐只）。
|
||
"""
|
||
if not symbols or not columns:
|
||
return pd.DataFrame()
|
||
streamer = getattr(basic_repo, "stream_range_many_columns", None)
|
||
if streamer is not None:
|
||
try:
|
||
return _frame_from_stream(
|
||
streamer(symbols, start, end, sorted(columns)), sorted(columns)
|
||
)
|
||
except Exception: # noqa: BLE001 —— 回退旧路径
|
||
pass
|
||
get_many = getattr(basic_repo, "get_range_many", None)
|
||
if get_many is not None:
|
||
rows = [
|
||
{"symbol": r.symbol, "trade_date": r.trade_date, **{c: getattr(r, c) for c in columns}}
|
||
for r in get_many(symbols, start, end)
|
||
]
|
||
else: # 兜底：逐只查询
|
||
rows = []
|
||
for sym in symbols:
|
||
for r in basic_repo.get_range(sym, start, end):
|
||
rows.append(
|
||
{
|
||
"symbol": r.symbol,
|
||
"trade_date": r.trade_date,
|
||
**{c: getattr(r, c) for c in columns},
|
||
}
|
||
)
|
||
df = pd.DataFrame(rows)
|
||
if df.empty:
|
||
return df
|
||
df["trade_date"] = pd.to_datetime(df["trade_date"])
|
||
for col in columns:
|
||
df[col] = pd.to_numeric(df[col], errors="coerce")
|
||
return df
|
||
|
||
|
||
def merge_basic_into_daily(daily: pd.DataFrame, basic: pd.DataFrame) -> pd.DataFrame:
|
||
"""把每日指标列并入行情长表（按 symbol + trade_date 左连接）。
|
||
|
||
命名冲突保护：daily_basic 也有 close 列，若与行情 close 冲突则**丢弃指标侧**列
|
||
（成交/估值口径以 stock_daily 不复权 close 为准，避免静默改口径，AGENT.md §8）。
|
||
"""
|
||
if basic is None or basic.empty:
|
||
return daily
|
||
if daily is None or daily.empty:
|
||
return daily
|
||
add_cols = [
|
||
c for c in basic.columns if c not in ("symbol", "trade_date") and c not in daily.columns
|
||
]
|
||
if not add_cols:
|
||
return daily
|
||
left = daily.copy()
|
||
left["trade_date"] = pd.to_datetime(left["trade_date"])
|
||
right = basic[["symbol", "trade_date", *add_cols]].copy()
|
||
right["trade_date"] = pd.to_datetime(right["trade_date"])
|
||
merged = left.merge(right, on=["symbol", "trade_date"], how="left")
|
||
return merged
|
||
|
||
|
||
def _fill_names(result: BacktestResult, stocks) -> BacktestResult:
|
||
"""把股票池的 `symbol → name` 回填进回测结果的各展示结构（就地修改并返回）。
|
||
|
||
为什么放在服务层而不是引擎里：名称只是展示增强，两个引擎（LocalEngine / QlibEngine）
|
||
都经本方法返回，改一处即同时覆盖；引擎保持「只认 symbol」的纯计算职责，不引入
|
||
名称查询/IO，也就不必在引擎内部为零散字段各写一次映射。
|
||
|
||
`stocks` 为空（universe 过滤后无标的、或调用方未装配）时**静默跳过**，不抛错：
|
||
名称缺失只影响展示，不应让一个已经算完的回测失败（AGENT.md §24 的诚实标注
|
||
由引擎的 unimplemented 承担，不由名称缺失承担）。
|
||
"""
|
||
if not stocks:
|
||
return result
|
||
name_map = {s.symbol: s.name for s in stocks if getattr(s, "name", None)}
|
||
if not name_map:
|
||
return result
|
||
|
||
def _fill(rows) -> None:
|
||
for row in rows:
|
||
if row.name is None:
|
||
row.name = name_map.get(row.symbol)
|
||
|
||
_fill(result.symbol_curves)
|
||
# marks 与 signal_history 在 LocalEngine 中共享同一批 ActionRecord 对象，
|
||
# 这里仍显式回填一次：对独立实现（如 Qlib 适配器）也成立，幂等无副作用。
|
||
for curve in result.symbol_curves:
|
||
_fill(curve.marks)
|
||
_fill(result.positions)
|
||
_fill(result.trades)
|
||
_fill(result.signal_history)
|
||
_fill(result.fills)
|
||
_fill(result.selection_history)
|
||
return result
|
||
|
||
|
||
class ResearchService:
|
||
"""研究用例入口（因子测试 / 回测）。依赖注入 Repository 与引擎。"""
|
||
|
||
def __init__(
|
||
self,
|
||
stock_repo: StockRepository,
|
||
daily_repo: DailyBarRepository,
|
||
engine: QuantEngine,
|
||
index_repo=None,
|
||
basic_repo=None,
|
||
financial_repo=None,
|
||
name_repo=None,
|
||
) -> None:
|
||
self._stock_repo = stock_repo
|
||
self._daily_repo = daily_repo
|
||
self._engine = engine
|
||
self._index_repo = index_repo
|
||
# 名称变更历史仓储：universe.exclude_st 的**时点**口径依据。
|
||
# 未注入 → 回退 stock.name 最新快照（旧行为，结果页会如实标注残余偏差）。
|
||
self._name_repo = name_repo
|
||
# 每日指标仓储（daily_basic）：仅当 spec 因子引用 dv_ratio 等列时使用；
|
||
# 未注入而在 spec 中引用 → 明确报错，不做静默降级（否则因子会全是 NaN）。
|
||
self._basic_repo = basic_repo
|
||
# 注：复权（qfq/hfq）折算在行情仓储 `stream_range_many_columns` 的 SQL 内完成
|
||
# （与成交价同源，v3 §20.5），业务层不需要 adjust_factor 仓储；缺口统计走
|
||
# `self._daily_repo.count_price_adjust_gaps`。
|
||
# 财务仓储：spec.conditions 引用 fundamental.* 时使用
|
||
self._financial_repo = financial_repo
|
||
# 最近一次装配的复权因子缺口报告（未复权时为 None），用于结果如实标注
|
||
self.last_adjust_gaps: dict | None = None
|
||
# 最近一次装配的 exclude_st 名称口径（None = 未启用 ST 过滤）
|
||
self.last_name_basis: dict | None = None
|
||
# 最近一次装配的 universe 股票列表（条件求值需要 static.* 元数据）
|
||
self._last_stocks: list = []
|
||
|
||
def run_factor_test(
|
||
self, spec: ResearchSpec, horizon_days: int = 21, on_stage=None
|
||
) -> FactorTestReport:
|
||
"""on_stage(str)：执行阶段回调（data_loading / factor_calculation / analysis），
|
||
供 Job 状态机上报 stage（v3 §23）。"""
|
||
if spec.type != "factor_test":
|
||
raise ValueError("factor_test 用例需要 spec.type=factor_test")
|
||
_stage(on_stage, "data_loading")
|
||
daily = self._load_daily(spec)
|
||
_stage(on_stage, "factor_calculation")
|
||
report = self._engine.run_factor_test(daily, spec, horizon_days=horizon_days)
|
||
_stage(on_stage, "analysis")
|
||
return report
|
||
|
||
def run_backtest(self, spec: ResearchSpec, on_stage=None) -> BacktestResult:
|
||
if spec.type != "backtest":
|
||
raise ValueError("backtest 用例需要 spec.type=backtest")
|
||
_stage(on_stage, "data_loading")
|
||
daily = self._load_daily(spec)
|
||
_stage(on_stage, "backtesting")
|
||
eligibility = self._build_eligibility(spec, daily)
|
||
result = self._engine.run_backtest(daily, spec, eligibility_fn=eligibility)
|
||
_stage(on_stage, "analysis")
|
||
self._annotate_price_basis(result, spec)
|
||
# 两个引擎都在此汇合：名称回填只做一次（`_last_stocks` 在 _load_daily 中装配）
|
||
return _fill_names(result, self._last_stocks)
|
||
|
||
def run_factor_correlation(self, spec: ResearchSpec) -> FactorCorrelationReport:
|
||
"""多因子两两相关（v3 §12）：同 universe/period 装配 → 横截面相关矩阵。"""
|
||
daily = self._load_daily(spec)
|
||
panels = {fs.name: build_factor_panels(daily, [fs])[0][1] for fs in spec.factors}
|
||
return factor_correlation_report(panels)
|
||
|
||
# ---- 数据装配 ----
|
||
|
||
def _load_daily(self, spec: ResearchSpec) -> pd.DataFrame:
|
||
start, end = spec.period
|
||
# 回测前预留因子 warmup（lookback≤120 交易日，取 300 自然日余量）
|
||
data_start = start - timedelta(days=300)
|
||
all_stocks = self._stock_repo.list()
|
||
name_at, applied = names_as_of(all_stocks, start, self._name_repo)
|
||
stocks = filter_stocks(
|
||
all_stocks, spec.universe, as_of=start,
|
||
members=resolve_members(self._index_repo, spec.universe, start),
|
||
name_at=name_at,
|
||
)
|
||
# 供结果如实标注：exclude_st 是否用了时点名称、覆盖了多少只
|
||
self.last_name_basis = (
|
||
{"point_in_time": applied[0], "covered": applied[1], "total": len(all_stocks)}
|
||
if spec.universe.exclude_st
|
||
else None
|
||
)
|
||
self._last_stocks = stocks
|
||
# 引擎所需列裁剪（LocalEngine 只取 close + 因子字段；Qlib 回测取全 OHLCV）
|
||
required = self._engine.required_columns(spec)
|
||
bar_cols, basic_cols = split_factor_columns(required)
|
||
symbols = [s.symbol for s in stocks]
|
||
# 行集口径恒为 none（主口径）；复权折算由 price_adjust 承担（v3 §20.5 严格分离）
|
||
daily = load_daily_df(
|
||
self._daily_repo,
|
||
symbols,
|
||
data_start,
|
||
end,
|
||
sorted(bar_cols),
|
||
adjust="none",
|
||
price_adjust=spec.price_adjustment,
|
||
)
|
||
if spec.price_adjustment != "none":
|
||
self.last_adjust_gaps = self._adjust_gap_report(symbols, data_start, end)
|
||
if basic_cols:
|
||
daily = self._attach_basic(daily, symbols, data_start, end, sorted(basic_cols))
|
||
return daily
|
||
|
||
def _adjust_gap_report(self, symbols: list[str], start: date, end: date) -> dict:
|
||
"""复权因子缺口统计（如实上报「按 1.0 兜底未折算」的行占比）。"""
|
||
counter = getattr(self._daily_repo, "count_price_adjust_gaps", None)
|
||
if counter is None:
|
||
return {"checked": False, "reason": "仓储未实现 count_price_adjust_gaps"}
|
||
total, missing = counter(symbols, start, end)
|
||
return {
|
||
"checked": True,
|
||
"rows": total,
|
||
"rows_without_factor": missing,
|
||
"missing_pct": round(missing / total * 100, 4) if total else 0.0,
|
||
}
|
||
|
||
def _build_eligibility(self, spec: ResearchSpec, daily: pd.DataFrame):
|
||
"""按 spec.conditions 构造「择股日 → 合格股票集合」的求值闭包（可选）。
|
||
|
||
与选股路径共用 `build_condition_fields` / `eligible_symbols`，因此
|
||
「历史某日 /api/selections 的候选」与「回测在该日的候选池」由同一实现产出
|
||
（v2 §25 / v3 §28 一致性要求）。逐择股日结果缓存，避免重复计算。
|
||
|
||
未来函数红线：条件字段只取 <= as_of 的截面；fundamental.* 只取
|
||
announce_date <= as_of 的已公告值（由仓储 list_announced_many 保证）。
|
||
"""
|
||
if not self._last_stocks:
|
||
if not spec.conditions:
|
||
return None
|
||
raise ValueError("universe 过滤结果为空，无法构造选股条件求值器")
|
||
statics = {s.symbol: s.model_dump() for s in self._last_stocks}
|
||
candidates = sorted(statics)
|
||
# 时点 ST 过滤：exclude_st 必须在**每个择股日**按当时名称重判，否则
|
||
# 「2020 年入池、2022 年才变 ST」的标的会一直被当作合格候选（与
|
||
# /api/selections 的单时点语义不一致，v2 §25）。名称历史缺失时降级为
|
||
# 池子基准日口径（旧行为），由结果 name_basis 如实标注。
|
||
st_fn = self._build_st_filter(spec, candidates)
|
||
if not spec.conditions:
|
||
if st_fn is None:
|
||
return None
|
||
# 只有 ST 约束时，eligible 必须是「候选 − 该日 ST」，不能直接返回 st_fn
|
||
# （st_fn 返回的是**不合格**集合，直接返回会把语义取反）
|
||
allowed_cache: dict[date, set[str]] = {}
|
||
|
||
def _st_only(as_of: date) -> set[str]:
|
||
if as_of not in allowed_cache:
|
||
allowed_cache[as_of] = set(candidates) - st_fn(as_of)
|
||
return allowed_cache[as_of]
|
||
|
||
return _st_only
|
||
uses_fundamental = any(
|
||
f.startswith("fundamental.")
|
||
for c in spec.conditions
|
||
for f in (c.field, c.ref or "")
|
||
)
|
||
cache: dict[date, set[str]] = {}
|
||
|
||
def _fn(as_of: date) -> set[str]:
|
||
if as_of in cache:
|
||
return cache[as_of]
|
||
financial = (
|
||
self._load_financial(candidates, as_of) if uses_fundamental else {}
|
||
)
|
||
fields = build_condition_fields(daily, spec.conditions, pd.Timestamp(as_of))
|
||
if not fields:
|
||
cache[as_of] = set()
|
||
return cache[as_of]
|
||
passed = set(
|
||
eligible_symbols(candidates, spec.conditions, statics, fields, financial)
|
||
)
|
||
if st_fn is not None:
|
||
passed -= st_fn(as_of) # 该日名称含 ST → 不合格
|
||
cache[as_of] = passed
|
||
return cache[as_of]
|
||
|
||
return _fn
|
||
|
||
def _build_st_filter(self, spec: ResearchSpec, candidates: list[str]):
|
||
"""「择股日 → 该日名称含 ST 的股票集合」；不可用（未启用/未注入）时返回 None。
|
||
|
||
与 `filter_stocks` 共用 `names_as_of`，因此回测在任一择股日的
|
||
`exclude_st` 结果与该日 `/api/selections` 的候选池同口径（v2 §25）。
|
||
逐日缓存；每次查询只取该日生效的名称区间。
|
||
"""
|
||
if not spec.universe.exclude_st or self._name_repo is None:
|
||
return None
|
||
cache: dict[date, set[str]] = {}
|
||
|
||
def _fn(as_of: date) -> set[str]:
|
||
if as_of not in cache:
|
||
name_at, applied = names_as_of(self._last_stocks, as_of, self._name_repo)
|
||
if not applied[0]:
|
||
# 无时点名称可用（表为空/查询失败）：整段时间都退回池子基准日口径，
|
||
# 不逐日过滤 —— 与 filter_stocks 的回退一致，且 name_basis 会标注 false
|
||
cache[as_of] = set()
|
||
else:
|
||
# 逐股优先时点名称、缺失回退最新名称快照 —— 必须与
|
||
# filter_stocks 的 `(name_at or {}).get(sym) or s.name` 完全同口径，
|
||
# 否则同一择股日「回测入选」与「/api/selections 排除」会打架（v2 §25）
|
||
st_syms: set[str] = set()
|
||
for st in self._last_stocks:
|
||
nm = (name_at or {}).get(st.symbol) or st.name
|
||
if nm and "ST" in nm.upper():
|
||
st_syms.add(st.symbol)
|
||
cache[as_of] = st_syms
|
||
return cache[as_of]
|
||
|
||
return _fn
|
||
|
||
def _load_financial(self, symbols: list[str], as_of: date) -> dict[str, Any]:
|
||
"""按 announce_date <= as_of 批量取财务，每 symbol 保留最新一版。"""
|
||
if self._financial_repo is None:
|
||
raise ValueError(
|
||
"回测条件引用了 fundamental.* 字段，但未注入 FinancialRepository。"
|
||
"请检查 API/CLI 的依赖装配。"
|
||
)
|
||
getter = getattr(self._financial_repo, "list_announced_many", None)
|
||
if getter is not None:
|
||
rows = list(getter(symbols, as_of))
|
||
else: # 兜底：逐只取最新已公告
|
||
rows = []
|
||
for sym in symbols:
|
||
latest = self._financial_repo.latest_announced(sym, as_of)
|
||
if latest is not None:
|
||
rows.append(latest)
|
||
out: dict[str, FinancialIndicator] = {}
|
||
for r in rows: # 约定升序返回 → 后者覆盖前者即「最新一版」
|
||
out[r.symbol] = r
|
||
return out
|
||
|
||
def _annotate_price_basis(self, result, spec: ResearchSpec) -> None:
|
||
"""把价格口径写入结果（v3 §20.5：adjust_mode / price_basis / execution_price_basis）。
|
||
|
||
同时如实标注复权因子缺口（AGENT.md §24：未覆盖部分必须显式说明，禁止假装支持）。
|
||
"""
|
||
mode = spec.price_adjustment
|
||
result.config_snapshot["price_basis"] = {
|
||
"adjust_mode": mode,
|
||
"price_basis": "adjust_factor" if mode != "none" else "raw_close",
|
||
"execution_price_basis": "close_adj" if mode != "none" else "close_raw",
|
||
"adjust_gaps": self.last_adjust_gaps,
|
||
# exclude_st 的名称口径：时点（stock_name_history）还是最新快照
|
||
"name_basis": self.last_name_basis,
|
||
}
|
||
nb = self.last_name_basis
|
||
if nb is not None:
|
||
if nb["point_in_time"]:
|
||
result.unimplemented.append(
|
||
f"exclude_st 采用**时点名称**（stock_name_history，覆盖 {nb['covered']} 只生效名称）："
|
||
"「曾为高股息、后变 ST」的标的在其非 ST 期间会被正常纳入（股息陷阱可见）"
|
||
)
|
||
else:
|
||
result.unimplemented.append(
|
||
"exclude_st 回退**最新名称快照**（名称变更历史表未同步/未注入）："
|
||
"「曾为高股息、后变 ST/退市」的标的会被整段排除，收益可能被高估；"
|
||
"修复：python -m app.cli.sync namechange"
|
||
)
|
||
if mode == "none":
|
||
return
|
||
label = "后复权（hfq）" if mode == "hfq" else "前复权（qfq）"
|
||
result.unimplemented.append(
|
||
f"价格口径为{label}：价格列 = 原始价 × adjust_factor"
|
||
+ ("；现金分红按复权因子隐含再投资处理" if mode == "hfq" else "（以前复权归一）")
|
||
+ "；volume/amount 不做复权折算"
|
||
)
|
||
gaps = self.last_adjust_gaps or {}
|
||
if gaps.get("checked") and gaps.get("rows_without_factor", 0) > 0:
|
||
result.unimplemented.append(
|
||
f"复权因子覆盖不全：{gaps['rows_without_factor']} / {gaps['rows']} 行"
|
||
f"（{gaps['missing_pct']}%）无对应 adjust_factor，已按系数 1.0 兜底（未折算）"
|
||
)
|
||
|
||
def _attach_basic(
|
||
self,
|
||
daily: pd.DataFrame,
|
||
symbols: list[str],
|
||
start: date,
|
||
end: date,
|
||
columns: list[str],
|
||
) -> pd.DataFrame:
|
||
"""把 daily_basic 列并入行情长表（dv_ratio 等因子依赖）。"""
|
||
if self._basic_repo is None:
|
||
raise ValueError(
|
||
f"因子需要每日指标列 {columns}（daily_basic），但未注入 DailyBasicRepository。"
|
||
"请检查 API/CLI 的依赖装配。"
|
||
)
|
||
basic = load_basic_df(self._basic_repo, symbols, start, end, columns)
|
||
if basic.empty:
|
||
raise ValueError(
|
||
f"daily_basic 表在 {start}~{end} 无数据，无法计算需要 {columns} 的因子。"
|
||
"请先运行：python -m app.cli.sync daily_basic --start 20200101"
|
||
)
|
||
return merge_basic_into_daily(daily, basic)
|
||
|
||
|
||
def _stage(cb, name: str) -> None:
|
||
if cb is not None:
|
||
cb(name)
|