4.1 手写回测引擎¶

这一节讲什么？
有了策略想法（上一节），有了指标工具（第三章），现在需要验证它们在历史上是否真的能赚钱。
这个验证过程叫做「回测（Backtesting）」。我们不用第三方框架，先自己手写一个，彻底搞清楚每一步在做什么。

学习目标¶

• 从零构建一个简单的事件驱动回测框架
• 实现双均线（SMA）交叉策略的完整回测
• 理解仓位管理、手续费扣除与绩效统计
• 深入理解 Look-ahead Bias（回测中最常见的错误）

In [1]:

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import yfinance as yf

plt.rcParams['figure.figsize'] = (13, 5)

# 下载数据（用 SPY 标普500 ETF，2018-2024 跨越牛熊周期）
data = yf.download('SPY', start='2018-01-01', end='2024-01-01', progress=False)
close = data['Close'].squeeze()
print('数据准备完成 ✅')

import numpy as np import pandas as pd import matplotlib.pyplot as plt import yfinance as yf plt.rcParams['figure.figsize'] = (13, 5) # 下载数据（用 SPY 标普500 ETF，2018-2024 跨越牛熊周期） data = yf.download('SPY', start='2018-01-01', end='2024-01-01', progress=False) close = data['Close'].squeeze() print('数据准备完成 ✅')

数据准备完成 ✅

In [2]:

import matplotlib.pyplot as plt

# 1. 设置系统自带的中文字体（这里使用黑体 SimHei）
plt.rcParams['font.sans-serif'] = ['SimHei']  # 如果你想用微软雅黑，可以改成 ['Microsoft YaHei']

# 2. 解决更换字体后，负号（-）显示为方块的问题
plt.rcParams['axes.unicode_minus'] = False

import matplotlib.pyplot as plt # 1. 设置系统自带的中文字体（这里使用黑体 SimHei） plt.rcParams['font.sans-serif'] = ['SimHei'] # 如果你想用微软雅黑，可以改成 ['Microsoft YaHei'] # 2. 解决更换字体后，负号（-）显示为方块的问题 plt.rcParams['axes.unicode_minus'] = False

1. 策略逻辑：为什么用均线金叉？¶

我们要实现的策略叫「双均线趋势跟随策略」，规则非常简单：

• 📈 买入：短期均线（20日）上穿长期均线（60日）→ 金叉
• 📉 卖出：短期均线下穿长期均线 → 死叉

为什么这个逻辑有道理？

回想均线的意义：短期均线反映近期价格，长期均线反映长期趋势。

• 当短期均线从下往上穿越长期均线，意味着「近期价格强于长期趋势」——趋势可能在转好，市场上涨力量在积累
• 反之，死叉意味着近期价格疲软，趋势可能在转坏

这背后是趋势跟随的逻辑：人群的买入行为有惯性，一旦趋势形成，往往持续一段时间。

但是！ 这个策略有明显的缺点：

• 滞后：均线是历史平均值，信号总是「慢半拍」
• 震荡市失效：价格横盘时会频繁给出假信号，被反复打止损

→ 知道缺陷，才能知道什么时候用它、什么时候不用它

2. 生成交易信号¶

⚠️ Look-ahead Bias — 回测最大的坑¶

在写代码之前，必须理解这个概念。

问题：今天的收盘价，你在今天收盘之后才知道。但在回测代码里，如果你用今天的收盘价计算出信号，然后今天就执行，这就用了「未来数据」——在真实交易中这是不可能的。

解决方法： 始终用 shift(1) 把信号延迟一天执行

# ❌ 错误：今天算出信号，今天就持仓（用了今天才知道的数据）
position = signal  

# ✅ 正确：今天算出信号，明天才执行（收盘后看信号，第二天开盘下单）
position = signal.shift(1)  

In [3]:

# 双均线策略参数
FAST = 20
SLOW = 60

