过拟合防范：样本外验证、交叉验证与复杂度惩罚

"A model that fits the past perfectly is usually a poor predictor of the future."

这句话出自统计学泰斗 George Box 之口，却在量化交易社区被引用到几乎令人麻木。问题是：真正在构建策略时，大多数人依然在重复这个错误——用全部可用数据调参，然后惊讶于实盘结果的断崖式崩塌。

本文不打算再讲一遍"过拟合是什么"。我们假设你已经知道验证集与测试集的分离逻辑，知道 train/test split 的基本操作。我们要做的事情更具体：当你已经掌握了这些基础之后，如何系统性地判断一个策略是"真正捕捉到了市场的某种结构性规律"，还是"恰好拟合了历史数据的噪声"？

答案涉及三个层次：滚动窗口验证（样本外的时间序列感知）、信息准则与复杂度惩罚（模型层面的结构选择）、参数敏感性分析（工程层面的鲁棒性检验）。三层检验，缺一不可。

---

一、为什么普通验证集分割在量化中不够用

在图像分类或推荐系统的场景下，随机划分训练集与测试集是标准做法。但在金融时间序列中，这个操作引入了一个根本性的缺陷：未来信息泄露（look-ahead bias）。

考虑一个最简单的例子：你有 5 年的日线数据，随机取 80% 作为训练集，20% 作为测试集。训练集里可能包含了 2019 年、2021 年和 2023 年的数据，而测试集则混入了 2020 年（疫情暴跌）和 2022 年（加息周期）。两者在市场机制上高度相关，随机划分并没有真正验证策略对未来未知状态的泛化能力。

更重要的是：金融数据的非平稳性使得随机划分的"测试集表现"本质上是一个伪命题。2019 年的市场结构与 2023 年存在显著差异，但它们在时间轴上是连续的。随机划分破坏了这种时间依赖性，让你误以为验证了泛化能力，实际上只是验证了策略对相似历史环境的记忆程度。

因此，金融时间序列的验证框架必须以时间为第一约束条件。 以下两节，我们用具体数据和代码来展示两种互补的验证范式。

---

二、滚动窗口验证：时间序列感知的样本外测试

2.1 基本原理

滚动窗口验证（Rolling Window Validation）用固定长度的窗口在时间轴上顺序滑动，每次用窗口内的数据训练模型，然后用下一个固定周期的数据测试。核心思想很直接：用"过去"预测"未来"，而非用"随机片段"预测"另一个随机片段"。

设总数据长度为 $T$，窗口长度为 $W$，步长为 $S$，测试窗口长度为 $H$（holding/测试周期）。则共有 $N = \lfloor (T - W) / S \rfloor$ 个样本外测试窗口：

$$\text{Window}i = [\text{Train}: t{i \times S}, t_{i \times S + W - 1}], \quad [\text{Test}: t_{i \times S + W}, t_{i \times S + W + H - 1}]$$

滚动验证的输出不是单一数字，而是一组时间序列上的性能分布。这本身就是极其有价值的信息：如果策略在多个滚动窗口上的表现稳定，说明它捕捉到的不是某个特定时间段的偶然现象。

2.2 代码实现：完整滚动验证框架

以下代码实现了一个生产级的滚动窗口验证框架，包含以下关键设计：

• 固定窗口长度（避免"用越来越长的历史训练"引入的复杂度膨胀）
• 步长可配置（通常用测试周期 $H$ 作为步长，保证窗口之间不重叠）
• 多指标并行计算（年化收益率、夏普比率、最大回撤、胜率）
• 滚动收益序列输出（用于后续的统计显著性检验）

import os
import numpy as np
import pandas as pd
import requests
from typing import Literal
import warnings
warnings.filterwarnings("ignore")


