第28章 量化回测系统开发
28.1 回测系统核心逻辑：数据输入、策略执行、绩效计算
28.1.1 先讲个真实需求：我把网上的回测框架改崩了三次
一开始做回测我直接用现成的开源框架，参数调一下就能出结果，看起来收益很高，实盘一跑完全不对。后来发现开源框架默认把停牌的股票当成涨停算、没算滑点、甚至用了未来函数，回测全是虚的。我干脆自己写了个回测系统，前前后后改崩了三次，才把各种坑踩完，现在回测和实盘vb.net教程C#教程python教程SQL教程access 2010教程的收益偏差能控制在5%以内。今天我就把回测系统的核心逻辑拆透，从数据输入到绩效计算每一步都讲明白，帮你避开回测最容易踩的坑。

28.1.2 核心逻辑：回测系统的三层架构
一个靠谱的回测系统必须分成三层，每层职责明确，避免耦合出问题：
1.数据输入层：负责喂高质量的原始数据，没有未来函数、没有缺失值、考虑停牌退市等真实场景，数据错了后面全白搭。
2.策略执行层：完全模拟实盘交易逻辑，包括调仓、下单、撮合、持仓计算，和实盘规则一模一样，不能有任何“开外挂”的操作。
3.绩效计算层：算准收益、风险、回撤等核心指标，客观评价策略好坏，不要为了好看乱改指标计算方式。

28.1.3 实战：从零搭建回测系统核心
我们完全从原生Python写，不依赖任何第三方回测框架，每一行代码都透明，避免黑箱问题。

