quant/PairsTrading.py

import vectorbt as vbt
import akshare as ak
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import datetime
from numba import njit
from collections import namedtuple

# 设置中文显示
plt.rcParams['font.sans-serif'] = ['SimHei']
plt.rcParams['axes.unicode_minus'] = False

# 获取股票数据
print("正在获取股票数据...")
stock_00981 = ak.stock_hk_daily(symbol="00981")
stock_01347 = ak.stock_hk_daily(symbol="01347")

print("中芯国际数据列名:", stock_00981.columns.tolist())
print("华虹半导体数据列名:", stock_01347.columns.tolist())

# 数据预处理
def preprocess_data(df, symbol):
    """预处理股票数据"""
    df = df.copy()
    # 检查列名并重命名（如果需要）
    if 'date' in df.columns:
        df['date'] = pd.to_datetime(df['date'])
        df.set_index('date', inplace=True)
    elif '日期' in df.columns:
        df['date'] = pd.to_datetime(df['日期'])
        df.set_index('date', inplace=True)
        # 重命名中文列为英文
        rename_dict = {
            '开盘': 'open',
            '最高': 'high', 
            '最低': 'low',
            '收盘': 'close',
            '成交量': 'volume'
        }
        df = df.rename(columns=rename_dict)
    else:
        # 如果已经有英文列名，直接使用
        df.index = pd.to_datetime(df.index)
    
    df = df.sort_index()
    return df[['open', 'high', 'low', 'close', 'volume']]

# 预处理数据
smic_data = preprocess_data(stock_00981, "00981")
hhic_data = preprocess_data(stock_01347, "01347")

print(f"中芯国际原始数据时间范围: {smic_data.index.min()} 到 {smic_data.index.max()}")
print(f"华虹半导体原始数据时间范围: {hhic_data.index.min()} 到 {hhic_data.index.max()}")

# 限制为最近三年数据
end_date = smic_data.index.max()
start_date = end_date - pd.Timedelta(days=3*365)  # 最近三年

print(f"\n限制回测时间范围: {start_date} 到 {end_date}")

smic_data = smic_data.loc[start_date:end_date]
hhic_data = hhic_data.loc[start_date:end_date]

print(f"限制后中芯国际数据形状: {smic_data.shape}")
print(f"限制后华虹半导体数据形状: {hhic_data.shape}")

# 对齐数据时间索引
common_index = smic_data.index.intersection(hhic_data.index)
smic_data = smic_data.loc[common_index]
hhic_data = hhic_data.loc[common_index]

print(f"对齐后数据时间范围: {common_index.min()} 到 {common_index.max()}")
print(f"总交易日数: {len(common_index)}")

if len(common_index) == 0:
    print("错误: 没有共同交易日数据!")
    exit()

# 配对交易参数
PERIOD = 30  # 回看周期
UPPER = 2.0  # 上界
LOWER = -2.0  # 下界
ORDER_PCT1 = 0.1  # 中芯国际交易比例
ORDER_PCT2 = 0.1  # 华虹半导体交易比例
COMMPERC = 0.002  # 手续费率 0.2%
MODE = 'OLS'  # 使用OLS方法
INITIAL_CASH = 100000  # 初始资金

# 定义数据结构
Memory = namedtuple("Memory", ('spread', 'zscore', 'status'))
Params = namedtuple("Params", ('period', 'upper', 'lower', 'order_pct1', 'order_pct2'))

@njit
def ols_spread_nb(a, b):
    """计算OLS价差"""
    a = np.log(a)
    b = np.log(b)
    _b = np.vstack((b, np.ones(len(b)))).T
    slope, intercept = np.dot(np.linalg.inv(np.dot(_b.T, _b)), np.dot(_b.T, a))
    spread = a - (slope * b + intercept)
    return spread[-1]

@njit
def pre_group_func_nb(c, _period, _upper, _lower, _order_pct1, _order_pct2):
    """准备当前组（资产对）"""
    assert c.group_len == 2
    
    # 初始化内存数组
    spread = np.full(c.target_shape[0], np.nan, dtype=np.float64)
    zscore = np.full(c.target_shape[0], np.nan, dtype=np.float64)
    status = np.full(1, 0, dtype=np.int64)
    memory = Memory(spread, zscore, status)
    
    # 简化参数处理 - 直接使用标量值
    period = _period
    upper = _upper
    lower = _lower
    order_pct1 = _order_pct1
    order_pct2 = _order_pct2
    
    params = Params(period, upper, lower, order_pct1, order_pct2)
    
    # 创建仓位大小数组
    size = np.empty(c.group_len, dtype=np.float64)
    
    return (memory, params, size)

