基于akshare的配对交易策略实战解析

1. 配对交易策略从“双胞胎股票”中寻找稳定收益如果你在股市里摸爬滚打过一阵子肯定有过这样的体验看着一只股票涨得欢另一只同行业的股票却趴着不动心里直痒痒。有没有一种方法能让我们不赌大盘涨跌只赚这两只股票之间“价差回归”的钱呢这就是配对交易策略的核心思想。我把它理解成一种“相对价值”投资不关心股票本身的绝对价格是涨是跌只关心两只高度相关的股票比如可口可乐和百事可乐或者工商银行和建设银行之间的价差关系是否稳定。简单来说配对交易就是找到两只历史走势高度同步的股票当它们的价差或者叫“价差序列”因为市场短期情绪而过度拉大时我们就做空涨得多的那只同时做空跌得多的那只赌它们最终会“和好如初”价差会回归到历史平均水平。这个策略的魅力在于它试图构建一个市场中性的投资组合对冲掉大部分市场系统性风险赚取的是统计套利的钱。听起来很美好对吧但实操起来从选股、建模到下单每一步都有不少坑。我自己刚开始做的时候就经常遇到“价差一去不复返”的尴尬局面明明历史走势很像一建仓就分道扬镳了。所以今天我就结合我自己的实战经验手把手带你用 Python 和 akshare 这个免费又强大的数据工具从头构建一个完整的配对交易策略。我们会从最基础的数据获取开始一步步走到策略回测和参数优化。你不用有很深的金融工程背景只要会一点 Python跟着我的思路走就能把这个策略跑起来看看它到底是不是像传说中那么“稳”。2. 实战第一步用 akshare 搞定免费又干净的金融数据工欲善其事必先利其器。做量化交易数据是基石。以前搞数据要么花钱买要么自己写爬虫既麻烦又不稳定。直到我发现了 akshare一个基于 Python 的免费金融数据接口库它整合了国内众多主流财经网站的数据源数据覆盖面广更新也还算及时对于个人研究和小资金实盘来说完全够用。2.1 快速安装与基础数据获取首先我们得把 akshare 请到我们的 Python 环境里。打开你的命令行终端输入下面这行命令通常就能搞定pip install akshare --upgrade有时候网络可能会抽风如果安装慢或者报错可以试试加上国内的镜像源比如清华的pip install akshare --upgrade -i https://pypi.tuna.tsinghua.edu.cn/simple安装成功后我们就能在 Python 脚本里愉快地调用它了。假设我们现在想研究一下 A 股市场上两只银行股——招商银行股票代码 600036和兴业银行股票代码 601166——看看它们是否适合做配对交易。第一步就是获取它们的历史行情数据。import akshare as ak import pandas as pd # 获取招商银行的后复权日线数据 stock_zsyh ak.stock_zh_a_hist(symbol600036, adjusthfq) # 获取兴业银行的后复权日线数据 stock_xyyh ak.stock_zh_a_hist(symbol601166, adjusthfq) print(stock_zsyh.head()) # 看看数据长什么样 print(stock_xyyh.head())运行上面这段代码你会得到两个 DataFrame里面包含了日期、开盘价、收盘价、最高价、最低价和成交量等字段。这里我特意用了adjusthfq参数来获取后复权数据这非常重要。后复权价格考虑了分红、送股等事件对股价的影响能真实反映股票的历史收益是做长期分析和配对交易的基础。如果你用前复权或者不复权的价格价差序列可能会因为除权除息而产生剧烈的、非交易性的跳空这会严重干扰我们的模型。2.2 数据清洗与格式化为 Backtrader 做好准备从 akshare 拿到的数据已经很规整了但为了喂给后续的回测框架 Backtrader我们还需要做一些简单的“美容”工作。Backtrader 对数据的格式有特定要求尤其是日期索引。# 以招商银行为例进行数据清洗 def clean_data(df): # 通常 akshare 返回的列名是中文我们重命名为英文方便后续处理 df.rename(columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume }, inplaceTrue) # 将日期列转换为 datetime 类型并设置为索引 df[date] pd.to_datetime(df[date]) df.set_index(date, inplaceTrue) # 确保数据按日期升序排列 df.sort_index(inplaceTrue) # 只保留我们需要的OHLCV列 df df[[open, high, low, close, volume]] return df stock_zsyh_cleaned clean_data(stock_zsyh.copy()) stock_xyyh_cleaned clean_data(stock_xyyh.copy()) print(stock_zsyh_cleaned.head())这一步完成后我们就得到了两个干净整洁的、以日期为索引的 Pandas DataFrame。你可能会问如果两只股票的交易日不完全一致怎么办比如一只股票因为停牌缺失了几天数据。这是个好问题也是实战中必须处理的细节。一个简单有效的方法是在将数据传入回测引擎前先使用 Pandas 的merge或join方法按照日期索引进行对齐只保留两只股票都有交易数据的共同交易日。这样可以避免在计算价差或回归时出现数据错位。3. 策略核心如何科学地“配对”与计算信号数据准备好了接下来就是策略最核心的部分怎么判断两只股票是“天生一对”以及什么时候该出手交易这里我们主要依赖两个统计学工具线性回归和 Z-Score。3.1 寻找“灵魂伴侣”协整性检验与价差序列构建不是随便抓两只同行业的股票就能做配对交易。它们必须满足“协整”关系。你可以把它想象成一对被绳子拴住的小狗每只小狗都可以自己跑来跑去价格各自波动但绳子长度长期均衡关系限制了它们分开的最大距离。协整关系保证了价差序列是平稳的会围绕一个均值上下波动这才是我们套利的基础。在实操中我们通常先用肉眼观察两只股票价格走势图是否相似然后再用统计方法验证。更常用的一个初步判断方法是计算滚动相关系数。我们可以用 Pandas 快速计算两只股票收盘价的滚动相关性# 计算60个交易日的滚动相关系数 rolling_corr stock_zsyh_cleaned[close].rolling(window60).corr(stock_xyyh_cleaned[close]) rolling_corr.plot(title60-Day Rolling Correlation between 600036 and 601166)如果滚动相关系数长期稳定在较高水平比如0.7以上说明它们短期走势联动性强是个不错的候选对。但这还不够我们需要构建价差序列。最经典的方法是使用线性回归。我们以股票A的价格为自变量X股票B的价格为因变量Y进行回归Y alpha beta * X epsilon。这里的残差项epsilon也就是实际价格与回归模型预测价格的差值就是我们的价差序列Spread。import statsmodels.api as sm # 假设我们决定用招商银行X来解释兴业银行Y X stock_zsyh_cleaned[close].values Y stock_xyyh_cleaned[close].values # 添加常数项截距 X sm.add_constant(X) # 进行普通最小二乘OLS回归 model sm.OLS(Y, X).fit() # 获取回归系数 beta 和 alpha beta, alpha model.params[1], model.params[0] # 计算价差序列Y - (alpha beta * X) spread Y - (alpha beta * X[:, 1])现在我们就得到了一个价差序列spread。理想情况下这个序列应该看起来像一条围绕0值上下波动的曲线。3.2 交易信号的生成器Z-Score 标准化得到了价差序列我们怎么定义“价差过大”或“价差过小”呢直接用价差的绝对值是不行的因为不同股票对的价差波动幅度天差地别。这里就需要引入Z-Score标准化。Z-Score 衡量的是当前价差偏离其历史均值的程度单位是历史标准差。计算公式很简单Z (当前价差 - 历史N期价差均值) / 历史N期价差标准差。这个 Z 值就是我们最终的交易信号。我通常把计算 Z-Score 的窗口期就是上面公式里的 N设为20到60个交易日具体需要优化。def calculate_zscore(spread_series, window30): 计算价差序列的滚动Z-Score mean spread_series.rolling(windowwindow).mean() std spread_series.rolling(windowwindow).std() zscore (spread_series - mean) / std return zscore # 将价差序列转为Pandas Series spread_series pd.Series(spread, indexstock_zsyh_cleaned.index) # 计算滚动Z-Score zscore_series calculate_zscore(spread_series, window30) zscore_series.plot(titleZ-Score of the Spread)Z-Score 图能直观地告诉我们交易时机。业界常用的交易阈值是 ±2。也就是说当 Z 2 时我们认为价差过大未来会收敛。此时应该做空价差卖出相对高估的股票价差为正说明Y实际值高于模型预测值即Y相对贵买入相对低估的股票X相对便宜。当 Z -2 时我们认为价差过小负得太多未来会扩大。此时应该做多价差买入相对低估的股票Y卖出相对高估的股票X。当 Z 回归到0附近比如 -0.5 Z 0.5时我们平掉所有头寸锁定利润。在实际代码中我们还需要考虑一个“开仓状态”的管理避免在Z值持续大于2时反复做空。这就是原始代码中status这个参数的作用它记录我们当前是处于空头价差status1、多头价差status2还是无仓位status0状态。4. 策略实现用 Backtrader 搭建完整的回测引擎理论部分搞清楚了现在我们把所有零件组装起来用 Backtrader 这个专业的回测框架来构建我们的策略。Backtrader 的好处是事件驱动更贴近真实交易而且能方便地管理多数据源、订单和仓位。4.1 策略类的完整结构解析下面我结合原始代码把关键部分拆开揉碎了讲。首先我们定义一个继承自bt.Strategy的策略类。import backtrader as bt import backtrader.indicators as btind class PairTradingStrategy(bt.Strategy): # 参数定义方便我们后续优化 params dict( period30, # 计算Z-Score的滚动窗口期 stake10, # 默认交易股数未使用我们动态计算 qty10, qty20, # 初始持仓数量 printoutTrue, # 是否打印日志 upper2.0, # Z-Score开仓上阈值 lower-2.0, # Z-Score开仓下阈值 up_medium0.5, # Z-Score平仓上界 low_medium-0.5, # Z-Score平仓下界 status0, # 仓位状态0无仓1空价差2多价差 portfolio_value100000, # 用于计算开仓数量的组合价值 ) def __init__(self): # 初始化指标计算data0和data1的OLS回归变换及Z-Score # 这里data0和data1是我们在主函数中按顺序加入Cerebro的数据 self.transform btind.OLS_TransformationN(self.data0, self.data1, periodself.p.period) self.zscore self.transform.zscore # 这就是我们需要的Z-Score序列 # 初始化仓位状态和动态计算的开仓数量 self.status self.p.status self.qty1 self.p.qty1 self.qty2 self.p.qty2这里用到了 Backtrader 内置的OLS_TransformationN指标它帮我们封装了滚动回归和 Z-Score 的计算非常方便。如果没有这个指标我们就得自己写循环计算麻烦很多。4.2 下单逻辑与仓位管理的细节策略的核心逻辑都在next函数里这个函数在每个交易日或每个Bar都会被调用。我们需要在这里检查当前的 Z-Score并根据规则决定买卖。def next(self): # 如果有未完成的订单则跳过本次逻辑 if self.orderid: return # 获取最新的Z-Score值 current_z self.zscore[0] # 情况1Z 上阈值且当前不是空价差状态 - 开空价差仓 if current_z self.p.upper and self.status ! 1: # 动态计算开仓数量将一半资金分配给每只股票 cash_per_leg 0.5 * self.p.portfolio_value # 计算能买多少股取整 size1 int(cash_per_leg / self.data0.close[0]) # 股票1数量 size2 int(cash_per_leg / self.data1.close[0]) # 股票2数量 # 下单卖空相对高估的股票(data0)买入相对低估的股票(data1) # 注意这里假设价差由 data0 和 data1 构成且当前Z0意味着 data0 相对 data1 被高估 self.sell(dataself.data0, sizesize1 self.qty1) # 卖出高估股 self.buy(dataself.data1, sizesize2 self.qty2) # 买入低估股 # 更新持仓状态和数量 self.qty1 size1 self.qty2 size2 self.status 1 # 标记状态为“空价差” self.log(fOPEN SHORT SPREAD. Z{current_z:.2f}) # 情况2Z 下阈值且当前不是多价差状态 - 开多价差仓 elif current_z self.p.lower and self.status ! 2: cash_per_leg 0.5 * self.p.portfolio_value size1 int(cash_per_leg / self.data0.close[0]) size2 int(cash_per_leg / self.data1.close[0]) # 下单买入相对低估的股票(data0)卖空相对高估的股票(data1) self.buy(dataself.data0, sizesize1 self.qty1) # 买入低估股 self.sell(dataself.data1, sizesize2 self.qty2) # 卖出高估股 self.qty1 size1 self.qty2 size2 self.status 2 # 标记状态为“多价差” self.log(fOPEN LONG SPREAD. Z{current_z:.2f}) # 情况3Z值回归到平仓区间内 - 平掉所有仓位 elif self.p.low_medium current_z self.p.up_medium: if self.status ! 0: # 如果有仓位才平仓 self.close(dataself.data0) self.close(dataself.data1) self.status 0 self.qty1 0 self.qty2 0 self.log(fCLOSE ALL. Z{current_z:.2f})这里有几个关键点需要理解。第一开仓数量是动态计算的基于固定的portfolio_value和当前股价确保每次开仓投入的资金量大致相同。第二self.qty1和self.qty2记录了理论上的持仓数量用于在下次开同向仓位时进行累加虽然原始策略可能更倾向于每次全平全开。第三status状态管理至关重要它防止了在趋势行情中Z值持续超阈值的反复开仓。4.3 组装并运行回测让策略“活”起来策略类写好了我们需要一个“大脑”Cerebro来协调数据、策略、经纪商Broker等所有组件。def run_backtest(stock1_code, stock2_code, start_date, end_date, initial_cash100000): cerebro bt.Cerebro() # 创建回测引擎 cerebro.broker.setcash(initial_cash) # 设置初始资金 cerebro.broker.setcommission(commission0.0005) # 设置交易佣金例如万五 # 使用前面写好的函数通过akshare获取并清洗数据 data1 get_akshare_data(stock1_code, start_date, end_date) data2 get_akshare_data(stock2_code, start_date, end_date) # 将数据加载到引擎中注意顺序data0, data1 cerebro.adddata(data1) # 这将成为策略中的 self.data0 cerebro.adddata(data2) # 这将成为策略中的 self.data1 # 添加我们写好的策略并传入参数 cerebro.addstrategy(PairTradingStrategy, period30, upper2.0, lower-2.0, portfolio_valueinitial_cash) print(启动资金: %.2f % cerebro.broker.getvalue()) cerebro.run() # 运行回测 print(期末资金: %.2f % cerebro.broker.getvalue()) # 绘制回测结果图 cerebro.plot(stylecandlestick, volumeFalse)把股票代码、回测日期区间和初始资金传进去运行这个函数你就能看到策略在整个历史区间内的表现包括资金曲线、买卖点标记等。第一次看到自己的策略在历史数据上跑出曲线那种感觉还是挺奇妙的。5. 策略优化与实战中的那些“坑”回测曲线看起来不错是不是就可以直接实盘了千万别急回测Backtest和实盘Live Trading之间隔着一个“银河系”。下面我分享几个自己踩过坑才总结出来的优化点和注意事项。5.1 关键参数如何优化不只是 Z 值阈值原始策略给了我们一组默认参数但这对你的股票对未必是最优的。最重要的三个参数是计算 Z-Score 的滚动窗口期period、开仓阈值upper/lower和平仓阈值up_medium/low_medium。滚动窗口期period这个值太小Z-Score 对噪声敏感会频繁交易太大则反应迟钝错过机会。我通常会在20到120之间进行测试。一个实用的方法是观察价差序列的自相关性选择一个能覆盖主要波动周期的长度。开仓阈值±2是经典值但市场波动率会变。在波动大的时期阈值可以设宽一些如±2.5以避免被“打脸”在波动小的时期可以设窄一些如±1.5以捕捉更多机会。你可以用历史波动率来动态调整但这会复杂很多。平仓阈值设置在0附近如±0.5是为了尽快获利了结。但有时价差回归到0后还会继续朝有利方向运动平得太早会损失利润。可以尝试更窄的平仓区间如±0.2或者结合其他指标比如当Z值穿过0轴并反向突破某个小阈值时再平仓。如何优化老老实实做参数扫描。写一个循环遍历不同的参数组合跑回测记录夏普比率、最大回撤、总收益率等指标。然后用热力图可视化找出在历史数据上表现稳健的“甜点”区域。记住不要过度优化去拟合历史数据要选择参数敏感性较低、表现稳定的区域。5.2 实盘必须考虑的滑点、流动性与手续费回测是理想世界实盘是残酷现实。最大的几个敌人是滑点Slippage你以为能以收盘价成交实际可能差几分钱。对于配对交易这种赚取微小价差的策略滑点的侵蚀效果非常显著。在 Backtrader 中可以通过cerebro.broker.set_slippage_perc()或set_slippage_fixed()来设置滑点模型让你的回测更接近现实。流动性你选的股票是否足够活跃如果一只股票成交量很小你的大单可能会显著影响市场价格导致成交价远差于预期甚至无法及时平仓。在选择股票对时一定要把日均成交额作为一个硬性筛选条件。手续费与冲击成本原始代码设置了佣金但没考虑印花税卖出时收取和冲击成本。A股印花税是千分之一这是一笔不小的固定成本。在计算预期收益时一定要把这些成本都算进去。冲击成本对于大资金尤其致命你可能需要将大单拆分成小单分批交易。5.3 策略失效与动态再平衡没有永远有效的策略。两只股票的协整关系可能会因为公司基本面变化、行业格局重塑而永久性打破。这就是所谓的“价差扩散”风险是配对交易最大的敌人。我遇到过最惨的一次是两只地产股配对结果其中一只暴雷价差一路扩大永不回头双边头寸都亏损。怎么办定期回顾不要以为建仓后就一劳永逸。至少每月检查一次价差序列的平稳性可以用ADF检验和滚动相关性。如果协整关系明显减弱要果断离场。设置硬性止损除了基于Z-Score的逻辑止损一定要设置一个绝对价差止损线。比如当价差超过历史平均值的3.5个标准差时无条件平仓承认这次配对可能已经失效。动态再平衡回归系数beta不是一成不变的。可以考虑每季度或每半年用最近一段时期的数据重新跑一次回归更新你的价差计算公式。但这会带来更高的交易成本和模型风险需要权衡。最后我想说配对交易是一个经典的统计套利思路它教给我们的是如何用数学工具去捕捉市场中的相对定价错误。用 akshare 和 Backtrader 把它实现出来是一个非常好的学习和练手过程。但请务必记住所有回测结果都是“过去式”实盘交易充满不确定性。从小资金开始充分理解策略的每一个环节和潜在风险比盲目追求高回测收益率重要得多。我自己也是交了学费才明白在量化交易里控制风险永远比追逐利润更优先。希望这篇长文能帮你少走一些弯路亲手搭建起属于自己的第一个配对交易模型。