简介压缩包内含72个Python脚本共44KB是一套面向量化交易入门与进阶者的源码集合。内容按章节组织覆盖金融数据处理、技术指标计算如MA、RSI、布林带、选股与成长模型、回测逻辑、交易API调用、风险控制及并发处理等常见模块适合希望理解量化策略代码实现、快速上手实战的Python开发者。源码目录结构清晰从基础的数据清洗到完整的选股模型与信号检测均有涉及可直接运行或二次修改也可作为编写自定义策略的参考蓝本。通过阅读这些代码可以学习如何用Pandas处理行情数据、如何设计回测流程、如何对接数据接口并理解从数据获取到策略落地的完整链路。已有6612人学习下载对于想系统学习Python量化编程、减少重复造轮子的读者这份源码能提供直观的代码参考和功能模块拆解帮助在实际项目中快速搭建量化研究环境。1. 量化交易源码包不是跑通就完事能落地的代码才值得你下载先说一句可能不太好听的大实话大多数从网上下载“Python量化交易-源码.rar”这类压缩包的人跑完一次回测看到漂亮的净值曲线就把代码丢进硬盘再也没打开过。原因不是代码写得不好而是包里缺一条从数据到策略再到下单信号的完整闭环。这个源码包的价值恰好在于把三层都串起来了——数据层用 tushare 和 akshare 双通道获取行情策略层实现了双均线、动量、网格三类经典模型执行层有信号落库与模拟盘推送的代码。你要做的不再是从零搭框架而是改参数、换标的、加过滤条件。适合谁用用 Excel 看盘多年、想转到 Python 做自动化交易的从业者以及需要一份能复现的量化课程设计源码的金融或计算机相关专业学生。2. 量化交易源码包的三层结构数据、策略、执行各改哪里下载解压之后第一件事不是双击运行而是先把目录结构读明白。量化交易的核心无非三件事数据能不能稳定拿到、策略逻辑对不对、信号能不能送到交易端。这个源码包也按这三层组织。按我的习惯拿到任何源码包都会先画一张模块调用关系图再逐个文件读注释最后才动手跑。2.1 目录结构与文件清单下载后先读这几处常见的压缩包结构像这样不同版本文件名略有差异但分层逻辑一致Python量化交易/ ├── config/ │ ├── config.yaml # 全局参数标的池、周期、资金 │ └── strategy_params.py # 策略参数独立模块 ├── data/ │ ├── data_loader.py # 行情数据统一加载入口 │ ├── tushare_worker.py # tushare 接口封装 │ └── akshare_worker.py # akshare 接口封装 ├── strategy/ │ ├── base_strategy.py # 策略基类定义统一接口 │ ├── dual_ma.py # 双均线策略 │ ├── momentum.py # 动量策略 │ └── grid.py # 网格策略 ├── backtest/ │ ├── backtest_engine.py # 向量化回测主逻辑 │ └── metrics.py # 绩效指标计算 ├── execution/ │ ├── signal_manager.py # 信号生成与落库 │ └── order_utils.py # 模拟下单工具 ├── utils/ │ └── logger_config.py # 日志与错误处理 ├── run_backtest.py # 回测入口 └── requirements.txt # 依赖清单打开requirements.txt看依赖版本这个包锁定的核心依赖是pandas1.3.0、numpy1.21.0、PyYAML5.4、tushare1.2.89、akshare1.10.0。如果你电脑里已经装过backtrader或vnpy这类框架注意别混着跑这个包的回测引擎是自实现的向量化版本和backtrader的事件驱动逻辑不同混用容易在信号数量上对不上账。2.2 数据层tushare 与 akshare 接口封装怎么改数据层是整套源码包与外部世界的连接器。data_loader.py提供统一入口先读配置文件里选定的数据源再决定调用哪个 worker。# data/data_loader.py 核心逻辑简化版 import pandas as pd from utils.logger_config import logger def load_daily_bars(source: str auto, symbol: str 600519.SH, start: str 20200101, end: str 20241231): 加载日线行情 :param source: auto 表示自动选择可用数据源也可指定 tushare / akshare :param symbol: 标的代码注意 .SH / .SZ 后缀不能省略 :param start: 起始日期格式 YYYYMMDD :param end: 结束日期 :return: DataFrame列名统一为 open/high/low/close/volume if source auto: try: from data.tushare_worker import fetch_daily df fetch_daily(symbol, start, end) except Exception as exc: logger.warning(ftushare 拉取失败: {exc}切换到 akshare) from data.akshare_worker import fetch_daily df fetch_daily(symbol, start, end) elif source tushare: from data.tushare_worker import fetch_daily df fetch_daily(symbol, start, end) else: from data.akshare_worker import fetch_daily df fetch_daily(symbol, start, end) df.rename(columns{trade_date: date, vol: volume}, inplaceTrue) df[date] pd.to_datetime(df[date]) df.sort_values(date, inplaceTrue) df.drop_duplicates(subsetdate, keeplast, inplaceTrue) return df这里有几个参数值得单独说。symbol必须带交易所后缀tushare 和 akshare 对代码格式的要求不一致统一在 worker 层做清洗。start和end直接透传给接口大概率会遇到“返回列名和文档不一致”的情况所以rename那一步是必须的。drop_duplicates很关键数据源偶尔会把同一根K线重复推送不去重的话后面算均线会出现一天两根K线信号数量直接翻倍。2.3 策略层双均线与动量策略的触发逻辑base_strategy.py定义了统一接口核心方法是generate_signal输入日线 DataFrame输出一个 0/1 序列1 表示持仓、0 表示空仓。双均线策略实现如下# strategy/dual_ma.py —— 双均线策略 import talib from strategy.base_strategy import BaseStrategy class DualMAStrategy(BaseStrategy): def __init__(self, fast: int 5, slow: int 20): self.fast fast self.slow slow def generate_signal(self, df: pd.DataFrame) - pd.Series: df df.copy() df[ma_fast] talib.SMA(df[close].values, timeperiodself.fast) df[ma_slow] talib.SMA(df[close].values, timeperiodself.slow) df[ma_fast].fillna(0, inplaceTrue) df[ma_slow].fillna(0, inplaceTrue) signal (df[ma_fast] df[ma_slow]).astype(int) return signal参数fast和slow是双均线的周期窗口源码包默认 5 和 20这是日线级别最常见的组合之一。用talib.SMA而不是 pandas 的rolling().mean()是因为 talib 在长序列计算上性能更稳定。astype(int)这一步不能省信号序列后续要跟收益率序列相乘如果不转数值类型pandas 会把布尔值参与运算产生隐式类型转换的警告结果虽然不变但会埋下隐患。动量策略在momentum.py里核心是close / close.shift(N) - 1得到过去 N 日收益率再与阈值比较产生入场信号。源码包默认 N20、阈值 0.05属于高波动高收益类型回测收益和回撤同时放大适合风险承受能力强的用户。3. 单标的改多标的轮动排序逻辑与回测指标怎么读源码包默认跑单标的但实际交易中单标的资金曲线平滑度太差。拿到包之后建议先改造成多标的用的逻辑这也是把“能跑的 demo”变成“能用的策略”的关键转折点。3.1 参数配置文件怎么设计才不翻车config/config.yaml里值得优先改的参数如下# config/config.yaml trading: initial_capital: 1000000 # 初始资金单位元 commission_rate: 0.0003 # 佣金费率万三 slippage: 0.001 # 滑点比例成交价加 0.1% universe: - 600519.SH - 000858.SZ - 300750.SZ strategy: dual_ma params: fast: 5 slow: 20universe改成你要跟踪的标的池建议从 3 个标的起步熟练后再扩到 10 个以内。标的太多时向量化回测的矩阵拼接会占用大量内存。commission_rate要按自己开户券商的真实费率填默认万三是保守值。slippage是最容易被低估的参数我见过很多回测漂亮的策略把滑点从 0 调到 0.1% 之后年化直接掉一半这不是源码包的 bug而是行情深度和打板速度的真实损耗。3.2 轮动策略的排序与换仓逻辑多标的轮动实现方式有很多最常见的是动量轮动每 5 个交易日排序一次持有涨幅最高的标的。# run_backtest.py 轮动改造节选 import pandas as pd from data.data_loader import load_daily_bars from strategy.momentum import MomentumStrategy universes [600519.SH, 000858.SZ, 300750.SZ] rebalance_days 5 mom MomentumStrategy(window20, threshold0.0) bars {} for symbol in universes: bars[symbol] load_daily_bars(sourceauto, symbolsymbol, start20200101, end20241231) panel pd.concat(bars, axis1) # 列名变成 (symbol, field) for i in range(rebalance_days, len(panel), rebalance_days): window panel.iloc[i - rebalance_days: i] momentums {} for symbol in universes: close window[symbol, close] momentum close.iloc[-1] / close.iloc[0] - 1 momentums[symbol] momentum best_symbol max(momentums, keymomentums.get) print(f第 {i} 个交易日换仓持仓标的: {best_symbol})这段代码展示核心排序逻辑。真正的回测还要把换仓信号转换成持仓权重序列再与各标的日收益率相乘得到组合收益。注意pd.concat(bars, axis1)之后列名变成二层 MultiIndex取数必须写panel[symbol, close]而不是panel[close]新手最容易在这里翻车报错看起来迷惑实际就是列索引层级没写对。3.3 回测结果怎么看年化、回撤、夏普的边界backtest/metrics.py已实现常用指标直接调用from backtest.metrics import ( calc_annual_return, calc_max_drawdown, calc_sharpe_ratio, ) def print_metrics(nav_series: pd.Series): ann_ret calc_annual_return(nav_series, periods_per_year252) mdd calc_max_drawdown(nav_series) sharpe calc_sharpe_ratio(nav_series, risk_free_rate0.03) print(f年化收益: {ann_ret:.2%}) print(f最大回撤: {mdd:.2%}) print(f夏普比率: {sharpe:.2f})periods_per_year252对应A股年交易日数如果标的是港股或加密货币改成 250 或 365。risk_free_rate0.03近似无风险利率不同年份可微调。三个指标里最值得盯的是最大回撤这个源码包的默认策略在沪深 300 成分股上回测最大回撤通常在 15% 到 25% 之间。如果某次改动参数后年化收益明显上升而回撤几乎不变先别高兴大概率是数据里掺了未来函数这个在第 5 章详细展开。4. 信号落库与订单状态机模拟盘阶段的关键改造回测跑通只完成了四分之一的工程。真正让源码包产生价值的是执行层——信号生成后怎么保存、怎么防止断线丢单、怎么留下可审计的交易记录。4.1 订单状态机与断线重连机制execution/order_utils.py里封装了一个简化版订单状态机# execution/order_utils.py —— 订单状态机简化版 from enum import Enum class OrderStatus(Enum): PENDING pending # 待发送 SUBMITTED submitted # 已提交 FILLED filled # 全部成交 CANCELED canceled # 已撤销 REJECTED rejected # 被拒 class Order: def __init__(self, symbol: str, side: str, quantity: int, price: float): self.symbol symbol self.side side # buy / sell self.quantity quantity self.price price self.status OrderStatus.PENDING def submit(self): if self.status ! OrderStatus.PENDING: raise ValueError(f非法状态迁移: {self.status}) self.status OrderStatus.SUBMITTED def mark_filled(self): self.status OrderStatus.FILLED订单状态机的价值在于回测里一根K线结束才知道结果实盘完全不同每一分钟的行情都可能是成交信号。把状态迁移做成显式状态机数据错乱时从报错里一眼就能定位问题。断线重连的常见做法是在signal_manager.py里用一个循环加try/except检测到连接异常就把未确认订单写回本地 JSON 文件恢复后重新上送。4.2 信号落地到 MySQL 再推送交易信号不建议只存在内存里。源码包提供了一个落库示例用pymysql写入信号表-- 信号表结构MySQL 8.0 CREATE TABLE IF NOT EXISTS trade_signal ( id BIGINT AUTO_INCREMENT PRIMARY KEY, signal_time DATETIME NOT NULL, symbol VARCHAR(20) NOT NULL, side VARCHAR(10) NOT NULL, signal_price DECIMAL(10, 4), strategy VARCHAR(50), status TINYINT DEFAULT 0, INDEX idx_time (signal_time), INDEX idx_symbol (symbol) ) ENGINE InnoDB DEFAULT CHARSET utf8mb4;写入逻辑用参数化 SQL# execution/signal_manager.py 写入信号节选 import pymysql from datetime import datetime def insert_signal(conn, symbol: str, side: str, price: float, strategy: str dual_ma): sql INSERT INTO trade_signal (signal_time, symbol, side, signal_price, strategy) VALUES (%s, %s, %s, %s, %s) params (datetime.now(), symbol, side, price, strategy) with conn.cursor() as cursor: cursor.execute(sql, params) conn.commit()status字段默认 0可以手动改成 1 表示“已推送至模拟盘”改成 2 表示“模拟盘确认成交”。这个方案在复盘时的价值很明显三个月后要回溯某天的信号直接查表就行不需要重新跑回测。另一个好处是策略改动后可以用历史信号表对比新旧版本在相同时间点的信号差异判断改动是否合理。5. 避坑与常见问题排查回测漂亮实盘亏钱的六个原因这一章是全文的核心。我从这份源码包的代码里挑了六个最常见的坑按“现象→原因→解决”写清楚都是实操中真实踩过的。5.1 未来函数回测收益虚高的第一嫌疑现象策略在回测里年化超过 80%胜率极高但拉到实盘两个月就亏损。原因常见未来函数有两种。一是用当天收盘价计算信号后立刻按当天收盘价成交二是对收益率序列做shift(-1)把次日收益错误归因于当日信号。解决检查generate_signal返回的信号序列是否在时间轴上滞后一天。正确做法是信号在第 T 日收盘生成第 T1 日开盘成交。源码包的backtest_engine对信号做了移位处理但你自己加的过滤条件不一定做了。建议在run_backtest.py里加一行断言assert signal.index.equals(ret.index), 信号索引与收益索引不一致还有一个省事的办法把策略信号整体shift(1)后再和收益序列相乘。如果收益明显下降说明之前确实存在偷价行为。5.2 手续费与滑点设得太乐观现象回测年化 25%把手续费从万三改成万五、滑点从 0 改成 0.2% 后年化变成 12%。原因A股实际交易中冲击成本和小盘股的滑点损耗远大于佣金。成交价不是你想的收盘价而是对手盘给出的最优报价。解决保守估计。源码包默认slippage0.001相对合理但做小盘股或涨停板策略时手动调到 0.002 以上。每次改完参数跑一遍对比表年化和回撤都变化太大就要警惕策略本身薄弱。5.3 复权方式不统一导致信号漂移现象同一策略在不同数据源上回测信号数量相差 20%。原因前复权、后复权、不复权三种数据算出的均线位置不同尤其在除权除息日附近指标值会发生跳变。解决统一用后复权数据做计算。源码包的tushare_worker默认adjhfqakshare_worker需要显式传入复权参数。在一个策略里混用两种复权数据是大忌建议在data_loader.py里加一个断言强制所有标的用同一种复权方式。5.4 数据源接口失效与限流现象某天早上启动脚本报出类似“抱歉您每天最多访问该接口 X 次”的错误。原因tushare 和 akshare 免费接口都有频率限制不同接口限制的指标不同。解决源码包设计了双数据源自动切换但这不够还要在data_loader.py里加本地缓存。把每天拉到的日线存成parquet文件第二天优先读本地读不到再走接口。这个改动大约二十行能救很多次早盘启动。5.5 信号重复推送与幂等性缺失现象用定时任务每 5 分钟跑一次策略某个信号被重复推送了三次。原因没有记录上一次信号状态每次跑完都当作“新信号”处理。解决在signal_manager.py里加一张last_signal_state表每次推送前对比当前信号与上次信号是否有变化没有变化就不推送。更简单的方法是把信号哈希存到内存字典里策略进程重启后清空。推荐用数据库重启后仍然能防重。5.6 回测撮合与实盘撮合差异现象回测中按收盘价成交实盘用限价单总是差几分钱。原因回测引擎假设收盘价可完全成交但实盘行情在收盘那一刻可能已经跳价尤其是流动性差的标的。解决把回测成交价改为收盘价 * (1 slippage)做压力测试。如果在这个条件下策略仍然有正收益才有讨论实盘的意义。我个人习惯再加一档压力测试把滑点翻倍跑一遍能扛得住才考虑真实资金。6. 把源码包变成策略库注册机制、日志统一与三层验证6.1 策略基类加注册机制新策略不用改主流程源码包的base_strategy.py里可以加一个策略注册字典让新策略一键生效# strategy/base_strategy.py 注册机制 _STRATEGY_REGISTRY {} def register(cls): _STRATEGY_REGISTRY[cls.__name__] cls return cls register class DualMAStrategy(BaseStrategy): pass def create_strategy(name: str, **kwargs): cls _STRATEGY_REGISTRY.get(name) if cls is None: raise KeyError(f未知策略: {name}当前可选 {list(_STRATEGY_REGISTRY.keys())}) return cls(**kwargs)这样run_backtest.py里只需要维护一个策略名字字符串新增策略不用碰主流程代码。6.2 用 loguru 统一日志回测和模拟盘共用一套代码源码包里用的是utils/logger_config.py自封装的日志模块我一般建议改成loguru好处是同时输出控制台和文件日志不需要手写 formatter。关键场景是回测阶段打 INFO实盘模拟阶段开到 DEBUG信号推送失败时自动保留完整堆栈事后排查方便得多。# utils/logger_config.py 使用 loguru 的配置方式 from loguru import logger import sys logger.remove() logger.add(sys.stderr, levelINFO) logger.add(logs/trading_{time}.log, rotation1 day, levelDEBUG, retention7 days)6.3 三层验证闭环回测、参数敏感性测试、样本外测试最后说验证闭环三个步骤缺一不可。第一步看回测指标年化、最大回撤、夏普、交易次数、平均持仓天数。只看收益不看回撤是新手常犯的错。第二步做参数敏感性测试把fast从 3 到 10、slow从 15 到 60 全部跑一遍观察指标是否剧烈波动。如果参数在某个区间内小幅变化时指标平稳说明策略具备一定鲁棒性如果参数稍微一动收益就从正变负那是过拟合实盘大概率翻车。第三步做样本外测试用近两年的数据做样本前面数据做样本内参数寻优。样本外表现如果明显劣于样本内说明策略可能只是拟合了历史噪声。我也曾经在这个地方栽过跟头觉得写完策略就等于上了实盘结果三周内亏掉回测里半年才会亏掉的钱。从那以后每拿到一份量化交易源码我都强制自己先过一遍复权校验、滑点压力测试和样本外验证这三道关再谈其他。希望帮到你。本文还有配套的精品资源点击获取