4、套利信号识别：阈值设定方法、信号生成逻辑、回测框架搭建、策略绩效评估

各位同学，欢迎来到第四讲。前面我们聊了集合竞价和连续竞价的价格差异是怎么产生的，也看了数据怎么清洗。今天这节，是真正动手干活的时候了——我们要把那些藏在数据里的套利机会，一个一个揪出来。

说白了，套利信号识别就是回答三个问题：什么时候该进场？什么时候该离场？这笔交易到底赚没赚钱？ 我做了这么多年量化，见过太多人把精力花在复杂的模型上，结果连最基本的阈值都没设对。嗯，咱们今天就从最基础、也最关键的阈值设定开始。

4.1 阈值设定方法：别拍脑袋，要讲数据

阈值怎么设？我见过有人直接拍脑袋设个“价差超过0.5%就进场”。这种做法，说白了就是赌运气。真正靠谱的方法，得从历史数据里找规律。

我个人习惯用以下三种方法，你可以根据品种特性选一个：

• 固定百分位法： 取过去N个交易日价差序列的90%或95%分位数作为阈值。比如，过去20天价差的95%分位数是0.8%，那价差超过0.8%就触发信号。
• 移动标准差法： 用滚动窗口计算价差的均值和标准差，阈值 = 均值 ± k × 标准差。k通常取2或3。这个方法能自适应市场波动。
• 动态分位数法： 结合时间衰减权重，近期数据权重更大。适合市场结构经常变化的品种。

这里给出一段阈值计算的代码示例，用的是移动标准差法：

import pandas as pd
import numpy as np

def calculate_threshold(price_df, window=20, k=2.5):
    """
    计算动态阈值
    :param price_df: DataFrame，包含'call_auction'和'continuous'两列价格
    :param window: 滚动窗口大小
    :param k: 标准差倍数
    :return: 带阈值的DataFrame
    """
    df = price_df.copy()
    # 计算价差（百分比）
    df['spread'] = (df['call_auction'] - df['continuous']) / df['continuous'] * 100
    
    # 滚动均值和标准差
    df['mean'] = df['spread'].rolling(window=window).mean()
    df['std'] = df['spread'].rolling(window=window).std()
    
    # 上下阈值
    df['upper_threshold'] = df['mean'] + k * df['std']
    df['lower_threshold'] = df['mean'] - k * df['std']
    
    return df

4.2 信号生成逻辑：从数据到交易指令

阈值设好了，接下来就是生成交易信号。这一步其实不复杂，但容易踩坑。我曾经犯过一个错误——信号生成时没考虑交易成本，结果实盘一跑，利润全被手续费吃掉了。

信号生成的基本逻辑是这样的：

1. 开仓信号： 当价差向上突破上阈值时，做空集合竞价、做多连续竞价（预期价差缩小）；当价差向下突破下阈值时，做多集合竞价、做空连续竞价。
2. 平仓信号： 当价差回归到均值附近（比如触及均值线或0轴）时，平掉所有头寸。
3. 止损信号： 如果价差继续扩大，超过某个更极端的阈值（比如3倍标准差），强制平仓止损。

你想想看，如果只设开仓不设止损，遇到极端行情怎么办？我记得2020年3月那波波动，很多套利策略就是因为没设止损，价差一直不回，最后亏得很惨。

下面是一个完整的信号生成函数：

def generate_signals(df, entry_k=2.5, exit_k=0.5, stop_k=4.0):
    """
    生成交易信号
    :param df: 包含spread, upper_threshold, lower_threshold的DataFrame
    :param entry_k: 开仓阈值倍数
    :param exit_k: 平仓阈值倍数（相对于均值）
    :param stop_k: 止损阈值倍数
    :return: 带信号的DataFrame
    """
    df = df.copy()
    df['signal'] = 0  # 0: 无持仓, 1: 多头, -1: 空头
    
    # 计算平仓线和止损线
    df['exit_upper'] = df['mean'] + exit_k * df['std']
    df['exit_lower'] = df['mean'] - exit_k * df['std']
    df['stop_upper'] = df['mean'] + stop_k * df['std']
    df['stop_lower'] = df['mean'] - stop_k * df['std']
    
    position = 0
    for i in range(1, len(df)):
        # 开仓逻辑
        if position == 0:
            if df['spread'].iloc[i] > df['upper_threshold'].iloc[i]:
                position = -1  # 做空价差
            elif df['spread'].iloc[i] < df['lower_threshold'].iloc[i]:
                position = 1   # 做多价差
        
        # 平仓逻辑
        elif position == -1:
            if df['spread'].iloc[i] <= df['exit_upper'].iloc[i]:
                position = 0
            elif df['spread'].iloc[i] >= df['stop_upper'].iloc[i]:
                position = 0  # 止损
        
        elif position == 1:
            if df['spread'].iloc[i] >= df['exit_lower'].iloc[i]:
                position = 0
            elif df['spread'].iloc[i] <= df['stop_lower'].iloc[i]:
                position = 0  # 止损
        
        df.loc[df.index[i], 'signal'] = position
    
    return df

