简介这是一套面向量化投资初学者与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_codets_code, start_datestart_date, end_dateend_date) adj pro.adj_factor(ts_codets_code, start_datestart_date, end_dateend_date) df df.merge(adj[[trade_date, adj_factor]], ontrade_date, howleft) 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, ascendingFalse)权重怎么定我没有用特别复杂的优化算法而是先用等权跑一段时间再看哪一类因子贡献最大手动微调。比如我发现动量因子在震荡市里经常帮倒忙就把它从 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, short5, long20, fee_rate0.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, jsonpayload)我实际使用中发现推送消息一定不要贪多。如果每 5 分钟就把所有满足条件的股票推一遍用不了几天你的手机就会被通知轰炸到麻木。我设置了冷静时间同一只股票触发通知后30 分钟内不再重复推送。这个机制看似简单却大大提升了通知的有效性。5.3 参数交互面板让策略调参不再改代码用户互动这块我认为最实用的是参数面板。我最初做配置时直接写在配置文件里每次调参都要重新启动程序非常影响调试节奏。后来我把选股权重、回测参数、通知阈值全部暴露到界面左侧面板用户修改后点击“保存参数”就写入本地config.json点击“立即执行”就触发重新选股或回测。更重要的是回测结果展示时做成了交互式图表底下有一个 QComboBox 可以切换不同参数组合上方图表会同步变化资金曲线和回撤曲线。这样横向对比参数的效果非常直观不需要每次都在脚本里改数然后重启重跑。这一块的核心设计思路是系统要允许用户通过界面完成日常 80% 的操作只有新增因子、新增策略这类结构性改动才需要写代码。把这一层做出来之后你会发现整个工具的使用频率会高很多因为“跑一次选股”的成本从改脚本、等启动、看输出变成了点一个按钮、等几秒、看表格。6. 打包发布与运行稳定性从开发机到生产环境系统在开发环境跑得好好的不代表换个电脑也能跑起来。我把这套系统分发给朋友使用时最大的障碍就是环境搭建。Python 版本不同、依赖缺失、路径不对任何一处都会让程序当场崩溃。后来我花了不少时间研究 PyInstaller 打包和运行守护这块经验我认为比写策略本身更值钱。6.1 PyInstaller 打包 exe 的实测记录PyInstaller 是打包 Python 程序的主流工具。我的打包命令和注意事项如下pyinstaller -F -w --name QuantSystem --hidden-importpyqtgraph main.py-F打包成单文件分发方便但启动稍慢因为要解压到临时目录。-wwindowed 模式运行时不弹出控制台窗口。--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_outputTrue, textTrue ) if QuantSystem.exe not in result.stdout: subprocess.Popen([QuantSystem.exe]) time.sleep(300)说实话我自己的体会是量化系统能不能长期稳定运转比的不是策略多高深而是“明天打开电脑它还能不能工作”。我踩过最大的坑不是选股模型不赚钱而是数据源接口某天突然返回空数据、程序直接异常退出第二天早上什么通知都没收到。所以稳定性这部分请务必当成核心功能来对待。最后再分享一条我从这套系统里沉淀出来的小技巧开发顺序一定要按照“数据底层 → 选股 → 回测 → 界面 → 通知 → 打包”来推进每层都先跑通再往上一层。不要一开始就想着把界面做得花团锦簇界面背后没有验证过的数据和逻辑最后只会变成一套好看但不能用的空壳。先把单条链路跑通再逐步完善细节这套工具才会真正成为你每天都愿意打开的东西。本文还有配套的精品资源点击获取
