简介：一套基于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字段的DataFrame，1代表持仓、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( symbol=symbol, period="daily", start_date="20200101", end_date=datetime.now().strftime("%Y%m%d"), adjust="qfq" # 前复权，默认拉取 ) # 字段重命名，统一用英文，避免每次查询都写中文别名 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_exists="replace", index=False) 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_exists="replace"是整表重建，数据量小的时候最简单可靠；如果股票数量多、每天全量拉一次会撞上接口频率限制，所以要改成增量更新。增量更新的做法是：先查库里已有最大日期，start_date填这个日期的后一天，然后to_sql用if_exists="append"追加，最后按date去重一次。下面是增量版的核心改动。
def update_daily_incremental(symbol: str, db_path: str = "quant.db") -> None: conn = sqlite3.connect(db_path) cur = conn.execute( f"SELECT 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( symbol=symbol, period="daily", start_date=start, end_date=datetime.now().strftime("%Y%m%d"), adjust="qfq" ) # 字段重命名与上面的全量版保持同一套逻辑 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_exists="append", index=False) conn.close()参数说明：start_date接口要求传的是字符串，格式YYYYMMDD，所以库里存的date也要统一成这种格式，避免比较时出现类型不一致。adjust="qfq"是前复权，必须显式写，漏了会用未复权价格，回测里除权除息日会出现假跳空。增量版里最后一行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个人量化交易系统时少走一段弯路。
本文还有配套的精品资源，点击获取
