Pandas DataFrame或SQLite数据库新增行时计算简单移动平均SMA的方法

最优标准实现方案 #

针对滑动窗口类指标的实时计算场景，行业通用的实现逻辑是增量计算+固定长度内存缓存，不需要全量拉取历史数据、也不需要维护完整DataFrame，性能和稳定性都远高于你列的三个基础方案：
你用到的SMA7、SMA25本质只依赖最近25个收盘价数据，完全不需要读取全量历史，只需要维护一个最大长度为25的固定长度缓存即可，核心逻辑如下：

from collections import deque
import sqlite3
import pandas as pd

# 服务启动初始化阶段：仅拉取最大窗口长度的历史数据
max_window = 25
conn = sqlite3.connect('binance_kline.db')
# 只查最近25条收盘价，避免全量读取的性能开销
recent_klines = pd.read_sql("SELECT Close FROM kline_table ORDER BY timestamp DESC LIMIT ?", conn, params=(max_window,))
# 初始化固定长度队列，新数据加入时自动淘汰最旧的数据
close_queue = deque(recent_klines['Close'].tolist(), maxlen=max_window)
conn.close()

# 实时websocket推送新K线时的处理逻辑
def on_new_kline(new_kline_dict):
    # 新收盘价加入缓存队列
    close_queue.append(new_kline_dict['Close'])
    # 计算指标，队列长度不足窗口时可返回空或自行补全逻辑
    new_kline_dict['SMA7'] = sum(list(close_queue)[-7:])/7 if len(close_queue)>=7 else None
    new_kline_dict['SMA25'] = sum(close_queue)/25 if len(close_queue)>=25 else None
    # 单条数据追加写入数据库，IO开销极低
    conn = sqlite3.connect('binance_kline.db')
    pd.DataFrame([new_kline_dict]).to_sql('kline_table', conn, if_exists='append', index=False)
    conn.close()

你提到的三个方案的适用场景 #

• 方案1（拉历史计算再追加）：仅作为服务重启、缓存失效、数据校验失败的兜底逻辑使用，日常实时流处理不要频繁调用，重复读库会增加不必要的IO开销。
• 方案2（同时维护数据库和缓存）：将你原本设想的全量DataFrame替换为上述固定长度队列后，就是当前场景的最优解，既不会出现内存溢出问题，计算效率也最高，不需要额外读库。
• 方案3（全量重写数据库）：完全不推荐日常使用，仅在需要全量回溯重新计算所有历史指标的特殊场景下偶尔使用，该方案的写入性能会随数据量增长线性下降，数据量过大会直接导致写入超时。

注意事项 #

• 可以按固定周期（比如每100条K线）做一次数据一致性校验：拉取最近1条入库数据和写入的新行做对比，防止websocket丢数、写入失败导致的缓存与数据库数据不一致。
• 如果后续新增更长周期的技术指标，只需要调整固定队列的maxlen为你用到的最大窗口长度即可，整体逻辑不需要修改。
• 其他调用数据库的脚本可以直接读取已经预计算好的SMA7、SMA25字段，不需要重复计算指标。

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