用 SQLite 给行情数据做一层本地缓存：把一次复盘从 8 分钟压到 20 秒

复盘系统的瓶颈往往不在策略，而在「每次运行都要重新把两年的行情拉一遍」。这篇文章记录我给本地 A 股复盘脚本加一层 SQLite 缓存的完整过程，包含表设计、增量更新、一致性校验和几个真实踩过的坑。

一、问题：慢在重复劳动上

我的每日复盘流程大概是：拉全市场日线 → 计算指标 → 跑多套筛选策略 → 输出报告。最早的版本是纯内存直连接口，一次全量拉取要 6~8 分钟，其中：

• 网络往返占了九成以上时间，CPU 基本闲着；
• 接口有频率限制，拉快了会被限流，拉慢了又拖长总时长；
• 中途想改个参数重跑一次，就得再等 8 分钟；
• 断网、接口抖动、盘中数据未更新，都会让跑到一半的流程直接失败。

后来我把「取数」和「算数」拆开，中间垫一层 SQLite，同样的流程降到 20 秒左右。核心思路只有一句：数据拉一次就落地，之后只拉增量。

二、表设计：让写入天然幂等

日线数据的自然主键是「代码 + 交易日」，直接用复合主键，配合 INSERT OR REPLACE，重复写入就是覆盖，不需要先查再判断：

CREATE TABLE IF NOT EXISTS daily_bar (
  code      TEXT NOT NULL,
  trade_date TEXT NOT NULL,     -- 统一用 YYYYMMDD 字符串，避免时区歧义
  open      REAL,
  high      REAL,
  low       REAL,
  close     REAL,
  volume    REAL,
  amount    REAL,
  adj_factor REAL,              -- 复权因子，后面讲为什么必须存
  updated_at INTEGER,           -- 写入时间戳，用于排查陈旧数据
  PRIMARY KEY (code, trade_date)
);
CREATE INDEX IF NOT EXISTS idx_bar_date ON daily_bar(trade_date);
CREATE INDEX IF NOT EXISTS idx_bar_code ON daily_bar(code);

两个习惯值得强调：

1. 日期用 YYYYMMDD 字符串而不是 DATETIME。行情数据本质是「交易日」这个业务概念，不是时间点；用字符串排序即时间序，还能彻底避开 SQLite 日期函数的时区坑。
2. 建两个索引，不是省不省的问题。按日期扫全市场、按代码拉单只股票，是复盘里最高频的两种读法，缺一个索引差距就是几十倍。

写入用 executemany 批量提交，并且放在一个事务里：

def upsert_bars(con, rows):
    sql = """INSERT OR REPLACE INTO daily_bar
             (code, trade_date, open, high, low, close, volume, amount, adj_factor, updated_at)
             VALUES (?,?,?,?,?,?,?,?,?,?)"""
    now = int(time.time())
    payload = [(r['code'], r['date'], r['open'], r['high'], r['low'],
                r['close'], r['volume'], r['amount'], r.get('adj', 1.0), now) for r in rows]
    with con:                      # 事务：要么全进，要么全不进
        con.executemany(sql, payload)

三、增量更新：只拉缺的那一段

缓存层的关键不是「存」，而是「知道该补什么」。做法很直接：对每个代码先查本地最大交易日，只请求它之后的区间。

def last_dates(con) -> dict:
    cur = con.execute("SELECT code, MAX(trade_date) FROM daily_bar GROUP BY code")
    return {c: d for c, d in cur.fetchall()}

def need_update(con, code, today) -> str | None:
    row = con.execute("SELECT MAX(trade_date) FROM daily_bar WHERE code=?", (code,)).fetchone()
    have = row[0] if row and row[0] else None
    if have is None:
        return '20240101'          # 首次拉取，给一个起始日
    if have >= today:
        return None                # 已是最新，跳过
    return next_trade_day(have)    # 从下一个交易日开始补

