Python股票量化系统实战：从数据清洗到回测引擎的完整实现-育师

简介：Python股票量化系统源码及教程是一套面向量化投资入门者与Python开发者的完整项目，提供可运行的量化系统与配套教学说明。系统依赖MySQL存储市场数据，通过pip命令批量安装依赖后即可启动，适合希望快速搭建本地量化策略测试环境的学习者。资源共244个文件，压缩包仅3.53MB，包含71个py源码、80个pyc文件、20个json配置、7个ui界面以及png图表和前端页面等，覆盖策略逻辑、配置管理、界面交互与可视化展示，目录结构清晰。教程涉及数据获取、历史分析、策略编写、回测执行、结果评估与风险控制，并对win_main.py主脚本进行解读，帮助用户从零理解量化系统开发全流程。目前已有38人学习下载，对想掌握Python量化交易系统搭建和策略开发思路的用户而言，是一份轻量且实用的参考资料。

1. Python股票量化系统的边界与起点

用Python写股票量化系统，多数人第一步就错了：不是先找源码，而是先问自己手里的K线数据从哪来。我见过太多人从网上下了一份源码，装上依赖，跑出漂亮的资金曲线，投入实盘后却连续止损。原因往往不在策略，而在数据复权、信号滞后、手续费模型这些代码里看不见的地方。围绕一个可运行的Python股票量化系统源码，下面会完整覆盖数据清洗、策略编写、回测验证到实盘部署前的检查，适合刚读完Python教程想要动手做完整系统的开发者，也适合想审视自己框架的老手。

2. 从行情源到本地数据仓库：Python量化系统的数据层实现

数据是一切量化系统的地基。如果数据里的复权因子算错，回测里的所有盈亏都会失真。常见做法是用akshare或tushare拉取历史行情，理由很简单：它们都免费，而且返回的字段足够支撑一般策略。相比自己写爬虫去抓网页，使用这些库可以将精力集中在数据校验上。另外，如果你还在对照python安装教程配环境，建议直接把Python升到3.10以上——pandas和sqlite3的新特性会减少后续兼容性问题。

2.1 选型：免费数据源与本地存储格式

akshare不需要token，tushare需要积分；我的建议是先用akshare，等需要分钟线或因子数据再迁移。存储格式常用SQLite或者HDF5。SQLite适合单机回测和增量更新，CSV虽然直观，但多标的时文件量大且难以事务性更新。如果需要按日期切片、跨股票查询，SQLite是更稳的选择。有些“免费python源码大全”里会把所有数据都存CSV，结果一运行就内存溢出，这不是代码问题，是存储选型问题。

2.2 数据获取与清洗代码：从API到DataFrame

import akshare as ak import pandas as pd import sqlite3 df = ak.stock_zh_a_hist( symbol="000001", period="daily", start_date="20200101", end_date="20241231", adjust="qfq" ) # 统一列名，避免后续代码使用中文索引 df.rename(columns={ "日期": "date", "开盘": "open", "收盘": "close", "最高": "high", "最低": "low", "成交量": "volume" }, inplace=True) print(df.head())

symbol是股票代码字符串，period支持daily/weekly/monthly，start_date和end_date按YYYYMMDD传，adjust设为qfq表示前复权。这里最需要注意的是返回值顺序是正序还是倒序，akshare默认正序，但最好主动按date排序。另外日期列要转换成datetime类型，避免后面对比时类型不统一。

存储到SQLite时，to_sql会做一次全量替换，适合首次写入；增量更新应使用INSERT OR REPLACE，否则重复写入会生成重复行。

conn = sqlite3.connect("market.db") df["date"] = pd.to_datetime(df["date"]) df.sort_values("date", inplace=True) df.to_sql("daily_000001", conn, if_exists="replace", index=False) conn.close()

2.3 复权、对齐与去重：量化数据仓库的几个硬指标

前复权是动态的，会导致历史价格不断变化；因此对于策略回测，我一般存两种：原始价和复权因子，方便切换。字段表如下。

多标的做因子计算时，最容易出问题的不是价格而是对齐。用pivot把多只股票的close转成宽表后，不同股票的停牌日会留下NaN。常见做法是ffill延续前值，或者在计算因子时逐列删除NaN，而不是整个数据集一起删。去重也一样，同一个交易日同时出现两个收盘价时，保留成交量较大的那条，另一条删除。

3. 因子计算与信号生成：策略代码的源码写法

策略代码的核心不是买卖时机，而是把原始价格转换成可解释的因子。常见的双均线、RSI、布林带都可以在几分钟内写出，但真正考验人的是信号生成的前视偏差。这里用一个成交量加权均线系统作为例子，演示如何写因子、生成信号并检查未来函数。这一节里的源码风格可以直接复用到其他因子项目中。

3.1 从行情到因子：用向量化计算避免循环

def add_factors(df: pd.DataFrame) -> pd.DataFrame: # 5日与20日均线 df["ma5"] = df["close"].rolling(5).mean() df["ma20"] = df["close"].rolling(20).mean() # 量能比值：当日成交量 / 过去10日平均成交量 df["vol_ratio"] = df["volume"] / df["volume"].rolling(10).mean() # 避免第一行出现NaN影响后续运算 df.dropna(inplace=True) return df

