简介:这是一套面向量化投资初学者与Python开发者的股票自动化分析系统源码,解决个人投资者缺乏专业工具进行选股、策略验证与实盘联动的痛点。系统基于Python 3.4+构建,集成PyQt5实现可视化交互界面,采用多线程事件引擎支撑高频数据处理与实时通知,覆盖数据获取、自动选股、策略回测、模拟交易(支持9个独立账号)及微信实盘提醒等核心环节。压缩包共317个文件,含294个Python源码(涵盖策略模板、数据下载、交易执行、UI逻辑等模块)、11个PNG/JPG界面资源图、4份Markdown文档(含历史数据下载、策略编写、配置说明等实用指南),以及必要依赖jar包与许可证文件,整体仅1.81MB,轻量易部署。已有89人学习下载,提供开箱即用的MongoDB数据存储方案、实盘与回测统一策略接口、单账户多策略支持,以及Wind/Tushare双数据源适配能力,助用户快速理解量化系统架构并开展本地化开发与实证。 真正决定一套量化分析系统能不能长期跑下去的因素,往往不是模型多复杂、策略多高端,而是数据链路稳不稳、界面会不会卡死、告警能不能及时到人。这句话是我把这套用 Python 写的股票量化分析系统从命令行脚本逐步改造成带 PyQt 图形界面的完整工具之后,最大的体会。整个项目覆盖了自动选股、交易策略测试、实时行情通知和用户互动,功能听起来很全,但每一块的落地方式都值得单独拆开讲。如果你正准备自己动手做一套差不多的系统,或者已经在写但是卡在某一环,这篇文章应该能帮你少走不少弯路。
我默认你是对 Python 有一定基础、想了解量化系统如何工程化的开发者。我不会只堆概念,而是按我实际开发时的顺序,从行情数据接口选型,到自动选股的多因子打分,再到回测引擎、PyQt 界面、实时通知、打包发布,完整讲一遍设计和踩坑记录。文中的代码都是可运行的简化示例,可以直接作为骨架去扩展。
1. 行情数据接口选型与预处理:所有量化模块的起点
很多人上来就写选股策略、写回测逻辑,结果数据源没选好,后面全部白干。行情数据是整个系统的地基,数据不全、复权错误、字段不统一,会让自动选股和策略回测的结果完全失真。我在这块花的时间比想象中多得多,也恰恰是这部分决定了系统能不能稳定跑三个月以上。
1.1 接口选型:免费与积分的取舍
目前 Python 生态里常用的免费/半免费数据接口,主要是 Tushare、AkShare、yfinance 这三类。我最后是 Tushare 为主、AkShare 为辅助,原因是 Tushare 的日线、周线、月线数据和基本面字段结构稳定,适合做自动化定时任务;而 AkShare 数据源多、覆盖面广,但接口变化频繁,不适合作为核心数据链路。
| 数据源 | 优点 | 缺点 | 我的用途 |
|---|---|---|---|
| Tushare | 数据结构稳定,字段规范,积分制免费 | 部分接口有积分门槛,需要注册攒积分 | 日线行情、基本面因子、复权因子 |
| AkShare | 数据源丰富,无需 token,接口灵活 | 接口经常调整,需常更新版本 | 补充实时行情、龙虎榜、公告等辅助信息 |
| yfinance | 海外股票数据方便 | A 股数据有延迟和缺失 | 偶尔看美股,不用于主链路 |
如果你只是自己研究,积分不够的时候可以先用 AkShare 顶着。但要做定时任务、要稳定推送,我建议还是把 Tushare 作为核心,至少把日线级别的数据跑通。另外,数据接口必须加一层缓存,否则每天重复拉全量数据既慢又容易被限流。我用的方案是:每天收盘后拉一次日线数据存到本地 SQLite,界面和选股模块只读本地库,只有实时行情走网络接口。
1.2 预处理环节:复权、停牌与 ST 处理
数据拿到手之后,第一步不是算因子,而是清洗。我遇到最典型的坑就是:不复权的价格数据算出来的动量因子、均线因子全是错的。尤其是分红除权日,股价凭空跳空一块,策略会误判成暴跌或暴涨。
所以我的预处理流程固定是这几步:
- 先用 Tushare 的
adj_factor接口拿复权因子,对收盘价、开盘价做后复权处理。 - 过滤掉 ST、*ST 股票,这些股票的涨跌停幅度和交易规则都特殊,一旦混入选股池会污染结果。
- 剔除上市不满 60 个交易日的次新股,因为历史数据不足,很多因子计算出来没有意义。
- 停牌股票标记出来,不能直接抛弃,否则第二天复牌时的成交量异常会被当成信号。
import pandas as pd import tushare as ts ts.set_token("你的token") pro = ts.pro_api() def load_daily(ts_code, start_date, end_date): df = pro.daily(ts_code=ts_code, start_date=start_date, end_date=end_date) adj = pro.adj_factor(ts_code=ts_code, start_date=start_date, end_date=end_date) df = df.merge(adj[["trade_date", "adj_factor"]], on="trade_date", how="left") df["adj_close"] = df["close"] * df["adj_factor"] df["adj_open"] = df["open"] * df["adj_factor"] return df.sort_values("trade_date")这一层做扎实之后,后面所有模块拿到的都是干净数据。我自己的经验是:宁可每天多花 10 秒做全量复权和过滤,也不要在策略里搞“脏数据也能跑”的妥协,因为错误数据产生的交易信号,会让你在回测里看到完全不存在的收益曲线。
2. 自动选股:多因子打分模块的落地实现
自动选股听起来很高大上,其实本质就是一个“过滤 + 打分 + 排名”的流程。市面上很多讲选股的文章喜欢强调某个神奇指标,但实际工程里,把若干基础因子组合在一起做打分,再用规则过滤掉风险股,效果会更稳定,也更便于后期维护。
2.1 因子池构建与标准化
我实现了一套简单的多因子打分系统,因子池包含四类:
- 估值因子:市盈率(PE)、市净率(PB),用来排除过分高估的标的。
- 动量因子:过去 20 个交易日的涨跌幅,衡量短期市场关注度。
- 质量因子:ROE(净资产收益率),衡量公司盈利能力。
- 波动率因子:过去 20 日收益率的年化标准差,用来控制风险,波动率过高的股票会被降权。
每个因子的量纲不同,不能直接相加。常见的做法是 z-score 标准化,也就是用(原始值 - 均值) / 标准差把因子转换到同一尺度。对于 PE、PB 这类越小越好的反向因子,取负值后再参与打分。
import pandas as pd import numpy as np def zscore(series): return (series - series.mean()) / series.std() def factor_score(df): df["pe_score"] = -zscore(df["pe"]) # 越低越好 df["pb_score"] = -zscore(df["pb"]) df["mom_score"] = zscore(df["ret_20"]) # 动量越高越好 df["roe_score"] = zscore(df["roe"]) df["vol_score"] = -zscore(df["vol_20"]) # 波动越低越好 df["total_score"] = ( df["pe_score"] * 0.2 + df["pb_score"] * 0.1 + df["mom_score"] * 0.3 + df["roe_score"] * 0.25 + df["vol_score"] * 0.15 ) return df.sort_values("total_score", ascending=False)权重怎么定?我没有用特别复杂的优化算法,而是先用等权跑一段时间,再看哪一类因子贡献最大,手动微调。比如我发现动量因子在震荡市里经常帮倒忙,就把它从 0.3 降到 0.2。这种可解释的权重调整,比玄学调参更容易沉淀经验。
2.2 打分排名与结果落地
自动选股不是只出前 10 名就结束的,还需要考虑行业分散度、避免同一板块扎堆。所以我在排名之后会做一次行业去重:同一行业最多保留 3 只,分散风险。这个逻辑在实盘体验中非常关键,全仓挤在同一个板块,回撤会非常吓人。
选股结果我存到 SQLite 的一张stock_pool表里,字段包括:股票代码、股票名称、行业、总分、每个因子的明细值、选股日期。这样做有几个好处:
- 可以在界面上直接展示每日选股结果,方便复盘。
- 可以回溯某一天为什么选出了某只股票,因子占比一目了然。
- 后续做回测时,可以直接把历史选股结果作为“持仓列表”输入,不用重新算一遍因子。
关于选股频率,我的建议是不要做得太激进。日频选股会导致交易频繁、手续费侵蚀利润;周频或双周频更现实。实盘经验是不如回测理想,其中很大一部分原因是回测时忽略了换仓成本和滑点,这一点在后面回测模块里还会进一步讨论。
3. 策略回测:从信号生成到资金曲线计算的完整链路
交易策略测试是系统的核心模块之一,目的是用历史数据验证策略是否可行,而不是拿着真金白银去试错。我早期犯过一个错误:只看了总收益率和年化收益率,没有关注最大回撤和夏普比率,结果某段历史行情里收益看起来不错,换成另一段行情就亏得一塌糊涂。
3.1 向量化回测与事件驱动回测怎么选
回测框架可以分成两类:向量化回测和事件驱动回测。
- 向量化回测:把整个时间序列作为数组运算,代码短、执行快、适合均线、动量这类规则清晰的策略。缺点是难以处理复杂的资金管理、分批建仓、涨跌停无法成交等情况。
- 事件驱动回测:按每个交易日逐个处理数据,每个 tick 或每个 bar 触发一次信号回调,更接近真实交易系统。缺点是代码量大、运行慢,但扩展性极好。
我推荐的落地方式是:策略研究阶段用向量化回测快速验证想法,确定有效后,再在事件驱动框架里做精细化模拟,加入手续费、滑点、涨跌停和停牌处理。如果一上来就搭事件驱动框架,光调试撮合逻辑就会耗尽精力,很难快速验证策略。
3.2 均线策略的代码实现与指标计算
下面是一个双均线策略的简化向量化回测实现。核心逻辑是:5 日均线上穿 20 日均线时买入,下穿时卖出。
import pandas as pd import numpy as np def backtest_double_ma(df, short=5, long=20, fee_rate=0.0003): df = df.copy() df["ma_short"] = df["adj_close"].rolling(short).mean() df["ma_long"] = df["adj_close"].rolling(long).mean() df["position"] = 0 # 上穿买入,下穿卖出 df.loc[df["ma_short"] > df["ma_long"], "position"] = 1 df.loc[df["ma_short"] <= df["ma_long"], "position"] = 0 df["position"] = df["position"].diff().fillna(0) # 只在信号发生日产生交易成本 df["trade_cost"] = df["position"].abs() * fee_rate df["daily_return"] = df["adj_close"].pct_change().fillna(0) df["strategy_return"] = df["position"].shift(1) * df["daily_return"] - df["trade_cost"] df["nav"] = (1 + df["strategy_return"]).cumprod() return df回测之后一定要计算这几个指标,缺一不可:
- 累计收益率:
nav[-1] - 1 - 年化收益率:
(nav[-1]) ** (252 / len(df)) - 1 - 最大回撤:
max(1 - nav / nav.cummax()) - 夏普比率:
annual_return / annual_volatility
最大回撤尤其重要。我见过很多策略年化收益很高,但某一段回撤超过 40%,这样的策略在实盘里根本扛不住,人的心理承受不了。我在系统里会专门画一条资金曲线和回撤曲线,就是为了直观看到“最坏的时候亏多少”。
3.3 手续费、滑点和涨跌停对回测的影响
回测看着盈利,实盘却亏损,最常见的元凶就是没有充分计入摩擦成本。我自己的处理方式是:
- 手续费:按双边万三计算,这已经算比较宽松了,实际上还有印花税和过户费。
- 滑点:每次买卖再按成交价的 0.1% 扣掉,对于流动性差的股票,这个比例可能还不够。
- 涨跌停:如果当日涨停,买单无法成交;跌停,卖单无法成交。回测里如果不处理,会把很多“买入后涨停”的理想情景算进去,严重高估收益。
一个非常真实的情况是:双均线策略在牛市回测效果很好,因为趋势一旦形成,连续上涨很容易被捕获。但到了震荡市,均线会反复交叉,每一次交叉都产生手续费,这就是“两头挨打”。我在策略测试模块里专门做了一组参数对比表格,把均线周期从 5/10、5/20、10/30、20/60 全部跑一遍,再对比不同市场状态下的表现,这样做能非常直观地看出策略的脆弱点。
4. PyQt 界面设计:线程模型、信号槽与实时更新
图形界面是整个系统从“能用”到“好用”的分水岭。我早期只有命令行脚本时,每天要看选股结果就得翻终端输出,后来用 PyQt 做了界面,一切才变得直观。但 PyQt 编程有一个必须跨过去的坎——线程模型。
4.1 为什么不能把耗时任务丢进主线程
PyQt 的主线程是事件循环,所有界面刷新都在这个线程里执行。如果你在按钮的点击回调里直接调用选股函数或回测函数,界面就会整个卡死,用户拖都拖不动窗口,看起来像程序崩溃了。原因是选股和回测都是 CPU 密集型任务,一旦长时间占据主线程,Qt 就没机会处理重绘事件。
解决方式是用QThread把耗时任务放到后台线程,任务完成后通过pyqtSignal把结果发回主线程更新界面。这是 PyQt 开发的黄金法则:所有耗时操作一律进工作线程,所有界面更新一律在主线程通过信号完成。
4.2 用 QThread 封装选股与回测任务
我封装了一个通用的Worker类,任何耗时任务都可以往里丢。
from PyQt6.QtCore import QThread, pyqtSignal class Worker(QThread): finished = pyqtSignal(object) error = pyqtSignal(str) def __init__(self, fn, *args, **kwargs): super().__init__() self.fn = fn self.args = args self.kwargs = kwargs def run(self): try: result = self.fn(*self.args, **self.kwargs) self.finished.emit(result) except Exception as e: self.error.emit(str(e))使用方式很简单,在按钮回调里启动线程:
def on_start_select_stock(self): self.worker = Worker(self.run_select_stock, self.params) self.worker.finished.connect(self.on_select_done) self.worker.error.connect(self.on_select_error) self.worker.start() self.status_label.setText("选股中,请稍候...") def on_select_done(self, result): self.table_widget.update_data(result) self.status_label.setText("选股完成")这里有一个非常容易踩的坑:Worker对象必须保存为实例属性,不能写成局部变量。否则线程对象会被 Python 的垃圾回收机制回收,任务跑到一半就没下文了。我刚开始写 PyQt 时遇到的问题基本都是这么来的。
4.3 界面组件布局与行情表格刷新
系统界面我采用的是左右分栏布局:
- 左侧:参数配置区域,包含选股因子权重、均线周期、回测区间等输入项,每次操作结束后都会把参数保存到本地 JSON 配置文件里。
- 右侧:主展示区域,用
QTabWidget分成三个页签——选股结果、策略回测、实时信号。
选股结果用QTableWidget展示,列包括代码、名称、行业、总分、PE、PB、20 日涨幅等。实时行情刷新用QTimer定时器完成,每 5 秒拉一次最新价并刷新指定单元格。这里要特别注意的是:QTimer回调里只做轻量级更新,如果网络请求有延迟,可以单独开一个QThread负责请求,然后通过信号更新 UI,不要阻塞 timer。
我做得比较顺手的一个交互是:双击选股结果的某一行,自动跳转到该股票的 K 线查看页。K 线图我用的pyqtgraph的CandlestickItem来绘制,比 matplotlib 集成进 Qt 更流畅,缩放和拖拽都很顺手。
5. 实时通知与用户互动:系统真正“可用”的关键环节
选股结果出来、回测跑完,这系统还只能算“半成品”。真正让它变成一个能每天使用的工具,靠的是实时通知和用户互动。前者解决“定时跑完之后我怎么知道”,后者解决“能不能不动代码就改参数”。
5.1 两种实时通知实现方式:轮询与订阅推送
实时通知的实现主要有两条路:
- 轮询拉取:定时去行情接口查最新价格,如果满足条件就触发通知。实现简单,但实时性和延迟受轮询频率限制。
- 订阅推送:通过 WebSocket 或第三方平台的消息订阅机制,实时收到行情变化。实现复杂,需要维护长连接,但对时效性极高。
我采用的方案是把两者结合起来:平时用 30 秒一次的轮询检查持仓池和自选股池的涨跌状态;对于某些重点监控的股票,如果涨幅超过阈值,立即触发推送通知。阈值和监控列表都可以在界面里配置,不用改代码。
5.2 钉钉/企业微信 webhook 推送实现
推送渠道我优先推荐企业微信机器人和钉钉机器人,因为配置简单,搜索结果就是一个 Webhook URL,用requests.post就能发消息。不推荐自己搭邮件服务,容易被运营商拦截,而且配置 SMTP 对很多小白来说也挺劝退。
import requests def send_dingtalk_message(webhook_url, content): payload = { "msgtype": "text", "text": {"content": content}, } requests.post(webhook_url, json=payload)我实际使用中发现,推送消息一定不要贪多。如果每 5 分钟就把所有满足条件的股票推一遍,用不了几天你的手机就会被通知轰炸到麻木。我设置了冷静时间:同一只股票触发通知后,30 分钟内不再重复推送。这个机制看似简单,却大大提升了通知的有效性。
5.3 参数交互面板:让策略调参不再改代码
用户互动这块,我认为最实用的是参数面板。我最初做配置时直接写在配置文件里,每次调参都要重新启动程序,非常影响调试节奏。后来我把选股权重、回测参数、通知阈值全部暴露到界面左侧面板,用户修改后点击“保存参数”就写入本地config.json,点击“立即执行”就触发重新选股或回测。
更重要的是,回测结果展示时做成了交互式图表:底下有一个 QComboBox 可以切换不同参数组合,上方图表会同步变化资金曲线和回撤曲线。这样横向对比参数的效果非常直观,不需要每次都在脚本里改数然后重启重跑。
这一块的核心设计思路是:系统要允许用户通过界面完成日常 80% 的操作,只有新增因子、新增策略这类结构性改动才需要写代码。把这一层做出来之后,你会发现整个工具的使用频率会高很多,因为“跑一次选股”的成本从改脚本、等启动、看输出,变成了点一个按钮、等几秒、看表格。
6. 打包发布与运行稳定性:从开发机到生产环境
系统在开发环境跑得好好的,不代表换个电脑也能跑起来。我把这套系统分发给朋友使用时,最大的障碍就是环境搭建。Python 版本不同、依赖缺失、路径不对,任何一处都会让程序当场崩溃。后来我花了不少时间研究 PyInstaller 打包和运行守护,这块经验我认为比写策略本身更值钱。
6.1 PyInstaller 打包 exe 的实测记录
PyInstaller 是打包 Python 程序的主流工具。我的打包命令和注意事项如下:
pyinstaller -F -w --name QuantSystem --hidden-import=pyqtgraph main.py-F:打包成单文件,分发方便,但启动稍慢,因为要解压到临时目录。-w:windowed 模式,运行时不弹出控制台窗口。--hidden-import:显式声明动态导入的模块,防止 PyInstaller 漏掉。
如果项目依赖太多,导致单文件打包后体积巨大且启动慢,可以改成-D目录模式。目录模式下启动速度更快,而且不容易误报杀毒软件。朋友那台电脑上实测,单文件模式 2 秒内能弹出界面,目录模式接近瞬时启动。
打包时最麻烦的是数据文件,比如选股结果数据库、配置文件、图标资源。需要把这类文件放在sys._MEIPASS临时目录下,否则打包后找不到文件。最典型的报错就是FileNotFoundError: config.json,我当时排查了很久才发现是路径问题。
6.2 日志记录与程序守护
系统跑起来之后,稳定性比功能更重要。我加了两层保障:
第一层是日志系统,使用 Python 的logging模块,把运行日志同时写入文件和输出到控制台(如果是在终端中运行)。日志要分级,INFO 记录每日选股和回测的开始结束,WARNING 记录接口限流和数据缺失,ERROR 记录崩溃堆栈。
第二层是程序守护。如果是部署在 Windows 服务器上,我建议直接做成 Windows 计划任务,每天开盘前自动启动程序,收盘后自动执行选股和回测。如果是在普通开发机上跑,那就需要保证程序崩溃后能快速重启。我写了简单的看门狗脚本,每隔 5 分钟检测一次程序进程是否存在,不存在就重新拉起。
import subprocess import time while True: result = subprocess.run( ["tasklist"], capture_output=True, text=True ) if "QuantSystem.exe" not in result.stdout: subprocess.Popen(["QuantSystem.exe"]) time.sleep(300)说实话,我自己的体会是,量化系统能不能长期稳定运转,比的不是策略多高深,而是“明天打开电脑它还能不能工作”。我踩过最大的坑不是选股模型不赚钱,而是数据源接口某天突然返回空数据、程序直接异常退出,第二天早上什么通知都没收到。所以稳定性这部分,请务必当成核心功能来对待。
最后再分享一条我从这套系统里沉淀出来的小技巧:开发顺序一定要按照“数据底层 → 选股 → 回测 → 界面 → 通知 → 打包”来推进,每层都先跑通再往上一层。不要一开始就想着把界面做得花团锦簇,界面背后没有验证过的数据和逻辑,最后只会变成一套好看但不能用的空壳。先把单条链路跑通,再逐步完善细节,这套工具才会真正成为你每天都愿意打开的东西。
本文还有配套的精品资源,点击获取