如何无需循环基于滚动条件生成DataFrame交易信号列？

背景 #

我手里有个包含收盘价（Price）、当日最高价（High）、**当日最低价（Low）**的股票价格DataFrame，正在回测做空入场信号，规则如下：

• 入场触发：过去7天内价格跌幅达到30%，计算公式为(Min Low - Max High)/Max High，触发信号后次日建仓（持仓标记pos==1，需收盘后确认信号）
• 止损平仓：持有仓位时，若当日收盘价相比建仓日（触发日+1）收盘价涨幅达到15%，则平仓离场
• 约束条件：仅空仓时可建仓，仅持仓时可平仓
• 核心逻辑：通过High和Low列触发建仓，通过收盘价列触发平仓

示例DataFrame：

Date      Price    High   Low     Entry  Exit
2024-01-15  3100.0  3230.0  2685.0
2024-01-16  3595.0  3635.0  3155.0
2024-01-17  4295.0  4295.0  4040.0
2024-01-18  3595.0  4550.0  3595.0 
2024-01-19  3280.0  3880.0  2929.0 (entry trigger)
2024-01-22  3505.0  3575.0  3185.0  Entry
2024-01-23  3945.0  3970.0  3565.0
2024-01-24  4075.0  4300.0  4020.0 (exit trigger)
2024-01-25  3990.0  4190.0  3910.0          Exit
2024-01-26  3855.0  4020.0  3765.0
2024-01-29  3740.0  2920.0  2775.0 (entry trigger)
2024-02-01  3810.0  3880.0  3460.0  Entry
2024-02-02  4150.0  4150.0  3830.0
2024-02-05  3915.0  4195.0  3915.0
2024-02-06  3855.0  4040.0  2805.0
2024-02-07  3785.0  3945.0  3755.0
2024-02-08  3790.0  3885.0  3720.

问题 #

我现在用循环遍历每行、跟踪仓位的方式实现了逻辑，但知道针对DataFrame有更高效的无循环实现方法，求优化方案。

现有循环代码 #

import numpy as np
import pandas as pd

df['Max7'] = df['High'].rolling(window=7, min_periods=1).max()
df['Min7'] = df['Low'].rolling(window=7, min_periods=1).min()
df['Min PctChange7'] = (df['High'] - df['Min7']) / df['Min7'] * 100
df["Entry"] = np.nan
df["Exit"] = np.nan

pos = 0
entry_date = np.nan

pos = 0
entry_date = np.nan

for i in range(len(df)):
    if pos == 0:
        if df["Min PctChange7"].iloc[[i]][0] < -30:
            pos = 1
            entry_date = i+1
            df["Entry"].iloc[[entry_date]] = "Entry"  
        
    if i > entry_date and pos == 1:
        if (df.iloc[[i]]["Price"][0]/df.iloc[[entry_date]]["Price"][0]) > 1.15:
            df["Exit"].iloc[[i]] = "Exit"  
            pos = 0    

优化方案 #

关键修正说明 #

原循环代码的跌幅计算公式写错了，正确公式应为(7天最低价 - 7天最高价)/7天最高价 *100，先修正这一错误再做优化。

完整无循环代码 #

import numpy as np
import pandas as pd

# 1. 计算7天滚动高低点及跌幅
df['Max7'] = df['High'].rolling(window=7, min_periods=1).max()
df['Min7'] = df['Low'].rolling(window=7, min_periods=1).min()
df['DropPct7'] = (df['Min7'] - df['Max7']) / df['Max7'] * 100

# 2. 标记入场信号与建仓日
# 先标记触发入场信号的日期（跌幅≤30%）
df['trigger_entry'] = df['DropPct7'] <= -30
# 建仓日是触发日的次日，用shift(1)实现
df['Entry'] = np.where(df['trigger_entry'].shift(1), 'Entry', np.nan)

# 3. 处理仓位状态与平仓信号
entry_points = df['Entry'].notna()
# 初始化仓位：0为空仓，1为持仓
df['pos'] = 0
df.loc[entry_points, 'pos'] = 1

# 把建仓日的价格填充到整个持仓周期
df['entry_price'] = np.where(entry_points, df['Price'], np.nan)
df['entry_price'] = df['entry_price'].ffill()

# 计算平仓触发条件：持仓时，当日价格相比建仓价涨幅≥15%
df['trigger_exit'] = (df['Price'] / df['entry_price']) >= 1.15
# 标记平仓点
df['Exit'] = np.where(df['pos'].eq(1) & df['trigger_exit'], 'Exit', np.nan)

# 更新仓位状态：平仓后重置为0，确保仓位只能是0或1
exit_points = df['Exit'].notna()
df['pos'] = (df['pos'].cumsum() - exit_points.cumsum()).clip(0, 1)
df['pos'] = df['pos'].ffill().fillna(0).astype(int)
# 清空空仓时的建仓价格
df['entry_price'] = np.where(df['pos'] == 1, df['entry_price'], np.nan)

# 4. 清理中间列（可选，不需要可以删掉）
df.drop(['Max7', 'Min7', 'DropPct7', 'trigger_entry', 'pos', 'entry_price', 'trigger_exit'], axis=1, inplace=True)

优化亮点 #

• 全向量化操作，性能比循环提升数倍，尤其适合大样本数据
• 通过shift、ffill、cumsum实现仓位状态的自动跟踪，严格遵守“空仓才能建仓、持仓才能平仓”的约束
• 每一步逻辑清晰，中间标记列便于调试和后续扩展

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