1. 为什么要把 tushare 数据落到 MySQL做过量化或者金融数据分析的朋友大概率都接触过 tushare这个库把国内股票、指数、基金、期货等行情数据封装得相当顺手几行代码就能把日线、复权因子、财务指标拉下来。但真正把项目往生产环境推的时候问题就来了每次跑策略都要重新调接口网络一抖动整个流程就断接口还有频率限制历史数据反复拉取既慢又浪费积分。这时候把数据落到本地数据库就成了绕不开的一步。我最早是用 CSV 存一个股票一个文件几百个文件散在目录里做截面分析的时候光是读文件就要等半天更别说 join 和聚合了。后来换成 SQLite单机跑跑还行一旦多进程并发写入就开始锁表而且数据量上到千万行级别查询明显变慢。最后切到 MySQL才算把写入、查询、并发这几件事都理顺了。这篇就围绕“python tushare 直接写入 mysql 数据库”这个主题把整套流程从环境准备、表结构设计、批量写入、增量更新到常见报错排查完整讲一遍都是我实际项目里踩过坑之后沉淀下来的做法。这篇文章适合三类人看一是刚接触 tushare、想把数据存下来的新手二是已经在用 CSV 或 SQLite、想迁移到 MySQL 的进阶用户三是需要搭建稳定数据管道、做定时增量更新的工程向开发者。不管你是哪种下面的内容都能直接抄作业代码改改连接参数就能跑。2. 整体方案设计与技术选型思路2.1 为什么是 MySQL 而不是其他存储先把选型这件事说清楚不然后面写代码心里没底。tushare 返回的数据本质是结构化的二维表字段固定、类型明确、行数大这种数据天然适合关系型数据库。对比几种常见方案存储方案优点缺点适用场景CSV 文件零依赖、易查看查询慢、无索引、并发差临时导出、一次性分析SQLite单文件、免安装写锁严重、并发弱单机小数据量MySQL索引强、并发好、生态成熟需部署、配置略复杂生产级数据管道时序数据库写入极快学习成本高、SQL 生态弱超高频行情选 MySQL 的核心理由有三个。第一tushare 的日线数据字段就那么十几个用关系表存最自然查询的时候 where、group by、join 都能用上索引。第二MySQL 的INSERT ... ON DUPLICATE KEY UPDATE语法天然适合做增量更新重复拉取不会产生脏数据。第三生态成熟pandas 的to_sql、SQLAlchemy、PyMySQL 都能无缝对接后续接 BI 工具或者做数据服务都方便。2.2 写入方式的选择to_sql 还是 executemany很多人第一反应是用 pandas 的df.to_sql()一行代码搞定。但我实测下来to_sql在数据量大时性能很差因为它默认逐行拼 SQL几万行数据能跑几分钟。真正高效的做法是用executemany批量提交或者用to_sql配合methodmulti参数。下面这张表是我在同样 5 万行日线数据下测出来的对比写入方式耗时说明to_sql 默认约 180 秒逐行插入最慢to_sql methodmulti约 25 秒批量拼 SQLexecutemany约 8 秒预编译批量提交LOAD DATA INFILE约 2 秒需文件中转最快所以我的方案是小批量用executemany大批量历史回补用LOAD DATA或者分块executemany。这篇主要讲executemany这条路线因为它最通用不依赖文件系统权限。2.3 整体数据流设计整个管道我拆成四层拉取层、清洗层、写入层、调度层。拉取层负责调 tushare 接口拿原始 DataFrame清洗层做字段重命名、类型转换、空值处理写入层负责建表、批量插入、增量更新调度层用定时任务每天收盘后跑一次。这样分层的好处是每层职责单一出问题好定位比如数据对不上先看清洗层写入慢先看写入层。提示不要把所有逻辑塞进一个函数里我早期就是这么干的后来加个字段要改十几处维护成本极高。3. 环境准备与数据库表结构设计3.1 依赖安装与连接配置先把环境搭起来。需要的库不多tushare拉数据pymysql做驱动pandas处理数据sqlalchemy可选但建议装上方便管理连接。pip install tushare pymysql pandas sqlalchemy连接配置我习惯单独放一个 config 文件不硬编码在业务代码里。下面是一个典型的配置# config.py DB_CONFIG { host: 127.0.0.1, port: 3306, user: your_user, password: your_password, database: stock_data, charset: utf8mb4 } TS_TOKEN 你的tushare_token这里有个细节要注意charset一定要用utf8mb4不要用utf8。MySQL 的utf8其实是阉割版只支持三字节遇到某些特殊字符会报错utf8mb4才是完整的四字节 UTF-8。虽然股票数据里中文不多但股票名称偶尔会有生僻字用utf8mb4一劳永逸。3.2 日线表结构设计表结构设计是整套方案的地基设计不好后面查询和更新都难受。以日线行情为例tushare 的daily接口返回字段包括 ts_code、trade_date、open、high、low、close、pre_close、change、pct_chg、vol、amount。我的建表语句是这样的CREATE TABLE IF NOT EXISTS stock_daily ( id BIGINT AUTO_INCREMENT PRIMARY KEY, ts_code VARCHAR(12) NOT NULL COMMENT 股票代码, trade_date DATE NOT NULL COMMENT 交易日期, open DECIMAL(10,3) COMMENT 开盘价, high DECIMAL(10,3) COMMENT 最高价, low DECIMAL(10,3) COMMENT 最低价, close DECIMAL(10,3) COMMENT 收盘价, pre_close DECIMAL(10,3) COMMENT 昨收价, change_amt DECIMAL(10,3) COMMENT 涨跌额, pct_chg DECIMAL(10,4) COMMENT 涨跌幅, vol DECIMAL(20,2) COMMENT 成交量(手), amount DECIMAL(20,4) COMMENT 成交额(千元), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_code_date (ts_code, trade_date), KEY idx_trade_date (trade_date) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT股票日线行情;几个设计要点解释一下。第一价格字段用DECIMAL而不是FLOAT因为浮点数有精度问题做金额计算时会出现0.10.20.30000000000000004这种鬼东西DECIMAL是精确存储。第二ts_code和trade_date建联合唯一索引这是增量更新的关键配合ON DUPLICATE KEY UPDATE就能实现“有则更新、无则插入”。第三单独给trade_date建索引因为做截面分析时经常按日期筛选全市场数据。第四change是 MySQL 保留字字段名改成change_amt避免转义麻烦。3.3 字段类型映射的坑tushare 返回的 DataFrame 里trade_date是字符串格式20240101直接写进 DATE 字段会报错或者变成 0000-00-00。必须转成2024-01-01格式。另外vol和amount数值可能很大用DECIMAL(20,2)足够别用 INT会溢出。下面是我常用的类型转换函数def clean_daily(df): df df.copy() df[trade_date] pd.to_datetime(df[trade_date], format%Y%m%d).dt.date df df.rename(columns{change: change_amt}) numeric_cols [open,high,low,close,pre_close,change_amt,pct_chg,vol,amount] for col in numeric_cols: df[col] pd.to_numeric(df[col], errorscoerce) return dferrorscoerce这个参数很关键遇到无法转换的值会变成 NaN 而不是直接抛异常保证整批数据不会因为一行脏数据全军覆没。4. 核心写入逻辑与增量更新实现4.1 批量写入的核心代码写入这块我用pymysql的executemany配合ON DUPLICATE KEY UPDATE实现幂等写入。所谓幂等就是同一批数据跑一次和跑十次结果一样这对定时任务特别重要万一某天任务重跑不会产生重复数据。import pymysql import pandas as pd from config import DB_CONFIG def get_conn(): return pymysql.connect(**DB_CONFIG) def save_daily(df): if df.empty: return 0 cols [ts_code,trade_date,open,high,low,close, pre_close,change_amt,pct_chg,vol,amount] df df[cols] sql INSERT INTO stock_daily (ts_code,trade_date,open,high,low,close,pre_close,change_amt,pct_chg,vol,amount) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE openVALUES(open), highVALUES(high), lowVALUES(low), closeVALUES(close), pre_closeVALUES(pre_close), change_amtVALUES(change_amt), pct_chgVALUES(pct_chg), volVALUES(vol), amountVALUES(amount) data [tuple(x) for x in df.itertuples(indexFalse, nameNone)] conn get_conn() try: with conn.cursor() as cur: cur.executemany(sql, data) conn.commit() return len(data) except Exception as e: conn.rollback() raise e finally: conn.close()这段代码有几个地方值得说。itertuples(indexFalse, nameNone)比iterrows()快很多因为它返回的是普通元组而不是 Series 对象。VALUES(字段名)是 MySQL 的语法表示“插入时那个位置的值”在更新时引用它就能把新值覆盖旧值。conn.rollback()放在异常分支里保证出错时不会留下半截数据。4.2 分块写入避免内存和超时问题一次性写入几十万行会出两个问题一是内存里拼出来的 data 列表太大二是单条 SQL 包太大可能超过max_allowed_packet限制。解决办法是分块每块 5000 到 10000 行比较合适。def save_daily_chunked(df, chunk_size5000): total 0 for i in range(0, len(df), chunk_size): chunk df.iloc[i:ichunk_size] total save_daily(chunk) return totalmax_allowed_packet默认是 4MB 或 16MB取决于 MySQL 版本。5000 行日线数据大概几百 KB留足余量。如果你要写财务数据这种字段特别多的表chunk_size 要相应调小。4.3 增量更新的日期判断逻辑每天定时任务只需要拉最新一天的数据不需要全量重拉。判断逻辑是先查库里已有的最大日期然后从那天之后开始拉。def get_last_date(ts_codeNone): conn get_conn() try: with conn.cursor() as cur: if ts_code: cur.execute(SELECT MAX(trade_date) FROM stock_daily WHERE ts_code%s, (ts_code,)) else: cur.execute(SELECT MAX(trade_date) FROM stock_daily) result cur.fetchone()[0] return result finally: conn.close()拿到 last_date 之后调 tushare 时传start_date参数即可。这里有个坑如果库里是空的MAX返回 None要单独处理否则传给 tushare 会报错。我一般写成start_date last_date.strftime(%Y%m%d) if last_date else 20100101。4.4 用 SQLAlchemy 管理连接池上面每次写入都新建连接在低频场景没问题但如果是高频批量任务频繁建连开销不小。用 SQLAlchemy 的 engine 可以复用连接from sqlalchemy import create_engine from config import DB_CONFIG engine create_engine( fmysqlpymysql://{DB_CONFIG[user]}:{DB_CONFIG[password]} f{DB_CONFIG[host]}:{DB_CONFIG[port]}/{DB_CONFIG[database]} f?charsetutf8mb4, pool_size5, max_overflow10, pool_recycle3600 )pool_recycle3600很重要MySQL 默认 8 小时会断开空闲连接如果不设置回收时间第二天早上任务跑的时候会拿到一个已经失效的连接报Lost connection错误。设成 3600 秒1 小时就能避开这个问题。5. 完整实操流程与调度落地5.1 从拉取到入库的完整脚本把前面的模块串起来一个完整的日线更新脚本长这样import tushare as ts import pandas as pd from datetime import datetime, timedelta from config import TS_TOKEN from db_utils import save_daily_chunked, get_last_date ts.set_token(TS_TOKEN) pro ts.pro_api() def update_daily_all(): last_date get_last_date() if last_date: start (last_date timedelta(days1)).strftime(%Y%m%d) else: start 20100101 end datetime.now().strftime(%Y%m%d) if start end: print(数据已是最新) return df pro.daily(start_datestart, end_dateend) if df.empty: print(无新数据) return df clean_daily(df) n save_daily_chunked(df) print(f写入 {n} 行区间 {start} ~ {end}) if __name__ __main__: update_daily_all()这个脚本每天收盘后跑一次就行。注意start是 last_date 加一天避免重复拉取已有数据虽然ON DUPLICATE KEY UPDATE能兜底但少拉点数据省积分。5.2 定时任务的两种落地方式调度这块我用过两种方式。Linux 环境下直接用 crontab简单可靠# 每个交易日 17:30 执行 30 17 * * 1-5 cd /path/to/project /usr/bin/python3 update_daily.py logs/update.log 211-5表示周一到周五 logs/update.log 21把标准输出和错误都重定向到日志文件方便排查。另一种是用 Python 的schedule库或者 APScheduler适合跨平台或者需要复杂调度逻辑的场景。我一般生产环境用 crontab开发调试用 APScheduler。5.3 首次全量回补的注意事项第一次建库需要回补历史数据这时候不能按增量逻辑走。全量回补要注意三点一是按年或者按季度分段拉避免单次请求数据量过大被接口限制二是加个 sleep 控制频率tushare 有每分钟调用次数限制三是做好断点续传万一中途失败不用从头再来。import time def backfill_by_year(start_year2010, end_year2024): for year in range(start_year, end_year 1): start f{year}0101 end f{year}1231 df pro.daily(start_datestart, end_dateend) if not df.empty: df clean_daily(df) save_daily_chunked(df) print(f{year} 年写入完成共 {len(df)} 行) time.sleep(1)time.sleep(1)是给接口留喘息时间具体间隔看你账号的积分等级积分高的可以短一点。断点续传的思路是每次写入前查一下该年份是否已有数据有就跳过。6. 常见报错与排查技巧实录6.1 高频报错速查表实际跑下来遇到的报错就那么几类整理成表方便对照报错信息原因解决办法Lost connection to MySQL连接空闲超时设置 pool_recyclePacket too large单次写入数据过大减小 chunk_sizeIncorrect date value日期格式不对转成 date 对象Duplicate entry唯一索引冲突用 ON DUPLICATE KEYAccess denied账号权限不足授权 GRANT ALL1366 Incorrect string字符集不匹配改 utf8mb46.2 日期格式问题的深度排查日期问题是最容易踩的坑。tushare 返回的trade_date是20240101这种字符串如果你直接塞进 DATE 字段MySQL 在严格模式下会报Incorrect date value: 20240101在非严格模式下会存成0000-00-00后者更坑因为不报错但数据是错的。排查方法很简单写入前打印一下df.dtypes和df.head()确认日期列已经是 date 类型。注意MySQL 5.7 之后默认开启严格模式日期格式不对会直接报错这其实是好事能帮你及早发现问题。6.3 写入速度慢的优化思路如果发现写入特别慢按这个顺序排查先看是不是用了to_sql默认模式换成executemany再看 chunk_size 是不是太小太小会导致事务提交次数过多然后看表上索引是不是太多每个索引在插入时都要维护索引越多写入越慢最后看 MySQL 的innodb_flush_log_at_trx_commit参数设成 2 能显著提升写入速度代价是极端情况下可能丢一秒数据对行情数据来说可以接受。6.4 数据对不上的排查方法有时候发现库里数据和 tushare 页面对不上先别慌。第一步查是不是复权问题tushare 的daily是不复权数据pro_bar才有复权选项。第二步查是不是日期范围问题start_date和end_date都是闭区间。第三步查是不是有重复数据用SELECT ts_code, trade_date, COUNT(*) FROM stock_daily GROUP BY ts_code, trade_date HAVING COUNT(*) 1查一下正常情况应该返回空。7. 我踩过的坑和几条实用经验第一个坑是字段名冲突。change是 MySQL 保留字我一开始没注意建表直接用了change结果每次查询都要加反引号后来统一改成change_amt才清爽。类似的还有desc、order、key这些建表前最好查一下保留字列表。第二个坑是时区问题。created_at用CURRENT_TIMESTAMP默认取的是服务器时区如果服务器是 UTC 而你在东八区看到的时间会差 8 小时。解决办法是在连接参数里加time_zone08:00或者干脆用 Python 生成时间戳写入。第三个坑是批量写入时 NaN 的处理。pandas 的 NaN 直接传给 pymysql 会变成字符串nan写进 DECIMAL 字段会报错。我的做法是在清洗阶段用df.where(pd.notnull(df), None)把 NaN 转成 Nonepymysql 会把 None 正确转成 SQL 的 NULL。第四个经验是关于索引的。我一开始给每个字段都建了索引觉得查询快结果写入慢得离谱。后来精简到只保留联合唯一索引和 trade_date 索引写入速度提升了一倍多。索引不是越多越好按实际查询需求来。第五个经验是日志。定时任务一定要打日志记录每次拉取的日期范围、写入行数、耗时。我用的格式是[2024-01-15 17:30:02] start20240115 end20240115 rows5234 cost3.2s出问题的时候一眼就能看出是哪天哪步出的错。这套方案我从最早的 CSV 一路迭代到现在的 MySQL 管道跑了两年多每天自动更新基本没出过大问题。核心就三点表结构设计好、写入用批量加幂等、调度加日志。把这三点做到位剩下的就是按需扩展字段和表了。