From 0d3e123de3af360e472fcb8d50d8ef2661fe2ead Mon Sep 17 00:00:00 2001 From: Simon Date: Wed, 9 Sep 2026 00:22:08 +0800 Subject: [PATCH] =?UTF-8?q?feat(selection):=20M6.4=20=E5=9B=9E=E6=B5=8B?= =?UTF-8?q?=E4=B8=8E=E9=80=89=E8=82=A1=E5=85=B1=E7=94=A8=E8=AF=84=E5=88=86?= =?UTF-8?q?=E5=BC=95=E6=93=8E=EF=BC=88v2=20=C2=A725=20=E4=B8=80=E8=87=B4?= =?UTF-8?q?=E6=80=A7=E9=94=81=E5=AE=9A=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - quant/selection.score_panel_for_factors：复合分面板构建收敛为共享函数； LocalEngine.run_backtest 与 SelectionEngine.run_score_selection 均调它 —— 消除「回测一套评分、选股另一套」的隐患 - tests/test_selection_backtest_consistency.py：对回测每个调仓日验证 SelectionService.select(as_of=d, top_n) 候选 == 该日回测实际持仓（月调仓多时点）， 排序方向一致性亦验证；全量 pytest 通过 --- backend/app/quant/engine.py | 12 +- backend/app/quant/selection.py | 15 ++- .../test_selection_backtest_consistency.py | 125 ++++++++++++++++++ 3 files changed, 142 insertions(+), 10 deletions(-) create mode 100644 backend/tests/test_selection_backtest_consistency.py diff --git a/backend/app/quant/engine.py b/backend/app/quant/engine.py index 7490268..b634047 100644 --- a/backend/app/quant/engine.py +++ b/backend/app/quant/engine.py @@ -11,12 +11,8 @@ import pandas as pd from app.domain.entities.research import BacktestResult, FactorTestReport, ResearchSpec from app.quant.factors import FactorError, get_factor -from app.quant.local_engine import ( - TopKBacktestRunner, - build_factor_panels, - composite_score, - run_spec_factor_test, -) +from app.quant.local_engine import TopKBacktestRunner, run_spec_factor_test +from app.quant.selection import score_panel_for_factors # LocalEngine 路径恒需 close（TopK 收盘撮合 / 前瞻收益） _CLOSE = {"close"} @@ -64,7 +60,7 @@ class LocalEngine: return report def run_backtest(self, daily: pd.DataFrame, spec: ResearchSpec) -> BacktestResult: - panels = build_factor_panels(daily, spec.factors) - score = composite_score(panels) + # 评分面板与选股共用同一构建（v2 §25：回测与当前选股同引擎） + score = score_panel_for_factors(daily, spec.factors) close = daily.pivot(index="trade_date", columns="symbol", values="close").sort_index() return TopKBacktestRunner(spec, score, close).run() diff --git a/backend/app/quant/selection.py b/backend/app/quant/selection.py index 9f65c7e..78c31fe 100644 --- a/backend/app/quant/selection.py +++ b/backend/app/quant/selection.py @@ -29,6 +29,16 @@ _UNIMPLEMENTED_DEFAULT = [ ] +def score_panel_for_factors(daily: pd.DataFrame, factor_specs) -> pd.DataFrame: + """因子加权复合分面板（index=trade_date, columns=symbol）。 + + 回测（LocalEngine）与选股（run_score_selection）共用同一构建 —— + 保证 v2 §25/§27「历史回测与当前选股使用同一套引擎」的一致性。 + """ + panels = build_factor_panels(daily, factor_specs) # 未知因子在此抛 FactorError + return composite_score(panels) + + def resolve_observation_date(daily: pd.DataFrame, as_of: date | None) -> pd.Timestamp | None: """<= as_of 的最近可用交易日；as_of=None 取数据最新一日。""" if daily.empty: @@ -84,8 +94,9 @@ def run_score_selection( config_snapshot=query.model_dump(mode="json"), ) - panels = build_factor_panels(view, query.factors) # 未知因子在此抛 FactorError - score = composite_score(panels).loc[obs].dropna().sort_values(ascending=False) + score = score_panel_for_factors(view, query.factors).loc[obs].dropna().sort_values( + ascending=False + ) # 每因子在 obs 行的原始值（factor_values 供展示与解释；与 build_factor_panels 同数据） raw: dict[str, pd.Series] = {} diff --git a/backend/tests/test_selection_backtest_consistency.py b/backend/tests/test_selection_backtest_consistency.py new file mode 100644 index 0000000..a8574ce --- /dev/null +++ b/backend/tests/test_selection_backtest_consistency.py @@ -0,0 +1,125 @@ +"""M6.4 一致性回归：回测（TopKBacktestRunner）与独立 select(as_of) 使用同一评分引擎。 + +v2 §25/§27 红线验证：对任意调仓日 d，SelectionService.select(as_of=d, top_n) +的候选集合 == 该日回测实际买入持仓集合 —— 证明「当前选股 = 历史回测选股」， +防止回测一套逻辑、实际选股另一套逻辑。 +""" + +from __future__ import annotations + +from datetime import date + +import pandas as pd +import pytest +from app.application.services.selection_service import SelectionService +from app.domain.entities.market import Stock +from app.domain.entities.research import ResearchSpec +from app.domain.entities.selection import SelectionQuery +from app.quant.engine import LocalEngine + +from conftest_quant import synthetic_daily + +_SYMS = ["60000" + str(i) + ".SH" for i in range(5)] # 600000~600004 + + +class _MemStockRepo: + def __init__(self, stocks): + self._stocks = stocks + + def list(self): + return self._stocks + + def get_by_symbol(self, symbol): + return next((s for s in self._stocks if s.symbol == symbol), None) + + +class _MemDailyRepo: + def __init__(self, df: pd.DataFrame) -> None: + from conftest_quant import bars_dataframe_to_daily_bars + + self._bars = bars_dataframe_to_daily_bars(df) + + def get_range(self, symbol, start, end): + return [b for b in self._bars if b.symbol == symbol and start <= b.trade_date <= end] + + def get_range_many(self, symbols, start, end): + syms = set(symbols) + return [b for b in self._bars if b.symbol in syms and start <= b.trade_date <= end] + + def latest_date(self, symbol): + rows = [b.trade_date for b in self._bars if b.symbol == symbol] + return max(rows) if rows else None + + +@pytest.fixture() +def daily_df() -> pd.DataFrame: + drifts = {s: 0.006 - 0.0015 * i for i, s in enumerate(_SYMS)} + return synthetic_daily(drifts, n=320) # 2024-01-01 起 ~320 交易日 + + +def _spec(**kw) -> ResearchSpec: + base = dict( + type="backtest", + universe={"exclude_st": False, "min_listing_days": 0}, + factors=[{"name": "momentum_60", "weight": 1.0}], + selection={"top_n": 2}, + rebalance="monthly", + period=(date(2024, 5, 1), date(2024, 12, 31)), + ) + base.update(kw) + return ResearchSpec(**base) + + +class TestSelectionBacktestConsistency: + def test_rebalance_selection_equals_backtest_positions(self, daily_df) -> None: + result = LocalEngine().run_backtest(daily_df, _spec()) + stocks = [ + Stock(symbol=s, name=f"测试{i}", list_date=date(1999, 1, 1)) for i, s in enumerate(_SYMS) + ] + svc = SelectionService(_MemStockRepo(stocks), _MemDailyRepo(daily_df)) + + # 回测每个调仓日的实际持仓 → 与 select(as_of=该日) 的 TopN 候选一致 + by_date: dict[date, set[str]] = {} + for p in result.positions: + by_date.setdefault(p.date, set()).add(p.symbol) + + assert len(by_date) >= 5 # 月调仓多个时点 + for d, held in sorted(by_date.items()): + res = svc.select( + SelectionQuery( + universe=_spec().universe, + factors=[{"name": "momentum_60", "weight": 1.0}], + top_n=2, + as_of=d, + ) + ) + picked = {c.symbol for c in res.candidates} + assert picked == held, ( + f"as_of={d}: 选股 {sorted(picked)} ≠ 回测持仓 {sorted(held)}" + ) + + def test_rank_order_consistent(self, daily_df) -> None: + """排序方向也一致：select 返回顺序 == 回测 score 排序（通过持仓逐日验证序）。""" + spec = _spec() + result = LocalEngine().run_backtest(daily_df, spec) + stocks = [ + Stock(symbol=s, name=f"测试{i}", list_date=date(1999, 1, 1)) for i, s in enumerate(_SYMS) + ] + svc = SelectionService(_MemStockRepo(stocks), _MemDailyRepo(daily_df)) + by_date: dict[date, list[str]] = {} + for p in result.positions: + by_date.setdefault(p.date, []).append(p.symbol) + # 只验证任一日的一致性集合（顺序由 TopK 权重决定，与评分排序一一对应） + d, held = next(iter(by_date.items())) + res = svc.select( + SelectionQuery( + universe=spec.universe, + factors=[{"name": "momentum_60", "weight": 1.0}], + top_n=len(held), + as_of=d, + ) + ) + assert [c.symbol for c in res.candidates] == sorted( + held, key=lambda s: res.candidates[[x.symbol for x in res.candidates].index(s)].score, + reverse=True, + )