回测框架搭建：Python环境准备、pandas/numpy基础、回测引擎架构设计
做量化交易，说白了就是跟数据打交道。而回测，就是验证你的策略在历史上能不能赚钱。我个人习惯把回测框架比作一个「时光机」——你得先造好这台机器，才能放心地穿越回去测试策略。
这一章，咱们就手把手把这台机器搭起来。
Python环境准备：别小看这一步
我见过太多人，上来就写代码，结果装包装到崩溃。嗯，这里要注意，环境隔离是基本功。
推荐用 conda 或 venv 创建独立环境。我个人偏爱 conda，因为它对科学计算包的支持更友好。
# 创建环境
conda create -n darkpool python=3.9
# 激活环境
conda activate darkpool
# 安装核心依赖
pip install pandas numpy matplotlib seaborn
pip install jupyter notebook # 调试用，强烈推荐
你想想看，一个干净的环境能省多少事？至少能避免「在我电脑上能跑」这种尴尬局面。
pandas/numpy基础：回测的左右手
做回测，90%的时间都在跟DataFrame打交道。pandas处理时间序列，numpy做数值计算，这两个库你得玩得转。
pandas核心操作
我习惯把pandas的操作分成三类：数据加载、数据清洗、数据计算。
| 操作类型 | 常用函数 | 说明 |
|---|---|---|
| 数据加载 | pd.read_csv() |
读取CSV，注意parse_dates参数 |
| 数据清洗 | dropna(), fillna() |
处理缺失值，暗池数据常有空值 |
| 数据计算 | rolling(), shift() |
滚动窗口、滞后操作，策略核心 |
举个例子，加载暗池交易数据时，我一般这么干：
import pandas as pd
import numpy as np
# 加载数据
df = pd.read_csv('darkpool_trades.csv',
parse_dates=['timestamp'],
index_col='timestamp')
# 看一眼数据长啥样
print(df.head())
# 处理缺失值——暗池数据经常有跳空
df = df.fillna(method='ffill') # 前向填充，模拟真实交易中的「无成交」状态
resample() 重采样到固定频率，能省很多事。
numpy加速计算
pandas底层就是numpy。但有些计算，直接用numpy更快。我在项目中遇到过，回测跑100万行数据，用pandas的apply要跑3分钟，换成numpy向量化操作，3秒搞定。
# 向量化计算收益率
prices = df['close'].values # 转成numpy数组
returns = np.diff(prices) / prices[:-1] # 一秒算完
# 计算滚动标准差（波动率）
window = 20
rolling_std = pd.Series(prices).rolling(window).std().values
为什么会这样？因为numpy的循环是在C语言层面跑的，比Python的for循环快几十倍。做回测，速度就是生命。
回测引擎架构设计：核心骨架
回测引擎，说白了就是一个事件循环。它模拟市场怎么走，你的策略怎么响应，最后算算赚了多少钱。
我设计过好几个版本的引擎，踩过不少坑。最终沉淀下来的架构，就三个核心模块：
- 数据模块：喂数据给引擎
- 策略模块：根据数据生成交易信号
- 风控与绩效模块：计算收益、回撤、夏普比率
下面这张图，是我个人习惯用的引擎架构：
你看，数据从左到右流，策略在中间处理，最后算绩效。底部那条虚线是回测循环——每来一笔新数据，就触发一次策略判断。
核心代码骨架
下面是一个极简的回测引擎框架。别嫌它简单，所有复杂的引擎都是从这个骨架长出来的。
class BacktestEngine:
def __init__(self, data, strategy):
self.data = data # DataFrame
self.strategy = strategy # 策略函数
self.positions = [] # 持仓记录
self.trades = [] # 成交记录
self.capital = 1_000_000 # 初始资金
def run(self):
"""主循环：逐行处理数据"""
for i in range(len(self.data)):
# 1. 获取当前数据切片
current_bar = self.data.iloc[:i+1]
# 2. 调用策略，生成信号
signal = self.strategy(current_bar)
# 3. 执行交易（简化版）
if signal == 1: # 买入
self._buy(current_bar.iloc[-1])
elif signal == -1: # 卖出
self._sell(current_bar.iloc[-1])
# 4. 记录持仓市值
self._update_pnl(current_bar.iloc[-1])
# 5. 生成绩效报告
return self._generate_report()
def _buy(self, bar):
"""买入逻辑——暗池交易要注意滑点"""
price = bar['close'] * 1.001 # 模拟0.1%滑点
shares = self.capital // price
self.positions.append({'price': price, 'shares': shares})
self.trades.append({'type': 'buy', 'price': price, 'shares': shares})
def _sell(self, bar):
"""卖出逻辑"""
if not self.positions:
return
pos = self.positions.pop(0)
price = bar['close'] * 0.999 # 模拟滑点
self.capital += price * pos['shares']
self.trades.append({'type': 'sell', 'price': price, 'shares': pos['shares']})
def _generate_report(self):
"""生成绩效指标"""
# 这里只是示意，实际要算夏普、最大回撤等
return {'final_capital': self.capital, 'total_trades': len(self.trades)}
避坑指南：我踩过的三个大坑
- 前视偏差：用未来的数据计算当前信号。比如用当天的收盘价计算当天的均线——这在实盘里根本做不到。解决办法：用
shift(1)把数据往后挪一天。 - 幸存者偏差：只用了现在还存活的股票做回测。退市的股票亏的钱，你全没算进去。解决办法：用全量历史数据，包括已退市的。
- 过度优化：参数调得刚刚好，历史回测曲线完美。换到实盘就崩。解决办法：做样本外测试，或者用交叉验证。
好了，环境搭好了，pandas/numpy也热身了，引擎骨架也画出来了。接下来就是往里面填策略逻辑。记住，回测框架的核心不是代码多花哨，而是真实、可复现、无偏差。
公众号:蓝海资料掘金营，微信deep3321