企业官网建设流程全解析

1. 为什么要把 Tushare 数据落到 MySQL

做量化研究或者数据分析的朋友，大概率都绕不开 Tushare 这个数据源。它把 A 股、基金、期货、宏观经济等数据封装成了统一的接口，调用起来确实方便。但用着用着就会碰到一个很现实的问题：每次跑策略或者做回测，都要重新调一遍接口，网络稍微抖一下，整个流程就卡住了。更麻烦的是，Tushare 的积分接口有频率限制，短时间内大量拉取会被限流，甚至影响账号的正常使用。

所以，把 Tushare 的数据一次性拉下来，落到本地的 MySQL 数据库里，后续所有分析都从本地读，这才是长久之计。这个项目标题说的就是这件事：用 Python 把 Tushare 返回的数据直接写入 MySQL。听起来简单，但实际操作里坑不少——字段类型怎么定、主键怎么设、重复数据怎么处理、批量插入怎么提速，每一个环节都值得展开聊。

这篇文章适合两类人：一类是刚开始接触量化数据管理的新手，想找一个能直接抄作业的方案；另一类是有一定经验、但每次写入库代码都要重新查文档的老手，希望能拿到一套稳定、可复用的模板。我会从整体设计思路讲起，把核心细节、实操步骤、常见问题都拆开揉碎，尽量让不同基础的人都能看懂、能用上。

2. 整体方案设计与技术选型

2.1 为什么选 MySQL 而不是 SQLite 或 CSV

先说存储选型。很多人第一反应是用 CSV 或者 SQLite，毕竟零配置、上手快。但数据量一上来，问题就暴露了。CSV 没有索引，每次查询都要全表扫描，几百万行数据查一个字段能等好几分钟；SQLite 虽然支持索引，但并发写入能力弱，多个进程同时写容易锁库。MySQL 在这方面优势明显：支持多用户并发、有成熟的索引机制、数据类型丰富、还能配合各种 BI 工具做可视化。

另一个考虑是数据更新。Tushare 的日线数据每天都会新增，如果每次都是全量覆盖，既浪费时间又浪费存储。MySQL 支持INSERT ... ON DUPLICATE KEY UPDATE这种语法，可以做到“有则更新、无则插入”，非常适合增量更新场景。CSV 和 SQLite 要实现同样的逻辑，得自己写一堆判断代码，维护成本高。

当然，MySQL 也不是没有代价。你需要装服务、建库建表、配置用户权限，前期准备工作比 SQLite 多。但考虑到量化数据动辄几十 GB、需要长期维护，这点前期投入完全值得。

2.2 为什么用 pandas 做中转而不是直接拼 SQL

Tushare 的接口返回的是 pandas DataFrame，这是最自然的数据结构。有人可能会想，那我直接遍历 DataFrame 拼 SQL 语句不就行了？理论上可以，但效率极低。逐行拼接字符串再执行，一万行数据可能要跑好几分钟，而且容易因为特殊字符导致 SQL 报错。

更合理的做法是用 pandas 的to_sql方法，配合 SQLAlchemy 引擎，让 pandas 帮你处理类型映射和批量插入。to_sql底层用的是 executemany，比逐行插入快一个数量级。如果数据量特别大，还可以调chunksize参数分批写入，避免一次性占用太多内存。

不过to_sql有个小毛病：默认的if_exists='fail'在表已存在时会直接报错，if_exists='replace'又会把整张表删掉重建，都不太适合增量更新。所以实际项目里，我一般会自己封装一个写入函数，结合ON DUPLICATE KEY UPDATE来实现真正的 upsert。

2.3 整体架构：从接口到入库的完整链路

整个流程可以拆成四步：第一步，用 Tushare 的 token 初始化接口，拉取指定股票、指定时间范围的数据；第二步，对返回的 DataFrame 做清洗，包括列名统一、类型转换、空值处理；第三步，通过 SQLAlchemy 建立到 MySQL 的连接，把 DataFrame 写入目标表；第四步，记录写入日志，方便后续排查和增量更新。

这个链路里，第二步和第三步是重点。清洗不到位，入库后查询会各种报错；写入方式选不对，速度慢得让人抓狂。后面我会分别展开讲。

3. 环境准备与核心依赖配置

