Day 1 — Rolling Window 回测（simplest version）

import pandas as pd
import numpy as np

# 假设已有 df：包含 close, signal
# signal: 1 多头，-1 空头，0 空仓

def rolling_backtest(df, window=252):
    results = []

    for start in range(0, len(df)-window):
        end = start + window
        sub = df.iloc[start:end]

        # 简单收益：signal * 下一根收益率
        sub['ret'] = sub['close'].pct_change()
        sub['strategy_ret'] = sub['signal'].shift(1) * sub['ret']

        cum_ret = (1 + sub['strategy_ret']).prod() - 1
        max_dd = (sub['strategy_ret'].cumsum().cummax() - sub['strategy_ret'].cumsum()).max()

        results.append([sub.index[0], sub.index[-1], cum_ret, max_dd])

    return pd.DataFrame(results, columns=['start', 'end', 'cum_ret', 'max_dd'])

rolling_result = rolling_backtest(df)
rolling_result.head()

Day 2 — 蒙地卡罗模拟（打乱 returns）

import numpy as np
import matplotlib.pyplot as plt

returns = df['strategy_ret'].dropna().values

def monte_carlo(returns, n=1000):
    simulations = []

    for _ in range(n):
        shuffled = np.random.permutation(returns)
        simulated = (1 + shuffled).prod() - 1
        simulations.append(simulated)

    return simulations

mc_results = monte_carlo(returns, 1000)

plt.hist(mc_results, bins=50)
plt.title("Monte Carlo Return Distribution")
plt.show()

Day 3 — Rolling + 导出 CSV

rolling_result.to_csv("rolling_result.csv", index=False)

# 可视化每个窗口的收益
plt.plot(rolling_result['cum_ret'])
plt.title("Rolling Window Performance")
plt.show()

# 可视化最大回撤
plt.plot(rolling_result['max_dd'])
plt.title("Rolling MaxDD")
plt.show()

Day 4 — 稳定性指标计算

# %窗口正收益
positive_ratio = (rolling_result['cum_ret'] > 0).mean()

# median
median_yield = rolling_result['cum_ret'].median()

# IQR
iqr = rolling_result['cum_ret'].quantile(0.75) - rolling_result['cum_ret'].quantile(0.25)

stability_report = {
    "positive_ratio": positive_ratio,
    "median_yield": median_yield,
    "iqr": iqr
}

stability_report

判定建议模板：

def evaluate_stability(stab):
    if stab["positive_ratio"] > 0.6 and stab["iqr"] < 0.15:
        return "策略整体稳定，适合进入实盘或继续优化。"
    else:
        return "策略稳定性偏弱，建议继续改进（过滤器、时段优化、特征）。"

evaluate_stability(stability_report)

Day 5 — 压力测试（费用 / 滑点 / 数据异常）

def stress_test(df, fee=0.0004, slippage=0.0005):
    df = df.copy()
    df['ret'] = df['close'].pct_change()

    # 模拟滑点 + 费用
    df['strategy_ret_stress'] = df['signal'].shift(1) * df['ret'] - fee - slippage

    result = (1 + df['strategy_ret_stress'].dropna()).prod() - 1
    return result

for fee in [0.0004, 0.0008, 0.001]:
    for slip in [0.0005, 0.001]:
        print(f"fee={fee}, slippage={slip}, return={stress_test(df, fee, slip)}")

模拟缺失数据：

df_missing = df.copy()
df_missing.loc[df_missing.sample(100).index, 'close'] = np.nan
df_missing = df_missing.fillna(method="ffill")

Day 6 — 自动生成稳定性报告（HTML 输出）