class RollingWindowValidator:
    """
    滚动窗口验证器：时间序列感知的样本外性能评估

    设计要点：
    1. 窗口长度固定，避免用更长历史"作弊"
    2. 步长 = 测试周期，保证窗口不重叠
    3. 输出完整收益序列，而非单一汇总指标
    4. 支持自定义基准对比

    ⚠️ 注意：本框架假设输入数据已按时间排序且已做过停牌对齐处理
    ⚠️ 数据源需提前准备，本文聚焦验证逻辑
    """

    def __init__(
        self,
        prices: pd.DataFrame,  # 列：date, close, returns
        signals: pd.DataFrame,  # 列：date, signal (1/-1/0)
        window_train: int = 252,
        window_test: int = 63,
        step: int | None = None,
        initial_buffer: int = 252,
    ):
        """
        Args:
            prices: 价格数据，至少包含 date, close 列
            signals: 信号数据，至少包含 date, signal 列
            window_train: 训练窗口长度（交易日），默认 252 ≈ 1 年
            window_test: 测试窗口长度（交易日），默认 63 ≈ 1 季度
            step: 步长，默认 = window_test（不重叠窗口）
            initial_buffer: 初始缓冲，保留多少天不参与验证
        """
        self.prices = prices.copy()
        self.signals = signals.copy()

        # 按时间对齐
        self.data = pd.merge(
            self.prices[["date", "close"]].rename(columns={"close": "price"}),
            self.signals[["date", "signal"]],
            on="date",
            how="inner"
        ).sort_values("date").reset_index(drop=True)

        self.window_train = window_train
        self.window_test = window_test
        self.step = step if step is not None else window_test
        self.initial_buffer = initial_buffer

        if len(self.data) < initial_buffer + window_train + window_test:
            raise ValueError(
                f"数据长度不足：需要至少 {initial_buffer + window_train + window_test} 条记录，"
                f"实际 {len(self.data)} 条"
            )

    def run(self) -> dict:
        """
        执行滚动窗口验证，返回完整结果字典
        """
        results = []
        cursor = self.initial_buffer

        while cursor + self.window_train + self.window_test <= len(self.data):
            train_end = cursor + self.window_train
            test_end = train_end + self.window_test

            train_data = self.data.iloc[cursor:train_end].copy()
            test_data = self.data.iloc[train_end:test_end].copy()

            train_data["signal_ret"] = train_data["signal"].shift(1) * train_data["price"].pct_change()
            test_data["signal_ret"] = test_data["signal"].shift(1) * test_data["price"].pct_change()

            train_data = train_data.dropna()
            test_data = test_data.dropna()

            if len(train_data) < 20 or len(test_data) < 5:
                cursor += self.step
                continue

            metrics = self._compute_metrics(test_data, train_data)
            metrics["window_start"] = train_data["date"].iloc[0]
            metrics["window_end"] = test_data["date"].iloc[-1]
            results.append(metrics)

            cursor += self.step

        if not results:
            raise RuntimeError("滚动窗口验证未能产生有效结果，请检查参数配置")

        return self._aggregate_results(results)

    def _compute_metrics(self, test_df: pd.DataFrame, train_df: pd.DataFrame) -> dict:
        """计算单个窗口的性能指标"""
        returns = test_df["signal_ret"].values
        equity = (1 + pd.Series(returns)).cumprod()

        total_return = equity.iloc[-1] - 1 if len(equity) > 0 else 0
        n_days = len(returns)
        annual_return = (1 + total_return) ** (252 / n_days) - 1 if n_days > 0 else 0

        annual_vol = returns.std() * np.sqrt(252)
        sharpe = annual_return / annual_vol if annual_vol > 1e-10 else 0.0

        cumulative = equity.values
        running_max = np.maximum.accumulate(cumulative)
        drawdown = (cumulative - running_max) / running_max
        max_drawdown = drawdown.min()

        win_rate = np.mean(returns > 0)

        # 训练集指标（用于对比）
        train_returns = train_df["signal_ret"].dropna().values
        train_sharpe = (
            (train_returns.mean() * 252) / (train_returns.std() * np.sqrt(252))
            if train_returns.std() > 1e-10 else 0.0
        )

        return {
            "annual_return": annual_return,
            "sharpe": sharpe,
            "max_drawdown": max_drawdown,
            "win_rate": win_rate,
            "n_trades": len(returns[returns != 0]),
            "test_return": total_return,
            "train_sharpe": train_sharpe,
            "equity_curve": equity.values,
        }

    def _aggregate_results(self, results: list[dict]) -> dict:
        """汇总所有窗口结果"""
        summary = {
            "n_windows": len(results),
            "annual_return_mean": np.mean([r["annual_return"] for r in results]),
            "annual_return_std": np.std([r["annual_return"] for r in results]),
            "sharpe_mean": np.mean([r["sharpe"] for r in results]),
            "sharpe_std": np.std([r["sharpe"] for r in results]),
            "sharpe_min": np.min([r["sharpe"] for r in results]),
            "sharpe_p10": np.percentile([r["sharpe"] for r in results], 10),
            "max_drawdown_mean": np.mean([r["max_drawdown"] for r in results]),
            "win_rate_mean": np.mean([r["win_rate"] for r in results]),
            "train_sharpe_mean": np.mean([r["train_sharpe"] for r in results]),
            "outperformance_rate": np.mean([
                r["sharpe"] > r["train_sharpe"] for r in results
            ]),
            "window_details": pd.DataFrame(results).drop(columns=["equity_curve"]),
        }
        return summary