3.1 Python 环境与依赖包清单

我习惯用 conda 建一个独立环境，避免和系统 Python 冲突。Python 版本建议 3.8 以上，太老的版本对 pandas 和 SQLAlchemy 的新特性支持不好。核心依赖就四个：

• tushare：数据源接口
• pandas：数据处理
• sqlalchemy：数据库连接和 ORM
• pymysql：MySQL 驱动，SQLAlchemy 用它来连 MySQL

安装命令很简单：

pip install tushare pandas sqlalchemy pymysql

如果你用的是 conda，也可以conda install对应的包，但 tushare 在 conda 源里不一定有最新版，建议还是用 pip 装。

注意：pymysql 是纯 Python 实现的驱动，安装方便但性能一般。如果对写入速度要求极高，可以换成mysqlclient，它是 C 扩展，速度快不少，但安装时需要系统有 MySQL 开发库，Windows 上可能要多折腾一会儿。

3.2 MySQL 建库建表的关键决策

建表这一步，很多人喜欢用to_sql自动建表，省事是省事，但自动生成的字段类型往往不是最优的。比如 Tushare 返回的trade_date是字符串格式的YYYYMMDD，pandas 会把它识别成 object，to_sql就建一个TEXT类型的列。TEXT 类型不能直接做范围查询，也没法建有效索引，后面查“某段时间的数据”会非常慢。

我的做法是手动建表，把字段类型定死。以日线行情为例，核心字段和推荐类型如下：

主键设成(ts_code, trade_date)联合主键，这样同一只股票同一天的数据只会有一条，天然去重。索引方面，除了主键自带的聚簇索引，我还会在trade_date上单独建一个索引，方便按日期范围查询。

3.3 连接字符串与字符集设置

SQLAlchemy 的连接字符串格式是：

from sqlalchemy import create_engine engine = create_engine( 'mysql+pymysql://user:password@host:3306/dbname?charset=utf8mb4', pool_recycle=3600, pool_pre_ping=True )

这里有几个细节值得说。charset=utf8mb4一定要加，否则遇到中文列名或者特殊字符会乱码。pool_recycle=3600是让连接池每小时回收一次连接，避免 MySQL 默认的 8 小时空闲断开导致“MySQL server has gone away”。pool_pre_ping=True是在每次取连接前先 ping 一下，确保连接可用，这两个参数配合使用，基本能杜绝连接失效的问题。

实操心得：如果你在本地跑，host 写127.0.0.1比localhost更稳。因为localhost在某些系统上会走 socket 连接，而127.0.0.1强制走 TCP，行为更一致。

4. 数据拉取与清洗的实操要点

4.1 Tushare 接口调用与 token 管理

Tushare 的接口调用需要 token，这个 token 不要硬编码在代码里，更不要提交到代码仓库。我一般用环境变量或者单独的配置文件来管理：

import os import tushare as ts token = os.getenv('TUSHARE_TOKEN') ts.set_token(token) pro = ts.pro_api()

拉日线数据的典型调用是：

df = pro.daily(ts_code='000001.SZ', start_date='20230101', end_date='20231231')

这里要注意，Tushare 的日期参数是字符串格式YYYYMMDD，不是YYYY-MM-DD。如果你从其他系统拿到的日期是带横杠的，记得先转换。

另外，单次调用返回的行数有限制，一般一次最多返回 5000 行左右。如果要拉多只股票或者很长时间的数据，需要写循环分批拉取。循环里记得加time.sleep，避免触发频率限制。

4.2 DataFrame 清洗：列名、类型、空值

Tushare 返回的 DataFrame 列名都是小写英文，比较规范，但类型需要手动处理。trade_date是字符串，要转成 datetime；价格字段是 float，入库前最好转成 Decimal 或者保留两位小数；vol和amount可能是 float，要转成 int。

清洗代码大概长这样：

import pandas as pd df['trade_date'] = pd.to_datetime(df['trade_date'], format='%Y%m%d') df['open'] = df['open'].round(2) df['high'] = df['high'].round(2) df['low'] = df['low'].round(2) df['close'] = df['close'].round(2) df['vol'] = df['vol'].fillna(0).astype('int64') df['amount'] = df['amount'].round(2)