上线后每天的增量只有几百行，网络耗时从几分钟变成几百毫秒。首次冷启动仍然要拉全量，但那是一次性成本，之后每天都只补当天那几根 K 线。

四、读侧：一次性读成 DataFrame

不要在循环里一条条查，把整个区间一次性读出来再交给 pandas：

def load_bars(con, start, end, codes=None):
    sql = "SELECT code, trade_date, open, high, low, close, volume, amount, adj_factor " \
          "FROM daily_bar WHERE trade_date BETWEEN ? AND ?"
    args = [start, end]
    if codes:
        sql += f" AND code IN ({','.join('?' * len(codes))})"
        args += codes
    df = pd.read_sql_query(sql, con, params=args, parse_dates=['trade_date'])
    return df.sort_values(['code', 'trade_date']).reset_index(drop=True)

配合上面的索引，读 5000 只股票 × 500 个交易日的量级，实测在 3 秒以内，而且这部分耗时是稳定的——不管接口今天快不快、限不限流。

五、一致性：缓存最怕的不是慢，是悄悄变脏

缓存层引入的最大风险是「你以为数据是对的，其实它是旧的或者错的」。我加了三道防线：

1. 复权因子单独存。 除权除息会让历史价格被追溯修改，如果只缓存收盘价，某只股票送股之后，你的历史数据就和真实行情对不上了。存下复权因子，读的时候按当次计算需要的口径（前复权 / 后复权 / 不复权）现算，而不是缓存死值。

2. 缺口检测。 每天拉完增量后扫一遍，找出「有开盘但无收盘」「连续交易日断档超过 5 天」「成交量为 0 但涨跌幅不为 0」这类异常，落到一个 data_issue 表里，报告里单独列出来。这个表帮我抓到过两次接口返回的部分字段缺失——如果不检测，它会 silently 变成策略里的 NaN。

3. 陈旧度水位。 记录每张表的 MAX(trade_date)，跑复盘前先比对「今天应该是哪个交易日」。如果缓存里的最大日期落后于当前交易日，就明确报出来，而不是拿 T-1 的数据假装是 T 的。这点在盘后马上跑复盘时尤其重要：接口数据往往延迟十几分钟才齐。

def freshness_check(con, expect_date):
    got = con.execute("SELECT MAX(trade_date) FROM daily_bar").fetchone()[0]
    if got != expect_date:
        raise RuntimeError(f"缓存陈旧：期望 {expect_date}，实际 {got}，请先执行增量更新")

宁可报错，也不要静默用旧数据出报告。

六、两个性能细节

开 WAL 模式。 默认的回滚日志模式下，读写会互相阻塞。复盘脚本一边跑一边想补数据时尤其明显：

con = sqlite3.connect(DB, timeout=30)
con.execute("PRAGMA journal_mode=WAL")
con.execute("PRAGMA synchronous=NORMAL")

定期 VACUUM。 每天 INSERT OR REPLACE 会留下页碎片，一周做一次 VACUUM 即可，不必每次都做。

七、实际收益

同一个复盘任务，改造前后的对比（本地实测，5000+ 只标的、约 500 个交易日）：

更有价值的是迭代速度：以前调一个参数要等 8 分钟才知道结果，现在 20 秒，一天能多试十几个想法。对量化这种「靠大量试错收敛」的工作来说，这个收益比单纯省时间大得多。

八、小结

这层缓存的代码总量不到 200 行，没有引入任何外部服务，但它把「取数」这个不可控环节变成了可控的本地操作。回头看，几个原则值得记住：

• 主键选自然业务键，让写入幂等，省掉大量判重逻辑；
• 存原始 + 存口径因子，而不是存算好的结果值；
• 任何缓存都要配一致性检查，慢可以忍，脏数据不能忍；
• 报错优于猜测：数据不新鲜就明确失败，不要让用户拿着旧报告做决策。

如果你的脚本现在还在每次运行都全量拉接口，值得花半小时加这一层——后面每一次运行都会把这半小时还给你。