简介作为开源的Python财经数据接口库tushare主打国内股票、期货、基金等金融数据的实时与历史获取面向金融数据分析初学者、量化研究者和Python开发者。这份最新源代码提供了从网络请求封装、数据解析、数据库存储、清洗预处理到可视化呈现的完整链路可以清晰看到作者在高并发、异步IO、错误处理与日志记录等方面的设计取舍。压缩包共四十四份文件以二十五份Python脚本为核心另有七份RST文档、三份TXT说明、两份Markdown文档以及Makefile、配置文件等辅助内容分别承载接口文档、安装说明与工程配置整体仅二百六十九KB结构精炼便于逐模块研读。目前已有一千四百九十四人学习查看具备一定关注度。通过阅读源码读者不仅能掌握tushare底层接口调用原理还能借鉴其中多线程调度、缓存策略、数据库索引优化等实战经验项目同时附带单元测试与使用文档适合在真实数据场景中对照练习、二次扩展有效提升金融数据分析与Python工程化能力。1. tushare 源码拿来就能跑先把 token 和积分这个门槛看明白做 A 股数据回测的开发者应该都有过这种经历网上一搜 tushare出来的代码五花八门有的还在用老接口有的拿到手就报权限错误。tushare 是一个免费但有门槛的金融数据接口库所谓「最新源代码」的关键不在代码本身而在你拿到代码之后怎么配 token、怎么过积分门槛、怎么把数据稳定落到本地。这篇笔记我会从接口分级逻辑讲到一套能跑的本地数据管道再把我实际踩过的几个坑原样摆出来。2. tushare 接口分级逻辑为什么新代码全员走 pro 请求2.1 老接口与 pro 接口的界线早年间 tushare 的使用方式很简单导入包之后直接调函数就能拿数据。后来数据源改版老接口逐渐停更新的数据接口全部迁到了 pro 体系下。现在的调用方式统一变成了pro_query或是直接调用ts.pro_api()返回的接口对象比如拿日线行情就是pro.daily()而不是老代码里的ts.get_k_data()。这两代代码最大的区别在于鉴权方式。老接口不需要 token拿着包就能用但数据源不稳定新接口强制要求设置 token并且在服务端做了积分体系——积分不够的账号连基础接口都调不通。如果你拿到一份标注「最新」的 tushare 源码打开之后发现里面没有ts.set_token()那这份代码基本可以判定是旧版本改造跑起来大概率报权限错误。新接口的调用链路是本地设置 token → 请求到达服务端鉴权 → 校验账号积分权限 → 返回数据或抛出异常。只要链路里任何一环失败你看到的不是数据而是一堆抱歉您没有权限之类的提示。所以拿到源码第一步不是看业务逻辑而是先确认鉴权那段代码的写法是否匹配新版 SDK。2.2 HTTP 直连与 Python SDK 怎么选pro 接口本质是一组 HTTP API所以你有两条路可以走直接用requests库 POST 参数过去或者通过官方 SDK 中转。SDK 做的事情其实就是把 HTTP 请求封装起来统一处理 token 注入、JSON 解析还有网络异常。我一般会建议用 SDK原因有三个一是参数校验更友好报错信息能直接告诉你缺了哪个字段二是返回结果已经解析成 DataFrame省去自己处理 JSON 的时间三是官方更新会跟随接口变动你要拿的是最新源码场景跟着 SDK 走不容易踩旧坑。以日线行情为例用 SDK 的写法非常短import tushare as ts ts.set_token(你的token字符串) pro ts.pro_api() df pro.daily(ts_code000001.SZ, start_date20240101, end_date20240110) print(df.head())这段代码先配置 token然后初始化 pro 接口对象接着调用daily()拉取指定股票在指定时间段的日线数据。ts_code是股票代码带后缀的写法000001.SZ表示深交所的平安银行600000.SH表示上交所的浦发银行start_date和end_date是闭区间格式固定为YYYYMMDD。返回的df直接就是 DataFrame字段里包含open、high、low、close、vol等常用行情列。如果你拿到手的是新版核心源码里面用的基本也是这套模式。HTTP 直连适合极端情况——比如 SDK 挂掉、你需要绕过它去排查接口本身的状态。判断标准是日常批量拉数用 SDK排查接口联通性再去看 HTTP 返回原文。3. 落地一套本地数据管道从 token 安全配置到增量更新3.1 目录结构与配置分离很多拿到的源码示例里token 就直接写在文件顶部短跑没问题但如果你想把拉数脚本沉淀成自己的工具这种写法很危险。token 泄露意味着别人可以用你的账号配额所以我在本地做了一套简单的分层data_pipeline/ ├── config.py # 从环境变量读取 token不硬编码 ├── fetcher.py # 封装拉数逻辑处理限频和重试 ├── storage.py # 落盘与增量合并 └── main.py # 任务编排入口config.py的核心作用是把 token 从代码里摘出去。用环境变量读既避免误提交到 git也方便在不同机器上切换账号import os TOKEN os.environ.get(TUSHARE_TOKEN, ) if not TOKEN: raise RuntimeError(请先设置环境变量 TUSHARE_TOKEN)这里不直接写raise之外的处理逻辑是为了让缺失配置的问题尽早暴露。你可以把TUSHARE_TOKEN加到 shell 的 profile 文件里或者在 IDE 的运行配置里注入。从资源复现的角度说拿到任何一份源码我都建议先做这个改造——它不影响业务逻辑但能避免后面所有因鉴权引发的连锁问题。3.2 认证、基础拉取与本地落盘fetcher.py里放的是拉数动作。单次请求的写法很直白但要注意两点一是pro_api()不要每次调用都重新初始化全局建一个对象就行二是日期参数尽量用变量维护别写死。我一般会这样设计import tushare as ts import pandas as pd from config import TOKEN ts.set_token(TOKEN) _pro ts.pro_api() def fetch_daily(ts_code: str, start_date: str, end_date: str) - pd.DataFrame: df _pro.daily( ts_codets_code, start_datestart_date, end_dateend_date ) if df is None or df.empty: return pd.DataFrame() return dfts_code和日期区间是必传参数不传就会报参数缺失。返回结果为空时函数返回空 DataFrame这样上层逻辑不需要额外判断None统一拿到的都是 DataFrame 类型。storage.py这边负责把数据落成本地文件。我习惯用 CSV 加日期分区的形式import os import pandas as pd def save_daily(df: pd.DataFrame, ts_code: str, save_dir: str) - str: os.makedirs(save_dir, exist_okTrue) file_path os.path.join(save_dir, f{ts_code}.csv) if os.path.exists(file_path): old_df pd.read_csv(file_path) combined pd.concat([old_df, df]).drop_duplicates(subset[trade_date]).sort_values(trade_date) else: combined df.sort_values(trade_date) combined.to_csv(file_path, indexFalse) return file_path这里的关键点是drop_duplicates(subset[trade_date])。tushare 的日线数据里trade_date是唯一键重复拉取同一区间时直接按日期去重避免数据翻倍。先用 concat 合并再去重比读一次写一次的覆盖式写入更稳。这个函数返回文件路径方便你后续打印日志确认落盘位置。3.3 增量更新与自动重试增量更新的思路很简单每次只拉最近缺失的区间。实际操作里我先把本地已有的最新交易日期读出来然后从那个日期的下一天开始拉。这里直接涉及daily()的参数设计——它的最大区间是单次调用两条日期之间的跨度但如果跨度太大单次请求的数据量会很重容易触发限频。常见做法是切到按月或按周迭代def update_daily(ts_code: str, end_date: str, save_dir: str) - None: local_file os.path.join(save_dir, f{ts_code}.csv) if os.path.exists(local_file): local_df pd.read_csv(local_file) last_date local_df[trade_date].max() else: last_date 19900101 start last_date df fetch_daily(ts_code, start, end_date) if not df.empty: save_daily(df, ts_code, save_dir) print(f{ts_code} 更新到 {df[trade_date].max()})重试机制放在fetch_daily外面包一层循环更好。网络抖动导致单次请求失败是很常见的情况重试时最怕把相同请求在短时间内反复打过去所以我习惯在重试之前加一个退避等待import time def fetch_daily_with_retry(ts_code: str, start_date: str, end_date: str, max_retries: int 3) - pd.DataFrame: for attempt in range(max_retries): try: return fetch_daily(ts_code, start_date, end_date) except Exception as e: if attempt max_retries - 1: raise time.sleep(2 * (attempt 1)) return pd.DataFrame()这个函数的逻辑是某次请求抛出异常先不慌睡两秒再试第二次失败睡四秒直到达到最大次数。max_retries3意味着最多三次重试。针对 tushare 的场景大部分异常是网络层面而非业务层面的重试能解决掉绝大部分偶发失败如果三次都失败那你要检查的是不是 token 失效或积分被限制了。4. 避坑笔记积分、限频与数据口径的三个血泪现场4.1 积分不是摆设接口额度逐级放开现象拿到源码配好 token跑pro.daily()报错提示积分不足。原因tushare 的接口按积分分级日线接口基础权限就要 120 积分积分通过完善资料和贡献数据获取。不是所有接口你有 token 就能调。解决先去官网个人主页看积分值确定当前账号能访问哪些接口。120 分以下只能调最基础的功能想拿财务数据或分钟线积分门槛会更高。如果你是本地自用先把账号的积分养起来靠日常签到和资料完善慢慢涨如果项目紧急就需要评估是否只用低门槛接口能完成任务。4.2 单次请求拉大跨度数据直接被截断现象我图省事一次把某只股票五年的日线全部拉下来结果返回的只有最近几百条。原因tushare 对单次请求返回的行数有限制日线接口单次最多返回 6000 行左右超过就自动截断。五年大约 1200 个交易日虽然没超 6000但如果你拉的是指数或全市场列表行数很容易顶到上限。解决严格按区间切片请求。每次拉一年的数据量是比较稳妥的做法再把结果合并。我后来在fetch_daily里加了一个按年份拆分的循环确保任何单次请求的条数都在安全线以内。4.3 交易日历不一致导致数据错位现象同一时间段某股票在本地数据里比官方多出几条记录回测结果对不上。原因A 股有节假日休市和临时停牌。tushare 的daily返回的是实际有成交的日子停牌日没有记录合并数据时空值处理不当就会造成日期错位。解决引入交易日历接口trade_cal做基准日对齐。例如用pro.trade_cal(exchangeSSE, start_date20240101, end_date20241231)拿到当年所有交易日再用交易日序列去左连接行情数据缺失的填充为空这样回测循环就不会因为日期错位而串行。4.4 复权因子没用对价格曲线断崖现象股票分红除权那天价格瞬间跳空回测的年化收益算出来严重失真。原因daily返回的是不复权价格除权日自然会出现价格断层。要算真实收益必须用前复权或后复权价格。解决用pro.adj_factor()获取复权因子自己计算前复权价格前复权价 原始价 × 最新因子 ÷ 当日因子。也可以直接用接口pro.daily(adjqfq)拿前复权数据但注意复权参数在不同接口里的支持情况不一样务必先确认接口文档。资金曲线和收益计算必须建立在复权价上这是量化回测的基础常识但太容易被人忽略。5. 用日线接口拼一个可复现的回测数据集5.1 标的选择与参数封装回测数据集的核心不只是行情字段还要包括股票代码列表和交易日历。我通常在main.py里显式维护一个标的清单避免在代码里写散落的字符串UNIVERSE [ 000001.SZ, 600000.SH, 000858.SZ, 600519.SH, ]这四个分别代表不同市场、不同行业的典型标的。000001.SZ是银行600000.SH是银行000858.SZ是白酒600519.SH也是白酒。选标的时覆盖两个交易所和不同板块便于验证数据管道对代码后缀的解析是否正确。配置好清单后用前一章写好的update_daily_with_retry循环拉取for code in UNIVERSE: try: file_path update_daily(code, end_date20241231, save_dir./data) print(f完成: {code} - {file_path}) except Exception as exc: print(f失败: {code}, 错误: {exc})循环体里把每个标的的失败原因打印出来便于排查是网络问题、积分问题还是代码问题。这里故意不做全局 try 包裹因为一个标的失败不应该中断整个批次打印完继续跑下一个。5.2 数据对齐与缺失值标记拉完所有标的之后下一步是把它们合并成一张宽表这样回测引擎直接读取一个 CSV 就行。合并逻辑以交易日为索引每个交易日一行标的的收盘价作为一列import pandas as pd def build_panel(data_dir: str, codes: list) - pd.DataFrame: panel pd.DataFrame() for code in codes: file_path f{data_dir}/{code}.csv df pd.read_csv(file_path, usecols[trade_date, close]) df df.rename(columns{close: code}) if panel.empty: panel df.set_index(trade_date) else: panel panel.join(df.set_index(trade_date), howouter) panel panel.sort_index() return panel.fillna(0.0)fillna(0.0)是一种直白的缺失值标记方式但回测时你要特别小心0 会被当成真实价格参与收益率计算导致结果出现异常。更稳妥的做法是先用一个哨兵值比如-1或者NaN回填在策略层面对缺失值做前向填充。我在这段代码里保留fillna(0.0)是为了让宽表可视化时一眼看到缺失位置真正进策略前你会换成ffill()或直接丢弃对应行。日期对齐方面join(howouter)保证不同标的的停牌日不会互相挤压缺的日期空出来后续统一处理。6. 验证数据集先看这些再开始写策略数据管道跑通并不代表数据能用。我每次拉完一批数据第一件事不是写策略而是花十分钟做一次数据完整性校验。校验工具就用 pandas 的透视表比什么脚本都直观。import pandas as pd panel pd.read_csv(panel.csv, index_col0) counts panel.notna().sum(axis1) print(counts.describe())counts.describe()会输出每行有效标的数量的分布。如果大部分交易日都有四个标的但某天突然只剩两个那一半可能是停牌一半可能是拉数时有遗漏。看到这种异常就回去单独检查那天的原始 CSV 是否存在。下一步验证字段值的合理性把每天的收益率极值打出来看看有没有因为除权未复权导致的异常跳变。正常情况下 A 股单日涨跌幅不超过 10%ST 股 5%科创板 20%如果算出超过 25% 的日收益数据里大概率混了未复权价格。我的习惯是直接打印异常行所在的日期和标的对照官网行情确认一遍再继续。复杂的校验方式很多但对大部分个人回测场景来说透视表加极值检查已经够用。从那以后我每次拉完数据都强制走一遍这个验证流程再进入策略编写。这样做不是为了仪式感而是因为数据错的代价太高——等你花一周写完策略发现是数据源不对那种返工的感觉真的很难受。希望这份从鉴权到落盘再到校验的完整链路能帮你少走一次这样的弯路。本文还有配套的精品资源点击获取
