4. 环境与数据接口：数据源接入、回测引擎搭建与实时行情模拟
做量化交易，说白了就三件事：数据、策略、执行。而数据，是所有策略的起点。我见过太多人花几个月写策略，结果一跑回测，发现数据源没接对，全白干了。今天咱们就把这块硬骨头啃下来。
4.1 数据源接入：Tushare 与 JoinQuant
国内做量化，绕不开两个数据源：Tushare 和 JoinQuant（聚宽）。我个人习惯是两者搭配用——Tushare 拿历史数据，JoinQuant 做在线回测。
4.1.1 Tushare 接入实战
Tushare 是个老牌数据平台，免费版够用，但 token 要自己去官网申请。嗯，这里有个坑：免费版有频率限制，每分钟最多 200 次请求。我曾经在回测脚本里忘了加 sleep，直接触发了封禁，数据停了半小时。
import tushare as ts
import time
# 初始化，token 换成你自己的
pro = ts.pro_api('你的token')
# 获取日线数据
df = pro.daily(ts_code='000001.SZ', start_date='20230101', end_date='20231231')
print(df.head())
# 注意：批量获取时加延时
for code in ['000001.SZ', '600036.SH']:
df = pro.daily(ts_code=code, start_date='20230101', end_date='20231231')
time.sleep(0.5) # 防止被封
4.1.2 JoinQuant 在线数据
JoinQuant 的好处是数据直接集成在回测环境里，不用自己管存储。你写策略时，直接调 get_price() 就行。但要注意，聚宽的数据是分钟级对齐的，如果你做高频策略，得用它的 tick 数据接口。
# JoinQuant 回测环境中的代码
def initialize(context):
g.security = '000001.XSHE'
set_benchmark('000300.XSHG')
def handle_data(context, data):
# 获取过去 20 天的收盘价
close_prices = attribute_history(g.security, 20, '1d', ['close'])
ma20 = close_prices['close'].mean()
current_price = data[g.security].close
if current_price > ma20:
order_target_percent(g.security, 0.8)
else:
order_target_percent(g.security, 0.2)
4.2 回测引擎搭建：从零开始写一个轻量级引擎
回测引擎，说白了就是模拟历史行情，让策略跑一遍。大厂用的事件驱动引擎，咱们自己写个简单的就够了。我当年第一个回测引擎只有 200 行代码，但跑通了多策略组合。
4.2.1 核心架构设计
一个回测引擎至少需要三个模块：数据加载器、策略执行器、绩效计算器。我习惯用面向对象的方式写，方便扩展。
class BacktestEngine:
def __init__(self, data, initial_cash=1000000):
self.data = data # DataFrame，包含 open/high/low/close/volume
self.cash = initial_cash
self.positions = {} # 持仓字典
self.trades = [] # 交易记录
def run(self, strategy_func):
for i in range(len(self.data)):
# 获取当前时刻的数据
current_bar = self.data.iloc[i]
# 调用策略函数
signals = strategy_func(current_bar, self.positions)
# 执行交易
self.execute_trades(signals, current_bar)
# 更新持仓市值
self.update_portfolio(current_bar)
def execute_trades(self, signals, bar):
# 这里实现具体的买卖逻辑
for signal in signals:
if signal['action'] == 'buy':
cost = signal['shares'] * bar['close']
if cost <= self.cash:
self.cash -= cost
self.positions[signal['code']] = signal['shares']
self.trades.append({
'time': bar.name,
'action': 'buy',
'price': bar['close'],
'shares': signal['shares']
})
4.2.2 滑点与手续费模拟
很多新手回测收益很高，一实盘就亏，原因就是没考虑滑点和手续费。我吃过这个亏，回测年化 30%，实盘只有 8%。
| 参数 | 建议值 | 说明 |
|---|---|---|
| 滑点 | 0.1% - 0.3% | 流动性差的股票取上限 |
| 佣金 | 万分之 2.5 | 最低 5 元 |
| 印花税 | 千分之 1 | 卖出时收取 |
def apply_slippage(price, direction, slippage=0.001):
"""模拟滑点，direction: 1 买入, -1 卖出"""
return price * (1 + direction * slippage)
def calculate_commission(trade_value):
"""计算佣金，最低 5 元"""
commission = trade_value * 0.00025
return max(commission, 5.0)
4.3 实时行情模拟：让策略跑起来
回测跑通了，下一步就是模拟实盘。实时行情模拟，说白了就是让策略在「假的时间流」里跑真数据。我常用的方法有两种：回放模式和加速模式。
4.3.1 回放模式
把历史数据按时间顺序一条条喂给策略，模拟实时行情。这种方式最真实，但速度慢。
import time
from datetime import datetime
class ReplayEngine:
def __init__(self, data):
self.data = data
self.current_index = 0
def next_bar(self):
"""模拟下一个 tick 到来"""
if self.current_index < len(self.data):
bar = self.data.iloc[self.current_index]
self.current_index += 1
# 模拟实时延迟
time.sleep(0.1) # 100ms 模拟网络延迟
return bar
return None
def run_realtime(self, strategy_func):
while True:
bar = self.next_bar()
if bar is None:
break
strategy_func(bar)
4.3.2 加速模式
如果你要测试策略在极端行情下的表现，加速模式更合适。比如把 1 分钟的数据压缩到 0.1 秒处理完。
class AcceleratedEngine:
def __init__(self, data, speed=10):
self.data = data
self.speed = speed # 加速倍数
def run(self, strategy_func):
for i in range(len(self.data)):
bar = self.data.iloc[i]
strategy_func(bar)
# 按加速倍数缩短等待时间
time.sleep(0.01 / self.speed)
4.4 知识体系总览
下面这张图，是我对本章内容的总结。数据源是地基，回测引擎是骨架，实时模拟是血液。三者缺一不可。
嗯，数据接口这块，说白了就是「拿来主义」——把别人的数据拿过来，用自己的引擎跑一遍，再用模拟环境验证。我做了这么多年，发现最花时间的不是写策略，而是把这三个环节串起来。你想想看，数据格式不统一、回测结果和模拟对不上、模拟环境又和实盘有差异...每一步都是坑。
但别怕，把这些坑一个个填平了，你的策略就稳了。我个人习惯是先跑通最简单的日线回测，再加分钟线，最后上实时模拟。一步一步来，别想一口吃成胖子。