fix(backtest): ETF/指数矩阵链路不再要求股本字段, 修复因子挖掘准备阶段失败

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

@@ -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(

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))

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(),
+        )