简介一套基于Python的个人量化交易系统源码面向个人投资者和量化爱好者覆盖从行情数据采集、因子计算、策略生成到回测、模拟交易与风险监控的完整流程。压缩包大小约457KB共91个文件其中包括79个Python源文件、CSV历史数据、Markdown说明文档、XLSX回测记录、HTML可视化页面以及低市盈率、短期强势、趋势加速、趋势等独立策略文件目录按factor、data、signal、stop_profit、stop_loss、strategy、monitor等模块清晰组织便于按需查阅。已有685人学习过这份源码。源码不仅实现了数据爬虫、因子计算、交易信号生成与回测模拟等核心功能还提供了多种止损止盈方案和基于HTML的可视化监控界面有助于理解量化系统的事件驱动架构和工程化实践适合希望快速搭建个人量化系统或深入策略研究的开发者参考、复用与二次开发。1. 个人量化交易系统免费源码很多能稳定跑完一年的很少网上随手一搜Python量化交易源码能翻出上百套双均线、MACD、各种“主力监测”指标一应俱全多数还配着一条漂亮的净值曲线。可真把源码拉到自己电脑上大多数人在数据更新这一步就卡住了库装不上、接口换了、行情格式对不上更别提从头到尾跑一年模拟盘。个人量化交易系统核心不是“会写策略”而是把数据获取、策略回测、风控执行这几块拼成一个每天能自动运转的闭环。下面按这个闭环拆开讲适合已经会一点Python、想自己搭一套能跑通全流程系统的从业者也适合那些在“免费源码”里反复试错、想找个靠谱落地路径的人。2. 先把系统拆成四层数据、策略、回测、执行怎么分工个人量化系统最容易犯的错是一上来就写策略写完了发现没有可靠的数据源也没有一个能验证收益的框架。我一般会先按四个层把目录结构定下来每层只干一件事层与层之间用数据接口通信。数据层负责把行情落进本地库策略层只负责根据K线算信号回测层拿历史数据验证这些信号执行层则负责把信号变成真实的成交记录。四层拆开之后任何一个环节换实现方式都不会牵动其他代码。2.1 为什么个人系统不需要微服务模块划分与选型见过有人把个人量化系统做成四个微服务用Docker Compose编排Kafka传消息最后连行情都还没收全就放弃了。日线级别、几百只股票、一天一次更新这种量级单机单进程完全扛得住复杂度才是最大的敌人。我的建议是一个Python项目仓库按src、data、strategy、backtest、risk、logs建目录数据库用SQLite定时任务用crontab或者系统计划任务就够了。模块划分上数据层对上层暴露的接口应该只有“给我某只股票某段时间的DataFrame”至于数据是从AkShare拉的、从Tushare下载的还是手工导入的调用方不需要关心。策略层输出的是带position字段的DataFrame1代表持仓、0代表空仓不掺仓位管理的逻辑。回测层拿这个position和真实行情算净值曲线执行层则读取最新信号决定下单或不下单。这样每个模块都能单独测试数据文件格式变了也不会改到策略代码。选型上还有两个容易纠结的点。一个是SQLite还是MySQL个人单机场景我选SQLite零部署、单文件备份方便几百万行日线数据查询毫秒级返回另一个是事件驱动回测还是向量化回测日线策略用向量化就够了pandas的shift和cumprod能覆盖绝大多数逻辑只有做tick级或逐笔撮合才需要事件驱动框架。对了密钥管理一开始就要做对。Tushare的token、数据源账号这类信息不要写进代码文件放在config.local.yaml里并且让这个文件进.gitignore。数据访问和存储安全不用做到机构那种带加密方案设计的程度但至少做到“本地库文件别带密钥、代码提交时别带token”这两个底线守住后面才敢把项目推到远端。2.2 用AkShare/Tushare把日线行情落进SQLite建表与增量更新数据获取我常用AkShare免费、接口多、国内网络直接访问。以A股日线为例ak.stock_zh_a_hist能拿到历史行情下面的函数把一只股票的日线写入SQLite表名直接用股票代码。# data_fetcher.py import akshare as ak import sqlite3 from datetime import datetime import pandas as pd def update_daily(symbol: str, db_path: str quant.db) - None: df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_date20200101, end_datedatetime.now().strftime(%Y%m%d), adjustqfq # 前复权默认拉取 ) # 字段重命名统一用英文避免每次查询都写中文别名 df df.rename(columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume })[[date, open, close, high, low, volume]] conn sqlite3.connect(db_path) df.to_sql(symbol, conn, if_existsreplace, indexFalse) conn.close() print(f{symbol} updated: {len(df)} rows)逻辑说明ak.stock_zh_a_hist返回的是带中文列名的DataFrame先重命名成date/open/close/high/low/volume五个核心字段再写入SQLite。to_sql用if_existsreplace是整表重建数据量小的时候最简单可靠如果股票数量多、每天全量拉一次会撞上接口频率限制所以要改成增量更新。增量更新的做法是先查库里已有最大日期start_date填这个日期的后一天然后to_sql用if_existsappend追加最后按date去重一次。下面是增量版的核心改动。def update_daily_incremental(symbol: str, db_path: str quant.db) - None: conn sqlite3.connect(db_path) cur conn.execute( fSELECT MAX(date) FROM {symbol} # 表名来自代码内部变量不拼用户输入 ) last_date cur.fetchone()[0] start last_date if last_date else 20200101 conn.close() df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart, end_datedatetime.now().strftime(%Y%m%d), adjustqfq ) # 字段重命名与上面的全量版保持同一套逻辑 df df.rename(columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume })[[date, open, close, high, low, volume]] df df[df[date] last_date] conn sqlite3.connect(db_path) df.to_sql(symbol, conn, if_existsappend, indexFalse) conn.close()参数说明start_date接口要求传的是字符串格式YYYYMMDD所以库里存的date也要统一成这种格式避免比较时出现类型不一致。adjustqfq是前复权必须显式写漏了会用未复权价格回测里除权除息日会出现假跳空。增量版里最后一行df[df[date] last_date]是保险措施防止接口返回重复数据把表写脏。这里还有一个坑数据源接口偶尔会改字段名AkShare改版频率不算低所以更新脚本里最好把“列名重命名类型转换”抽成一个函数接口变动时只改一处。提示首次建库不要只拉策略会用到的股票把自选股和备选池一次性全量拉下来因为历史数据接口是按次请求的以后补数据比现在麻烦得多。3. 写一个可复用的策略与回测引擎从双均线到绩效指标策略层和回测层是个人量化系统的核心。我看过不少“Python量化交易策略代码”很多是把K线算成几个指标再画买卖点代码只有几十行但换一只股票、换一个周期就不出信号了。问题不在于策略本身而在于策略代码没有抽象信号、持仓、成交、绩效全混在同一个循环里。这一章先讲策略怎么写才可复用再讲回测引擎凭什么让你相信收益数字。3.1 策略基类与双均线示例信号只在下一根K线生效有一个思路和同花顺supermind这类平台很像平台帮你把数据准备好你只用十行左右的代码描述买卖条件就能快速验证想法。自己搭系统的做法是反过来的先把策略的输入输出约定好再填条件这样换策略不换框架。我一般定义一个Strategy基类子类只实现generate_signals方法输入是历史DataFrame输出是带signal列的DataFrame。# strategy.py import pandas as pd class Strategy: def generate_signals(self, df: pd.DataFrame) - pd.DataFrame: raise NotImplementedError class DualMA(Strategy): def __init__(self, fast: int 5, slow: int 20): self.fast fast self.slow slow def generate_signals(self, df: pd.DataFrame) - pd.DataFrame: df df.copy() df[ma_fast] df[close].rolling(self.fast).mean() df[ma_slow] df[close].rolling(self.slow).mean() # signal: 1 持有多头, -1 持有空头, 0 空仓 df[signal] 0 df.loc[df[ma_fast] df[ma_slow], signal] 1 df.loc[df[ma_fast] df[ma_slow], signal] -1 # 关键信号延迟一根K线生效避免用当天收盘价成交 df[position] df[signal].shift(1).fillna(0) return df逻辑说明双均线策略里快线上穿慢线做多、下穿做空这是最基础的择时逻辑适合用来验证回测框架不适合直接拿去实盘。signal列代表“收盘后计算出的目标状态”position列代表“下一根K线真正执行的状态”shift(1)这个动作就是把计算和成交隔离开。很多源码里直接拿signal去乘当天收益率等于当天收盘价知道结果、当天收盘价成交这在回测里属于典型的未来函数会让收益看起来远好于实际。参数说明fast和slow是均线窗口默认5和20对应一周和一个月左右。这两个参数不要一上来就调成你觉得“最赚钱”的值参数越敏感的策略样本外越容易失效。策略基类的好处是回测引擎不需要知道策略内部逻辑只用调用generate_signals拿position后面接MACD、接布林带甚至接那些“三步点金”“主力监测器”指标都只是多写一个子类的事。3.2 回测引擎的核心循环手续费、滑点与最大回撤计算有了position序列回测就只剩下两件事算收益率、算绩效指标。收益率计算最朴素的写法是把每日收益率乘以当日的position然后累乘成净值曲线。但这里必须扣掉交易成本否则回测结果就是一块注水猪肉。# backtest.py import pandas as pd def run_backtest(df: pd.DataFrame, position_col: str position, fee: float 0.0003, slippage: float 0.001): df df.copy() df[ret] df[close].pct_change().fillna(0) df[position_shift] df[position_col] # 成交发生时才扣成本仓位从0变1、1变0都算一次换手 df[trade] df[position_shift].diff().abs().fillna(0) cost df[trade] * (fee slippage) df[strategy_ret] df[position_shift] * df[ret] - cost df[cum] (1 df[strategy_ret]).cumprod() return df逻辑说明trade列反映仓位变动从0到1、从1到0都算一次换手每次换手扣fee加slippage。这里的fee是佣金比例默认万分之三slippage是滑点比例默认千分之一已经粗略包含买卖价差和冲击成本。A股还有卖出印花税更严格的模型应该按方向区分费率但初版回测先统一按成交金额比例扣等策略稳定了再细化。真正的重点是只要交易频率高成本对净值的侵蚀会非常明显很多日线策略把cost设成0年化收益率能翻一倍这就是网上那些漂亮收益曲线的来源。净值序列出来之后接着算三个核心绩效指标年化收益率、最大回撤、夏普比率。最大回撤用净值序列的累计高点算它直接回答“这策略最惨的时候亏多少”比年化收益更能说明风险。def max_drawdown(equity: pd.Series) - float: peak equity.cummax() drawdown equity / peak - 1 return drawdown.min() def sharpe_ratio(returns: pd.Series, periods: int 252) - float: return returns.mean() / returns.std() * (periods ** 0.5)回测结果里我建议把最大回撤和年化收益放在同一张表里看不要只看年化。比如双均线在沪深300上的典型结果是年化8%到12%、最大回撤15%左右这在个人量化里属于能接受的区间如果看到一个策略年化40%同时最大回撤只有3%先别高兴回去查有没有未来函数、有没有把手续费设成0。常见参数组合双均线窗口5/20、10/30、5/60手续费万分之三滑点千分之一回测周期至少包含一轮完整的涨跌、五年起步样本外预留最近半年到一年不参与调参。4. 从回测到模拟盘实盘接入前先过一遍风控回测再漂亮也只是历史数据上的一个假设。个人量化系统走到这一步最稳妥的路径不是直接接实盘而是先开一个模拟盘把自己当成真钱在管。这一章讲两件事怎么设计模拟盘阶段才算数以及下单前那几道风控检查到底检查什么。4.1 模拟盘与Paper Trading先跑四到八周样本外模拟盘的目的是验证“从信号到成交”这条链路而不只是验证策略。回测里我们假设每天收盘价能成交实际跑模拟盘时你会发现收盘前信号还没算出来、收盘后下单要等第二天开盘、开盘价可能跳空高开两三个点这些偏差才是回测和实盘之间的黑匣子。我一般建议模拟盘至少跑四到八周而且要覆盖调仓日。具体做法是每天收盘后跑一遍数据更新和策略信号把当日应执行的调仓动作写进一个orders表第二天开盘手动或半自动下单再把实际成交价记录回库里。这个流程里模拟盘用的数据和回测数据必须严格分开不能用已经算过结果的同一段历史数据再来一遍要等新数据自然产生。把模拟盘当作一次真实演习记录三个关键指标信号到成交的延迟、实际成交价与信号价的偏差、每天未成交或撤单的次数。这些数据积累下来用来修正回测里的滑点参数。比如回测设滑点千分之一模拟盘实际统计下来平均偏差到了千分之三那回测参数就要改否则后面实盘的收益预期全是虚的。4.2 下单前的风控函数单笔仓位、当日熔断与日志落库模拟盘跑顺之后下单逻辑一定要和风控绑定在一起不能“信号出来就无脑买”。个人最容易犯的错是重仓一只票一天亏掉全年收益。我习惯把风控写成独立的risk_check函数在生成订单之前拦一道。# risk_check.py def risk_check(cash: float, price: float, position_value: float, equity: float, daily_pnl: float, max_single_pct: float 0.2, max_position_pct: float 0.8, daily_stop: float -0.05): # 单笔最大仓位不能超过总资金的两成 amount cash * max_single_pct shares int(amount / (price * 100)) * 100 if shares 0: return None # 单票总仓位已有持仓加上本次买入后占比不能超过八成 if (position_value shares * price) / equity max_position_pct: return None # 当日熔断当日亏损超过5%全天停止开仓 if daily_pnl equity * daily_stop: return None return shares逻辑说明risk_check返回可买入的股数按手取整取不了整就返回None表示放弃本次开仓。第一个检查控制单笔投入第二个检查控制单票集中度第三个检查是当日熔断净值回撤到阈值就躺平不再开新仓。max_single_pct设0.2相当于最多同时持有五只票比较适合A股散户如果策略本身是集中持仓风格可以调到0.3但不要超过0.4否则一次极端行情就把自己带走了。风控函数的参数不要写死在函数里建议放在配置文件里。与之配套的是一张strategy_log表记录每次信号的日期、股票、方向、信号价、实际成交价和状态。这张表除了用来复盘还能在下一次回测时做校准查一查实际成交价和信号价的偏差如果系统性偏高说明滑点估计偏乐观。日志落库的SQL很简单关键是只在事务提交后记录避免回滚时把假信号写进表里。CREATE TABLE IF NOT EXISTS strategy_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, symbol TEXT, signal_date TEXT, direction TEXT, signal_price REAL, exec_price REAL, status TEXT DEFAULT pending, created_at TEXT );说明signal_price是信号产生时的参考价exec_price是模拟盘实际成交价status记录pending、filled、canceled三种状态。这张表不要和行情数据表混在一起行情表是只读的日志表是频繁写入的分开能减少锁冲突也让对账更清晰。5. 避坑清单个人量化系统最常见的5个翻车点这一章是我在搭建和测试中反复踩过的坑。每个坑都按“现象、原因、解决”来记录遇到类似问题时可以直接对照排查。5.1 数据与回测层的3个坑第一个坑是未来函数。现象回测曲线非常漂亮年化收益高得离谱但一模一样的逻辑拿到模拟盘上完全不是那么回事。原因常见的有两种一种是拿当天收盘价计算信号又用当天收盘价成交另一种是把当日收盘后才确定的数据直接用于当日持仓判断。解决信号一律shift(1)之后再参与收益计算需要严格排除当天的均值指标rolling之后再做一次shift(1)这两步是回测代码里最容易偷懒的地方也最值得反复检查。第二个坑是把手续费和滑点设成0。现象周度调仓策略回测年化30%实盘跑下来连手续费都不够付。原因A股佣金、印花税、买卖价差叠加起来高频调仓一年能吃掉十几个点回测不扣成本等于默认交易免费。解决初版回测就把fee设成万分之三、slippage设成千分之一如果策略对这几个参数极其敏感先怀疑策略本身的换手率是不是太高而不是怀疑参数设错了。第三个坑是前复权数据“漂移”。现象同一个股票代码上周拉的历史K线和今天拉的对不上回测结果也跟着变净值曲线每次跑出来都不一样。原因前复权以最新价为基准股票分红送股之后接口返回的历史价格会被重新计算存量数据没更新就会和最新K线拼接错位。解决历史数据入库后不要反复重拉固定一个版本做事件驱动类研究改用后复权或者同时保存复权因子在计算收益率时自行处理除权除息。5.2 实盘与配置层的2个坑第四个坑是SQLite并发写锁。现象行情更新脚本和回测程序同时运行回测程序直接报database is locked跑出来的结果缺一段数据。原因SQLite是单写多读模型一个写事务没结束另一个写请求就只能等长事务或者频繁写会把库锁很久。解决行情更新写库用短事务每只股票写完立即commit连接串里加上timeout参数如果并发确实多备份和更新分开时段回测只在行情更新结束后开始。我自己的习惯是每天盘后先跑更新脚本更新完再跑回测任务顺序执行避免锁。第五个坑是迷信网上各类指标源码。现象跟着“主力监测器”“三步点金”“麒麟三红”这类指标源码把买卖点代码抄进系统结果历史回测还行最近一年的走势里完全失效。原因这些指标公式本质上是K线或量价关系的某种计算规则源码本身只是特征描述不是策略加上它们大多是在特定历史行情里被“拟合”出来的换到样本外自然失灵。解决把这类指标当特征接入策略层用样本外数据和参数敏感性验证先画出最近一年的净值对比有效再保留无效就删掉不用心疼。6. 上线前用一份checklist验证你的系统回测可信度与实盘一致性最后一个环节是验证不是写代码是给前面所有模块挑毛病。我每次上新策略前都会按下面这份checklist走一遍任何一项不过关就不进模拟盘。第一项是样本外回测。把最近六到十二个月的数据从调参数据里隔离出去策略确定参数后只在样本外数据上跑一次记录年化和最大回撤与样本内的差异超过一倍就要警惕过拟合。第二项是参数敏感性。拿双均线来说快线参数从5改到15慢线从20改到60如果绩效指标大幅波动说明策略是在特定参数上“刻”出来的。第三项是成本敏感性。把手续费和滑点分别放大两倍看净值曲线还撑不撑得住撑不住的策略通常没有实盘意义。第四项是成交偏差统计用strategy_log表做一次对账。SELECT signal_date, signal_price, exec_price FROM strategy_log WHERE status filled AND abs(exec_price - signal_price) / signal_price 0.01;这条SQL查的是成交价偏离信号价超过1%的记录如果这类记录数量很多并且系统性偏向不利方向说明回测的滑点参数需要调大或者下单时点需要改。第五项是检查订单状态里有没有大量canceled记录频繁撤单通常意味着流动性不足或者信号触发条件太窄。我自己的习惯是保留每一次回测的净值文件和对应的数据版本号这样后续复盘还能复现原结果。没有放之四海而皆准的盈利圣杯但把样本外、成本、成交偏差这几个环节盯死至少能排除掉一大半“回测好看、实盘翻车”的玄学问题。希望这套从数据层到风控层的落地路径能帮你在搭自己的Python个人量化交易系统时少走一段弯路。本文还有配套的精品资源点击获取