简介面向量化交易初学者与Python开发者的一份tushare 0.2.8源码学习材料围绕金融数据接口的获取、清洗、存储与实时行情等核心环节展开适合希望读懂数据接口实现、提升网络请求与异常处理能力的人群。压缩包共26个文件约181KB以19个py源码为主体另含3个txt说明、2个pkg-info元数据、1个csv样例和1个cfg配置便于按包结构梳理模块职责与调用关系。已有6257人学习下载具备一定参考热度。读者可结合源码理解数据请求与解析流程、数据处理函数、本地CSV/Excel存储、实时行情轮询及技术指标计算等实现思路并从中借鉴接口设计、异常处理与金融数据工程的组织方式用于课程实践、项目参考或量化分析入门。1. 早上 8 点 40 的那次跑批才是这套源码的真正用法早上 8 点 40定时任务把昨天的日线补进本地库8 点 50全市场选股结果生成好推到手机9 点 20 之前你只需要确认一下有没有要下单的标的。这套 tushare 量化交易 Python 源码干的就是这件事——以 tushare 作为行情数据源把数据拉取、因子计算、策略回测、每日信号生成串成一条能定时跑、能重复跑、结果可复现的流水线。它适合两类人一类是 Python 已经能写点脚本、想把“看盘感觉”变成可回测规则的交易者另一类是手里有策略想法、缺一套数据加回测骨架的开发者。源码不是黑盒交易机器人它更像一套骨架策略、风控、执行三个位置都留了明确的改造入口改起来不用推倒重来。2. 项目骨架与技术选型从数据接口到回测引擎的一条链路这套源码的技术选型非常“务实”没有上重型框架。核心依赖就四五个tushare 负责取数pandas 负责所有的表格运算和时间序列对齐numpy 做向量化计算matplotlib 出净值曲线存储侧用 SQLite 或 Parquet 落本地文件定时任务交给系统 crontab 或者 apscheduler。整套东西在笔记本上就能跑起来不需要额外部署数据库服务这对个人开发者来说是决定性的——量化项目最容易死在“环境还没搭好就没兴趣了”。架构上它走的是经典的四层切分数据层、策略层、回测层、执行层。数据层只关心“拿到干净的行情”策略层只关心“给定数据输出信号”回测层只关心“按规则撮合并算绩效”执行层只关心“今天该买什么该卖什么”。这样切的好处是回测和实盘共用同一个策略函数避免出现“回测一套逻辑、实盘另写一套”的翻车现场——血泪经验两边代码一旦分叉对不上账就是时间问题。2.1 目录结构data / strategy / backtest / trade 四层怎么分源码的目录大致长这样压缩包解开后先对照着看一遍后面读代码会快很多quant-tushare/ ├── config/ │ ├── config.yaml # 数据源、数据库路径、日志级别 │ └── strategy.yaml # 策略参数均线周期、持仓数量、费率 ├── data/ │ ├── raw/ # 原始日线按年分 Parquet │ └── cache/ # 中间计算结果可随时删 ├── quant/ │ ├── datasource/ # 数据层只跟 tushare 打交道 │ │ ├── api.py # 接口封装、限流、重试 │ │ ├── loader.py # 拉取 复权 字段裁剪 │ │ └── storage.py # 落库、增量更新、读取 │ ├── strategy/ # 策略层纯计算不碰网络 │ │ ├── base.py # 策略基类约定输入输出 │ │ └── ma_cross.py # 双均线示例策略 │ ├── backtest/ # 回测层撮合、持仓、绩效 │ │ ├── engine.py # 主循环 │ │ ├── broker.py # 手续费、滑点、T1 │ │ └── metrics.py # 年化、回撤、夏普 │ ├── trade/ # 执行层生成每日清单 │ │ ├── signal.py │ │ └── notify.py │ └── utils/ │ ├── logger.py │ └── calendar.py # 交易日历判断是否开市 ├── scripts/ │ ├── update_data.py # 增量更新行情 │ ├── run_backtest.py # 跑回测 │ └── daily_signal.py # 生成每日信号 ├── notebooks/ # 研究用的临时脚本可删 ├── tests/ ├── requirements.txt └── README.md四层的边界是硬的datasource里不允许出现策略逻辑strategy里不允许发网络请求。这条纪律看起来刻板但它决定了后面能不能把回测结果和实盘信号对齐。我见过太多项目把取数和计算揉在一个函数里最后想加个缓存、想换个数据源牵一发动全身。notebooks/这个目录别小看源码里留它是为了研究阶段快速试因子试完把稳定的逻辑搬进strategy/临时脚本就扔在那不动。生产代码和研究代码物理隔离能省掉大量“这段到底跑不跑”的困惑。2.2 依赖与配置一份 requirements 和一个 config.yaml 说清的事依赖清单很短装的时候照着来就行# requirements.txt 的核心几项 tushare1.2.89 # 行情数据接口 pandas1.5.0 # 表格与时间序列 numpy1.23.0 matplotlib3.5.0 PyYAML6.0 # 读配置 pyarrow10.0.0 # Parquet 读写 loguru0.6.0 # 日志比 logging 少写样板代码安装就是常规两步建议单独建虚拟环境避免和系统里的 pandas 版本打架python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install -r requirements.txt配置文件是整个项目的“参数面板”token 千万不要写死在代码里提交上去# config/config.yaml datasource: token: ${TS_TOKEN} # 从环境变量读不落盘 rate_limit: 480 # 每分钟调用次数上限按账号权限调整 retry: 3 # 失败重试次数 retry_wait: 5 # 重试间隔秒数 storage: db_path: data/quant.db raw_dir: data/raw keep_years: 10 # 本地保留多少年历史 backtest: start: 20190101 end: 20241231 init_cash: 1000000 commission: 0.00025 # 佣金万2.5 stamp_tax: 0.001 # 卖出印花税千1 slippage: 0.001 # 滑点千1 max_positions: 10${TS_TOKEN}这种写法需要在加载时做一次环境变量替换源码里通常用一个load_config()函数处理。环境变量在 shell 里设一次即可export TS_TOKEN你的token字符串rate_limit这个参数是给数据层用的不同权限的账号每分钟能调的接口次数不一样写死一个偏保守的值最安全。slippage和commission直接决定回测收益的真实程度很多人回测年化 80%、实盘年化 8%八成就是这两项没设或者设得太乐观。2.3 一次完整跑批的命令链路从零到有信号三条命令按顺序走每条都是独立可重跑的# 1. 增量更新行情只补本地缺失的交易日 python scripts/update_data.py --start 20190101 --end 20241231 # 2. 跑回测输出净值曲线和绩效指标 python scripts/run_backtest.py --strategy ma_cross --config config/strategy.yaml # 3. 生成最近一个交易日的选股信号 python scripts/daily_signal.py --strategy ma_cross --date 20241231第一条命令的--start是本地水位线的兜底值实际起点由数据库里已有的最大交易日决定不会重复拉。第二条命令的--strategy对应quant/strategy/下的模块名加新策略就是加一个文件加一个同名参数。第三条命令输出的是一个 CSV字段通常是ts_code, name, signal, price, reasonreason里写清触发了哪条规则方便第二天复盘时知道当时为什么买。这三条命令的组合方式决定了项目能不能自动跑放到 crontab 里第一条每天 17:30 跑收盘后数据更新第二条每周跑一次第三条每个交易日 8:50 跑。定时任务里记得用绝对路径和虚拟环境的 python否则 crontab 里找不到模块是经典问题。3. 数据层拆解tushare 拉数、复权与增量更新数据层是整个项目最脏最累的部分也是最值得先读透的部分。策略写得再花哨数据错了全白搭。这一层的目标很明确给定一个日期区间本地库里就有对齐好的、复权正确的、没有缺口的日线数据其他模块只管读不管怎么来的。3.1 接口封装限流、重试和一处集中管理tushare 的 pro 接口是有调用频率限制的全市场几千只票一只一只拉不做限流必然被拒。源码里通常会在datasource/api.py里包一层# quant/datasource/api.py import os import time import tushare as ts from loguru import logger class TSApi: def __init__(self, config): token os.getenv(config[datasource][token].strip(${})) if not token: raise RuntimeError(TS_TOKEN 环境变量未设置) self.pro ts.pro_api(token) self.rate_limit config[datasource][rate_limit] self.retry config[datasource][retry] self.wait config[datasource][retry_wait] self._last_call 0.0 def _throttle(self): # 简单匀速限流保证两次调用间隔不小于 60/rate_limit 秒 interval 60.0 / self.rate_limit delta time.time() - self._last_call if delta interval: time.sleep(interval - delta) self._last_call time.time() def call(self, func_name, **kwargs): self._throttle() for i in range(self.retry): try: df getattr(self.pro, func_name)(**kwargs) if df is None: return None return df except Exception as e: logger.warning(f{func_name} 第{i1}次失败: {e}) time.sleep(self.wait * (i 1)) # 退避越往后等越久 logger.error(f{func_name} 重试{self.retry}次仍失败, 参数{kwargs}) return None_throttle用的是最朴素的匀速限流比令牌桶简单但对单机脚本够用。call里的重试是线性退避不是指数退避因为 tushare 的限流通常是分钟级的等太久反而拖慢整体进度。getattr(self.pro, func_name)这种写法让调用方只传字符串加新接口不用改封装层。参数上要注意rate_limit别设太满。账号标称每分钟 500 次你就写 480留一点余量给网络抖动和重试。我有次把限流关掉想快点拉完结果触发了临时封禁等了半小时才恢复这就是图快的代价。3.2 日线拉取与复权别拿不复权的数据算收益拉日线有两个坑一是字段太多全拉下来内存和磁盘都吃不消二是复权口径不复权的价格遇到除权除息会产生假跌均线策略直接失真。# quant/datasource/loader.py import pandas as pd # 只保留策略真正要用的列减少存储和内存压力 KEEP_COLS [ts_code, trade_date, open, high, low, close, pre_close, vol, amount] def fetch_daily(api, ts_code, start, end): df api.call(daily, ts_codets_code, start_datestart, end_dateend) if df is None or df.empty: return pd.DataFrame(columnsKEEP_COLS) df df[KEEP_COLS].copy() df[trade_date] pd.to_datetime(df[trade_date], format%Y%m%d) # 取复权因子按后复权对齐保证历史价格不随新的除权变化 adj api.call(adj_factor, ts_codets_code, start_datestart, end_dateend) if adj is not None and not adj.empty: adj[trade_date] pd.to_datetime(adj[trade_date], format%Y%m%d) df df.merge(adj[[trade_date, adj_factor]], ontrade_date, howleft) df[adj_factor] df[adj_factor].ffill().fillna(1.0) for col in [open, high, low, close, pre_close]: df[col] df[col] * df[adj_factor] return df.sort_values(trade_date).reset_index(dropTrue)这段代码的关键在复权因子的处理。adj_factor在某些日期可能缺失用ffill()向前填充是常见做法首日缺失则fillna(1.0)兜底。乘上因子得到的是后复权价格历史价格不会因为新的分红送股而变化回测才可复现。如果你只用close不复权遇到 10 送 10 那天价格直接腰斩均线策略会给出一个完美的假卖出信号。另一个细节是pre_close也要复权因为算涨跌幅、算涨跌停要用它。vol和amount不复权成交量金额本来就是实际值。3.3 落库与增量更新用最大交易日做水位线全市场拉一遍动辄几十分钟不可能每次重跑。增量更新的逻辑是查本地库里每只票的最大交易日从它的下一天开始拉。# quant/datasource/storage.py import sqlite3 import pandas as pd def get_last_date(conn, ts_code): cur conn.execute( SELECT MAX(trade_date) FROM daily WHERE ts_code ?, (ts_code,)) row cur.fetchone() return row[0] if row and row[0] else None def upsert_daily(conn, df): if df.empty: return 0 # 主键 (ts_code, trade_date)重复写入直接覆盖保证可重入 rows df.where(pd.notnull(df), None).values.tolist() conn.executemany( INSERT OR REPLACE INTO daily (ts_code, trade_date, open, high, low, close, pre_close, vol, amount, adj_factor) VALUES (?,?,?,?,?,?,?,?,?,?), rows) conn.commit() return len(rows)配套的建表语句CREATE TABLE IF NOT EXISTS daily ( ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, open REAL, high REAL, low REAL, close REAL, pre_close REAL, vol REAL, amount REAL, adj_factor REAL DEFAULT 1.0, PRIMARY KEY (ts_code, trade_date) ); CREATE INDEX IF NOT EXISTS idx_daily_date ON daily(trade_date);INSERT OR REPLACE让写入天然幂等同一天跑十次结果一样这对定时任务太重要了——某次跑到一半崩了重跑不会产生重复行。idx_daily_date这个索引别省按日期切片查全市场是回测里最高频的操作没索引在几十万行上能慢十倍。水位线用MAX(trade_date)而不是“上次运行时间”是因为补历史数据、手动插数据都会破坏时间假设只有数据本身说的算。我一般还会在update_data.py里加一个--force参数强制重拉最近 N 天用来修数据。3.4 数据校验缺口、停牌和异常值拉完不校验等于没拉。最常检查的三件事# quant/datasource/loader.py 追加 def check_daily(df, calendar): issues [] # 1. 缺口交易日历里有、数据里没有的日期 have set(df[trade_date]) expect set(calendar[(calendar df[trade_date].min()) (calendar df[trade_date].max())]) missing sorted(expect - have) if missing: issues.append(f缺口 {len(missing)} 天首个 {missing[0]}) # 2. 异常涨跌复权后单日涨跌超过 21% 基本可判定为脏数据 pct df[close].pct_change() abnormal df[pct.abs() 0.21] if not abnormal.empty: issues.append(f异常波动 {len(abnormal)} 条) # 3. 零成交停牌日 vol 为 0回测里不能当成可交易日 zero df[df[vol] 0] if not zero.empty: issues.append(f零成交 {len(zero)} 天) return issues缺口要区分两种情况全天停牌导致的无数据是正常的但如果是数据接口漏了就必须补。常见做法是拿交易日历对比缺口天数超过阈值比如连续 3 天就记日志人工看。异常涨跌在复权后超过 21% 基本不可能是真实行情多半是复权因子缺失或者单位错乱。零成交日在选股时要直接过滤掉不然回测会在一个根本买不进的票上成交收益全是假的。这三项检查的输出写进日志文件每天跑完扫一眼比事后发现回测结果离谱再回头查要省事得多。4. 策略与回测内核从信号到净值曲线的那几行关键代码策略和回测是这套源码里最“像量化”的部分但代码量其实不大。真正决定成败的是接口约定和撮合细节而不是策略本身有多复杂。4.1 策略基类一个方法约定输入输出所有策略继承同一个基类只实现generate_signal输入是行情 DataFrame输出是同索引的信号序列# quant/strategy/base.py from abc import ABC, abstractmethod import pandas as pd class BaseStrategy(ABC): def __init__(self, params: dict): self.params params abstractmethod def generate_signal(self, df: pd.DataFrame) - pd.Series: 返回 -1/0/1 组成的 Seriesindex 与 df 对齐 -1 卖出0 持有不动1 买入 raise NotImplementedError def prepare(self, df: pd.DataFrame) - pd.DataFrame: # 预留钩子算因子、加指标子类可覆盖 return df这个约定的价值在于回测引擎和实盘信号生成都调generate_signal同一份代码同一个输入结果必然一致。prepare是给因子计算留的把耗时计算和信号生成分开方便缓存中间结果。params从strategy.yaml读改参数不改代码。4.2 双均线策略二十行写完一个能跑的示例# quant/strategy/ma_cross.py from .base import BaseStrategy class MaCross(BaseStrategy): def prepare(self, df): fast self.params.get(fast, 5) slow self.params.get(slow, 20) df[ma_fast] df[close].rolling(fast).mean() df[ma_slow] df[close].rolling(slow).mean() return df def generate_signal(self, df): sig 0 signals [] for fast, slow in zip(df[ma_fast], df[ma_slow]): if fast slow: sig 1 elif fast slow: sig -1 signals.append(sig) return __import__(pandas).Series(signals, indexdf.index)逻辑很直白快线上穿慢线给买入信号下穿给卖出信号。rolling(fast).mean()里的fast和slow从配置读改周期不用动代码。信号用 0/1/-1 三态而不是布尔值是为了给后面加“清仓”“减半仓”留扩展空间。注意prepare里的均线是在复权后的close上算的如果哪天有人把复权逻辑改坏了这个策略会第一时间表现为“信号乱跳”而不是收益慢慢变差这种敏感性在调试时反而好用。4.3 回测引擎手续费、滑点、T1 一个都不能省回测主循环的核心是“按信号撮合、按规则扣费”# quant/backtest/broker.py class Broker: def __init__(self, cash, commission, stamp_tax, slippage): self.cash cash self.commission commission self.stamp_tax stamp_tax self.slippage slippage def buy(self, price, shares): # 买入按滑点上浮佣金双边收 exec_price price * (1 self.slippage) cost exec_price * shares fee max(cost * self.commission, 5) # 最低5元 if cost fee self.cash: return None self.cash - (cost fee) return {price: exec_price, shares: shares, fee: fee} def sell(self, price, shares): # 卖出按滑点下浮多收一笔印花税 exec_price price * (1 - self.slippage) amount exec_price * shares fee amount * self.commission amount * self.stamp_tax self.cash (amount - fee) return {price: exec_price, shares: shares, fee: fee}buy里的max(cost * commission, 5)是 A 股真实规则很多简化回测直接按比例算小资金账户会明显高估收益。sell多收的印花税只在卖出方向收这是 A 股和很多海外市场不一样的地方漏掉就是系统性高估。T1 的处理不在Broker里而在引擎主循环的持仓状态上当天买入的票记一个buy_date卖出前检查buy_date today。涨跌停的处理是另一个细节开盘一字板买不进、跌停卖不出严格回测里需要判断开盘价是否等于涨停价不严格的话至少要在日志里标注这笔成交是“理想成交”。回测引擎的主循环大致是按交易日遍历每天先更新持仓市值再调用策略拿信号然后按信号调仓最后记录净值。净值序列是后面所有绩效指标的输入也是画曲线的数据源。4.4 绩效指标年化、回撤、夏普的口径要写清# quant/backtest/metrics.py import numpy as np def performance(nav, periods_per_year252): ret nav.pct_change().dropna() total nav.iloc[-1] / nav.iloc[0] - 1 years len(nav) / periods_per_year annual (1 total) ** (1 / years) - 1 if years 0 else 0 drawdown (nav / nav.cummax() - 1).min() sharpe (ret.mean() / ret.std() * np.sqrt(periods_per_year) if ret.std() 0 else 0) return {total_return: total, annual_return: annual, max_drawdown: drawdown, sharpe: sharpe}年化用的复合口径而不是简单年化years用交易日数除以 252 估算比按自然日算更贴近实际交易节奏。最大回撤用净值除以历史新高再取最小值负值表示回撤幅度。夏普用日收益均值除标准差再乘根号 252 年化无风险利率没扣简单策略这么算够用如果拿去和别人比记得先确认对方扣没扣无风险利率口径不一致比出来的排名毫无意义。指标里最容易被忽略的是total_return和annual_return的区别短周期回测比如三个月年化后会变成一个夸张的数字评审时只看年化容易自欺欺人。5. 避坑与排查跑不通、对不上、跑得慢的三类现场这套源码真正让人花时间的不是写策略是排查。下面五条都是实际跑的时候大概率会撞上的按“现象 → 原因 → 解决”记下来能省不少来回。5.1 现象拉数报权限不足或频繁超时原因tushare 的接口按积分和权限分级某些接口需要更高权限另一个常见原因是每分钟调用超限尤其在全市场循环拉取时rate_limit设得太满或者根本没设。解决先单独调一次目标接口确认权限报抱歉您没有访问该接口的权限就是权限问题换接口或用已有权限的字段自己算。超时的话把rate_limit降到标称值的 80%重试间隔从 5 秒起线性递增。全市场拉取建议分批每批 500 只批与批之间 sleep 几秒别一口气冲到底。5.2 现象回测年化高得离谱实盘完全不是一个量级原因九成是未来函数或者成本没算。未来函数最常见的是用了当天收盘价生成信号、又用当天收盘价成交成本漏算则是没设手续费、滑点、印花税或者设了但方向搞反。解决信号在 T 日生成成交必须放在 T1 日的开盘价或均价这条规则写进引擎主循环不给策略留“当天买当天卖”的口子。成本检查用一个最小例子验证单笔 10 万买入佣金应该是100000 * 0.00025 25元如果算出来是 0说明配置没生效。滑点至少给千分之一做小盘股的给千二。5.3 现象回测信号和每日清单对不上原因两边用的数据版本不同。回测跑的是三个月前拉的库每日信号用的是今天刚更新的库中间的除权除息让复权价格变了均线自然不一样。还有一种可能是prepare里的窗口长度依赖历史长度回测给足了 250 天实盘只取了 60 天。解决每日信号生成前强制跑一次增量更新保证库是最新的。prepare里的滚动窗口统一从配置读并且在取数时多取slow * 2天的前置数据算完再截断到目标区间。诊断方法很简单把同一天的回测信号和实盘信号打出来做 diff哪只票不一致就单独查它的数据。5.4 现象跑全市场回测慢到无法接受原因逐票循环里每次都在查库几千只票就是几千次查询或者每次都用pd.read_sql拉全表再过滤内存和 IO 双重浪费。解决一次性把区间内的全市场数据读进内存用groupby(ts_code)分组处理别在循环里查库。数据量大的话按年切片一年一个 Parquet 文件读的时候只读需要的年份。指标计算尽量向量化rolling().mean()比 for 循环快两个数量级实在要循环就用numba或者干脆接受慢、放到晚上跑。5.5 现象日期格式错乱、时区偏移、切片切出空结果原因tushare 返回的trade_date是YYYYMMDD字符串数据库里存成 TEXTpandas 里又变成 datetime三种形态混用比较的时候静默返回空。时区问题多见于用了datetime.now()又做日期比较的场景。解决约定一条铁律——进数据库之前全部转成YYYY-MM-DD字符串读出后立刻pd.to_datetime中间不保留其他形态。日期比较统一用 pandas 的 datetime 类型不用字符串比大小。涉及“今天”的逻辑统一走交易日历模块别直接datetime.today()周末和节假日跑出来的结果会让你怀疑人生。6. 二次开发把单策略脚本改成多策略框架的三个改造点源码跑通之后大部分人的下一步是加自己的策略。这里给三个改造方向按投入产出比排序。第一个改造点是策略注册。现在加策略要改run_backtest.py里的分支判断改成用装饰器自动注册# quant/strategy/registry.py STRATEGIES {} def register(name): def wrapper(cls): STRATEGIES[name] cls return cls return wrapper # 使用在策略文件顶部加一行 register(ma_cross) class MaCross(BaseStrategy): ...这样--strategy参数直接查字典加策略只加文件不动入口脚本。第二个改造点是参数扫描。单策略跑一次只能看一组参数改成网格扫描后输出一张表改造项单策略现状多策略/多参数目标参数来源strategy.yaml 单份参数网格生成器结果存储终端打印SQLite 结果表评估方式看年化年化 回撤 夏普综合排序复现方式手动改配置参数哈希做 run_id结果表至少存run_id, strategy, params_json, annual, drawdown, sharpe, start, end同一个run_id能复现出完全一样的净值曲线。参数扫描最容易犯的错是拿全样本调参再报全样本收益正确做法是留一段样本外调参只用样本内。第三个改造点是执行层解耦。daily_signal.py现在直接打印 CSV改成输出到一个signals表再写一个独立的推送模块读表发消息。这样换推送渠道、加人工确认环节都不用碰信号生成逻辑。风控也可以挂在这一层单票最大仓位、行业集中度、当日最大回撤熔断全部在生成清单之后、推送之前过滤一遍。验证改造有没有改坏方法很土但有效改造前后用同一份数据、同一组参数跑一次把净值曲线叠在一起看差异超过千分之一就要查。我吃过一次亏改注册机制的时候不小心把默认参数从 20 改成了 10回测收益直接翻倍差点当成策略突破。从那以后我每次动框架代码都强制先跑一遍“同一天、同一只票”的信号 diff确认输出没变化再谈收益。希望帮到你。本文还有配套的精品资源点击获取