@njit
def pre_segment_func_nb(c, memory, params, size, mode):
    """准备当前段（组内行）"""
    
    # 等待足够的数据
    if c.i < params.period - 1:
        size[0] = np.nan
        size[1] = np.nan
        return (size,)
    
    # 计算窗口切片
    window_slice = slice(max(0, c.i + 1 - params.period), c.i + 1)
    
    # 根据模式计算价差
    if mode == 'OLS':
        a = c.close[window_slice, c.from_col]      # 中芯国际
        b = c.close[window_slice, c.from_col + 1]  # 华虹半导体
        memory.spread[c.i] = ols_spread_nb(a, b)
    elif mode == 'log_return':
        # 对数收益率方法
        logret_a = np.log(c.close[c.i, c.from_col] / c.close[c.i - 1, c.from_col])
        logret_b = np.log(c.close[c.i, c.from_col + 1] / c.close[c.i - 1, c.from_col + 1])
        memory.spread[c.i] = logret_a - logret_b
    else:
        raise ValueError("Unknown mode")
    
    # 计算z-score
    spread_mean = np.mean(memory.spread[window_slice])
    spread_std = np.std(memory.spread[window_slice])
    memory.zscore[c.i] = (memory.spread[c.i] - spread_mean) / spread_std
    
    # 使用前一个z-score生成交易信号（避免未来数据）
    if c.i > 0 and not np.isnan(memory.zscore[c.i - 1]):
        # 做空信号：z-score > 上界 且 当前不是做空状态
        if memory.zscore[c.i - 1] > params.upper and memory.status[0] != 1:
            size[0] = -params.order_pct1  # 卖空中芯国际
            size[1] = params.order_pct2   # 买入华虹半导体
            # 执行顺序：先卖后买
            c.call_seq_now[0] = 0
            c.call_seq_now[1] = 1
            memory.status[0] = 1  # 设置为做空状态
            
        # 做多信号：z-score < 下界 且 当前不是做多状态  
        elif memory.zscore[c.i - 1] < params.lower and memory.status[0] != 2:
            size[0] = params.order_pct1   # 买入中芯国际
            size[1] = -params.order_pct2  # 卖空华虹半导体
            # 执行顺序：先卖后买
            c.call_seq_now[0] = 1
            c.call_seq_now[1] = 0
            memory.status[0] = 2  # 设置为做多状态
            
        else:
            size[0] = np.nan
            size[1] = np.nan
    else:
        size[0] = np.nan
        size[1] = np.nan
    
    # 设置估值价格
    c.last_val_price[c.from_col] = c.close[c.i - 1, c.from_col]
    c.last_val_price[c.from_col + 1] = c.close[c.i - 1, c.from_col + 1]
        
    return (size,)

# 修复订单函数 - 使用正确的参数顺序
@njit
def order_func_nb(c, size):
    """执行订单 - 简化版本，只接受必要的参数"""
    group_col = c.col - c.from_col
    
    # 直接使用固定手续费率，避免参数传递问题
    commperc = 0.002  # 0.2%
    
    return vbt.portfolio.nb.order_nb(
        size=size[group_col], 
        price=c.close[c.i, c.col],  # 使用当前价格
        size_type=vbt.portfolio.enums.SizeType.TargetPercent,
        fees=commperc
    )

# 准备价格数据
print("准备回测数据...")
price_data = pd.DataFrame({
    'SMIC': smic_data['close'],
    'HHIC': hhic_data['close']
})

print(f"价格数据形状: {price_data.shape}")
print(f"价格数据时间范围: {price_data.index.min()} 到 {price_data.index.max()}")
print(f"价格数据前5行:\n{price_data.head()}")

# 调试信息：检查vectorbt版本和可用参数
print("\n=== 调试信息 ===")
print(f"vectorbt版本: {vbt.__version__}")
print(f"pandas版本: {pd.__version__}")
print(f"numpy版本: {np.__version__}")

# 运行配对交易回测
print("\n运行配对交易回测...")

# 首先尝试不使用cash参数
print("尝试方法1: 不使用cash参数...")
portfolio = vbt.Portfolio.from_order_func(
    price_data,
    order_func_nb,
    pre_group_func_nb=pre_group_func_nb,
    pre_segment_func_nb=pre_segment_func_nb,
    pre_group_args=(
        PERIOD,      # period (标量)
        UPPER,       # upper (标量)  
        LOWER,       # lower (标量)
        ORDER_PCT1,  # order_pct1 (标量)
        ORDER_PCT2   # order_pct2 (标量)
    ),
    pre_segment_args=(MODE,),    # mode
    group_by=True,   # 将两列分为同一组
    cash_sharing=True,           # 共享现金
    init_cash=INITIAL_CASH,
    freq='d'
)
print("方法1成功!")

