如何用Python识别日内非重叠的1%涨幅行情（Run-up）

因为expanding().min()是从交易日开始持续累积的最小值，没法在Run-up完成后重置计算起点，所以需要按交易日分组后，用状态跟踪的方式分段处理每个区间。

具体实现代码 #

import pandas as pd

# 示例数据转换（实际使用替换为你的数据）
data = {
    'Open': [308.0, 305.2, 304.5, 305.1, 304.3],
    'High': [308.0, 306.1, 304.9, 305.1, 304.9],
    'Low': [304.6, 304.3, 304.3, 304.2, 303.9],
    'Close': [305.1, 304.6, 304.9, 304.3, 304.8],
    'Volume': [31477.0, 20433.0, 14094.0, 14412.0, 28365.0],
    'barCount': [122, 92, 81, 79, 99]
}
df = pd.DataFrame(data, index=pd.to_datetime([
    '2021-11-15 09:00:00+01:00', 
    '2021-11-15 09:05:00+01:00', 
    '2021-11-15 09:10:00+01:00', 
    '2021-11-15 09:15:00+01:00', 
    '2021-11-15 09:20:00+01:00'
]))

# 初始化标记列
df['Run_start'] = False
df['Run_end'] = False
df['Run_duration'] = False
df['Current_Min'] = df['Low']

def process_trading_day(group):
    n = len(group)
    in_run = False
    current_low = None
    start_idx = None
    
    for i in range(n):
        # 未触发Run-up时，持续更新当前区间的最低低点
        if not in_run:
            if current_low is None or group['Low'].iloc[i] < current_low:
                current_low = group['Low'].iloc[i]
                start_idx = i  # 记录潜在的Run起点
            
            # 检查当前高点是否达到1%涨幅
            if group['High'].iloc[i] >= current_low * 1.01:
                in_run = True
                # 标记行情状态
                group['Run_start'].iloc[start_idx] = True
                group['Run_end'].iloc[i] = True
                if i > start_idx:
                    group['Run_duration'].iloc[start_idx+1:i] = True
        
        # Run-up完成后，重置状态开始下一轮搜索
        else:
            in_run = False
            current_low = group['Low'].iloc[i]
            start_idx = i
    
    # 生成分段的滚动低点列
    group['Current_Min'] = group['Low'].copy()
    current_min = None
    reset = True
    for i in range(n):
        if group['Run_end'].iloc[i]:
            reset = True
        if reset:
            current_min = group['Low'].iloc[i]
            reset = False
        else:
            if group['Low'].iloc[i] < current_min:
                current_min = group['Low'].iloc[i]
        group['Current_Min'].iloc[i] = current_min
    
    return group

# 按交易日分组执行逻辑
df = df.groupby(df.index.date).apply(process_trading_day)

print(df[['Low', 'High', 'Current_Min', 'Run_start', 'Run_end', 'Run_duration']])

核心逻辑说明 #

1. 交易日独立处理：按index.date分组，确保每个交易日的搜索逻辑完全重置，互不干扰。
2. 状态机跟踪：用in_run变量标记是否处于Run-up完成后的重置阶段，未触发时持续更新当前区间的最低低点，直到找到首个满足1%涨幅的高点。
3. 行情状态标记：
   • Run_start：标记当前Run-up的起始低点位置
   • Run_end：标记Run-up的结束K线（首次达到1%涨幅的K线）
   • Run_duration：标记起始和结束之间的所有K线
4. 分段滚动低点：每次Run-up完成后，从结束K线的下一根开始，重新计算新区间的滚动低点，实现非重叠搜索。

优化提示 #

如果你的K线数据量极大，循环遍历效率偏低，可以考虑用cumsum标记区间，再对每个区间单独计算expanding().min()，但循环方式逻辑更直观，便于调试和修改规则。

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