Python交易算法实现同一时间仅允许1笔持仓的信号过滤方法咨询

实现逻辑 #

核心通过维护冷却计数器实现信号过滤，规则如下：

• 初始冷却计数器为0，代表当前无持仓可触发开仓
• 若冷却计数器>0：当前周期信号设为0，计数器减1
• 若冷却计数器=0且Long_Trade值为1：当前周期信号设为1，计数器重置为3（对应t入场后屏蔽t+1~t+3共3个周期）
• 若冷却计数器=0且Long_Trade值为0：当前周期信号设为0，计数器保持0

代码实现 #

通用Pandas版本（适合普通数据量） #

import pandas as pd
import numpy as np

def generate_filtered_signal(long_trade_series, cool_period=3):
    signal = []
    cool_counter = 0
    for val in long_trade_series:
        if cool_counter > 0:
            signal.append(0)
            cool_counter -= 1
        else:
            if val == 1:
                signal.append(1)
                cool_counter = cool_period
            else:
                signal.append(0)
    return pd.Series(signal, index=long_trade_series.index)

# 测试用例（匹配你给出的示例）
df = pd.DataFrame({'Long_Trade': [1,1,1,1,1]})
df['Signal'] = generate_filtered_signal(df['Long_Trade'])
print(df)
# 输出结果：
#   Long_Trade  Signal
# 0          1       1
# 1          1       0
# 2          1       0
# 3          1       0
# 4          1       1

高性能向量化版本（适合百万级以上数据量，Numba加速） #

from numba import jit

@jit(nopython=True)
def generate_signal_vectorized(long_trade_arr, cool_period=3):
    n = len(long_trade_arr)
    signal = np.zeros(n, dtype=np.int8)
    cool_counter = 0
    for i in range(n):
        if cool_counter > 0:
            signal[i] = 0
            cool_counter -= 1
        else:
            if long_trade_arr[i] == 1:
                signal[i] = 1
                cool_counter = cool_period
    return signal

# 调用方式
df['Signal'] = generate_signal_vectorized(df['Long_Trade'].values)

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