4.1 手写回测引擎¶

这一节讲什么？
本节把前面的策略假设和指标转换为明确的历史回测，并检查信号时点、仓位、成本和基线。历史回测不能证明未来盈利能力。
使用历史数据检验策略规则的过程称为回测（Backtesting）。本节先直接实现信号、持仓、收益和成本计算，再与回测框架的处理方式比较。

学习目标¶

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

1. 这节究竟解决什么实际问题？¶

4.1 手写回测引擎 的具体任务是：手写滞后信号、交易成本和净值计算，直观看到未来函数如何夸大利润。 本节将明确输入数据、计算过程、输出及其在后续研究中的用途。

2. 前置知识¶

理解收益率、仓位、交易成本与基本技术指标。 对公式的理解应包括变量定义、单位、时间索引和适用条件。

3. 数据说明¶

默认固定历史样本。历史数据只能说明过去在既定假设下发生了什么，不能承诺未来收益。 运行前先检查 shape、日期范围、列名与缺失值。数据看起来能画图，不代表它适合回答研究问题。

4. 学习目标与完成标准¶

完成本节后，你应能用自己的话解释 4.1 手写回测引擎 的输入、计算步骤、输出和局限；尤其能说明：手写滞后信号、交易成本和净值计算，直观看到未来函数如何夸大利润。 你还应能独立改动一个参数并解释结果为什么变化，指出至少一个会让结论失真的假设。

5. 核心概念与公式¶

回测按照明确的成交和成本假设重放历史数据。它不能完整复现未来流动性、订单竞争和系统故障。

策略收益 = 滞后仓位 × 资产收益 − 换手 × 成本；每一项都要有明确单位。 看到公式时依次确认：每个符号的单位是什么？它属于哪个时点？分母会不会为零？它是描述性统计量、预测量，还是交易规则？

6. 从零实现¶

原始代码按照“准备输入 → 计算中间量 → 得到结果 → 解释输出”的顺序组织。每个单元应检查变量类型、索引、维度和代表性数值。中间变量用于核对公式实现和定位计算错误。

7. 成熟库实现¶

回测框架替你管理订单和记录，不会自动判断数据、成本或策略假设是否合理。 使用库函数前先阅读参数含义、默认值和返回对象；库输出一个数字，并不说明这个数字适合你的交易假设。

8. 两种实现的差异¶

从零实现便于核对公式、调试和识别隐含假设；成熟库适合在计算逻辑已验证后复用。两种实现依赖相同的数据口径与时点假设，使用库函数不会改变输入数据的质量。

9. 预期结果与检查范围¶

你应看到维度正确的 Series/DataFrame、日期单调递增的索引，以及能支持“手写滞后信号、交易成本和净值计算，直观看到未来函数如何夸大利润。”这一任务的表格或图形。金融日收益率均值通常接近 0；若出现极大的连续收益、空表、全 NaN 或索引错位，应先排查数据和对齐逻辑，而不是立即相信结果。

10. 常见错误与诊断¶

信号未滞后、把手续费只扣一次、忽略空仓收益、年化频率错配、将净值曲线当显著性证明。

诊断顺序：先打印类型与 shape，再看 head()/tail()，接着检查日期和 NaN，最后才比较数值。这样能避免在错误的对象上反复改参数。

11. 研究偏差检查¶

检查未来函数、幸存者偏差、重复测试集调参、成交可得性和失败区间；必须报告基线。 结果至少应与一个简单基线比较，并明确哪些部分只是教学演示。

12. 三层练习与答案¶

• 基础练习：验证一个中间变量、补全一段小计算，并用一句话解释输出。
• 研究练习：改变窗口、样本区间、约束或标的，记录哪些结论仍成立。
• 开放项目：把本节方法嵌入模块项目，报告基线、失败结果和限制。

题目在 ../exercises/04_backtesting/01_simple_backtest_exercises.ipynb；完整推导、可运行参考实现与结果检查在 ../solutions/04_backtesting/01_simple_backtest_solutions.ipynb。请先独立作答，再打开答案。

13. 延伸阅读¶

延伸材料：Advances in Financial Machine Learning（Lopez de Prado）。阅读记录应包括研究假设、数据频率、样本范围和结论适用条件。

14. 下一节¶

下一节是 02_backtrader_intro。本节的概念会成为理解它的输入，而不是孤立的名词。

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 [ ]:

# 默认使用固定样本，保证课程与 CI 不依赖实时接口。
USE_OFFLINE_DATA = True
if USE_OFFLINE_DATA:
    from pathlib import Path
    import sys
    docs_root = next(parent for parent in [Path.cwd(), *Path.cwd().parents]
                     if (parent / 'utils' / 'offline_data.py').exists())
    sys.path.insert(0, str(docs_root))
    from utils.offline_data import configure_yfinance_offline
    configure_yfinance_offline(yf)
else:
    print('在线下载为扩展步骤；请核对数据字段、时区和下载日期。')

# 默认使用固定样本，保证课程与 CI 不依赖实时接口。 USE_OFFLINE_DATA = True if USE_OFFLINE_DATA: from pathlib import Path import sys docs_root = next(parent for parent in [Path.cwd(), *Path.cwd().parents] if (parent / 'utils' / 'offline_data.py').exists()) sys.path.insert(0, str(docs_root)) from utils.offline_data import configure_yfinance_offline configure_yfinance_offline(yf) else: 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) 将信号滞后一期，避免同一期数据生成信号并计入收益
# 今天收盘后计算信号，明天才执行
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) 将信号滞后一期，避免同一期数据生成信号并计入收益 # 今天收盘后计算信号，明天才执行 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 [ ]: