import vectorbt as vbt import pandas as pd import numpy as np import matplotlib.pyplot as plt # 设置中文字体和显示格式 plt.rcParams['font.sans-serif'] = ['SimHei'] plt.rcParams['axes.unicode_minus'] = False pd.set_option('display.float_format', '{:.4f}'.format) stock_00981 = ak.stock_hk_daily(symbol="00981") stock_01347 = ak.stock_hk_daily(symbol="01347") class PairsTradingStrategy: def __init__(self, price_a, price_b, lookback_window=60, k=1.5): """ 初始化配对交易策略 """ self.price_a = price_a self.price_b = price_b self.lookback_window = lookback_window self.k = k def calculate_ratio(self): """计算价格比率 RS = 华虹/中芯""" return self.price_b / self.price_a def generate_signals(self): """生成交易信号""" ratio = self.calculate_ratio() # 计算滚动均值和标准差 ratio_mean = ratio.rolling(window=self.lookback_window).mean() ratio_std = ratio.rolling(window=self.lookback_window).std() # 计算上下轨道 upper_band = ratio_mean + self.k * ratio_std lower_band = ratio_mean - self.k * ratio_std # 生成信号 short_b_long_a = ratio > upper_band # 做空华虹，做多中芯 long_b_short_a = ratio < lower_band # 做多华虹，做空中芯 exit_signal = (ratio <= ratio_mean) & (ratio >= ratio_mean) # 回归均值平仓 return { 'ratio': ratio, 'ratio_mean': ratio_mean, 'upper_band': upper_band, 'lower_band': lower_band, 'short_b_long_a': short_b_long_a, 'long_b_short_a': long_b_short_a, 'exit_signal': exit_signal } def backtest(self, initial_cash=100000, transaction_cost=0.001): """执行回测 - 修复后的版本""" signals = self.generate_signals() # 创建价格DataFrame prices = pd.DataFrame({ 'SMIC': self.price_a, # 中芯国际 'HuaHong': self.price_b # 华虹半导体 }) # 生成仓位信号 positions = self._generate_positions(signals, prices) # 使用vectorbt进行回测 portfolio = vbt.Portfolio.from_holdings( prices, size=positions, # 仓位大小 init_cash=initial_cash, fees=transaction_cost, freq='1D' ) return portfolio, signals def _generate_positions(self, signals, prices): """生成仓位序列""" positions = pd.DataFrame(0, index=prices.index, columns=prices.columns) cash_per_trade = 0.2 # 每次交易使用20%资金 current_position = 0 # 0: 无仓位, 1: 做空华虹做多中芯, -1: 做多华虹做空中芯 for i in range(len(prices)): if i < self.lookback_window: continue current_date = prices.index[i] # 退出信号 if current_position != 0 and signals['exit_signal'].iloc[i]: positions.loc[current_date, :] = 0 current_position = 0 # 开仓信号 - 只在没有持仓时开仓 elif current_position == 0: if signals['short_b_long_a'].iloc[i]: # 做空华虹，做多中芯 positions.loc[current_date, 'HuaHong'] = -cash_per_trade # 做空 positions.loc[current_date, 'SMIC'] = cash_per_trade # 做多 current_position = 1 elif signals['long_b_short_a'].iloc[i]: # 做多华虹，做空中芯 positions.loc[current_date, 'HuaHong'] = cash_per_trade # 做多 positions.loc[current_date, 'SMIC'] = -cash_per_trade # 做空 current_position = -1 # 前向填充仓位，直到下一个信号 positions = positions.replace(0, np.nan).ffill().fillna(0) return positions # 数据预处理函数 def prepare_data(stock_00981, stock_01347): """准备回测数据""" # 复制数据避免修改原数据 smic_data = stock_00981.copy() huahong_data = stock_01347.copy() # 设置日期索引 smic_data['date'] = pd.to_datetime(smic_data['date']) huahong_data['date'] = pd.to_datetime(huahong_data['date']) smic_data = smic_data.set_index('date') huahong_data = huahong_data.set_index('date') # 对齐数据 - 只保留两个股票都有的交易日 common_dates = smic_data.index.intersection(huahong_data.index) smic_aligned = smic_data.loc[common_dates] huahong_aligned = huahong_data.loc[common_dates] return smic_aligned['close'], huahong_aligned['close'] # 执行回测 print("准备数据...") smic_close, huahong_close = prepare_data(stock_00981, stock_01347) print(f"数据时间范围: {smic_close.index.min()} 到 {smic_close.index.max()}") print(f"总交易日数: {len(smic_close)}") print(f"中芯国际价格范围: {smic_close.min():.2f} - {smic_close.max():.2f}") print(f"华虹半导体价格范围: {huahong_close.min():.2f} - {huahong_close.max():.2f}") # 创建策略实例 strategy = PairsTradingStrategy( price_a=smic_close, # 中芯国际 price_b=huahong_close, # 华虹半导体 lookback_window=60, # 60日滚动窗口 k=1.5 # 1.5倍标准差 ) print("执行回测...") portfolio, signals = strategy.backtest( initial_cash=100000, # 初始资金10万 transaction_cost=0.001 # 交易成本0.1% ) # 分析结果 print("\n" + "="*50) print("回测结果分析") print("="*50) # 基本统计 try: print(f"总收益率: {portfolio.total_return():.2%}") print(f"年化收益率: {portfolio.annualized_return():.2%}") print(f"最大回撤: {portfolio.max_drawdown():.2%}") print(f"夏普比率: {portfolio.sharpe_ratio():.2f}") except: print("部分指标计算失败，继续显示其他结果...") # 交易统计 try: stats = portfolio.stats() print(f"总交易次数: {stats['Total Trades']}") print(f"胜率: {stats.get('Win Rate [%]', 'N/A')}%") print(f"盈亏比: {stats.get('Profit Factor', 'N/A')}") except: print("交易统计获取失败") # 可视化结果 fig = plt.figure(figsize=(15, 12)) # 1. 价格比率和交易信号 ax1 = plt.subplot(3, 1, 1) plt.plot(signals['ratio'].index, signals['ratio'].values, label='价格比率(华虹/中芯)', linewidth=1) plt.plot(signals['ratio_mean'].index, signals['ratio_mean'].values, label='滚动均值', linestyle='--', alpha=0.7) plt.plot(signals['upper_band'].index, signals['upper_band'].values, label=f'上轨(μ+{strategy.k}σ)', linestyle='--', color='red', alpha=0.7) plt.plot(signals['lower_band'].index, signals['lower_band'].values, label=f'下轨(μ-{strategy.k}σ)', linestyle='--', color='green', alpha=0.7) # 标记交易信号 short_signals = signals['ratio'][signals['short_b_long_a'] & (signals['ratio'].notna())] long_signals = signals['ratio'][signals['long_b_short_a'] & (signals['ratio'].notna())] if len(short_signals) > 0: plt.scatter(short_signals.index, short_signals.values, color='red', marker='v', s=50, label='做空华虹/做多中芯') if len(long_signals) > 0: plt.scatter(long_signals.index, long_signals.values, color='green', marker='^', s=50, label='做多华虹/做空中芯') plt.title('价格比率与交易信号') plt.legend() plt.grid(True, alpha=0.3) # 2. 累积收益 ax2 = plt.subplot(3, 1, 2) try: portfolio.value.vbt.plot(ax=ax2, label='策略价值') (portfolio.init_cash * (1 + portfolio.returns).cumprod()).vbt.plot(ax=ax2, label='买入持有') plt.title('策略价值 vs 买入持有') plt.legend() plt.grid(True, alpha=0.3) except: plt.text(0.5, 0.5, '收益数据无法显示', ha='center', va='center', transform=ax2.transAxes) # 3. 仓位变化 ax3 = plt.subplot(3, 1, 3) try: portfolio.positions.records_readable['Size'].groupby(portfolio.positions.records_readable['Timestamp']).sum().vbt.plot(ax=ax3) plt.title('仓位变化') plt.grid(True, alpha=0.3) except: # 如果上面的方法失败，使用备选方法 try: portfolio.holdings.vbt.plot(ax=ax3) plt.title('持仓价值') plt.grid(True, alpha=0.3) except: plt.text(0.5, 0.5, '仓位数据无法显示', ha='center', va='center', transform=ax3.transAxes) plt.tight_layout() plt.show() # 显示详细的交易记录 print("\n交易记录详情:") try: trades = portfolio.trades.records_readable if len(trades) > 0: print(trades[['Entry Index', 'Column', 'Size', 'Entry Price', 'Exit Price', 'PnL']].tail(10)) else: print("没有交易记录") except: print("无法获取交易记录") # 参数优化分析 print("\n" + "="*50) print("参数优化分析") print("="*50) # 测试不同的k值 k_values = [1.0, 1.3, 1.5, 1.7, 2.0] results = [] for k in k_values: try: test_strategy = PairsTradingStrategy( price_a=smic_close, price_b=huahong_close, lookback_window=60, k=k ) test_portfolio, _ = test_strategy.backtest(initial_cash=100000) results.append({ 'k': k, '总收益率': test_portfolio.total_return(), '年化收益率': test_portfolio.annualized_return(), '最大回撤': test_portfolio.max_drawdown(), '夏普比率': test_portfolio.sharpe_ratio(), '总交易次数': getattr(test_portfolio.stats(), 'get', lambda x: 'N/A')('Total Trades') }) except Exception as e: print(f"参数k={k}测试失败: {e}") continue if results: results_df = pd.DataFrame(results) print(results_df.round(4)) else: print("所有参数测试都失败了") # 显示价格比率的基本统计信息 print("\n价格比率统计信息:") ratio = signals['ratio'].dropna() print(f"均值: {ratio.mean():.4f}") print(f"标准差: {ratio.std():.4f}") print(f"最小值: {ratio.min():.4f}") print(f"最大值: {ratio.max():.4f}") print(f"当前值: {ratio.iloc[-1]:.4f}") # 显示策略信号统计 print(f"\n策略信号统计:") print(f"做空华虹/做多中芯信号次数: {signals['short_b_long_a'].sum()}") print(f"做多华虹/做空中芯信号次数: {signals['long_b_short_a'].sum()}")