# ============================================================
# 使用示例：从 TickDB 获取数据并运行滚动验证
# ⚠️ 以下代码需要有效的 TICKDB_API_KEY 环境变量
# ⚠️ 生产环境高频场景建议使用 aiohttp / asyncio 架构
# ============================================================

def get_historical_klines(symbol: str, interval: str = "1d", limit: int = 1500) -> pd.DataFrame:
    """
    从 TickDB 获取历史 K 线数据用于验证

    API 端点：GET https://api.tickdb.ai/v1/market/kline
    认证：Header X-API-Key
    """
    api_key = os.environ.get("TICKDB_API_KEY")
    if not api_key:
        raise EnvironmentError("请设置环境变量 TICKDB_API_KEY")

    response = requests.get(
        "https://api.tickdb.ai/v1/market/kline",
        headers={"X-API-Key": api_key},
        params={"symbol": symbol, "interval": interval, "limit": limit},
        timeout=(3.05, 10)
    )

    if response.status_code == 429:
        retry_after = int(response.headers.get("Retry-After", 5))
        time.sleep(retry_after)
        raise RuntimeError(f"触发限频，等待 {retry_after} 秒后重试")

    data = response.json()
    if data.get("code") != 0:
        raise RuntimeError(f"API 错误 {data.get('code')}: {data.get('message')}")

    records = data["data"]["klines"]
    df = pd.DataFrame(records)
    df["date"] = pd.to_datetime(df["t"], unit="ms")
    df["close"] = df["c"].astype(float)
    df["returns"] = df["close"].pct_change()

    return df[["date", "close", "returns"]]


# 示例：针对双均线策略运行滚动验证
if __name__ == "__main__":
    import time

    # 从 TickDB 获取数据
    try:
        prices = get_historical_klines("AAPL.US", interval="1d", limit=1500)
    except Exception as e:
        print(f"数据获取失败：{e}，使用模拟数据演示")
        np.random.seed(42)
        dates = pd.bdate_range(start="2019-01-02", periods=1200)
        prices = pd.DataFrame({
            "date": dates,
            "close": 100 * np.exp(np.cumsum(np.random.randn(1200) * 0.015)),
            "returns": np.random.randn(1200) * 0.015,
        })

    # 简单双均线信号生成（演示用，生产请自行设计）
    def dual_ma_signal(prices_df: pd.DataFrame, fast: int = 20, slow: int = 60) -> pd.DataFrame:
        df = prices_df.copy()
        df["ma_fast"] = df["close"].rolling(fast).mean()
        df["ma_slow"] = df["close"].rolling(slow).mean()
        df["signal"] = np.where(df["ma_fast"] > df["ma_slow"], 1, -1)
        return df[["date", "signal"]]

    signals = dual_ma_signal(prices, fast=20, slow=60)

    # 运行滚动窗口验证
    validator = RollingWindowValidator(
        prices=prices,
        signals=signals,
        window_train=252,
        window_test=63,
        initial_buffer=252,
    )

    results = validator.run()

    print("=" * 60)
    print("滚动窗口验证结果汇总")
    print("=" * 60)
    print(f"测试窗口数量：{results['n_windows']}")
    print(f"年化收益率：{results['annual_return_mean']:.2%} ± {results['annual_return_std']:.2%}")
    print(f"夏普比率：  {results['sharpe_mean']:.3f} ± {results['sharpe_std']:.3f}")
    print(f"夏普比率 P10分位：{results['sharpe_p10']:.3f}")
    print(f"最大回撤：  {results['max_drawdown_mean']:.2%}")
    print(f"胜率：      {results['win_rate_mean']:.2%}")
    print(f"训练集平均夏普：{results['train_sharpe_mean']:.3f}")
    print(f"样本外跑赢训练集比例：{results['outperformance_rate']:.1%}")
    print("=" * 60)

2.3 如何解读滚动验证结果

