fix(backtest): 矩阵回测接入扩展时序数据

扩展数值列按设计不落 enriched parquet (只存 14 列), 此前矩阵回测引用
ext_* 字段直接报 fields unavailable — 叠加条件/评分引用扩展数据的策略
无法回测, 扩展数据失去回测意义。

修复: 引擎在矩阵加载后按矩阵自身 (日期×标的) 骨架现场注入时序模式数值
扩展字段 (attach_matrix_ext_fields, 与策略页帧路径同一台 PIT 安全 join,
缺失补空值=条件不命中), 与财务因子 attach 同构; parquet 矩阵缓存不受
影响。快照模式与字符串扩展字段仍拒绝 (快照回看历史=未来函数), 引用时
前置校验给出明确报错。

实测: 隔夜溢价精选·迭代优化 + 人气排名 20~100 叠加条件, 回测 1.1s 跑通,
交易数 29→22 证明过滤真实生效; 新增 7 个单测 (对齐/NaN/拦截/叠加联动/
端到端正反例), 全量回归 3131 通过。

Signed-off-by: shy3130 <shy3130@qq.com>
This commit is contained in:
shy3130
2026-10-25 20:14:17 +08:00
parent da9dbca422
commit 62c8a8bd56
4 changed files with 393 additions and 0 deletions
+25
View File
@@ -520,6 +520,22 @@ class BacktestEngine:
| set(feature_plan.instrument_columns)
| set(feature_plan.matrix_columns)
)
# 扩展时序数值字段不进 parquet 解析 (enriched 只落 14 存储列), 矩阵构建后
# 按 (日期, 标的) 对齐现场注入 — 与财务因子 attach 同构。快照模式/字符串
# 字段/未注册名在此拦截 (快照回看历史 = 未来函数)。
from app.factors.ext_factors import numeric_timeseries_ext_columns
ext_wanted = frozenset(c for c in field_columns if c.startswith("ext_"))
if ext_wanted:
invalid_ext = ext_wanted - numeric_timeseries_ext_columns(
self.repo.store.data_dir
)
if invalid_ext:
raise ValueError(
f"扩展字段不可用于矩阵回测 (需为时序模式的数值扩展字段): "
f"{sorted(invalid_ext)}"
)
field_columns = field_columns - ext_wanted
cache_root = (
self.repo.store.data_dir / ".backtest_matrix_cache"
if settings.backtest_matrix_disk_cache_enabled
@@ -530,6 +546,7 @@ class BacktestEngine:
if cache_profile is not None
else frozenset(field_columns)
)
cache_fields = frozenset(cache_fields) - ext_wanted
cache_max_bytes = (
cache_profile.max_disk_bytes
if cache_profile is not None
@@ -575,6 +592,14 @@ class BacktestEngine:
self.repo.store.data_dir if self.repo is not None else None,
fundamental_names,
)
if ext_wanted:
from app.factors.ext_factors import attach_matrix_ext_fields
market = attach_matrix_ext_fields(
market,
self.repo.store.data_dir if self.repo is not None else None,
ext_wanted,
)
return market
except EnrichedGenerationUnavailableError:
if attempt + 1 >= attempts:
+66
View File
@@ -28,6 +28,7 @@ import contextlib
import logging
import re
from pathlib import Path
from typing import Any
import polars as pl
@@ -76,6 +77,21 @@ def _numeric_fields(config) -> list:
return [f for f in config.fields if f.dtype in _NUMERIC_DTYPES]
def numeric_timeseries_ext_columns(data_dir: Path | None = None) -> frozenset[str]:
"""时序模式数值扩展字段的列名集合 — 回测矩阵可注入的白名单。
快照模式 (mode=snapshot) 刻意排除: 快照是"最新值", 回看历史注入即未来函数,
帧路径的 include_snapshot=False 同理只放行时序。
"""
root = _resolve_dir(data_dir)
return frozenset(
ext_column_name(cfg.id, f.name)
for cfg in _load_configs(root)
if cfg.mode == "timeseries"
for f in _numeric_fields(cfg)
)
def _signal_fields(config) -> list:
"""帧 join / 信号条件可用的字段 (数值 + 字符串)。"""
return [f for f in config.fields if f.dtype in _SIGNAL_DTYPES]
@@ -326,6 +342,56 @@ def attach_ext_columns(
return df
def attach_matrix_ext_fields(market: Any, data_dir: Path | None, names: Any) -> Any:
"""把时序扩展数值列作为 matrix fields 附加到 (frozen) MarketDataMatrix 副本。
回测矩阵只读 enriched parquet 的 14 个存储列 (扩展列按设计不落盘, 见
pipeline._select_storage_cols); 扩展数据在这里按矩阵自身的 (日期×标的)
骨架现场 join — 与策略页帧路径同一台 attach_ext_columns (PIT 安全:
include_snapshot=False, 只进时序模式)。left join 保持骨架行序, 直接
reshape 为 (T, A); 缺失 (标的, 日期) 为 NaN, 与叠加条件掩码"NaN 不命中"
及帧路径 fill_null(False) 同语义。财务因子的 attach_matrix_fundamental_fields
是同构先例。
"""
import dataclasses
from types import MappingProxyType
import numpy as np
wanted = sorted({str(n) for n in names})
if not wanted:
return market
labels = list(market.timestamp_labels)
symbols = list(market.symbols)
n_t, n_a = len(labels), len(symbols)
if n_t == 0 or n_a == 0:
return market
skeleton = pl.DataFrame({
"date": np.repeat(np.array(labels, dtype=object), n_a),
"symbol": np.tile(np.array(symbols, dtype=object), n_t),
})
enriched = attach_ext_columns(skeleton, include_snapshot=False, data_dir=data_dir)
missing = [c for c in wanted if c not in enriched.columns]
if missing:
raise ValueError(
f"扩展字段数据不可用: {missing} — 检查扩展数据是否已拉取、字段是否仍存在"
)
extra: dict[str, np.ndarray] = {}
for name in wanted:
arr = (
enriched[name]
.cast(pl.Float64)
.fill_null(float("nan"))
.to_numpy()
.astype(np.float32)
)
arr = arr.reshape(n_t, n_a)
arr.flags.writeable = False
extra[name] = arr
merged = {**dict(market.fields), **extra}
return dataclasses.replace(market, fields=MappingProxyType(merged))
def invalidate_ext_caches(data_dir: Path | None = None, *, keep_strategy_cache: bool = False) -> None:
"""扩展数据/配置变更后的失效入口 (写入端自动调用)。
@@ -0,0 +1,295 @@
"""矩阵回测的扩展数据字段注入 — 对齐 / 缺失补空 / 模式拦截 / 叠加条件联动。
背景: enriched parquet 只落 14 个存储列 (扩展列按设计不落盘), 扩展时序数值
字段由引擎在矩阵加载后按 (日期, 标的) 骨架现场 join (attach_matrix_ext_fields),
与财务因子 attach_matrix_fundamental_fields 同构。设计对齐 docs/features.md
「第三方数据接入」与 strategy.py 叠加条件矩阵列的注入约定。
"""
from __future__ import annotations
import json
from datetime import date
from pathlib import Path
from types import SimpleNamespace
import numpy as np
import polars as pl
import pytest
from app.backtest.engine import BacktestEngine
from app.factors.ext_factors import (
numeric_timeseries_ext_columns,
)
from app.tickflow.repository import DataStore, KlineRepository
SYMS = ["600001.SH", "600002.SH"]
DATES = [date(2026, 1, 5), date(2026, 1, 6), date(2026, 1, 7), date(2026, 1, 8)]
EXT_COL = "ext_ext_demo_rank" # ext_column_name("ext_demo", "rank")
SNAP_COL = "ext_ext_snapdemo_heat" # 快照模式数值列 — 必须被拒
STR_COL = "ext_ext_demo_sname" # 字符串列 — 必须被拒
def _write_enriched(data_dir: Path) -> None:
for offset, d in enumerate(DATES):
rows = []
for i, sym in enumerate(SYMS):
close = 10.0 + i + offset
rows.append({
"symbol": sym, "date": d,
"open": close, "high": close * 1.01, "low": close * 0.99, "close": close,
"volume": 1000.0, "amount": close * 100000.0,
"raw_close": close, "raw_high": close * 1.01, "raw_low": close * 0.99,
"turnover_rate": 1.0 + i,
"consecutive_limit_ups": 0, "consecutive_limit_downs": 0,
})
part = data_dir / "kline_daily_enriched" / f"date={d.isoformat()}"
part.mkdir(parents=True)
pl.DataFrame(rows).write_parquet(part / "part.parquet")
inst = data_dir / "instruments"
inst.mkdir(parents=True)
pl.DataFrame({
"symbol": SYMS, "name": ["甲", "乙"],
"total_shares": [1e9, 1e9], "float_shares": [1e9, 1e9],
}).write_parquet(inst / "part.parquet")
def _write_ext_config(data_dir: Path, config_id: str, mode: str, fields: list[dict]) -> None:
cfg_dir = data_dir / "ext_data" / config_id
cfg_dir.mkdir(parents=True)
(cfg_dir / "config.json").write_text(json.dumps({
"id": config_id, "label": config_id, "mode": mode, "fields": fields,
"symbol_map": {"type": "mapped", "col": "symbol"},
}, ensure_ascii=False), encoding="utf-8")
def _write_ext_timeseries(data_dir: Path, rows: list[dict]) -> None:
"""rows: [{symbol, date, rank, sname}] — 缺行即缺失组合 (期望 NaN)。"""
by_date: dict[str, list[dict]] = {}
for r in rows:
by_date.setdefault(r["date"], []).append(r)
ts = data_dir / "ext_data" / "ext_demo" / "timeseries"
ts.mkdir(parents=True)
for d, day_rows in by_date.items():
part = ts / f"date={d}"
part.mkdir(parents=True)
pl.DataFrame(day_rows).write_parquet(part / "part.parquet")
@pytest.fixture()
def data_dir(tmp_path: Path) -> Path:
_write_enriched(tmp_path)
_write_ext_config(tmp_path, "ext_demo", "timeseries", [
{"name": "rank", "dtype": "int", "label": "排名"},
{"name": "sname", "dtype": "string", "label": "名称"},
])
_write_ext_config(tmp_path, "ext_snapdemo", "snapshot", [
{"name": "heat", "dtype": "float", "label": "热度"},
])
# 数据布局: 600001 全 4 天有排名; 600002 只有前 2 天 (后 2 天缺失 → NaN);
# 1/5 顺带带字符串列验证只注入数值。
rows = []
for k, d in enumerate(DATES):
# 600001: 首日 15 (区间外), 其余 50 (20<50<=100 命中叠加区间)
rows.append({"symbol": SYMS[0], "date": d.isoformat(), "rank": 15 if k == 0 else 50, "sname": "甲"})
for d in DATES[:2]:
rows.append({"symbol": SYMS[1], "date": d.isoformat(), "rank": 150, "sname": "乙"})
_write_ext_timeseries(tmp_path, rows)
return tmp_path
def _engine(data_dir: Path) -> BacktestEngine:
return BacktestEngine(KlineRepository(DataStore(data_dir)))
def _plan(matrix_extra: frozenset[str]) -> SimpleNamespace:
return SimpleNamespace(
execution_backend="matrix_native",
base_columns=frozenset({"turnover_rate"}),
instrument_columns=frozenset(),
matrix_columns=matrix_extra,
fundamental_columns=frozenset(),
)
def test_numeric_timeseries_columns_filter(data_dir: Path):
cols = numeric_timeseries_ext_columns(data_dir)
assert EXT_COL in cols
assert SNAP_COL not in cols # 快照模式排除 (回看历史 = 未来函数)
assert STR_COL not in cols # 字符串列不进矩阵
def test_matrix_load_attaches_ext_fields_aligned(data_dir: Path):
engine = _engine(data_dir)
market = engine.load_market_data_matrix_for_backtest(
None, DATES[0], DATES[-1], _plan(frozenset({EXT_COL})),
)
assert EXT_COL in market.fields
arr = np.asarray(market.fields[EXT_COL])
assert arr.shape == (len(DATES), len(SYMS))
labels = list(market.timestamp_labels)
syms = list(market.symbols)
i0, i1 = labels.index(DATES[0].isoformat()), syms.index(SYMS[0])
j1 = syms.index(SYMS[1])
assert arr[i0, i0 - i0] == pytest.approx(15.0) # 600001 首日排名 15
assert arr[labels.index(DATES[-1].isoformat()), j1] != arr[labels.index(DATES[-1].isoformat()), j1] # 600002 末日缺失 → NaN
assert arr[labels.index(DATES[1].isoformat()), j1] == pytest.approx(150.0)
# 基础列照常来自 parquet
assert "turnover_rate" in market.fields
# 数组只读 (frozen 副本约定, 同财务因子)
with pytest.raises(ValueError):
arr[0, 0] = 1.0
def test_matrix_load_rejects_snapshot_or_unknown_ext(data_dir: Path):
engine = _engine(data_dir)
with pytest.raises(ValueError, match=r"不可用于矩阵回测"):
engine.load_market_data_matrix_for_backtest(
None, DATES[0], DATES[-1], _plan(frozenset({SNAP_COL})),
)
with pytest.raises(ValueError, match=r"不可用于矩阵回测"):
engine.load_market_data_matrix_for_backtest(
None, DATES[0], DATES[-1], _plan(frozenset({"ext_ext_demo_ranks"})), # 拼错的名
)
def test_overlay_mask_on_ext_field(data_dir: Path):
"""端到端语义: 叠加条件在注入后的 ext 字段上评估, NaN 不命中。"""
from app.backtest.strategy import StrategyBacktestService
engine = _engine(data_dir)
market = engine.load_market_data_matrix_for_backtest(
None, DATES[0], DATES[-1], _plan(frozenset({EXT_COL})),
)
mask, err = StrategyBacktestService._overlay_matrix_mask(market, [
{"left": EXT_COL, "op": ">", "right": "20", "leftDays": 0, "rightDays": 0},
{"left": EXT_COL, "op": "<=", "right": "100", "leftDays": 0, "rightDays": 0},
])
assert err is None and mask is not None
arr = np.asarray(market.fields[EXT_COL])
# 掩码与数组逐元素一致: 15 命中 (20<15? 不 — 15>20 为假!)
np.testing.assert_array_equal(mask, (arr > 20) & (arr <= 100))
# 正命中样本: 600001 第 2~4 天 (rank=50) 命中, 首日 15 与 600002 全部不命中
assert mask[1, 0] and mask[2, 0] and mask[3, 0]
assert not mask[0, 0] and not mask.all(axis=0)[1]
def test_attach_requires_existing_data(data_dir: Path):
"""配置存在但数据全空 → 明确报错而非静默全 NaN (fail-closed)。"""
empty_dir = data_dir / "ext_data" / "ext_empty"
empty_dir.mkdir(parents=True)
(empty_dir / "config.json").write_text(json.dumps({
"id": "ext_empty", "label": "e", "mode": "timeseries",
"fields": [{"name": "v", "dtype": "float"}],
"symbol_map": {"type": "mapped", "col": "symbol"},
}), encoding="utf-8")
engine = _engine(data_dir)
with pytest.raises(ValueError, match=r"扩展字段数据不可用"):
engine.load_market_data_matrix_for_backtest(
None, DATES[0], DATES[-1], _plan(frozenset({"ext_ext_empty_v"})),
)
# ---------------------------------------------------------------
# 端到端: 矩阵策略 + 扩展叠加条件真实回测 (回归: SignalMatrix 只读修复)
# ---------------------------------------------------------------
_ALWAYS_ENTRY_STRATEGY = '''import numpy as np
from app.backtest.matrix import make_signal_matrix
META = {
"id": "ext_overlay_probe",
"name": "ext叠加探针",
"asset_types": ["stock"],
"timeframes": ["1d"],
"params": [],
"scoring": {},
"required_features": ["open", "high", "low", "close", "volume"],
}
EXECUTION_BACKEND = "matrix_native"
ENTRY_SIGNALS = []
EXIT_SIGNALS = []
STOP_LOSS = None
MAX_HOLD_DAYS = 1
class AlwaysEntry:
def required_fields(self):
return frozenset({"open", "high", "low", "close", "volume"})
def required_warmup_bars(self, params):
return 1
def compute_signals(self, market, params):
entry = np.ones(market.shape, dtype=np.uint8)
return make_signal_matrix(market.shape, entry=entry)
MATRIX_STRATEGY = AlwaysEntry()
'''
def _write_strategy(data_dir: Path) -> None:
sd = data_dir / "strategies" / "custom"
sd.mkdir(parents=True, exist_ok=True)
(sd / "ext_overlay_probe.py").write_text(_ALWAYS_ENTRY_STRATEGY, encoding="utf-8")
def _service_for(data_dir: Path):
from app.backtest.strategy import StrategyBacktestService, StrategyEngine
_write_strategy(data_dir)
engine = StrategyEngine(strategy_dirs=[data_dir / "strategies" / "custom"])
return StrategyBacktestService(BacktestEngine(KlineRepository(DataStore(data_dir))), engine)
def _ext_overlay_conditions() -> list[dict]:
return [{"left": EXT_COL, "op": "<=", "right": "100", "leftDays": 0, "rightDays": 0}]
def test_matrix_backtest_runs_with_ext_overlay(data_dir: Path):
"""扩展叠加条件端到端: 全命中版本正常出信号 (回归 SignalMatrix 只读修复)。"""
from app.backtest.strategy import StrategyBacktestConfig
service = _service_for(data_dir)
res = service.run(StrategyBacktestConfig(
strategy_id="ext_overlay_probe", symbols=None,
start=DATES[0], end=DATES[-1],
overrides={
"basic_filter": {"enabled": False}, # 合成小盘股过不了默认市值过滤
"overlay_filter": _ext_overlay_conditions(),
},
matching="close_t", fees_pct=0, slippage_bps=0, max_positions=2,
))
assert res.error is None, res.error
assert (res.stats.get("n_trades") or 0) > 0
def test_matrix_backtest_ext_overlay_can_block_all(data_data_dir=None):
"""反证过滤真实生效: 永不满足的叠加条件 → 无买入信号。"""
import tempfile
from app.backtest.strategy import StrategyBacktestConfig
with tempfile.TemporaryDirectory() as td:
data_dir = Path(td)
_write_enriched(data_dir)
_write_ext_config(data_dir, "ext_demo", "timeseries", [
{"name": "rank", "dtype": "int", "label": "排名"},
{"name": "sname", "dtype": "string", "label": "名称"},
])
rows = [
{"symbol": SYMS[0], "date": d.isoformat(), "rank": 15, "sname": "甲"}
for d in DATES
] + [
{"symbol": SYMS[1], "date": d.isoformat(), "rank": 150, "sname": "乙"}
for d in DATES[:2]
]
_write_ext_timeseries(data_dir, rows)
service = _service_for(data_dir)
res = service.run(StrategyBacktestConfig(
strategy_id="ext_overlay_probe", symbols=None,
start=DATES[0], end=DATES[-1],
overrides={
"basic_filter": {"enabled": False},
"overlay_filter": [
{"left": EXT_COL, "op": ">", "right": "1000", "leftDays": 0, "rightDays": 0},
],
},
matching="close_t", fees_pct=0, slippage_bps=0, max_positions=2,
))
assert res.error is not None and "未产生买入信号" in res.error
+7
View File
@@ -60,6 +60,10 @@
**结果沉淀**:回测结果可**导出 CSV**(概要 / 净值曲线 / 交易明细 / 分标的统计四段),并可**保存候选**(完整参数与指标快照);候选在回测页「候选方案」中查看,支持**一键载入复测**,回填全部参数继续研究。
**稳健参数筛选(验证 tab)**:参数不再按单窗样本内得分排名——每组参数在 K 个互不重叠验证段上独立回测,按稳健分 R = 0.6×中位 + 0.4×最差 排名,只有在多段行情上都好的参数才能当冠军。配套:主目标白名单(排除胜率/持仓天数等退化目标)、K 段合计最小交易数门槛(fail-closed,不足者降序垫底不静默剔除)、Top-20 全窗 IS 细化(样本内乐观偏差/退化率)、冠军收敛度诊断(各段冠军是否收敛于同一组)、高原判据(冠军网格邻域得分衰减,孤峰=过拟合警报)、运行前验证窗预览。冠军参数可一键应用(走策略配置 override,策略页可还原)。旧单窗样本内排名路径已移除。
**滚动样本外检验(验证 tab)**:每折在训练区间网格优化选参、紧邻测试区间做样本外(OOS)验证,滚动前移;核心产出是 OOS 拼接复利净值与 IS→OOS 退化(过拟合信号)。补强:每折 OOS 最小交易数门槛(不足标记跳过不混入复利)、运行前折预览(<3 折告警)、参数漂移轨迹(逐折最优值 + 稳定/漂移徽章)、「采用最新折参数」(确认框注明来源折,走策略配置 override)。
**因子与策略挖掘**:复用已有日频因子和 matrix-native 策略,通过 T-1 市场环境、相关去重和嵌套样本外验证生成研究候选。任务在 spawn worker 中运行,支持持久 run ID、取消、刷新重连和显式发布;自动周度任务默认关闭且永不自动发布。完整口径见 [因子与策略挖掘](./mining.md)。
**策略设置统一**:回测页「高级策略设置」抽屉与策略页设置弹窗采用同一信息架构(策略条件 / 策略权重 / 风控触发 / 回测范围)。风控与持仓纪律(止损 · 止盈 · 移动止损 · 回撤止盈 · 最长持仓)以策略设置保存的值为默认,回测抽屉内可临时覆盖,也可「应用到策略」持久化;策略页保存的叠加条件随回测请求继承,与扫描 / 监控同一口径;回测范围(资产类型 / 股票池)为回测专属概念。
@@ -181,6 +185,9 @@
详见 [custom-data-source.md](./custom-data-source.md) 与 [plugin-development.md](./plugin-development.md)。
### 扩展数据进回测
数值型**时序模式**扩展字段(人气排名等)可在回测中使用:叠加条件 / 策略评分引用 `ext_*` 字段时,引擎在矩阵加载后按 (日期, 标的) 对齐现场注入(与策略页帧路径同一台 PIT 安全 join,缺失补空值=条件不命中)。快照模式扩展字段与字符串扩展字段不进回测(快照回看历史即未来函数);引用时回测前置校验直接拒绝并说明原因。
### 全量分钟(盘中增量落盘)
「全量分钟」是独立能力(`full_minute`),**与其他能力同样可路由**:盘中将全市场当日 1 分钟 K 持续增量落盘 `data/kline_minute/`,分钟策略与分时视图即取即用。接入方式:TickFlow Expert,或声明 `full_minute` 数据集的插件/自定义源(设置 → 数据源 → 全量分钟 路由);配好能力后开启实时行情即自动运行(冷启动全天修复轮 + 稳态增量轮)。