《高频交易入门:Python实现毫秒级交易策略》(二)

class StatisticalArbitrageHFT:    """    高频统计套利策略    寻找两个高度相关资产的短暂定价错误    """    def __init__(self, symbol1, symbol2, initial_capital=1000000):        self.symbol1 = symbol1        self.symbol2 = symbol2        self.capital = initial_capital        self.position1 = 0  # symbol1持仓        self.position2 = 0  # symbol2持仓        self.spread_history = []        self.mean_spread = 0        self.std_spread = 1        self.z_score_threshold = 2.0  # Z-score阈值        self.max_position = 1000  # 最大单边持仓        # 交易记录        self.trades = []        self.pnl = 0    def calculate_spread(self, price1, price2):        """        计算价差（标准化）        简单使用价格比的对数        """        spread = np.log(price1) - np.log(price2)        return spread    def update_spread_statistics(self, spread):        """        更新价差统计量（使用滚动窗口）        """        self.spread_history.append(spread)        # 保持最近1000个观测值        if len(self.spread_history) > 1000:            self.spread_history.pop(0)        if len(self.spread_history) >= 100:            self.mean_spread = np.mean(self.spread_history[-100:])            self.std_spread = np.std(self.spread_history[-100:])    def generate_signal(self, current_spread):        """        生成交易信号        """        if self.std_spread == 0:            return 0, 0        z_score = (current_spread - self.mean_spread) / self.std_spread        # 当价差偏离超过阈值时交易        if z_score > self.z_score_threshold:            # symbol1相对高估，symbol2相对低估            # 卖空symbol1，买入symbol2            return -1, 1        elif z_score < -self.z_score_threshold:            # symbol1相对低估，symbol2相对高估            # 买入symbol1，卖空symbol2            return 1, -1        elif abs(z_score) < 0.5 and (self.position1 != 0 or self.position2 != 0):            # 价差回归，平仓            return -np.sign(self.position1), -np.sign(self.position2)        return 0, 0    def execute_trade(self, signal1, signal2, price1, price2, timestamp):        """        执行交易        """        # 确定交易数量（简化：固定数量）        trade_size = min(100, self.max_position - abs(self.position1))        if trade_size <= 0:            return        # 执行交易        cost1 = 0        cost2 = 0        if signal1 == 1:  # 买入symbol1            cost1 = -trade_size * price1            self.position1 += trade_size        elif signal1 == -1:  # 卖出symbol1            cost1 = trade_size * price1            self.position1 -= trade_size        if signal2 == 1:  # 买入symbol2            cost2 = -trade_size * price2            self.position2 += trade_size        elif signal2 == -1:  # 卖出symbol2            cost2 = trade_size * price2            self.position2 -= trade_size        total_cost = cost1 + cost2        self.capital += total_cost        # 记录交易        self.trades.append({            'timestamp': timestamp,            'symbol1_action': 'buy' if signal1 == 1 else 'sell' if signal1 == -1 else 'hold',            'symbol2_action': 'buy' if signal2 == 1 else 'sell' if signal2 == -1 else 'hold',            'price1': price1,            'price2': price2,            'size': trade_size,            'position1': self.position1,            'position2': self.position2,            'capital': self.capital,            'spread': self.calculate_spread(price1, price2),            'z_score': (self.calculate_spread(price1, price2) - self.mean_spread) / self.std_spread                       if self.std_spread > 0 else 0        })    def run_backtest(self, price_data1, price_data2):        """        运行回测        price_data1, price_data2: 包含'timestamp'和'price'列的DataFrame        """        # 对齐时间戳（简化处理）        merged_data = pd.merge(price_data1, price_data2, on='timestamp',                               suffixes=('_1', '_2'))        results = []        for i, row in merged_data.iterrows():            timestamp = row['timestamp']            price1 = row['price_1']            price2 = row['price_2']            # 计算当前价差            current_spread = self.calculate_spread(price1, price2)            # 更新统计量            self.update_spread_statistics(current_spread)            # 生成信号            signal1, signal2 = self.generate_signal(current_spread)            # 执行交易            if signal1 != 0 or signal2 != 0:                self.execute_trade(signal1, signal2, price1, price2, timestamp)            # 记录每日表现            position_value = self.position1 * price1 + self.position2 * price2            total_value = self.capital + position_value            results.append({                'timestamp': timestamp,                'price1': price1,                'price2': price2,                'spread': current_spread,                'z_score': (current_spread - self.mean_spread) / self.std_spread                           if self.std_spread > 0 else 0,                'position1': self.position1,                'position2': self.position2,                'capital': self.capital,                'position_value': position_value,                'total_value': total_value,                'signal1': signal1,                'signal2': signal2            })        return pd.DataFrame(results)# 生成模拟数据（两个相关资产）np.random.seed(42)n_points = 1000# 基础价格序列base_trend = np.cumsum(np.random.randn(n_points) * 0.001) + 100# 两个相关但略有差异的价格序列price1 = base_trend + np.random.randn(n_points) * 0.01price2 = base_trend * 0.99 + np.random.randn(n_points) * 0.01  # 略有系统差异# 添加一些短暂背离for i in range(100, 150):    price1[i] += 0.1  # 短暂高估for i in range(300, 320):    price2[i] += 0.08  # 短暂高估timestamps = pd.date_range('2024-06-01 09:30:00', periods=n_points, freq='1s')price_data1 = pd.DataFrame({'timestamp': timestamps, 'price': price1})price_data2 = pd.DataFrame({'timestamp': timestamps, 'price': price2})# 运行统计套利策略stat_arb = StatisticalArbitrageHFT('ASSET1', 'ASSET2', initial_capital=1000000)results_df = stat_arb.run_backtest(price_data1, price_data2)# 分析结果initial_value = 1000000final_value = results_df['total_value'].iloc[-1]total_return = (final_value - initial_value) / initial_value# 计算策略收益序列results_df['strategy_return'] = results_df['total_value'].pct_change().fillna(0)print(f"\n统计套利策略表现:")print(f"初始资金: {initial_value:,.0f}")print(f"最终资金: {final_value:,.0f}")print(f"总收益率: {total_return:.2%}")print(f"交易次数: {len(stat_arb.trades)}")print(f"最大持仓symbol1: {results_df['position1'].abs().max()}")print(f"最大持仓symbol2: {results_df['position2'].abs().max()}")# 可视化结果fig, axes = plt.subplots(3, 1, figsize=(14, 12))# 价格和价差ax1 = axes[0]ax1.plot(results_df['timestamp'], results_df['price1'],         label='资产1价格', linewidth=2, alpha=0.7)ax1.plot(results_df['timestamp'], results_df['price2'],         label='资产2价格', linewidth=2, alpha=0.7)ax1.set_ylabel('价格')ax1.set_title('资产价格走势', fontsize=14)ax1.legend()ax1.grid(True, alpha=0.3)# Z-score和交易信号ax2 = axes[1]ax2.plot(results_df['timestamp'], results_df['z_score'],         label='Z-score', linewidth=2, color='purple')ax2.axhline(y=stat_arb.z_score_threshold, color='r', linestyle='--',            alpha=0.5, label=f'阈值 (+{stat_arb.z_score_threshold})')ax2.axhline(y=-stat_arb.z_score_threshold, color='r', linestyle='--',            alpha=0.5, label=f'阈值 (-{stat_arb.z_score_threshold})')ax2.axhline(y=0, color='gray', linestyle='-', alpha=0.3)# 标记交易点trade_times = [t['timestamp'] for t in stat_arb.trades]if trade_times:    # 获取这些时间点的Z-score    trade_z_scores = []    for t in stat_arb.trades:        mask = results_df['timestamp'] == t['timestamp']        if mask.any():            trade_z_scores.append(results_df.loc[mask, 'z_score'].iloc[0])    ax2.scatter(trade_times, trade_z_scores, color='red', s=50,                zorder=5, label='交易点')ax2.set_ylabel('Z-score')ax2.set_title('价差Z-score与交易信号', fontsize=14)ax2.legend()ax2.grid(True, alpha=0.3)# 资金曲线ax3 = axes[2]ax3.plot(results_df['timestamp'], results_df['total_value'],         label='总资产', linewidth=2, color='green')ax3.set_xlabel('时间')ax3.set_ylabel('资金')ax3.set_title('策略资金曲线', fontsize=14)ax3.legend()ax3.grid(True, alpha=0.3)plt.tight_layout()plt.show()# 显示交易记录if stat_arb.trades:    trades_df = pd.DataFrame(stat_arb.trades)    print("\n最近5笔交易记录:")    print(trades_df.tail(5).to_string(index=False))