tushare源码解析：Python量化数据接口封装与本地缓存实践

简介这份资源是 tushare 开源 Python 财经数据接口库的最新源代码包面向金融数据分析从业者、量化研究者以及希望深入理解数据接口实现原理的开发者。包内共 44 个文件以 25 个 py 源码文件为核心辅以 7 个 rst 文档、3 个 txt 说明、2 个 md 文档以及 makefile、yml、csv、bat、manifest 等配置与测试文件压缩包约 269KB结构紧凑、模块划分清晰。目前已有 1494 人学习下载。通过研读源码读者可以掌握股票、期货、期权、基金、外汇、债券等数据的获取封装方式理解 HTTP 请求、数据解析与异常处理逻辑同时能学习 SQLite、MongoDB 等数据库的存储与检索设计数据清洗、缺失值填充与时间序列处理方案以及 matplotlib、seaborn 可视化接口的调用方法。多线程与异步 IO 的并发调度、错误处理与日志记录机制也值得借鉴有助于提升大规模金融数据处理的效率与程序健壮性。1. 从一份源码包说起tushare 到底能帮你拿到什么数据如果你写过 A 股相关的策略回测大概率遇到过这种场景想拉一份完整的日线行情从某财经页面复制粘贴格式乱、字段缺、除权除息还得自己算。折腾半天策略逻辑一行没写数据清洗倒写了三百行。tushare 这个项目解决的正是这个痛点——它把行情、财务、指数、基金等数据封装成统一的 Python 接口一行调用就能拿到结构化 DataFrame。这份最新源代码适合两类人一是想快速搭建数据管道的量化新手二是想读懂数据接口层设计、准备做二次封装的老手。源码包里通常包含核心请求模块、接口定义、字段映射和示例脚本拿到手就能本地跑通。2. 项目定位与技术选型为什么它成了数据接口层的常客2.1 它解决的不是「取数」而是「统一取数」很多人第一次接触 tushare以为它就是个爬虫封装。实际拆开源码会发现它的核心价值在于把不同来源、不同格式、不同频率的数据统一成一套调用约定。比如日线行情、复权因子、财务指标在底层可能是完全不同的接口协议但对外都暴露成pro.daily()、pro.adj_factor()、pro.fina_indicator()这种风格一致的函数。这种设计让策略代码和数据源解耦——哪天某个接口换了地址你只需要改配置不用动策略逻辑。从源码结构看它通常分成三层最底层是 HTTP 请求与重试机制中间层是接口路由与参数校验最上层是面向用户的 API 对象。这种分层在 Python 项目里很常见但 tushare 做得比较克制没有过度抽象读起来不费劲。2.2 技术栈轻依赖、重约定源码包里的依赖通常不多核心就是requests做网络请求pandas做数据容器simplejson或标准库处理序列化。这种轻依赖的好处是部署成本低一台普通开发机pip install就能跑。但轻依赖也意味着它不帮你做太多事——比如缓存、并发、断点续传这些得自己加。我一般会在它的请求层外面包一层本地缓存用 SQLite 或 Parquet 存已经拉过的数据避免重复请求。源码里通常留了timeout和retry参数但默认值偏保守实际用的时候建议根据网络情况调大。2.3 接口鉴权与 token 机制tushare 的接口大多需要 token源码里一般会有一个set_token()方法把 token 存到全局配置或环境变量里。拆开看它其实就是在请求头里加了一个字段。这里有个容易翻车的点token 不要硬编码在脚本里尤其是要提交到代码仓库的时候。常见做法是读环境变量或者用一个单独的config.py并加入.gitignore。import os import tushare as ts # 从环境变量读取 token避免硬编码 token os.environ.get(TUSHARE_TOKEN) if not token: raise ValueError(请先设置 TUSHARE_TOKEN 环境变量) ts.set_token(token) pro ts.pro_api() # 拉取单只股票日线指定日期区间 df pro.daily(ts_code000001.SZ, start_date20240101, end_date20240131) print(df[[trade_date, open, close, vol]].head())这段代码的逻辑很直白先拿 token再初始化 API 对象最后调具体接口。参数说明上ts_code是带交易所后缀的股票代码start_date和end_date格式是YYYYMMDD。返回的 DataFrame 字段名是英文缩写源码里通常有一份字段映射表想改成中文列名可以自己写个 rename。3. 目录结构与核心模块拆解一个请求从发起到返回3.1 源码目录长什么样拿到源码包后先别急着跑setup.py花五分钟把目录扫一遍。典型的 tushare 源码结构大致如下tushare/ ├── __init__.py # 暴露顶层 API如 pro_api、set_token ├── pro/ │ ├── client.py # 请求客户端封装 HTTP 调用与重试 │ ├── data_pro.py # 各接口的具体实现按数据类别分文件 │ └── utils.py # 参数校验、日期处理、字段映射 ├── stock/ │ └── cons.py # 股票列表、交易日历等基础数据 ├── examples/ │ └── demo_daily.py # 官方示例脚本 └── setup.py # 安装配置这个结构不是固定的不同版本会有调整但核心思路一致client.py管通信data_pro.py管业务utils.py管工具。读源码时建议从__init__.py入手看它导出了哪些名字再顺着调用链往下追。3.2 请求客户端重试与超时怎么做的client.py是整个项目的咽喉。它通常定义一个Client类内部持有requests.Session并在query()方法里处理重试。下面是一段常见的实现风格import time import requests class Client: def __init__(self, token, timeout15, max_retry3): self.token token self.timeout timeout self.max_retry max_retry self.session requests.Session() self.session.headers.update({Authorization: fBearer {token}}) def query(self, url, params): 带重试的 GET 请求失败时指数退避 for attempt in range(self.max_retry): try: resp self.session.get(url, paramsparams, timeoutself.timeout) resp.raise_for_status() return resp.json() except requests.RequestException as e: if attempt self.max_retry - 1: raise # 指数退避避免瞬间打满 time.sleep(2 ** attempt)逻辑说明query()是统一入口所有接口最终都走这里。参数上timeout控制单次请求超时max_retry控制重试次数2 ** attempt实现指数退避。这里有个细节raise_for_status()会把 4xx 和 5xx 都抛出来但实际业务里 4xx 重试没意义可以加个判断只对 5xx 重试。我一般会改成if resp.status_code 500才重试。3.3 接口实现层参数校验与字段裁剪data_pro.py里通常是一堆方法每个方法对应一个接口。以daily()为例它做三件事校验参数、拼请求、裁剪字段。def daily(self, ts_codeNone, trade_dateNone, start_dateNone, end_dateNone): 日线行情支持单只或按日期区间 if not ts_code and not trade_date: raise ValueError(ts_code 和 trade_date 至少传一个) params { api_name: daily, token: self.token, params: {ts_code: ts_code, trade_date: trade_date, start_date: start_date, end_date: end_date}, fields: ts_code,trade_date,open,high,low,close,vol,amount } data self.client.query(self.url, params) return pd.DataFrame(data[items], columnsdata[fields].split(,))参数说明ts_code和trade_date是互斥的常用组合源码里一般会做校验。fields字段控制返回列不传就返回全部但全部字段往往很多按需裁剪能省带宽。返回时用pd.DataFrame包装列名从fields里切分。这里有个坑如果接口返回空data[items]可能是空列表直接构造 DataFrame 会得到空表后续计算要判空。3.4 基础数据模块交易日历与股票列表stock/cons.py这类文件通常放的是「不常变但每次都要用」的数据比如交易日历、股票列表、复权因子。这些数据的特点是拉一次可以缓存很久但每次策略启动都要读。源码里一般会提供一个trade_cal()接口返回指定年份的交易日。# 获取 2024 年交易日历只保留开市日 cal pro.trade_cal(exchangeSSE, start_date20240101, end_date20241231) open_days cal[cal[is_open] 1][cal_date].tolist() print(f2024 年开市天数{len(open_days)})这段代码的用途很实际回测时遍历交易日而不是遍历自然日。参数exchange指定交易所is_open为 1 表示开市。我一般会把结果存成 CSV下次直接读省得每次都请求。4. 本地跑通与数据落库从安装到第一份 Parquet4.1 安装与初始化源码包通常支持两种安装方式pip install .从本地安装或者直接python setup.py develop做开发模式安装。开发模式的好处是改源码立即生效不用反复重装。安装完先跑一遍examples/demo_daily.py确认能拿到数据。# 进入源码目录开发模式安装 cd tushare-src pip install -e . # 设置 token 后运行示例 export TUSHARE_TOKEN你的token python examples/demo_daily.py如果示例报错先看报错类型ModuleNotFoundError说明依赖没装全HTTPError 401说明 token 无效KeyError说明返回字段和预期不一致。这三种是最常见的。4.2 批量拉取与本地缓存单次拉一只股票没意思实际用的时候都是批量。下面这段代码演示了如何遍历股票列表把日线数据存成 Parquet下次直接读本地。import os import pandas as pd import tushare as ts pro ts.pro_api() cache_dir ./data/daily os.makedirs(cache_dir, exist_okTrue) # 获取全市场股票列表 stocks pro.stock_basic(exchange, list_statusL, fieldsts_code) codes stocks[ts_code].tolist() for code in codes[:50]: # 先跑 50 只试试 path os.path.join(cache_dir, f{code}.parquet) if os.path.exists(path): continue # 已有缓存就跳过 df pro.daily(ts_codecode, start_date20230101, end_date20241231) if df.empty: continue df.to_parquet(path, indexFalse) print(f{code} 已缓存 {len(df)} 行)逻辑说明先拿股票列表再逐个拉取存成 Parquet。Parquet 比 CSV 省空间读取也快。参数上list_statusL表示只取上市状态的股票避免拉到退市的。这里有个优化点pro.daily()一次只能拉一只但有些接口支持按日期拉全市场那种方式更适合做全量回测。4.3 数据校验拉完不等于能用拉完数据别急着跑策略先做几项校验行数是否合理、日期是否连续、价格是否有异常值。下面是一个简单的校验脚本。import pandas as pd df pd.read_parquet(./data/daily/000001.SZ.parquet) df[trade_date] pd.to_datetime(df[trade_date], format%Y%m%d) df df.sort_values(trade_date) # 检查缺失交易日 full_range pd.date_range(df[trade_date].min(), df[trade_date].max(), freqB) missing set(full_range) - set(df[trade_date]) print(f缺失交易日数量{len(missing)}) # 检查价格是否为负 neg df[(df[[open, high, low, close]] 0).any(axis1)] print(f异常价格行数{len(neg)})这段代码检查两件事工作日是否有缺失、价格是否有非正数。缺失交易日可能是停牌也可能是数据没拉到非正价格基本可以判定为脏数据。校验通过再入库能省掉后面很多玄学问题。5. 避坑与常见问题那些年我们踩过的数据接口坑5.1 现象接口返回空 DataFrame但代码没报错原因ts_code格式不对或者日期区间内确实没有数据。tushare 的股票代码必须带交易所后缀比如000001.SZ只写000001会返回空。另外停牌期间也可能返回空。解决先确认代码格式再用trade_cal查一下区间内是否有交易日。如果都没问题换一只活跃股票试试排除是单只股票的问题。5.2 现象频繁请求后接口变慢或返回限流提示原因tushare 对请求频率有限制具体阈值和账号权限有关。源码里的重试机制如果没做退避容易触发限流。解决在client.py里加请求间隔比如每次请求后time.sleep(0.3)。批量拉取时用队列控制并发别开太多线程。我一般会把批量任务放到晚上跑避开高峰期。5.3 现象复权价格和行情软件对不上原因tushare 返回的daily()是不复权价格需要配合adj_factor()自己算。很多人直接拿close做回测结果除权日出现巨大跳空。解决拉取复权因子按公式后复权价 不复权价 × 复权因子 / 最新复权因子计算。或者直接用pro_bar()接口它内部帮你处理了复权。# 用 pro_bar 直接拿后复权数据 df ts.pro_bar(ts_code000001.SZ, adjhfq, start_date20230101, end_date20241231) print(df[[trade_date, close]].head())参数adjhfq表示后复权qfq是前复权。回测一般用后复权避免未来函数。5.4 现象字段名和文档不一致KeyError 频发原因不同版本源码的字段命名可能有调整或者接口返回的字段比你请求的少。解决不要硬编码字段名先用df.columns打印出来看。源码里的fields参数建议显式指定避免依赖默认值。如果做二次封装把字段映射单独放一个字典里改起来方便。5.5 现象Parquet 文件越存越多磁盘吃紧原因每只股票一个文件全市场几千只就是几千个文件小文件多了既占空间又影响读取。解决按年份或按交易所合并比如daily_2024.parquet存全市场当年数据。或者用分区表按trade_date分区。我一般按月合并单文件控制在几百 MB读取时用pandas的columns参数只读需要的列。6. 二次开发把接口层改造成自己的数据管道6.1 封装一个带缓存的 DataFeed 类源码里的pro_api()是裸接口每次调用都走网络。二次开发的第一步是在外面包一层带缓存的DataFeed。下面是一个简化实现import os import pandas as pd import tushare as ts class DataFeed: def __init__(self, token, cache_dir./cache): ts.set_token(token) self.pro ts.pro_api() self.cache_dir cache_dir os.makedirs(cache_dir, exist_okTrue) def daily(self, code, start, end): 优先读本地缓存没有则请求并落盘 path os.path.join(self.cache_dir, f{code}_{start}_{end}.parquet) if os.path.exists(path): return pd.read_parquet(path) df self.pro.daily(ts_codecode, start_datestart, end_dateend) if not df.empty: df.to_parquet(path, indexFalse) return df def adj_daily(self, code, start, end): 后复权行情内部调用 pro_bar return ts.pro_bar(ts_codecode, adjhfq, start_datestart, end_dateend)这个类的价值在于策略代码只依赖DataFeed不直接依赖 tushare。哪天换数据源改DataFeed内部实现就行。参数上cache_dir建议放在项目根目录下的data/里方便统一管理。6.2 加一层字段标准化tushare 返回的字段是英文缩写不同接口风格还不完全一致。我一般会在DataFeed里加一个_standardize()方法把列名统一成date, open, high, low, close, volume, amount这种通用格式。COLUMN_MAP { trade_date: date, vol: volume, amount: amount, ts_code: code } def _standardize(self, df): df df.rename(columnsCOLUMN_MAP) if date in df.columns: df[date] pd.to_datetime(df[date], format%Y%m%d) return df.sort_values(date).reset_index(dropTrue)这样上层策略就不用关心数据源字段差异换接口时只改映射表。表格里列出常用映射关系原始字段标准字段说明trade_datedate交易日期volvolume成交量手amountamount成交额千元ts_codecode股票代码6.3 验证改造是否成功改完之后怎么验证我一般跑三个检查一是同一只股票缓存前后数据是否一致二是复权价格和不复权价格在除权日的比例是否等于复权因子三是随机抽几个交易日和行情软件对一下收盘价。这三个过了基本就能放心用。feed DataFeed(tokenos.environ[TUSHARE_TOKEN]) df1 feed.daily(000001.SZ, 20240101, 20240131) df2 feed.daily(000001.SZ, 20240101, 20240131) # 第二次走缓存 assert df1.equals(df2), 缓存前后不一致 print(缓存校验通过)6.4 一个容易忽略的点异常处理要分层源码里的异常处理比较粗网络错误和业务错误混在一起。二次开发时建议分层网络层抛ConnectionError业务层抛DataNotFoundError上层策略只捕获业务异常。这样排查问题时能快速定位是网络挂了还是数据没有。从那以后我每次封装数据接口都强制先写一个最小可用的DataFeed跑通缓存和字段标准化再往上叠策略。这个习惯帮我省了很多重复调试的时间。希望帮到你。本文还有配套的精品资源点击获取