print("回测完成!")

# 计算额外指标用于分析
def calculate_additional_metrics(price_data, period, mode):
    """计算额外指标"""
    smic_close = price_data['SMIC'].values
    hhic_close = price_data['HHIC'].values
    
    if mode == 'OLS':
        # 使用OLS方法计算价差
        spread = np.full(len(smic_close), np.nan)
        for i in range(period, len(smic_close)):
            window_slice = slice(i - period + 1, i + 1)
            a = smic_close[window_slice]
            b = hhic_close[window_slice]
            spread[i] = ols_spread_nb(a, b)
    else:
        # 对数收益率方法
        spread = np.full(len(smic_close), np.nan)
        spread[1:] = np.log(smic_close[1:] / smic_close[:-1]) - np.log(hhic_close[1:] / hhic_close[:-1])
    
    # 计算z-score
    zscore = np.full(len(spread), np.nan)
    for i in range(period, len(spread)):
        window_slice = slice(i - period + 1, i + 1)
        spread_mean = np.nanmean(spread[window_slice])
        spread_std = np.nanstd(spread[window_slice])
        if spread_std > 0:
            zscore[i] = (spread[i] - spread_mean) / spread_std
    
    return pd.Series(spread, index=price_data.index), pd.Series(zscore, index=price_data.index)

# 计算指标
spread_series, zscore_series = calculate_additional_metrics(price_data, PERIOD, MODE)

# 生成交易信号
short_signals = (zscore_series > UPPER).rename('short_signals')
long_signals = (zscore_series < LOWER).rename('long_signals')

# 可视化结果 - 使用matplotlib而不是vectorbt的绘图
print("生成分析图表...")

# 创建子图
fig, axes = plt.subplots(3, 1, figsize=(15, 12))

# 1. 价格走势
ax1 = axes[0]
ax1.plot(price_data.index, price_data['SMIC'], label='中芯国际(00981)', linewidth=1)
ax1.plot(price_data.index, price_data['HHIC'], label='华虹半导体(01347)', linewidth=1)
ax1.set_title(f'股票价格走势 ({start_date.strftime("%Y-%m-%d")} 至 {end_date.strftime("%Y-%m-%d")})')
ax1.set_ylabel('价格(港元)')
ax1.legend()
ax1.grid(True, alpha=0.3)

# 2. 价差和z-score
ax2 = axes[1]
ax2.plot(spread_series.index, spread_series.values, label='价差', color='blue', linewidth=1)
ax2.set_title('价差走势')
ax2.set_ylabel('价差')
ax2.legend(loc='upper left')
ax2.grid(True, alpha=0.3)

ax3 = ax2.twinx()
ax3.plot(zscore_series.index, zscore_series.values, label='Z-Score', color='red', linewidth=1, alpha=0.7)
ax3.axhline(y=UPPER, color='red', linestyle='--', alpha=0.5, label=f'上界({UPPER})')
ax3.axhline(y=0, color='black', linestyle='-', alpha=0.3)
ax3.axhline(y=LOWER, color='green', linestyle='--', alpha=0.5, label=f'下界({LOWER})')
ax3.set_ylabel('Z-Score')
ax3.legend(loc='upper right')

# 标记交易信号
short_points = zscore_series[short_signals]
long_points = zscore_series[long_signals]
ax3.scatter(short_points.index, short_points.values, color='red', marker='v', s=30, label='做空信号')
ax3.scatter(long_points.index, long_points.values, color='green', marker='^', s=30, label='做多信号')

# 3. 资产价值
ax4 = axes[2]
portfolio_value = portfolio.value()
ax4.plot(portfolio_value.index, portfolio_value.values, label='投资组合价值', color='purple', linewidth=1)
ax4.set_title('投资组合价值')
ax4.set_ylabel('资产价值(港元)')
ax4.set_xlabel('日期')
ax4.legend()
ax4.grid(True, alpha=0.3)

plt.tight_layout()
plt.show()

# 输出回测结果
print("\n" + "="*50)
print("配对交易回测结果（最近三年）")
print("="*50)

# 调试信息：检查投资组合属性
print("\n=== 投资组合调试信息 ===")
print(f"投资组合类型: {type(portfolio)}")
print(f"投资组合值形状: {portfolio.value().shape}")
print(f"投资组合值前5行:\n{portfolio.value().head()}")
print(f"投资组合值统计:")
print(f"  最小值: {portfolio.value().min():.2f}")
print(f"  最大值: {portfolio.value().max():.2f}")
print(f"  最终值: {portfolio.value().iloc[-1]:.2f}")