1. 实战代码：回测系统核心实现
   python

	# 1. 导入需要的库
	import pandas as pd
	import numpy as np
	from datetime import datetime, timedelta
	import tushare as ts
	import os
	from dataclasses import dataclass
	from typing import List, Dict, Optional
	
	# 2. 初始化Tushare
	pro = ts.pro_api()
	
	# 3. 定义基础数据结构（用dataclass避免字典混乱）
	@dataclass
	class Bar:
	"""单根K线数据结构，对应一个交易日的行情"""
	ts_code: str
	trade_date: datetime
	open: float
	high: float
	low: float
	close: float
	volume: float
	is_trading: bool # 是否可交易：False=停牌/退市
	
	@dataclass
	class Order:
	"""订单数据结构"""
	ts_code: str
	trade_date: datetime
	direction: str # buy/sell
	price: float
	volume: int
	status: str # filled/canceled/partial
	
	@dataclass
	class Position:
	"""持仓数据结构"""
	ts_code: str
	volume: int
	avg_cost: float # 平均持仓成本
	market_value: float # 最新市值
	
	# 4. 核心回测类
	class BacktestEngine:
	def __init__(self, start_date: str, end_date: str, initial_capital: float = 1000000):
	"""
	初始化回测引擎
	:param start_date: 回测起始日期，格式YYYYMMDD
	:param end_date: 回测结束日期，格式YYYYMMDD
	:param initial_capital: 初始资金，默认100万
	"""
	self.start_date = datetime.strptime(start_date, '%Y%m%d')
	self.end_date = datetime.strptime(end_date, '%Y%m%d')
	self.initial_capital = initial_capital
	self.current_capital = initial_capital # 可用资金
	self.current_date: Optional[datetime] = None
	
	# 核心数据存储
	self.price_data: Dict[str, List[Bar]] = {} # 所有股票的行情数据
	self.calendar: List[datetime] = [] # 交易日历
	self.positions: Dict[str, Position] = {} # 当前持仓
	self.orders: List[Order] = [] # 历史订单
	self.net_value_history: List[Dict] = [] # 净值历史
	
	# 交易成本设置（和实盘完全一致）
	self.commission_rate = 0.00025 # 佣金万分之2.5
	self.stamp_duty_rate = 0.001 # 印花税千分之1，仅卖出收
	self.slip_rate = 0.001 # 滑点千分之1，买卖都收
	self.min_commission = 5 # 最低佣金5元
	
	# -------------------------- 第一层：数据输入模块 --------------------------
	def load_data(self, stock_list: List[str], adjust: str = 'hfq'):
	"""
	加载回测需要的行情和交易日历数据
	:param stock_list: 要回测的股票代码列表
	:param adjust: 复权方式：hfq后复权，qfq前复权，None不复权
	"""
	print('开始加载回测数据...')
	# 1. 加载交易日历
	self.calendar = self._get_trade_calendar()
	print(f'加载交易日历完成，共{len(self.calendar)}个交易日')
	
	# 2. 加载每只股票的行情数据
	total = len(stock_list)
	for idx, ts_code in enumerate(stock_list):
	# 获取行情数据
	df = pro.daily(
	ts_code=ts_code,
	start_date=self.start_date.strftime('%Y%m%d'),
	end_date=self.end_date.strftime('%Y%m%d'),
	fields='trade_date,open,high,low,close,vol'
	)
	# 获取复权因子
	if adjust in ['hfq', 'qfq']:
	adj = pro.adj_factor(
	ts_code=ts_code,
	start_date=self.start_date.strftime('%Y%m%d'),
	end_date=self.end_date.strftime('%Y%m%d'),
	fields='trade_date,adj_factor'
	)
	df = df.merge(adj, on='trade_date', how='left')
	# 复权计算
	if adjust == 'hfq':
	for col in ['open', 'high', 'low', 'close']:
	df[col] = df[col] * df['adj_factor'] / df['adj_factor'].iloc[-1]
	else:
	for col in ['open', 'high', 'low', 'close']:
	df[col] = df[col] * df['adj_factor'] / df['adj_factor'].iloc[0]
	
	# 转换日期格式
	df['trade_date'] = pd.to_datetime(df['trade_date'], format='%Y%m%d')
	# 对齐到交易日历，填充停牌数据
	stock_bars = []
	for date in self.calendar:
	day_data = df[df['trade_date'] == date]
	if not day_data.empty:
	bar = Bar(
	ts_code=ts_code,
	trade_date=date,
	open=day_data['open'].iloc[0],
	high=day_data['high'].iloc[0],
	low=day_data['low'].iloc[0],
	close=day_data['close'].iloc[0],
	volume=day_data['vol'].iloc[0],
	is_trading=True
	)
	else:
	# 停牌：用前一天收盘价填充，标记为不可交易
	prev_bar = stock_bars[-1] if stock_bars else Bar(ts_code, date, 0,0,0,0,0,False)
	bar = Bar(
	ts_code=ts_code,
	trade_date=date,
	open=prev_bar.close,
	high=prev_bar.close,
	low=prev_bar.close,
	close=prev_bar.close,
	volume=0,
	is_trading=False
	)
	stock_bars.append(bar)
	self.price_data[ts_code] = stock_bars
	print(f'加载{ts_code}数据完成，进度：{int((idx+1)/total*100)}%')
	print('所有数据加载完成！')
	
	def _get_trade_calendar(self) -> List[datetime]:
	"""获取A股交易日历，避免把非交易日算进去"""
	df = pro.trade_cal(
	exchange='SSE',
	start_date=self.start_date.strftime('%Y%m%d'),
	end_date=self.end_date.strftime('%Y%m%d'),
	fields='cal_date,is_open'
	)
	df = df[df['is_open'] == 1]
	df['cal_date'] = pd.to_datetime(df['cal_date'], format='%Y%m%d')
	return df['cal_date'].sort_values().tolist()
	
	# -------------------------- 第二层：策略执行模块 --------------------------
	def run_backtest(self, strategy_func):
	"""
	运行回测
	:param strategy_func: 策略函数，输入当前日期、当前持仓、可用资金，输出要交易的订单
	"""
	if not self.price_data:
	raise ValueError('请先加载数据！')
	print('开始运行回测...')
	
	# 按交易日逐天回测
	for date in self.calendar:
	self.current_date = date
	# 1. 先更新当前持仓的市值
	self._update_position_value()
	# 2. 记录当日净值
	self._record_net_value()
	# 3. 调用策略函数，生成交易信号
	orders = strategy_func(
	current_date=date,
	positions=self.positions,
	available_capital=self.current_capital,
	get_bar_func=self._get_current_bar
	)
	# 4. 撮合订单，执行交易
	if orders:
	self._match_orders(orders)
	# 回测结束，记录最后一天净值
	self._update_position_value()
	self._record_net_value()
	print('回测运行完成！')
	
	def _get_current_bar(self, ts_code: str) -> Bar:
	"""获取当前日期的K线数据，策略函数只能调用这个接口拿数据，避免未来函数"""
	bars = self.price_data[ts_code]
	# 找到当前日期对应的K线
	for bar in bars:
	if bar.trade_date == self.current_date:
	return bar
	raise ValueError(f'找不到{ts_code}在{self.current_date}的行情数据')
	
	def _update_position_value(self):
	"""更新当前持仓的市值"""
	for ts_code, pos in self.positions.items():
	current_bar = self._get_current_bar(ts_code)
	pos.market_value = pos.volume * current_bar.close
	
	def _match_orders(self, orders: List[Order]):
	"""撮合订单，完全模拟实盘交易规则"""
	for order in orders:
	# 校验订单时间
	if order.trade_date != self.current_date:
	order.status = 'canceled'
	self.orders.append(order)
	continue
	# 校验股票是否可交易
	current_bar = self._get_current_bar(order.ts_code)
	if not current_bar.is_trading:
	order.status = 'canceled'
	self.orders.append(order)
	continue
	# 计算滑点后的实际成交价格
	if order.direction == 'buy':
	# 买滑点：价格向上滑千分之1，不能超过当日最高价
	trade_price = min(order.price * (1 + self.slip_rate), current_bar.high)
	else:
	# 卖滑点：价格向下滑千分之1，不能低于当日最低价
	trade_price = max(order.price * (1 - self.slip_rate), current_bar.low)
	# 计算交易成本
	trade_amount = trade_price * order.volume
	commission = max(trade_amount * self.commission_rate, self.min_commission)
	stamp_duty = trade_amount * self.stamp_duty_rate if order.direction == 'sell' else 0
	total_cost = trade_amount + commission + stamp_duty
	
	# 执行买入
	if order.direction == 'buy':
	if self.current_capital >= total_cost:
	# 扣钱，加持仓
	self.current_capital -= total_cost
	if order.ts_code in self.positions:
	# 已有持仓，更新平均成本
	pos = self.positions[order.ts_code]
	new_volume = pos.volume + order.volume
	new_cost = (pos.avg_cost * pos.volume + trade_price * order.volume + commission) / new_volume
	pos.volume = new_volume
	pos.avg_cost = new_cost
	pos.market_value = new_volume * current_bar.close
	else:
	# 新开仓
	self.positions[order.ts_code] = Position(
	ts_code=order.ts_code,
	volume=order.volume,
	avg_cost=trade_price + commission/order.volume,
	market_value=order.volume * current_bar.close
	)
	order.status = 'filled'
	else:
	order.status = 'canceled'
	# 执行卖出
	else:
	if order.ts_code in self.positions and self.positions[order.ts_code].volume >= order.volume:
	# 加钱，减持仓
	self.current_capital += (trade_amount - commission - stamp_duty)
	pos = self.positions[order.ts_code]
	pos.volume -= order.volume
	if pos.volume == 0:
	del self.positions[order.ts_code]
	else:
	pos.market_value = pos.volume * current_bar.close
	order.status = 'filled'
	else:
	order.status = 'canceled'
	self.orders.append(order)
	
	def _record_net_value(self):
	"""记录当日净值"""
	total_value = self.current_capital
	for pos in self.positions.values():
	total_value += pos.market_value
	net_value = total_value / self.initial_capital
	self.net_value_history.append({
	'trade_date': self.current_date,
	'total_value': total_value,
	'net_value': net_value,
	'position_count': len(self.positions),
	'available_capital': self.current_capital
	})
	
	# -------------------------- 第三层：绩效计算模块 --------------------------
	def calculate_performance(self, benchmark_code: str = '000300.SH') -> pd.DataFrame:
	"""
	计算回测绩效指标
	:param benchmark_code: 基准指数代码，默认沪深300
	"""
	if not self.net_value_history:
	raise ValueError('请先运行回测！')
	print('开始计算绩效指标...')
	
	# 1. 处理净值数据
	nv_df = pd.DataFrame(self.net_value_history)
	nv_df['daily_return'] = nv_df['net_value'].pct_change()
	nv_df['benchmark_return'] = self._get_benchmark_return(benchmark_code)
	
	# 2. 计算核心收益指标
	total_return = nv_df['net_value'].iloc[-1] - 1
	annual_return = (1 + total_return) ** (252 / len(nv_df)) - 1
	benchmark_total_return = nv_df['benchmark_net'].iloc[-1] - 1
	benchmark_annual_return = (1 + benchmark_total_return) ** (252 / len(nv_df)) - 1
	excess_return = annual_return - benchmark_annual_return
	
	# 3. 计算风险指标
	max_drawdown = (nv_df['net_value'] / nv_df['net_value'].cummax() - 1).min()
	volatility = nv_df['daily_return'].std() * np.sqrt(252)
	# 夏普比率（无风险利率按3%年化）
	daily_risk_free = 0.03 / 252
	sharpe_ratio = np.sqrt(252) * (nv_df['daily_return'] - daily_risk_free).mean() / nv_df['daily_return'].std()
	# 信息比率
	active_return = nv_df['daily_return'] - nv_df['benchmark_return']
	information_ratio = np.sqrt(252) * active_return.mean() / active_return.std()
	# 胜率
	win_rate = len(nv_df[nv_df['daily_return'] > 0]) / len(nv_df.dropna(subset=['daily_return']))
	
	# 4. 整理成表格
	metrics = pd.DataFrame({
	'指标名称': ['总收益率', '年化收益率', '基准总收益率', '基准年化收益率', '超额年化收益率',
	'最大回撤', '年化波动率', '夏普比率', '信息比率', '胜率', '交易次数'],
	'数值': [
	f'{total_return*100:.2f}%', f'{annual_return*100:.2f}%',
	f'{benchmark_total_return*100:.2f}%', f'{benchmark_annual_return*100:.2f}%',
	f'{excess_return*100:.2f}%', f'{max_drawdown*100:.2f}%',
	f'{volatility*100:.2f}%', f'{sharpe_ratio:.2f}',
	f'{information_ratio:.2f}', f'{win_rate*100:.2f}%',
	len([o for o in self.orders if o.status == 'filled'])
	]
	})
	print('绩效计算完成：')
	print(metrics)
	return metrics, nv_df
	
	def _get_benchmark_return(self, benchmark_code: str) -> pd.Series:
	"""获取基准指数的收益率序列"""
	# 获取基准行情
	df = pro.index_daily(
	ts_code=benchmark_code,
	start_date=self.start_date.strftime('%Y%m%d'),
	end_date=self.end_date.strftime('%Y%m%d'),
	fields='trade_date,close'
	)
	df['trade_date'] = pd.to_datetime(df['trade_date'], format='%Y%m%d')
	df = df.set_index('trade_date').reindex(self.calendar).sort_index()
	df['benchmark_return'] = df['close'].pct_change()
	df['benchmark_net'] = df['close'] / df['close'].iloc[0]
	# 把基准净值合并到净值历史
	for i, record in enumerate(self.net_value_history):
	record['benchmark_net'] = df['benchmark_net'].iloc[i]
	return df['benchmark_return']
	
	# -------------------------- 策略示例：测试回测系统 --------------------------
	def demo_strategy(current_date: datetime, positions: Dict[str, Position], available_capital: float, get_bar_func):
	"""示例策略：每个季度第一个交易日等权买入沪深300成分股前10只"""
	orders = []
	# 只在每个季度第一个交易日调仓
	if current_date.month not in [1,4,7,10] or current_date.day > 10:
	return orders
	# 选股示例：选PE最低的10只沪深300成分股（实际用你的选股逻辑）
	selected_stocks = ['600519.SH', '000858.SZ', '002594.SZ', '601318.SH', '600036.SH',
	'000002.SZ', '601888.SH', '600276.SH', '000651.SZ', '000333.SZ']
	# 先卖出不在持仓里的股票
	for ts_code in list(positions.keys()):
	if ts_code not in selected_stocks:
	bar = get_bar_func(ts_code)
	orders.append(Order(
	ts_code=ts_code,
	trade_date=current_date,
	direction='sell',
	price=bar.open, # 开盘价下单
	volume=positions[ts_code].volume,
	status='pending'
	))
	# 等权买入新股票
	if available_capital > 10000:
	per_stock_amount = available_capital / len(selected_stocks)
	for ts_code in selected_stocks:
	if ts_code not in positions:
	bar = get_bar_func(ts_code)
	volume = int(per_stock_amount / (bar.open * 100)) * 100 # 买整手
	if volume >= 100:
	orders.append(Order(
	ts_code=ts_code,
	trade_date=current_date,
	direction='buy',
	price=bar.open,
	volume=volume,
	status='pending'
	))
	return orders
	
	# -------------------------- 运行回测示例 --------------------------
	if __name__ == '__main__':
	# 初始化回测引擎
	engine = BacktestEngine(start_date='20180101', end_date='20231231', initial_capital=1000000)
	# 加载数据
	stock_list = ['600519.SH', '000858.SZ', '002594.SZ', '601318.SH', '600036.SH',
	'000002.SZ', '601888.SH', '600276.SH', '000651.SZ', '000333.SZ']
	engine.load_data(stock_list, adjust='hfq')
	# 运行回测
	engine.run_backtest(demo_strategy)
	# 计算绩效
	metrics, nv_df = engine.calculate_performance(benchmark_code='000300.SH')
	# 保存结果
	os.makedirs('回测结果', exist_ok=True)
	metrics.to_csv('回测结果/绩效指标.csv', index=False, encoding='utf-8-sig')
	nv_df.to_csv('回测结果/净值序列.csv', index=False, encoding='utf-8-sig')

