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 逐类验证归档页）。
428 lines
18 KiB
Python
428 lines
18 KiB
Python
"""名称变更历史（时点 ST）测试：Repository、Provider 归一、universe 时点口径。
|
||
|
||
背景（实测）：`stock.name` 只是最新名称快照。用它做 `exclude_st` 会把
|
||
「曾是高股息、后来才变 ST/退市」的标的在**整段历史**里排除 —— 而那正是
|
||
「股息陷阱」样本。实测对照（同一 spec 仅改 exclude_st）：+35.71% → +32.01%，
|
||
即约 3.70pp 收益被名称快照口径隐藏。本模块锁定修复后的时点语义。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
|
||
import pandas as pd
|
||
import pytest
|
||
from app.domain.entities.market import Stock, StockNameHistory
|
||
from app.infrastructure.persistence.sqlalchemy.base import Base
|
||
from app.infrastructure.persistence.sqlalchemy.models.market import (
|
||
DailyBasicModel,
|
||
StockDailyModel,
|
||
StockModel,
|
||
StockNameHistoryModel,
|
||
)
|
||
from app.infrastructure.persistence.sqlalchemy.repositories.market_impl import (
|
||
SqlAlchemyStockNameHistoryRepository,
|
||
)
|
||
from sqlalchemy import create_engine
|
||
from sqlalchemy.orm import Session
|
||
|
||
|
||
@pytest.fixture
|
||
def session(tmp_path):
|
||
engine = create_engine(f"sqlite:///{tmp_path / 'name.db'}", future=True)
|
||
Base.metadata.create_all(engine)
|
||
s = Session(engine)
|
||
yield s
|
||
s.close()
|
||
|
||
|
||
def _seed_dima(session: Session) -> None:
|
||
"""600565.SH 迪马股份：2002-07-23 上市 → 2024-05-06 变 ST迪马（真实数据）。"""
|
||
session.add_all(
|
||
[
|
||
StockModel(
|
||
symbol="600565.SH", name="ST迪马", list_date=date(2002, 7, 23), status="L"
|
||
),
|
||
StockModel(
|
||
symbol="600519.SH", name="贵州茅台", list_date=date(2001, 8, 27), status="L"
|
||
),
|
||
StockNameHistoryModel(
|
||
symbol="600565.SH", name="迪马股份", start_date=date(2002, 7, 23),
|
||
end_date=date(2024, 5, 5), ann_date=date(2002, 7, 23), change_reason="其他",
|
||
),
|
||
StockNameHistoryModel(
|
||
symbol="600565.SH", name="ST迪马", start_date=date(2024, 5, 6),
|
||
end_date=None, ann_date=date(2024, 4, 30), change_reason="ST",
|
||
),
|
||
StockNameHistoryModel(
|
||
symbol="600519.SH", name="贵州茅台", start_date=date(2001, 8, 27),
|
||
end_date=None, ann_date=date(2001, 8, 27), change_reason="其他",
|
||
),
|
||
]
|
||
)
|
||
session.commit()
|
||
|
||
|
||
class TestRepository:
|
||
def test_names_as_of_picks_effective_span(self, session) -> None:
|
||
_seed_dima(session)
|
||
repo = SqlAlchemyStockNameHistoryRepository(session)
|
||
symbols = ["600565.SH", "600519.SH"]
|
||
# 2020 年：迪马股份（不是 ST）
|
||
assert repo.names_as_of(symbols, date(2020, 1, 2))["600565.SH"] == "迪马股份"
|
||
# 区间边界：末日仍为旧名，次日起为新名
|
||
assert repo.names_as_of(["600565.SH"], date(2024, 5, 5))["600565.SH"] == "迪马股份"
|
||
assert repo.names_as_of(["600565.SH"], date(2024, 5, 6))["600565.SH"] == "ST迪马"
|
||
# end_date 为空的区间至今有效
|
||
assert repo.names_as_of(["600565.SH"], date(2026, 9, 4))["600565.SH"] == "ST迪马"
|
||
|
||
def test_unknown_symbol_absent_from_map(self, session) -> None:
|
||
"""无记录 → 不返回该键，由调用方回退最新名称（不抛错）。"""
|
||
_seed_dima(session)
|
||
repo = SqlAlchemyStockNameHistoryRepository(session)
|
||
assert repo.names_as_of(["000001.SZ"], date(2020, 1, 2)) == {}
|
||
|
||
def test_upsert_is_idempotent(self, session) -> None:
|
||
repo = SqlAlchemyStockNameHistoryRepository(session)
|
||
row = StockNameHistory(
|
||
symbol="600565.SH", name="迪马股份", start_date=date(2002, 7, 23),
|
||
end_date=date(2024, 5, 5), change_reason="其他",
|
||
)
|
||
assert repo.upsert_many([row]) == 1
|
||
assert repo.upsert_many([row]) == 1 # 重跑无副作用
|
||
assert repo.count_rows() == 1
|
||
assert repo.namechange_dates() == (date(2002, 7, 23), date(2002, 7, 23))
|
||
|
||
def test_name_spans_grouped_by_symbol(self, session) -> None:
|
||
_seed_dima(session)
|
||
repo = SqlAlchemyStockNameHistoryRepository(session)
|
||
spans = repo.name_spans(["600565.SH"])
|
||
assert [n for _s, _e, n in spans["600565.SH"]] == ["迪马股份", "ST迪马"]
|
||
|
||
|
||
class TestUniversePointInTime:
|
||
"""filter_stocks(name_at=...) 的时点语义（案例收益口径的关键）。"""
|
||
|
||
def _stocks(self) -> list[Stock]:
|
||
return [
|
||
Stock(symbol="600565.SH", name="ST迪马", list_date=date(2002, 7, 23)),
|
||
Stock(symbol="600519.SH", name="贵州茅台", list_date=date(2001, 8, 27)),
|
||
]
|
||
|
||
def test_name_at_overrides_snapshot(self) -> None:
|
||
from app.domain.entities.research import UniverseSpec
|
||
from app.quant.universe import filter_stocks
|
||
|
||
u = UniverseSpec(exclude_st=True, min_listing_days=0)
|
||
# 旧口径（无 name_at）：最新名称含 ST → 2020 年就被排除（股息陷阱被隐藏）
|
||
assert "600565.SH" not in {s.symbol for s in filter_stocks(self._stocks(), u, date(2020, 1, 2))}
|
||
# 时点口径：2020 年它叫「迪马股份」→ 必须纳入
|
||
got = {
|
||
s.symbol
|
||
for s in filter_stocks(
|
||
self._stocks(), u, date(2020, 1, 2), name_at={"600565.SH": "迪马股份"}
|
||
)
|
||
}
|
||
assert "600565.SH" in got
|
||
# 2024-05-06 起为 ST迪马 → 排除
|
||
got2 = {
|
||
s.symbol
|
||
for s in filter_stocks(
|
||
self._stocks(), u, date(2024, 6, 3), name_at={"600565.SH": "ST迪马"}
|
||
)
|
||
}
|
||
assert "600565.SH" not in got2
|
||
|
||
def test_name_at_missing_falls_back_to_snapshot(self) -> None:
|
||
from app.domain.entities.research import UniverseSpec
|
||
from app.quant.universe import filter_stocks
|
||
|
||
u = UniverseSpec(exclude_st=True, min_listing_days=0)
|
||
got = filter_stocks(self._stocks(), u, date(2020, 1, 2), name_at={})
|
||
assert "600565.SH" not in {s.symbol for s in got} # 回退快照，行为不变
|
||
|
||
def test_exclude_st_false_ignores_names(self) -> None:
|
||
from app.domain.entities.research import UniverseSpec
|
||
from app.quant.universe import filter_stocks
|
||
|
||
u = UniverseSpec(exclude_st=False, min_listing_days=0)
|
||
got = filter_stocks(self._stocks(), u, date(2020, 1, 2), name_at={"600565.SH": "ST迪马"})
|
||
assert "600565.SH" in {s.symbol for s in got}
|
||
|
||
|
||
class TestNamesAsOfHelper:
|
||
def test_none_repo_reports_snapshot_basis(self) -> None:
|
||
from app.quant.universe import names_as_of
|
||
|
||
name_at, applied = names_as_of([], date(2020, 1, 2), None)
|
||
assert name_at is None
|
||
assert applied == (False, 0)
|
||
|
||
def test_repo_failure_degrades_without_raising(self) -> None:
|
||
"""名称历史查询异常不得让选股/回测整体失败（降级为快照口径）。"""
|
||
from app.quant.universe import names_as_of
|
||
|
||
class Boom:
|
||
def names_as_of(self, symbols, as_of):
|
||
raise RuntimeError("表不存在")
|
||
|
||
name_at, applied = names_as_of([], date(2020, 1, 2), Boom())
|
||
assert name_at is None and applied == (False, 0)
|
||
|
||
|
||
class TestProviderNormalize:
|
||
def test_normalize_name_history(self) -> None:
|
||
from app.infrastructure.data_sources.tushare import TushareProvider
|
||
|
||
rows = TushareProvider.normalize_name_history(
|
||
[
|
||
{
|
||
"ts_code": "600565.SH", "name": "ST迪马", "start_date": "20240506",
|
||
"end_date": None, "ann_date": "20240430", "change_reason": "ST",
|
||
},
|
||
{
|
||
"ts_code": "600565.SH", "name": "迪马股份", "start_date": "20020723",
|
||
"end_date": "20240505", "ann_date": "20020723", "change_reason": "其他",
|
||
},
|
||
]
|
||
)
|
||
assert rows[1].start_date == date(2002, 7, 23)
|
||
assert rows[0].end_date is None
|
||
assert rows[0].is_risk_warned is True
|
||
assert rows[1].is_risk_warned is False
|
||
|
||
def test_nan_end_date_and_bad_code_skipped(self) -> None:
|
||
"""实测：namechange 的 end_date 是 float NaN，曾让 32/37 个分片整体失败。"""
|
||
from app.infrastructure.data_sources.tushare import TushareProvider
|
||
|
||
rows = TushareProvider.normalize_name_history(
|
||
[
|
||
{
|
||
"ts_code": "000001.SZ", "name": "平安银行", "start_date": "19910403",
|
||
"end_date": float("nan"), "ann_date": float("nan"), "change_reason": None,
|
||
},
|
||
{"ts_code": "T600018.SH", "name": "上港集箱(退)", "start_date": "19960101"},
|
||
]
|
||
)
|
||
assert len(rows) == 1
|
||
assert rows[0].end_date is None and rows[0].ann_date is None
|
||
|
||
def test_to_date_rejects_nan_variants(self) -> None:
|
||
from app.infrastructure.data_sources.tushare import _to_date
|
||
|
||
assert _to_date(float("nan")) is None
|
||
assert _to_date("nan") is None
|
||
assert _to_date("None") is None
|
||
assert _to_date("") is None
|
||
assert _to_date(None) is None
|
||
assert _to_date("20240506") == date(2024, 5, 6)
|
||
|
||
|
||
class TestPerDateStFilter:
|
||
"""回测的 exclude_st 必须**逐择股日**重判（与 /api/selections 单时点同口径）。
|
||
|
||
否则「入池时非 ST、之后才变 ST」的标的会在之后所有择股日继续被选中 ——
|
||
正是高股息策略最危险的股息陷阱路径。
|
||
"""
|
||
|
||
def _build(self, tmp_path):
|
||
"""A 高股息但 2024-06-01 变 ST；B 低股息且始终非 ST。"""
|
||
from app.infrastructure.persistence.sqlalchemy.repositories.market_impl import (
|
||
SqlAlchemyDailyBarRepository,
|
||
SqlAlchemyDailyBasicRepository,
|
||
SqlAlchemyStockNameHistoryRepository,
|
||
SqlAlchemyStockRepository,
|
||
)
|
||
|
||
engine = create_engine(f"sqlite:///{tmp_path / 'pit.db'}", future=True)
|
||
Base.metadata.create_all(engine)
|
||
session = Session(engine)
|
||
syms = ["600000.SH", "600001.SH"]
|
||
session.add_all(
|
||
[
|
||
StockModel(symbol=syms[0], name="ST阿甲", list_date=date(2000, 1, 1), status="L"),
|
||
StockModel(symbol=syms[1], name="阿乙", list_date=date(2000, 1, 1), status="L"),
|
||
# 名称历史：A 2024-06-01 起为 ST阿甲
|
||
StockNameHistoryModel(
|
||
symbol=syms[0], name="阿甲", start_date=date(2000, 1, 1),
|
||
end_date=date(2024, 5, 31), change_reason="其他",
|
||
),
|
||
StockNameHistoryModel(
|
||
symbol=syms[0], name="ST阿甲", start_date=date(2024, 6, 1),
|
||
end_date=None, change_reason="ST",
|
||
),
|
||
StockNameHistoryModel(
|
||
symbol=syms[1], name="阿乙", start_date=date(2000, 1, 1),
|
||
end_date=None, change_reason="其他",
|
||
),
|
||
]
|
||
)
|
||
days = pd.bdate_range("2023-12-01", "2024-12-31")
|
||
for sym, dv in ((syms[0], 12.0), (syms[1], 4.0)):
|
||
session.add_all(
|
||
[
|
||
StockDailyModel(
|
||
symbol=sym, trade_date=d.date(), open=Decimal("10"),
|
||
high=Decimal("10"), low=Decimal("10"), close=Decimal("10"),
|
||
volume=Decimal("1000"), amount=Decimal("10000"), source="tushare",
|
||
)
|
||
for d in days
|
||
]
|
||
)
|
||
session.add_all(
|
||
[
|
||
DailyBasicModel(symbol=sym, trade_date=d.date(), dv_ratio=Decimal(str(dv)))
|
||
for d in days
|
||
]
|
||
)
|
||
session.commit()
|
||
from app.quant.engine import LocalEngine
|
||
from app.quant.service import ResearchService
|
||
|
||
svc = ResearchService(
|
||
SqlAlchemyStockRepository(session),
|
||
SqlAlchemyDailyBarRepository(session),
|
||
LocalEngine(),
|
||
basic_repo=SqlAlchemyDailyBasicRepository(session),
|
||
name_repo=SqlAlchemyStockNameHistoryRepository(session),
|
||
)
|
||
return svc, session, syms
|
||
|
||
def _spec(self):
|
||
from app.domain.entities.research import (
|
||
FactorSpec,
|
||
ResearchSpec,
|
||
SelectionSpec,
|
||
UniverseSpec,
|
||
)
|
||
|
||
return ResearchSpec(
|
||
type="backtest",
|
||
universe=UniverseSpec(exclude_st=True, min_listing_days=0),
|
||
factors=[FactorSpec(name="dividend_yield", weight=1.0)],
|
||
selection=SelectionSpec(top_n=2, hold_top_x=1, allow_substitute=True),
|
||
period=("2024-01-01", "2024-12-31"),
|
||
selection_interval_months=6,
|
||
rebalance_interval_months=6,
|
||
initial_capital=100000.0,
|
||
)
|
||
|
||
def test_st_stock_dropped_from_later_selection(self, tmp_path) -> None:
|
||
svc, session, syms = self._build(tmp_path)
|
||
try:
|
||
result = svc.run_backtest(self._spec())
|
||
finally:
|
||
session.close()
|
||
picks = {}
|
||
for p in result.selection_history:
|
||
picks.setdefault(p.date, []).append(p.symbol)
|
||
dates = sorted(picks)
|
||
assert len(dates) >= 2, picks
|
||
first, second = dates[0], dates[-1]
|
||
# 首个择股日：A 非 ST（高股息）→ 入选
|
||
assert syms[0] in picks[first]
|
||
# 变 ST 之后的择股日：A 必须消失，B 顶上
|
||
assert syms[0] not in picks[second]
|
||
assert syms[1] in picks[second]
|
||
# 口径标注：时点名称（不是快照回退）
|
||
assert result.config_snapshot["price_basis"]["name_basis"]["point_in_time"] is True
|
||
|
||
def test_snapshot_fallback_without_repo(self, tmp_path) -> None:
|
||
"""未注入名称历史 → 回退最新名称（旧行为），并如实标注 point_in_time=False。"""
|
||
svc, session, syms = self._build(tmp_path)
|
||
try:
|
||
svc._name_repo = None
|
||
result = svc.run_backtest(self._spec())
|
||
finally:
|
||
session.close()
|
||
# 最新名称是 ST阿甲 → 首个择股日就被排除（股息陷阱被隐藏，已标注）
|
||
first = min(p.date for p in result.selection_history)
|
||
assert syms[0] not in [p.symbol for p in result.selection_history if p.date == first]
|
||
assert result.config_snapshot["price_basis"]["name_basis"]["point_in_time"] is False
|
||
|
||
|
||
class TestReviewRegressions:
|
||
"""代码审查发现的缺陷回归（P1/P2/P3）：不得复活。"""
|
||
|
||
def test_empty_table_does_not_claim_point_in_time(self) -> None:
|
||
"""[P1] 表存在但为空时必须降级为快照口径，不得声称 point_in_time=true。
|
||
|
||
否则结果页会把「未被修正的 10.85pp 股息陷阱偏差」当成已修正上报（AGENT.md §24）。
|
||
"""
|
||
from app.quant.universe import names_as_of
|
||
|
||
class EmptyRepo:
|
||
def names_as_of(self, symbols, as_of):
|
||
return {}
|
||
|
||
stocks = [Stock(symbol="600565.SH", name="ST迪马", list_date=date(2000, 1, 1))]
|
||
name_at, applied = names_as_of(stocks, date(2020, 1, 2), EmptyRepo())
|
||
assert name_at is None
|
||
assert applied == (False, 0)
|
||
|
||
def test_empty_table_end_to_end_reports_snapshot_basis(self, tmp_path) -> None:
|
||
"""[P1] 端到端：名称表为空 → name_basis.point_in_time 必须为 False。"""
|
||
svc, session, _syms = TestPerDateStFilter()._build(tmp_path)
|
||
try:
|
||
session.query(StockNameHistoryModel).delete()
|
||
session.commit()
|
||
result = svc.run_backtest(TestPerDateStFilter()._spec())
|
||
finally:
|
||
session.close()
|
||
assert result.config_snapshot["price_basis"]["name_basis"]["point_in_time"] is False
|
||
assert any("最新名称快照" in n for n in result.unimplemented)
|
||
|
||
def test_missing_span_falls_back_to_snapshot_in_backtest(self, tmp_path) -> None:
|
||
"""[P2] 某股无生效区间时，逐择股日 ST 必须与 filter_stocks 同口径回退最新名称。
|
||
|
||
否则同一择股日会出现「回测入选、/api/selections 排除」的打架（v2 §25）。
|
||
"""
|
||
svc, session, syms = TestPerDateStFilter()._build(tmp_path)
|
||
try:
|
||
# 删掉 A 的**全部**名称区间：只剩最新名称 ST阿甲
|
||
session.query(StockNameHistoryModel).filter(
|
||
StockNameHistoryModel.symbol == syms[0]
|
||
).delete()
|
||
session.commit()
|
||
result = svc.run_backtest(TestPerDateStFilter()._spec())
|
||
dates = sorted({p.date for p in result.selection_history})
|
||
first = dates[0]
|
||
picks_first = [p.symbol for p in result.selection_history if p.date == first]
|
||
finally:
|
||
session.close()
|
||
# 快照口径：A 名称含 ST → 首个择股日即被排除（与 filter_stocks 一致）
|
||
assert syms[0] not in picks_first
|
||
assert syms[1] in picks_first
|
||
|
||
def test_protocol_declares_name_changes(self) -> None:
|
||
"""[P2] MarketDataProvider 必须声明 get_name_changes（§6 业务层只依赖抽象）。"""
|
||
|
||
from app.domain.providers import MarketDataProvider
|
||
|
||
assert hasattr(MarketDataProvider, "get_name_changes")
|
||
assert "get_name_changes" in dir(MarketDataProvider)
|
||
|
||
def test_sina_declares_not_supported(self) -> None:
|
||
"""[P2] 备用源必须显式 NotSupported，不得静默返回空列表（否则时点 ST 静默降级）。"""
|
||
from app.infrastructure.data_sources.errors import DataSourceNotSupported
|
||
from app.infrastructure.data_sources.sina import SinaProvider
|
||
|
||
provider = SinaProvider.__new__(SinaProvider)
|
||
with pytest.raises(DataSourceNotSupported):
|
||
provider.get_name_changes(date(2024, 1, 1), date(2024, 12, 31))
|
||
|
||
def test_qlib_engine_accepts_eligibility_fn(self) -> None:
|
||
"""[P2] 引擎协议一致性：所有引擎都必须接受 eligibility_fn。
|
||
|
||
否则注入 QlibEngine 后任何回测都会 TypeError（且条件/时点 ST 会被静默忽略）。
|
||
"""
|
||
import inspect
|
||
|
||
from app.quant.engine import QuantEngine
|
||
from app.quant.qlib_adapter.engine import QlibEngine
|
||
|
||
for cls in (QuantEngine, QlibEngine):
|
||
params = inspect.signature(cls.run_backtest).parameters
|
||
assert "eligibility_fn" in params, cls.__name__
|