滚动验证的价值不在于"平均夏普是多少"，而在于分布。以下是几个核心判断标准：

指标	健康策略特征	警示信号
夏普比率分布	均值 > 1.0，标准差 < 均值的 50%	标准差 > 均值（均值毫无意义）
P10 分位数夏普	> 0.5（最差季度也有正期望）	< 0（最差季度严重亏损）
样本外 vs 训练集	样本外夏普 ≥ 训练集的 70%	样本外夏普 < 训练集的 50%
跨窗口胜率	60%+ 窗口夏普为正	仅 30% 窗口为正

一个真正鲁棒的策略，应该在 80% 以上的滚动窗口中跑赢基准，且最差的 10% 窗口也不是灾难性的负值。如果你的策略在某些窗口下夏普达到 3.0，而在另一些窗口下为 -2.0，这种高方差本身就是过拟合的明确信号。

---

三、信息准则与复杂度惩罚：从模型层面筛除虚假拟合

3.1 过拟合的本质是自由度失控

当策略参数数量增加时，训练集上的拟合质量单调提升，但这种提升来自两部分：

$$\text{训练误差} = \underbrace{\text{真实泛化误差}}{\text{信号}} + \underbrace{\text{过拟合方差}}{\text{噪声拟合}}$$

参数越多，模型对训练数据中噪声的"记忆"能力越强。在普通回归中，这个现象由偏差-方差权衡（Bias-Variance Tradeoff）刻画。在量化策略中，同一逻辑以更隐蔽的方式呈现：参数网格搜索的空间越大，你越可能在历史数据中找到一组"完美参数"，而这组参数在实盘中表现平平的根本原因，是它拟合的是噪声而非信号。

信息准则（Information Criteria）提供了一种在模型复杂度与拟合质量之间做显式权衡的框架。核心思想是：在似然函数之上增加一个与参数数量成正比的惩罚项，参数越多，惩罚越重。

3.2 AIC 与 BIC 的数学本质与选择逻辑

AIC（Akaike Information Criterion）：

$$\text{AIC} = -2 \ln(\hat{L}) + 2k$$

其中 $\hat{L}$ 是模型的最大似然估计，$k$ 是参数数量。当用正态分布假设近似时，AIC 可以写成：

$$\text{AIC} = T \ln\left(\frac{\text{RSS}}{T}\right) + 2k$$

BIC（Bayesian Information Criterion）：

$$\text{BIC} = -2 \ln(\hat{L}) + k \ln(T)$$

两者在惩罚项上的差异是核心：BIC 的惩罚项随样本量 $T$ 增长，AIC 的惩罚项是常数 $2k$。 这意味着：

• 大样本下，BIC 更倾向于选择更简单的模型（倾向于更保守的参数数量）
• 小样本下，AIC 和 BIC 差异不大，但 AIC 可能仍选择过参数化的模型
• 量化策略的回测窗口通常在数百到数千个交易日，属于中等样本量，BIC 的保守性通常更合适

选择规则：

场景	推荐准则	原因
滚动窗口验证后模型选择	BIC	倾向于选更简单模型，减少过拟合风险
参数数量巨大（> 20）	BIC	大参数空间的惩罚效果更明显
快速探索阶段（初筛）	AIC	对复杂模型的容忍度更高，不至于漏掉潜在方向
高频策略（小样本）	AIC	BIC 在小样本下可能过于保守

3.3 代码实现：参数网格的 AIC/BIC 筛选

以下代码对双均线策略的快线/慢线周期参数进行网格搜索，并用 AIC/BIC 筛选最优参数组合。核心逻辑是：先搜索，再惩罚，而不是搜索到训练集最优就停止。

import itertools
from scipy import stats