4.3 回测框架搭建：别让回测骗了你

回测框架，说白了就是一台时光机。你把策略放进去，它告诉你“如果过去这么干，能赚多少钱”。但这里有个大坑——过拟合。我见过太多人在回测里跑出漂亮的曲线，一上实盘就崩。

搭建回测框架，我建议至少包含以下几个模块：

• 数据模块： 加载历史行情数据，包括集合竞价和连续竞价的价格、成交量、时间戳。
• 信号模块： 调用上面写的信号生成函数，产生交易指令。
• 执行模块： 模拟交易执行，考虑滑点和手续费。这里要注意，集合竞价的成交规则和连续竞价不一样，别搞混了。
• 绩效模块： 计算收益率、夏普比率、最大回撤等指标。

下面是一个简化版的回测框架：

def backtest(df, initial_capital=100000, fee_rate=0.0003, slippage=0.0001):
    """
    回测主函数
    :param df: 带信号的DataFrame
    :param initial_capital: 初始资金
    :param fee_rate: 手续费率（双边）
    :param slippage: 滑点（百分比）
    :return: 回测结果
    """
    df = df.copy()
    df['position'] = df['signal'].shift(1)  # 信号滞后一期，模拟实际交易延迟
    df['position'].fillna(0, inplace=True)
    
    # 计算每笔交易的收益率
    df['return'] = df['position'] * df['spread'].diff() / 100
    
    # 扣除交易成本
    df['trade'] = df['position'].diff().abs()
    df['cost'] = df['trade'] * fee_rate * 2 + df['trade'] * slippage  # 双边手续费+滑点
    df['net_return'] = df['return'] - df['cost']
    
    # 计算净值曲线
    df['equity'] = initial_capital * (1 + df['net_return']).cumprod()
    
    return df

4.4 策略绩效评估：别只看收益率

很多人看策略好不好，第一眼就看收益率。嗯，这其实是个误区。收益率高不代表策略好，可能是运气好，也可能是风险太高。

我一般会从以下几个维度评估策略：

指标	计算公式	说明
年化收益率	(最终净值/初始净值)^(252/交易日数) - 1	衡量赚钱能力
夏普比率	(策略收益率 - 无风险利率) / 收益率标准差	衡量风险调整后收益，一般大于1算不错
最大回撤	净值从峰值到谷底的最大跌幅	衡量策略的最大亏损风险
胜率	盈利交易次数 / 总交易次数	衡量信号准确度
盈亏比	平均盈利 / 平均亏损	衡量单笔交易的盈亏平衡

这里有一个我自己的经验：对于套利策略，我更看重夏普比率和最大回撤。为什么？因为套利本身就是赚取低风险的钱，如果回撤太大，说明策略逻辑可能有问题。

下面是一个绩效评估函数：

def evaluate_performance(df, risk_free_rate=0.03):
    """
    策略绩效评估
    :param df: 回测结果DataFrame
    :param risk_free_rate: 无风险利率（年化）
    :return: 绩效指标字典
    """
    equity = df['equity']
    daily_returns = df['net_return']
    
    # 年化收益率
    total_days = len(daily_returns)
    annual_return = (equity.iloc[-1] / equity.iloc[0]) ** (252 / total_days) - 1
    
    # 夏普比率
    excess_returns = daily_returns - risk_free_rate / 252
    sharpe = np.sqrt(252) * excess_returns.mean() / excess_returns.std()
    
    # 最大回撤
    peak = equity.expanding().max()
    drawdown = (equity - peak) / peak
    max_drawdown = drawdown.min()
    
    # 胜率
    trades = df[df['trade'] > 0]
    winning_trades = trades[trades['net_return'] > 0]
    win_rate = len(winning_trades) / len(trades) if len(trades) > 0 else 0
    
    # 盈亏比
    avg_win = winning_trades['net_return'].mean() if len(winning_trades) > 0 else 0
    losing_trades = trades[trades['net_return'] <= 0]
    avg_loss = losing_trades['net_return'].mean() if len(losing_trades) > 0 else 0
    profit_loss_ratio = abs(avg_win / avg_loss) if avg_loss != 0 else float('inf')
    
    return {
        'annual_return': annual_return,
        'sharpe_ratio': sharpe,
        'max_drawdown': max_drawdown,
        'win_rate': win_rate,
        'profit_loss_ratio': profit_loss_ratio
    }

4.5 知识体系总览

为了让你更直观地理解本章的知识结构，我画了一张流程图：

这张图把今天的内容串起来了。从数据输入开始，经过阈值设定、信号生成，最后到回测和绩效评估。每一步都有坑，每一步也都有优化空间。

---

公众号:蓝海资料掘金营，微信deep3321