空值处理也很关键。有些股票在某些日期可能没有成交，vol和amount会是 NaN。如果直接入库，MySQL 里会出现 NULL，后续计算容易出错。我一般用 0 填充，或者根据业务逻辑决定是丢弃还是保留。

注意：pd.to_datetime如果不指定 format，pandas 会自己推断，速度慢而且可能推断错。指定format='%Y%m%d'既快又准。

4.3 批量拉取多只股票的策略

如果要拉全市场几千只股票的数据，一只一只拉效率太低。Tushare 的daily接口支持按交易日拉取，也就是指定trade_date而不指定ts_code，一次返回当天所有股票的数据。这种方式效率高很多，一天的数据一次就能拉完。

我的做法是：先拉一个交易日历，然后遍历每个交易日，拉取当天全市场数据，清洗后入库。这样几千只股票几年的数据，几个小时就能全部落库。当然，前提是你的 Tushare 积分够高，能支持这种批量调用。

5. 写入 MySQL 的核心实现

5.1 用 to_sql 快速写入及其局限

最简单的写入方式就是df.to_sql：

df.to_sql('daily', engine, if_exists='append', index=False)

这行代码能把 DataFrame 直接写进 MySQL，index=False表示不把 pandas 的索引写进去。速度方面，一万行数据大概几秒钟，可以接受。

但它的局限也很明显。if_exists='append'是纯追加，如果表里已经有相同主键的数据，会直接报主键冲突错误。if_exists='replace'会先 drop 表再重建，历史数据全丢。所以to_sql只适合首次全量导入，不适合日常增量更新。

5.2 实现 upsert：ON DUPLICATE KEY UPDATE

真正的增量更新，得用 MySQL 的INSERT ... ON DUPLICATE KEY UPDATE语法。这个语法的作用是：尝试插入，如果主键冲突，就执行 UPDATE 更新其他字段。

用 pandas 实现这个逻辑，需要自己拼 SQL。思路是：把 DataFrame 转成字典列表，然后构造带占位符的 SQL 语句，用executemany批量执行。核心代码如下：

from sqlalchemy import text def upsert_dataframe(df, table_name, engine): columns = list(df.columns) cols_str = ', '.join(columns) placeholders = ', '.join([f':{c}' for c in columns]) update_str = ', '.join([f'{c}=VALUES({c})' for c in columns if c not in ('ts_code', 'trade_date')]) sql = text(f""" INSERT INTO {table_name} ({cols_str}) VALUES ({placeholders}) ON DUPLICATE KEY UPDATE {update_str} """) records = df.to_dict('records') with engine.begin() as conn: conn.execute(sql, records)

这段代码里，update_str排除了主键字段，因为主键不需要更新。engine.begin()会自动开启事务，执行完自动提交，出错自动回滚，比手动 commit 安全。

5.3 批量写入的性能优化

上面的写法已经比逐行插入快很多，但如果数据量特别大，还可以进一步优化。几个方向：

第一，调整chunksize。把大 DataFrame 切成小块，每块几千行，分批执行。这样内存占用低，而且每批执行完就提交，不会因为一个错误导致全部回滚。

第二，关闭自动提交，手动控制事务。engine.begin()本身就是事务，但如果用engine.connect()，默认是 autocommit 模式，每条语句都提交一次，非常慢。一定要用begin()。

第三，调整 MySQL 参数。写入前临时设置innodb_flush_log_at_trx_commit=2，可以减少磁盘 I/O，写入速度能提升好几倍。当然，这个设置会稍微降低数据安全性，适合批量导入场景，导入完再改回来。

实操心得：我实测过，一万行数据用to_sql追加大概 3 秒，用 upsert 大概 5 秒，差距不大。但十万行数据时，to_sql要 30 秒以上，upsert 只要 15 秒左右。数据量越大，upsert 的优势越明显。

6. 常见问题与排查技巧实录

6.1 连接报错：Access denied 和 Gone away

Access denied for user一般是用户名密码错了，或者该用户没有从当前主机连接的权限。MySQL 的用户是user@host的形式，user@localhost和user@%是两个不同的用户。如果你从另一台机器连，需要确保用户的主机部分是%或者对应的 IP。