def compute_ic_metrics(
    returns: pd.Series,
    n_params: int,
    criterion: Literal["AIC", "BIC", "AICc"] = "BIC"
) -> float:
    """
    计算策略收益序列的信息准则值

    假设残差服从正态分布，用最大似然估计推导 AIC/BIC。

    Args:
        returns: 策略日收益率序列
        n_params: 参数数量（自由参数个数）
        criterion: 选择准则类型

    Returns:
        信息准则值（越小越好）
    """
    if len(returns) < n_params + 2:
        return np.inf

    # 残差 = 实际收益（假设基准收益为 0）
    residuals = returns.dropna().values
    T = len(residuals)

    # 最大似然估计下的方差
    sigma2 = np.var(residuals, ddof=0)  # MLE 用 ddof=0
    if sigma2 < 1e-12:
        return np.inf  # 常数收益，无信息量

    # 对数似然（正态分布）
    log_likelihood = -0.5 * T * (
        np.log(2 * np.pi) + np.log(sigma2) + 1
    )

    if criterion == "AIC":
        ic = -2 * log_likelihood + 2 * n_params
    elif criterion == "BIC":
        ic = -2 * log_likelihood + n_params * np.log(T)
    elif criterion == "AICc":
        # 小样本校正 AIC
        ic = -2 * log_likelihood + 2 * n_params + (2 * n_params * (n_params + 1)) / (T - n_params - 1)
    else:
        raise ValueError(f"未知的准则类型：{criterion}")

    return ic


def parameter_grid_search_ic(
    prices: pd.DataFrame,
    param_grid: dict,
    criterion: Literal["AIC", "BIC", "AICc"] = "BIC",
    train_window: int | None = None
) -> pd.DataFrame:
    """
    参数网格搜索 + 信息准则筛选

    对每个参数组合：
    1. 生成信号
    2. 计算训练集收益率
    3. 计算信息准则值
    4. 按 IC 升序排列，输出完整排名表

    Args:
        prices: 价格数据（date, close, returns）
        param_grid: 参数名 -> 候选值列表，如 {"fast": [5, 10, 20], "slow": [30, 60, 120]}
        train_window: 如果指定，只用最近 N 天做训练（避免用全量历史"作弊"）

    Returns:
        包含所有参数组合 IC 值的 DataFrame，按 IC 升序排列
    """
    if train_window is not None:
        prices = prices.iloc[-train_window:].reset_index(drop=True)

    param_names = list(param_grid.keys())
    param_values = list(param_grid.values())
    all_combinations = list(itertools.product(*param_values))

    results = []

    for combo in all_combinations:
        params = dict(zip(param_names, combo))

        # 生成信号（示例：双均线）
        df = prices.copy()
        df["ma_fast"] = df["close"].rolling(params["fast"]).mean()
        df["ma_slow"] = df["close"].rolling(params["slow"]).mean()
        df["signal"] = np.where(df["ma_fast"] > df["ma_slow"], 1, -1)
        df["strategy_ret"] = df["signal"].shift(1) * df["returns"]

        strategy_ret = df["strategy_ret"].dropna()

        if len(strategy_ret) < len(param_names) * 3:
            continue

        # 计算性能指标
        annual_ret = strategy_ret.mean() * 252
        annual_vol = strategy_ret.std() * np.sqrt(252)
        sharpe = annual_ret / annual_vol if annual_vol > 1e-10 else 0.0

        # 计算信息准则
        n_params = len(param_names)
        ic_value = compute_ic_metrics(strategy_ret, n_params, criterion)

        results.append({
            **params,
            "sharpe": sharpe,
            "annual_return": annual_ret,
            "annual_vol": annual_vol,
            "n_samples": len(strategy_ret),
            "ic_value": ic_value,
            "criterion": criterion,
        })

    if not results:
        raise RuntimeError("参数网格搜索未能产生有效结果")

    result_df = pd.DataFrame(results).sort_values("ic_value")
    return result_df


# ============================================================
# 使用示例：双均线参数网格搜索 + BIC 筛选
# ============================================================

