当前位置：首页 > news >正文

用AKShare和Backtrader实现股票配对交易策略：从数据获取到回测全流程

news 2026/3/26 21:53:31

用AKShare和Backtrader实现股票配对交易策略：从数据获取到回测全流程

在量化交易领域，配对交易（Pairs Trading）是一种经典的市场中性策略，它通过寻找历史价格走势高度相关的两只股票，当它们的价差偏离正常范围时进行交易，待价差回归时获利。本文将详细介绍如何利用Python生态中的AKShare金融数据接口和Backtrader回测框架，从零开始构建一个完整的配对交易策略。

1. 配对交易基础与原理

配对交易的核心思想源于统计套利。当两只具有长期均衡关系的股票价格出现短期偏离时，做空相对强势的股票，同时做多相对弱势的股票，等待价差回归后平仓获利。这种策略的优势在于：

市场中性：多空头寸相互对冲，降低市场系统性风险
收益稳定：依赖统计规律而非市场方向判断
适用性广：可在牛市、熊市和震荡市中应用

关键概念解析：

术语	说明	计算公式
价差(Spread)	两只股票价格的线性组合	Price_A - β * Price_B
Z-score	标准化后的价差	(Spread - 均值) / 标准差
协整检验	检验两只股票是否存在长期均衡关系	ADF检验、Johansen检验

注意：成功的配对交易依赖于选取具有稳定统计关系的股票对，这需要通过严格的统计检验而非主观判断。

2. 环境准备与数据获取

2.1 安装必要库

pip install akshare backtrader pandas numpy statsmodels

2.2 使用AKShare获取股票数据

AKShare提供了丰富的A股市场数据接口。以下代码演示如何获取两只股票的历史数据：

import akshare as ak import pandas as pd # 获取贵州茅台(600519)和五粮液(000858)的复权历史数据 maotai = ak.stock_zh_a_hist(symbol="600519", adjust="hfq") wuliangye = ak.stock_zh_a_hist(symbol="000858", adjust="hfq") # 数据清洗与格式化 def clean_data(df): df = df.iloc[:, :6] df.columns = ['date', 'open', 'close', 'high', 'low', 'volume'] df['date'] = pd.to_datetime(df['date']) df.set_index('date', inplace=True) return df maotai_clean = clean_data(maotai) wuliangye_clean = clean_data(wuliangye)

2.3 数据预处理与协整检验

在构建配对交易策略前，必须验证两只股票是否存在统计上的协整关系：

from statsmodels.tsa.stattools import coint # 合并两只股票的收盘价 merged = pd.merge(maotai_clean['close'], wuliangye_clean['close'], left_index=True, right_index=True, suffixes=('_mt', '_wly')) # 协整检验 score, pvalue, _ = coint(merged['close_mt'], merged['close_wly']) print(f'协整检验p值: {pvalue:.4f}') # p值<0.05表示存在协整关系

3. 策略设计与实现

3.1 Backtrader框架基础

Backtrader是一个功能强大的Python回测框架，其主要组件包括：

Cerebro：策略回测引擎
Strategy：交易策略基类
Data Feed：市场数据接口
Analyzer：绩效分析工具

3.2 配对交易策略核心逻辑

import backtrader as bt import backtrader.indicators as btind class PairTradingStrategy(bt.Strategy): params = ( ('period', 10), # 计算Z-score的窗口期 ('upper', 2.0), # 上轨阈值 ('lower', -2.0), # 下轨阈值 ('printlog', True), # 是否打印交易日志 ) def __init__(self): # 计算价差和Z-score self.spread = self.data0.close - self.data1.close self.zscore = (self.spread - btind.SMA(self.spread, period=self.p.period)) / btind.StdDev(self.spread, period=self.p.period) # 跟踪持仓状态 self.position_status = 0 # 0:无持仓, 1:做空价差, 2:做多价差 def next(self): if self.position_status == 0: # 无持仓时寻找交易机会 if self.zscore[0] > self.p.upper: # Z-score突破上轨，做空价差 self.sell(data=self.data0, size=100) # 做空股票A self.buy(data=self.data1, size=100) # 做多股票B self.position_status = 1 elif self.zscore[0] < self.p.lower: # Z-score突破下轨，做多价差 self.buy(data=self.data0, size=100) # 做多股票A self.sell(data=self.data1, size=100) # 做空股票B self.position_status = 2 elif self.position_status == 1 and self.zscore[0] < 0: # 平仓做空头寸 self.close(data=self.data0) self.close(data=self.data1) self.position_status = 0 elif self.position_status == 2 and self.zscore[0] > 0: # 平仓做多头寸 self.close(data=self.data0) self.close(data=self.data1) self.position_status = 0

3.3 策略参数优化

通过网格搜索寻找最优参数组合：

class OptimizeStrategy(PairTradingStrategy): params = ( ('period', range(5, 21, 5)), # 测试5-20天的窗口 ('upper', [1.5, 2.0, 2.5]), # 上轨阈值选项 ('lower', [-1.5, -2.0, -2.5]), # 下轨阈值选项 ) cerebro = bt.Cerebro() cerebro.optstrategy(OptimizeStrategy) results = cerebro.run()

4. 回测执行与绩效分析

4.1 配置回测环境

def run_backtest(): cerebro = bt.Cerebro() # 添加数据 data0 = bt.feeds.PandasData(dataname=maotai_clean) data1 = bt.feeds.PandasData(dataname=wuliangye_clean) cerebro.adddata(data0) cerebro.adddata(data1) # 添加策略 cerebro.addstrategy(PairTradingStrategy) # 设置初始资金和佣金 cerebro.broker.setcash(100000.0) cerebro.broker.setcommission(commission=0.001) # 0.1%佣金 # 添加分析器 cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe') cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown') # 运行回测 results = cerebro.run() # 打印结果 strat = results[0] print('夏普比率:', strat.analyzers.sharpe.get_analysis()['sharperatio']) print('最大回撤:', strat.analyzers.drawdown.get_analysis()['max']['drawdown']) # 绘制结果 cerebro.plot(style='candlestick')