signals = pd.DataFrame(index=close.index)
signals['price']    = close
signals['sma_fast'] = close.rolling(FAST).mean()  # 短期均线（近20天平均）
signals['sma_slow'] = close.rolling(SLOW).mean()  # 长期均线（近60天平均）

# 核心信号：快线 > 慢线 = 多头（趋势向上）
signals['signal'] = 0
signals.loc[signals['sma_fast'] > signals['sma_slow'], 'signal'] = 1   # 金叉：持多
signals.loc[signals['sma_fast'] < signals['sma_slow'], 'signal'] = -1  # 死叉：空仓

# ⚠️  关键：用 shift(1) 避免 Look-ahead Bias！
# 今天收盘后计算信号，明天才执行
signals['position'] = signals['signal'].shift(1)  # 次日持仓
signals['trade']    = signals['position'].diff().fillna(0)  # 持仓变化（1=买，-1=卖）

n_trades = (signals['trade'] != 0).sum()
print(f'策略期间共交易 {n_trades} 次（每次均线交叉就触发一次）')
signals[['price', 'sma_fast', 'sma_slow', 'signal', 'position']].tail(5)

# 双均线策略参数 FAST = 20 SLOW = 60 signals = pd.DataFrame(index=close.index) signals['price'] = close signals['sma_fast'] = close.rolling(FAST).mean() # 短期均线（近20天平均） signals['sma_slow'] = close.rolling(SLOW).mean() # 长期均线（近60天平均） # 核心信号：快线 > 慢线 = 多头（趋势向上） signals['signal'] = 0 signals.loc[signals['sma_fast'] > signals['sma_slow'], 'signal'] = 1 # 金叉：持多 signals.loc[signals['sma_fast'] < signals['sma_slow'], 'signal'] = -1 # 死叉：空仓 # ⚠️ 关键：用 shift(1) 避免 Look-ahead Bias！ # 今天收盘后计算信号，明天才执行 signals['position'] = signals['signal'].shift(1) # 次日持仓 signals['trade'] = signals['position'].diff().fillna(0) # 持仓变化（1=买，-1=卖） n_trades = (signals['trade'] != 0).sum() print(f'策略期间共交易 {n_trades} 次（每次均线交叉就触发一次）') signals[['price', 'sma_fast', 'sma_slow', 'signal', 'position']].tail(5)

策略期间共交易 26 次（每次均线交叉就触发一次）

Out[3]:

	price	sma_fast	sma_slow	signal	position
Date
2023-12-22	462.223328	450.823489	430.869441	1	1.0
2023-12-26	464.175049	451.945998	431.680981	1	1.0
2023-12-27	465.014343	453.088603	432.509264	1	1.0
2023-12-28	465.190063	454.255548	433.433133	1	1.0
2023-12-29	463.843292	455.268166	434.284825	1	1.0

3. 计算策略收益率¶

每天的策略收益 = 持仓方向 × 当天市场涨跌幅

• position = 1 且市场涨 1%→ 策略赚 1%
• position = 1 且市场跌 1% → 策略亏 1%
• position = 0（空仓）→ 策略收益 0%（待在场外，不赚不亏）

发生交易时，还要扣手续费。

In [4]:

COMMISSION = 0.001  # 0.1% 手续费（单边，这是比较接近基准的合理假设）

daily_ret = close.pct_change()  # 每日市场涨跌幅

# 策略收益 = 你的仓位方向 × 市场当天涨跌
signals['strategy_ret'] = signals['position'] * daily_ret

# 扣除手续费（只有发生买卖时才扣，不发生则不扣）
trade_cost = signals['trade'].abs() * COMMISSION
signals['strategy_ret'] -= trade_cost
signals['market_ret']    = daily_ret  # 买入持有的基准

# 计算累积收益（复利增长）
signals = signals.dropna()
signals['cum_strategy'] = (1 + signals['strategy_ret']).cumprod()
signals['cum_market']   = (1 + signals['market_ret']).cumprod()

print('最终累积收益：')
print(f'  策略:     {signals["cum_strategy"].iloc[-1]-1:.2%}')
print(f'  买入持有: {signals["cum_market"].iloc[-1]-1:.2%}')

