如何基于Python pandas将实时tick数据转换为OHLC数据

你贴的示例代码仅实现了币安ETHUSDT实时成交价的轮询拉取逻辑，本身没有包含OHLC聚合部分，自然无法直接输出K线数据。
实时场景下不需要攒全量历史tick再调用pandas的resample函数，通过维护滚动时间窗口的OHLC状态，就能在每个时间窗口边界触发时输出完整K线，全程内存占用极低，延迟可以做到毫秒级。

核心实现逻辑 #

• 提前定义目标K线周期（比如1分钟、5分钟），写工具函数把任意时间戳对齐到对应K线周期的起始点
• 内存中维护一个当前正在生成的K线缓存，存储当前K线的开盘时间、open/high/low/close四个核心值，需要成交量的话可以额外加volume字段
• 每收到一条新的tick数据，先计算它所属的K线周期起始时间：
  • 如果和当前缓存K线属于同一个周期：更新high为当前价和缓存high的最大值，low为当前价和缓存low的最小值，close更新为最新成交价
  • 如果属于后续的新周期：先把当前缓存的K线作为完整K线输出（可以对接策略计算、存库等逻辑），再初始化新周期的K线缓存，新K线的open/high/low/close初始值都取这条新tick的价格
• 生产环境可以按需补全跨周期断层的空K线，避免逻辑异常

可直接运行的参考代码 #

import time
import requests
from datetime import datetime

# 基础配置
SYMBOL = "ETHUSDT"
# K线周期单位为秒，示例为1分钟K线，修改为300即为5分钟K线
KLINE_INTERVAL = 60

def fetch_latest_tick():
    """拉取最新成交价，加简单重试逻辑避免接口报错中断"""
    while True:
        try:
            resp = requests.get(
                f'https://api1.binance.com/api/v3/ticker/price?symbol={SYMBOL}',
                timeout=5
            ).json()
            return time.time(), float(resp["price"])
        except Exception as e:
            print(f"Tick拉取失败，正在重试: {str(e)}")
            time.sleep(0.5)

def align_to_kline_start(ts: float, interval: int) -> int:
    """将任意时间戳对齐到对应周期K线的起始时间"""
    return int(ts // interval * interval)

if __name__ == "__main__":
    # 初始化第一根K线
    init_ts, init_price = fetch_latest_tick()
    current_kline_start = align_to_kline_start(init_ts, KLINE_INTERVAL)
    current_kline = {
        "open_ts": current_kline_start,
        "open": init_price,
        "high": init_price,
        "low": init_price,
        "close": init_price
    }
    print(f"实时K线生成服务启动，当前K线周期: {KLINE_INTERVAL}秒")

    while True:
        tick_ts, tick_price = fetch_latest_tick()
        tick_kline_start = align_to_kline_start(tick_ts, KLINE_INTERVAL)

        # 到达新K线周期，输出上一根完整K线
        if tick_kline_start > current_kline["open_ts"]:
            kline_end_ts = current_kline["open_ts"] + KLINE_INTERVAL
            print("="*60)
            print(f"完整K线时间范围: {datetime.fromtimestamp(current_kline['open_ts'])} ~ {datetime.fromtimestamp(kline_end_ts)}")
            print(
                f"OHLC数据: 开盘{current_kline['open']:.2f} | 最高{current_kline['high']:.2f} | "
                f"最低{current_kline['low']:.2f} | 收盘{current_kline['close']:.2f}"
            )
            # 此处可以插入自定义逻辑：K线存库、传给策略计算等

            # 初始化新周期的K线
            current_kline = {
                "open_ts": tick_kline_start,
                "open": tick_price,
                "high": tick_price,
                "low": tick_price,
                "close": tick_price
            }
        # 同周期内更新当前未走完K线的字段
        else:
            current_kline["high"] = max(current_kline["high"], tick_price)
            current_kline["low"] = min(current_kline["low"], tick_price)
            current_kline["close"] = tick_price

        # 控制请求频率，避免触发接口频次限制
        time.sleep(1)

落地注意事项 #

• 最简示例仅实现了价格OHLC聚合，如果需要成交量、成交额维度，把拉取接口换成逐笔成交接口，在K线缓存里增加对应字段累加即可
• 不建议在实时场景下攒固定数量tick就反复调用resample做聚合，效率低且容易出现K线边界计算错误，滚动维护状态的方案性能高、延迟低，完全满足实盘需求
• 轮询接口的延迟相对较高，如果对实时性要求高，可以替换为对应交易场所的websocket逐笔推送流，逻辑不需要改动
• 生产环境建议增加跨周期空K线补全逻辑：如果相邻两个tick间隔跨了多个K线周期，可按业务规则补全中间空缺的K线，避免后续策略计算出错

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