Python实现Binance 5分钟K线收盘后拉取数据，规避高频请求与数据偏差

核心实现逻辑 #

5分钟K线的收盘时间固定为每小时的0、5、10……55分的整分时间点，我们只需要每次计算下一个收盘时间点，精准等待到该时间再触发拉取，同时增加K线收盘校验，即可保证无论程序何时启动，都只会拿到完全收盘的K线数据。

实现要点 #

• 动态计算下一个5分钟整分收盘点的等待时长，避免固定休眠5分钟导致的时间漂移问题
• 拉取K线后校验最新K线的收盘时间，自动过滤未完结的K线数据
• 出错时增加重试等待，避免触发接口限流

修正后完整代码 #

依赖导入 #

import time
import datetime
import requests
import pandas as pd

K线拉取函数（增加收盘校验） #

def getKlines(symbol, length):
    headers = {'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/66.0.3359.139 Safari/537.36',
               "Connection": "keep-alive",
               "Accept-Encoding": "*"
               }
    payload = {'symbol': symbol, 'interval': '5m', 'limit': str(length)}
    result = requests.get(
        "https://api.binance.com/api/v3/klines", headers=headers, params=payload)
    # 替换原eval解析为官方json方法，安全性更高
    candles = pd.DataFrame(result.json())
    candles.columns = ['Date', 'Open', 'High', 'Low', 'Close', 'Volume', 'Close_Time',
                       'Quote_Volume', 'Trades_Count', 'BUY_VOL', 'BUY_VOL_VAL', 'x']    
    candles.insert(loc=1, column="Symbol", value=symbol)  
    candles["Date"] = pd.to_datetime(candles["Date"], unit="ms")
    candles["Close"] = candles["Close"].astype(float)
    candles["Volume"] = candles["Volume"].astype(float)
    # 校验最新K线是否已完全收盘
    latest_close_ts = candles["Close_Time"].iloc[-1]
    current_ts = int(time.time() * 1000)
    if latest_close_ts > current_ts:
        # 未收盘则丢弃最后一条未完结K线
        candles = candles.iloc[:-1]
    return candles

主循环（动态对齐收盘时间） #

# 替换为你需要拉取的交易对列表
symbols = ["BTCUSDT", "ETHUSDT"]
while True:
    try:
        # 计算下一个5分钟整分收盘时间
        now = datetime.datetime.now()
        add_min = 5 - (now.minute % 5)
        next_run_time = now + datetime.timedelta(minutes=add_min)
        # 对齐到整分，忽略秒和微秒
        next_run_time = next_run_time.replace(second=0, microsecond=0)
        # 计算需要等待的秒数
        wait_sec = (next_run_time - now).total_seconds()
        time.sleep(wait_sec)
        
        # 到点后批量拉取K线
        for ticker in symbols:
            candles = getKlines(symbol=ticker, length=48)
            print(candles)
    except Exception as e:
        print(f"运行异常: {str(e)}")
        # 异常后休眠10秒重试，避免频繁请求触发限流
        time.sleep(10)

方案说明 #

该方案无需依赖外部定时工具，每次循环都会自动对齐5分钟K线的收盘时间点，不会出现提前拉取、时间漂移导致的数据不匹配问题，即使程序中途重启，也会自动等待到下一个收盘点再执行拉取，完全符合需求。

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