COMMISSION = 0.001 # 0.1% 手续费（单边，这是比较接近基准的合理假设） daily_ret = close.pct_change() # 每日市场涨跌幅 # 策略收益 = 你的仓位方向 × 市场当天涨跌 signals['strategy_ret'] = signals['position'] * daily_ret # 扣除手续费（只有发生买卖时才扣，不发生则不扣） trade_cost = signals['trade'].abs() * COMMISSION signals['strategy_ret'] -= trade_cost signals['market_ret'] = daily_ret # 买入持有的基准 # 计算累积收益（复利增长） signals = signals.dropna() signals['cum_strategy'] = (1 + signals['strategy_ret']).cumprod() signals['cum_market'] = (1 + signals['market_ret']).cumprod() print('最终累积收益：') print(f' 策略: {signals["cum_strategy"].iloc[-1]-1:.2%}') print(f' 买入持有: {signals["cum_market"].iloc[-1]-1:.2%}')

最终累积收益：
  策略:     -11.56%
  买入持有: 100.87%

4. 可视化结果¶

In [5]:

fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(13, 9), sharex=True)

# 价格 + 均线 + 买卖点
ax1.plot(signals.index, signals['price'], linewidth=1, label='SPY', alpha=0.7)
ax1.plot(signals.index, signals['sma_fast'], label=f'SMA{FAST}（短期）', linewidth=1.3)
ax1.plot(signals.index, signals['sma_slow'], label=f'SMA{SLOW}（长期）', linewidth=1.3)
buy_signals  = signals[signals['trade'] > 0]
sell_signals = signals[signals['trade'] < 0]
ax1.scatter(buy_signals.index, buy_signals['price'],
             marker='^', color='green', s=80, zorder=5, label='买入（金叉）')
ax1.scatter(sell_signals.index, sell_signals['price'],
             marker='v', color='red', s=80, zorder=5, label='卖出（死叉）')
ax1.set_title(f'双均线策略 (SMA{FAST}/SMA{SLOW}) | SPY', fontsize=13)
ax1.legend(ncol=3)
ax1.grid(alpha=0.3)

# 累积收益对比（策略 vs 买入持有）
ax2.plot(signals.index, signals['cum_strategy'], label='双均线策略', linewidth=1.5, color='steelblue')
ax2.plot(signals.index, signals['cum_market'],   label='买入持有（基准）', linewidth=1.5,
          color='orange', linestyle='--')
ax2.set_title('累积收益对比 — 策略能否跑赢「什么都不做」？', fontsize=13)
ax2.legend()
ax2.grid(alpha=0.3)

plt.tight_layout()
plt.show()

fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(13, 9), sharex=True) # 价格 + 均线 + 买卖点 ax1.plot(signals.index, signals['price'], linewidth=1, label='SPY', alpha=0.7) ax1.plot(signals.index, signals['sma_fast'], label=f'SMA{FAST}（短期）', linewidth=1.3) ax1.plot(signals.index, signals['sma_slow'], label=f'SMA{SLOW}（长期）', linewidth=1.3) buy_signals = signals[signals['trade'] > 0] sell_signals = signals[signals['trade'] < 0] ax1.scatter(buy_signals.index, buy_signals['price'], marker='^', color='green', s=80, zorder=5, label='买入（金叉）') ax1.scatter(sell_signals.index, sell_signals['price'], marker='v', color='red', s=80, zorder=5, label='卖出（死叉）') ax1.set_title(f'双均线策略 (SMA{FAST}/SMA{SLOW}) | SPY', fontsize=13) ax1.legend(ncol=3) ax1.grid(alpha=0.3) # 累积收益对比（策略 vs 买入持有） ax2.plot(signals.index, signals['cum_strategy'], label='双均线策略', linewidth=1.5, color='steelblue') ax2.plot(signals.index, signals['cum_market'], label='买入持有（基准）', linewidth=1.5, color='orange', linestyle='--') ax2.set_title('累积收益对比 — 策略能否跑赢「什么都不做」？', fontsize=13) ax2.legend() ax2.grid(alpha=0.3) plt.tight_layout() plt.show()

