9.1 案例背景

今天我们将通过一个综合案例，将前 8 天的知识综合应用。我们将构建一个基于动量和均值回归的多策略组合，并在实盘环境中运行。

9.2 数据准备

我们将使用 AkShare 获取浦**行（60**00）的历史数据，并进行预处理。

获取数据：

import akshare as ak 
import pandas as pd 
# 获取浦**行的历史数据
stock_data = ak.stock_zh_a_hist( symbol="600000", period="daily", start_date="2020-01-01", end_date="2023-01-01" )
# 数据预处理
stock_data.columns = ['date', 'open', 'close', 'high', 'low', 'volume', 'amount', 'amplitude', '涨跌幅', '换手率'] 
stock_data['date'] = pd.to_datetime(stock_data['date']) 
stock_data.set_index('date', inplace=True) 
stock_data.fillna(method='ffill', inplace=True)

9.3 策略开发

我们将开发两个策略：趋势跟踪策略和均值回归策略，并将它们组合在一起。

趋势跟踪策略：

class TrendFollowingStrategy(bt.Strategy): 
				params = (('period', 20),) 
				def __init__(self): 
        				self.sma = bt.indicators.SimpleMovingAverage(self.data.close, period=self.params.period) 
				def next(self): 
        				if self.data.close[0] > self.sma[0] and not self.position: 
                				self.buy() 
								elif self.data.close[0] < self.sma[0] and self.position: 
                				self.sell()

均值回归策略：

class MeanReversionStrategy(bt.Strategy):
				params = (('period', 14), ('threshold', 2)) 
				def __init__(self): 
        				self.sma = bt.indicators.SimpleMovingAverage(self.data.close, period=self.params.period) 
								self.std = bt.indicators.StandardDeviation(self.data.close, period=self.params.period) 
				def next(self): 
        				upper_band = self.sma[0] + self.std[0] * self.params.threshold 
								lower_band = self.sma[0] - self.std[0] * self.params.threshold 
								if self.data.close[0] < lower_band and not self.position: 
                				self.buy() 
								elif self.data.close[0] > upper_band and self.position:
                				self.sell()

9.4 回测与优化

我们将对策略进行回测，并优化参数。

运行回测：

import backtrader as bt 
# 初始化 
Cerebro cerebro = bt.Cerebro() 
# 添加策略 
cerebro.addstrategy(TrendFollowingStrategy) 
cerebro.addstrategy(MeanReversionStrategy) 
# 加载数据
data = bt.feeds.PandasData(dataname=stock_data)
cerebro.adddata(data) 
# 设置初始资金和佣金
cerebro.broker.setcash(100000.0) 
cerebro.broker.setcommission(commission=0.001) 
# 添加分析器
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe') 
cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown') 
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trade') 
# 运行回测
results = cerebro.run() 
# 打印分析结果 
sharpe = results[0].analyzers.sharpe.get_analysis() 
drawdown = results[0].analyzers.drawdown.get_analysis() 
trade = results[0].analyzers.trade.get_analysis()
print(f"夏普比率: {sharpe['sharperatio']}") 
print(f"最大回撤: {drawdown['max']['drawdown']:.2f}%") 
print(f"总交易次数: {trade.total.closed}") 
print(f"盈利交易次数: {trade.won.total}") 
print(f"亏损交易次数: {trade.lost.total}") 
print(f"胜率: {trade.won.total / trade.total.closed:.2%}")

参数优化：

cerebro.optstrategy(TrendFollowingStrategy, period=range(10, 30, 5)) 
cerebro.optstrategy(MeanReversionStrategy, period=range(10, 30, 5), threshold=range(1, 3))
results = cerebro.run() 
for result in results: 
				for param in result: 
        				print(f"趋势跟踪周期: {param.params.period}") 
								print(f"均值回归周期: {param.params.period}") 
								print(f"阈值: {param.params.threshold}") 
								print(f"夏普比率: {param.analyzers.sharpe.get_analysis()['sharperatio']}")

9.5 风险控制与实盘部署

添加风险控制：

class RiskControlStrategy(bt.Strategy): 
				params = (('max_drawdown', 0.1),) 
				def __init__(self): 
        				self.peak_value = self.broker.getvalue() 
				def next(self): 
        				current_value = self.broker.getvalue()
								drawdown = (self.peak_value - current_value) / self.peak_value 
								if drawdown > self.params.max_drawdown: 
                				self.order = self.close() 
												# 平仓 
                        self.peak_value = current_value

实盘部署：

from ib_insync import IB, Stock 
# 连接 Interactive 
Brokers ib = IB() 
ib.connect('127.0.0.1', 7497, clientId=1) 
# 定义数据接口
class IBData(bt.feeds.PandasData):
				params = ( ('datetime', 0), ('open', 1), ('high', 2), ('low', 3), ('close', 4), ('volume', 5), ) 
	
# 初始化
Cerebro cerebro = bt.Cerebro() 
# 添加策略 
cerebro.addstrategy(TrendFollowingStrategy) 
cerebro.addstrategy(MeanReversionStrategy) 
cerebro.addstrategy(RiskControlStrategy) 
# 加载数据
data = IBData(dataname=stock_data)
cerebro.adddata(data)
# 设置经纪商
cerebro.setbroker(ibbroker.IBBroker(ib)) 
# 设置初始资金 
cerebro.broker.setcash(100000.0) 
# 运行实盘交易 
cerebro.run()

9.6 总结

今天通过一个综合案例，将前 8 天的知识综合应用：

使用 AkShare 获取并预处理 A 股数据。
开发了趋势跟踪和均值回归两个策略，并将它们组合在一起。
对策略进行了回测和参数优化。
添加了风险控制，并将策略部署到实盘交易中。

通过这个案例，我们掌握了从数据加载到实盘交易的完整流程，为后续的量化交易实践打下了坚实的基础。明天我们将总结整个课程，并规划下一步的学习方向。

泽兴芝士网

一站式 IT 编程学习资源平台

第 9 天:综合案例（综合案例分析）