研究问题与数据约定
本章目录
学完后,你应能独立完成
- 定义股票日期面板的一行、信息集和标签
- 区分训练损失、选模指标和交易目标
- 手工核对开盘到开盘的未来收益
改变一个条件,观察发生什么
浏览器内即时教学计算;Python 实验结果另列于代码区。图形与数值一起核对。
一行样本是一项预测决策
假设你在 t 日收盘后,用当时可得的价量因子,为每只股票产生一个分数。长表的一行由 信号日期 × 股票唯一确定,X 的列是因子,y 是之后实现的目标。模型参数从允许的训练样本估计;λ、树深、窗长、训练频率属于需要预先约定或在开发期选择的超参数。
1000 只股票 × 250 日不等于 25 万个独立试验。同日股票共享市场和行业冲击,相邻日期又共享状态。行数说明计算规模,日期与依赖结构决定有多少新信息。更复杂的模型不能凭行数自动获得可信的泛化能力。
把标签画在时间线上
本课程统一采用:t 收盘形成信号,t+1 开盘进入,t+k+1 开盘退出。k 表示两次开盘之间的交易日间隔。k=5 时,是 t+1 到 t+6,不是 t 到 t+5。这里的日期位置是教学交易日索引;真实数据需要交易所日历。
构建离线标签当然会用到未来价格;关键是训练时该标签是否已成熟。特征可得时点 ≤ 决策时点,标签可得时点 ≤ 训练截止时点,二者需要分别检查。发布、到达和清洗延迟也属于时点约定。
- 股票内排序,检查日期与股票键唯一。
- 按股票分别 shift(-(k+1)) 与 shift(-1),避免跨股票取未来价格。
- 保留入口、出口与实际 label_available_time;尾部缺少退出价格时标签为 NA。
回归、分类和排序回答不同问题
回归预测收益水平;分类估计上涨或超过阈值的概率;排序强调同日股票之间的先后。把未来同日收益转成 rank 可以作为标签,但它不允许出现在当时输入中。排序的 query 通常是一个日期,而不是把所有年份的股票混成一组。
绝对收益、市场超额、行业超额、波动调整收益会改变目标。波动调整的分母若用于输入处理,应在决策时可得。目标不能在查看最终测试表现后悄悄换成效果更好的版本。
训练好、排名好、交易好是三种判断
MSE 衡量收益水平误差,逐日 IC/Rank IC 衡量截面相关或次序,交易表现还依赖持仓、成交、换手、费用和风险约束。一个正向缩放后的分数可能保留排序,但 MSE 明显变化。
五日 top-bottom 标签差是预测诊断。每天计算的五日标签彼此重叠,不能把它们直接作为每日收益连乘净值。现实长仓选股还需单独检查多空两端、行业/规模暴露和不能成交的股票。
先约定能复核的研究合同
在模型之前写清:历史股票资格、数据可得性、标签、训练截止、滚动政策、开发折、候选空间、选择指标、最终测试与评价集合。未知字段先标未知,不能用模型名称填补。
合理基线至少包括预先规定的单因子、有依据的等权和 Ridge。各模型可以使用明确声明的特征子集,但应基于同一允许信息集、同一标签与同一评价集合。后面的完整案例会落实这些约定。
从算法到一次真实运行
代码在本地 Python 中执行,HTML 内展示参考实现和本轮运行结果。请先独立完成练习,再展开完整参考文件。
def make_panel(seed=42, T=180, N=24, k=5, mechanism="linear"):
"""Illustrative weekdays, 6 AR factors, future open-to-open k-day label."""
if mechanism not in {"null","linear","interaction","regime"}:
raise ValueError("Unknown mechanism")
rng=np.random.default_rng(seed)
dates=pd.bdate_range("2020-01-02",periods=T)
factors=np.empty((T,N,6)); factors[0]=rng.normal(size=(N,6))
for t in range(1,T):
factors[t]=.6*factors[t-1]+.8*rng.normal(size=(N,6))
factors[:,:,3]=.9*factors[:,:,0]+.15*rng.normal(size=(T,N))
prices=np.full((T,N),100.)
for t in range(T-1):
x=factors[max(0,t-1)]
signal=np.zeros(N)
if mechanism!="null": signal=.0015*x[:,0]
if mechanism in {"interaction","regime"}: signal += .0015*x[:,1]*x[:,2]
if mechanism=="regime" and t>=int(.65*T): signal -= .002*x[:,0]
noise=.006*rng.standard_t(5,N)/np.sqrt(5/3)+rng.normal(0,.002)
prices[t+1]=prices[t]*np.exp(signal+noise)
panel=pd.DataFrame(factors.reshape(T*N,6),columns=[f"x{i}" for i in range(6)])
panel["date"]=np.repeat(dates,N); panel["stock"]=np.tile([f"S{i:03}" for i in range(N)],T)
panel["tpos"]=np.repeat(np.arange(T),N); panel["open"]=prices.ravel()
panel["signal_time"]=panel.date+pd.Timedelta(hours=15)
panel["feature_available_time"]=panel.signal_time
labels=panel.groupby("stock").open
panel["y"]=labels.shift(-(k+1))/labels.shift(-1)-1
exit_dates=pd.Series(dates).shift(-(k+1)).to_numpy()[panel.tpos]
panel["label_available_time"]=exit_dates+pd.Timedelta(hours=9,minutes=30)
return panel
def validate_panel(frame):
if frame.duplicated(["date","stock"]).any():
raise ValueError("Duplicate (date,stock) keys")
if frame[["signal_time","feature_available_time"]].isna().any().any():
raise ValueError("Missing availability timestamps")
if (frame.feature_available_time>frame.signal_time).any():
raise ValueError("Feature was unavailable at decision time")
if (frame.y.notna() & frame.label_available_time.isna()).any():
raise ValueError("Known label has unknown availability time")
def time_fold(frame, start, end, train_start=0):
"""Train at prior signal day's 15:00; dates [start,end) are held out."""
validate_panel(frame)
dates=frame.date.drop_duplicates().sort_values().reset_index(drop=True)
if not 1<=start<end<=len(dates) or not 0<=train_start<start:
raise ValueError("Invalid date fold")
cutoff=dates.iloc[start-1]+pd.Timedelta(hours=15)
train=frame[(frame.date>=dates.iloc[train_start]) & (frame.date<dates.iloc[start])
& frame.y.notna() & (frame.label_available_time<=cutoff)].copy()
valid=frame[frame.date.isin(dates.iloc[start:end]) & frame.y.notna()].copy()
if train.empty or valid.empty: raise ValueError("Empty fold")
return train,valid,cutoff"""Standalone synthetic teaching example. Run locally with Python, not in the HTML."""
import json
import numpy as np
import pandas as pd
from scipy.special import expit
def clean(value):
if isinstance(value, dict): return {str(k):clean(v) for k,v in value.items()}
if isinstance(value,(list,tuple,np.ndarray)): return [clean(v) for v in value]
if isinstance(value,(np.integer,)): return int(value)
if isinstance(value,(float,np.floating)): return float(value) if np.isfinite(value) else None
if isinstance(value,(np.bool_,)): return bool(value)
return value
def print_result(result):
print(json.dumps(clean(result),ensure_ascii=False,allow_nan=False))
def make_panel(seed=42, T=180, N=24, k=5, mechanism="linear"):
"""Illustrative weekdays, 6 AR factors, future open-to-open k-day label."""
if mechanism not in {"null","linear","interaction","regime"}:
raise ValueError("Unknown mechanism")
rng=np.random.default_rng(seed)
dates=pd.bdate_range("2020-01-02",periods=T)
factors=np.empty((T,N,6)); factors[0]=rng.normal(size=(N,6))
for t in range(1,T):
factors[t]=.6*factors[t-1]+.8*rng.normal(size=(N,6))
factors[:,:,3]=.9*factors[:,:,0]+.15*rng.normal(size=(T,N))
prices=np.full((T,N),100.)
for t in range(T-1):
x=factors[max(0,t-1)]
signal=np.zeros(N)
if mechanism!="null": signal=.0015*x[:,0]
if mechanism in {"interaction","regime"}: signal += .0015*x[:,1]*x[:,2]
if mechanism=="regime" and t>=int(.65*T): signal -= .002*x[:,0]
noise=.006*rng.standard_t(5,N)/np.sqrt(5/3)+rng.normal(0,.002)
prices[t+1]=prices[t]*np.exp(signal+noise)
panel=pd.DataFrame(factors.reshape(T*N,6),columns=[f"x{i}" for i in range(6)])
panel["date"]=np.repeat(dates,N); panel["stock"]=np.tile([f"S{i:03}" for i in range(N)],T)
panel["tpos"]=np.repeat(np.arange(T),N); panel["open"]=prices.ravel()
panel["signal_time"]=panel.date+pd.Timedelta(hours=15)
panel["feature_available_time"]=panel.signal_time
labels=panel.groupby("stock").open
panel["y"]=labels.shift(-(k+1))/labels.shift(-1)-1
exit_dates=pd.Series(dates).shift(-(k+1)).to_numpy()[panel.tpos]
panel["label_available_time"]=exit_dates+pd.Timedelta(hours=9,minutes=30)
return panel
def validate_panel(frame):
if frame.duplicated(["date","stock"]).any():
raise ValueError("Duplicate (date,stock) keys")
if frame[["signal_time","feature_available_time"]].isna().any().any():
raise ValueError("Missing availability timestamps")
if (frame.feature_available_time>frame.signal_time).any():
raise ValueError("Feature was unavailable at decision time")
if (frame.y.notna() & frame.label_available_time.isna()).any():
raise ValueError("Known label has unknown availability time")
def time_fold(frame, start, end, train_start=0):
"""Train at prior signal day's 15:00; dates [start,end) are held out."""
validate_panel(frame)
dates=frame.date.drop_duplicates().sort_values().reset_index(drop=True)
if not 1<=start<end<=len(dates) or not 0<=train_start<start:
raise ValueError("Invalid date fold")
cutoff=dates.iloc[start-1]+pd.Timedelta(hours=15)
train=frame[(frame.date>=dates.iloc[train_start]) & (frame.date<dates.iloc[start])
& frame.y.notna() & (frame.label_available_time<=cutoff)].copy()
valid=frame[frame.date.isin(dates.iloc[start:end]) & frame.y.notna()].copy()
if train.empty or valid.empty: raise ValueError("Empty fold")
return train,valid,cutoff
if __name__ == "__main__":
panel = make_panel(T=20,N=3,k=5)
train, valid, cutoff = time_fold(panel,10,14)
assert sorted(train.tpos.unique()) == [0,1,2,3]
print_result({"cutoff":str(cutoff),"train_signal_positions":sorted(train.tpos.unique()),
"train_rows":len(train),"validation_rows":len(valid),
"label_head":panel.loc[panel.stock=="S000",["tpos","open","y"]].head(4).to_dict("records")})
查看本轮本地运行记录 · 2026-10-12
{
"cutoff": "2020-01-15 15:00:00",
"train_signal_positions": [
0,
1,
2,
3
],
"train_rows": 12,
"validation_rows": 12,
"label_head": [
{
"tpos": 0,
"open": 100.0,
"y": 0.004185886405229544
},
{
"tpos": 1,
"open": 100.18749552002413,
"y": 0.009503821620656216
},
{
"tpos": 2,
"open": 99.87298188241014,
"y": 0.011144839929125405
},
{
"tpos": 3,
"open": 99.99758346812865,
"y": 0.018177726537848216
}
]
}先回答,再看提示和答案
在 s−1 收盘训练,五日标签最新可用的信号位置是多少?
一级提示
先写出 entry、exit 和 cutoff。
参考答案与条件
退出位置 t+6 必须不晚于 s−1,因此 t≤s−7;前提是退出开盘标签当天已经到达。
未来收益做同日 rank 是泄漏吗?
一级提示
先写出目标、信息集和所需条件。
参考答案与条件
作为目标 y 可以;作为信号时的输入 X 不可以。训练还要等待该日所有必要目标成熟。
随机拆股票日期行的分数很稳定,是否证明未来有效?
一级提示
先写出目标、信息集和所需条件。
参考答案与条件
不能。训练与验证可能共享日期状态、共同冲击或未来信息。需要符合未来预测目的的日期验证。
把知识迁移到一个新任务
造 3 只股票 × 12 个交易日的价格表,手算 k=1 与 k=5 的三个标签;写出每个标签的可用时刻。把一只股票的缺失价格保留为 NA,再解释它如何影响评价集合。
这里保存学习笔记,不在浏览器执行 Python。A/B/C 由你按独立完成程度填写。
参考与进一步阅读
在线文档可能比本地固定版本更新;API 行为以已验证代码和版本记录为准。
掌握状态与复习
关掉资料独立完成后记 A;提示后完成记 B;需要重学记 C。
记录尽量保存在当前浏览器,建议定期导出备份。