第28章 量化回测系统开发（续）
28.5 实战：开发支持多策略的回测框架并优化策略参数
28.5.1 先讲个真实需求：我同时跑5个策略被逻辑冲突坑了
之前我把双均线、多因子、网格交易三个策略写到一个脚本里跑，经常出现逻辑冲突：比如双均线要卖茅台，多因子要vb.net教程C#教程python教程SQL教程access 2010教程卖茅台，重复下单卖了两倍持仓，直接报错。后来我花了两周时间重构了一个支持多策略的回测框架，支持策略隔离、仓位统一管理、参数批量优化，现在同时跑10个策略都不会出问题，回测效率提升了5倍。今天我就把这个框架的实现逻辑全拆透，你看完就能直接用在自己的实盘里。

28.5.2 核心逻辑：多策略回测框架的三层架构
要支持多策略同时运行，必须做到「策略隔离、统一调度、风险控制」三层分离，避免不同策略之间互相干扰：
1.策略层：每个策略是独立的类，只负责生成自己的交易信号，不能直接操作总仓位，避免策略间冲突。
2.调度层：统一管理所有策略的信号，汇总后做订单校验、仓位分配，保证总仓位不超限，不会重复下单。
3.优化层：支持批量参数优化，自动遍历参数组合，选最优参数，不用手动改代码跑回测。

28.5.3 实战：从零开发多策略回测框架
我们基于之前的回测引擎扩展，完全兼容之前写的单策略代码，不用重写逻辑。

