简介这是一套面向量化交易初学者与Python开发者的股票分析系统开源实现解决个人投资者缺乏专业工具进行策略验证、自动选股与实盘联动的痛点。系统基于Python 3.4构建采用PyQt5实现跨平台图形界面集成多线程事件引擎与MongoDB本地数据库支持历史数据自动下载含个股、指数、ETF日线及分笔数据、策略回测、模拟交易9个独立账号、微信实时通知及一键全自动实盘交易且实盘与回测共用同一策略逻辑显著降低策略迁移成本。压缩包共317个文件主体为294个Python源码文件涵盖引擎、数据模块、策略模板、UI逻辑等辅以11张界面与流程图PNG、4份核心说明文档如策略编写、历史数据下载指南、2个Java验证码工具jar包及配置类JSON文件整体仅1.81MB轻量易部署。已有89人学习下载提供完整可运行框架、标准化策略接口、实盘/模拟双模式切换能力及WindTuShare双数据源适配方案是理解量化系统架构与快速启动实盘实践的高价值参考项目。 做股票量化分析这件事最头疼的往往不是策略本身而是“怎么把这个东西落地成一套能天天用的工具”。命令行脚本跑完看结果太原始。天天盯盘手动操作太累。我这次就把整套流程做成一个带 PyQt5 图形界面的桌面系统把自动选股、交易策略回测、实时通知全串起来。最终效果是每天开盘前自动跑一遍选股逻辑符合条件直接弹通知历史回测结果用图表直观展示止损止盈、均线交叉这类经典策略都能一键测试。这套系统最适合三类人一是刚入门量化、想把 Python 技术真正接到股票数据上的朋友二是手里有固定交易逻辑、但懒得天天手动筛股的散户三是想学习 PyQt5 桌面应用开发、需要一个完整实战案例的开发者。内容不涉及任何“稳赚”的玄学重点讲工程实现讲“怎么把想法变成一个可用的软件”。1. 项目整体框架与模块划分1.1 核心需求拆解动手之前先把需求拆成四个模块数据层、策略层、界面层、通知层。这是整个项目最核心的一步。我见过太多人一上来就写界面结果数据接口没定好后面全部返工。先说数据层。量化系统的地基就是数据没有干净的数据策略写得再漂亮也是垃圾进垃圾出。数据层要解决三个问题行情数据从哪来、怎么存、怎么保证实时性。策略层是系统的灵魂。自动选股解决的是“哪些票值得关注”策略回测解决的是“我的买卖逻辑到底靠不靠谱”。这两个功能看似独立实际共享同一套数据接口所以我把它们放在同一层方便互相调用。界面层负责把结果呈现在用户面前。PyQt5 在这个项目里承担两个任务一是用表格、图表把选股结果和回测曲线直观展示出来二是提供交互入口比如调整参数、手动触发选股、查看个股详情。通知层解决的是“人不盯着程序时怎么办”这个问题。选股结果跑出来后通过系统托盘弹窗、声音提醒、邮件推送等方式主动通知用户这是桌面量化工具最能体现价值的地方。1.2 技术选型为什么是 Python PyQt5 akshare选型这件事我的原则是“够用、好维护、生态成熟”。Python 做量化分析是绝对的主流pandas 和 numpy 处理行情数据非常顺手策略实现速度快不用像 C 那样纠结内存管理。GUI 框架我对比过 PyQt5、Tkinter 和 wxPython。Tkinter 太简陋做表格和图表交互很吃力wxPython 文档少、社区冷清遇到问题很难搜到答案。PyQt5 的优势在于QTableWidget 和 QTableView 做行情列表非常成熟配合 pyqtgraph 或 matplotlib 画 K 线和回测曲线有现成方案而且信号槽机制天然适合处理“按钮点击 → 触发选股 → 刷新表格”这种交互流程。更实际的一点是PyQt5 的资料和案例在国内社区很多踩坑成本低。数据源我推荐 akshare。它免费、无需 token、接口覆盖 A 股全部行情数据和财务数据对一个个人项目来说足够用了。tushare 更稳定但新版接口需要积分个人开发者起步阶段没必要。有些朋友可能会问为什么不用现成的量化平台比如聚宽、米筐它们确实强大但有两个问题一是策略逻辑被平台锁死想自定义界面和通知机制很麻烦二是数据出不了平台没法做个性化的本地分析。自己用 Python 搭一套自由度完全不一样。2. 数据层行情获取与本地缓存2.1 数据源接口封装akshare 的接口我用得最多的是这几个ak.stock_zh_a_spot_em()获取全部 A 股实时行情返回所有股票的代码、名称、最新价、涨跌幅、成交量等字段ak.stock_zh_a_hist()获取个股历史 K 线数据支持日线、周线、月线和前复权ak.stock_individual_info_em()查询个股基本面数据。这里有一个非常重要的点akshare 的字段名是中文的比如“最新价”“涨跌幅”而 pandas 处理时用中文列名在后续代码里容易出编码问题。我习惯在获取数据后立刻做一次字段重命名统一改成英文。下面的代码就是数据层的基础封装import akshare as ak import pandas as pd class DataFetcher: def __init__(self): # 统一字段映射避免中文列名带来的编码问题 self.spot_columns { 代码: code, 名称: name, 最新价: price, 涨跌幅: pct_chg, 成交量: volume, 成交额: amount, 换手率: turnover, 市盈率: pe, 总市值: mkt_cap } def get_all_stocks(self) - pd.DataFrame: 获取全市场实时行情 df ak.stock_zh_a_spot_em() df df.rename(columnsself.spot_columns) # 只保留需要的字段减少内存占用 cols [code, name, price, pct_chg, volume, amount, turnover, pe, mkt_cap] return df[cols].copy() def get_history(self, code: str, days: int 250) - pd.DataFrame: 获取个股历史日线数据默认250个交易日 df ak.stock_zh_a_hist(symbolcode, perioddaily, adjustqfq, start_date20200101) df df.rename(columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume }) return df[[date, open, close, high, low, volume]].tail(days)这段代码里我特意做了两件事重命名字段、裁剪列。不要小看这一步全市场 5000 多只股票每多保留一列无关数据选股时的内存占用和遍历耗时都会上升。2.2 数据缓存与增量更新策略每次启动程序都重新拉全市场数据会有两个问题一是 akshare 接口有频率限制短时间内频繁请求会被限流二是全量下载耗时十几秒用户体验很差。我的方案是本地缓存 定时刷新。缓存用 CSV 就够了文件按日期命名程序启动时先读当天缓存如果不存在或太久没更新再触发网络请求。增量更新的逻辑是每天只有收盘后需要全量拉取盘中跑选股只需要获取实时行情两者分开处理。import os import time class DataCache: def __init__(self, cache_dircache): self.cache_dir cache_dir os.makedirs(cache_dir, exist_okTrue) def get_spot_data(self, fetcher: DataFetcher, max_age: int 300) - pd.DataFrame: 获取实时行情5分钟内的缓存直接复用 today time.strftime(%Y%m%d) cache_file os.path.join(self.cache_dir, fspot_{today}.csv) if os.path.exists(cache_file): mtime os.path.getmtime(cache_file) if time.time() - mtime max_age: return pd.read_csv(cache_file) df fetcher.get_all_stocks() df.to_csv(cache_file, indexFalse) return df注意这里 max_age 参数我设置的是 300 秒。A 股盘中每分钟都在变化缓存太久选股结果会失真但 5 分钟内的数据对中短线策略来说足够新鲜还能避免反复请求被限流。3. 自动选股模块多因子筛选与实现3.1 选股逻辑设计从单因子到多因子打分自动选股的核心不是“预测涨跌”而是“用可量化的规则缩小关注范围”。全市场五千多只股票人不可能一个个看但程序可以。选股模块的设计思路是先用硬性条件过滤掉明显不行的票再用多因子打分排序。硬性过滤条件我用的是剔除 ST 股票、剔除上市不足 60 天的次新股、剔除价格低于 2 元的股票、剔除当日停牌的股票。这些条件都是基于常识的风险控制避免选出一堆根本没法参与的标的。打分因子我实现了几类技术面因子均线多头排列、MACD 金叉、RSI 在合理区间、资金面因子换手率适中、成交额排名靠前、基本面因子市盈率、市净率、市值。每个因子按规则打分最终加权求和排序。这里有一个关键认知多因子模型的目的不是找到“必涨”的票而是找到“值得研究”的票。我的组合逻辑很朴素——均线多头排列说明趋势向上MACD 金叉说明动能转强成交额放大说明有资金关注三者同时满足的票起码值得放进自选股盯一盯。3.2 选股策略核心代码下面是选股模块的核心实现我把策略拆成独立的函数方便日后加新因子import pandas as pd def filter_stocks(df: pd.DataFrame) - pd.DataFrame: 硬性条件过滤 df df[~df[name].str.contains(ST|退, naFalse)] df df[df[price] 2] df df[df[turnover] 1] df df[df[turnover] 20] return df.copy() def score_stocks(df: pd.DataFrame, history_repo: dict) - pd.DataFrame: 多因子打分 scores [] for _, row in df.iterrows(): score 0 code row[code] # 技术面均线多头排列 hist history_repo.get(code) if hist is not None and len(hist) 20: ma5 hist[close].rolling(5).mean().iloc[-1] ma10 hist[close].rolling(10).mean().iloc[-1] ma20 hist[close].rolling(20).mean().iloc[-1] if ma5 ma10 ma20: score 40 # 资金面成交额超过 2 亿 if row[amount] 2e8: score 20 # 基本面市盈率在 0~80 之间剔除亏损和过度高估 if 0 row[pe] 80: score 20 # 涨跌幅温和上涨优于暴涨暴跌 if -3 row[pct_chg] 5: score 10 scores.append(score) df[score] scores return df.sort_values(score, ascendingFalse)评分逻辑里我特别说明一点涨跌幅因子我给的是“温和上涨”的票加分而不是“涨得越多分越高”。原因很简单暴涨票追进去很容易高位接盘量化选股要的是可复现的规则不是追涨杀跌的冲动。代码里如果取不到历史数据就直接跳过技术面评分避免因数据缺失误判。4. 交易策略回测让历史数据说话4.1 回测框架的核心设计选股解决的是“买什么”回测解决的是“怎么买、怎么卖、能不能赚钱”。我一直跟朋友说回测是量化系统里最不能省的一环——你凭什么相信一个策略凭历史数据算出来的收益曲线、最大回撤、胜率。没有回测支撑的策略就是赌。回测框架我拆成四个部分数据准备、信号生成、模拟撮合、指标统计。信号生成是最核心的部分。我以均线策略为例当 5 日均线上穿 20 日均线金叉时产生买入信号5 日均线下穿 20 日均线死叉时产生卖出信号。这是最经典的策略但实现细节里坑很多。比如金叉后如果第二天下跌要不要立即执行历史回测里是按当日收盘价成交还是次日开盘价成交这些细节直接决定回测结果的真实程度。模拟撮合部分我采用一个简化但合理的方案信号产生次日开盘价成交同时扣除双边手续费和滑点。手续费按万 2.5 计算滑点按 0.1% 计算。这两项成本看着不多但对高频策略来说可能就是盈利和亏损的分界线。4.2 均线策略回测完整实现def backtest_ma(hist: pd.DataFrame, short5, long20, fee0.00025, slippage0.001) - dict: 均线策略回测 :param hist: 历史行情含 open/close/high/low/volume :param short: 短期均线周期 :param long: 长期均线周期 :param fee: 双边手续费率 :param slippage: 滑点比例 :return: 回测指标字典 df hist.copy() df[ma_short] df[close].rolling(short).mean() df[ma_long] df[close].rolling(long).mean() # 金叉死叉信号1 表示持仓0 表示空仓 df[signal] 0 df.loc[df[ma_short] df[ma_long], signal] 1 df[signal_diff] df[signal].diff() df[position] 0 cash 100000 # 初始资金 10 万 position 0 # 持仓股数 record [] # 逐笔交易记录 for i in range(1, len(df)): # 次日开盘价成交 if df[signal_diff].iloc[i] 1 and position 0: buy_price df[open].iloc[i] * (1 slippage) position int(cash / (buy_price * (1 fee))) cash - position * buy_price * (1 fee) record.append((B, df[date].iloc[i], buy_price)) elif df[signal_diff].iloc[i] -1 and position 0: sell_price df[open].iloc[i] * (1 - slippage) cash position * sell_price * (1 - fee) record.append((S, df[date].iloc[i], sell_price)) position 0 # 回测指标统计 final_value cash position * df[close].iloc[-1] total_return (final_value - 100000) / 100000 * 100 # 简化计算最大回撤 closes df[close].values peak closes[0] max_drawdown 0 for c in closes: peak max(peak, c) dd (peak - c) / peak max_drawdown max(max_drawdown, dd) return { total_return: round(total_return, 2), max_drawdown: round(max_drawdown * 100, 2), trade_count: len(record) // 2, records: record }这段代码里的撮合细节值得多说几句。我特意用df[signal_diff]而不是直接判断均线位置是因为金叉只有当“前一天没有金叉、今天才金叉”的时候才算有效信号否则每天都会重复触发买入。这就是量化实现和理论描述的差距一个 diff 就能避免重复开仓的 bug。回测只用单一股票跑一遍是看不出问题的真实的回测模块应该支持批量回测。我的做法是把回测函数套在一个循环里对选股结果中的每只股票跑一遍最后汇总成一张收益对比表。你一眼就能看出这个策略在哪些票上亏钱、在哪些票上赚钱然后决定要不要调整参数。5. PyQt5 图形界面从命令行到可视化5.1 GUI 整体布局设计命令行工具做得再强没有可视化界面使用门槛就摆在那里。PyQt5 界面我分三个区域顶部是工具栏选股触发、回测触发、参数设置按钮左侧是股票列表表格显示自动选股的结果右侧是详情区域个股 K 线图、回测收益曲线。布局用 QSplitter 还是 QTabWidget我最终选择 QTabWidget原因很简单左侧选股结果、回测报告、策略日志三个页面互不干扰切换方便。如果你想要更灵活的窗口布局QSplitter 可以拖拽调节比例但代码复杂度会高一截。个人项目优先保证功能稳定布局朴素一点没关系。底部再加一个 QStatusBar 显示状态信息比如“数据加载完成”“选股进行中”“回测已完成”。状态栏是很多人忽略的细节但它能让用户明确感知程序在干什么大大减少“程序是不是卡死了”的疑虑。5.2 多线程处理防止界面卡死的关键这是整篇文章里最值得划线的一段。PyQt5 的界面运行在主线程数据请求、选股计算这些耗时操作如果也放在主线程界面会像死机一样卡住按钮点了没反应窗口拖不动。Windows 甚至会弹出“程序未响应”的提示。解决办法是用 QThread 把耗时任务放到子线程通过信号把结果传回主线程更新界面。我封装了一个通用的任务线程类from PyQt5.QtCore import QThread, pyqtSignal class TaskThread(QThread): finished pyqtSignal(object) # 正常结果 error pyqtSignal(str) # 错误信息 def __init__(self, task_func, *args): super().__init__() self.task_func task_func self.args args def run(self): try: result self.task_func(*self.args) self.finished.emit(result) except Exception as e: self.error.emit(str(e))用这个类启动选股的按钮事件就非常简洁def on_btn_select_clicked(self): self.statusbar.showMessage(选股进行中请稍候...) self.btn_select.setEnabled(False) # 防止重复点击 self.thread TaskThread(run_select_stocks) self.thread.finished.connect(self.on_select_done) self.thread.error.connect(self.on_task_error) self.thread.start()两个细节大家一定要注意。第一线程对象必须保存为 self 的成员变量否则局部变量会被垃圾回收程序直接崩溃。第二任务执行期间要禁用触发按钮防止用户重复点击开了多个线程既浪费资源又容易产生数据竞争。5.3 图表绘制与实时刷新图表我用 pyqtgraph 而不是 matplotlib。matplotlib 画静态图没问题但做实时刷新时性能明显不足来回切换交互模式也麻烦。pyqtgraph 是专门为 PyQt 设计的绘图库刷新速度快交互流畅。画 K 线图需要用到 pyqtgraph 的 candlestick 功能底层是基于 GraphicsObject 的实现起来比普通折线图复杂一些。提供一个最小实现思路用pg.GraphicsObject子类在 paint 方法里遍历每一根 K 线用 drawLine 画上下影线用 drawRect 画实体。只要数据量在几百根以内性能完全没问题。回测收益曲线的绘制就简单多了直接调用plot()就能画出净值曲线。我的做法是把策略回测的每日资产净值保存为列表传到绘图函数里即可。记住一点图表控件的刷新要么在信号回调里直接 update要么用一个 QTimer 定时刷新不要用 time.sleep 来“假装刷新”那会让整个界面卡住。6. 实时通知与用户互动从被动查看到主动触达6.1 桌面通知系统托盘与声音提醒自动选股跑完结果怎么送到用户面前我的方案是三层递进系统托盘弹窗、声音提醒、邮件推送。系统托盘弹窗用 QSystemTrayIcon这是桌面应用最自然的消息提醒方式。设置好图标后调用showMessage()方法就能在屏幕右下角弹出通知。要注意 Windows 下如果程序没有托盘图标showMessage 是不生效的所以必须先把 QSystemTrayIcon 创建并 show 出来。from PyQt5.QtWidgets import QSystemTrayIcon, QApplication from PyQt5.QtGui import QIcon class Notifier: def __init__(self): self.tray QSystemTrayIcon() self.tray.setIcon(QIcon(icon.png)) self.tray.setVisible(True) def notify(self, title: str, message: str): 弹系统通知并播放声音 self.tray.showMessage(title, message, QSystemTrayIcon.Information, 10000) QApplication.beep()声音提醒用 QApplication.beep()简单有效。更进阶的做法是用 QSound 播放自定义音频适合用不同声音区分提醒类型比如选股完成一个音效止损触发另一个音效。6.2 邮件推送与用户交互设计系统托盘通知有一个天然缺陷程序必须开着人必须坐在电脑前。如果你想让选股结果在收盘后自动发到手机上邮件推送是最简单可靠的方案。用 Python 的 smtplib 调 QQ 邮箱或 163 邮箱的 SMTP 服务几十行代码就能实现。import smtplib from email.mime.text import MIMEText def send_email(receiver: str, subject: str, content: str): sender 你的邮箱qq.com auth_code 你的授权码 msg MIMEText(content, plain, utf-8) msg[Subject] subject msg[From] sender msg[To] receiver server smtplib.SMTP_SSL(smtp.qq.com, 465) server.login(sender, auth_code) server.sendmail(sender, [receiver], msg.as_string()) server.quit()注意这里用的不是邮箱登录密码而是邮箱服务商提供的“授权码”。QQ 邮箱需要在设置里开启 SMTP 服务并生成授权码这是很多人第一次写邮件推送脚本时卡住的地方。用户交互方面我做了三个功能选股参数面板用户可以调整均线周期、市盈率范围等、白名单管理用户可以把某些股票加入白名单即使不满足条件也保留在列表中、手动刷新按钮。这些看似简单的交互组件决定了这个系统是“玩具”还是“工具”。没有交互的软件永远只是脚本有交互的软件才叫产品。7. 常见问题与排查实录7.1 典型问题速查表我把开发过程中遇到的典型问题整理成一个表格方便大家对照排查。这些都是实际踩过的坑网上教程很少会提到现象根本原因解决方案界面加载行情时卡死并提示未响应网络请求放在了主线程用 QThread 把耗时操作移到子线程akshare 偶尔返回空数据接口限流或网络波动增加重试机制最多重试3次失败后用缓存选股结果和行情软件对不上数据未复权或字段理解错误明确使用前复权数据核对字段含义打包 exe 后程序很大PyQt5 依赖的库都被打进去了用 PyInstaller 排除不需要的模块中文字体显示为方块运行环境缺少中文字体设置 QFont 指定微软雅黑或思源黑体QThread 线程运行一次后无法再次启动QThread 对象被垃圾回收把线程保存为实例变量每次任务重新创建线程7.2 打包 exe 的踩坑经历程序写完后要面对的现实问题是别人想装这个工具但总不能让他们每人配一个 Python 环境。把 PyQt5 程序打包成 exe 是必经之路。我用的是 PyInstaller命令很简单pyinstaller -F -w -i icon.ico main.py-F 表示打包成单文件-w 表示不显示控制台窗口-i 指定程序图标。但实际踩坑不少。打包出来的 exe 可能比预期大很多因为 PyQt5 会把一堆组件打进去用不上的模块可以通过 spec 文件排除。更大的坑是数据文件路径问题。如果程序里引用了图标、样式表等外部资源打包后相对路径会失效。因为 PyInstaller 解压运行时当前工作目录和 exe 所在目录不一样。我最终统一用sys._MEIPASS来获取资源路径import sys import os def resource_path(relative_path): 获取打包后资源文件的绝对路径 if hasattr(sys, _MEIPASS): base sys._MEIPASS else: base os.path.abspath(.) return os.path.join(base, relative_path)7.3 多进程在量化场景里的使用边界很多人搜“python多进程”我也聊聊多进程在量化系统里的合理用法。前面说的 QThread 是线程解决的是界面不卡死的问题。但如果你的选股策略要对 5000 只股票逐一计算技术指标单线程跑可能要几十秒这时多进程就派上用场了。Python 有 GIL 锁多线程对 CPU 密集任务没用必须用 multiprocessing 或 concurrent.futures 的 ProcessPoolExecutor。我的做法是把选股计算拆成多个 chunk每个进程处理一个股票子集最后合并结果。实测下来4 个进程大概能把选股时间从 40 秒压缩到 15 秒左右提升明显。from concurrent.futures import ProcessPoolExecutor def parallel_score(stock_list, num_workers4): chunk_size len(stock_list) // num_workers 1 chunks [stock_list[i:ichunk_size] for i in range(0, len(stock_list), chunk_size)] with ProcessPoolExecutor(max_workersnum_workers) as executor: results list(executor.map(score_one_chunk, chunks)) return pd.concat(results)不过要提醒一点多进程不是越多越好。进程间通信和内存拷贝是有开销的数据量小时反而更慢。我实测过2000 只股票以内用单进程更快超过 4000 只上多进程才有明显收益。盲目堆进程数是新手常见的性能优化误区。最后聊点实在的。这个系统从零搭到能稳定跑前后花了大概两个周末最耗时的不是写代码而是调数据和排界面问题。我的建议是从最小可用版本起步先跑通数据获取再做选股然后加回测最后才上 GUI。每做一个模块就验证一个模块别想着一次性全部做完那样出了问题你根本不知道是哪个环节的锅。另外一个心得是量化工具只是辅助决策它帮你把全市场五千只股票缩小到十几只值得关注的标的省下的是大量翻资料的时间。真正的买卖决策还是要结合自己对市场的理解。别把程序输出的任何信号当成“圣旨”它只是给了你一个再次确认的机会。如果你也想动手做不必一上来就追求功能大而全建议先把自动选股 表格展示这条链路打通那种“程序帮我从五千只票里挑出十几只”的感觉是推动你继续往下做的最好动力。本文还有配套的精品资源点击获取
用PyQt5打造股票量化分析桌面系统：自动选股与策略回测实战
简介这是一套面向量化交易初学者与Python开发者的股票分析系统开源实现解决个人投资者缺乏专业工具进行策略验证、自动选股与实盘联动的痛点。系统基于Python 3.4构建采用PyQt5实现跨平台图形界面集成多线程事件引擎与MongoDB本地数据库支持历史数据自动下载含个股、指数、ETF日线及分笔数据、策略回测、模拟交易9个独立账号、微信实时通知及一键全自动实盘交易且实盘与回测共用同一策略逻辑显著降低策略迁移成本。压缩包共317个文件主体为294个Python源码文件涵盖引擎、数据模块、策略模板、UI逻辑等辅以11张界面与流程图PNG、4份核心说明文档如策略编写、历史数据下载指南、2个Java验证码工具jar包及配置类JSON文件整体仅1.81MB轻量易部署。已有89人学习下载提供完整可运行框架、标准化策略接口、实盘/模拟双模式切换能力及WindTuShare双数据源适配方案是理解量化系统架构与快速启动实盘实践的高价值参考项目。 做股票量化分析这件事最头疼的往往不是策略本身而是“怎么把这个东西落地成一套能天天用的工具”。命令行脚本跑完看结果太原始。天天盯盘手动操作太累。我这次就把整套流程做成一个带 PyQt5 图形界面的桌面系统把自动选股、交易策略回测、实时通知全串起来。最终效果是每天开盘前自动跑一遍选股逻辑符合条件直接弹通知历史回测结果用图表直观展示止损止盈、均线交叉这类经典策略都能一键测试。这套系统最适合三类人一是刚入门量化、想把 Python 技术真正接到股票数据上的朋友二是手里有固定交易逻辑、但懒得天天手动筛股的散户三是想学习 PyQt5 桌面应用开发、需要一个完整实战案例的开发者。内容不涉及任何“稳赚”的玄学重点讲工程实现讲“怎么把想法变成一个可用的软件”。1. 项目整体框架与模块划分1.1 核心需求拆解动手之前先把需求拆成四个模块数据层、策略层、界面层、通知层。这是整个项目最核心的一步。我见过太多人一上来就写界面结果数据接口没定好后面全部返工。先说数据层。量化系统的地基就是数据没有干净的数据策略写得再漂亮也是垃圾进垃圾出。数据层要解决三个问题行情数据从哪来、怎么存、怎么保证实时性。策略层是系统的灵魂。自动选股解决的是“哪些票值得关注”策略回测解决的是“我的买卖逻辑到底靠不靠谱”。这两个功能看似独立实际共享同一套数据接口所以我把它们放在同一层方便互相调用。界面层负责把结果呈现在用户面前。PyQt5 在这个项目里承担两个任务一是用表格、图表把选股结果和回测曲线直观展示出来二是提供交互入口比如调整参数、手动触发选股、查看个股详情。通知层解决的是“人不盯着程序时怎么办”这个问题。选股结果跑出来后通过系统托盘弹窗、声音提醒、邮件推送等方式主动通知用户这是桌面量化工具最能体现价值的地方。1.2 技术选型为什么是 Python PyQt5 akshare选型这件事我的原则是“够用、好维护、生态成熟”。Python 做量化分析是绝对的主流pandas 和 numpy 处理行情数据非常顺手策略实现速度快不用像 C 那样纠结内存管理。GUI 框架我对比过 PyQt5、Tkinter 和 wxPython。Tkinter 太简陋做表格和图表交互很吃力wxPython 文档少、社区冷清遇到问题很难搜到答案。PyQt5 的优势在于QTableWidget 和 QTableView 做行情列表非常成熟配合 pyqtgraph 或 matplotlib 画 K 线和回测曲线有现成方案而且信号槽机制天然适合处理“按钮点击 → 触发选股 → 刷新表格”这种交互流程。更实际的一点是PyQt5 的资料和案例在国内社区很多踩坑成本低。数据源我推荐 akshare。它免费、无需 token、接口覆盖 A 股全部行情数据和财务数据对一个个人项目来说足够用了。tushare 更稳定但新版接口需要积分个人开发者起步阶段没必要。有些朋友可能会问为什么不用现成的量化平台比如聚宽、米筐它们确实强大但有两个问题一是策略逻辑被平台锁死想自定义界面和通知机制很麻烦二是数据出不了平台没法做个性化的本地分析。自己用 Python 搭一套自由度完全不一样。2. 数据层行情获取与本地缓存2.1 数据源接口封装akshare 的接口我用得最多的是这几个ak.stock_zh_a_spot_em()获取全部 A 股实时行情返回所有股票的代码、名称、最新价、涨跌幅、成交量等字段ak.stock_zh_a_hist()获取个股历史 K 线数据支持日线、周线、月线和前复权ak.stock_individual_info_em()查询个股基本面数据。这里有一个非常重要的点akshare 的字段名是中文的比如“最新价”“涨跌幅”而 pandas 处理时用中文列名在后续代码里容易出编码问题。我习惯在获取数据后立刻做一次字段重命名统一改成英文。下面的代码就是数据层的基础封装import akshare as ak import pandas as pd class DataFetcher: def __init__(self): # 统一字段映射避免中文列名带来的编码问题 self.spot_columns { 代码: code, 名称: name, 最新价: price, 涨跌幅: pct_chg, 成交量: volume, 成交额: amount, 换手率: turnover, 市盈率: pe, 总市值: mkt_cap } def get_all_stocks(self) - pd.DataFrame: 获取全市场实时行情 df ak.stock_zh_a_spot_em() df df.rename(columnsself.spot_columns) # 只保留需要的字段减少内存占用 cols [code, name, price, pct_chg, volume, amount, turnover, pe, mkt_cap] return df[cols].copy() def get_history(self, code: str, days: int 250) - pd.DataFrame: 获取个股历史日线数据默认250个交易日 df ak.stock_zh_a_hist(symbolcode, perioddaily, adjustqfq, start_date20200101) df df.rename(columns{ 日期: date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume }) return df[[date, open, close, high, low, volume]].tail(days)这段代码里我特意做了两件事重命名字段、裁剪列。不要小看这一步全市场 5000 多只股票每多保留一列无关数据选股时的内存占用和遍历耗时都会上升。2.2 数据缓存与增量更新策略每次启动程序都重新拉全市场数据会有两个问题一是 akshare 接口有频率限制短时间内频繁请求会被限流二是全量下载耗时十几秒用户体验很差。我的方案是本地缓存 定时刷新。缓存用 CSV 就够了文件按日期命名程序启动时先读当天缓存如果不存在或太久没更新再触发网络请求。增量更新的逻辑是每天只有收盘后需要全量拉取盘中跑选股只需要获取实时行情两者分开处理。import os import time class DataCache: def __init__(self, cache_dircache): self.cache_dir cache_dir os.makedirs(cache_dir, exist_okTrue) def get_spot_data(self, fetcher: DataFetcher, max_age: int 300) - pd.DataFrame: 获取实时行情5分钟内的缓存直接复用 today time.strftime(%Y%m%d) cache_file os.path.join(self.cache_dir, fspot_{today}.csv) if os.path.exists(cache_file): mtime os.path.getmtime(cache_file) if time.time() - mtime max_age: return pd.read_csv(cache_file) df fetcher.get_all_stocks() df.to_csv(cache_file, indexFalse) return df注意这里 max_age 参数我设置的是 300 秒。A 股盘中每分钟都在变化缓存太久选股结果会失真但 5 分钟内的数据对中短线策略来说足够新鲜还能避免反复请求被限流。3. 自动选股模块多因子筛选与实现3.1 选股逻辑设计从单因子到多因子打分自动选股的核心不是“预测涨跌”而是“用可量化的规则缩小关注范围”。全市场五千多只股票人不可能一个个看但程序可以。选股模块的设计思路是先用硬性条件过滤掉明显不行的票再用多因子打分排序。硬性过滤条件我用的是剔除 ST 股票、剔除上市不足 60 天的次新股、剔除价格低于 2 元的股票、剔除当日停牌的股票。这些条件都是基于常识的风险控制避免选出一堆根本没法参与的标的。打分因子我实现了几类技术面因子均线多头排列、MACD 金叉、RSI 在合理区间、资金面因子换手率适中、成交额排名靠前、基本面因子市盈率、市净率、市值。每个因子按规则打分最终加权求和排序。这里有一个关键认知多因子模型的目的不是找到“必涨”的票而是找到“值得研究”的票。我的组合逻辑很朴素——均线多头排列说明趋势向上MACD 金叉说明动能转强成交额放大说明有资金关注三者同时满足的票起码值得放进自选股盯一盯。3.2 选股策略核心代码下面是选股模块的核心实现我把策略拆成独立的函数方便日后加新因子import pandas as pd def filter_stocks(df: pd.DataFrame) - pd.DataFrame: 硬性条件过滤 df df[~df[name].str.contains(ST|退, naFalse)] df df[df[price] 2] df df[df[turnover] 1] df df[df[turnover] 20] return df.copy() def score_stocks(df: pd.DataFrame, history_repo: dict) - pd.DataFrame: 多因子打分 scores [] for _, row in df.iterrows(): score 0 code row[code] # 技术面均线多头排列 hist history_repo.get(code) if hist is not None and len(hist) 20: ma5 hist[close].rolling(5).mean().iloc[-1] ma10 hist[close].rolling(10).mean().iloc[-1] ma20 hist[close].rolling(20).mean().iloc[-1] if ma5 ma10 ma20: score 40 # 资金面成交额超过 2 亿 if row[amount] 2e8: score 20 # 基本面市盈率在 0~80 之间剔除亏损和过度高估 if 0 row[pe] 80: score 20 # 涨跌幅温和上涨优于暴涨暴跌 if -3 row[pct_chg] 5: score 10 scores.append(score) df[score] scores return df.sort_values(score, ascendingFalse)评分逻辑里我特别说明一点涨跌幅因子我给的是“温和上涨”的票加分而不是“涨得越多分越高”。原因很简单暴涨票追进去很容易高位接盘量化选股要的是可复现的规则不是追涨杀跌的冲动。代码里如果取不到历史数据就直接跳过技术面评分避免因数据缺失误判。4. 交易策略回测让历史数据说话4.1 回测框架的核心设计选股解决的是“买什么”回测解决的是“怎么买、怎么卖、能不能赚钱”。我一直跟朋友说回测是量化系统里最不能省的一环——你凭什么相信一个策略凭历史数据算出来的收益曲线、最大回撤、胜率。没有回测支撑的策略就是赌。回测框架我拆成四个部分数据准备、信号生成、模拟撮合、指标统计。信号生成是最核心的部分。我以均线策略为例当 5 日均线上穿 20 日均线金叉时产生买入信号5 日均线下穿 20 日均线死叉时产生卖出信号。这是最经典的策略但实现细节里坑很多。比如金叉后如果第二天下跌要不要立即执行历史回测里是按当日收盘价成交还是次日开盘价成交这些细节直接决定回测结果的真实程度。模拟撮合部分我采用一个简化但合理的方案信号产生次日开盘价成交同时扣除双边手续费和滑点。手续费按万 2.5 计算滑点按 0.1% 计算。这两项成本看着不多但对高频策略来说可能就是盈利和亏损的分界线。4.2 均线策略回测完整实现def backtest_ma(hist: pd.DataFrame, short5, long20, fee0.00025, slippage0.001) - dict: 均线策略回测 :param hist: 历史行情含 open/close/high/low/volume :param short: 短期均线周期 :param long: 长期均线周期 :param fee: 双边手续费率 :param slippage: 滑点比例 :return: 回测指标字典 df hist.copy() df[ma_short] df[close].rolling(short).mean() df[ma_long] df[close].rolling(long).mean() # 金叉死叉信号1 表示持仓0 表示空仓 df[signal] 0 df.loc[df[ma_short] df[ma_long], signal] 1 df[signal_diff] df[signal].diff() df[position] 0 cash 100000 # 初始资金 10 万 position 0 # 持仓股数 record [] # 逐笔交易记录 for i in range(1, len(df)): # 次日开盘价成交 if df[signal_diff].iloc[i] 1 and position 0: buy_price df[open].iloc[i] * (1 slippage) position int(cash / (buy_price * (1 fee))) cash - position * buy_price * (1 fee) record.append((B, df[date].iloc[i], buy_price)) elif df[signal_diff].iloc[i] -1 and position 0: sell_price df[open].iloc[i] * (1 - slippage) cash position * sell_price * (1 - fee) record.append((S, df[date].iloc[i], sell_price)) position 0 # 回测指标统计 final_value cash position * df[close].iloc[-1] total_return (final_value - 100000) / 100000 * 100 # 简化计算最大回撤 closes df[close].values peak closes[0] max_drawdown 0 for c in closes: peak max(peak, c) dd (peak - c) / peak max_drawdown max(max_drawdown, dd) return { total_return: round(total_return, 2), max_drawdown: round(max_drawdown * 100, 2), trade_count: len(record) // 2, records: record }这段代码里的撮合细节值得多说几句。我特意用df[signal_diff]而不是直接判断均线位置是因为金叉只有当“前一天没有金叉、今天才金叉”的时候才算有效信号否则每天都会重复触发买入。这就是量化实现和理论描述的差距一个 diff 就能避免重复开仓的 bug。回测只用单一股票跑一遍是看不出问题的真实的回测模块应该支持批量回测。我的做法是把回测函数套在一个循环里对选股结果中的每只股票跑一遍最后汇总成一张收益对比表。你一眼就能看出这个策略在哪些票上亏钱、在哪些票上赚钱然后决定要不要调整参数。5. PyQt5 图形界面从命令行到可视化5.1 GUI 整体布局设计命令行工具做得再强没有可视化界面使用门槛就摆在那里。PyQt5 界面我分三个区域顶部是工具栏选股触发、回测触发、参数设置按钮左侧是股票列表表格显示自动选股的结果右侧是详情区域个股 K 线图、回测收益曲线。布局用 QSplitter 还是 QTabWidget我最终选择 QTabWidget原因很简单左侧选股结果、回测报告、策略日志三个页面互不干扰切换方便。如果你想要更灵活的窗口布局QSplitter 可以拖拽调节比例但代码复杂度会高一截。个人项目优先保证功能稳定布局朴素一点没关系。底部再加一个 QStatusBar 显示状态信息比如“数据加载完成”“选股进行中”“回测已完成”。状态栏是很多人忽略的细节但它能让用户明确感知程序在干什么大大减少“程序是不是卡死了”的疑虑。5.2 多线程处理防止界面卡死的关键这是整篇文章里最值得划线的一段。PyQt5 的界面运行在主线程数据请求、选股计算这些耗时操作如果也放在主线程界面会像死机一样卡住按钮点了没反应窗口拖不动。Windows 甚至会弹出“程序未响应”的提示。解决办法是用 QThread 把耗时任务放到子线程通过信号把结果传回主线程更新界面。我封装了一个通用的任务线程类from PyQt5.QtCore import QThread, pyqtSignal class TaskThread(QThread): finished pyqtSignal(object) # 正常结果 error pyqtSignal(str) # 错误信息 def __init__(self, task_func, *args): super().__init__() self.task_func task_func self.args args def run(self): try: result self.task_func(*self.args) self.finished.emit(result) except Exception as e: self.error.emit(str(e))用这个类启动选股的按钮事件就非常简洁def on_btn_select_clicked(self): self.statusbar.showMessage(选股进行中请稍候...) self.btn_select.setEnabled(False) # 防止重复点击 self.thread TaskThread(run_select_stocks) self.thread.finished.connect(self.on_select_done) self.thread.error.connect(self.on_task_error) self.thread.start()两个细节大家一定要注意。第一线程对象必须保存为 self 的成员变量否则局部变量会被垃圾回收程序直接崩溃。第二任务执行期间要禁用触发按钮防止用户重复点击开了多个线程既浪费资源又容易产生数据竞争。5.3 图表绘制与实时刷新图表我用 pyqtgraph 而不是 matplotlib。matplotlib 画静态图没问题但做实时刷新时性能明显不足来回切换交互模式也麻烦。pyqtgraph 是专门为 PyQt 设计的绘图库刷新速度快交互流畅。画 K 线图需要用到 pyqtgraph 的 candlestick 功能底层是基于 GraphicsObject 的实现起来比普通折线图复杂一些。提供一个最小实现思路用pg.GraphicsObject子类在 paint 方法里遍历每一根 K 线用 drawLine 画上下影线用 drawRect 画实体。只要数据量在几百根以内性能完全没问题。回测收益曲线的绘制就简单多了直接调用plot()就能画出净值曲线。我的做法是把策略回测的每日资产净值保存为列表传到绘图函数里即可。记住一点图表控件的刷新要么在信号回调里直接 update要么用一个 QTimer 定时刷新不要用 time.sleep 来“假装刷新”那会让整个界面卡住。6. 实时通知与用户互动从被动查看到主动触达6.1 桌面通知系统托盘与声音提醒自动选股跑完结果怎么送到用户面前我的方案是三层递进系统托盘弹窗、声音提醒、邮件推送。系统托盘弹窗用 QSystemTrayIcon这是桌面应用最自然的消息提醒方式。设置好图标后调用showMessage()方法就能在屏幕右下角弹出通知。要注意 Windows 下如果程序没有托盘图标showMessage 是不生效的所以必须先把 QSystemTrayIcon 创建并 show 出来。from PyQt5.QtWidgets import QSystemTrayIcon, QApplication from PyQt5.QtGui import QIcon class Notifier: def __init__(self): self.tray QSystemTrayIcon() self.tray.setIcon(QIcon(icon.png)) self.tray.setVisible(True) def notify(self, title: str, message: str): 弹系统通知并播放声音 self.tray.showMessage(title, message, QSystemTrayIcon.Information, 10000) QApplication.beep()声音提醒用 QApplication.beep()简单有效。更进阶的做法是用 QSound 播放自定义音频适合用不同声音区分提醒类型比如选股完成一个音效止损触发另一个音效。6.2 邮件推送与用户交互设计系统托盘通知有一个天然缺陷程序必须开着人必须坐在电脑前。如果你想让选股结果在收盘后自动发到手机上邮件推送是最简单可靠的方案。用 Python 的 smtplib 调 QQ 邮箱或 163 邮箱的 SMTP 服务几十行代码就能实现。import smtplib from email.mime.text import MIMEText def send_email(receiver: str, subject: str, content: str): sender 你的邮箱qq.com auth_code 你的授权码 msg MIMEText(content, plain, utf-8) msg[Subject] subject msg[From] sender msg[To] receiver server smtplib.SMTP_SSL(smtp.qq.com, 465) server.login(sender, auth_code) server.sendmail(sender, [receiver], msg.as_string()) server.quit()注意这里用的不是邮箱登录密码而是邮箱服务商提供的“授权码”。QQ 邮箱需要在设置里开启 SMTP 服务并生成授权码这是很多人第一次写邮件推送脚本时卡住的地方。用户交互方面我做了三个功能选股参数面板用户可以调整均线周期、市盈率范围等、白名单管理用户可以把某些股票加入白名单即使不满足条件也保留在列表中、手动刷新按钮。这些看似简单的交互组件决定了这个系统是“玩具”还是“工具”。没有交互的软件永远只是脚本有交互的软件才叫产品。7. 常见问题与排查实录7.1 典型问题速查表我把开发过程中遇到的典型问题整理成一个表格方便大家对照排查。这些都是实际踩过的坑网上教程很少会提到现象根本原因解决方案界面加载行情时卡死并提示未响应网络请求放在了主线程用 QThread 把耗时操作移到子线程akshare 偶尔返回空数据接口限流或网络波动增加重试机制最多重试3次失败后用缓存选股结果和行情软件对不上数据未复权或字段理解错误明确使用前复权数据核对字段含义打包 exe 后程序很大PyQt5 依赖的库都被打进去了用 PyInstaller 排除不需要的模块中文字体显示为方块运行环境缺少中文字体设置 QFont 指定微软雅黑或思源黑体QThread 线程运行一次后无法再次启动QThread 对象被垃圾回收把线程保存为实例变量每次任务重新创建线程7.2 打包 exe 的踩坑经历程序写完后要面对的现实问题是别人想装这个工具但总不能让他们每人配一个 Python 环境。把 PyQt5 程序打包成 exe 是必经之路。我用的是 PyInstaller命令很简单pyinstaller -F -w -i icon.ico main.py-F 表示打包成单文件-w 表示不显示控制台窗口-i 指定程序图标。但实际踩坑不少。打包出来的 exe 可能比预期大很多因为 PyQt5 会把一堆组件打进去用不上的模块可以通过 spec 文件排除。更大的坑是数据文件路径问题。如果程序里引用了图标、样式表等外部资源打包后相对路径会失效。因为 PyInstaller 解压运行时当前工作目录和 exe 所在目录不一样。我最终统一用sys._MEIPASS来获取资源路径import sys import os def resource_path(relative_path): 获取打包后资源文件的绝对路径 if hasattr(sys, _MEIPASS): base sys._MEIPASS else: base os.path.abspath(.) return os.path.join(base, relative_path)7.3 多进程在量化场景里的使用边界很多人搜“python多进程”我也聊聊多进程在量化系统里的合理用法。前面说的 QThread 是线程解决的是界面不卡死的问题。但如果你的选股策略要对 5000 只股票逐一计算技术指标单线程跑可能要几十秒这时多进程就派上用场了。Python 有 GIL 锁多线程对 CPU 密集任务没用必须用 multiprocessing 或 concurrent.futures 的 ProcessPoolExecutor。我的做法是把选股计算拆成多个 chunk每个进程处理一个股票子集最后合并结果。实测下来4 个进程大概能把选股时间从 40 秒压缩到 15 秒左右提升明显。from concurrent.futures import ProcessPoolExecutor def parallel_score(stock_list, num_workers4): chunk_size len(stock_list) // num_workers 1 chunks [stock_list[i:ichunk_size] for i in range(0, len(stock_list), chunk_size)] with ProcessPoolExecutor(max_workersnum_workers) as executor: results list(executor.map(score_one_chunk, chunks)) return pd.concat(results)不过要提醒一点多进程不是越多越好。进程间通信和内存拷贝是有开销的数据量小时反而更慢。我实测过2000 只股票以内用单进程更快超过 4000 只上多进程才有明显收益。盲目堆进程数是新手常见的性能优化误区。最后聊点实在的。这个系统从零搭到能稳定跑前后花了大概两个周末最耗时的不是写代码而是调数据和排界面问题。我的建议是从最小可用版本起步先跑通数据获取再做选股然后加回测最后才上 GUI。每做一个模块就验证一个模块别想着一次性全部做完那样出了问题你根本不知道是哪个环节的锅。另外一个心得是量化工具只是辅助决策它帮你把全市场五千只股票缩小到十几只值得关注的标的省下的是大量翻资料的时间。真正的买卖决策还是要结合自己对市场的理解。别把程序输出的任何信号当成“圣旨”它只是给了你一个再次确认的机会。如果你也想动手做不必一上来就追求功能大而全建议先把自动选股 表格展示这条链路打通那种“程序帮我从五千只票里挑出十几只”的感觉是推动你继续往下做的最好动力。本文还有配套的精品资源点击获取