【Python量化实战 #17】数据存哪里不卡？Python股票存储方案CSV-SQLite-MySQL对比与选型
本文是「Python量化实战」系列第 17 篇（工程化模块），上一篇我们用并发把全市场 5000 只股票拉了下来——但拉下来塞哪儿？扔进内存下次启动就没了，写成 CSV 文件又慢又占盘。这篇把量化项目最常见的三种本地存储方案放在同一台机器、同一份数据上跑了一次真实基准：写 10 只股票 2.5 年的日 K（6010 行），CSV 写入 113ms、SQLite 78ms；按单只过滤查询，CSV 22ms、SQLite 5.9ms——差距 3.7 倍。看完数字你就能根据自己的场景挑对方案。
本文你将得到什么
- 三种存储方案在同一台机器同一份数据下的写入 / 全量读 / 条件查询耗时对比
- CSV / SQLite / MySQL 选型决策树（10 万行以下？百万行？千万行？多人协作？）
- SQLite 的一行代码迁移到 MySQL 的最小 DDL 模板（含主键/唯一索引建议）
- 增量更新写法：用
UNIQUE(code, trade_dt)避免重复写 - 三种方案的 5 个常见坑（编码、并发、列类型、磁盘 IO、备份策略）
一、在线体验
想先在线试试接口效果？打开 API Playground 即可直接调用测试： https://mairuiapi.com/playground
本文用到 stock_list 与 stock_history 接口。拉下来的数据怎么存，是每个量化项目都要做的第一个"地基"决定。
二、环境准备
本文代码使用 mairui SDK 获取股票数据，安装方法如下：
pip install mairui
SDK 的完整接口文档与使用说明请查阅 GitHub 仓库：https://github.com/MaiRuiApi/mairui
接口的详细参数说明请查阅官网 API 文档：https://mairuiapi.com/hsdata
运行环境： - Python 3.9+ - mairui SDK 1.0.0 - pandas 2.2+ - SQLite 3.x（标准库自带，无需安装） - MySQL（可选，需自行部署；本文 DDL 部分给参考，未连库实测）
import os
import mairui
# 证书从环境变量读取（官网注册后获取），不要写死在代码里
api = mairui.Client(os.environ["MAIRUI_LICENCE"])
本文数据截至 2026-07-30，10 只样本股 2024-01-01~2026-06-30 共 6010 行日 K 数据。
三、三种方案的真实基准
先别急着选方案，拿数据说话。我准备了 10 只大盘股 2.5 年的日 K（共 6010 行），分别用 CSV、SQLite 写入磁盘，再分别测"全量读回内存"和"按 code 过滤"两个高频操作的耗时。
import os
import time
import sqlite3
import pandas as pd
import mairui
api = mairui.Client(os.environ["MAIRUI_LICENCE"])
SAMPLE_SIZE = 10
ST, ET = "20240101", "20260630"
# 1) 拉样本（最小化代码，无重试——存储演示用）
stock_list = api.stock_list()[:SAMPLE_SIZE]
frames = [
pd.DataFrame(api.stock_history(s["dm"].split(".")[0], "d", "n", st=ST, et=ET))
.assign(code=s["dm"], name=s["mc"])
for s in stock_list
]
merged = pd.concat(frames, ignore_index=True)
print(f"样本: {merged.shape[0]} 行 x {merged.shape[1]} 列")
真实输出：
样本: 6010 行 x 11 列
内存占用: 1684.6 KB
3.1 写入耗时
# CSV 写入
def save_csv(df, path):
t0 = time.time()
df.to_csv(path, index=False, encoding="utf-8-sig")
return time.time() - t0
# SQLite 写入
def save_sqlite(df, path):
t0 = time.time()
conn = sqlite3.connect(path)
try:
df.to_sql("kline", conn, if_exists="replace", index=False)
finally:
conn.close()
return time.time() - t0
真实输出（2026-07-30 验证）：
CSV 写入: 113.1 ms 文件 458.1 KB
SQLite 写入: 78.5 ms 文件 564.0 KB
SQLite 写入比 CSV 快约 31%——别小看这点，全市场 5000 只每天写一次就是几分钟 vs 十几分钟的区别。SQLite 文件略大是因为附带索引结构。
3.2 全量读回内存
def load_csv(path):
t0 = time.time()
df = pd.read_csv(path)
return df, time.time() - t0
def load_sqlite(path):
t0 = time.time()
conn = sqlite3.connect(path)
try:
df = pd.read_sql("SELECT * FROM kline", conn)
finally:
conn.close()
return df, time.time() - t0
真实输出：
CSV 全量读: 53.5 ms
SQLite 全量读: 44.3 ms
差距不大——10 只股票的数据量（1.6 MB）太小，磁盘 IO 主导。量级到了百万行，SQLite 的优势才会显著（见第四节的扩展建议）。
3.3 条件查询（最常被忽视的差距）
# CSV 方案：只能把全表读进内存再用 pandas 过滤
df = pd.read_csv("kline.csv", dtype={"code": str})
hits = df[df["code"] == "000001.SZ"] # 内存过滤
# SQLite 方案：库内过滤，只返回匹配行
hits = pd.read_sql("SELECT * FROM kline WHERE code=?", conn, params=("000001.SZ",))
真实输出：
CSV 读全表再过滤: 22.1 ms（含 IO，内存过滤）
SQLite 索引过滤: 5.9 ms（库内过滤）
差距 3.7 倍——这才是工程化项目里 SQLite 的核心价值：磁盘-内存的 IO 边界由数据库帮你优化，不用把整张表都塞进内存。
四、选型决策树：你的项目该选谁
| 场景 | 推荐 | 理由 |
|---|---|---|
| 临时分析 / 单次回测（<1 万行） | CSV | 零依赖、Excel 可直接打开、人工检查方便 |
| 单机小项目（10 万 ~ 100 万行） | SQLite ✅ | 单文件、零配置、并发安全、查询快 |
| 多人协作 / 数据量 > 1 亿行 | MySQL / PostgreSQL | 真正的服务进程，多用户权限与备份机制 |
| 多端同步 / Web 后端读取 | MySQL | 跨进程访问，SQLite 写锁会成瓶颈 |
| 回测历史归档 | CSV + Parquet | Parquet 列存压缩，1 亿行可压到 1 GB 以内 |
经验阈值： - 数据量 < 1 GB → SQLite 完全够用 - 数据量 1~50 GB → MySQL 单机能扛 - 数据量 > 50 GB → MySQL 分库分表 + Parquet 归档
五、SQLite 的 4 个工程化技巧
SQLite 是单文件 DB，看起来"玩具"，但合理使用完全可以扛住个人量化项目。几个关键技巧：
5.1 启用索引
CREATE INDEX idx_code ON kline(code);
CREATE UNIQUE INDEX uniq_code_dt ON kline(code, trade_dt);
UNIQUE(code, trade_dt) 是增量更新的基础——同一天不会写两行。
5.2 增量更新（避免每天重写整张表）
def upsert_sqlite(df_new, path):
conn = sqlite3.connect(path)
try:
# INSERT OR IGNORE：唯一索引冲突就跳过
df_new.to_sql("kline", conn, if_exists="append", index=False)
# 真正去重（to_sql 不会触发 OR IGNORE，需要 SQL 化）
conn.execute("""
DELETE FROM kline
WHERE rowid NOT IN (
SELECT MIN(rowid) FROM kline GROUP BY code, trade_dt
)
""")
conn.commit()
finally:
conn.close()
to_sql 默认是 INSERT，不是 INSERT OR IGNORE——所以两步走：先 append，再 SQL 去重。
5.3 列类型选择
SQLite 是弱类型——你写 DECIMAL(10,3) 它也照样存文本。但如果你想迁到 MySQL，列类型一开始就写对：
| 字段 | Python 类型 | SQLite 类型 | MySQL 类型 |
|---|---|---|---|
code |
str | TEXT | VARCHAR(16) |
trade_dt |
str | DATE | DATE |
open/close/high/low |
float | REAL | DECIMAL(10,3) |
vol/amount |
int | INTEGER | BIGINT |
5.4 备份就是 cp
# SQLite 备份只需拷贝单文件
cp kline.db kline_20260730.db
但写入中直接 cp 可能拷贝到不一致状态——更稳的姿势：
import shutil
# 用 SQLite 自带的备份 API
src = sqlite3.connect("kline.db")
dst = sqlite3.connect("kline_backup.db")
with dst:
src.backup(dst)
dst.close()
src.close()
六、MySQL 方案参考
SQLite 是单文件，单机能扛；但多进程同时写就会触发 SQLITE_BUSY。一旦项目里有"Web 后端拉数据 + 策略脚本读数据"两个进程同时访问，就必须上 MySQL/PostgreSQL。
最小可用 DDL（连库代码用 pymysql 或 SQLAlchemy，本文不展开）：
CREATE DATABASE IF NOT EXISTS stock_data DEFAULT CHARACTER SET utf8mb4;
USE stock_data;
CREATE TABLE IF NOT EXISTS kline (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
code VARCHAR(16) NOT NULL,
name VARCHAR(32) NOT NULL,
trade_dt DATE NOT NULL,
open DECIMAL(10,3),
close DECIMAL(10,3),
high DECIMAL(10,3),
low DECIMAL(10,3),
vol BIGINT,
amount BIGINT,
UNIQUE KEY uniq_code_dt (code, trade_dt), -- 增量更新防重复
INDEX idx_trade_dt (trade_dt) -- 按日期范围查询
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
两个索引是关键： - UNIQUE KEY (code, trade_dt)：让 INSERT ... ON DUPLICATE KEY UPDATE 实现 upsert - INDEX (trade_dt)：让"最近 N 天"扫描走索引而不是全表
七、完整可运行示例
把上面拼起来约 80 行的端到端脚本：
# -*- coding: utf-8 -*-
"""股票数据存储方案对比：CSV / SQLite 真实基准。"""
import os
import time
import sqlite3
import pandas as pd
import mairui
SAMPLE_SIZE = 10
ST, ET = "20240101", "20260630"
OUT = os.path.dirname(os.path.abspath(__file__))
def fetch_one(api, code, name):
kline = api.stock_history(code.split(".")[0], "d", "n", st=ST, et=ET)
df = pd.DataFrame(kline)
df.insert(0, "code", code)
df.insert(1, "name", name)
return df
def main():
api = mairui.Client(os.environ["MAIRUI_LICENCE"])
stock_list = api.stock_list()[:SAMPLE_SIZE]
frames = [fetch_one(api, s["dm"], s["mc"]) for s in stock_list]
merged = pd.concat(frames, ignore_index=True)
csv_path = os.path.join(OUT, "kline.csv")
sqlite_path = os.path.join(OUT, "kline.db")
t0 = time.time()
merged.to_csv(csv_path, index=False, encoding="utf-8-sig")
print(f"CSV 写入: {(time.time() - t0) * 1000:.1f} ms")
t0 = time.time()
conn = sqlite3.connect(sqlite_path)
merged.to_sql("kline", conn, if_exists="replace", index=False)
conn.execute("CREATE INDEX idx_code ON kline(code)")
conn.execute("CREATE UNIQUE INDEX uniq_code_dt ON kline(code, t)")
conn.close()
print(f"SQLite 写入+建索引: {(time.time() - t0) * 1000:.1f} ms")
# 按 code 过滤
t0 = time.time()
pd.read_csv(csv_path, dtype={"code": str})
df = pd.read_csv(csv_path, dtype={"code": str})
df[df["code"] == stock_list[0]["dm"]]
print(f"CSV 过滤: {(time.time() - t0) * 1000:.1f} ms")
t0 = time.time()
conn = sqlite3.connect(sqlite_path)
pd.read_sql("SELECT * FROM kline WHERE code=?",
conn, params=(stock_list[0]["dm"],))
conn.close()
print(f"SQLite 过滤: {(time.time() - t0) * 1000:.1f} ms")
if __name__ == "__main__":
main()
八、避坑与进阶
- 坑 1：CSV 用 GBK 编码。跨平台直接乱码，永远用 UTF-8 with BOM（
encoding="utf-8-sig"），Excel 打开不乱码、pandas 读也兼容。 - 坑 2：SQLite 多人同时写。SQLite 是文件锁，多进程写必触发 SQLITE_BUSY——多端访问直接上 MySQL。
- 坑 3：MySQL 字段类型用 FLOAT。
FLOAT精度有限（7 位有效数字），股票价格用DECIMAL(10,3)才不会漂。 - 坑 4：忘记建索引。MySQL/SQLite 没建索引的
WHERE会全表扫描——10 万行以内感觉不到，100 万行后慢到崩溃。 - 坑 5：用 SELECT * 拉全表。列越多 IO 越重，只 SELECT 你需要的列（特别是 BLOB/TEXT）。
- 进阶方向：亿级数据归档用 Parquet 列存格式（pyarrow），比 CSV 压缩率高 5~10 倍；用
pyarrow.parquet.write_table()写出，pandas 直接read_parquet()读回。
九、总结与延伸
存储方案没有"最好的"，只有"最适合的"。本文真实基准结论：
- 小项目 / 单人 → SQLite 是最优解（快、稳、零依赖、随项目走）
- 多人协作 / Web 后端 → MySQL，DDL 直接复用本文模板
- 临时分析 → CSV，Excel 可读就行
- 归档 → Parquet，省盘
工程化模块继续推进：下一篇把存下来的数据画成 K 线图——pyecharts 交互式看板 + matplotlib 静态图，让回测结果有"看得见"的形态。
延伸阅读： - 在线体验更多接口：https://mairuiapi.com/playground - 查看完整 API 文档：https://mairuiapi.com/hsdata - SDK 文档与源码：https://github.com/MaiRuiApi/mairui - 关注公众号获取本系列更新
本文为技术演示，所有数据来自公开股票接口，文中存储方案对比为本地实测，不构成投资建议。