量化交易代码前瞻偏差排查及无交易信号问题咨询

一、前瞻偏差位置排查 #

1. support函数的核心前瞻问题 #

原代码及修改后的support函数均存在前瞻偏差：
第二个循环range(index + 1, index + n2 + 1)直接调用了index位置之后的n2根K线数据。实盘场景下，判断当前K线是否为支撑/阻力位时，不可能获取未来未发生的K线数据，这是典型的回溯时误用未来数据的错误。

2. identify_df_trends函数的前瞻问题 #

该函数的趋势识别逻辑基于全量数据集回溯：遍历所有K线后才确定趋势起止点，相当于提前知晓后续价格走势，不符合实盘仅能使用历史数据的要求。

3. support_resistance函数的遍历范围问题 #

循环range(2, df.shape[0] - 2)虽限制遍历到倒数第3根K线，但结合support函数的n2=2参数，仍会用到倒数第2、第1根K线的未来数据，未彻底规避前瞻偏差。

二、修改后无交易信号的原因 #

1. 未彻底消除前瞻偏差：修改后的support函数仅调整了第一个循环范围，第二个循环仍依赖未来数据，导致实盘逻辑不成立，无法识别有效支撑/阻力位。
2. 条件过于严苛：原逻辑要求支撑位前后K线必须严格单调变化（前n1根下跌、后n2根上涨），实际行情中符合该条件的点位极少。
3. 过滤逻辑过严：distance_from_mean函数会过滤掉与已有支撑/阻力位距离小于平均波幅的点位，进一步减少了有效信号数量。

三、纯历史K线判断趋势的可行方案 #

所有方案严格基于当前及之前的K线数据，不得使用任何未来信息，以下是几种实用方案：

1. 移动平均线交叉法 #

通过短期均线与长期均线的交叉判断趋势，逻辑简单且无前瞻：

def identify_trend_with_ma(df, short_window=5, long_window=20):
    # 计算短期、长期移动平均线
    df['short_ma'] = df['close'].rolling(window=short_window).mean()
    df['long_ma'] = df['close'].rolling(window=long_window).mean()
    df['Trend'] = 0
    # 短期均线上穿长期均线，标记上升趋势
    df.loc[df['short_ma'] > df['long_ma'], 'Trend'] = 1
    # 短期均线下穿长期均线，标记下降趋势
    df.loc[df['short_ma'] < df['long_ma'], 'Trend'] = -1
    # 填充初始窗口的空值
    df['Trend'].fillna(0, inplace=True)
    return df

2. 高低点突破法 #

通过突破历史高低点判断趋势延续或反转：

def identify_trend_with_high_low(df, window=5):
    # 计算前window根K线的最高高点、最低低点
    df['highest_high'] = df['high'].rolling(window=window).max().shift(1)
    df['lowest_low'] = df['low'].rolling(window=window).min().shift(1)
    df['Trend'] = 0
    # 当前高点突破历史高点，标记上升趋势
    df.loc[df['high'] > df['highest_high'], 'Trend'] = 1
    # 当前低点跌破历史低点，标记下降趋势
    df.loc[df['low'] < df['lowest_low'], 'Trend'] = -1
    # 趋势延续：未突破时保持之前的趋势
    df['Trend'] = df['Trend'].ffill()
    df['Trend'].fillna(0, inplace=True)
    return df

3. 线性回归斜率法 #

通过计算近期K线收盘价的线性回归斜率判断趋势方向：

import numpy as np

def identify_trend_with_slope(df, window=5):
    def calculate_slope(series):
        x = np.arange(len(series))
        y = series.values
        # 计算线性回归斜率
        slope, _ = np.polyfit(x, y, 1)
        return slope
    
    df['slope'] = df['close'].rolling(window=window).apply(calculate_slope, raw=True)
    df['Trend'] = 0
    # 斜率为正，上升趋势
    df.loc[df['slope'] > 0, 'Trend'] = 1
    # 斜率为负，下降趋势
    df.loc[df['slope'] < 0, 'Trend'] = -1
    df['Trend'].fillna(0, inplace=True)
    return df

四、彻底修正支撑位判断函数 #

修正后的support函数仅使用当前及之前的K线数据，无前瞻偏差：

def support(df, index, n, col):
    # 避免索引越界
    if index < n:
        return False
    # 检查当前index是否为最近n根K线的最低点
    window_data = df[col].iloc[index - n + 1 : index + 1]
    if df[col].iloc[index] != window_data.min():
        return False
    # 检查最近n根K线是逐步下跌至当前低点
    for i in range(index - n + 1, index + 1):
        if df[col].iloc[i] > df[col].iloc[i - 1]:
            return False
    return True

def is_Support_Level(df, i):
    return support(df, i, 2, 'low')

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