简介一套基于Python的个人量化交易系统设计源码面向个人投资者、量化交易爱好者和Python开发者覆盖数据采集、因子计算、信号生成、策略回测、模拟与实盘交易、止损止盈及可视化监控等环节能够把分散的交易想法固化为可执行的程序化流程减少情绪化主观决策。压缩包共91个文件其中79个为Python源文件另有CSV市场数据、Markdown文档、XLSX交易记录、HTML界面以及低市盈率、短期强势、趋势加速和趋势等策略文件包体仅457KB。Python源码按数据、因子、信号、策略、监控、交易、止损止盈、工具等模块组织便于定位与二次开发。从源码结构看这套包包含数据爬取、因子计算、回测、模拟/实盘交易、可视化渲染等脚本并不是单纯的策略集合而是覆盖了从数据抓取到信号生成、再到交易执行和结果呈现的完整链路同时内置多种止盈止损与风险监控方法适合作为个人量化框架的起点。源码包目前已有685人学习下载对希望系统理解量化交易系统模块拆分、并在此基础上改造自己策略的读者来说具有较高的参考价值。1. 个人量化交易系统源码它不该是一个策略脚本而是一条可验证的数据流水线很多人到网上搜“基于Python的个人量化交易系统设计源码”心里想要的多半是一段能立刻回测的买卖逻辑。这类代码在多数场景下能跑回测曲线也漂亮可真拿去用第一个月就会遇到数据库断更、信号重复触发、手续费吞掉利润、电脑重启后任务不跑等问题。我做了几年个人量化后最深的感受是个人量化系统源码的核心价值不在某个预测模型而是把一个完整闭环固定下来——数据怎么来、策略怎么写、回测怎么证明它有效、下单怎么留痕。这篇文章就沿着这个闭环给出一套可以复现的骨架同时把我在成本参数、调度方式和风险控制上的取舍讲清楚。它适合已经有Python基础、但没完整搭过系统的从业者也适合打算把研究转成长期可运行程序的人。2. 先立骨架再写策略个人量化系统的模块划分与数据链路2.1 为什么个人系统不能只抄“策略代码”量化交易听起来最容易吸引人的部分是“策略”但个人系统和研究脚本最大的差别不是模型复杂程度而是“能不能每天对着同一个入口跑下去”。只把买入卖出写成一段独立脚本会出现几个很现实的问题数据源换了要去改策略里的取数逻辑回测和实盘用了两套信号计算逻辑结果对不上日志和订单没有落库亏了都不知道是哪一步引起的。所以我更愿意把“基于Python的个人量化交易系统设计源码”理解成一套骨架源码而不是某个策略源码。骨架要包含四件事数据接入、策略信号、回测验证、交易执行与记录。每一块都只用最少的代码先跑通后面再逐步替换。这里面策略反而是最好换的真正决定系统能不能长期用的是数据链路和风控边界。2.2 最小目录结构一眼就能看懂的模块边界我自己搭个人量化系统时会先用一个非常扁平的目录把边界立住。不要一上来就上微服务也不要所有函数都塞进一个main.py。下面这个结构是典型的个人项目做法quant/ ├── config.py # 路径、品种、手续费、滑点、数据库文件 ├── data/ │ ├── loader.py # 从数据源取数 │ └── storage.py # 写 SQLite去重 ├── strategy/ │ └── ma_cross.py # 策略信号生成 ├── backtest/ │ ├── engine.py # 向量化回测 │ └── metrics.py # 年化收益、回撤、夏普 ├── trade/ │ ├── risk.py # 下单前约束 │ └── paper.py # 模拟盘双账本 └── run_daily.py # 每日调度入口这个目录最大的好处是“边界靠文件夹而不是靠注释”。回测引擎不直接依赖具体数据源策略不关心数据库长什么样交易模块只接收标准字段的订单字典。这样你换数据商、换策略、换券商接口时都只需要替换某一块而不是重写整个源码。2.3 用数据源拉取日线10分钟跑通第一段数据链路个人量化最常见的数据频率是日线。以 A 股为例常见做法是用 baostock、tushare、akshare 这类公开接口不需要自己维护爬虫。下面这段代码用 baostock 拉取日线重点是把结果统一成 DataFrame 并转成数值类型。# data/loader.py import baostock as bs import pandas as pd def fetch_daily(symbol: str, start: str, end: str) - pd.DataFrame: # symbol 格式为 sh.600000 或 sz.000001 lg bs.login() if lg.error_code ! 0: bs.logout() raise RuntimeError(f登录失败: {lg.error_msg}) rs bs.query_history_k_data_plus( symbol, date,code,open,high,low,close,volume,amount,adjustflag, start_datestart, end_dateend, frequencyd, adjustflag2 # 2前复权1后复权3不复权 ) rows [] while rs.error_code 0 and rs.next(): rows.append(rs.get_row_data()) bs.logout() df pd.DataFrame(rows, columnsrs.fields) for col in [open, high, low, close, volume, amount]: df[col] df[col].astype(float) df[date] pd.to_datetime(df[date]) if df.empty: raise ValueError(f{symbol} 在 {start} ~ {end} 没有数据) return df这段代码看着长但每一行都在解决一个实际问题。bs.login()和bs.logout()要成对出现否则连接数会被占满接口返回的字段是字符串所以open、close这些列必须astype(float)不然之后算收益率会得到全零或者报类型错误adjustflag2表示前复权这是回测里比较常用的口径某一天的复权价会随着历史分红不断变化后复权则相反。数据拿到之后下一步是落地。个人系统用 SQLite 最省事既不需要起服务器也能直接用 pandas 读写。下面是一个带去重的入库函数# data/storage.py import sqlite3 import pandas as pd def upsert_bars(db_path: str, symbol: str, df: pd.DataFrame) - None: df df.copy() df[symbol] symbol with sqlite3.connect(db_path) as conn: df.to_sql(daily_bar, conn, if_existsappend, indexFalse) # 同一 symbol 同一天可能重复入库保留最新一条 conn.execute( DELETE FROM daily_bar WHERE id NOT IN ( SELECT MAX(id) FROM daily_bar GROUP BY symbol, date ) ) conn.commit()这里没有用数据库的唯一索引而是靠MAX(id)去重。原因是数据源偶尔会重复推送同一天的数据直接INSERT OR REPLACE依赖表结构里必须建唯一索引而增量拼接时漏建索引很容易忽略。这个写法在个人量级的数据量下足够健康百万行以内都不需要换 PostgreSQL。2.4 数据访问安全密钥不要写进源码标题里既然提到“设计源码”很多人会把接口的 token 直接写进配置再提交到 Git。个人量化虽然没有外部审计压力但这仍是个坏习惯。数据商的 token 一旦泄露别人可以消耗你的调用额度甚至伪造请求。我一般会把它放到环境变量里config.py只读环境# config.py import os DB_PATH os.getenv(QUANT_DB_PATH, data/market.db) DATA_TOKEN os.getenv(QUANT_DATA_TOKEN, )如果数据源要求 token就从终端注入export QUANT_DATA_TOKEN你的token python run_daily.py源代码里没有明文 token数据库文件也只放在项目目录下配合系统用户权限就能避免大部分“源码外流导致凭据泄露”的风险。这个习惯在项目刚起步时就值得养好。3. 回测引擎与策略解耦让双均线策略在真实成本下现原形3.1 回测为什么必须和策略分离很多新手写回测时会把均线计算、买卖点、资金曲线全部写在一个函数里。这样确实能很快看到结果但一旦策略从双均线换成布林带或者从日线换到小时线就要重写整个回测。更麻烦的是同一个信号在回测里和实盘里用了两套实现最后很难定位是策略问题还是实现问题。我的做法是让回测引擎只认“策略生成出来的持仓序列”。策略负责输出仓位引擎负责计算收益和成本两者通过 DataFrame 里的position列解耦。这样换策略不需要改成本计算换成本模型也不需要动策略。3.2 一个不骗自己的向量化回测实现先写一个最简单的双均线策略类# strategy/ma_cross.py import pandas as pd class MaCross: 双均线策略快线上穿慢线做多下穿清仓 def __init__(self, fast: int 5, slow: int 20): self.fast fast self.slow slow if fast slow: raise ValueError(fast 必须小于 slow) def generate(self, df: pd.DataFrame) - pd.DataFrame: df df.copy() df[fast_ma] df[close].rolling(self.fast).mean() df[slow_ma] df[close].rolling(self.slow).mean() # 当日收盘后比较均线得到目标仓位 df[target] (df[fast_ma] df[slow_ma]).astype(int) # 关键目标仓位后移一天避免用当日收盘信号当日成交 df[position] df[target].shift(1).fillna(0).astype(int) return df这里的shift(1)是整个回测能不能站住脚的核心。双均线在当天收盘后才能算出如果你在当天就以收盘价买入本质上用了“当天已经走完的行情”去成交这在实盘里不可能做到。position代表的是“今天这一根K线里我实际持有的仓位”它必须来自昨天的信号。回测引擎接收这个策略对象统一计算收益和成本# backtest/engine.py import pandas as pd def run_vectorized( df: pd.DataFrame, strategy, init_cash: float 100_000, fee_rate: float 0.0003, slippage: float 0.001, ) - pd.DataFrame: data strategy.generate(df) data[ret] data[close].pct_change().fillna(0) # 每天目标仓位是 position 本身因为策略已经做过 shift data[turnover] data[position].diff().abs().fillna(data[position].abs()) # 双边成本手续费 滑点都按成交金额比例处理 data[cost] data[turnover] * (fee_rate slippage) data[strategy_ret] data[position] * data[ret] - data[cost] data[equity_curve] init_cash * (1 data[strategy_ret]).cumprod() return data这套向量化回测假设你每天收盘时按收盘价结算仓位变化发生在当日滑点则用一个固定比例来吸收“开盘价和收盘价之间的差距”。它当然不是一个逐笔撮合模型但对日线级别的个人策略足够用。成本参数里最容易被低估的是slippage如果你交易的是小盘股0.001 可能都不够建议先按 0.002 起步。3.3 绩效指标年化、最大回撤、夏普的计算口径有了资金曲线还需要三个指标来判断策略能不能用。很多平台会直接算好但自己源码里建议保留一份因为不同平台的口径差异会导致你误判策略。# backtest/metrics.py import numpy as np import pandas as pd def annual_return(equity: pd.Series, periods: int 252) - float: n len(equity) if n 2: return 0.0 return (equity.iloc[-1] / equity.iloc[0]) ** (periods / n) - 1 def max_drawdown(equity: pd.Series) - float: peak equity.cummax() return float(((equity - peak) / peak).min()) def sharpe(returns: pd.Series, periods: int 252) - float: if returns.std() 0: return 0.0 return float(np.sqrt(periods) * returns.mean() / returns.std())annual_return用的是几何年化不是把每天收益率相加再除以样本年数。max_drawdown返回的是负数比如 -0.18 表示最大回撤 18%。sharpe用日收益率的均值除以标准差再年化这是最基础的口径虚线处要注意std()为 0 时会除零。把引擎跑起来后策略的参数不能只测一组就下结论。双均线(5,20)和(10,60)可能得出完全不同的结论如果你只挑最好看的一组参数对外展示那本质上是在做数据挖掘。个人系统的做法是先把一组参数扔进样本外验证看它是否还能保持正收益而不是看样本内回测曲线有多平滑。4. 从历史数据到定时任务SQLite 增量更新与调度设计4.1 为什么个人系统用 SQLite 而不是 CSV 或重型数据库CSV 是最容易入门的存储方式但增量更新很麻烦。每天往同一个 CSV 尾部追加如果行情接口重复推送某一天你要么手动去重要么把整个文件重新读一遍再覆盖。当数据覆盖几十只股票、每只股票几千条日线时CSV 的读写效率和一致性都会拖后腿。SQLite 是个人量化系统的低配首选。它只有一个文件数据库连接支持事务还能直接用 SQL 做去重和取最近日期。如果你的数据量已经到了千万行级别再考虑换 PostgreSQL 或者 ClickHouse但那是后话。个人系统最重要的不是存储引擎多厉害而是“每天的增量能安全合并进去”。4.2 增量更新别每次全量覆盖增量更新的逻辑很简单先看数据库里某个品种最近的一条数据日期再从数据源补上之后的每一天。这里要注意节假日和停牌所以取数窗口最好留出缓冲。# run_daily.py import os from datetime import datetime, timedelta from data.loader import fetch_daily from data.storage import upsert_bars from config import DB_PATH, SYMBOLS def main(): # 每次最多往回拉 7 天覆盖假期和接口延迟 end datetime.now().strftime(%Y-%m-%d) start (datetime.now() - timedelta(days7)).strftime(%Y-%m-%d) for symbol in SYMBOLS: df fetch_daily(symbol, startstart, endend) if df.empty: print(f{symbol} 无新数据) continue upsert_bars(DB_PATH, symbol, df) print(f{symbol} 更新 {len(df)} 行) if __name__ __main__: main()这里的关键参数是回拉的窗口大小。如果你只从数据库最大日期加一天开始拉遇到节假日会一直等不到数据遇到接口调整又容易漏掉某一天。回拉 7 天之后再通过去重逻辑合并是目前比较稳妥的个人做法。4.3 调度入口Windows 任务计划与 Linux cron数据落地后量化系统需要每天在固定时间跑一遍。开发环境里手动跑没问题长期运行就要交给系统调度器不需要在源码里写一个 while 循环让自己这台电脑一直挂着。Windows 用户可以用任务计划程序新建一个每日任务启动程序指向你的 Python 解释器参数填run_daily.py的绝对路径。Linux 或者 WSL 下用 cron 更直接30 17 * * 1-5 cd /path/to/quant /usr/bin/python3 run_daily.py logs/quant.log 21这条 crontab 表示每个交易日 17:30 执行一次数据更新。日志重定向到logs/quant.log非常重要否则定时任务报错时你看不到任何提示。个人系统最容易出的问题不是代码本身而是“任务没跑”这件事没有通知机制。先把日志落下来后面可以在run_daily.py里加一个失败时发送通知的入口。4.4 关键参数放在配置中心而不是散落在各个函数我在文章前面就强调过源码设计里要有一个config.py把通用参数管起来。这里的参数至少包括参数示例值说明DB_PATHdata/market.db数据库文件路径SYMBOLS[sh.600000, sz.000001]交易或研究标的FEE_RATE0.0003双边手续费率按成交金额计算SLIPPAGE0.002滑点比率用来吸收成交价差INIT_CASH100000回测初始资金TRADING_DAYS252A 股年化交易日数这些值一旦被写死在策略或回测函数里换参数就要开源码改逻辑很容易改出边界问题。个人量化系统设计源码的意义就是让代码本身变成“可配置的流程”而不是“一锤子买卖的脚本”。5. 量化系统源码落地的五类翻车点未来函数、滑点、复权、调度和类型5.1 回测里用了未来数据曲线越漂亮越可疑现象回测年化 50%最大回撤只有 5%但模拟盘或实盘连续亏损。原因最常见的未来函数是用“当天收盘后才能知道的信号”在当天收盘价成交。比如双均线策略里你在(12, 26)均线信号出现的当天就以收盘价买入等于比别人多知道了当天收盘后才确认的结果。解决信号必须向后平移一天。上面的策略代码里已经写了df[target].shift(1)这就是在强制让持仓晚一天生效。排查自己代码时重点看策略返回的position列里是不是有“当天信号出现、当天就有收益贡献”的情况。5.2 手续费和滑点参数不是摆设是吞吐成本的镜子现象回测里交易 200 次每次收益率看着是正的但扣掉手续费后收益变成负的。原因个人量化最常见的“看起来很赚”就是每次交易毛利只有 0.1%手续费和滑点却吃掉 0.3%。尤其是 A 股卖出还有印花税如果只在回测里算了佣金实际成本会被严重低估。解决把回测成本分成两部分。fee_rate按佣金率填slippage按你实际交易的品种流动性填。流动性差的股票按 0.002 起步流动性好的蓝筹可以试 0.001。开盘下单和收盘下单的冲击也不一样保守一点没有坏处。5.3 前复权、后复权数据混用关键位置上出现断崖现象某只股票在分红除权日前后回测资金曲线突然跳变收益率出现一个单日 20% 或 -15% 的假数据。原因前复权和后复权的价格是基于不同基准日算出来的。如果你把历史数据用前复权近期数据用不复权或者从两个数据源各取一段拼接处会因为除权缺口产生假收益。解决全库统一用一种复权口径。我的习惯是固定adjustflag2前复权入库并且在同一张表里不要一遍跑前复权、一遍跑后复权。如果必须混合使用至少增加字段adjustflag查询时按该字段过滤。5.4 定时任务在重启后蒸发量化变成“手动跑批”现象电脑重启一段时间后数据库最后更新日期停在几天前日志里也没有报错。原因Windows 任务计划程序在某些系统更新或休眠唤醒后会错过触发cron 任务如果服务器时间不对也会跳过。解决数据的增量更新入口要做成“可重复执行且幂等”。即使某一个交易日没跑下次任务运行时也能通过回拉 7 天窗口补上数据。这就是我在 4.2 节坚持不能只拉“最大日期1”的原因。同时把日志放在固定目录每次手动巡检时先看quant.log最后几行。5.5 数据源返回的是字符串不转换类型直接计算现象df[close].pct_change()出来的结果全是 0或者open close变成字符串拼接成12.312.5。原因很多量化数据接口返回的每一行是List[str]即使字段名是数值也仍然是字符串。解决在入库前统一做astype(float)并且要对所有数值列做不只做close一列。我踩过只转换了 close、没转换 volume结果成交量排序时全乱的坑。数据进入数据库之前把 schema 固定下来后面所有模块才不被类型问题反复打断。6. 最后一关样本外验证、风险预算和模拟盘双账本6.1 不要把参数档位留在样本内反复挑策略进入实盘前我会先把最近 12 个月的数据单独切出来作为样本外区间。样本内用来定参数样本外只跑一次不再回头调参。如果双均线在样本外还能保持和样本内同一量级的收益这个策略才有继续打磨的意义。6.2 下单前套一个风险约束函数模拟盘阶段就直接连真实的券商下单接口并不适合所有人。我更推荐先把下单函数封装成一个统一的入口所有策略信号都要经过风险检查才能进入模拟盘。下面是一个最小约束函数# trade/risk.py FEE_RATE 0.0003 SLIPPAGE 0.002 def pre_trade_check(order: dict, account: dict) - dict: price order[price] shares order[shares] if order[side] buy: notional price * shares if notional * (1 FEE_RATE SLIPPAGE) account[available]: raise ValueError(可用资金不足) # A股按手成交 if shares % 100 ! 0: shares (shares // 100) * 100 order[shares] shares # 单笔风险不超过账户权益的 1% risk_per_share abs(price - order[stop_loss]) max_shares int(account[equity] * 0.01 / risk_per_share / 100) * 100 order[shares] min(order[shares], max_shares) return order这个函数把资金不足、一手约束、单笔风险上限全部收进来。策略只负责给出“想买多少”交易模块负责决定“能不能买、买多少”。这样即便策略信号写得再激进实盘资金也不会一次性被某个异常参数打穿。6.3 双账本把“策略信号”和“实际成交”分开记模拟盘最有意义的地方不是验证策略收益而是验证“信号到成交”之间的偏差。我会在数据库里建两张表一张记录策略产生的目标订单另一张记录模拟盘实际成交结果。两者对比后你会发现信号价格和成交价格之间经常相差好几个滑点这些差异会直接影响策略评估。我个人这几年最大的教训是宁可让一个简单的双均线在数据、交易成本上做到 99% 可信也不要让黑匣子策略在回测里多赚 20%。把上面这套骨架搭好后续换策略只是替换strategy目录里的一个类数据链路、回测口径和风险约束都不用重写。希望这套思路和代码能帮你少走一段弯路也希望你能找到属于自己的实盘边界。本文还有配套的精品资源点击获取