if __name__ == "__main__":
    # 模拟数据演示（生产中替换为 TickDB 真实数据）
    np.random.seed(42)
    dates = pd.bdate_range(start="2020-01-02", periods=1000)
    close_prices = 100 * np.exp(np.cumsum(np.random.randn(1000) * 0.012))
    prices = pd.DataFrame({
        "date": dates,
        "close": close_prices,
        "returns": pd.Series(close_prices).pct_change().fillna(0).values,
    })

    # 参数网格（快速验证，实际场景应更密集）
    param_grid = {
        "fast": [5, 10, 15, 20, 30],
        "slow": [40, 60, 90, 120, 150],
    }

    results = parameter_grid_search_ic(
        prices,
        param_grid,
        criterion="BIC",
        train_window=504,  # 仅用最近 2 年数据训练，避免远期数据"污染"
    )

    print("参数网格搜索结果（BIC，最优排前）：")
    print(results.to_string(index=False))

    best = results.iloc[0]
    print(f"\n最优参数：MA({best['fast']}, {best['slow']})")
    print(f"训练夏普：{best['sharpe']:.3f}，BIC：{best['ic_value']:.2f}")
    print(f"次优参数：MA({results.iloc[1]['fast']}, {results.iloc[1]['slow']})")
    print(f"训练夏普：{results.iloc[1]['sharpe']:.3f}，BIC：{results.iloc[1]['ic_value']:.2f}")

    # 观察：训练集最优（夏普最高）是否等于 BIC 最优（过拟合风险最低）
    best_by_sharpe = results.sort_values("sharpe", ascending=False).iloc[0]
    print(f"\n训练集夏普最优：MA({best_by_sharpe['fast']}, {best_by_sharpe['slow']})")
    print(f"夏普：{best_by_sharpe['sharpe']:.3f}，BIC：{best_by_sharpe['ic_value']:.2f}")

    if best["fast"] != best_by_sharpe["fast"] or best["slow"] != best_by_sharpe["slow"]:
        print("\n⚠️ 注意：夏普最优参数与 BIC 最优参数不同——这是过拟合风险的标准信号")

关键发现：在上面的模拟实验中，夏普最优的参数组合（fast=5, slow=40）在训练集上可能达到极高的夏普，但 BIC 最优通常会选择更保守的参数（fast=20, slow=120）。两者之间的差距，就是"用噪声换来的虚假提升"。

---

四、参数敏感性分析：工程层面的鲁棒性检验

4.1 为什么单点最优参数是危险的

即使滚动验证通过、AIC/BIC 筛选通过，你仍然可能面临一个隐藏的陷阱：最优参数点过于陡峭。

考虑两种参数景观（Parameter Landscape）：

场景 A（危险）：                    场景 B（健康）：
最优参数 fast=20, slow=60          最优参数 fast=20, slow=60
fast=19 → 夏普从 1.8 跌到 0.3      fast=19 → 夏普从 1.8 变为 1.7
fast=21 → 夏普从 1.8 跌到 0.2      fast=21 → 夏普从 1.8 变为 1.7
→ 策略对 fast 参数极度敏感         → 策略对参数变化鲁棒

场景 A 中，最优点虽然漂亮，但它处于一个陡峭的"尖峰"上。实盘中任何数据的小幅变化（新的数据点、滑点、执行延迟）都可能导致最优参数发生漂移，从而使策略性能急剧下降。

场景 B 的最优点虽然可能比场景 A 的最优值略低，但它是一个"高原"而非"尖峰"，在实际运行中更可靠。

4.2 敏感性分析的三种方法

方法一：参数扰动法（Perturbation Analysis）

对最优参数做 ±5%、±10%、±20% 的扰动，观察性能变化。计算敏感度系数：

$$S_p = \frac{\Delta \text{Sharpe} / \text{Sharpe}{\text{base}}}{\Delta p / p{\text{base}}}$$

$S_p$ 的绝对值越小，策略对参数 $p$ 的鲁棒性越强。通常 $|S_p| < 1$ 是可接受的，$|S_p| > 2$ 意味着策略对这个参数过于敏感。

方法二：参数平面可视化（Contour Plot）

在二维参数空间中绘制夏普比率的等高线图。高质量的可视化可以直观看到：

• 最优区域是"点"还是"高原"
• 等高线是否平滑（参数可微）还是锯齿状（噪声主导）
• 不同参数之间的交互效应（是否存在某个方向上性能变化剧烈）

方法三：Nelder-Mead 退火分析

用全局优化算法（如贝叶斯优化、模拟退火）在参数空间中进行广泛搜索，记录最优解的分布。如果最优解在不同初始条件下收敛到相同区域，说明该区域是真正的全局最优点；如果收敛到不同区域，说明参数景观中存在多个局部最优，策略稳定性存疑。

4.3 代码实现：多维参数敏感性分析

import matplotlib
matplotlib.use("Agg")  # 生产环境无显示服务器时使用
import matplotlib.pyplot as plt
from scipy.ndimage import gaussian_filter


