先说结论OpenStock是我近段时间从零搭起来的一套开源股票行情监控服务核心目标只有一个——把盯盘、提醒、复盘这三件事从手动操作变成半自动流程让信息在正确的时间主动来找我而不是我每天多个App来回切。取名带Open一方面代码完全开源另一方面数据尽量走公开接口不依赖任何闭源终端和服务。如果你也面临这几个问题股票自选一大堆但记不住每家公司的关键价位、盘中没空一直盯分时、收盘后复盘还是靠翻聊天记录和截图那这篇“手把手搭建OpenStock”的教程就是写给你的。整个方案不需要买服务器、不需要付费数据源一台能跑Python的旧电脑甚至树莓派就能运行最后会得到一个能定时拉行情、按自定义规则触发告警、把提醒推送到手机的小系统。1. 先拆需求OpenStock到底要解决什么问题动手写代码之前我花了大半天时间在纸上列需求。很多个人项目做不下去不是代码写不出来而是需求没想清楚就开干结果功能越加越多最后变成了一堆没人用的脚本。OpenStock的定位我一开始就定得很死它不是自动交易系统不做荐股不做量化策略研究只做三件事——收数据、判断规则、发通知。1.1 散户口中的三个真实痛点第一是信息分散。我平时看A股会用行情软件看行业研报又要切换到另一个平台偶尔还要去财经网站上翻资金流向信息源太多没有一个统一的地方能把我关心的东西集中起来。第二是提醒不及时。行情软件里设置价格提醒是可以但条件太简单只能填一个目标价。我想实现的是“涨跌幅超过3%同时成交量放大到5日均量两倍”这种组合条件普通App根本做不到。第三是复盘基本靠手动。收盘以后把自选股挨个点开截图、记录当日涨跌、写下备注这些事情特别耗时间而且坚持不了几天就放弃了。这三个痛点的本质是缺少一个完全由自己控制的中间层。OpenStock要做的就是把这个中间层补上行情数据通过公开接口获取规则逻辑用Python自己写告警通知走邮件或者Webhook所有数据都落在本地数据库里。这样既灵活也能保证数据隐私不把自选股和盯盘行为上传给任何第三方平台。1.2 自建监控和现成App相比值不值有人可能会问行情软件自带的预警功能就能满足大部分需求为什么还要自己搭这个问题我在做之前也纠结过。我的判断标准是如果只需要单条件价格提醒用现成App就够了没必要折腾。但我需要组合条件的告警、需要把监控结果沉淀到数据库里做后续分析还需要把每天的复盘记录统一管理这些需求现成产品给不了。而且自建方案的一个隐性好处是你可以完全掌控数据的流向。数据落在自己的SQLite文件里想怎么查询就怎么查询想怎么统计就怎么统计。行情软件里的历史自选数据导出起来很麻烦而OpenStock从一开始就把历史监控记录、告警记录、每日快照都结构化存储了后续做复盘分析、写周报、甚至训练自己的简单策略模型数据基础都在。2. 技术选型与整体架构为什么是这套组合在写第一行代码之前我把技术方案完整过了一遍。OpenStock的整体架构可以拆成五个模块数据采集模块、存储模块、规则判断模块、告警通知模块、可视化展示模块。每个模块都选最简单成熟的技术因为个人项目最大的风险是复杂度失控。2.1 系统模块划分数据采集模块负责从公开数据源拉取A股实时行情和日K线数据存储模块把行情快照、监控规则、告警记录存入本地数据库规则判断模块按用户自定义的条件扫描自选股输出“触发/未触发”的结论告警通知模块负责把触发结果推送到邮箱、微信或者任意Webhook可视化展示模块提供简单的Web页面让我能随时看到最新行情和监控状态。这种模块化划分的好处是每个部分可以独立替换。比如今天用邮件通知明天想换成企业微信机器人只需要改通知模块其他代码不用动。数据源也是一样AKShare不稳定了可以把采集模块切换到其他公开接口上层逻辑不受影响。2.2 技术栈选型理由技术栈我选了Python 3.10、AKShare、FastAPI、SQLite、APScheduler和requests没有用任何重框架。理由很直接Python生态里做财经数据采集最方便的就是AKShare它把东方财富、新浪财经等公开网页数据统一封装成了函数不需要自己处理Cookie和请求头FastAPI写一个几十行的后端API非常顺手自带Swagger文档调试方便SQLite零配置文件一个文件就是整个数据库个人项目完全够用。调度任务我选了APScheduler而不是自己写while循环加sleep。APScheduler的CronTrigger可以直接按交易时段配置执行时间代码可读性高而且自带线程池任务阻塞不会影响其他定时任务。通知部分先用smtplib发邮件再兼容一个Webhook接口后续接任何即时通讯工具都方便。2.3 项目目录设计项目目录我设计成下面的结构这是后期维护中最关键的一步openstock/ ├── config.py # 全局配置自选股、阈值、通知方式 ├── database.py # SQLite连接与建表语句 ├── collector.py # 数据采集实时行情、历史K线 ├── monitor.py # 规则判断涨跌幅、突破、量能异动 ├── notifier.py # 告警通知邮件、Webhook ├── scheduler.py # APScheduler定时启动入口 ├── webapp.py # FastAPI可视化页面 └── data/ └── openstock.db # SQLite数据文件这样的结构让每个文件的职责非常单一。新手最容易犯的错误是把所有功能写进一个几千行的脚本里当时觉得很爽过两周想改个阈值都无从下手。分开写以后每个文件都可以单独测试排查问题也快。3. 手把手搭建从零跑通OpenStock接下来进入实操环节我会按照真实搭建顺序从环境准备、数据采集、规则判断、通知发布到定时调度一步一步走完整个流程。先交代一下环境我用的是一台装了Ubuntu 22.04的迷你主机2核4G内存跑了几个月非常稳。你没有Linux机器也没关系macOS和Windows上操作基本一致。3.1 环境准备与项目初始化第一步先创建虚拟环境避免依赖污染系统Pythonmkdir openstock cd openstock python3 -m venv venv source venv/bin/activate pip install akshare fastapi uvicorn[standard] apscheduler requests pandas安装完成以后推荐第一时间验证AKShare是否正常因为这一步网络环境不同结果差异很大import akshare as ak df ak.stock_zh_a_spot_em() print(df.columns.tolist()) print(df.head())如果能看到全A股实时行情并且打印出字段名列表数据源就通了。这里有个很重要的细节AKShare的接口名和字段名会随着上游网页改版而变化网上教程里的代码很可能过时养成打印columns的习惯能帮你少踩很多坑。3.2 行情数据接入与入库数据接入我分两层设计历史日K线数据在每天收盘后更新一次用于计算5日均量等指标实时快照数据在盘中每5分钟拉取一次用于触发监控规则。先看建表语句CREATE TABLE IF NOT EXISTS daily_kline ( symbol TEXT NOT NULL, trade_date TEXT NOT NULL, open REAL, high REAL, low REAL, close REAL, volume REAL, PRIMARY KEY (symbol, trade_date) ); CREATE TABLE IF NOT EXISTS realtime_snapshot ( symbol TEXT NOT NULL, ts TEXT NOT NULL, name TEXT, price REAL, change_pct REAL, volume REAL, amount REAL, PRIMARY KEY (symbol, ts) ); CREATE TABLE IF NOT EXISTS alert_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, symbol TEXT, rule TEXT, message TEXT, ts TEXT );历史K线更新我用的是AKShare的日线接口以平安银行为例import akshare as ak import pandas as pd from database import get_conn def update_daily_kline(symbol000001, days20): end pd.Timestamp.now().strftime(%Y%m%d) start (pd.Timestamp.now() - pd.Timedelta(daysdays * 1.5)).strftime(%Y%m%d) df ak.stock_zh_a_hist(symbolsymbol, perioddaily, start_datestart, end_dateend, adjustqfq) if df is None or df.empty: return df.columns [date, open, close, high, low, volume, amount, amplitude, change_pct, change, turnover] df[symbol] symbol conn get_conn() df[[symbol, date, open, close, high, low, volume, amount]].to_sql(daily_kline, conn, if_existsappend, indexFalse) conn.close()注意adjust参数我用了qfq也就是前复权这样计算均量时不会因为除权除息导致数据突变。实时行情用ak.stock_zh_a_spot_em()一次就能拿到全市场快照不需要按代码逐个请求效率高得多。这个接口返回的volume单位通常是手我习惯除以100换算成股后面算量比的时候统一口径。3.3 监控规则与触发逻辑规则判断是整个项目最核心的部分。我目前实现了三类规则涨跌幅超阈值、价格突破关键价位、量能异动。涨跌幅规则最简单实时快照里有change_pct字段直接和阈值比较就行。价格突破需要配置自定义的压力位比如某只股票如果连续两分钟价格超过15.50元就触发“突破压力位”告警。量能异动稍微复杂一点需要对比当前成交量和近5日均量当前量是日均量两倍以上才触发。核心逻辑可以抽象成一个函数def check_rules(symbol, price, change_pct, volume): alerts [] cfg config.STOCKS.get(symbol) if cfg is None: return alerts # 规则1涨跌幅超阈值 if abs(change_pct) cfg[change_pct_limit]: alerts.append(f{symbol} 涨跌幅 {change_pct:.2f}% 超过阈值 {cfg[change_pct_limit]}%) # 规则2价格突破压力位 if cfg.get(resistance) and price cfg[resistance]: alerts.append(f{symbol} 价格 {price:.2f} 突破压力位 {cfg[resistance]:.2f}) # 规则3量能放大到日均量2倍 avg_volume_5d get_avg_volume(symbol, 5) if avg_volume_5d and volume avg_volume_5d * 2: alerts.append(f{symbol} 当前量 {volume:.0f} 超过5日均量2倍) return alerts这里有一个经验分享配置规则的时候阈值不要设得太敏感。我刚开始把涨跌幅阈值设为1%结果开盘半小时内几乎每分钟都有告警直接把我的注意力炸碎了。后来改成3%并且加了一个冷却时间——同一只股票同一规则在一小时内只提醒一次告警噪音瞬间下降。3.4 告警通知渠道配置通知模块我留了两个接口邮件和Webhook。邮件是最保险的方案只要配好SMTP就能用。我用的方式很简单通过smtplib发送纯文本邮件import smtplib from email.mime.text import MIMEText from email.header import Header import config def send_email(subject, content): msg MIMEText(content, plain, utf-8) msg[Subject] Header(subject, utf-8) msg[From] config.SMTP_USER msg[To] config.ALERT_MAIL_TO with smtplib.SMTP_SSL(config.SMTP_HOST, config.SMTP_PORT) as server: server.login(config.SMTP_USER, config.SMTP_PASS) server.sendmail(config.SMTP_USER, [config.ALERT_MAIL_TO], msg.as_string())Webhook接口更通用只需要一个HTTP POST请求import requests import config def send_webhook(title, content): requests.post(config.WEBHOOK_URL, json{ title: title, description: content, msgtype: text }, timeout10)我用的是通用Webhook格式无论是接企业微信机器人、钉钉机器人还是飞书机器人只要改一下webhook地址和JSON字段名就行。日常使用中我推荐优先用Webhook渠道因为手机通知的实时性比邮件好很多邮件更适合做每日汇总记录不适合做盘中紧急提醒。3.5 定时任务与常驻运行调度任务我用APScheduler实现重点要处理好交易时段。A股交易时间是周一到周五的9:30到11:30、13:00到15:00节假日不开盘。我采用两段式配置盘中监控每5分钟执行一次收盘后在15:10执行数据归档和每日复盘推送。from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger scheduler BlockingScheduler() # 盘中监控工作日9:30-15:00之间每5分钟执行一次 scheduler.add_job( monitor_job, CronTrigger(day_of_weekmon-fri, hour9-14, minute*/5, timezoneAsia/Shanghai) ) # 每日收盘归档15:10执行 scheduler.add_job( daily_archive_job, CronTrigger(day_of_weekmon-fri, hour15, minute10, timezoneAsia/Shanghai) ) scheduler.start()为什么minute写成*/5而不是5,10,15...55因为这样更简洁。但要注意9:30到9:35之间的数据受开盘集合竞价影响波动剧烈我在monitor_job里加了一个判断在前5分钟自动跳过触发条件只记录快照不做告警。为了让服务在后台常驻我把整个项目注册成了一个systemd服务。这是Linux下最稳妥的方式开机自启、崩溃自动重启、日志统一管理都解决了。配置参考如下[Unit] DescriptionOpenStock Monitor Afternetwork-online.target [Service] User你的用户名 WorkingDirectory/home/你的用户名/openstock ExecStart/home/你的用户名/openstock/venv/bin/python /home/你的用户名/openstock/scheduler.py Restartalways RestartSec30 EnvironmentTZAsia/Shanghai [Install] WantedBymulti-user.target4. 常见问题与排查技巧实录把这套系统跑起来的难度其实不大真正折磨人的是运行过程中的各种小毛病。我总结了四个最常遇到的问题每一个都是我踩过的坑对应的排查思路也都是验证过的。4.1 数据接口异常与字段变更AKShare底层是从公开网页抓数据上游页面一改版接口就会出现各种奇怪问题。最常见的是字段名变了导致KeyError或者返回DataFrame为空。排查方法只有一条不要依赖网上的字段名每次调用后先打印columns。我写数据采集模块时每跑一个接口都会输出一次列名确认没问题才进入下一步。另外AKShare偶尔会因请求频率过高触发限流表现是请求卡住几十秒甚至抛出超时异常。解决方式是在循环请求之间加time.sleep(1)并且做好异常捕获单次失败不影响整体任务。4.2 交易时段与集合竞价干扰刚上线第一周我发现每天9:30到9:31之间总会触发一大批虚假告警后来查日志发现是集合竞价数据的问题。9:15到9:25是集合竞价时段9:25产生开盘价9:30才连续竞价所以9:30到9:35之间的快照数据噪声很大。我的解决办法是两层过滤第一层在调度代码里9:35之前只记录快照不触发规则第二层在规则判断代码里对涨跌幅超过阈值的情况再额外校验一次要求连续两次快照都满足条件才告警。这个“二次确认”机制效果很好几乎消除了瞬时数据抖动造成的告警。4.3 通知发送失败与告警轰炸通知模块出过两个问题都很有意思。第一个是邮箱被判定为垃圾邮件用户体验极差。原因是邮件内容太短、无正文、标题带“alert”这类词容易被邮箱服务商过滤。解决方法是把邮件标题和正文都规范化去掉敏感词增加落款和链接命中率恢复正常。第二个问题是告警轰炸。有段时间我设置了单只股票涨跌幅超过2%就提醒结果行情波动大的时候同一只股票五分钟触发一次一天下来收到几十条重复信息。解决方式是引入了一个冷却机制在数据库里记录每条告警的触发时间同一股票同一规则在冷却窗口内重复触发时直接丢弃不通知。冷却窗口我默认设置为1小时这个值可以按自己的盯盘习惯调整。4.4 SQLite并发写入与数据膨胀SQLite个人使用很简单但也有一个需要注意的点。磁盘慢或者写入频繁时可能出现“database is locked”错误。我把连接配置加了WAL模式和超时时间问题基本解决conn.execute(PRAGMA journal_modeWAL) conn.execute(PRAGMA busy_timeout5000)数据膨胀问题出现在运行一段时间后realtime_snapshot表积累了海量历史快照磁盘占用越来越大查询变慢。我的做法是写一个归档任务保留最近30天的快照数据更早的只做每日汇总指标删掉明细。这样数据库体积控制在几百兆以内查询速度一直很稳定。5. 从一个监控工具扩展到你的个人研究平台OpenStock跑通以后我很快发现它不只是一个告警工具而是一个天然的行情数据平台。因为所有快照、日线、告警都结构化存下来了后续加功能特别顺畅。这里分享两个我觉得价值最大的扩展方向。5.1 加一个简易回测既然有历史K线数据做规则回测也就顺理成章。我加了一个简单的回测脚本把涨跌幅规则和量能异动规则代入历史数据统计触发后的5日收益分布。这个回测不追求精确就为了验证规则在历史上是否有效避免凭感觉拍脑袋。实现起来不复杂核心就是遍历历史数据找到规则触发点然后计算从触发点到未来N天的收益。需要注意这种回测有明显的幸存者偏差和过拟合风险毕竟样本量有限、规则参数是自己调出来的。我把它定位成辅助决策参考而不是自动交易信号。用一段时间的统计结果来判断“这个规则是噪音还是有规律”顺带筛选出值得重点关注的股票池。5.2 收盘自动生成复盘日报另一个立竿见影的功能是每日复盘日报。每天收盘后自动生成一份Markdown格式日报内容包括自选股当日涨跌排序、哪些股票触发了告警、当日大盘概况、我的盘中备注自动汇总。这份日报通过Webhook推送到手机或者通过邮件发送给自己晚上通勤时花五分钟就能完成当日复盘。日报的模式很简单从数据库里读取当天的realtime_snapshot、alert_log和手动输入的note表用模板拼成Markdown再通过通知模块发出去。坚持跑了一个月之后我对自选股状态的把握明显比之前清晰很多因为所有的复盘动作都有了历史沉淀不再需要依赖记忆和零散的截图。我自己把这套OpenStock跑了几个月最大的体会是一开始不要追求大而全先把“数据能拉下来、规则能触发、通知能发出去”这个最小闭环跑通再用真实行情慢慢打磨规则。工具本身不会替你赚钱它的价值是把散落的信息按时整理好送到你面前让你在需要做判断的时候手里有足够的数据支撑而不是凭感觉。后面我计划继续完善回测模块和盘中异动排行榜让这套系统真正成为我自己的行情研究基础设施。