rolling窗口默认右对齐，意味着当根K线收盘后计算，不含未来数据。这里要明确指出，pandas的rolling天然避免了未来函数，但如果后续手动用shift，方向就很容易搞反。比如要取“昨天均线”，应该用df["ma5"].shift(1)，而不是shift(-1)。成交量比值的分母用10日平滑，能过滤单日放量的噪音。

3.2 信号生成与交易指令的分离

df["signal"] = 0 df.loc[df["ma5"] > df["ma20"], "signal"] = 1 # 多头持仓 df.loc[df["ma5"] < df["ma20"], "signal"] = 0 # 空仓或空头 # 根据信号转变点发出交易单 df["trade"] = df["signal"].diff().fillna(0) # trade=1 买入，trade=-1 卖出 orders = df[df["trade"] != 0]

signal是目标仓位，trade是仓位变化量。这样设计的好处是：策略逻辑只负责给出目标状态，回测引擎只负责按trade字段执行，二者解耦。很多新手在策略里直接写买入卖出，导致换仓逻辑难以测试。diff函数计算相邻两行的差值，等于1说明刚从空仓变成持仓，等于-1说明刚刚清仓。

3.3 未来函数检查：最容易写错的一段

未来函数的典型错误是使用shift(-1)提前取下一根K线的数据。比如想在收盘前判断均线交叉，就会写出df["ma20"].iloc[i+1]这类代码。自查方法也很简单：把信号日为1的样本，对比当天收盘价和次一天开盘价，如果平均跳空收益显著为正，说明信号可能泄漏了未来信息。另一个技巧是打开回测引擎的逐笔打印，检查每一笔委托的发出时间一定晚于信号K线的收盘时间。

4. 事件驱动回测引擎：逐K线模拟交易与绩效评估

在量化系统里，因子和信号解决“什么时候买”，回测引擎解决“买了之后会发生什么”。很多人直接用向量化方式对资金曲线做一次性计算，但这样做无法准确模拟涨跌停、停牌和账户资金约束。我一般会实现一个极简的事件驱动引擎：每一根K线触发一个事件，引擎更新账户状态。这是量化系统源码里最值得反复打磨的部分。

4.1 回测引擎的最小架构：K线循环、账户状态、订单队列

class Backtester: def __init__(self, df, initial_cash=100000, fee=0.001, slippage=0.002): self.df = df.reset_index(drop=True) self.cash = initial_cash self.position = 0 self.fee = fee self.slippage = slippage self.equity_curve = [] def run(self): for i in range(1, len(self.df)): bar = self.df.iloc[i] signal = self.df.iloc[i - 1]["trade"] # 上一根K线的信号 if signal == 1 and self.position == 0: buy_price = bar["open"] * (1 + self.slippage) qty = int(self.cash / (buy_price * (1 + self.fee))) self.cash -= qty * buy_price * (1 + self.fee) self.position = qty elif signal == -1 and self.position > 0: sell_price = bar["open"] * (1 - self.slippage) self.cash += self.position * sell_price * (1 - self.fee) self.position = 0 self.equity_curve.append(self.cash + self.position * bar["close"])

信号在K线结束后生成，用下一根K线的开盘价成交，这样的实现天然避免前视偏差。int(self.cash / ...)做向下取整，保证买入数量是手数整数倍。如果没有把trade信号与执行分离，这段代码里的signal就只能是历史值。这里我把循环从range(1, len(self.df))开始，是因为第0根K线没有前一根信号。

4.2 撮合细节：手续费、滑点与涨跌停限制

参数	默认值	影响
fee	0.001	高频率策略对费率敏感
slippage	0.002	小盘股需要调大
initial_cash	100000	仓位计算基数

真实交易中，涨停板时可能买不进，跌停板时卖不出。如果bar["open"]已经涨停，即open == round(previous_close * 1.1, 2)，应该跳过买入。否则回测资金会在实盘中变成一堆无法成交的委托单。滑点参数在小市值股票上需要调到0.005以上，因为流动性不足会导致更大的买卖价差。

4.3 绩效指标计算：年化收益、最大回撤、夏普比率

equity = pd.Series(self.equity_curve) returns = equity.pct_change().dropna() annual_return = (equity.iloc[-1] / equity.iloc[0]) ** (252 / len(equity)) - 1 max_drawdown = (equity / equity.cummax() - 1).min() sharpe = returns.mean() / returns.std() * (252 ** 0.5)

年化收益按252个交易日计算，最大回撤是每个时点净值相对历史最高点的最大回落。夏普比率使用日收益率均值除以标准差再开方年化。这里不要忽略无风险利率，但A股散户实盘通常直接用0代替，误差不大。equity.cummax()计算历史最高净值，任何时间点的回撤都是相对这个最高点的。

5. 参数敏感性、样本外测试与上线前检查

5.1 参数敏感性热力图：找出过拟合区间

双均线参数ma1和ma2的固定组合如果在某个窄区间内表现极好，而邻近参数收益骤降，说明是过拟合。用两层循环生成热力图数据，并输出所有参数结果分布。

import numpy as np results = [] for ma1 in range(3, 30, 2): for ma2 in range(20, 90, 5): if ma1 >= ma2: continue ret = run_backtest(df, ma1, ma2) # 重复执行策略回测 results.append((ma1, ma2, ret))

观察最优参数周边区域的收益均值。如果最优参数的收益和次优参数差别超过20%，且周边参数收益骤降，说明模型参数没有稳定性。正确做法是选择一块收益高原的中央参数，而不是单纯选最高点。