如何用 Python 写一个均线策略回测？（附避免未来函数的正确姿势）

import numpy as np
import pandas as pd
from alphafeed import AlphaFeed

def backtest_ma(symbol, fast=5, slow=20, count=500,
                fee=0.0003, slippage=0.001, init_cash=1_000_000):
    af = AlphaFeed(api_key="your-api-key")

    # 1. 取足够长的前复权日 K 线
    df = af.klines.get(symbol, period="1d", count=count, adjust="forward", to_dataframe=True)
    df = df.sort_values("trade_date").reset_index(drop=True)

    # 2. 计算均线与信号（都基于"当日收盘"，属于 T 日可知的信息）
    df["ma_fast"] = df["close"].rolling(fast).mean()
    df["ma_slow"] = df["close"].rolling(slow).mean()
    df["signal"] = (df["ma_fast"] > df["ma_slow"]).astype(int)

    # 3. 关键：信号 shift(1)，即 T 日信号 → T+1 日才持仓，用 T+1 开盘价成交
    df["position"] = df["signal"].shift(1).fillna(0)

    # 用开盘价计算持仓期收益（避免用收盘价成交造成的前视）
    df["open_ret"] = df["open"].pct_change().shift(-1)  # 今开→明开的收益，归属到今日持仓
    # 更严谨：持仓收益 = 次日开盘买入后到再次日开盘的价格变化
    df["daily_ret"] = df["open"].pct_change()
    df["strat_ret"] = df["position"] * df["daily_ret"]

    # 4. 扣成本：仓位变化的那天计一次手续费+滑点
    df["turnover"] = df["position"].diff().abs().fillna(0)
    df["cost"] = df["turnover"] * (fee + slippage)
    df["strat_ret_net"] = df["strat_ret"] - df["cost"]

    # 5. 净值与绩效
    df["nav"] = (1 + df["strat_ret_net"].fillna(0)).cumprod()
    total = df["nav"].iloc[-1] - 1
    n = len(df)
    ann = (1 + total) ** (250 / n) - 1
    vol = df["strat_ret_net"].std() * np.sqrt(250)
    sharpe = (ann - 0.02) / vol if vol else np.nan
    max_dd = (df["nav"] / df["nav"].cummax() - 1).min()

    return df, {
        "symbol": symbol, "total_return": round(total, 4),
        "ann_return": round(ann, 4), "sharpe": round(sharpe, 2),
        "max_drawdown": round(max_dd, 4),
        "trades": int(df["turnover"].sum() // 1),
    }

df, stats = backtest_ma("600519.SH", fast=5, slow=20)
print(stats)

df["bench_nav"] = df["close"] / df["close"].iloc[0]
print(f"策略最终净值: {df['nav'].iloc[-1]:.3f}")
print(f"基准(买入持有): {df['bench_nav'].iloc[-1]:.3f}")

symbols = ["600519.SH", "000001.SZ", "601318.SH"]
results = [backtest_ma(s)[1] for s in symbols]
print(pd.DataFrame(results).to_string(index=False))