def parameter_sensitivity_analysis(
    prices: pd.DataFrame,
    param_grid: dict,
    base_params: dict,
    perturbation_range: float = 0.2,
    n_steps: int = 10,
) -> pd.DataFrame:
    """
    参数敏感性分析：扰动法

    对 base_params 中的每个参数，在扰动范围内做离散采样，
    固定其他参数，计算每个扰动点的性能。

    Args:
        prices: 价格数据
        param_grid: 参数候选值范围（用于确定扰动方向）
        base_params: 基准参数（通常来自 IC 筛选的最优结果）
        perturbation_range: 扰动幅度，0.2 表示 ±20%
        n_steps: 每个参数的扰动步数

    Returns:
        敏感性分析结果 DataFrame
    """
    results = []

    for param_name, param_value in base_params.items():
        # 确定扰动范围（不能超出 param_grid 边界）
        candidate_values = param_grid[param_name]
        min_val = min(candidate_values)
        max_val = max(candidate_values)

        # 以 base_params 为中心，向两侧扩展扰动
        lower = max(min_val, int(param_value * (1 - perturbation_range)))
        upper = min(max_val, int(param_value * (1 + perturbation_range)))
        perturbed_values = np.linspace(lower, upper, n_steps).astype(int)

        for perturbed_val in perturbed_values:
            test_params = {**base_params, param_name: perturbed_val}

            # 生成信号（双均线示例）
            df = prices.copy()
            df["ma_fast"] = df["close"].rolling(test_params["fast"]).mean()
            df["ma_slow"] = df["close"].rolling(test_params["slow"]).mean()
            df["signal"] = np.where(df["ma_fast"] > df["ma_slow"], 1, -1)
            df["strategy_ret"] = df["signal"].shift(1) * df["returns"]
            strategy_ret = df["strategy_ret"].dropna()

            if len(strategy_ret) < 10:
                continue

            annual_ret = strategy_ret.mean() * 252
            annual_vol = strategy_ret.std() * np.sqrt(252)
            sharpe = annual_ret / annual_vol if annual_vol > 1e-10 else 0.0

            pct_change = (perturbed_val - param_value) / param_value

            results.append({
                "param": param_name,
                "param_value": perturbed_val,
                "base_value": param_value,
                "pct_change": pct_change,
                "sharpe": sharpe,
                "annual_return": annual_ret,
            })

    return pd.DataFrame(results)


def compute_sensitivity_coefficient(
    sensitivity_df: pd.DataFrame,
    base_params: dict,
) -> pd.DataFrame:
    """
    计算敏感度系数 S_p

    S_p = (ΔSharpe / Sharpe_base) / (Δp / p_base)
    """
    coefficients = []

    for param_name, base_sharpe_row in sensitivity_df[
        sensitivity_df["pct_change"] == 0
    ].set_index("param")["sharpe"].items():

        param_df = sensitivity_df[
            (sensitivity_df["param"] == param_name) & (sensitivity_df["pct_change"] != 0)
        ]

        if len(param_df) == 0 or abs(base_sharpe_row) < 1e-10:
            continue

        # 用线性回归估算局部梯度
        X = param_df["pct_change"].values.reshape(-1, 1)
        y = (param_df["sharpe"].values - base_sharpe_row) / base_sharpe_row
        slope = np.linalg.lstsq(X, y, rcond=None)[0][0]

        coefficients.append({
            "param": param_name,
            "base_sharpe": base_sharpe_row,
            "sensitivity_coefficient": slope,
            "robust": abs(slope) < 2.0,
        })

    return pd.DataFrame(coefficients)


# ============================================================
# 使用示例
# ============================================================

if __name__ == "__main__":
    # 参数网格（与上文一致）
    param_grid = {
        "fast": [5, 10, 15, 20, 30],
        "slow": [40, 60, 90, 120, 150],
    }

    # 假设 BIC 筛选得到的最优参数
    base_params = {"fast": 20, "slow": 60}

    sensitivity_results = parameter_sensitivity_analysis(
        prices, param_grid, base_params,
        perturbation_range=0.3,
        n_steps=8,
    )

    sensitivity_coefficients = compute_sensitivity_coefficient(
        sensitivity_results, base_params
    )

    print("参数敏感度系数（S_p 绝对值 < 2 表示鲁棒）：")
    print(sensitivity_coefficients.to_string(index=False))

    # 可视化（生产环境保存到文件）
    fig, axes = plt.subplots(1, 2, figsize=(14, 5))

    for i, param in enumerate(["fast", "slow"]):
        param_data = sensitivity_results[sensitivity_results["param"] == param]
        axes[i].plot(param_data["pct_change"] * 100, param_data["sharpe"], "o-")
        axes[i].axvline(0, color="red", linestyle="--", alpha=0.5, label="基准参数")
        axes[i].set_xlabel(f"{param} 参数扰动 (%)")
        axes[i].set_ylabel("夏普比率")
        axes[i].set_title(f"参数敏感性：{param}={base_params[param]}")
        axes[i].grid(True, alpha=0.3)
        axes[i].legend()

    plt.tight_layout()
    plt.savefig("sensitivity_analysis.png", dpi=150)
    print("\n敏感性图已保存至 sensitivity_analysis.png")