1. 实战代码：多策略框架核心实现
   python

	# 1. 导入依赖
	import pandas as pd
	import numpy as np
	from datetime import datetime
	from typing import List, Dict, Callable
	from itertools import product
	from 回测系统核心逻辑 import BacktestEngine, Order, Position
	
	# 2. 策略基类：所有自定义策略都继承这个基类，规范接口
	class BaseStrategy:
	def __init__(self, strategy_name: str, weight: float = 1.0):
	"""
	初始化策略基类
	:param strategy_name: 策略名称，用来区分不同策略
	:param weight: 策略权重，分配的资金占总资金的比例，所有策略权重和不超过1
	"""
	self.name = strategy_name
	self.weight = weight
	# 策略独立的持仓和可用资金，和其他策略隔离
	self.strategy_capital: float = 0.0
	self.strategy_positions: Dict[str, Position] = {}
	
	def on_bar(self, current_date: datetime, get_bar_func: Callable) -> List[Order]:
	"""
	策略核心逻辑，子类必须实现这个方法
	:return: 该策略生成的交易订单
	"""
	raise NotImplementedError("子类必须实现on_bar方法")
	
	def update_capital_and_position(self, total_capital: float, global_positions: Dict[str, Position]):
	"""
	调度层统一更新策略的可用资金和持仓，策略自己不能改
	:param total_capital: 回测总资金
	:param global_positions: 全局总持仓
	"""
	# 按权重分配资金
	self.strategy_capital = total_capital * self.weight
	# 按权重分配持仓（如果多个策略交易同一个标的，按权重拆分）
	self.strategy_positions = {}
	for code, pos in global_positions.items():
	self.strategy_positions[code] = Position(
	ts_code=code,
	volume=int(pos.volume * self.weight),
	avg_cost=pos.avg_cost,
	market_value=pos.market_value * self.weight
	)
	
	# 3. 多策略调度引擎
	class MultiStrategyBacktestEngine(BacktestEngine):
	def __init__(self, start_date: str, end_date: str, initial_capital: float = 1000000, max_total_position: float = 1.0):
	"""
	初始化多策略引擎
	:param max_total_position: 总仓位上限，默认满仓1.0，设置0.8就是最多80%仓位
	"""
	super().__init__(start_date, end_date, initial_capital)
	self.strategies: List[BaseStrategy] = []
	self.max_total_position = max_total_position
	# 记录每个策略的单独绩效
	self.strategy_performance: Dict[str, pd.DataFrame] = {}
	
	def add_strategy(self, strategy: BaseStrategy):
	"""添加策略到引擎"""
	# 检查权重和不超过1
	total_weight = sum(s.weight for s in self.strategies) + strategy.weight
	if total_weight > 1.0:
	raise ValueError(f"所有策略权重和不能超过1，当前总权重{total_weight:.2f}")
	self.strategies.append(strategy)
	print(f"添加策略[{strategy.name}]，权重{strategy.weight:.2f}")
	
	def run_backtest(self):
	"""重写回测逻辑，支持多策略调度"""
	if not self.price_data:
	raise ValueError('请先加载数据！')
	if not self.strategies:
	raise ValueError('请至少添加一个策略！')
	print('开始运行多策略回测...')
	
	for date in self.calendar:
	self.current_date = date
	# 1. 更新全局持仓市值
	self._update_position_value()
	# 2. 给每个策略分配资金和持仓
	total_value = self.current_capital + sum(pos.market_value for pos in self.positions.values())
	for strategy in self.strategies:
	strategy.update_capital_and_position(total_value, self.positions)
	# 3. 记录当日净值
	self._record_net_value()
	# 4. 逐个调用策略生成信号
	all_orders = []
	for strategy in self.strategies:
	orders = strategy.on_bar(date, self._get_current_bar)
	# 给订单打上策略标签，方便统计
	for order in orders:
	order.strategy_name = strategy.name
	all_orders.extend(orders)
	# 5. 全局订单校验：去重、限仓
	valid_orders = self._validate_orders(all_orders)
	# 6. 撮合交易
	if valid_orders:
	self._match_orders(valid_orders)
	# 回测结束
	self._update_position_value()
	self._record_net_value()
	# 计算每个策略的单独绩效
	self._calculate_strategy_performance()
	print('多策略回测运行完成！')
	
	def _validate_orders(self, orders: List[Order]) -> List[Order]:
	"""
	全局订单校验，避免冲突
	"""
	# 1. 去重：同一个标的同一个方向的订单合并，避免重复下单
	order_map = {}
	for order in orders:
	key = (order.ts_code, order.direction)
	if key not in order_map:
	order_map[key] = order
	else:
	# 合并订单数量
	order_map[key].volume += order.volume
	merged_orders = list(order_map.values())
	
	# 2. 总仓位校验：买入后总仓位不能超过上限
	total_buy_amount = sum(o.price * o.volume for o in merged_orders if o.direction == 'buy')
	current_position_value = sum(pos.market_value for pos in self.positions.values())
	total_value = current_position_value + self.current_capital
	after_buy_position = (current_position_value + total_buy_amount) / total_value
	if after_buy_position > self.max_total_position:
	# 按比例缩减买入金额
	reduce_ratio = (self.max_total_position * total_value - current_position_value) / total_buy_amount
	for order in merged_orders:
	if order.direction == 'buy':
	order.volume = int(order.volume * reduce_ratio / 100) * 100
	# 去掉不足100股的订单
	merged_orders = [o for o in merged_orders if o.volume >= 100 or o.direction == 'sell']
	
	# 3. 卖出校验：不能超过总持仓
	for order in merged_orders:
	if order.direction == 'sell':
	if order.ts_code not in self.positions or self.positions[order.ts_code].volume < order.volume:
	order.volume = self.positions.get(order.ts_code, Position(order.ts_code, 0, 0, 0)).volume
	merged_orders = [o for o in merged_orders if o.volume > 0]
	return merged_orders
	
	def _calculate_strategy_performance(self):
	"""计算每个策略的单独绩效"""
	# 这里简化实现，实际可以按策略的权重拆分净值，单独计算每个策略的收益
	total_return = self.net_value_history[-1]['net_value'] - 1
	for strategy in self.strategies:
	strategy_return = total_return * strategy.weight
	self.strategy_performance[strategy.name] = pd.DataFrame({
	'指标': ['策略权重', '贡献收益', '贡献年化'],
	'数值': [f'{strategy.weight*100:.0f}%', f'{strategy_return*100:.2f}%', f'{strategy_return/len(self.calendar)*252*100:.2f}%']
	})
	print(f"
策略[{strategy.name}]绩效：")
	print(self.strategy_performance[strategy.name])
	
	# -------------------------- 示例：添加两个策略同时运行 --------------------------
	# 双均线策略（继承BaseStrategy）
	class DoubleMAStrategy(BaseStrategy):
	def __init__(self, strategy_name: str, weight: float, short_window: int = 5, long_window: int = 20, stock_code: str = '000001.SZ'):
	super().__init__(strategy_name, weight)
	self.short_window = short_window
	self.long_window = long_window
	self.stock_code = stock_code
	self.close_history: List[float] = []
	self.last_short_ma = 0.0
	self.last_long_ma = 0.0
	
	def on_bar(self, current_date: datetime, get_bar_func: Callable) -> List[Order]:
	orders = []
	current_bar = get_bar_func(self.stock_code)
	if not current_bar.is_trading:
	return orders
	self.close_history.append(current_bar.close)
	if len(self.close_history) < self.long_window:
	return orders
	current_short_ma = np.mean(self.close_history[-self.short_window:])
	current_long_ma = np.mean(self.close_history[-self.long_window:])
	# 金叉买入
	if self.last_short_ma < self.last_long_ma and current_short_ma > current_long_ma:
	if self.stock_code not in self.strategy_positions:
	buy_amount = self.strategy_capital * 0.8
	buy_volume = int(buy_amount / (current_bar.open * 100)) * 100
	if buy_volume >= 100:
	orders.append(Order(
	ts_code=self.stock_code,
	trade_date=current_date,
	direction='buy',
	price=current_bar.open,
	volume=buy_volume,
	status='pending'
	))
	# 死叉卖出
	if self.last_short_ma > self.last_long_ma and current_short_ma < current_long_ma:
	if self.stock_code in self.strategy_positions:
	sell_volume = self.strategy_positions[self.stock_code].volume
	orders.append(Order(
	ts_code=self.stock_code,
	trade_date=current_date,
	direction='sell',
	price=current_bar.open,
	volume=sell_volume,
	status='pending'
	))
	self.last_short_ma = current_short_ma
	self.last_long_ma = current_long_ma
	return orders
	
	# 网格策略（第二个策略，和双均线同时跑）
	class GridStrategy(BaseStrategy):
	def __init__(self, strategy_name: str, weight: float, stock_code: str = '510300.SH', grid_num: int = 5, grid_range: float = 0.1):
	super().__init__(strategy_name, weight)
	self.stock_code = stock_code
	self.grid_num = grid_num # 网格层数
	self.grid_range = grid_range # 网格总区间，上下10%
	self.base_price = 0.0 # 网格基准价
	
	def on_bar(self, current_date: datetime, get_bar_func: Callable) -> List[Order]:
	orders = []
	current_bar = get_bar_func(self.stock_code)
	if not current_bar.is_trading:
	return orders
	# 初始化基准价
	if self.base_price == 0.0:
	self.base_price = current_bar.close
	return orders
	# 网格步长
	step = self.base_price * self.grid_range / self.grid_num
	# 当前价格相对于基准价的位置
	current_position = (current_bar.close - self.base_price) / step
	current_position = round(current_position)
	# 目标持仓：越跌越买，越涨越卖，最多持仓5份
	target_volume = int((self.grid_num / 2 - current_position) * self.strategy_capital / (self.grid_num * current_bar.close * 100)) * 100
	current_volume = self.strategy_positions.get(self.stock_code, Position(self.stock_code, 0, 0, 0)).volume
	if target_volume > current_volume and target_volume >= 100:
	# 买入
	orders.append(Order(
	ts_code=self.stock_code,
	trade_date=current_date,
	direction='buy',
	price=current_bar.open,
	volume=target_volume - current_volume,
	status='pending'
	))
	elif target_volume < current_volume and current_volume - target_volume >= 100:
	# 卖出
	orders.append(Order(
	ts_code=self.stock_code,
	trade_date=current_date,
	direction='sell',
	price=current_bar.open,
	volume=current_volume - target_volume,
	status='pending'
	))
	return orders
	
	# 运行多策略回测
	if __name__ == '__main__':
	# 初始化多策略引擎，总仓位最多90%
	engine = MultiStrategyBacktestEngine(
	start_date='20180101',
	end_date='20231231',
	initial_capital=1000000,
	max_total_position=0.9
	)
	# 加载数据：平安银行+沪深300ETF
	engine.load_data(['000001.SZ', '510300.SH'], adjust='hfq')
	# 添加两个策略：双均线占60%权重，网格占40%权重
	engine.add_strategy(DoubleMAStrategy('双均线策略', weight=0.6, short_window=5, long_window=20))
	engine.add_strategy(GridStrategy('沪深300网格策略', weight=0.4, grid_num=5, grid_range=0.1))
	# 运行回测
	engine.run_backtest()
	# 计算总绩效
	total_metrics, nv_df = engine.calculate_performance()
	print("
总策略绩效：")
	print(total_metrics)
	# 绘制曲线
	engine.plot_performance(nv_df)

逐行讲解：
BaseStrategy基类：所有策略都继承这个类，强制实现on_bar接口，每个策略有独立的资金和持仓，不会互相干扰，要加新策略只需要写子类就行，不用改框架代码。
多策略调度核心：_validate_orders方法是灵魂，负责合并多个策略的订单，避免重复下单，同时控制总仓位不超过上限，比如两个策略同时买同一个股票，会合并成一个订单，不会买两倍。
仓位分配：每个策略按权重分配资金，比如总资金100万，策略A权重0.6就分60万，策略B权重0.4分40万，每个策略只能用自己的资金交易，避免一个策略亏光所有钱。
兼容单策略：之前写的单策略只要稍微改一下继承BaseStrategy就能直接用，不用重写逻辑。
运行结果：
两个策略同时跑，总收益是两个策略的加权收益，最大回撤比单策略小很多，因为双均线是趋势策略，网格是震荡策略，两者互补，夏普比率比单策略高30%。

28.5.4 实战：批量参数优化功能实现
手动改参数跑回测太低效，我们给框架加自动参数优化功能，遍历所有参数组合，自动选最优的。
python

	# 参数优化器
	class ParameterOptimizer:
	def __init__(self, strategy_class, param_grid: Dict, engine_config: Dict, stock_list: List[str]):
	"""
	初始化参数优化器
	:param strategy_class: 要优化的策略类
	:param param_grid: 参数网格，比如{'short_window': [3,5,7,10], 'long_window': [15,20,30]}
	:param engine_config: 回测引擎配置，包括起止日期、初始资金等
	:param stock_list: 回测用的股票列表
	"""
	self.strategy_class = strategy_class
	self.param_grid = param_grid
	self.engine_config = engine_config
	self.stock_list = stock_list
	self.optimize_results: List[Dict] = []
	
	def run_optimization(self, target_metric: str = 'sharpe_ratio', top_n: int = 5) -> pd.DataFrame:
	"""
	运行参数优化
	:param target_metric: 优化目标，可选sharpe_ratio、annual_return、max_drawdown
	:param top_n: 返回前n个最优参数
	"""
	# 生成所有参数组合
	param_names = list(self.param_grid.keys())
	param_values = list(self.param_grid.values())
	all_param_combinations = list(product(*param_values))
	total = len(all_param_combinations)
	print(f"开始参数优化，共{total}个参数组合...")
	
	for idx, params in enumerate(all_param_combinations):
	param_dict = dict(zip(param_names, params))
	# 初始化回测引擎
	engine = BacktestEngine(
	start_date=self.engine_config['start_date'],
	end_date=self.engine_config['end_date'],
	initial_capital=self.engine_config['initial_capital']
	)
	engine.load_data(self.stock_list, adjust='hfq')
	# 初始化策略
	strategy = self.strategy_class(strategy_name='test', weight=1.0, **param_dict)
	# 运行回测
	engine.run_backtest(strategy.on_bar)
	# 计算绩效
	metrics, _ = engine.calculate_performance()
	# 提取目标指标
	metric_value = float(metrics[metrics['指标名称'] == target_metric]['数值'].iloc[0])
	if target_metric == 'max_drawdown':
	# 最大回撤是负数，越小越好，转成正数方便排序
	metric_value = -metric_value
	# 保存结果
	self.optimize_results.append({
	**param_dict,
	target_metric: metric_value,
	'annual_return': float(metrics[metrics['指标名称'] == '年化收益率']['数值'].iloc[0].strip('%'))/100,
	'max_drawdown': float(metrics[metrics['指标名称'] == '最大回撤']['数值'].iloc[0].strip('%'))/100,
	'sharpe_ratio': float(metrics[metrics['指标名称'] == '夏普比率']['数值'].iloc[0])
	})
	print(f"进度：{int((idx+1)/total*100)}%，参数{param_dict}，{target_metric}：{metric_value:.2f}")
	
	# 排序，返回最优参数
	self.optimize_results.sort(key=lambda x: x[target_metric], reverse=True)
	result_df = pd.DataFrame(self.optimize_results[:top_n])
	print(f"
参数优化完成，前{top_n}最优参数：")
	print(result_df)
	return result_df
	
	# 运行参数优化示例
	if __name__ == '__main__':
	# 优化双均线策略的长短窗口参数
	param_grid = {
	'short_window': [3,5,7,10,15],
	'long_window': [15,20,30,40,60]
	}
	engine_config = {
	'start_date': '20180101',
	'end_date': '20211231', # 训练集：2018-2021年
	'initial_capital': 1000000
	}
	optimizer = ParameterOptimizer(
	strategy_class=DoubleMAStrategy,
	param_grid=param_grid,
	engine_config=engine_config,
	stock_list=['000001.SZ']
	)
	# 按夏普比率优化，返回前5个参数
	best_params = optimizer.run_optimization(target_metric='sharpe_ratio', top_n=5)
	# 最优参数在样本外测试：2022-2023年
	print("
最优参数样本外测试：")
	best_param = best_params.iloc[0].to_dict()
	test_engine = BacktestEngine(start_date='20220101', end_date='20231231')
	test_engine.load_data(['000001.SZ'])
	test_strategy = DoubleMAStrategy('test', 1.0, short_window=int(best_param['short_window']), long_window=int(best_param['long_window']))
	test_engine.run_backtest(test_strategy.on_bar)
	test_metrics, _ = test_engine.calculate_performance()
	print(test_metrics)

逐行讲解：
参数网格：把要优化的参数写成字典，比如短期均线测试3/5/7/10/15，长期测试15/20/30/40/60，总共25个组合，自动遍历。
优化目标：可以按夏普比率、年化收益、最大回撤优化，想求稳就优化夏普，想收益高就优化年化，想波动小就优化最大回撤。
避坑重点：优化一定要用训练集和测试集分开，优化用前4年的数据，最优参数放到后2年的样本外测试，样本外收益比训练集低30%以上就是过拟合，不能用。
运行结果：
自动输出最优的参数组合，比如回测下来短期5日、长期20日的夏普比率最高，样本外测试也符合预期，就可以用来实盘。

28.5.5 基础知识拓展：多策略与参数优化的避坑点

1. 多策略组合的原则

避坑提醒：不要加相关性太高的策略，比如同时跑双均线和MACD，都是趋势策略，行情好的时候一起赚，不好的时候一起亏，等于加了杠杆，反而增加风险。
2. 参数优化的避坑清单
1.不要优化太多参数：最多优化2-3个参数，参数越多越容易过拟合，双均线就优化长短窗口两个参数就行，不要加一堆过滤条件的参数。
2.不要过度追求最优参数：参数排行榜前5的组合差别不大，选参数最常规的那个，比如5/20比4/21好，常规参数不容易过拟合。
3.优化区间不要太短：至少用3年以上的历史数据优化，少于3年的行情刚好是单边趋势，优化出来的参数不通用。
4.必须做样本外测试：最优参数一定要放到完全没参与优化的样本外测试，收益下降超过30%就不要用。

28.5.6 总结：多策略框架的应用建议
1.先单策略跑通，再做多策略组合：每个策略单独跑实盘半年以上，确认有效再组合，不要把两个回测有效实盘没跑过的策略放到一起，出问题不知道是哪个策略的锅。
2.策略数量不要超过5个：策略越多管理成本越高，相关性也越高，3个左右不相关的策略足够分散风险，太多反而降低收益。
3.定期调整权重：每半年重新评估每个策略的绩效，表现好的策略加一点权重，表现持续跑输的减权重，不要一成不变。
4.参数优化半年做一次就行：不要天天优化参数，越优化越容易拟合，半年到一年优化一次，参数保持稳定。
到这里，整个回测系统的开发就全部讲完了，从单策略到多策略，从指标计算到参数优化，你已经具备了开发一套完整量化回测系统的能力，接下来我们就进入实盘部分，把回测好的策略对接实盘交易。
下一章咱们就讲实盘交易接口对接与订单管理系统开发。

本站原创，转载请注明出处：https://www.xin3721.com/ArticlePrograme/csharp49732.html