导出入选的样本外预测，避免验证数据泄漏
代码 《交易机器学习》
总结
此工具为实时演示笔记本准备预测产物，将不同的列名和日期格式规范为统一的时间戳、资产和预测结构。它会对重复的时间戳—资产观测分组，并计算其评分均值。加载器先搜索已提交的模型输出，找不到这些文件时再查询注册表。
注册表回退路径优先选择在已记录的信号、配置和风险叠加层回测中验证夏普比率最高的模型运行，并导出该次运行的样本外预测。这样导出的预测与用于选择配置的验证数据分开，降低在模型选择样本上评估入选模型的风险。如果没有符合条件的回测优胜者，回退路径会按请求期限的平均信息系数选择一组样本外预测。这是实用的恢复路径，但采用了不同的选择标准；此代码只负责加载产物，并不能证明任何入选模型在样本外表现良好。
核心观点
- 将预测列和时间戳规范化后，再把产物传给下游笔记本。
- 验证数据参与模型选择时，外部评估应优先使用样本外预测。
- 主要注册表路径依据指定回测阶段的验证夏普比率选择训练运行。
- 没有符合条件的回测阶段记录时，可按信息系数排名选择回退结果。
- 已提交文件和符合条件的注册表预测都不可用时，加载器不返回任何产物。
标签
全文
# demo_artifacts.py
```py
from __future__ import annotations
import polars as pl
from utils.paths import get_case_study_source_dir
def normalize_demo_predictions(df: pl.DataFrame, asset_column: str) -> pl.DataFrame:
"""Normalize prediction artifacts to ``timestamp``, ``asset_column``, ``prediction``."""
rename_map = {}
# Normalize entity column: accept "asset" or "symbol" as source
if asset_column not in df.columns:
if "asset" in df.columns:
rename_map["asset"] = asset_column
elif "symbol" in df.columns:
rename_map["symbol"] = asset_column
if "prediction" not in df.columns and "y_score" in df.columns:
rename_map["y_score"] = "prediction"
# Normalize legacy "date" column to canonical "timestamp"
if "date" in df.columns and "timestamp" not in df.columns:
rename_map["date"] = "timestamp"
if rename_map:
df = df.rename(rename_map)
required = {"timestamp", asset_column, "prediction"}
if required - set(df.columns):
raise ValueError(
f"Prediction artifact missing required columns: {required - set(df.columns)}"
)
ts = df["timestamp"]
if ts.dtype == pl.Utf8:
df = df.with_columns(pl.col("timestamp").str.to_date())
elif ts.dtype in (pl.Datetime, pl.Date):
df = df.with_columns(pl.col("timestamp").dt.date())
return (
df.select(["timestamp", asset_column, "prediction"])
.group_by(["timestamp", asset_column])
.agg(pl.col("prediction").mean().alias("prediction"))
.sort([asset_column, "timestamp"])
)
def load_demo_predictions(strategy_id: str, horizon: int, asset_column: str) -> pl.DataFrame | None:
"""Load prediction artifacts for live-demo notebooks.
Searches committed model directories first, then falls back to the
content-addressed registry (run_log/predictions/) which is the
primary output of the model training pipeline. The registry fallback
returns the sealed **holdout** split — the once-touched out-of-sample
set — never validation, so a deployment export never ships predictions
from the data used to select the model.
"""
import os
import sqlite3
from pathlib import Path
source_dir = get_case_study_source_dir(strategy_id)
models_dir = source_dir / "models"
candidates = [
models_dir / "gbm" / f"fwd_ret_{horizon}d" / "predictions.parquet",
models_dir / "linear" / f"fwd_ret_{horizon}d" / "predictions.parquet",
models_dir / "deep_learning" / f"fwd_ret_{horizon}d" / "predictions.parquet",
models_dir / "tabular_dl" / f"fwd_ret_{horizon}d" / "predictions.parquet",
]
# Also check seeded predictions in ML4T_OUTPUT_DIR (CI / test mode)
output_dir = os.environ.get("ML4T_OUTPUT_DIR", "")
if output_dir:
seeded_dir = Path(output_dir) / strategy_id / "models"
candidates.append(seeded_dir / f"predictions_reg_{horizon}d.parquet")
for path in candidates:
if not path.exists():
continue
df = pl.read_parquet(path)
has_time = "date" in df.columns or "timestamp" in df.columns
has_entity = "asset" in df.columns or "symbol" in df.columns
has_score = "y_score" in df.columns or "prediction" in df.columns
if not (has_time and has_entity and has_score):
continue
return normalize_demo_predictions(df, asset_column)
# Fall back to registry — export the sealed HOLDOUT prediction set, never
# validation. A deployment bridge that exported validation predictions would
# be backtesting on the data used to select the model (leakage). The holdout
# is the once-touched out-of-sample set, so it is the only honest thing to
# ship to an external backtester.
#
# Pick that holdout by the *selected winner*, not by raw IC: the selected configuration is
# the cross-stage validation-Sharpe rank-1 config, pooling the selection
# stages (signal/allocation/risk_overlay). cost_sensitivity is a
# perturbation, not a selection axis, and the holdout stage is the sealed
# set itself, so both are excluded. The winner's training run owns exactly
# one holdout prediction set (the "one holdout per case study" rule), which
# is what we export. This also derives the label/horizon from the winner
# rather than trusting the caller's `horizon`. The IC-sorted query is kept
# only as a fallback for case studies that have predictions but no backtest
# stages recorded.
registry_db = source_dir / "run_log" / "registry.db"
if registry_db.exists():
conn = sqlite3.connect(str(registry_db))
pred_hash = None
winner = conn.execute(
"""SELECT ps.training_hash
FROM backtest_runs br
JOIN backtest_metrics bm ON br.backtest_hash = bm.backtest_hash
JOIN prediction_sets ps ON br.prediction_hash = ps.prediction_hash
WHERE br.stage IN ('signal', 'allocation', 'risk_overlay')
ORDER BY bm.sharpe DESC LIMIT 1""",
).fetchone()
if winner:
row = conn.execute(
"""SELECT prediction_hash FROM prediction_sets
WHERE training_hash = ? AND split = 'holdout'
ORDER BY prediction_hash LIMIT 1""",
(winner[0],),
).fetchone()
if row:
pred_hash = row[0]
if pred_hash is None:
label = f"fwd_ret_{horizon}d"
row = conn.execute(
"""SELECT ps.prediction_hash
FROM prediction_sets ps
JOIN training_runs tr ON ps.training_hash = tr.training_hash
JOIN prediction_metrics pm ON ps.prediction_hash = pm.prediction_hash
WHERE tr.label = ? AND ps.split = 'holdout'
ORDER BY pm.ic_mean DESC LIMIT 1""",
(label,),
).fetchone()
if row:
pred_hash = row[0]
conn.close()
if pred_hash:
pred_path = source_dir / "run_log" / "predictions" / pred_hash / "predictions.parquet"
if pred_path.exists():
df = pl.read_parquet(pred_path)
return normalize_demo_predictions(df, asset_column)
return None
```在遵守原作品许可的前提下，附作者信息全文展示。 许可协议: MIT
此摘要由 Stratmill 研究智能体根据原文撰写，并非原文副本。