Python实现艾略特振荡器突破带计算异常，求故障排查

我正在Python中计算艾略特振荡器（EWO）的突破带数值，已有经过验证的C#实现逻辑。目前已在Python中计算并确认了突破带依赖的直方图（EWO_Std）数值正确，但当前Python逻辑输出的突破带结果不正确，多次调整仍未找到问题，请求帮忙排查。

Python现有实现代码 #

# 下轨计算，当前直接返回0：
osc_fast = 5
osc_slow = 35
lens = osc_fast + osc_slow
pr = 2 / lens
strength = 100
            
df = df.assign(Lwr_Line=0)
df['Lwr_Line'] = np.where(df["EWO_Std"] < 0, (df['EWO_Std']*pr) + (df['Lwr_Line'].shift(1)*(1-pr)), df['Lwr_Line'].shift(1))
df['LineEWOLwr'] = strength / 100 * df['Lwr_Line']
df.drop(columns='Lwr_Line', inplace=True)

# 上轨计算：
            
df = df.assign(Upr_Line=0)
df['Upr_Line'] = np.where(df["EWO_Std"] > 0, (df['EWO_Std']*pr) + (df['Upr_Line'].shift(1)*(1-pr)), df['Upr_Line'].shift(1))
df['LineEWOUpr'] = strength / 100 * df['Upr_Line']
df.drop(columns='Upr_Line', inplace=True)

验证通过的C#实现代码 #

{
    MP[0]       =   ( High[0] + Low[0] ) / 2;
    
    UprLine[0]  =   0;
    LwrLine[0]  =   0;
    Lens        =   OscFast + OscSlow;
    Pr = 2.0/Lens; 

    if(CurrentBar < OscSlow){
        OscAG = 0;
        if (OscAG > 0){
            OscAGUpr[0] = OscAG;
            if (OscAGUpr[0] > OscAGUpr[1]){
                OscAGUprDiv[0] = OscAG;
            }   
        }
        else{
            OscAGLwr[0] = OscAG;
            OscAGLwrDiv[0] = OscAG;             
        }
    }
    
    else{
        OscAG   =   SMA(MP,OscFast)[0] - SMA(MP,OscSlow)[0];
        if (OscAG > 0){
                UprLine[0]  = (OscAG*Pr) + (UprLine[1]*(1-Pr));
                LwrLine[0]  = LwrLine[1];
            OscAGUpr[0] = OscAG;
                if (OscAGUpr[0] > OscAGUpr[1])
                    {
                        OscAGUprDiv[0] = OscAG;
                    }

        }
        else{
            UprLine[0]      = UprLine[1];
            LwrLine[0]      = (OscAG*Pr) + (LwrLine[1]*(1-Pr));
            OscAGLwr[0]     = OscAG;
                if (OscAGLwr[0] > OscAGLwr[1])
                {
                    OscAGLwrDiv[0] = OscAG;
                }   
                
        }
    }
    
    LineEWOUpr[0]   =   BOBStrength / 100 * UprLine[0];
    LineEWOLwr[0]   =   BOBStrength / 100 * LwrLine[0];
    
}

问题排查与修复 #

核心差异点 #

1. 初始阶段逻辑缺失：C#中明确处理了CurrentBar < OscSlow（即前35根K线）的情况，此时OscAG固定为0，不会更新UprLine/LwrLine；但Python代码直接对所有K线应用条件判断，没有跳过初始阶段，导致初始值计算错误。
2. 向量化操作不适用递归逻辑：Python用np.where+shift(1)的向量化方式无法实现C#的逐行迭代更新——shift(1)取的是列的原始移位值，而非每一步计算后的实时更新值，这会导致递推逻辑失效。

修复后的Python代码 #

import numpy as np

osc_fast = 5
osc_slow = 35
lens = osc_fast + osc_slow
pr = 2 / lens
strength = 100

# 初始化中间计算列
df['Upr_Line'] = 0.0
df['Lwr_Line'] = 0.0

# 逐行迭代，完全对齐C#逻辑
for i in range(1, len(df)):
    if i < osc_slow:
        # 前osc_slow根K线，保持上一根的数值
        df.loc[i, 'Upr_Line'] = df.loc[i-1, 'Upr_Line']
        df.loc[i, 'Lwr_Line'] = df.loc[i-1, 'Lwr_Line']
    else:
        ewo_std = df.loc[i, 'EWO_Std']
        if ewo_std > 0:
            # 更新上轨，下轨沿用前值
            df.loc[i, 'Upr_Line'] = ewo_std * pr + df.loc[i-1, 'Upr_Line'] * (1 - pr)
            df.loc[i, 'Lwr_Line'] = df.loc[i-1, 'Lwr_Line']
        else:
            # 更新下轨，上轨沿用前值
            df.loc[i, 'Upr_Line'] = df.loc[i-1, 'Upr_Line']
            df.loc[i, 'Lwr_Line'] = ewo_std * pr + df.loc[i-1, 'Lwr_Line'] * (1 - pr)

# 计算最终突破带
df['LineEWOUpr'] = strength / 100 * df['Upr_Line']
df['LineEWOLwr'] = strength / 100 * df['Lwr_Line']

# 清理中间列
df.drop(columns=['Upr_Line', 'Lwr_Line'], inplace=True)

修复说明 #

• 采用逐行循环模拟C#的迭代逻辑，确保每一步的UprLine/LwrLine都是基于上一根的计算结果，符合递推要求。
• 明确处理前35根K线的特殊情况，和C#逻辑完全对齐。
• 直接使用已验证正确的EWO_Std列进行计算，保证数据源准确。

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