简介这套Python编写的股票量化交易系统源码基于PyQt构建可视化界面面向具备一定Python基础、希望搭建自动选股、策略回测及实盘交易框架的开发者与量化学习者。压缩包共327个文件、大小约2.35MB以301个py源码文件为主体辅以界面图片、MD/PDF说明文档、验证码识别JAR包以及项目配置工程文件便于独立研读与直接复用。已有119人学习下载。系统覆盖数据管理、选股、策略回测和实盘交易四大核心模块实盘与回测共用同一套策略代码并提供实盘及选股策略编写模板。历史数据可自动拉取至MongoDB涵盖股票代码、交易日、个股/指数ETF的日线与分笔数据同时支持微信提醒与交互、一键挂机、9个模拟账号的模拟交易等功能是构建个人量化交易系统的实用参考。1. 用Python和PyQt自建量化交易系统先想清楚你写的是工具还是产品“股票量化交易系统”在网上的开源项目一抓一大把但真正能让人用起来的不到三成。我见过太多从免费Python源码仓库里拉下来的PyQt项目装好依赖一跑选股按钮点下去界面直接转圈十几秒回测跑完日志糊满一屏微信通知压根没动静。问题多半不在策略而在架构界面线程、策略线程、数据源谁都没说清楚该怎么配合。这篇我会从PyQt界面骨架讲到自动选股、策略测试和微信通知与互动每一层都给出能直接复现的代码和参数最后把实盘前最要命的几个坑列出来。适合手里有现成策略、想把它串成一个桌面工具的人也适合刚入门、想找一个完整落地路径的Python开发者。2. 系统骨架PyQt界面、策略线程与数据流怎么拧在一起2.1 线程分离是PyQt量化工具的第一道门槛凡是跑过下载的免费Python源码的人大概率见过这个场面点击“开始选股”界面卡死标题栏变成“未响应”过十几秒缓过来结果一下刷出几十条日志。这不是电脑慢是耗时任务被塞进了GUI线程。PyQt的消息循环一次只能干一件事你让它去遍历3000只股票的K线它就没空重绘按钮和窗口。正确做法是把耗时逻辑扔进QThread再用信号把结果传回界面。信号槽是线程安全的能跨线程调用这是PyQt提供的正规通道。不要在工作线程里直接操作QLabel、QTableWidget这些控件也不要图省事用全局变量加锁去传状态——后期加一个选股任务、加一个回测任务全局变量会让你改到怀疑人生。我一般会给每个耗时任务单独建一个Worker类继承QThread把任务逻辑放在run()里只通过自定义信号对外发消息。界面端拿到信号后再更新控件。这套结构写起来不复杂但能一次性解决掉“界面卡死”、“日志乱序”、“任务重复启动”这三个最常见的问题。2.2 数据源选型tushare、baostock、akshare怎么挑量化系统离了数据就是空壳。选数据源时很多人只看接口数量忽略了积分、频率和复权因子。我这边常用的三选一先看一张对比表。数据源免费额度日线分钟线复权因子稳定性baostock全免费支持有前/后/不复权稳定接口老但够用tushare pro积分制新号积分低支持有有数据全但积分不够时受限akshare全免费支持部分部分接口多经常随网页改版波动如果你做的是A股日线级别的自动选股和策略测试baostock是最省心的登录即用不卡积分复权因子直接给到前复权、后复权、不复权三个选项字段也干净。tushare的数据质量确实好但积分不够时每分钟调用次数卡得很难受不适合做盘后批量扫描几百只股票。akshare胜在接口全但同一个接口可能这周能用、下周就报错用在长期跑批的自动选股模块里会让你不断返工。选型时还要注意数据缓存。不要每次选股都全量拉数据建议把日线数据按股票代码存成本地文件或SQLite每日增量更新。这一方面省流量另一方面也降低对数据源接口频率的限制。把数据落地到本地之后选股模块和回测模块就可以共用同一份数据避免界面和策略拿到两个不一样的数据集。2.3 最小可跑骨架选股线程加状态栏回调光说没用先把一个最小骨架搭起来。这个骨架只有一个按钮和一个状态栏点击按钮后启动一个后台线程线程跑完通过信号把消息传回界面更新状态栏。import sys import time from PyQt5.QtCore import QThread, pyqtSignal from PyQt5.QtWidgets import QApplication, QMainWindow, QPushButton, QLabel, QVBoxLayout, QWidget class SelectWorker(QThread): # 线程向界面回传消息的唯一出口信号 progress pyqtSignal(str) def run(self): # 这里放选股、回测这类耗时逻辑 for i in range(20): time.sleep(0.1) self.progress.emit(f正在遍历第 {i 1} 个候选标的...) self.progress.emit(选股完成) class MainWindow(QMainWindow): def __init__(self): super().__init__() self.worker None self.status_label QLabel(就绪) self.start_btn QPushButton(开始选股) self.start_btn.clicked.connect(self.start_select) layout QVBoxLayout() layout.addWidget(self.start_btn) layout.addWidget(self.status_label) container QWidget() container.setLayout(layout) self.setCentralWidget(container) def start_select(self): # 防止重复点击导致多个线程同时跑 if self.worker and self.worker.isRunning(): return self.worker SelectWorker() self.worker.progress.connect(self.status_label.setText) self.worker.start() if __name__ __main__: app QApplication(sys.argv) window MainWindow() window.show() sys.exit(app.exec_())这段代码里最关键的是第11行的progress pyqtSignal(str)和第34行的self.worker.progress.connect(...)。信号携带一个字符串参数线程里emit出去的消息会自动排队送到主线程由主线程更新控件这就避免了跨线程操作控件导致的崩溃和闪烁。start_select里的isRunning()判断必须加否则快速连点两次按钮会启动两个线程同时扫股票轻则重复推送重则把数据库写串。保存worker引用也很重要如果直接SelectWorker().start()而不存引用这个线程随时可能被垃圾回收掉。跑通这个骨架后后面加自动选股、回测、微信推送模块只需要按同样模式新建Worker类并注册信号界面部分基本不用动。这就是把工具当产品写的起点。3. 自动选股模块把选股因子变成可落地的定时任务3.1 指标计算与因子筛选别把未来函数带进选股有了数据源和线程骨架接下来做自动选股。选股的核心是把“我想买什么”翻译成可计算的条件。以最常见的均线多头排列加量比放大为例逻辑很清楚5日均线在20日均线上方且当日成交量比过去5日的均量放大1.5倍以上。这个逻辑写起来不难难在别算出未来数据。很多人用pandas的rolling窗口时忽略了shift导致当天信号用到了当天的收盘价这在回测里是作弊在实盘里会直接导致选股结果和次日走势完全对不上。正确做法是计算完指标后统一用iloc[-1]取最后一个已完成交易日的值并且所有指标只基于当前时刻之前的数据。下面的代码用baostock拉取日线并计算两个基础因子import baostock as bs import pandas as pd lg bs.login() # 获取单只股票的前复权日线 rs bs.query_history_k_data_plus( sh.600000, date,code,close,volume, start_date2024-01-01, end_date2024-12-31, frequencyd, adjustflag2, # 1后复权 2前复权 3不复权 ) rows [] while rs.error_code 0 and rs.next(): rows.append(rs.get_row_data()) df pd.DataFrame(rows, columnsrs.fields) bs.logout() # 因子计算 df[close] df[close].astype(float) df[volume] df[volume].astype(float) df[ma5] df[close].rolling(5).mean() df[ma20] df[close].rolling(20).mean() df[vol_base] df[volume].rolling(5).mean().shift(1) # 避开当日量 df[vol_ratio] df[volume] / df[vol_base]adjustflag2表示前复权这是处理分红送转的标准姿势。如果不复权一只股票除权后K线会跳空均线系统直接失真。shift(1)是这里最容易踩的坑它把量能基准推到前一日保证当日信号不包含当日成交量的反馈偏差。筛选时只取最后一行的数据做判断因为前面所有行都是历史只有最后一根K线是“当下”。如果拿整段数据里的最大值或末日数据去选股等于提前知道了答案。3.2 定时调度与结果落库让选股每天自己跑选股逻辑写好还不够自动选股的重点在“自动”。常见做法是APScheduler做cron定时每个交易日早上开盘前跑一次把结果写进SQLite。这样你打开界面时能看到的是“昨天的选股结果”而不是现场慢慢扫。from apscheduler.schedulers.background import BackgroundScheduler def run_daily_select(): # 这里调用选股主流程把结果写入sqlite print(开始执行自动选股任务) # select_stocks_and_save() scheduler BackgroundScheduler() # 周一到周五的 9:35 执行一次 scheduler.add_job( run_daily_select, cron, day_of_weekmon-fri, hour9, minute35, misfire_grace_time300, ) scheduler.start()misfire_grace_time300是容易被忽略的参数意思是任务原定执行时间点错过了5分钟之内仍然补执行。电脑休眠、系统卡顿都会导致定时任务错过触发点没有这个参数今天没开盘选股结果就是早上空跑一整天。定时任务不能只依赖cron表达式还要处理交易日历。A股有法定节假日和调休周一到周五不一定是交易日。最可靠的办法是维护一份交易日列表比如用baostock的query_trade_dates接口拉全年交易日在任务回调里先判断当天是不是交易日不是就直接返回。不要指望cron表达式自己能分辨清明节和国庆节。选股结果落库我建议建两张表一张存选股日期和股票代码另一张存当天的因子快照。理由很实在如果你只存结果股票后面想复盘“为什么选了这只没选那只”时没有依据。因子快照能让你回看历史也能在策略调整后做对比。3.3 别惦记通达信自动选股的配置文件自己在Python里跑更透明很多人在搜索“通达信自动选股设置在哪个文件”其实通达信有它自己的公式系统配置文件可以设置公式选股但那个机制有两个先天问题一是它在软件内部运行你没法方便地拿到选股结果去做二次处理二是公式语言调试起来黑匣子一样出错了你连日志都没得看。如果你已经走上Python这条路完全没必要把选股逻辑写回通达信公式里。我自己做过一次迁移把通达信公式里的条件选股翻译成Python pandas逻辑总共不过几十行代码但换来的东西是三个实打实的好处选股结果能落库能自动化定时跑能直接对接下一章的回测模块。通达信的公式系统更适合作盘中人工参考而自动选股、策略测试这类场景用Python生态去处理才是主流做法。还有一个细节自动选股的频率不要太高。日线级别的因子一天跑一次足够了。把频率调到分钟级意味着你要处理分钟数据的复权、停牌、集合竞价复杂度翻好几倍而且大部分散户策略根本不依赖分钟级选股。4. 策略测试回测不是跑通历史就完事4.1 自写循环还是Backtrader两条路各有代价策略测试是量化系统里最容易被高估的一环。很多人把“回测曲线很漂亮”等同于“策略能赚钱”实际上回测框架选错、参数填错结果会失真到完全没参考价值。先决定用什么框架自己写循环还是用现成库。自己写循环的优点是逻辑完全透明每一笔成交都能控制缺点是要处理复权、停牌、涨跌停、成交价格偏离所有细节都得自己兜。Backtrader这类现成框架把撮合规则、滑点、手续费都封装好了但学习曲线陡内部机制隔了一层出问题不好排查。维度自写循环Backtrader逻辑透明程度完全透明中等需读源码手续费/滑点控制需自己实现内置commission与slippage分钟级回测工作量很大支持但数据格式要求严格调试难度出错直接定位报错堆栈较绕上手速度快适合验证单策略慢但功能全我的选择标准是单策略验证用自写循环跑通后要做参数扫描或多策略对比再迁到Backtrader。不要一开始就上Backtrader很多新手会因为它的数据格式要求和订单状态机设计而卡住也不要永远自写循环等你需要计算夏普比率、最大回撤、连续亏损时现成框架里这些都是一个函数的事。4.2 回测参数里的隐形成本手续费、滑点与复权回测的收益不等于实盘收益最大的偏差来源就是成本。以下面的双均线策略为例它能在历史数据上频繁交易但每多交易一次手续费和滑点就多刮一层。import backtrader as bt class GoldenCross(bt.Strategy): params dict(fast5, slow20) def __init__(self): self.ma_fast bt.indicators.SMA(self.data.close, periodself.p.fast) self.ma_slow bt.indicators.SMA(self.data.close, periodself.p.slow) self.crossover bt.indicators.CrossOver(self.ma_fast, self.ma_slow) def next(self): if not self.position: if self.crossover 0: size int(self.broker.getcash() * 0.95 / self.data.close[0]) self.buy(sizesize) elif self.crossover 0: self.close() cerebro bt.Cerebro() cerebro.addstrategy(GoldenCross) # 假设 data 是已经读取并复权的数据源 # cerebro.adddata(data) cerebro.broker.setcash(100000.0) cerebro.broker.setcommission(commission0.0003, mult10.0) # 万三佣金 cerebro.addsizer(bt.sizers.PercentSizer, percents95) result cerebro.run()setcommission里的mult10.0表示A股每手100股佣金按成交金额和手数计算。不要忽略这个参数否则一手股票的手续费会被当成一股来算误差放大十倍。PercentSizer按账户资金的95%开仓是为了接近实际仓位而不是满仓梭哈。真正让回测虚高的是滑点。Backtrader默认滑点为零这等于假设你每次都能以收盘价成交。现实是买的时候要高一点买卖的时候要低一点卖。我一般加上固定滑点cerebro.broker.set_slippage_perc(perc0.001) # 单边千一滑点千一的滑点加万三的佣金一个完整的买卖来回成本将近千分之三。高频策略如果没把这笔账算进去回测收益可能比实盘高出一倍甚至更多。复权数据在回测里同样关键。用前复权数据回测时要注意复权因子是动态变化的新一次除权发生后之前所有历史价格会被重新计算。这意味着你昨天的回测结果可能因为今天出现一次新的分红送转而整体改变。常见做法是固定一个“截至某个日期”的复权因子别让它随最新行情漂移否则你的回测结果每天都在变怎么复盘都对不上。4.3 回测结果怎么验证可信度分段和样本外跑完一次回测曲线好看不代表策略成立。先做分段验证把历史数据切成两段前一段用来调参数后一段用来验证。如果后一段的表现明显差于前一段多半是过拟合。这是最费时间但是最值得做的一步。还可以做一个更硬核的检查把参数随机扰动一下比如快线从5改成4或6慢线从20改成30或40如果收益曲线剧烈塌陷说明策略对参数极度敏感实盘基本没法用。真正靠谱的策略在参数邻域内应当保持类似的盈利逻辑而不是只在某个精确数字上灵光一现。我见过不少人在这一步翻车策略在海量参数扫描里挑出了历史收益最高的组合兴冲冲上实盘结果连续亏损。原因很朴素——那一组参数精确拟合了过去的噪声。策略测试的目标是找到“稳定的行为模式”不是“最大的历史收益”。5. 微信通知与互动把交易状态推到手机再把指令收回来5.1 微信通知的三条通道怎么选选股结果和回测结论做好了接微信通知让结果跑到手机上这一步很多人卡住。个人微信自动化这块早期有人用itchat网页协议但已经大面积失效而且风控严重不建议碰。目前稳定靠谱的通道只有两个方向一是企业微信群机器人webhook二是Server酱这类推送服务。企业微信群机器人的好处是免费、不限制个人微信号、可以在群里刷markdown消息而且收件人不限一个。你把自己的量化机器人和几个信得过的朋友拉一个群每天推送选股结果和交易信号这是最贴近“实时交易微信通知”的落地方式。Server酱的逻辑是推送到你自己的微信服务号配置简单适合只给自己看。通道是否免费推送形式限频是否支持互动企业微信群机器人免费群内markdown每分钟20条不支持Server酱免费版够用服务号消息每日有限额支持表单回执个人微信自动化免费但风险高单聊/群聊易封号支持但不稳定实时交易通知的实际瓶颈不在功能而在频率。策略一天最多触发几次交易每分钟20条完全够用。真正要防的是自己的程序出错疯狂刷消息比如策略在循环里没加节流故障时十几分钟推上百条直接被限频拉黑。5.2 用webhook推成交与风控消息企业微信群机器人推送的代码很简单关键是消息内容要精简。我一般只推三类消息选股结果、策略买卖信号、风控预警。每类用不同的前缀方便在群里扫一眼就定位。import requests def push_wx(title: str, content: str) - None: webhook https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key你的key payload { msgtype: markdown, markdown: { content: f## {title}\n{content} } } try: resp requests.post(webhook, jsonpayload, timeout5) data resp.json() if data.get(errcode) ! 0: print(f微信推送失败: {data}) except Exception as e: print(f微信推送异常: {e}) # 使用示例 push_wx(买卖信号, **买入** 浦发银行 600000\n价格: 10.85\n理由: 均线金叉量比放大)timeout5要加上webhook请求偶尔会慢不设超时会把策略线程卡住。这里要注意推送本身失败不应影响主交易流程所以我用try-except把它包住只打日志不抛出异常。实际运行中推送失败很常见群被解散、webhook被回收、网络抖动。一个实用的兜底方案是把推送消息同时追加到本地日志文件这样微信通道挂了你还能在复盘时看到当时到底有没有触发信号。5.3 从单向通知到双向互动本地小服务收指令微信通知做到位了很多人会发现下一个需求是“互动”能不能在手机上让系统停一下、跑一次回测、看看持仓。企业微信机器人不支持接收消息Server酱虽然支持表单回执但互动能力有限。更通用的方案是让量化系统本地起一个HTTP小服务你通过微信里的链接访问或发送指令服务端接收后控制策略的运行状态。我用FastAPI做了一个极简控制接口策略侧在关键循环里检查一个全局标志位from fastapi import FastAPI from pydantic import BaseModel app FastAPI() control_flag {paused: False, run_backtest: False} class Command(BaseModel): action: str app.post(/cmd) def handle(cmd: Command): if cmd.action pause: control_flag[paused] True return {ok: True, msg: 策略已暂停控制} if cmd.action resume: control_flag[paused] False return {ok: True, msg: 策略已恢复} if cmd.action backtest: control_flag[run_backtest] True return {ok: True, msg: 回测任务已触发} return {ok: False, msg: f未知指令: {cmd.action}}策略线程每次迭代时只检查标志位不直接调用“暂停”这样既避免锁竞争也让控制的粒度可控。这个服务只监听127.0.0.1外部访问怎么进来常见做法是把你本机的服务通过局域网地址暴露给手机或者用内网穿透类工具映射出一个临时地址。这里要注意安全控制接口只建议绑定本机回环地址防局域网内其他人乱发指令。有了这个控制接口后微信通知就真正闭环了——通知触达手机指令从手机收回系统变成一个可以远程操作的交易终端。6. 实盘前的避坑清单5条踩过的坑和一条验证路径界面假死。现象点按钮后窗口变白、标题栏提示未响应。原因耗时逻辑直接跑在PyQt主线程。解决一切耗时任务进QThread通过信号回传界面。这条也是免费Python源码里最常见的问题拿到别人项目先看线程怎么写的能省很多排查时间。回测收益虚高。现象回测年化翻倍实盘连续亏损。原因没设滑点、手续费按单边忽略、复权方式不对。解决setcommission加set_slippage_perc复权因子固定日期回测成本按万三佣金加千一滑点起步。停牌股和涨跌停股买不进。现象信号触发了但界面提示下单失败。原因实盘撮合时停牌股无法成交涨停股买不进跌停股卖不出。解决下单前先检查交易状态和涨停跌幅。回测时也要把这些条件模拟进去否则回测里全成交实盘里成不了。微信推送限频与漏发。现象某天群里突然几十条重复消息之后推送全部失效。原因策略异常导致循环内重复推送触发企业微信限频。解决推送加节流和失败重试异常情况只推一条预警其余写本地日志。历史回测与实盘对不上。现象回测该买没买、该卖没卖实盘成交价格总是更差。原因数据源不一致或策略里存在未来函数。解决选股和回测共用同一份本地数据指标计算统一加shift避免未来数据。最后说一条验证路径先用模拟盘跑两周每笔信号和推送都留日志每天收盘后对比模拟盘成交和回测信号。两周下来如果日志和实盘成交基本吻合再考虑投入真金白银。我自己吃过一次亏——回测跑通直接上实盘结果数据源复权口径不一致策略前三个月完全跑偏。从那以后模拟盘和日志核对成了雷打不动的流程。这条验证路径能帮你把前面五个坑一次性兜住希望帮到你。本文还有配套的精品资源点击获取
简介这套Python编写的股票量化交易系统源码基于PyQt构建可视化界面面向具备一定Python基础、希望搭建自动选股、策略回测及实盘交易框架的开发者与量化学习者。压缩包共327个文件、大小约2.35MB以301个py源码文件为主体辅以界面图片、MD/PDF说明文档、验证码识别JAR包以及项目配置工程文件便于独立研读与直接复用。已有119人学习下载。系统覆盖数据管理、选股、策略回测和实盘交易四大核心模块实盘与回测共用同一套策略代码并提供实盘及选股策略编写模板。历史数据可自动拉取至MongoDB涵盖股票代码、交易日、个股/指数ETF的日线与分笔数据同时支持微信提醒与交互、一键挂机、9个模拟账号的模拟交易等功能是构建个人量化交易系统的实用参考。1. 用Python和PyQt自建量化交易系统先想清楚你写的是工具还是产品“股票量化交易系统”在网上的开源项目一抓一大把但真正能让人用起来的不到三成。我见过太多从免费Python源码仓库里拉下来的PyQt项目装好依赖一跑选股按钮点下去界面直接转圈十几秒回测跑完日志糊满一屏微信通知压根没动静。问题多半不在策略而在架构界面线程、策略线程、数据源谁都没说清楚该怎么配合。这篇我会从PyQt界面骨架讲到自动选股、策略测试和微信通知与互动每一层都给出能直接复现的代码和参数最后把实盘前最要命的几个坑列出来。适合手里有现成策略、想把它串成一个桌面工具的人也适合刚入门、想找一个完整落地路径的Python开发者。2. 系统骨架PyQt界面、策略线程与数据流怎么拧在一起2.1 线程分离是PyQt量化工具的第一道门槛凡是跑过下载的免费Python源码的人大概率见过这个场面点击“开始选股”界面卡死标题栏变成“未响应”过十几秒缓过来结果一下刷出几十条日志。这不是电脑慢是耗时任务被塞进了GUI线程。PyQt的消息循环一次只能干一件事你让它去遍历3000只股票的K线它就没空重绘按钮和窗口。正确做法是把耗时逻辑扔进QThread再用信号把结果传回界面。信号槽是线程安全的能跨线程调用这是PyQt提供的正规通道。不要在工作线程里直接操作QLabel、QTableWidget这些控件也不要图省事用全局变量加锁去传状态——后期加一个选股任务、加一个回测任务全局变量会让你改到怀疑人生。我一般会给每个耗时任务单独建一个Worker类继承QThread把任务逻辑放在run()里只通过自定义信号对外发消息。界面端拿到信号后再更新控件。这套结构写起来不复杂但能一次性解决掉“界面卡死”、“日志乱序”、“任务重复启动”这三个最常见的问题。2.2 数据源选型tushare、baostock、akshare怎么挑量化系统离了数据就是空壳。选数据源时很多人只看接口数量忽略了积分、频率和复权因子。我这边常用的三选一先看一张对比表。数据源免费额度日线分钟线复权因子稳定性baostock全免费支持有前/后/不复权稳定接口老但够用tushare pro积分制新号积分低支持有有数据全但积分不够时受限akshare全免费支持部分部分接口多经常随网页改版波动如果你做的是A股日线级别的自动选股和策略测试baostock是最省心的登录即用不卡积分复权因子直接给到前复权、后复权、不复权三个选项字段也干净。tushare的数据质量确实好但积分不够时每分钟调用次数卡得很难受不适合做盘后批量扫描几百只股票。akshare胜在接口全但同一个接口可能这周能用、下周就报错用在长期跑批的自动选股模块里会让你不断返工。选型时还要注意数据缓存。不要每次选股都全量拉数据建议把日线数据按股票代码存成本地文件或SQLite每日增量更新。这一方面省流量另一方面也降低对数据源接口频率的限制。把数据落地到本地之后选股模块和回测模块就可以共用同一份数据避免界面和策略拿到两个不一样的数据集。2.3 最小可跑骨架选股线程加状态栏回调光说没用先把一个最小骨架搭起来。这个骨架只有一个按钮和一个状态栏点击按钮后启动一个后台线程线程跑完通过信号把消息传回界面更新状态栏。import sys import time from PyQt5.QtCore import QThread, pyqtSignal from PyQt5.QtWidgets import QApplication, QMainWindow, QPushButton, QLabel, QVBoxLayout, QWidget class SelectWorker(QThread): # 线程向界面回传消息的唯一出口信号 progress pyqtSignal(str) def run(self): # 这里放选股、回测这类耗时逻辑 for i in range(20): time.sleep(0.1) self.progress.emit(f正在遍历第 {i 1} 个候选标的...) self.progress.emit(选股完成) class MainWindow(QMainWindow): def __init__(self): super().__init__() self.worker None self.status_label QLabel(就绪) self.start_btn QPushButton(开始选股) self.start_btn.clicked.connect(self.start_select) layout QVBoxLayout() layout.addWidget(self.start_btn) layout.addWidget(self.status_label) container QWidget() container.setLayout(layout) self.setCentralWidget(container) def start_select(self): # 防止重复点击导致多个线程同时跑 if self.worker and self.worker.isRunning(): return self.worker SelectWorker() self.worker.progress.connect(self.status_label.setText) self.worker.start() if __name__ __main__: app QApplication(sys.argv) window MainWindow() window.show() sys.exit(app.exec_())这段代码里最关键的是第11行的progress pyqtSignal(str)和第34行的self.worker.progress.connect(...)。信号携带一个字符串参数线程里emit出去的消息会自动排队送到主线程由主线程更新控件这就避免了跨线程操作控件导致的崩溃和闪烁。start_select里的isRunning()判断必须加否则快速连点两次按钮会启动两个线程同时扫股票轻则重复推送重则把数据库写串。保存worker引用也很重要如果直接SelectWorker().start()而不存引用这个线程随时可能被垃圾回收掉。跑通这个骨架后后面加自动选股、回测、微信推送模块只需要按同样模式新建Worker类并注册信号界面部分基本不用动。这就是把工具当产品写的起点。3. 自动选股模块把选股因子变成可落地的定时任务3.1 指标计算与因子筛选别把未来函数带进选股有了数据源和线程骨架接下来做自动选股。选股的核心是把“我想买什么”翻译成可计算的条件。以最常见的均线多头排列加量比放大为例逻辑很清楚5日均线在20日均线上方且当日成交量比过去5日的均量放大1.5倍以上。这个逻辑写起来不难难在别算出未来数据。很多人用pandas的rolling窗口时忽略了shift导致当天信号用到了当天的收盘价这在回测里是作弊在实盘里会直接导致选股结果和次日走势完全对不上。正确做法是计算完指标后统一用iloc[-1]取最后一个已完成交易日的值并且所有指标只基于当前时刻之前的数据。下面的代码用baostock拉取日线并计算两个基础因子import baostock as bs import pandas as pd lg bs.login() # 获取单只股票的前复权日线 rs bs.query_history_k_data_plus( sh.600000, date,code,close,volume, start_date2024-01-01, end_date2024-12-31, frequencyd, adjustflag2, # 1后复权 2前复权 3不复权 ) rows [] while rs.error_code 0 and rs.next(): rows.append(rs.get_row_data()) df pd.DataFrame(rows, columnsrs.fields) bs.logout() # 因子计算 df[close] df[close].astype(float) df[volume] df[volume].astype(float) df[ma5] df[close].rolling(5).mean() df[ma20] df[close].rolling(20).mean() df[vol_base] df[volume].rolling(5).mean().shift(1) # 避开当日量 df[vol_ratio] df[volume] / df[vol_base]adjustflag2表示前复权这是处理分红送转的标准姿势。如果不复权一只股票除权后K线会跳空均线系统直接失真。shift(1)是这里最容易踩的坑它把量能基准推到前一日保证当日信号不包含当日成交量的反馈偏差。筛选时只取最后一行的数据做判断因为前面所有行都是历史只有最后一根K线是“当下”。如果拿整段数据里的最大值或末日数据去选股等于提前知道了答案。3.2 定时调度与结果落库让选股每天自己跑选股逻辑写好还不够自动选股的重点在“自动”。常见做法是APScheduler做cron定时每个交易日早上开盘前跑一次把结果写进SQLite。这样你打开界面时能看到的是“昨天的选股结果”而不是现场慢慢扫。from apscheduler.schedulers.background import BackgroundScheduler def run_daily_select(): # 这里调用选股主流程把结果写入sqlite print(开始执行自动选股任务) # select_stocks_and_save() scheduler BackgroundScheduler() # 周一到周五的 9:35 执行一次 scheduler.add_job( run_daily_select, cron, day_of_weekmon-fri, hour9, minute35, misfire_grace_time300, ) scheduler.start()misfire_grace_time300是容易被忽略的参数意思是任务原定执行时间点错过了5分钟之内仍然补执行。电脑休眠、系统卡顿都会导致定时任务错过触发点没有这个参数今天没开盘选股结果就是早上空跑一整天。定时任务不能只依赖cron表达式还要处理交易日历。A股有法定节假日和调休周一到周五不一定是交易日。最可靠的办法是维护一份交易日列表比如用baostock的query_trade_dates接口拉全年交易日在任务回调里先判断当天是不是交易日不是就直接返回。不要指望cron表达式自己能分辨清明节和国庆节。选股结果落库我建议建两张表一张存选股日期和股票代码另一张存当天的因子快照。理由很实在如果你只存结果股票后面想复盘“为什么选了这只没选那只”时没有依据。因子快照能让你回看历史也能在策略调整后做对比。3.3 别惦记通达信自动选股的配置文件自己在Python里跑更透明很多人在搜索“通达信自动选股设置在哪个文件”其实通达信有它自己的公式系统配置文件可以设置公式选股但那个机制有两个先天问题一是它在软件内部运行你没法方便地拿到选股结果去做二次处理二是公式语言调试起来黑匣子一样出错了你连日志都没得看。如果你已经走上Python这条路完全没必要把选股逻辑写回通达信公式里。我自己做过一次迁移把通达信公式里的条件选股翻译成Python pandas逻辑总共不过几十行代码但换来的东西是三个实打实的好处选股结果能落库能自动化定时跑能直接对接下一章的回测模块。通达信的公式系统更适合作盘中人工参考而自动选股、策略测试这类场景用Python生态去处理才是主流做法。还有一个细节自动选股的频率不要太高。日线级别的因子一天跑一次足够了。把频率调到分钟级意味着你要处理分钟数据的复权、停牌、集合竞价复杂度翻好几倍而且大部分散户策略根本不依赖分钟级选股。4. 策略测试回测不是跑通历史就完事4.1 自写循环还是Backtrader两条路各有代价策略测试是量化系统里最容易被高估的一环。很多人把“回测曲线很漂亮”等同于“策略能赚钱”实际上回测框架选错、参数填错结果会失真到完全没参考价值。先决定用什么框架自己写循环还是用现成库。自己写循环的优点是逻辑完全透明每一笔成交都能控制缺点是要处理复权、停牌、涨跌停、成交价格偏离所有细节都得自己兜。Backtrader这类现成框架把撮合规则、滑点、手续费都封装好了但学习曲线陡内部机制隔了一层出问题不好排查。维度自写循环Backtrader逻辑透明程度完全透明中等需读源码手续费/滑点控制需自己实现内置commission与slippage分钟级回测工作量很大支持但数据格式要求严格调试难度出错直接定位报错堆栈较绕上手速度快适合验证单策略慢但功能全我的选择标准是单策略验证用自写循环跑通后要做参数扫描或多策略对比再迁到Backtrader。不要一开始就上Backtrader很多新手会因为它的数据格式要求和订单状态机设计而卡住也不要永远自写循环等你需要计算夏普比率、最大回撤、连续亏损时现成框架里这些都是一个函数的事。4.2 回测参数里的隐形成本手续费、滑点与复权回测的收益不等于实盘收益最大的偏差来源就是成本。以下面的双均线策略为例它能在历史数据上频繁交易但每多交易一次手续费和滑点就多刮一层。import backtrader as bt class GoldenCross(bt.Strategy): params dict(fast5, slow20) def __init__(self): self.ma_fast bt.indicators.SMA(self.data.close, periodself.p.fast) self.ma_slow bt.indicators.SMA(self.data.close, periodself.p.slow) self.crossover bt.indicators.CrossOver(self.ma_fast, self.ma_slow) def next(self): if not self.position: if self.crossover 0: size int(self.broker.getcash() * 0.95 / self.data.close[0]) self.buy(sizesize) elif self.crossover 0: self.close() cerebro bt.Cerebro() cerebro.addstrategy(GoldenCross) # 假设 data 是已经读取并复权的数据源 # cerebro.adddata(data) cerebro.broker.setcash(100000.0) cerebro.broker.setcommission(commission0.0003, mult10.0) # 万三佣金 cerebro.addsizer(bt.sizers.PercentSizer, percents95) result cerebro.run()setcommission里的mult10.0表示A股每手100股佣金按成交金额和手数计算。不要忽略这个参数否则一手股票的手续费会被当成一股来算误差放大十倍。PercentSizer按账户资金的95%开仓是为了接近实际仓位而不是满仓梭哈。真正让回测虚高的是滑点。Backtrader默认滑点为零这等于假设你每次都能以收盘价成交。现实是买的时候要高一点买卖的时候要低一点卖。我一般加上固定滑点cerebro.broker.set_slippage_perc(perc0.001) # 单边千一滑点千一的滑点加万三的佣金一个完整的买卖来回成本将近千分之三。高频策略如果没把这笔账算进去回测收益可能比实盘高出一倍甚至更多。复权数据在回测里同样关键。用前复权数据回测时要注意复权因子是动态变化的新一次除权发生后之前所有历史价格会被重新计算。这意味着你昨天的回测结果可能因为今天出现一次新的分红送转而整体改变。常见做法是固定一个“截至某个日期”的复权因子别让它随最新行情漂移否则你的回测结果每天都在变怎么复盘都对不上。4.3 回测结果怎么验证可信度分段和样本外跑完一次回测曲线好看不代表策略成立。先做分段验证把历史数据切成两段前一段用来调参数后一段用来验证。如果后一段的表现明显差于前一段多半是过拟合。这是最费时间但是最值得做的一步。还可以做一个更硬核的检查把参数随机扰动一下比如快线从5改成4或6慢线从20改成30或40如果收益曲线剧烈塌陷说明策略对参数极度敏感实盘基本没法用。真正靠谱的策略在参数邻域内应当保持类似的盈利逻辑而不是只在某个精确数字上灵光一现。我见过不少人在这一步翻车策略在海量参数扫描里挑出了历史收益最高的组合兴冲冲上实盘结果连续亏损。原因很朴素——那一组参数精确拟合了过去的噪声。策略测试的目标是找到“稳定的行为模式”不是“最大的历史收益”。5. 微信通知与互动把交易状态推到手机再把指令收回来5.1 微信通知的三条通道怎么选选股结果和回测结论做好了接微信通知让结果跑到手机上这一步很多人卡住。个人微信自动化这块早期有人用itchat网页协议但已经大面积失效而且风控严重不建议碰。目前稳定靠谱的通道只有两个方向一是企业微信群机器人webhook二是Server酱这类推送服务。企业微信群机器人的好处是免费、不限制个人微信号、可以在群里刷markdown消息而且收件人不限一个。你把自己的量化机器人和几个信得过的朋友拉一个群每天推送选股结果和交易信号这是最贴近“实时交易微信通知”的落地方式。Server酱的逻辑是推送到你自己的微信服务号配置简单适合只给自己看。通道是否免费推送形式限频是否支持互动企业微信群机器人免费群内markdown每分钟20条不支持Server酱免费版够用服务号消息每日有限额支持表单回执个人微信自动化免费但风险高单聊/群聊易封号支持但不稳定实时交易通知的实际瓶颈不在功能而在频率。策略一天最多触发几次交易每分钟20条完全够用。真正要防的是自己的程序出错疯狂刷消息比如策略在循环里没加节流故障时十几分钟推上百条直接被限频拉黑。5.2 用webhook推成交与风控消息企业微信群机器人推送的代码很简单关键是消息内容要精简。我一般只推三类消息选股结果、策略买卖信号、风控预警。每类用不同的前缀方便在群里扫一眼就定位。import requests def push_wx(title: str, content: str) - None: webhook https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key你的key payload { msgtype: markdown, markdown: { content: f## {title}\n{content} } } try: resp requests.post(webhook, jsonpayload, timeout5) data resp.json() if data.get(errcode) ! 0: print(f微信推送失败: {data}) except Exception as e: print(f微信推送异常: {e}) # 使用示例 push_wx(买卖信号, **买入** 浦发银行 600000\n价格: 10.85\n理由: 均线金叉量比放大)timeout5要加上webhook请求偶尔会慢不设超时会把策略线程卡住。这里要注意推送本身失败不应影响主交易流程所以我用try-except把它包住只打日志不抛出异常。实际运行中推送失败很常见群被解散、webhook被回收、网络抖动。一个实用的兜底方案是把推送消息同时追加到本地日志文件这样微信通道挂了你还能在复盘时看到当时到底有没有触发信号。5.3 从单向通知到双向互动本地小服务收指令微信通知做到位了很多人会发现下一个需求是“互动”能不能在手机上让系统停一下、跑一次回测、看看持仓。企业微信机器人不支持接收消息Server酱虽然支持表单回执但互动能力有限。更通用的方案是让量化系统本地起一个HTTP小服务你通过微信里的链接访问或发送指令服务端接收后控制策略的运行状态。我用FastAPI做了一个极简控制接口策略侧在关键循环里检查一个全局标志位from fastapi import FastAPI from pydantic import BaseModel app FastAPI() control_flag {paused: False, run_backtest: False} class Command(BaseModel): action: str app.post(/cmd) def handle(cmd: Command): if cmd.action pause: control_flag[paused] True return {ok: True, msg: 策略已暂停控制} if cmd.action resume: control_flag[paused] False return {ok: True, msg: 策略已恢复} if cmd.action backtest: control_flag[run_backtest] True return {ok: True, msg: 回测任务已触发} return {ok: False, msg: f未知指令: {cmd.action}}策略线程每次迭代时只检查标志位不直接调用“暂停”这样既避免锁竞争也让控制的粒度可控。这个服务只监听127.0.0.1外部访问怎么进来常见做法是把你本机的服务通过局域网地址暴露给手机或者用内网穿透类工具映射出一个临时地址。这里要注意安全控制接口只建议绑定本机回环地址防局域网内其他人乱发指令。有了这个控制接口后微信通知就真正闭环了——通知触达手机指令从手机收回系统变成一个可以远程操作的交易终端。6. 实盘前的避坑清单5条踩过的坑和一条验证路径界面假死。现象点按钮后窗口变白、标题栏提示未响应。原因耗时逻辑直接跑在PyQt主线程。解决一切耗时任务进QThread通过信号回传界面。这条也是免费Python源码里最常见的问题拿到别人项目先看线程怎么写的能省很多排查时间。回测收益虚高。现象回测年化翻倍实盘连续亏损。原因没设滑点、手续费按单边忽略、复权方式不对。解决setcommission加set_slippage_perc复权因子固定日期回测成本按万三佣金加千一滑点起步。停牌股和涨跌停股买不进。现象信号触发了但界面提示下单失败。原因实盘撮合时停牌股无法成交涨停股买不进跌停股卖不出。解决下单前先检查交易状态和涨停跌幅。回测时也要把这些条件模拟进去否则回测里全成交实盘里成不了。微信推送限频与漏发。现象某天群里突然几十条重复消息之后推送全部失效。原因策略异常导致循环内重复推送触发企业微信限频。解决推送加节流和失败重试异常情况只推一条预警其余写本地日志。历史回测与实盘对不上。现象回测该买没买、该卖没卖实盘成交价格总是更差。原因数据源不一致或策略里存在未来函数。解决选股和回测共用同一份本地数据指标计算统一加shift避免未来数据。最后说一条验证路径先用模拟盘跑两周每笔信号和推送都留日志每天收盘后对比模拟盘成交和回测信号。两周下来如果日志和实盘成交基本吻合再考虑投入真金白银。我自己吃过一次亏——回测跑通直接上实盘结果数据源复权口径不一致策略前三个月完全跑偏。从那以后模拟盘和日志核对成了雷打不动的流程。这条验证路径能帮你把前面五个坑一次性兜住希望帮到你。本文还有配套的精品资源点击获取
