Python下使用WebSocket与SQLAlchemy积累SQL数据及避免K线重复

导致K线重复入库的核心原因有两点：

1. SQLite表未对时间维度设置唯一性约束，相同时间的K线数据可被重复写入
2. 现有代码中WebSocket的使用姿势错误，加上网络波动重连时币安会重推最近的K线闭合通知，导致同一条闭合消息被多次处理

方案1：数据库层面加唯一约束（最稳妥） #

建表时将time字段设为唯一主键，从底层直接拒绝重复时间的数据写入，无需在业务代码中做额外判断。
首次运行时先执行一次建表语句添加约束：

from sqlalchemy import text

# 仅需执行一次，后续运行可注释
with engine.connect() as conn:
    conn.execute(text("""
    CREATE TABLE IF NOT EXISTS BTCUSDT (
        time TEXT PRIMARY KEY,
        open REAL,
        high REAL,
        low REAL,
        close REAL,
        volume REAL
    )
    """))
    conn.commit()

后续写入时自动遵循主键约束，重复数据会被自动忽略，可配合INSERT OR IGNORE语法避免冲突报错。

方案2：新增本地缓存校验（性能最优） #

在内存中记录上一次成功入库的K线起始时间戳（即消息内k.t字段），每次收到闭合K线时先比对时间戳，相同则直接跳过写入流程：

last_saved_kline_ts = 0  # 记录上次入库的K线起始时间戳

async with socket as sock:
    while True:
        msg = await sock.recv()
        kline_data = msg['k']
        # 仅处理闭合且未入库的K线
        if kline_data['x'] and kline_data['t'] != last_saved_kline_ts:
            frame = createframe(msg)
            frame.to_sql(kline_data['s'], engine, if_exists='append', index=False)
            last_saved_kline_ts = kline_data['t'] # 更新缓存时间戳

方案3：修正WebSocket连接逻辑 #

原有代码中每次循环都调用await socket.__aenter__()属于错误用法，会导致连接反复重连，重复接收相同消息。正确的异步WebSocket使用方式应将上下文管理器放在循环外层：

# 全局创建一次连接，不要在循环内重复初始化
async with bsm.kline_socket('BTCUSDT', interval=KLINE_INTERVAL_1MINUTE) as socket:
    last_saved_ts = 0
    while True:
        msg = await socket.recv()
        k = msg['k']
        if k['x'] and k['t'] != last_saved_ts:
            frame = createframe(msg)
            frame.to_sql(k['s'], engine, if_exists='append', index=False)
            last_saved_ts = k['t']

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