【Python量化选股流水线 #05】过滤阈值拍脑袋？窗口与阈值的敏感度实验
> 回看验收发现选股器整体跑输，第一反应往往是「换因子」——但很多时候问题不在因子本身，而在参数：窗口太大还是太小？Top-N 取 3 还是取 7？动量权重给 0.4 合适吗？这篇教你用参数扫描的方法，量化地知道每个参数对结果的影响方向与幅度。 ## 一、为什么参数比因子更重要（在你还没有海量数据之前） 因子研究需要长时间跨度的真实数据来验证有效性；但在数据有限的情况下，**先把现有因子的参数调到一个相对合理的区间**，比盲目堆新因子更有工程价值。 参数敏感度实验的核心思路：固定因子定义，系统性地改变窗口大小、候选数量和权重配比，用同一段历史数据跑一遍，对比平均超额。最终产出的不是「一个稳赚组合」，而是「参数-结果」的映射矩阵——让你知道往哪个方向调参数是有效的。 **本文你将得到：** 1. 二维参数扫描框架（窗口 × Top-N），输出敏感度矩阵； 2. 权重扫描框架（动量权重 0.1~0.9），找到当前样本下的相对最优配比； 3. 可直接落地的结论格式，方便写入你的策略配置文件。 ## 二、准备环境 ```bash pip install mairui pandas numpy ``` ## 三、完整代码（已实测跑通） ```python from mairui import Client import random import pandas as pd import numpy as np LICENCE = "LICENCE-66D8-9F96-0C7F0FBCD073" # 演示证书，请换成你自己的正式证书 SEED = 20260918 POOL = { "600519": "贵州茅台", "000858": "五粮液", "601398": "工商银行", "600036": "招商银行", "300750": "宁德时代", "002594": "比亚迪", "600276": "恒瑞医药", "603259": "药明康德", "600030": "中信证券", "601688": "华泰证券", "000333": "美的集团", "600900": "长江电力", } WEIGHTS = {"动量": 0.4, "低波": 0.3, "流动性": 0.3} WINDOWS = [15, 21, 30] TOP_NS = [2, 3, 5, 7] def perturb(bars, seed): rng = random.Random(seed) out = [] for b in bars: k = 1.0 + (rng.random() - 0.5) * 0.10 kv = 1.0 + (rng.random() - 0.5) * 0.40 nb = dict(b) for f in ("o", "c", "h", "l"): if f in nb and nb[f] is not None: nb[f] = round(float(nb[f]) * k, 2) if nb.get("v") is not None: nb["v"] = round(float(nb["v"]) * kv) out.append(nb) return out def fetch_bars(code): bars = Client(LICENCE).stock_history(code=code, st="", et="", period="d", dividend="f") if not bars: raise RuntimeError(f"{code} 未获取到数据，请检查证书是否有效") return perturb(bars, seed=SEED + int(code)) def to_frame(bars): df = pd.DataFrame(bars) for col in ("o", "h", "l", "c", "v"): df[col] = pd.to_numeric(df[col], errors="coerce") df["t"] = df["t"].astype(str) return df.sort_values("t").reset_index(drop=True) def factor_scores(df, w): c, v = df["c"], df["v"] mom = c.iloc[-1] / c.iloc[0] - 1 lowvol = -c.pct_change().tail(min(20, len(df) - 1)).std() liq = (v * c).tail(min(20, len(df) - 1)).mean() return {"动量": mom, "低波": lowvol, "流动性": liq} def zscore(vals): a = np.asarray(vals, dtype=float) sd = a.std() return (a - a.mean()) / sd if sd > 0 else np.zeros_like(a) def avg_return_of_picks(frames, day, window, top_n, weights): rows = [] for code, df in frames.items(): hist = df.iloc[day - window:day] rows.append({"代码": code, **factor_scores(hist, window)}) table = pd.DataFrame(rows) for k in weights: table[k + "_z"] = zscore(table[k]) table["综合分"] = sum(table[k + "_z"] * w for k, w in weights.items()) picks = table.sort_values("综合分", ascending=False).head(top_n)["代码"].tolist() rets = [] for code in picks: rets.append(frames[code]["c"].iloc[day] / frames[code]["c"].iloc[day - 1] - 1) return float(np.mean(rets)) def main(): frames = {code: to_frame(fetch_bars(code)) for code in POOL} n = min(len(df) for df in frames.values()) print(f"每只标的 {n} 条日线；扫描窗口 {WINDOWS} × 候选数 {TOP_NS}\n") all_pool_ret = [] for day in range(max(WINDOWS), n): r = np.mean([frames[c]["c"].iloc[day] / frames[c]["c"].iloc[day - 1] - 1 for c in frames]) all_pool_ret.append(r) bench = float(np.mean(all_pool_ret)) print(f"全池基准平均次日涨幅: {bench * 100:+.4f}%\n") results = [] for window in WINDOWS: days = list(range(window, n)) row_label = f"窗口{window}({len(days)}日)" for top_n in TOP_NS: rets = [avg_return_of_picks(frames, day, window, top_n, WEIGHTS) for day in days] avg_ret = float(np.mean(rets)) results.append({"窗口": row_label, "Top-N": top_n, "候选均涨": avg_ret, "超额": avg_ret - bench}) table = pd.DataFrame(results) pivot = table.pivot(index="窗口", columns="Top-N", values="超额") print("===== 敏感度矩阵（超额 = 候选均涨 - 全池基准）=====") print((pivot * 100).round(2).astype(str) + "%") best = table.loc[table["超额"].idxmax()] print(f"\n最佳组合: 窗口{best['窗口'].split('(')[0].replace('窗口', '')}, Top-N={int(best['Top-N'])}") print(f"平均超额: {best['超额'] * 100:+.4f}%") print("\n===== 权重敏感度（固定窗口=21, Top-N=5）=====") mom_ws = [0.1, 0.3, 0.5, 0.7, 0.9] weight_rows = [] window = 21 days = list(range(window, n)) for mw in mom_ws: lw = (1 - mw) / 2 ww = {"动量": mw, "低波": lw, "流动性": lw} rets = [avg_return_of_picks(frames, day, window, 5, ww) for day in days] avg_ret = float(np.mean(rets)) weight_rows.append({"动量权重": mw, "低波权重": lw, "候选均涨": avg_ret, "超额": avg_ret - bench}) wdf = pd.DataFrame(weight_rows) print(wdf.to_string(index=False)) best_w = wdf.loc[wdf["超额"].idxmax()] print(f"\n最佳权重: 动量={best_w['动量权重']}, 低波={best_w['低波权重']:.2f}, 超额={best_w['超额'] * 100:+.4f}%") if __name__ == "__main__": main() ``` ## 四、真实运行结果 ```text 全池基准平均次日涨幅: +0.1828% ===== 敏感度矩阵（超额 = 候选均涨 - 全池基准）===== Top-N 2 3 5 7 窗口 窗口15(35日) -2.49% -2.18% -1.44% -0.86% 窗口21(29日) -2.19% -1.83% -0.96% -0.83% 窗口30(20日) -1.0% -1.21% -1.19% -0.98% 最佳组合: 窗口21, Top-N=7 平均超额: -0.8342% ===== 权重敏感度（固定窗口=21, Top-N=5）===== 动量权重 低波权重 候选均涨 超额 0.1 0.45 -0.001501 -0.003328 0.3 0.35 -0.004194 -0.006022 0.5 0.25 -0.011103 -0.012930 0.7 0.15 -0.014869 -0.016696 0.9 0.05 -0.014281 -0.016108 最佳权重: 动量=0.1, 低波=0.45, 超额=-0.3328% ``` 从矩阵里能读出三个明确方向： - **Top-N 越大，亏损越小**：Top-N=7 在所有窗口下都优于 Top-N=2。这说明在这个样本上，「多买几只摊薄」比「精选少数」更稳妥——可能是因为单因子选出来的前几名波动性大。 - **动量权重越低越好**：从 0.9 降到 0.1，超额从 -1.61% 改善到 -0.33%。这说明当前样本期内动量因子反而是个反向指标，降低它的权重、抬高低波和流动性的权重，能显著减少亏损。 - **窗口 30 并没有明显优于 21**：在数据量有限的情况下，窗口加大反而减少了滚动回看天数（30 窗口只剩 20 个回看日），样本可信度下降。 > 说明：演示证书返回固定样例数据，代码用确定性扰动模拟行情起伏；换成你自己的正式证书后删掉 `perturb` 调用即为真实行情，扫描逻辑零改动。 ## 五、两个实操建议 1. **敏感度实验的价值不是「找到稳赚参数」**：在有限样本上，所有参数都可能跑输基准。实验的真正价值是告诉你「往哪个方向调是有效的」——比如这里明确告诉你「降低动量权重、增加 Top-N」是改善方向。 2. **把最佳参数写入配置文件，定期重跑**：策略参数不应该写死在代码里。把 `WINDOW`、`TOP_N`、`WEIGHTS` 抽成 JSON/YAML 配置，每个月用最新数据重跑一次敏感度实验，更新参数。 ## 六、小结与下篇预告 参数敏感度实验是量化工程中容易被忽略但回报极高的一步。它能让你在数据有限的情况下，快速排除明显有害的参数组合，把资源集中在真正有效的方向上。 **下篇预告**：《【Python量化选股流水线 #06】从候选清单到每日盯盘：流水线结果进监控看板》——把选股结果输出为 JSON 清单 + 文本看板 + 走势图，每天收盘后一键生成候选监控面板。 ## 免责声明 本文所有代码与数据仅用于技术学习与演示，不构成任何投资建议。股市有风险，投资需谨慎。 ## 代码与文档 SDK 与更多语言版本：https://github.com/MaiRuiApi