MySQL server has gone away是连接空闲太久被服务器断开了。解决办法就是前面说的pool_recycle和pool_pre_ping两个参数。如果还是不行，检查 MySQL 的wait_timeout参数，默认是 28800 秒（8 小时），可以适当调大。

6.2 中文乱码与字符集问题

乱码的根源是字符集不匹配。MySQL 有服务器字符集、数据库字符集、表字符集、连接字符集四层。任何一层不是 utf8mb4，都可能出问题。建库时指定CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci，连接字符串里加?charset=utf8mb4，基本就能解决。

如果已经建了表发现乱码，可以用ALTER TABLE修改字符集，但已有数据可能已经损坏，需要重新导入。

6.3 主键冲突与重复数据清理

主键冲突说明表里已经有相同主键的数据。如果你用的是to_sql的 append 模式，就会报错。解决办法有两个：一是改用 upsert，二是先删掉旧数据再插入。

如果表里已经有重复数据（比如之前没设主键），可以用这条 SQL 清理：

DELETE t1 FROM daily t1 INNER JOIN daily t2 WHERE t1.id > t2.id AND t1.ts_code = t2.ts_code AND t1.trade_date = t2.trade_date;

这条语句保留 id 较小的那条，删掉重复的。清理完记得加上主键约束，防止再次出现重复。

6.4 写入速度慢的排查思路

写入慢的原因可能有很多。先看是不是逐行插入，如果是，改成批量。再看是不是 autocommit 模式，如果是，改成事务。然后看索引，索引太多会拖慢写入，可以先把非必要索引删掉，导入完再重建。最后看硬件，机械硬盘的写入速度就是比 SSD 慢很多，这个没办法，只能换硬件。

下面这张表整理了几个典型问题和对应的解决方向：

7. 增量更新与自动化调度

7.1 如何判断哪些数据需要更新

增量更新的核心是知道“上次更新到哪了”。最简单的办法是查表里的最大日期：

last_date = pd.read_sql('SELECT MAX(trade_date) FROM daily', engine).iloc[0, 0]

然后从last_date的下一天开始拉取。但这样有个问题：如果某天的数据之前拉取失败了，这个办法就漏掉了。更稳妥的做法是维护一张更新日志表，记录每次拉取的日期范围和状态，失败的下次重试。

7.2 定时任务的几种实现方式

自动化调度可以用操作系统的定时任务，比如 Linux 的 crontab 或者 Windows 的任务计划程序。也可以用一个常驻的 Python 进程，配合schedule库或者APScheduler来做。

我一般用 crontab，简单可靠。每天收盘后跑一次，拉取当天数据入库。crontab 配置大概是这样：

30 16 * * 1-5 /path/to/python /path/to/update_daily.py >> /path/to/log.log 2>&1

周一到周五下午 4 点半执行，输出重定向到日志文件，方便排查。

7.3 日志记录与异常告警

日志不用太复杂，记录时间、拉取的股票范围、写入行数、耗时、是否成功就够了。我习惯用 Python 的 logging 模块，同时输出到文件和控制台。

异常告警可以用邮件或者即时通讯工具的机器人。比如写入失败时，发一条消息到工作群。这样不用天天盯着，出问题能第一时间知道。

实操心得：日志文件一定要按日期切割，不然跑几个月就变成一个巨大的文件，打开都费劲。可以用TimedRotatingFileHandler，每天自动切一个新文件，保留最近 30 天。

8. 一些踩坑之后的个人体会

这套方案我前后迭代了好几个版本，最早用 CSV，后来换 SQLite，最后定在 MySQL。中间踩过的坑包括但不限于：字符集没设对导致中文乱码、连接池没配导致半夜任务失败、主键没设导致数据重复、逐行插入导致跑一晚上还没跑完。每一个坑都让我对“数据入库”这件事有了更深的理解。

如果让我给新手一个建议，那就是：先把表结构设计好，字段类型、主键、索引都定清楚，再写写入代码。表结构是地基，地基没打好，后面怎么优化都是事倍功半。另外，不要怕麻烦，增量更新和日志记录这些看似多余的工作，在数据量上来之后会救你无数次。

最后分享一个小技巧：写入之前先用df.head()和df.dtypes检查一下数据，确认列名、类型、空值都符合预期。这个习惯能帮你提前发现 80% 的入库问题，比事后排查高效得多。