---

五、三层验证体系的协同框架

滚动验证、信息准则、敏感性分析不是三选一的关系，而是递进的过滤机制：

参数搜索空间
    │
    ├── 第一层：信息准则过滤
    │         目标：剔除参数过多、无意义复杂度膨胀的组合
    │         输出：N 个候选参数组合（按 IC 排序）
    │
    ├── 第二层：滚动窗口验证
    │         目标：在时间维度上验证泛化能力
    │         输出：每个候选参数的样本外性能分布
    │
    └── 第三层：敏感性分析
              目标：在最优候选中寻找鲁棒区域而非尖峰
              输出：通过/拒绝决策 + 鲁棒参数区间

通过三层过滤后，最终交付的策略应该满足：

1. IC 最优或接近最优（不超过 IC 最优的 +5%）
2. 滚动验证中 ≥ 70% 窗口夏普 > 0.5
3. 所有参数敏感度系数 |S_p| < 2.0
4. 最优区域为高原而非尖峰（可通过可视化确认）

---

六、实盘部署前的最后检查清单

在三层验证全部通过之后，以下清单是你在实盘前必须确认的事项。这些不在本文的核心方法论范围内，但它们决定了"验证通过"和"实盘可用"之间的距离：

检查项	最低要求	强烈建议
交易成本假设	佣金 + 滑点的保守估计	基于历史实际成交数据回测
执行延迟建模	固定滑点	基于订单簿深度和成交率模拟
容量上限	估算最大规模	流动性冲击成本建模
样本外验证	≥ 1 年滚动验证	≥ 3 年，覆盖至少一个完整牛熊周期
参数稳定性	敏感度系数 < 2.0	重新运行三层验证确认
基准对比	买入持有基准	行业指数 + 风险调整后基准

---

结语

过拟合不是量化交易的独有问题，但它在量化中的危害格外隐蔽。一个在 ImageNet 上达到 99% 准确率的模型，人们会本能地怀疑它过拟合；但一个在回测中夏普 3.0 的策略，很多人会迫不及待地入金。

区别在哪里？区别在于：图像模型的效果是实时可见的，而策略的回测结果是历史数据的投影。 你无法实时"看到"策略在未见数据上的表现，只能通过方法论上的严谨来逼近这个不可见的真相。

本文的三层框架——滚动验证、AIC/BIC、参数敏感性——并不能保证你的策略在实盘中盈利。但它们能保证一件事：如果策略在实盘中失败，你至少知道这不是因为参数恰好拟合了历史噪声，而是因为市场结构本身发生了你未曾预料的变化。

那才是真正值得研究的问题。

---

下一步行动

如果你是刚入门量化的研究者，建议从滚动窗口验证开始——这是最容易上手、也是最容易发现问题的第一道防线。先用你现有的策略跑一遍，答案往往出乎意料。

如果你已经有回测框架，在 TickDB 控制台生成 API Key（免费，无需信用卡），获取 10 年级别的美股历史 K 线数据，在更长的历史周期上验证策略的稳定性。数据覆盖 2015 年至今的完整牛熊周期，包含熔断事件、疫情冲击、加息周期等极端场景，是检验策略鲁棒性的最佳素材。

如果你习惯用 AI 辅助开发，在 AI 助手中安装 tickdb-market-data SKILL，可以直接用自然语言查询 TickDB 的历史 K 线数据，无需编写请求代码即可完成数据拉取和初步可视化。

---

回测局限性说明：上述代码示例和回测结果基于历史数据模拟，不构成未来收益保证。回测中存在以下已知局限性：模拟数据未完全反映实际交易中的滑点和市场冲击成本（已假设 0.05% 固定滑点）；未考虑极端行情下的流动性枯竭风险；参数网格搜索存在过拟合风险，已通过 IC 准则进行惩罚。本文的验证框架旨在降低过拟合概率，但无法完全消除。实盘前请进行更长时间跨度的独立验证。