逐行讲解：
基础数据结构：用dataclass定义Bar、Order、Position，比字典更清晰，避免字段名写错，代码可读性高很多。
数据输入层：
load_data()：加载行情数据，重点是对齐交易日历，填充停牌数据，停牌的股票标记为不可交易，不会被误买入，和实盘完全一致。
复权处理：支持前后复权，用官方复权因子计算，不会出现复权价格错误的问题。
策略执行层：
run_backtest()：逐天回测，流程是：更新持仓市值→记录净值→调用策略生成订单→撮合交易，和实盘的每日流程一模一样。
_match_orders()：撮合逻辑是核心，考虑了滑点、佣金、印花税、最低佣金、停牌不能交易、资金不足不能买入、持仓不足不能卖出等所有实盘规则，完全没有“开外挂”的操作。
未来函数防护：策略只能通过_get_current_bar()拿当前日期的数据，不能访问未来的行情，从接口层面避免未来函数。
绩效计算层：
calculate_performance()：计算所有常用的绩效指标，从收益、风险、胜率三个维度全面评价策略，指标计算方式和业内标准完全一致，不会为了好看乱改。
自动对比基准指数，直接看超额收益，不用自己再算基准收益。
运行结果：
生成两个CSV文件，一个是核心绩效指标，一个是每日净值序列，直接可以用来画图或者做进一步分析。

