如何用Python实现基于日内数据的每日高低价实时更新功能

原有代码问题根源 #

• 新日判断逻辑错误：self.day[i] > self.day[i - 1]的判断在跨月场景会失效（比如上月31号到本月1号，day[i] < day[i-1]），且额外增加的self.day[i] == self.day[i+1]判断会排除每日最后一根K线；同时触发新日时没有重置高低价初始值，导致数值跨日累计。
• 位置错位：你把当前计算位置i的结果关联到i-1的key上，自然会出现值偏移到前一行的问题。

修复后的类代码版本 #

import pandas as pd

class DailyRange:
    def __init__(self, high, low, length, time, day):
        self.high = high
        self.low = low
        self.length = length
        self.time = time
        self.day = day

    def daily_high(self):
        keys = []
        highs = []
        times = []
        date_times = []
        current_high = 0
        for i in range(self.length):
            # 处理第一根K线的边界情况
            if i == 0:
                current_high = self.high[i]
                keys.append(i)
                highs.append(current_high)
                times.append(self.day[i])
                date_times.append(self.time[i])
                continue
            # 新日判断，只要日期变化就触发重置
            if self.day[i] != self.day[i-1]:
                current_high = self.high[i]
            # 同日期判断是否新高
            elif self.high[i] > current_high:
                current_high = self.high[i]
            # 每根K线都存入当前的日高，不需要只存变化点
            keys.append(i)
            highs.append(current_high)
            times.append(self.day[i])
            date_times.append(self.time[i])
        high_data = {'Key': keys,
                     'Current Daily High': highs,
                     'Day of Month - High': times,
                     'Datetime - High': date_times}
        hd_df = pd.DataFrame(high_data)
        return hd_df

    def daily_low(self):
        keys = []
        lows = []
        times = []
        date_times = []
        current_low = float('inf')
        for i in range(self.length):
            if i == 0:
                current_low = self.low[i]
                keys.append(i)
                lows.append(current_low)
                times.append(self.day[i])
                date_times.append(self.time[i])
                continue
            if self.day[i] != self.day[i-1]:
                current_low = self.low[i]
            elif self.low[i] < current_low:
                current_low = self.low[i]
            keys.append(i)
            lows.append(current_low)
            times.append(self.day[i])
            date_times.append(self.time[i])
        low_data = {'Key': keys,
                    'Current Daily Low': lows,
                    'Day of Month - Low': times,
                    'Datetime - Low': date_times}
        ld_df = pd.DataFrame(low_data)
        return ld_df

注：修改后每根K线都会返回对应的当前日高/日低，不需要后续做外连接补空值，直接按Key匹配主DataFrame的索引即可

更简洁的Pandas原生实现（推荐） #

不需要自己写循环，直接用Pandas的分组+扩展窗口功能，三行代码即可实现全量计算，逻辑更稳定不易出错：
假设你的主行情DataFrame命名为df，包含字段：

• datetime：K线时间，格式为pandas datetime类型
• high：K线最高价
• low：K线最低价

# 提取日期分组字段
df['date'] = df['datetime'].dt.date
# 计算逐K线更新的日内最高值，每日自动重置
df['daily_high'] = df.groupby('date')['high'].expanding().max().reset_index(level=0, drop=True)
# 计算逐K线更新的日内最低值，每日自动重置
df['daily_low'] = df.groupby('date')['low'].expanding().min().reset_index(level=0, drop=True)

实时增量更新方案 #

如果是实盘逐根K线更新的场景，不需要每次全量重算，用三个变量跟踪即可，效率更高：

last_date = None
current_daily_high = 0
current_daily_low = float('inf')

# 每次新K线到来时执行以下逻辑
def update_daily_range(new_kline):
    global last_date, current_daily_high, current_daily_low
    cur_date = new_kline['datetime'].date()
    # 新日重置
    if cur_date != last_date:
        last_date = cur_date
        current_daily_high = new_kline['high']
        current_daily_low = new_kline['low']
    else:
        # 同日期更新高低点
        if new_kline['high'] > current_daily_high:
            current_daily_high = new_kline['high']
        if new_kline['low'] < current_daily_low:
            current_daily_low = new_kline['low']
    # 返回当前K线对应的日高低点
    return current_daily_high, current_daily_low

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