如何用向量化提升Python量化交易回测的运行性能？

问题背景 #

你有两个核心DataFrame用于每日交易策略回测：

• 每日交易信号DataFrame：包含Date（交易日期）和trading_signal（1=买入，-1=卖出，0=无交易）两列
• 每日分钟级行情DataFrame：包含Date、Time、open、high、low、close六列

策略规则：

1. 依据当日交易信号，以当日首分钟开盘价开仓
2. 若盘中触发止损（幅度20），则在下一分钟反向开仓
3. 若盘中触发止盈，当日收益为止盈额
4. 未触发止损/止盈时，收益为当日最后一分钟收盘价减首分钟开盘价（乘以开仓方向）

原嵌套循环实现因数据量较大性能不佳，需优化为向量化实现，生成包含交易行为及收益的结果DataFrame。

向量化优化方案 #

核心思路是利用Pandas的groupby聚合、向量化条件判断（np.where）替代循环，依托Pandas底层C实现提升运算效率，步骤如下：

1. 数据预处理与关键指标提取 #

先统一日期格式，再按日期分组提取每个交易日的首分钟开盘价、最后分钟收盘价，同时识别盘中止损/止盈触发点。

import pandas as pd
import numpy as np

# 假设输入数据为signal_df（信号）、minute_df（分钟行情）
signal_df['Date'] = pd.to_datetime(signal_df['Date'])
minute_df['Date'] = pd.to_datetime(minute_df['Date'])

# 提取每日首开盘价、最后收盘价
daily_prices = minute_df.groupby('Date').agg(
    first_open=('open', 'first'),
    last_close=('close', 'last')
).reset_index()

# 为分钟数据添加触发止损/止盈的标记
def mark_trigger(group, stop_loss=20, take_profit=30):
    # 合并当日交易信号
    group = group.merge(signal_df[['Date', 'trading_signal']], on='Date', how='left')
    init_pos = group['trading_signal'].iloc[0]
    
    if init_pos == 0:
        group['trigger_type'] = None
        return group
    
    # 根据开仓方向计算止损/止盈触发条件
    first_open = group['first_open'].iloc[0]
    if init_pos == 1:  # 做多：止损=首开-20，止盈=首开+30
        stop_trigger = group['low'] <= first_open - stop_loss
        tp_trigger = group['high'] >= first_open + take_profit
    else:  # 做空：止损=首开+20，止盈=首开-30
        stop_trigger = group['high'] >= first_open + stop_loss
        tp_trigger = group['low'] <= first_open - take_profit
    
    # 标记第一个触发的类型
    trigger_mask = stop_trigger | tp_trigger
    first_trigger_idx = trigger_mask.idxmax() if trigger_mask.any() else None
    
    if first_trigger_idx is not None:
        group['trigger_type'] = 'stop_loss' if stop_trigger.loc[first_trigger_idx] else 'take_profit'
    else:
        group['trigger_type'] = None
    
    return group

# 应用到每个交易日分组
minute_with_trigger = minute_df.merge(daily_prices, on='Date').groupby('Date').apply(mark_trigger).reset_index(drop=True)

# 提取每日触发信息
daily_trigger = minute_with_trigger.groupby('Date').agg(
    trigger_type=('trigger_type', 'first')
).reset_index()

2. 计算第一笔交易的收益与信号 #

通过向量化条件判断np.where批量计算每个交易日的第一笔交易结果：

# 合并所有数据到结果DataFrame
result_df = signal_df.merge(daily_prices, on='Date').merge(daily_trigger, on='Date')

# 计算第一笔交易信号与收益
result_df['trade_1st'] = result_df['trading_signal']
result_df['pnl_1st'] = np.where(
    result_df['trading_signal'] == 0,
    0,
    np.where(
        result_df['trigger_type'] == 'take_profit',
        result_df['trading_signal'] * 30,  # 止盈额，可根据需求调整
        np.where(
            result_df['trigger_type'] == 'stop_loss',
            result_df['trading_signal'] * (-20),  # 止损额
            (result_df['last_close'] - result_df['first_open']) * result_df['trading_signal']
        )
    )
)

3. 计算第二笔反向交易（仅止损触发时） #

针对触发止损的交易日，批量计算反向开仓的收益：

# 初始化第二笔交易默认值
result_df['trade_2nd'] = 0
result_df['pnl_2nd'] = 0

# 处理触发止损的交易日，计算反向开仓收益
stop_loss_mask = result_df['trigger_type'] == 'stop_loss'
if stop_loss_mask.any():
    # 获取每个止损交易日的触发后下一分钟开盘价
    def get_reverse_pnl(group):
        init_pos = group['trading_signal'].iloc[0]
        first_open = group['first_open'].iloc[0]
        # 找到第一个触发止损的分钟索引
        trigger_idx = group[group['low'] <= first_open -20 if init_pos ==1 else group['high'] >= first_open +20].index[0] +1
        # 下一分钟开盘价（若触发在最后一分钟则用收盘价）
        reverse_open = group['open'].iloc[trigger_idx] if trigger_idx < len(group) else group['close'].iloc[-1]
        # 反向仓位收益：(收盘价 - 反向开仓价) * 反向方向
        reverse_pnl = (group['last_close'].iloc[0] - reverse_open) * (-init_pos)
        return pd.Series({'trade_2nd': -init_pos, 'pnl_2nd': reverse_pnl})
    
    # 对止损交易日分组计算
    reverse_results = minute_with_trigger[minute_with_trigger['trigger_type'] == 'stop_loss'].groupby('Date').apply(get_reverse_pnl).reset_index()
    # 更新结果
    result_df.loc[stop_loss_mask, ['trade_2nd', 'pnl_2nd']] = reverse_results[['trade_2nd', 'pnl_2nd']].values

4. 整理最终结果 #

提取需要的列，得到目标格式的DataFrame：

final_result = result_df[['Date', 'trade_1st', 'pnl_1st', 'trade_2nd', 'pnl_2nd']]
print(final_result)

性能优势说明 #

• 避免了嵌套循环的Python层逐行运算，改用Pandas内置的C级聚合与向量化操作，运算速度可提升数倍至数十倍（数据量越大优势越明显）
• 所有逻辑均基于Pandas原生API，代码可维护性与扩展性更强

内容的提问来源于stack exchange，提问作者tender