28.1.4 基础知识拓展：回测的常见坑与规避方法

1. 数据层面的坑

2. 交易层面的坑

3. 绩效计算层面的坑

28.1.5 实战优化：添加绩效可视化功能

1. 实战代码：添加净值曲线和回撤曲线绘制
   python

	# 在BacktestEngine类中添加可视化方法
	import matplotlib.pyplot as plt
	plt.rcParams['font.sans-serif'] = ['SimHei']
	plt.rcParams['axes.unicode_minus'] = False
	
	def plot_performance(self, nv_df: pd.DataFrame, save_fig: bool = True):
	"""绘制净值曲线和回撤曲线"""
	fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8), gridspec_kw={'height_ratios': [3, 1]})
	
	# 第一层：净值曲线
	ax1.plot(nv_df['trade_date'], nv_df['net_value'], label='策略净值', color='#FF4B4B', linewidth=2)
	ax1.plot(nv_df['trade_date'], nv_df['benchmark_net'], label='基准净值', color='#3498DB', linewidth=2)
	ax1.set_title('策略净值vs基准净值', fontsize=15)
	ax1.legend(fontsize=12)
	ax1.grid(alpha=0.3)
	ax1.set_ylabel('净值（初始=1）')
	
	# 第二层：回撤曲线
	drawdown = nv_df['net_value'] / nv_df['net_value'].cummax() - 1
	ax2.fill_between(nv_df['trade_date'], drawdown, 0, color='#FF4B4B', alpha=0.3)
	ax2.set_title('策略回撤曲线', fontsize=15)
	ax2.grid(alpha=0.3)
	ax2.set_ylabel('回撤比例')
	ax2.set_xlabel('日期')
	ax2.set_ylim([drawdown.min()*1.1, 0])
	
	plt.tight_layout()
	if save_fig:
	plt.savefig('回测结果/绩效曲线.png', dpi=300, bbox_inches='tight')
	plt.show()
	print('绩效曲线已保存到回测结果/绩效曲线.png')
	
	# 运行回测后调用
	# engine.plot_performance(nv_df)

优化效果：
直观展示策略的收益和回撤情况，比干巴巴的数字更容易理解，汇报的时候直接用图就行。

28.1.6 总结：回测系统的应用原则
1.回测不是为了得到高收益，而是为了证伪策略：回测收益高的策略实盘不一定赚钱，但回测收益低的实盘一定赚不到钱，回测的核心是帮你淘汰垃圾策略。
2.宁可回测收益低一点，也不要做任何美化：滑点设高一点，交易成本算足一点，参数少一点，回测越苛刻，实盘越靠谱。
3.样本外测试是试金石：策略在回测区间表现好不算好，在你没调过参数的样本外区间表现好，才是真的好。
4.实盘跟踪至少3个月：回测没问题的策略，先小资金实盘跑3个月，和回测对比偏差在5%以内再上大资金。
通过这一节的学习，你已经掌握了回测系统的核心逻辑，自己写的回测系统每一行都透明，不会有黑箱问题。接下来我们讲如何把回测好的策略对接实盘交易接口，实现自动化交易。
下一节咱们就讲实盘交易接口的对接与订单路由系统开发。

转载请注明出处：https://www.xin3721.com/ArticlePrograme/csharp49728.html