计算水下周期时starting_dates与ending_dates长度不匹配排查

我写了一个计算标普500序列水下周期的period_under_water函数，用来识别高水位线（HMW）——也就是价格指数中高于此前所有值且随后出现下跌的点位，同时记录这些HMW的起始日期，以及它们被再次超越的结束日期。

原函数代码 #

import pandas as pd
import numpy as np

def period_under_water(series: pd.Series):

    max_to_date = np.empty(len(series))
    max_to_date[0] = series.iloc[0]

    idx_max = np.empty(len(series), dtype='int')
    idx_max[0] = 0

    T = len(series.index)

    HMW = []
    control = False

    for t in range(1, T):
        current_val = series.iloc[t]
        
        if current_val < max_to_date[t-1]:
            max_to_date[t] = max_to_date[t-1]
            idx_max[t] = idx_max[t-1]
            
            if not control:
                HMW.append(series.iloc[t-1])
                control = True
            
        else:
            max_to_date[t] = current_val
            idx_max[t] = t
            control = False

    starting_dates = series[series.isin(HMW)].index.tolist()

    rel_loss = (max_to_date - series) / max_to_date
    ending_dates = []

    for i in range(1, len(rel_loss)):
        if rel_loss[i] == 0 and rel_loss[i-1] > 0:  # 回撤结束的条件
            ending_dates.append(series.index[i])
  
    return HMW, starting_dates, ending_dates

测试代码 #

import yfinance as yf

start_date = "2000-1-1"
end_date = "2022-12-31"

sp_500 = yf.download("SPY", start = start_date, end = end_date)[["Adj Close"]]
sp_500 = sp_500.squeeze()

现在运行后发现starting_dates长度是240，ending_dates长度是234，不符合预期（预期二者长度相等或starting_dates比ending_dates多1），请问哪里出错了？

错误原因分析&修正方案 #

1. 核心问题：starting_dates的获取逻辑错误

原代码用series[series.isin(HMW)].index.tolist()来获取HMW的日期，这会导致重复匹配：如果不同时间点出现了相同的收盘价（SPY的调整收盘价可能存在重复值），isin会把所有等于HMW值的日期都捞出来，而不是只取那个触发下跌的新高点，直接导致starting_dates长度被夸大。

2. 次要问题：浮点数精度导致结束日期漏判

原代码用rel_loss[i] == 0判断回撤结束，由于浮点数计算的精度误差，实际中rel_loss[i]几乎不可能精确等于0，导致很多本该被识别的结束点被漏掉，造成ending_dates长度不足。

修正后的函数代码

import pandas as pd
import numpy as np

def period_under_water(series: pd.Series):
    max_to_date = np.empty(len(series))
    max_to_date[0] = series.iloc[0]
    idx_max = np.empty(len(series), dtype='int')
    idx_max[0] = 0
    T = len(series.index)
    
    HMW = []
    starting_dates = []
    control = False  # 标记是否处于回撤状态
    
    for t in range(1, T):
        current_val = series.iloc[t]
        prev_max = max_to_date[t-1]
        
        if current_val < prev_max:
            max_to_date[t] = prev_max
            idx_max[t] = idx_max[t-1]
            
            if not control:
                # 第一次进入回撤，前一个点就是HMW，直接记录日期
                HMW.append(prev_max)
                starting_dates.append(series.index[t-1])
                control = True
        else:
            max_to_date[t] = current_val
            idx_max[t] = t
            control = False
    
    # 计算回撤幅度，用近似等于处理浮点数精度问题
    rel_loss = (max_to_date - series) / max_to_date
    ending_dates = []
    
    for i in range(1, len(rel_loss)):
        # 用np.isclose判断是否回到新高，同时确保前一个时刻确实在回撤中
        if np.isclose(rel_loss[i], 0, atol=1e-8) and rel_loss[i-1] > 1e-8:
            ending_dates.append(series.index[i])
    
    return HMW, starting_dates, ending_dates

修正点说明

• 直接记录起始日期：在识别到HMW的瞬间（第一次进入回撤时），直接把对应日期加入starting_dates，彻底避免isin带来的重复匹配问题。
• 浮点数精度兼容：用np.isclose替代精确等于判断，设置合理的容差（atol=1e-8），确保能正确识别回撤结束的节点。
• 逻辑一致性：严格遵循HMW的定义——只有当新高之后第一次出现下跌时，才记录这个新高为HMW，避免无效记录。

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