# 基本统计
try:
    stats = portfolio.stats()
    print(f"\n回测期间: {start_date.strftime('%Y-%m-%d')} 至 {end_date.strftime('%Y-%m-%d')}")
    print(f"初始资金: {INITIAL_CASH:,.2f} 港元")
    print(f"最终资产: {portfolio.value().iloc[-1]:,.2f} 港元")
    print(f"总收益率: {stats['Total Return']:.2%}")
    print(f"年化收益率: {stats['Annual Return']:.2%}")
    print(f"夏普比率: {stats['Sharpe Ratio']:.2f}")
    print(f"最大回撤: {stats['Max Drawdown']:.2%}")
    print(f"总交易次数: {stats['Total Trades']}")
except Exception as e:
    print(f"统计计算错误: {e}")

# 交易统计
try:
    orders = portfolio.orders()
    if len(orders) > 0:
        print(f"\n交易统计:")
        print(f"总订单数: {len(orders)}")
        print(f"中芯国际交易次数: {len(orders[orders['Column'] == 'SMIC'])}")
        print(f"华虹半导体交易次数: {len(orders[orders['Column'] == 'HHIC'])}")
        
        # 计算平均持仓时间
        positions = portfolio.positions()
        if len(positions) > 0:
            avg_holding = positions.duration.mean()
            print(f"平均持仓时间: {avg_holding:.1f} 天")
            
        # 显示交易盈亏
        if hasattr(portfolio, 'trades'):
            trades = portfolio.trades()
            if len(trades) > 0:
                print(f"总交易盈亏: {trades.pnl.sum():.2f} 港元")
                print(f"平均每笔交易盈亏: {trades.pnl.mean():.2f} 港元")
    else:
        print("没有交易记录")
except Exception as e:
    print(f"交易统计错误: {e}")

# 信号统计
print(f"\n信号统计:")
print(f"做空信号数量: {short_signals.sum()}")
print(f"做多信号数量: {long_signals.sum()}")
print(f"总信号数量: {short_signals.sum() + long_signals.sum()}")

# 相关性分析
correlation = price_data['SMIC'].corr(price_data['HHIC'])
print(f"\n股票相关性: {correlation:.4f}")

# 基本价格统计
print(f"\n价格统计:")
print(f"中芯国际 - 期初价格: {price_data['SMIC'].iloc[0]:.2f} 港元")
print(f"中芯国际 - 期末价格: {price_data['SMIC'].iloc[-1]:.2f} 港元")
print(f"中芯国际 - 期间涨跌幅: {(price_data['SMIC'].iloc[-1] / price_data['SMIC'].iloc[0] - 1):.2%}")
print(f"华虹半导体 - 期初价格: {price_data['HHIC'].iloc[0]:.2f} 港元")
print(f"华虹半导体 - 期末价格: {price_data['HHIC'].iloc[-1]:.2f} 港元")
print(f"华虹半导体 - 期间涨跌幅: {(price_data['HHIC'].iloc[-1] / price_data['HHIC'].iloc[0] - 1):.2%}")
print(f"中芯国际 - 价格波动率: {price_data['SMIC'].pct_change().std():.4f}")
print(f"华虹半导体 - 价格波动率: {price_data['HHIC'].pct_change().std():.4f}")

# 保存结果到文件
try:
    results_df = pd.DataFrame({
        'Date': price_data.index,
        'SMIC_Price': price_data['SMIC'],
        'HHIC_Price': price_data['HHIC'],
        'Spread': spread_series,
        'ZScore': zscore_series,
        'Portfolio_Value': portfolio.value(),
        'Short_Signals': short_signals,
        'Long_Signals': long_signals
    })

    results_df.to_csv('pair_trading_results_3years.csv', index=False)
    print(f"\n详细结果已保存到: pair_trading_results_3years.csv")
except Exception as e:
    print(f"保存结果错误: {e}")

# 显示最近交易
try:
    orders = portfolio.orders()
    if len(orders) > 0:
        recent_trades = orders.tail(10)
        print(f"\n最近10笔交易:")
        for idx, trade in recent_trades.iterrows():
            print(f"  时间: {trade['Timestamp']}, 股票: {trade['Column']}, "
                  f"数量: {trade['Size']:.4f}, 价格: {trade['Price']:.2f}, "
                  f"手续费: {trade['Fees']:.2f}")
    else:
        print("没有交易记录")
except Exception as e:
    print(f"显示最近交易错误: {e}")

# 年度绩效分析
try:
    print(f"\n年度绩效分析:")
    yearly_returns = portfolio.annual_returns()
    for year, ret in yearly_returns.items():
        print(f"{year}年收益率: {ret:.2%}")
except Exception as e:
    print(f"年度绩效分析错误: {e}")

print("\n回测完成!")