1. 为什么要把 tushare 数据落到 MySQL 而不是停在 CSV做量化研究或者数据回测的人几乎都绕不开一个环节把行情数据从接口拉下来然后存到一个能反复查询、能增量更新、能多进程共享的地方。很多人一开始图省事直接df.to_csv()存本地文件跑单机小策略没问题但一旦数据量上来、标的变多、需要按日期切片查询CSV 的短板就暴露得很彻底——每次读取都要全量加载没法建索引多个脚本同时写还容易冲突。tushare 本身提供的是 HTTP 接口返回的是 pandas DataFrame它不负责帮你持久化。所以真正的问题不是怎么调接口而是怎么把接口返回的数据高效、稳定、可重复地写进关系型数据库。MySQL 在这个场景里是一个很务实的选择免费、生态成熟、Python 驱动稳定、支持INSERT ... ON DUPLICATE KEY UPDATE这种幂等写入配合索引之后按ts_code trade_date查询几乎是毫秒级。这篇内容适合三类人看一是刚接触 tushare、想把数据管起来的量化新手二是已经用 CSV 跑了一段时间、被数据管理折磨到想换方案的人三是需要把行情数据接入自己后端服务的开发者。我会把整个链路的选型理由、表结构设计、批量写入写法、增量更新逻辑、以及我自己踩过的坑都摊开讲代码可以直接抄。需要先说明一点tushare 的接口有积分门槛不同接口权限不同本文不讨论积分获取只讨论拿到 DataFrame 之后怎么进 MySQL这一段。另外所有涉及具体接口名的地方你按自己账号实际能调的接口替换即可逻辑是通用的。2. 表结构设计先想清楚主键和字段类型再动手2.1 为什么主键设计决定了后面所有的写入逻辑很多人建表的时候随手给个自增 id 当主键然后每次拉数据都INSERT结果跑第二遍就出现重复行只能靠DELETE全表再重灌。这种做法在小数据量下能忍但日线数据动辄几百万行全表重灌的时间成本完全不可接受。正确的思路是用业务字段做唯一约束。对于日线行情天然的唯一键就是某只股票在某一天这一组合也就是(ts_code, trade_date)。有了这个唯一键写入时就可以用INSERT ... ON DUPLICATE KEY UPDATE实现存在则更新、不存在则插入的幂等语义。这样无论你跑多少次、补多少次数据结果都是一致的不会产生脏数据。我一般会这样建表CREATE TABLE stock_daily ( ts_code VARCHAR(12) NOT NULL COMMENT 股票代码, trade_date DATE NOT NULL COMMENT 交易日期, open DECIMAL(10,3) DEFAULT NULL, high DECIMAL(10,3) DEFAULT NULL, low DECIMAL(10,3) DEFAULT NULL, close DECIMAL(10,3) DEFAULT NULL, pre_close DECIMAL(10,3) DEFAULT NULL, change DECIMAL(10,3) DEFAULT NULL, pct_chg DECIMAL(10,4) DEFAULT NULL, vol DECIMAL(20,2) DEFAULT NULL, amount DECIMAL(20,2) DEFAULT NULL, update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (ts_code, trade_date), KEY idx_trade_date (trade_date) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;这里有几个细节值得展开说。第一价格字段用DECIMAL而不是FLOAT或DOUBLE。浮点数在数据库里做等值比较和聚合时会出现精度漂移比如0.1 0.2不等于0.3这种经典问题。行情数据虽然对精度要求没有金融账务那么苛刻但用DECIMAL(10,3)能保证你查出来的价格和接口返回的完全一致省去很多对账的麻烦。第二vol和amount用DECIMAL(20,2)。成交量单位是手成交额单位是千元大盘股单日成交额可能到千万级别DECIMAL(20,2)足够覆盖不会溢出。第三主键顺序是(ts_code, trade_date)而不是反过来。这取决于你最常用的查询模式。如果你的策略是按股票查历史那ts_code在前更优如果经常按日期查全市场那trade_date在前更好。我两种都试过实际做多因子回测时按股票切片更频繁所以选了前者同时额外给trade_date建了一个普通索引来兼顾按日查询。2.2 字段类型和接口返回值的对齐陷阱tushare 返回的 DataFrame 里trade_date通常是字符串格式YYYYMMDD比如20240115。如果你直接往DATE类型的列里塞MySQL 会尝试隐式转换大多数情况能成功但遇到空值或者格式异常就会报错或者写入0000-00-00。稳妥的做法是在 Python 侧就把它转成标准格式。我习惯在写入前统一做一次类型清洗import pandas as pd def normalize_daily(df: pd.DataFrame) - pd.DataFrame: df df.copy() # 日期字符串转 datetime.date df[trade_date] pd.to_datetime(df[trade_date], format%Y%m%d).dt.date # 数值列强制转 float避免 object 类型导致写入报错 num_cols [open, high, low, close, pre_close, change, pct_chg, vol, amount] for col in num_cols: if col in df.columns: df[col] pd.to_numeric(df[col], errorscoerce) return dferrorscoerce的作用是把无法转换的值变成NaN而不是直接抛异常中断整个流程。这样即使某一行数据有问题也不会导致整批写入失败后续可以在数据库里用WHERE open IS NULL把异常行捞出来单独排查。注意pd.to_datetime转换后是datetime64加.dt.date才是 Python 的date对象。直接传datetime64给 MySQL 驱动有时会带出时间部分虽然DATE列会截断但显式转换更干净。3. 连接与写入方案选型pymysql、SQLAlchemy 还是原生批量3.1 三种写入路径的实际性能差异Python 往 MySQL 写数据常见有三条路pandas.DataFrame.to_sql、pymysql的executemany、以及拼装成单条大 SQL。我实测过同样 10 万行日线数据在本地 MySQL 8.0 上的表现结果差异很明显。方案10万行耗时优点缺点df.to_sql默认约 180 秒代码最短逐行插入极慢df.to_sqlmethodmulti约 25 秒无需手写 SQL受 max_allowed_packet 限制pymysql.executemany约 12 秒快、可控需手写 SQL 和参数拼装单条大 SQL约 8 秒最快SQL 长度受限易出错to_sql默认是逐行INSERT每行一次网络往返10 万行就是 10 万次往返慢是必然的。methodmulti会把多行合并成一条INSERT INTO ... VALUES (...), (...), ...速度快很多但默认每批只有几十行需要手动调chunksize。我最终选的是pymysql的executemany配合ON DUPLICATE KEY UPDATE原因是它既能批量、又能幂等而且 SQL 完全可控出问题好排查。to_sql虽然也能通过method自定义但封装层太厚遇到字段类型不匹配时错误信息很不直观。3.2 连接配置里那些容易忽略的参数连接 MySQL 时除了 host、port、user、password、database 这几个必填项有几个参数直接影响写入稳定性import pymysql conn pymysql.connect( host127.0.0.1, port3306, useryour_user, passwordyour_password, databasequant_data, charsetutf8mb4, autocommitFalse, local_infileFalse, connect_timeout10, read_timeout60, write_timeout120, )autocommitFalse是关键。批量写入时如果每行都自动提交InnoDB 会频繁刷盘速度慢几个数量级。关掉自动提交手动commit()让整个批次作为一个事务提交性能提升非常明显。但要注意事务太大也会有问题后面会讲批次大小怎么定。write_timeout设长一点因为大批量写入时单次网络交互可能持续几十秒默认值容易触发超时断开。charsetutf8mb4而不是utf8。MySQL 里的utf8其实是残缺的三字节实现存不了某些特殊字符。虽然股票代码和数字用不上但表里如果有股票名称之类的字段用utf8mb4更保险。3.3 幂等写入的 SQL 到底怎么写核心就是这一条INSERT INTO stock_daily (ts_code, trade_date, open, high, low, close, pre_close, change, pct_chg, vol, amount) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE open VALUES(open), high VALUES(high), low VALUES(low), close VALUES(close), pre_close VALUES(pre_close), change VALUES(change), pct_chg VALUES(pct_chg), vol VALUES(vol), amount VALUES(amount);VALUES(col)在 MySQL 8.0.20 之后被标记为 deprecated官方推荐用别名写法INSERT INTO stock_daily (ts_code, trade_date, open, high, low, close, pre_close, change, pct_chg, vol, amount) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) AS new ON DUPLICATE KEY UPDATE open new.open, high new.high, low new.low, close new.close, pre_close new.pre_close, change new.change, pct_chg new.pct_chg, vol new.vol, amount new.amount;两种写法在 8.0 上都能跑但如果你用的是较新版本建议用别名写法避免未来版本升级后报 warning。5.7 及以下只能用VALUES()写法这个要看你数据库版本决定。提示ON DUPLICATE KEY UPDATE触发的前提是存在唯一索引或主键冲突。如果你的表只有自增主键没有业务唯一键这条语句永远不会触发更新只会一直插入新行。这是最常见的踩坑点。4. 完整可跑的写入脚本与增量更新逻辑4.1 从接口到入库的完整链路把前面几块拼起来一个完整的写入函数大概长这样import pymysql import pandas as pd from typing import List INSERT_SQL INSERT INTO stock_daily (ts_code, trade_date, open, high, low, close, pre_close, change, pct_chg, vol, amount) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) AS new ON DUPLICATE KEY UPDATE open new.open, high new.high, low new.low, close new.close, pre_close new.pre_close, change new.change, pct_chg new.pct_chg, vol new.vol, amount new.amount COLUMNS [ts_code, trade_date, open, high, low, close, pre_close, change, pct_chg, vol, amount] def df_to_rows(df: pd.DataFrame) - List[tuple]: df df.copy() df[trade_date] pd.to_datetime(df[trade_date], format%Y%m%d).dt.date for col in COLUMNS: if col not in df.columns: df[col] None df df[COLUMNS] # 把 NaN 转成 None否则 pymysql 会写成字符串 nan df df.where(pd.notnull(df), None) return [tuple(row) for row in df.itertuples(indexFalse, nameNone)] def write_daily(df: pd.DataFrame, conn, batch_size: int 2000) - int: rows df_to_rows(df) total 0 with conn.cursor() as cur: for i in range(0, len(rows), batch_size): batch rows[i:i batch_size] cur.executemany(INSERT_SQL, batch) conn.commit() total len(batch) return total这里有几个我反复调试后才定下来的细节。df.where(pd.notnull(df), None)这一步非常关键。pandas 里的缺失值是NaN而NaN是 float 类型。如果你直接把含NaN的 tuple 传给 pymysql它会把NaN当成字符串nan写进去DECIMAL列收到nan会报错或者写成 0。转成None之后pymysql 才会正确地写成 SQLNULL。itertuples比iterrows快很多因为iterrows每行都会构造一个 Series 对象开销大。itertuples返回的是 namedtuple转 tuple 后直接可用。批次大小batch_size2000是我在本地和云数据库上都测过的折中值。太小则事务提交次数多太大会撞上max_allowed_packet限制。MySQL 默认max_allowed_packet是 4MB 或 64MB版本不同2000 行日线数据大概几百 KB留足余量。如果你字段特别多可以降到 1000。4.2 增量更新怎么知道该拉哪一段全量拉取一次之后后续每天只需要拉最新一天的数据。但补数据的时候比如发现某段时间有缺失就需要按区间拉。核心是维护一个已入库的最大日期。def get_last_date(conn, ts_code: str): sql SELECT MAX(trade_date) FROM stock_daily WHERE ts_code %s with conn.cursor() as cur: cur.execute(sql, (ts_code,)) result cur.fetchone() return result[0] if result and result[0] else None拿到last_date之后接口调用时把start_date设为last_date 1 天end_date设为今天就只拉增量部分。这样每天的定时任务只需要处理一天的数据几秒钟就跑完。但这里有个坑停牌和退市。如果某只股票停牌了几个月MAX(trade_date)会停在停牌前那天你每次拉增量都会从那天开始反复拉同一段空数据。解决办法是记录最后检查日期而不是最后数据日期或者干脆按固定窗口拉最近 N 天靠幂等写入去重。我倾向于后者简单可靠每天拉最近 30 天数据反正ON DUPLICATE KEY UPDATE会覆盖多拉一点不影响正确性还能自动修复历史数据的小错误。4.3 多进程写入时的锁竞争问题如果你用多进程并行拉多只股票的数据每个进程各开一个连接往同一张表写会遇到行锁竞争。InnoDB 在ON DUPLICATE KEY UPDATE时会对冲突行加锁如果两个进程同时写同一只股票同一天的数据后到的会等待。实际影响不大因为不同股票的数据行不冲突锁粒度是行级。但要注意两点一是每个进程必须用独立的连接不能共享pymysql 的连接不是线程安全的二是commit的粒度不要太细否则锁持有时间虽然短但提交次数多整体吞吐反而下降。我一般的做法是按股票代码分片每个进程负责一批股票进程内串行写入进程间互不干扰。这样既避免了锁竞争又能利用多核。5. 那些让我熬夜排查的坑5.1 日期格式不一致导致的静默错误有一次我发现数据库里某段时间的数据查不出来明明接口返回了。排查半天发现那批数据的trade_date被写成了2024-01-15格式的字符串而其他数据是2024-01-15的 DATE 类型。原因是那批数据我是从另一个 CSV 导入的CSV 里的日期格式和接口返回的不一样我的清洗函数只处理了%Y%m%d格式遇到%Y-%m-%d就to_datetime失败errors默认是raise但那次我用了coerce结果全变成NaT写入时变成NULL。这个坑的教训是清洗函数要对多种日期格式做兼容或者至少在转换后检查空值比例异常时报警而不是静默通过。def parse_date(s): for fmt in (%Y%m%d, %Y-%m-%d, %Y/%m/%d): try: return pd.to_datetime(s, formatfmt).date() except (ValueError, TypeError): continue return None5.2 事务太大导致连接断开早期我把batch_size设成 50000想着一次提交省事。结果跑大盘股全历史数据时写到一半报MySQL server has gone away。原因是单个事务太大超过了max_allowed_packet或者服务端wait_timeout到了把连接掐了。后来改成 2000 一批每批提交问题消失。这里的原则是单批数据量控制在几百 KB 到 1MB 之间既能减少提交次数又不会撑爆包大小。你可以用len(str(batch))粗略估算一下。5.3 字段顺序和 SQL 占位符对不上executemany的第二个参数是 tuple 列表tuple 里的元素顺序必须和 SQL 里%s的顺序严格一致。我因为调整了COLUMNS列表的顺序忘了同步改 SQL结果价格和成交量写串了close列里出现了成交量。这种错误不会报错只会静默写错数据非常隐蔽。我的应对办法是SQL 里的列名和 Python 里的 COLUMNS 列表用同一个常量生成避免手工维护两份。COLS [ts_code, trade_date, open, high, low, close, pre_close, change, pct_chg, vol, amount] PLACEHOLDERS , .join([%s] * len(COLS)) UPDATE_CLAUSE , .join([f{c} new.{c} for c in COLS if c not in (ts_code, trade_date)]) INSERT_SQL f INSERT INTO stock_daily ({, .join(COLS)}) VALUES ({PLACEHOLDERS}) AS new ON DUPLICATE KEY UPDATE {UPDATE_CLAUSE} 这样只要改COLSSQL 自动跟着变不会出现列和值错位。5.4 时区问题让今天变成昨天服务器时区和数据库时区不一致时CURRENT_TIMESTAMP写入的update_time可能和你预期差几个小时。更麻烦的是如果你用 Python 的datetime.now()生成查询日期而服务器是 UTC本地是东八区凌晨跑任务时今天会算成前一天导致拉错数据区间。我的做法是所有涉及今天的逻辑统一用数据库的CURDATE()或者显式指定时区不在 Python 侧用裸的datetime.now()。如果一定要在 Python 侧算用datetime.now(ZoneInfo(Asia/Shanghai))明确时区。6. 查询侧优化让回测读数据不再等6.1 索引不是越多越好前面建表时我加了PRIMARY KEY (ts_code, trade_date)和KEY idx_trade_date (trade_date)。这两个索引覆盖了绝大多数查询场景按股票查历史走主键按日期查全市场走二级索引。但索引会拖慢写入速度因为每次插入都要更新索引树。如果你的表只写不读那索引越少越好。但量化场景一定是读写混合所以要在写入速度和查询速度之间找平衡。我的经验是主键必须建按日期查询的二级索引看需求。如果你只做按股票的回测idx_trade_date可以省掉写入能快 10% 左右。6.2 用覆盖索引避免回表如果某个查询只需要ts_code、trade_date、close三列而主键是(ts_code, trade_date)那么 MySQL 从主键索引里就能拿到前两列但close需要回表到聚簇索引取。如果这个查询非常频繁可以考虑建一个覆盖索引ALTER TABLE stock_daily ADD INDEX idx_code_date_close (ts_code, trade_date, close);这样查询直接在索引里完成不用回表。代价是索引体积变大写入更慢。是否值得取决于你的查询模式。我一般先用EXPLAIN看执行计划确认有回表且查询频繁才加覆盖索引。6.3 分批读取避免内存爆掉回测时如果把全市场十年的日线一次性SELECT *读进 DataFrame内存可能直接爆掉。几百万行乘以十几列轻松上 GB。稳妥的做法是按股票或按年份分批读。def iter_daily(conn, ts_code: str, start: str, end: str, chunk_days: int 250): cur_date pd.to_datetime(start) end_date pd.to_datetime(end) while cur_date end_date: chunk_end min(cur_date pd.Timedelta(dayschunk_days), end_date) sql SELECT * FROM stock_daily WHERE ts_code %s AND trade_date BETWEEN %s AND %s ORDER BY trade_date yield pd.read_sql(sql, conn, params(ts_code, cur_date.date(), chunk_end.date())) cur_date chunk_end pd.Timedelta(days1)用生成器逐批产出调用方处理完一批再取下一批内存占用恒定。chunk_days250大约是一年的交易日数量这个粒度对大多数策略够用。7. 定时任务与数据校验的工程化收尾7.1 每天收盘后自动跑一次的骨架把上面的逻辑串成一个定时任务用schedule或者系统的 cron 都行。核心是拉数据、清洗、写入、记录日志、异常重试。import logging from datetime import datetime, timedelta logging.basicConfig(levellogging.INFO, format%(asctime)s %(levelname)s %(message)s) def daily_job(): conn pymysql.connect(**DB_CONFIG) try: end datetime.now().date() start end - timedelta(days30) df fetch_from_api(start.strftime(%Y%m%d), end.strftime(%Y%m%d)) if df is None or df.empty: logging.warning(接口返回空数据跳过) return n write_daily(df, conn) logging.info(f写入完成共 {n} 行) except Exception as e: logging.error(f任务失败: {e}, exc_infoTrue) finally: conn.close()拉最近 30 天而不是只拉一天是为了自动修复可能的历史缺失。反正幂等写入多拉不会出错。7.2 写入后的数据校验不能省写完不校验等于没写。我一般会做三个检查行数是否合理、关键字段是否有空值、日期是否连续。def validate(conn, trade_date: str): with conn.cursor() as cur: cur.execute(SELECT COUNT(*) FROM stock_daily WHERE trade_date %s, (trade_date,)) cnt cur.fetchone()[0] if cnt 1000: logging.warning(f{trade_date} 仅 {cnt} 行可能不完整) cur.execute(SELECT COUNT(*) FROM stock_daily WHERE trade_date %s AND close IS NULL, (trade_date,)) null_cnt cur.fetchone()[0] if null_cnt 0: logging.warning(f{trade_date} 有 {null_cnt} 行 close 为空)正常交易日全市场应该有几千行如果只有几百行说明接口返回不完整或者写入中断了。close为空更严重说明数据有问题。这些检查跑一遍只要几毫秒但能帮你提前发现大问题。7.3 关于重试和断点续传网络请求偶尔失败是常态尤其是拉全市场数据时。我的做法是按股票代码循环拉取每只股票单独 try-except失败的记到一个列表里全部跑完后对失败列表重试一次。这样单只股票失败不会中断整个任务。failed [] for code in all_codes: try: df fetch_one(code, start, end) write_daily(df, conn) except Exception as e: logging.error(f{code} 失败: {e}) failed.append(code) for code in failed: try: df fetch_one(code, start, end) write_daily(df, conn) except Exception as e: logging.error(f{code} 重试仍失败: {e})重试一次就够了连续两次失败通常是接口权限或者代码本身有问题再重试也是浪费时间不如记下来人工排查。这套流程我跑了两年多从最初的 CSV 到现在的 MySQL最大的体会是数据管道的稳定性不取决于最复杂的部分而取决于最不起眼的细节——日期格式、空值处理、批次大小、事务提交。这些地方任何一个出问题都会让你在回测结果对不上的时候排查到怀疑人生。把幂等写入和校验做扎实后面无论加多少数据源、跑多少策略底层都是稳的。