5. 绩效指标汇总¶

光看图表不够，我们需要量化指标来客观评估策略：

In [6]:

def performance_summary(ret_series, name='策略'):
    """计算策略的主要绩效指标"""
    total_ret  = (1 + ret_series).prod() - 1
    n_days     = len(ret_series)
    annual_ret = (1 + total_ret) ** (252 / n_days) - 1
    annual_vol = ret_series.std() * np.sqrt(252)
    sharpe     = annual_ret / annual_vol if annual_vol > 0 else 0
    cum        = (1 + ret_series).cumprod()
    roll_max   = cum.cummax()
    mdd        = ((cum - roll_max) / roll_max).min()
    calmar     = annual_ret / abs(mdd) if mdd != 0 else 0
    return pd.Series({
        '总收益率':   f'{total_ret:.2%}',
        '年化收益率': f'{annual_ret:.2%}',
        '年化波动率': f'{annual_vol:.2%}',
        '夏普比率':   f'{sharpe:.2f}',
        '最大回撤':   f'{mdd:.2%}',
        'Calmar 比率': f'{calmar:.2f}',
    }, name=name)

strat = performance_summary(signals['strategy_ret'], f'双均线策略(SMA{FAST}/{SLOW})')
mkt   = performance_summary(signals['market_ret'],   'Buy & Hold (SPY)')
pd.DataFrame([strat, mkt]).T

def performance_summary(ret_series, name='策略'): """计算策略的主要绩效指标""" total_ret = (1 + ret_series).prod() - 1 n_days = len(ret_series) annual_ret = (1 + total_ret) ** (252 / n_days) - 1 annual_vol = ret_series.std() * np.sqrt(252) sharpe = annual_ret / annual_vol if annual_vol > 0 else 0 cum = (1 + ret_series).cumprod() roll_max = cum.cummax() mdd = ((cum - roll_max) / roll_max).min() calmar = annual_ret / abs(mdd) if mdd != 0 else 0 return pd.Series({ '总收益率': f'{total_ret:.2%}', '年化收益率': f'{annual_ret:.2%}', '年化波动率': f'{annual_vol:.2%}', '夏普比率': f'{sharpe:.2f}', '最大回撤': f'{mdd:.2%}', 'Calmar 比率': f'{calmar:.2f}', }, name=name) strat = performance_summary(signals['strategy_ret'], f'双均线策略(SMA{FAST}/{SLOW})') mkt = performance_summary(signals['market_ret'], 'Buy & Hold (SPY)') pd.DataFrame([strat, mkt]).T

Out[6]:

	双均线策略(SMA20/60)	Buy & Hold (SPY)
总收益率	-11.56%	100.87%
年化收益率	-2.11%	12.89%
年化波动率	20.41%	20.40%
夏普比率	-0.10	0.63
最大回撤	-44.01%	-33.72%
Calmar 比率	-0.05	0.38

🎯 练习¶

1. 参数敏感性测试：更改均线参数 FAST=5, SLOW=20，对比结果。参数变了收益变化多大？（如果变化很大，小心过拟合）

2. 手续费影响：增加手续费至 COMMISSION = 0.005（0.5%），观察策略表现如何变化。
   → 双均线策略交易频率较低，手续费影响没你想象的大还是大？

3. 验证 Look-ahead Bias 的影响：把 signals['position'] = signals['signal'].shift(1) 改成 signals['position'] = signals['signal']（去掉 shift），对比回测结果的差异。

4. 换市场测试：将 SPY 换成 TSLA 或 BTC-USD，哪个市场上双均线策略效果更好？为什么？（趋势市？震荡市？）

---

下一节 → 02_backtrader_intro.ipynb（用专业框架 Backtrader 实现同样的策略）

In [ ]: