From 61130cfdff433da405898e13db4d5c3b3650e17b Mon Sep 17 00:00:00 2001 From: shy3130 Date: Sun, 23 Aug 2026 17:46:54 +0800 Subject: [PATCH] =?UTF-8?q?fix(backtest):=20ETF/=E6=8C=87=E6=95=B0?= =?UTF-8?q?=E7=9F=A9=E9=98=B5=E9=93=BE=E8=B7=AF=E4=B8=8D=E5=86=8D=E8=A6=81?= =?UTF-8?q?=E6=B1=82=E8=82=A1=E6=9C=AC=E5=AD=97=E6=AE=B5,=20=E4=BF=AE?= =?UTF-8?q?=E5=A4=8D=E5=9B=A0=E5=AD=90=E6=8C=96=E6=8E=98=E5=87=86=E5=A4=87?= =?UTF-8?q?=E9=98=B6=E6=AE=B5=E5=A4=B1=E8=B4=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit etf 维表(index_sync)物理上只有 symbol/name/code/asset_type, enriched 窄表 也不落盘股本字段; 而 DEFAULT_BASIC_FILTER 与矩阵缓存档 common_filter 均带 非 None 市值界, 依赖解析会无条件要求 total_shares/float_shares, 导致 _resolve_matrix_storage_fields 抛 "matrix parquet fields unavailable: ['float_shares', 'total_shares']" (用户反馈: ETF 因子挖掘死于「准备共享撮合矩阵」阶段)。 - 新增 _basic_filter_for_asset: 非股票资产将市值/流通市值四项界置 None, 依赖解析前中和; 运行期过滤本就对缺失股本列降级 no-op, 无需改动 - StrategyDependencyResolver.resolve 新增 asset_type 参数(默认 stock, 旧调用行为不变), 四个调用点传入真实资产类型: build_matrix_cache_profile / prepare_matrix_optimization / 单次回测 / composite 子计划 - mining_runtime._prepare_base_market 两处 resolve 传入 request.asset_type - 新增回归测试: ETF 计划与缓存档不含股本字段、股票行为不变、 ETF 维表下字段解析成功、反向对照锁定失败模式 验证: 全量 pytest 976 passed; 真实组件端到端复现用户失败路径已转绿 (ETF enriched + 四列维表, 共享矩阵 (280,3) 构建成功)。 --- backend/app/backtest/mining_runtime.py | 2 + backend/app/backtest/strategy.py | 38 ++++- backend/tests/test_matrix_etf_share_fields.py | 141 ++++++++++++++++++ 3 files changed, 180 insertions(+), 1 deletion(-) create mode 100644 backend/tests/test_matrix_etf_share_fields.py diff --git a/backend/app/backtest/mining_runtime.py b/backend/app/backtest/mining_runtime.py index 0733d7c..f351ccd 100644 --- a/backend/app/backtest/mining_runtime.py +++ b/backend/app/backtest/mining_runtime.py @@ -926,6 +926,7 @@ def _prepare_base_market( entry_signals=research.entry_signals, exit_signals=research.exit_signals, overrides={}, + asset_type=request.asset_type, )) for strategy_id in request.strategy_ids: strategy = strategy_engine.get(strategy_id) @@ -942,6 +943,7 @@ def _prepare_base_market( overrides, "exit_signals", strategy.exit_signals ), overrides=overrides, + asset_type=request.asset_type, )) merged = _merge_resolved_feature_plans(plans) profile = build_matrix_cache_profile( diff --git a/backend/app/backtest/strategy.py b/backend/app/backtest/strategy.py index a8af4d4..9ae6803 100644 --- a/backend/app/backtest/strategy.py +++ b/backend/app/backtest/strategy.py @@ -129,8 +129,10 @@ class StrategyDependencyResolver: exit_signals: list[str], overrides: dict | None = None, minute_fill: bool = False, + asset_type: str = "stock", ) -> ResolvedFeaturePlan: overrides = overrides or {} + basic_filter = _basic_filter_for_asset(basic_filter, asset_type) if strategy.execution_backend == "matrix_native": return self._resolve_matrix_native( strategy, @@ -329,6 +331,7 @@ def build_matrix_cache_profile( exit_signals=strategy.exit_signals, overrides={}, minute_fill=False, + asset_type=asset_type, )) forward_bars = max(forward_bars, int(strategy.max_hold_days or 0)) @@ -470,6 +473,31 @@ def _basic_filter_dependencies(config: dict) -> set[str]: return dependencies +_SHARE_CAP_FILTER_KEYS = ( + "market_cap_min", + "market_cap_max", + "float_cap_min", + "float_cap_max", +) + + +def _basic_filter_for_asset(basic_filter: dict, asset_type: str) -> dict: + """非股票资产没有股本数据 (etf/index 维表只有 symbol/name), 市值与流通 + 市值界对它们既无意义也不可满足: 依赖解析前先置 None, 避免解析出 + total_shares/float_shares 字段需求导致矩阵加载直接失败。 + + 运行期过滤无需同步修改 —— polars 侧有列守卫 (engine._basic_filter_expr), + 矩阵侧 _optional_field 对缺失字段返回全 NaN 且 _apply_bound 跳过全 NaN + 界, 二者对缺失股本列本就降级为 no-op。 + """ + if asset_type == "stock" or not basic_filter: + return basic_filter + sanitized = dict(basic_filter) + for key in _SHARE_CAP_FILTER_KEYS: + sanitized[key] = None + return sanitized + + def _resolve_base_columns(features: set[str]) -> frozenset[str]: storage = set(ENRICHED_STORAGE_COLS) base = {"symbol", "date"} | (features & storage) @@ -648,6 +676,7 @@ class StrategyBacktestService: params: dict, basic_filter: dict, overrides: dict, + asset_type: str = "stock", ) -> tuple[ResolvedFeaturePlan, list[tuple[StrategyDef, dict, dict]]]: """解析 composite 回测的特征计划: 所有子策略 feature_plan 的并集。 @@ -696,6 +725,7 @@ class StrategyBacktestService: entry_signals=[], exit_signals=[], overrides=child_override, + asset_type=asset_type, ) plans.append(child_plan) # pipeline 用 composite 统一的 basic_filter; scoring 用子策略自己的 @@ -812,6 +842,7 @@ class StrategyBacktestService: exit_signals=exit_signals, overrides=overrides, minute_fill=config.minute_fill, + asset_type=config.asset_type, )) feature_plan = _merge_resolved_feature_plans(plans) @@ -1018,7 +1049,11 @@ class StrategyBacktestService: # composite 回测: 子策略必须全为 matrix_native(否则 fail-closed), # feature_plan 取所有子策略计划的并集(_merge_resolved_feature_plans)。 feature_plan, composite_children_resolved = self._resolve_composite_feature_plan( - s, params=params, basic_filter=basic_filter, overrides=overrides + s, + params=params, + basic_filter=basic_filter, + overrides=overrides, + asset_type=config.asset_type, ) else: composite_children_resolved = None @@ -1030,6 +1065,7 @@ class StrategyBacktestService: exit_signals=exit_signals, overrides=overrides, minute_fill=config.minute_fill, + asset_type=config.asset_type, ) except ValueError as e: return _err(str(e)) diff --git a/backend/tests/test_matrix_etf_share_fields.py b/backend/tests/test_matrix_etf_share_fields.py new file mode 100644 index 0000000..2c3a0a6 --- /dev/null +++ b/backend/tests/test_matrix_etf_share_fields.py @@ -0,0 +1,141 @@ +"""ETF/指数矩阵链路不得依赖股本字段 (total_shares/float_shares)。 + +etf 维表 (index_sync) 物理上只有 symbol/name/code/asset_type, enriched +窄表也不落盘股本字段; 而矩阵缓存档的 common_filter 与 DEFAULT_BASIC_FILTER +都带非 None 的市值界, 依赖解析会因此无条件要求股本字段, 最终在 +_resolve_matrix_storage_fields 抛 +"matrix parquet fields unavailable: ['float_shares', 'total_shares']" +(用户反馈: ETF 因子挖掘死于「准备共享撮合矩阵」阶段)。非股票资产必须在 +依赖解析前中和市值界, 股票行为保持不变。 +""" +from __future__ import annotations + +from datetime import date +from pathlib import Path + +import polars as pl +import pyarrow.dataset as pads +import pytest + +from app.backtest.matrix import ( + _normalize_matrix_cache_fields, + _resolve_matrix_storage_fields, +) +from app.backtest.strategy import ( + StrategyDependencyResolver, + build_matrix_cache_profile, +) +from app.strategy.engine import StrategyEngine + +_BUILTIN = Path(__file__).resolve().parents[1] / "app" / "strategy" / "builtin" + + +def _engine() -> StrategyEngine: + return StrategyEngine(strategy_dirs=[_BUILTIN, _BUILTIN.parent / "custom"]) + + +def _resolve_research_plan(engine: StrategyEngine, asset_type: str = "stock"): + research = engine.get("factor_rank_research") + return StrategyDependencyResolver().resolve( + research, + params={ + "scoring": {"turnover_rate": 1.0}, + "directions": {"turnover_rate": "high"}, + }, + basic_filter=dict(research.basic_filter), + entry_signals=research.entry_signals, + exit_signals=research.exit_signals, + overrides={}, + asset_type=asset_type, + ) + + +def test_etf_plan_does_not_require_share_fields(): + engine = _engine() + assert engine.get("factor_rank_research").basic_filter.get("market_cap_min") is not None + + etf_plan = _resolve_research_plan(engine, asset_type="etf") + assert "total_shares" not in etf_plan.instrument_columns + assert "float_shares" not in etf_plan.instrument_columns + + stock_plan = _resolve_research_plan(engine, asset_type="stock") + assert "total_shares" in stock_plan.instrument_columns + + +def test_etf_cache_profile_excludes_share_fields(): + engine = _engine() + etf_plan = _resolve_research_plan(engine, asset_type="etf") + etf_profile = build_matrix_cache_profile( + engine, "etf", requested_plan=etf_plan, requested_forward_bars=5, + ) + assert "total_shares" not in etf_profile.field_columns + assert "float_shares" not in etf_profile.field_columns + + stock_plan = _resolve_research_plan(engine, asset_type="stock") + stock_profile = build_matrix_cache_profile( + engine, "stock", requested_plan=stock_plan, requested_forward_bars=5, + ) + assert {"total_shares", "float_shares"} <= set(stock_profile.field_columns) + + +def _etf_instruments() -> pl.DataFrame: + # index_sync._fetch_instruments_by_type("etf","etf") 的实际落盘 schema + return pl.DataFrame({ + "symbol": ["510300.SH", "510500.SH"], + "name": ["沪深300ETF", "中证500ETF"], + "code": ["510300", "510500"], + "asset_type": ["etf", "etf"], + }) + + +def _etf_dataset(tmp_path: Path) -> pads.Dataset: + pl.DataFrame({ + "symbol": ["510300.SH", "510500.SH"], + "date": [date(2024, 1, 2)] * 2, + "open": [4.0, 6.0], + "high": [4.1, 6.1], + "low": [3.9, 5.9], + "close": [4.0, 6.0], + "volume": [100.0, 200.0], + "amount": [400.0, 1200.0], + "raw_close": [4.0, 6.0], + "raw_high": [4.1, 6.1], + "raw_low": [3.9, 5.9], + "turnover_rate": [0.01, 0.02], + }).write_parquet(tmp_path / "part.parquet") + return pads.dataset(str(tmp_path / "part.parquet"), format="parquet") + + +def test_matrix_storage_fields_resolve_with_etf_instruments(tmp_path): + engine = _engine() + etf_plan = _resolve_research_plan(engine, asset_type="etf") + profile = build_matrix_cache_profile( + engine, "etf", requested_plan=etf_plan, requested_forward_bars=5, + ) + requested = ( + set(etf_plan.base_columns) + | set(etf_plan.instrument_columns) + | set(etf_plan.matrix_columns) + ) + build_fields = frozenset( + _normalize_matrix_cache_fields(frozenset(requested)) + | _normalize_matrix_cache_fields(profile.field_columns) + ) + assert "total_shares" not in build_fields + + _parquet_fields, matrix_fields, vector_fields = _resolve_matrix_storage_fields( + _etf_dataset(tmp_path), build_fields, _etf_instruments(), + ) + assert "total_shares" not in matrix_fields + vector_fields + assert vector_fields == [] + + +def test_share_fields_still_unavailable_without_sanitization(tmp_path): + # 反向对照: 若市值界未被中和, ETF 维表下解析仍然失败 —— 锁定失败模式, + # 防止未来把中和逻辑误删。 + with pytest.raises(ValueError, match=r"matrix parquet fields unavailable"): + _resolve_matrix_storage_fields( + _etf_dataset(tmp_path), + frozenset({"name", "total_shares", "float_shares"}), + _etf_instruments(), + )