简介这份资源是tushare 0.2.8版本的Python源码包面向金融数据分析与量化交易方向的开发者、学生及研究者帮助读者通过阅读真实库源码理解金融数据接口的设计与实现。压缩包共26个文件以19个py源码文件为核心辅以3个txt说明、2个pkg-info元数据、1个csv数据样例和1个cfg配置文件整体约181KB体积轻量便于本地研读。源码覆盖数据获取、清洗转换、本地存储、实时行情轮询、技术指标计算、模拟交易接口及网络与数据异常处理等模块可帮助读者掌握HTTP请求封装、数据对象构建与工程化异常处理的思路。目前已有6234人学习下载适合希望从调用API进阶到理解底层实现、提升Python工程能力并积累量化策略开发经验的读者参考。1. 从一份 tushare 量化交易 Python 源码说起数据、策略、回测到底怎么串起来很多人第一次接触量化交易卡住的地方不是策略逻辑而是数据从哪来、怎么存、怎么喂给回测框架。这份 tushare 量化交易 Python 源码解决的就是这条链路用 tushare 拉 A 股日线、财务指标、指数成分等数据落到本地做缓存再驱动一个轻量回测引擎跑均线、动量、轮动这类经典策略。它适合两类人一是刚学完 pandas 想找个完整项目练手的开发者二是手里有策略想法、但不想从零搭数据管道的熟手。源码结构不复杂但把「取数 → 清洗 → 信号 → 回测 → 绩效」这条主线走通了拿过来改策略比看十篇教程都直接。2. 项目骨架与技术选型为什么是 tushare pandas 本地缓存2.1 目录结构与模块职责拿到源码包先别急着跑main.py。我一般会先花五分钟把目录扫一遍搞清楚每个文件在链路里的位置。这个项目的典型结构大致是这样tushare_quant/ ├── config/ │ └── settings.py # token、路径、回测参数 ├── data/ │ ├── fetch.py # tushare 取数封装 │ ├── cache.py # 本地 parquet/csv 缓存读写 │ └── clean.py # 复权、缺失值、对齐处理 ├── strategy/ │ ├── base.py # 策略基类定义 generate_signal │ ├── ma_cross.py # 双均线 │ └── momentum.py # 动量轮动 ├── backtest/ │ ├── engine.py # 逐日循环、撮合、持仓 │ └── metrics.py # 年化、回撤、夏普 ├── utils/ │ └── logger.py ├── main.py # 入口取数 → 回测 → 输出 └── requirements.txt这个划分的好处是职责边界清楚data只管把数据弄干净strategy只管出信号backtest只管按信号算钱。你改策略时基本不用碰取数和回测代码改数据源时也不用动策略。常见做法是把config/settings.py里的 token 和路径抽成环境变量避免提交到仓库。2.2 技术栈与依赖说明核心依赖就四个tushare负责数据pandas负责表格运算numpy负责数值计算pyarrow负责 parquet 缓存。回测引擎没有用 backtrader、vnpy 这类重型框架而是自己写了一个逐日循环原因是这类教学向项目需要让读者看清每一笔成交怎么来的用现成框架反而把逻辑藏进了黑匣子。requirements.txt里通常长这样tushare1.2.89 pandas1.5.0 numpy1.23.0 pyarrow10.0.0 matplotlib3.6.0版本不用卡太死但 pandas 2.x 和 1.x 在append、resample上有差异如果源码是按 1.x 写的升到 2.x 要留意FutureWarning。我一般会先建虚拟环境再装依赖避免和系统里的包打架。2.3 数据获取层tushare 接口封装与 token 配置data/fetch.py是整个项目的入口水源。它做的事很朴素把 tushare 的pro_api包一层加上重试和限频。# data/fetch.py import tushare as ts import time from config.settings import TUSHARE_TOKEN class DataFetcher: def __init__(self): # 初始化 pro 接口token 从配置读取 ts.set_token(TUSHARE_TOKEN) self.pro ts.pro_api() def get_daily(self, ts_code, start_date, end_date, retry3): 拉取单只股票日线带重试 for i in range(retry): try: df self.pro.daily( ts_codets_code, start_datestart_date, end_dateend_date ) # tushare 返回按日期倒序这里统一成正序 return df.sort_values(trade_date).reset_index(dropTrue) except Exception as e: if i retry - 1: raise time.sleep(1) # 简单退避避免触发限频这里有几个参数要盯住ts_code是带交易所后缀的代码比如000001.SZstart_date和end_date格式是YYYYMMDD字符串不是 datetime。tushare 对积分和调用频率有要求免费额度下日线接口每分钟能调的次数有限所以retry和sleep不是可有可无的装饰是防止批量拉取时中途断掉的后悔药。如果你要拉全市场几千只股票建议按交易日分批而不是按股票循环。2.4 本地缓存设计parquet 落盘与增量更新每次回测都去调接口既慢又容易触发限频。data/cache.py的思路是第一次全量拉之后只补增量。# data/cache.py import os import pandas as pd CACHE_DIR data/cache def save_cache(df, ts_code): 按股票代码存 parquet os.makedirs(CACHE_DIR, exist_okTrue) path os.path.join(CACHE_DIR, f{ts_code}.parquet) df.to_parquet(path, indexFalse) def load_cache(ts_code): 读取本地缓存没有则返回 None path os.path.join(CACHE_DIR, f{ts_code}.parquet) if not os.path.exists(path): return None return pd.read_parquet(path) def update_cache(fetcher, ts_code, start_date, end_date): 增量更新只拉缓存里最新日期之后的数据 old load_cache(ts_code) if old is not None and len(old) 0: last_date old[trade_date].max() if last_date end_date: return old # 已是最新直接返回 new fetcher.get_daily(ts_code, last_date, end_date) # 去掉重复日期再拼接 merged pd.concat([old, new]).drop_duplicates(trade_date) merged merged.sort_values(trade_date).reset_index(dropTrue) else: merged fetcher.get_daily(ts_code, start_date, end_date) save_cache(merged, ts_code) return merged用 parquet 而不是 csv是因为它列式存储、带类型、读得快几千只股票的量级下差距很明显。drop_duplicates(trade_date)这行是防止增量拉取时把边界日期重复算进去。注意last_date直接当start_date传给接口tushare 是闭区间所以会重复拉一天靠去重兜住这是常见做法。3. 策略与回测引擎信号怎么生成、钱怎么算3.1 策略基类统一 generate_signal 接口strategy/base.py定义了所有策略的契约。不管你是均线、动量还是别的只要实现generate_signal回测引擎就能跑。# strategy/base.py import pandas as pd class BaseStrategy: def __init__(self, paramsNone): self.params params or {} def generate_signal(self, df): 输入单只股票或多只股票的行情 DataFrame 输出与 df 等长的 signal 列1 表示持有0 表示空仓 raise NotImplementedError def prepare(self, df): 可选计算指标默认原样返回 return df这个设计的关键是信号只有 0 和 1不做仓位管理。对教学项目来说够用也避免了「信号和仓位混在一起」的常见混乱。如果你要加多档仓位把 signal 改成 0 到 1 的浮点数即可引擎那边乘一下就行。3.2 双均线策略实现从指标到信号strategy/ma_cross.py是最经典的入门策略短均线上穿长均线买入下穿卖出。# strategy/ma_cross.py import pandas as pd from strategy.base import BaseStrategy class MACrossStrategy(BaseStrategy): def __init__(self, short5, long20): super().__init__({short: short, long: long}) def prepare(self, df): # 用收盘价算均线注意 min_periods 避免前期全是 NaN df[ma_short] df[close].rolling( self.params[short], min_periodsself.params[short] ).mean() df[ma_long] df[close].rolling( self.params[long], min_periodsself.params[long] ).mean() return df def generate_signal(self, df): df self.prepare(df) # 短均线在长均线上方则持有 df[signal] (df[ma_short] df[ma_long]).astype(int) # 均线未形成前不交易 df.loc[df[ma_long].isna(), signal] 0 return dfmin_periods设成窗口长度是为了让前 N-1 天明确是 NaN 而不是用不完整窗口算出的假值。signal用astype(int)把布尔转成 0/1。最后一行把长均线还没形成的区间强制置 0否则回测会在数据不足时凭空开仓这是新手最容易翻车的地方之一。3.3 回测引擎逐日循环与撮合逻辑backtest/engine.py是项目的核心。它按交易日推进每天根据信号决定买还是卖并记录净值和持仓。# backtest/engine.py import pandas as pd import numpy as np class BacktestEngine: def __init__(self, init_cash100000, fee_rate0.0003): self.init_cash init_cash self.fee_rate fee_rate # 单边手续费 def run(self, df): df 需包含 trade_date, open, close, signal cash self.init_cash position 0 # 持股数量 records [] for i in range(1, len(df)): row df.iloc[i] prev_signal df.iloc[i - 1][signal] today_signal row[signal] # 信号变化时以今日开盘价成交 if today_signal 1 and prev_signal 0 and position 0: price row[open] # 留出手续费能买多少买多少 buy_amount int(cash / (price * (1 self.fee_rate)) / 100) * 100 if buy_amount 0: cost buy_amount * price fee cost * self.fee_rate cash - (cost fee) position buy_amount elif today_signal 0 and prev_signal 1 and position 0: price row[open] revenue position * price fee revenue * self.fee_rate cash (revenue - fee) position 0 # 每日收盘后记录总资产 market_value position * row[close] records.append({ trade_date: row[trade_date], cash: cash, position: position, total: cash market_value }) return pd.DataFrame(records)几个关键点成交价用次日开盘而不是当日收盘避免用未来数据买入数量按 100 股取整符合 A 股规则手续费单边收买卖各一次。prev_signal和today_signal的比较是为了只在信号翻转时动手不然每天都会重复买卖。这个引擎没有处理停牌、涨跌停属于简化版但逻辑透明适合拿来改。3.4 绩效指标年化、最大回撤、夏普backtest/metrics.py把净值序列转成几个能看的数字。# backtest/metrics.py import numpy as np import pandas as pd def annual_return(total_series, periods_per_year252): 年化收益率 total total_series.values n len(total) if n 2: return 0.0 return (total[-1] / total[0]) ** (periods_per_year / n) - 1 def max_drawdown(total_series): 最大回撤 total total_series.values peak np.maximum.accumulate(total) dd (total - peak) / peak return dd.min() def sharpe_ratio(total_series, rf0.0, periods_per_year252): 夏普比率rf 为无风险利率 ret pd.Series(total_series).pct_change().dropna() if ret.std() 0: return 0.0 excess ret - rf / periods_per_year return np.sqrt(periods_per_year) * excess.mean() / ret.std()periods_per_year252是 A 股一年交易日的大致数量。max_drawdown用累积最大值算峰值再求相对回撤的最小值返回的是负数。夏普里rf默认 0实际用的时候可以填 0.02 左右。这三个指标配合看单看年化容易被高波动策略骗。4. 跑起来环境、配置、执行与结果验证4.1 环境准备与依赖安装先确认 Python 版本建议 3.9 到 3.11太新的版本有些包还没跟上。python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install -r requirements.txt装完可以pip list看一眼 tushare 和 pandas 的版本确认没装串。如果公司网络对 pip 有限制提前配好镜像源不然卡在下载上很浪费时间。4.2 token 与参数配置config/settings.py里需要填 tushare token。没有的话去 tushare 官网注册一个免费账号有基础积分日线接口够用。# config/settings.py import os TUSHARE_TOKEN os.getenv(TUSHARE_TOKEN, 你的token) START_DATE 20200101 END_DATE 20231231 STOCK_POOL [000001.SZ, 600519.SH, 000858.SZ] INIT_CASH 100000把 token 放环境变量是习惯避免误提交。STOCK_POOL先放几只票试水别一上来就全市场拉数据能等到怀疑人生。4.3 执行回测与结果解读main.py把取数、策略、回测串起来。# main.py from data.fetch import DataFetcher from data.cache import update_cache from strategy.ma_cross import MACrossStrategy from backtest.engine import BacktestEngine from backtest.metrics import annual_return, max_drawdown, sharpe_ratio from config.settings import START_DATE, END_DATE, STOCK_POOL, INIT_CASH fetcher DataFetcher() strategy MACrossStrategy(short5, long20) engine BacktestEngine(init_cashINIT_CASH) for code in STOCK_POOL: df update_cache(fetcher, code, START_DATE, END_DATE) df strategy.generate_signal(df) result engine.run(df) print(code, 年化, round(annual_return(result[total]), 4), 回撤, round(max_drawdown(result[total]), 4), 夏普, round(sharpe_ratio(result[total]), 4))跑完你会看到每只票的三个指标。如果年化高得离谱、回撤又小先别高兴大概率是用了未来数据或者没扣手续费。正常双均线在震荡市里回撤不会太好看这才是真实的样子。4.4 结果可视化与常见输出源码里通常带一个画净值曲线的函数用 matplotlib 把result[total]画出来再叠加买卖点。看曲线比看数字直观如果净值在某段时间横着不动说明策略空仓如果回撤深且恢复慢说明参数太激进。我一般会把回测结果存成 csv方便和不同参数对比。5. 避坑与常见问题那些跑不通和跑偏的地方5.1 现象接口报权限或积分不足 → 原因token 无效或积分不够 → 解决检查 token、降低调用频率tushare 的日线接口对积分有门槛新账号可能调不了某些接口。先确认 token 填对再确认账号积分。批量拉取时加time.sleep别连续猛调。5.2 现象回测收益异常高 → 原因用了未来数据或未扣费 → 解决成交价用次日开盘扣双边手续费最常见的翻车是信号当天用当天收盘价成交等于偷看了未来。引擎里已经用prev_signal和次日开盘规避如果你自己改策略务必保持这个时序。手续费别省双边加起来千分之几长期影响不小。5.3 现象均线前期全是 NaN 导致不开仓 → 原因min_periods 设置不当 → 解决显式置 0 并检查窗口rolling默认min_periods1会用不完整窗口算出假均线。设成窗口长度后前期是 NaN比较结果是 Falsesignal 自然是 0这是对的。但如果你手动填了值就可能凭空开仓。5.4 现象增量更新后数据重复 → 原因闭区间重复拉取 → 解决drop_duplicates 按日期去重last_date当start_date会重复拉一天靠drop_duplicates(trade_date)兜住。如果哪天去重逻辑被删了回测里会出现同一天两条记录净值直接算错。5.5 现象pandas 版本升级后报错 → 原因API 变更 → 解决锁定版本或改写法pandas 2.x 去掉了appendresample的closed参数也有变化。如果源码是 1.x 写的要么锁版本要么按警告逐条改。我一般先在虚拟环境里锁版本跑通再考虑升级。6. 二次开发换策略、换数据、加风控的改造思路拿到这份源码真正的价值在于改。我一般从三个方向下手。第一换策略。strategy/base.py的接口很干净你新建一个rsi_reversal.py继承BaseStrategy在prepare里算 RSI在generate_signal里出 0/1 信号然后在main.py里把MACrossStrategy换掉就行。不用动引擎和取数。改完先跑单只票确认信号数量合理再上股票池。第二换数据。现在只用了日线如果你想加财务因子在data/fetch.py里加一个get_fina_indicator方法注意财务数据是按报告期发布的要按公告日对齐到交易日否则又是未来数据。对齐这步最容易出错建议单独写个测试打印几个日期的因子值人工核对。第三加风控。引擎现在是满仓进满仓出你可以在run里加止损持仓期间如果浮亏超过阈值强制置 signal 为 0。改的时候注意止损判断要用当日收盘价卖出仍用次日开盘保持时序一致。验证改造是否有效我习惯用同一段区间、同一组参数跑新旧两版对比年化、回撤、夏普和交易次数。如果新版交易次数暴增但收益没提升多半是信号抖动加个平滑或最小持有期。参数优化别过度网格搜出来的最优参数往往在样本外就失效这是血泪经验。从那以后我每次改完策略都强制走一遍「单票 → 小池 → 全池」的流程并且把每次回测的配置和结果存成带时间戳的文件不然过两天就忘了哪版是哪版。希望帮到你。本文还有配套的精品资源点击获取