简介这份 tushare 最新源代码面向具备一定 Python 基础、希望深入理解财经数据接口实现的金融数据分析学习者可用于研究行情数据获取、存储与预处理的完整链路。压缩包共 44 个文件以 25 个 py 源码为主体辅以 7 个 rst 文档及 txt、md、yml 等配置与说明文件整体约 269KB目录覆盖接口调用、测试脚本和文档模块。已有 1494 人学习说明其对入门与进阶读者均有参考价值。源码中可重点研读 HTTP 请求封装、数据解析与异常处理SQLite、MongoDB 等存储操作以及缺失值填充、时间序列处理、matplotlib/seaborn 可视化和多线程、异步 IO 调度等实现从而借鉴数据获取模块设计、理解并发控制与日志记录思路提升金融数据项目的开发与排错能力。1. 拿到 tushare 最新源代码先看的是两百行的 client.py第一次拆 tushare 最新源代码的时候我没急着跑示例而是先打开pro/client.py。整份源码里最能说明这个项目性质的就是它一个两三百行的文件把 token 怎么存、请求往哪发、返回怎么变成 DataFrame全交代清楚了。tushare 本身是 Python 的金融数据接口包把 A 股行情、财务、指数、基金、期货、宏观这些数据统一收成「一个函数一次调用、直接返回 DataFrame」的形态。它适合三类人做量化回测需要干净历史数据的、搭内部数据管道要稳定增量更新的、以及想拿真实行情练手又不想碰网页解析的。读它的价值不在于会调pro.daily()而在于你能看到一个接口层项目怎么把几百个数据接口压进一套统一的请求骨架里。2. 包结构拆解从 set_token 到 DataFrame 的一条完整链路tushare 的源码不大但分层非常清楚。真正值得逐行看的是三个地方client.py负责网络与鉴权data_pro.py铺开所有业务接口util/upass.py管 token 的落盘。其余文件基本是常量和格式化工具。把这三块读懂整份源码就没什么黑匣子了。2.1 目录骨架三层分工别找错文件先看结构。不同版本的文件名会有些出入但职责划分是一致的tushare/ ├── __init__.py # 对外只暴露 set_token / pro_api / pro_bar 等少数入口 ├── pro/ │ ├── __init__.py │ ├── client.py # 网络层拼 payload、发 POST、异常码判断 │ ├── data_pro.py # 接口层daily / stock_basic / income 等几百个薄函数 │ ├── cons.py # 常量层接口名、字段列表、权限要求 │ └── data_api.py # 早期版本的动态分发实现新版多为显式函数 └── util/ ├── upass.py # token 的读写 └── formatter.py # ts_code、日期这类参数的正规化这个分层的关键在于接口层不碰网络网络层不认识业务。data_pro.py里的daily()只是把参数原样递给client.query()自己不做任何校验client.query()也不知道daily是什么业务它只认api_name这个字符串。好处是加接口的成本极低——服务端上线一个新接口客户端只需要补一个同名函数网络层一行都不用动。代价是所有参数校验的责任被推给了调用方这一点后面会反复提到。提示如果你只想搞清楚某个字段是什么意思去cons.py里搜接口名比翻文档快想搞清楚某个接口为什么报错去client.py看异常是怎么抛的。2.2 token 存取一个隐藏文件撑起整套鉴权tushare 的鉴权模型很朴素一个字符串 token跟着每次请求发过去。源码里负责这件事的是util/upass.py逻辑大致如下# tushare/util/upass.py 的等价写法 import os TOKEN_FILE os.path.join(os.path.expanduser(~), .tushare.csv) def set_token(token: str) - None: 把 token 落到用户目录之后所有进程共用 with open(TOKEN_FILE, w, encodingutf-8) as f: f.write(token.strip()) def get_token() - str: 读不到就返回空串由上层决定抛什么错 if not os.path.exists(TOKEN_FILE): return with open(TOKEN_FILE, r, encodingutf-8) as f: return f.read().strip()逻辑说明set_token()写文件、get_token()读文件中间没有加密也没有缓存失效机制。~/.tushare.csv这个名字带 csv 后缀但其实只是一行纯文本文件名是历史遗留不用纠结。参数上唯一的坑是os.path.expanduser(~)——它依赖HOME环境变量。这意味着三件事。第一在 Docker 容器里跑HOME可能是/root也可能是空token 文件写进去下次重建就没了。第二CI 环境里没有交互式终端你不能指望人工执行一次set_token。第三同一台机器上多个项目共用一个 token 文件想切账号就得手动改。我一般的做法是不依赖落盘文件而是在程序启动时从环境变量注入import os import tushare as ts def init_api(): token os.environ.get(TS_TOKEN, ) if not token: raise RuntimeError(缺少 TS_TOKEN 环境变量拒绝启动) # 显式设置不依赖 ~/.tushare.csv 是否存在 ts.set_token(token) return ts.pro_api()把 token 从「文件状态」变成「启动参数」容器化、CI、多账号三个问题一次解决。这也是读这份源码最容易拿到的一个改造点。2.3 请求发出与响应组装三段式 payloadclient.py里的query()是整个包的心脏剥掉注释大概长这样# tushare/pro/client.py 核心简化 import requests import pandas as pd class Client: def __init__(self): self.__token self.__http_url http://api.tushare.pro # 服务端入口源码里的常量 self.__timeout 30 def query(self, api_name, fields, **kwargs): payload { api_name: api_name, # 接口名daily / stock_basic / income ... token: self.__token or get_token(), params: kwargs, # 业务参数原样透传不做清洗 fields: fields, # 逗号分隔的字段串空串表示走服务端默认 } resp requests.post(self.__http_url, jsonpayload, timeoutself.__timeout) result resp.json() if result.get(code) ! 0: # 服务端把业务错误也放在 200 响应里只能靠 code 判断 raise Exception(result.get(msg)) data result[data] return pd.DataFrame(data[items], columnsdata[fields])逐段拆。payload是固定的四个键多一个少一个都不行api_name决定调哪个接口token决定你有没有权限params是业务参数原样塞进去fields是字段裁剪。注意params这一层完全没有校验——你传start_date2024-01-01带横杠还是20240101它都照发服务端认不认是另一回事。响应体是{code, msg, data}三段。code ! 0时msg里是人话错误信息但 HTTP 状态码依然是 200所以不能用resp.raise_for_status()判错必须看code。data里fields是列名数组、items是二维数组两者按位置对应——这也是为什么最终能用pd.DataFrame(items, columnsfields)一行搞定。这种「列存 行存分离」的传输格式在行情类接口里很常见因为一次返回几千行、每行十几个字段时它比对象数组省一半以上的体积。timeout30是个容易被忽略的点源码里给了超时但没有重试。网络抖一下就抛异常循环跑到一半断掉前面拉的几百个交易日全白费。这个问题我在第 3 章单独处理。3. 核心实现细读重试包装、接口模板与复权分支这一章讲三个真正需要动手改的地方。client.py缺重试data_pro.py缺校验pro_bar的复权逻辑跟大多数人想的不一样。这三处搞明白你就能把这份源码变成自己能用的东西。3.1 给 query 补一层重试区分可重试与不可重试源码里的query()是一次性买卖失败就抛。生产环境下你必须区分两类错误限频、超时这类是「等一会儿再来」参数错误、权限不足这类是「再来一万次也一样」。混在一起重试只会把配额烧光。import time def query_with_retry(client, api_name, retries3, backoff2.0, **kwargs): 在 Client.query 外面套一层有差别的重试 last_err None for i in range(retries): try: return client.query(api_name, **kwargs) except Exception as e: last_err e msg str(e) # 限频固定退避指数增长 if 每分钟 in msg or 频率 in msg: time.sleep(backoff ** i) continue # 网络层超时同样值得重试 if timeout in msg.lower() or connection in msg.lower(): time.sleep(backoff ** i) continue # 其余权限、参数直接抛重试没意义 raise raise last_err逻辑说明retries控制总次数backoff是退避基数第 i 次等待backoff ** i秒——三次重试分别等 1 秒、2 秒、4 秒backoff2.0。判断依据是错误信息里的关键词这不算优雅但足够用因为服务端的msg文案相对固定。真正不可重试的是权限类错误返回信息里通常带「积分」或「权限」字样这类错误重试只会让你在日志里刷屏看不出真正原因。参数上有个细节**kwargs必须原样透传给client.query()因为你不知道调用方会传ts_code还是trade_date。这也是接口层薄的好处——包装层不需要认识业务参数。3.2 data_pro一个模板铺开几百个接口data_pro.py打开会看到大量长得几乎一样的函数这是它最没技术含量也最省事的地方# tushare/pro/data_pro.py 里成百上千个这样的薄函数 def daily(ts_code, trade_date, start_date, end_date, fields): return pro.query(daily, ts_codets_code, trade_datetrade_date, start_datestart_date, end_dateend_date, fieldsfields) def stock_basic(exchange, list_statusL, fields): return pro.query(stock_basic, exchangeexchange, list_statuslist_status, fieldsfields)逻辑说明每个函数只做两件事——固定api_name把形参搬进 kwargs。默认值全是空串空串也会被塞进params发出去。多数情况下服务端会把空串当成「不筛选」但个别接口会因此返回全量数据这就是「我只想查一只股票结果拉回来五千行」的来源。所以接口层之上我通常会加一个参数收敛函数把常用的坑一次堵掉import re def normalize_code(code: str) - str: 把 000001 / sz000001 / 000001.sz 统一成 000001.SZ code code.strip().upper() m re.match(r^([A-Z]{2})(\d{6})$, code) # sz000001 if m: return f{m.group(2)}.{m.group(1)} m re.match(r^(\d{6})\.([A-Z]{2})$, code) # 000001.SZ if m: return code if re.match(r^\d{6}$, code): # 裸代码默认深市 return f{code}.SZ raise ValueError(f无法识别的证券代码: {code}) def check_date(d: str) - str: 只接受 YYYYMMDD其他格式一律报错而不是透传 if not re.match(r^\d{8}$, d): raise ValueError(f日期必须是 YYYYMMDD: {d}) return dnormalize_code处理的是 tushare 里最常见的一类翻车裸代码不带后缀服务端查不到但不报错返回空 DataFrame。这比抛异常更难查因为你看到的是「没数据」而不是「出错了」。check_date则是把格式问题挡在本地省一次网络往返和被扣的调用次数。3.3 pro_bar复权行情为什么要单独走一条路pro_bar是整份源码里唯一带业务逻辑的函数也是最容易用错的一个。它不是一个独立接口而是「不除权行情 复权因子」在本地算出来的def pro_bar(ts_code, adjNone, start_dateNone, end_dateNone): adj: None 不复权 / qfq 前复权 / hfq 后复权 df pro.daily(ts_codets_code, start_datestart_date, end_dateend_date) if adj in (qfq, hfq): factor pro.adj_factor(ts_codets_code, start_datestart_date, end_dateend_date) df df.merge(factor[[trade_date, adj_factor]], ontrade_date, howleft) # 前复权以最新一天为基准后复权以首日为基准 base df.sort_values(trade_date)[adj_factor].iloc[-1] if adj qfq else 1.0 for col in [open, high, low, close, pre_close]: df[col] df[col] * df[adj_factor] / base return df.sort_values(trade_date).reset_index(dropTrue)逻辑说明adj_factor是复权因子除权除息日会跳变。前复权让最新价保持不变、历史价等比缩放后复权让首日价保持不变、后面价格往上叠。base取最新一天的因子做前复权基准是常见做法。两个必须注意的点。第一复权不是只复权收盘价open/high/low/pre_close都得乘同一个因子只改close会导致你画出来的 K 线是断的回测里算出来的振幅也是错的。第二merge用left——因为adj_factor接口在某些停牌区间可能缺行用内连接会静默丢数据比报错更难发现。注意前复权价格会随最新交易日滚动变化今天存下来的 qfq 数据明天再拉一遍就对不上了。做长期存储建议存不复权价 复权因子用的时候现算。4. 跑起来从装包到全市场日线落盘源码读到这里该动手了。这一章按「能不能跑通 → 能不能批量拉 → 能不能每天自动跑」三级推进每一步都给可抄的代码。4.1 环境与安装python -m venv .venv source .venv/bin/activate pip install tushare pandas pyarrow python -c import tushare as ts; print(ts.__version__)三件事说明一下。用虚拟环境是为了避免和系统里其他 pandas 版本打架tushare 对 pandas 版本不算特别敏感但也不是完全无感。装pyarrow是为了后面落 Parquet——CSV 存全市场日线一年大概几百兆Parquet 能压到十分之一读回来还快。最后那行打印版本号是确认装上了而不是装了个同名空包。4.2 第一个能跑通的脚本import tushare as ts ts.set_token(你的token) # 只需一次会写进 ~/.tushare.csv pro ts.pro_api() df pro.daily(ts_code000001.SZ, start_date20240101, end_date20240331) print(df.shape, df.columns.tolist()) print(df.head(3))set_token只需执行一次之后pro_api()会自动去读。daily的四个参数全都有默认值最省事的写法是pro.daily(ts_code000001.SZ)但它默认只返回最近若干个交易日想要完整区间必须显式给start_date和end_date。日期格式是YYYYMMDD字符串带横杠的2024-01-01不会报错但会返回空表——这是新手最常见的第一次翻车。跑通之后立刻做一件事打印df.dtypes。trade_date回来是字符串vol单位是手amount单位是千元这几个单位不搞清楚后面算出来的换手率和成交额全是错的。4.3 拉全市场日线并按日分片落盘按股票循环拉是错的——五千只股票就是五千次请求配额瞬间见底。正确姿势是按交易日拉一次拿全市场import os import time import pandas as pd import tushare as ts pro ts.pro_api() os.makedirs(data/daily, exist_okTrue) def fetch_by_date(trade_date: str) - pd.DataFrame: 按交易日拉全市场日线一次请求覆盖五千多只股票 df pro.daily(trade_datetrade_date) if df.empty: return df df[trade_date] pd.to_datetime(df[trade_date]) # 单位统一手 - 股千元 - 元 df[vol] df[vol] * 100 df[amount] df[amount] * 1000 return df for d in pd.date_range(2024-01-02, 2024-01-31, freqB): ds d.strftime(%Y%m%d) out fdata/daily/{ds}.parquet if os.path.exists(out): # 断点续传别重复拉 continue df fetch_by_date(ds) if df.empty: print(f{ds} 无数据跳过) continue df.to_parquet(out, indexFalse) print(f{ds} 写入 {len(df)} 行) time.sleep(0.4) # 别贴着限频跑逻辑说明freqB是工作日但 A 股有节假日所以空表要正常跳过而不是当失败。os.path.exists做断点续传脚本中断后重跑不会重复消耗配额。time.sleep(0.4)是给自己留余量具体间隔取决于你的账号等级——这个值宁大勿小。按日分片存的好处是增量极其简单想知道拉到哪天了看目录里最后一个文件名就行。4.4 增量更新用交易日历别用工作日def pending_days(start20240101, end20241231): 返回还没落盘的交易日列表 cal pro.trade_cal(exchangeSSE, start_datestart, end_dateend, is_open1) days sorted(cal[cal_date].tolist()) return [d for d in days if not os.path.exists(fdata/daily/{d}.parquet)]trade_cal是必须用的is_open1只返回开市日这样春节、国庆不会被误判成「拉了但没数据」。这个接口一年只需要调一次把结果缓存成 CSV后面每天比对文件是否存在即可。整套流程挂到定时任务上每天收盘后跑一次pending_days()正常情况下只会返回当天一天的数据几秒钟跑完。5. 排查手册token、限频、空表和复权这四个坑下面这五条是我自己在用这套源码时真实踩过的按「现象 → 原因 → 解决」写遇到了直接对号入座。5.1 权限与 token 相关的坑现象一报错信息里带「积分」或「权限」但同样的代码别人能跑。原因tushare 的接口是分层的不同接口要求不同的积分门槛daily这种基础接口门槛低income、moneyflow这类就要高一些。你调不通不是代码问题是账号权限不够。另一种情况是 token 根本没读到——get_token()读不到文件时返回空串请求照样发出去服务端返回的是权限错误看起来像是积分问题。解决先用最基础的接口探活pro.stock_basic()能返回数据说明 token 是通的问题出在接口权限上。探活这一步建议写进启动自检里def health_check(pro): try: df pro.stock_basic(exchange, list_statusL, fieldsts_code) print(f[OK] token 有效股票列表 {len(df)} 条) return True except Exception as e: print(f[FAIL] 鉴权或网络异常: {e}) return False现象二本地能跑部署到服务器就报鉴权失败。原因~/.tushare.csv在服务器的HOME下不存在或者进程用户没有写权限。容器每次重建这个文件就没了。解决不要依赖落盘文件改成环境变量注入见 2.2 的init_api()。如果必须用文件就在镜像构建时COPY进去并把权限设成 600。5.2 限频与超时相关的坑现象三循环跑到一半抛异常提示每分钟访问次数超限前面的数据白拉了。原因client.query()里没有重试也没有节流循环写得多快就发得多快。5000 次请求打出去服务端一定拦。解决两层防护。一层是固定间隔time.sleep()简单粗暴但有效另一层是query_with_retry里的退避重试。真正要注意的是不要把重试写成死循环退避次数用完必须抛出去否则你会盯着一个卡住的进程不知道发生了什么。现象四偶发超时重跑就好了但自动化脚本没人盯着。原因源码给了timeout30但没给重试网络抖动直接变成任务失败。解决把重试包在批量循环的最外层而不是每个接口里。单个接口失败重试三次仍失败就跳过这一天并记录到失败清单下一轮增量更新会自动补上——因为pending_days()是靠文件存在性判断的天然支持补偿。5.3 数据本身的坑现象五接口没报错但返回空 DataFrame查半天查不出原因。原因三种可能——证券代码少了后缀000001而不是000001.SZ、日期格式带横杠、查的那天不是交易日。这三种情况服务端都当成「查询无结果」处理不报错。参数原样透传的设计让这些错误在客户端完全无法拦截。解决所有入参走一遍normalize_code()和check_date()把格式问题挡在本地。批量拉取时对空表做显式判断并打日志别让它静默写进一个空文件——否则下次增量更新会误以为这天已经拉过了。现象六复权后的价格和行情软件对不上差的那点不多不少刚好是分红。原因只复权了closeopen/high/low/pre_close还是原始价或者前复权的基准日选错了用了区间首日而不是最新交易日。解决全列同因子相乘基准统一取最新交易日的adj_factor。长期存储改成「不复权价 因子」用的时候现算这样历史数据永远不需要回溯修正。6. 二次开发把接口包改造成自己的数据层读完源码最该带走的一个认知是tushare 是个传输层不是数据层。它负责把请求发出去、把 JSON 变成 DataFrame至于缓存、幂等、字段裁剪、失败补偿一概不管。所以二次开发的方向不是改它而是在它外面包一层。我一般的结构是三层最底下是Client中间是「接口注册表」最上面是业务函数。注册表用一个字典描述每个接口的特性比散落在各处的if好维护得多配置项作用常见取值api_name对应服务端接口名daily/adj_factorkey_fields幂等去重用的主键[ts_code, trade_date]min_interval两次调用的最小间隔秒0.3~1.0cache_ttl本地缓存有效期秒静态表 86400行情 0有了这张表缓存和限频就能写成通用逻辑调之前先查min_interval决定要不要 sleep静态表stock_basic、trade_cal走本地缓存行情表不缓存直接落 Parquet。第二个值得做的改造是把网络层抽成接口。这样单元测试里可以塞一个假 client返回固定的{fields, items}整个数据管道不用联网就能跑。我吃过这个亏——早期测试直接连真实接口跑一次测试烧掉半天配额还把测试结果搞得依赖网络状态时快时慢。后来改成离线回放把真实响应存成 JSON测试时按api_name取跑得快还没副作用。验证改造是否成功我只看两个指标同一批参数跑两遍落的文件字节数完全一致幂等断网情况下重跑能从断点继续而不是从头再来补偿。这两个都对这套管道才算能用。从那以后我每次包一层 SDK都强制先写一个假的 client 把离线测试跑通再连真实接口——不然你永远分不清是代码写错了还是配额用完了。希望帮到你。本文还有配套的精品资源点击获取
