OpenStock—— 如果只看名字你可能会觉得它又是一个聚合行情 App 的套壳。但把时间拉回到我做这个项目的上半年真正让我下决心动手的是我受够了手里那七八个行情软件之间的信息割裂自选股在一个 App 里历史数据在另一个网站消息面又被算法推荐搅成一锅粥。刚好那段时间我一直在折腾自己的技术栈就想着干脆用周末时间搭一个完全属于自己的开源行情监控系统顺便把数据、提醒、策略信号全部收拢到一个面板里。OpenStock 就是那时候起步的。这个项目本质上是一个轻量级的个人行情终端用 Python 抓取公开行情数据落库到本地 SQLite再用 Web 页面展示自选股、K 线和技术指标同时支持价格提醒和简单的策略信号。它不追求高频交易的毫秒级速度也不指望替代专业付费终端但如果你只想拥有一套数据自己说了算、界面自己改、逻辑自己写的行情体系它非常值得折腾一遍。适合有 Python 基础、想拿真实金融数据练手、或者对现有行情工具不满意的开发者。今天这篇文章我就从设计思路开始把整个搭建过程从头到尾拆开讲清楚包括一些容易踩的坑。1. 为什么我要自己搭一套行情系统1.1 现成工具用着不顺手的几个痛点市面上的行情工具其实已经非常强大了但当你真正高频使用一段时间后总会碰到几个团队和产品定位决定它们无法解决的问题。第一是数据所有权的问题。你在 App 里看自选股、存 K 线数据本质上都在人家的服务器上。今天这个接口收费、明天那个功能改版或者某天平台策略调整你的自选分组、预警条件、历史数据可能说没就没了。自己做系统数据存在自己的数据库里想怎么导、怎么算、怎么备份都行这种掌控感是第三方工具给不了的。第二是策略验证的流程太绕。很多终端工具提供了指标公式和选股器但公式语言是封闭的调试起来不顺手更没法把信号直接推送到自己的微信、钉钉或者邮件里。我习惯用 Python 写点简单的均线、动量策略如果每次验证都要手动从行情软件里导出数据效率太低了。OpenStock 把所有逻辑收在同一个项目里从数据到信号从信号到通知是一条完整的链路。第三是界面和交互的个性化。我不想每天打开五个页面来回切我想把自选股、大盘指数、持仓盈亏、当日提醒放在一屏里最好还可以自己写配色、加模块、调整布局。开源项目的价值就在这里代码是公开的看得见摸得着改起来没有边界。1.2 OpenStock 的整体设计思路这个项目在设计上我给自己定了三个原则够用、可改、好维护。够用是不追求庞大架构。我不需要分布式爬虫、也不需要消息队列就是一个常驻进程定时抓数据一个 Web 服务展示数据加一个 SQLite 做存储。整个系统最核心的部件就是三块采集层、存储层、展示层。采集层负责对接行情数据源把日线数据、实时行情、指数数据抓下来存储层把抓到的数据清洗后写入 SQLite按代码、日期建立索引展示层用 Web 方式把数据库里的内容渲染成页面支持自选股分组、K 线图表和指标叠加。提醒和策略信号放在采集层和展示层之间作为独立的调度模块运行。为什么用 SQLite 而不是 MySQL 或者 PostgreSQL个人项目的数据量其实不大一只股票一天的日线记录也就几十字节几千只股票积累十年也就几百万行SQLite 完全可以轻松支撑。而且 SQLite 是单文件数据库备份只需要拷贝一个文件部署时也不需要单独维护数据库服务对小项目来说性价比极高。我把这个思路叫按数据规模选型很多人一上来就上重型组件实际上是在给运维添负担。1.3 技术选型背后的取舍技术栈上我选了 Python FastAPI SQLite ECharts 的组合。Python 不用多说做数据抓取和策略回测的生态太成熟了Pandas、Requests、APScheduler 这些库都是现成的写起来效率高。FastAPI 作为 Web 框架自带 OpenAPI 文档写接口非常顺手而且性能对于个人项目来说绰绰有余。前端没有用重型框架直接使用服务端模板加 ECharts 画图这样整个项目的前端依赖最少部署时也省心。数据接口我优先选公开的、不需要付费的 HTTP 接口比如一些公开行情源提供的日线、周线、分钟线数据。它们响应速度虽然不算快但胜在免费稳定对于日线级别的个人监控需求完全足够。需要注意的一点是公开接口没有 SLA 承诺所以代码里必须有完善的重试、缓存和异常兜底不能因为接口抖动就把整个服务搞挂。这套选型在初期帮了我大忙因为每个环节都是相对通用、资料丰富的技术遇到问题很容易搜索到答案。等到项目跑顺以后再逐步加功能也不会因为底层选的太偏导致寸步难行。2. 环境准备先把地基打牢2.1 这台机器怎么选OpenStock 对硬件的要求很低普通的家用电脑、一台小型 ARM 开发板或者一台 1 核 2G 的云服务器都能跑。我自己最初是跑在一台树莓派 4B 上后来换了云服务器主要原因是要 7x24 小时跑提醒服务家用的网络和断电情况不太可控。如果你需要全天候运行我的建议是准备一台常开的 Linux 机器系统用 Ubuntu 22.04 LTS 或者 Debian 12 就行。如果只是白天盯盘时用Mac、Windows 都无所谓Python 是跨平台的语言代码不用改。内存方面只跑 OpenStock 的话 1G 内存就够但如果还要跑策略回测建议 2G 以上。磁盘按一年数据量估算日线数据日志大约占 200M 左右所以 20G 的磁盘剩余空间就非常宽裕了。2.2 Python 环境与项目初始化我建议用虚拟环境来隔离项目依赖。项目依赖的第三方库不算多核心就是这几个requests负责从数据源抓取 HTTP 数据pandas处理表格型行情数据fastapi和uvicorn提供 Web API 和静态页面服务apscheduler做定时任务调度jinja2渲染模板页面ecosystem可选数据可视化时前端会用到 ECharts这个通过 CDN 引入即可不占 Python 依赖初始化项目的命令很简单mkdir openstock cd openstock python3 -m venv venv source venv/bin/activate pip install requests pandas fastapi uvicorn apscheduler jinja2项目结构我建议按模块拆分从一开始就别把所有代码塞在一个文件里openstock/ ├── app.py # Web 服务入口 ├── collector.py # 数据采集调度 ├── models.py # 数据模型与数据库操作 ├── indicators.py # 技术指标计算 ├── alert.py # 价格提醒和信号推送 ├── config.py # 全局配置 ├── templates/ │ └── index.html # 前端页面模板 ├── static/ # 静态资源 └── data/ └── stock.db # SQLite 数据库这样的模块划分思路简单说就是各管一段采集和展示不互相干扰指标计算单独成模块提醒逻辑好单独测试。后续加功能时也从这些模块边界上扩展不会出现改一行代码把另一个功能弄挂的尴尬。2.3 数据库选型与初始化我用 SQLitePython 自带sqlite3不需要额外安装。首次启动项目时程序会自动创建数据库和表结构。核心的几张表先设计好比如股票列表、日线行情、自选分组、提醒规则。初始化表的 SQL 可以这么写CREATE TABLE IF NOT EXISTS stocks ( code TEXT PRIMARY KEY, name TEXT, exchange TEXT, updated_at DATETIME ); CREATE TABLE IF NOT EXISTS daily_bars ( code TEXT, date TEXT, open REAL, high REAL, low REAL, close REAL, volume INTEGER, PRIMARY KEY (code, date) ); CREATE TABLE IF NOT EXISTS watchlists ( group_name TEXT, code TEXT, PRIMARY KEY (group_name, code) ); CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, code TEXT, condition_type TEXT, threshold REAL, enabled INTEGER DEFAULT 1, last_triggered_at DATETIME );为什么daily_bars表要把(code, date)作为联合主键因为同一只股票同一个交易日只会有一条日线数据。设置联合主键可以在数据库层面防止重复插入不需要每次入库前都先查询一遍是否存在。这个细节虽然小但在跑历史数据回填时能省下大量时间。初始化逻辑我写在models.py里每次启动时调用一次初始化函数即可SQLite 的CREATE TABLE IF NOT EXISTS会保证幂等重复执行也不会报错。3. 数据层让行情数据流进系统3.1 数据源怎么选数据源是整个系统的根基也是最容易出问题的部分。市面上的免费行情源不少但稳定性参差不齐。我在选型时主要看三个维度数据质量、访问限制、字段完整性。数据质量方面最好选择返回 JSON 格式的公开数据接口字段要包含开高低收、成交量这些基础 OHLCV 信息。访问限制要特别注意频率限制有些接口限制每分钟最多请求多少次超过了会把 IP 封一段时间。字段完整性指的是除了日线是否还提供周线、月线或者分时数据这样后续扩展功能时不用换数据源。我的做法是在config.py里把数据源配置成可替换的结构。初始阶段先用一个稳定的公开接口跑通全流程后面根据需求更换数据源时只影响采集模块不影响存储和展示。这个设计用了面向接口的思维虽然是个人项目但预留扩展点带来的好处非常明显。3.2 数据抓取逻辑与入库采集模块的核心是把定时抓取和数据入库拆开。定时调度用 APScheduler每天收盘后自动启动一次抓取任务也可以手动触发补数。抓取日线数据的核心逻辑如下import requests import pandas as pd from datetime import datetime DATA_SOURCE_URL https://example-finance-api.com/historical def fetch_daily_bars(code, start_date, end_date): params { code: code, start: start_date, end: end_date, adjust: qfq # 前复权处理分红送股 } resp requests.get(DATA_SOURCE_URL, paramsparams, timeout10) resp.raise_for_status() data resp.json() df pd.DataFrame(data[bars]) df[code] code df[date] pd.to_datetime(df[date]).dt.strftime(%Y-%m-%d) return df[[code, date, open, high, low, close, volume]]这里有个细节抓取历史 K 线时复权处理非常关键。如果一只股票在历史上经历过分红送股不复权的价格会在除权日出现跳空缺口导致技术指标计算失真。我选用前复权数据这样看历史走势时当前价格与历史价格的相对关系是连续的均线、MACD 这类指标算出来才有意义。入库时利用前面设置的联合主键配合INSERT OR REPLACE语法可以非常简洁地做增量更新def save_daily_bars(df): conn sqlite3.connect(data/stock.db) df.to_sql(daily_bars, conn, if_existsappend, indexFalse) conn.close()to_sql配合if_existsappend会把待插入的每一行与主键比对重复的自动替换不重复的正常插入。这样每天收盘后跑一次数据表就是最新的不用自己写复杂的去重逻辑。3.3 增量更新与历史补数首次搭建系统时需要把历史数据一次性拉下来这个过程叫历史补数。我之前跑全市场五千多只股票的五年日线花了大概三个多小时中间还因为接口限流断了几次。所以补数逻辑一定要支持断点续传从一个开始日期启动每只股票抓完之后记录完成状态失败了就记录失败原因下次启动时只处理未完成的部分。我用的方式是维护一个简单的任务表CREATE TABLE IF NOT EXISTS sync_status ( code TEXT PRIMARY KEY, last_sync_date TEXT, status TEXT );每只股票抓完更新last_sync_date下次补数时只处理last_sync_date 目标日期的股票。同时用time.sleep(0.5)在两次请求之间加一点延迟避免频率过高被限流。等到增量更新阶段每天的调度任务就不用全量跑一遍了只需要对自选股和持仓股票做日更即可压力小很多。3.4 数据源的限流与重试策略公开数据接口最让人头疼的就是限流和超时。我总结了一套实用策略三次重试、指数退避、随机抖动、失败降级。三次重试的意思是接口请求失败后分别等待 1 秒、2 秒、4 秒后重试如果三次都失败就放弃这次请求并记录日志。指数退避就是等待时间按指数增长避免在接口还没恢复时反复请求加重负担。随机抖动是在等待时间上加上一个随机偏移量防止多个任务同时重试导致请求集中。失败降级是指如果实时数据源挂了可以暂时用上一次成功抓取的数据顶上保证页面不会空白。这个策略看起来简单实际运行时非常管用。最初的版本我没有做重试每天总有几只股票因为网络波动抓不到数据数据库里的数据缺一块后面算出来的指数和提醒就不准确。加了这套机制之后抓取成功率稳定在 99.9% 以上。4. 核心功能实现从数据到展示4.1 自选股分组与行情面板自选股分组是我用得最多的功能。我按照自己的习惯分成了几组核心持仓、观察仓、指数 ETF、商品期货。每个分组对应数据库里watchlists表的一条记录页面加载时通过分组名过滤出对应股票列表再实时拉取这些股票的最新行情。核心的行接口是 FastAPI 提供的from fastapi import FastAPI from fastapi.responses import HTMLResponse import sqlite3 app FastAPI() app.get(/api/group/{group_name}) def get_group(group_name: str): conn sqlite3.connect(data/stock.db) cur conn.cursor() cur.execute( SELECT s.code, s.name, b.close, b.change_pct FROM watchlists w JOIN stocks s ON w.code s.code JOIN ( SELECT code, close, (close - open) / open * 100 AS change_pct FROM daily_bars WHERE date (SELECT MAX(date) FROM daily_bars) ) b ON s.code b.code WHERE w.group_name ? , (group_name,)) rows cur.fetchall() conn.close() return {group: group_name, stocks: rows}这个 SQL 里嵌入了子查询先取出每个股票代码的最新交易日价格再与分组表关联。关键点在于用MAX(date)来定位最新交易日这样不会因为停牌或者数据缺失导致取到旧数据。API 返回值直接给前端模板用模板里渲染成表格。涨跌幅我用红涨绿跌的配色逻辑但这种个人习惯属于可修改项你完全可以根据自己的审美调整。4.2 K 线图与指标叠加前端展示层最核心的组件就是 K 线图。我用了 ECharts 的 candlestick 系列搭配 MA5、MA10、MA20 三条均线。数据从后端接口获取接口直接返回用于绘图的结构化数据。app.get(/api/kline/{code}) def get_kline(code: str, limit: int 250): conn sqlite3.connect(data/stock.db) df pd.read_sql_query( SELECT date, open, close, low, high, volume FROM daily_bars WHERE code ? ORDER BY date DESC LIMIT ? , conn, params(code, limit)) df df.sort_values(date) conn.close() return { dates: df[date].tolist(), kline: df[[open, close, low, high]].values.tolist(), volumes: df[volume].tolist() }返回给前端的数据格式要与 ECharts 的要求对应五个数组的列表是开盘、收盘、最低、最高这是 ECharts 的固定顺序。一开始我搞反过把最低最高放反了K 线的影线全是反的盯了好半天才发现是字段顺序问题。前端模板里这样初始化图表const chart echarts.init(document.getElementById(kline)); fetch(/api/kline/${code}) .then(res res.json()) .then(data { chart.setOption({ tooltip: { trigger: axis }, xAxis: { data: data.dates }, yAxis: { scale: true }, series: [{ type: candlestick, data: data.kline }] }); });这部分的核心逻辑不复杂但为了交互体验我额外做了两个细节图表区域自适应窗口大小、鼠标滚轮缩放时间范围。自适应是在窗口resize事件里调用chart.resize()缩放是 ECharts 自带的dataZoom组件只要配置一下即可。这样看长周期走势时可以自由拖拽查看不同区间的形态。4.3 技术指标计算与应用指标计算放在indicators.py模块里主要实现均线、MACD、RSI 三个常用指标。均线最简单就是对收盘价做滚动平均MACD 涉及 EMA 的计算RSI 则是基于涨跌幅均值的震荡指标。以 MACD 为例代码并不长def ema(series, period): return series.ewm(spanperiod, adjustFalse).mean() def macd(close, fast12, slow26, signal9): dif ema(close, fast) - ema(close, slow) dea ema(dif, signal) hist (dif - dea) * 2 return dif, dea, histEWM 是 Pandas 自带的指数加权移动平均函数adjustFalse表示从序列第一个值开始计算。计算出的dif与dea的差值再乘以 2就是常见的柱状图指标。这里的乘以 2 是因为国内股票软件通常把 MACD 柱放大一倍为了让指标观感与主流软件一致。这种细节属于各家习惯不同的范畴如果你觉得不乘以 2 更合适也可以改关键是代码要给你可配置的空间。指标算出来后我直接把结果加入 K 线数据接口的返回结构里前端多画一组折线即可。这样页面展示的就是完整的K线 均线 MACD组合几乎复刻了主流行情软件的核心形态。4.4 提醒与策略信号推送提醒功能是我做得最满意的部分。它有两种触发逻辑价格条件触发和技术指标交叉触发。价格条件好理解比如股价跌破 20 日均线时提醒我。技术指标交叉触发则更复杂一些比如MACD 的 DIF 上穿 DEA 形成金叉时提醒。我把这些逻辑全部收在alert.py里用 APScheduler 每分钟检查一次。def check_alerts(): alerts get_enabled_alerts() for alert in alerts: latest get_latest_bar(alert[code]) if not latest: continue triggered evaluate_condition( latest, alert[condition_type], alert[threshold] ) if triggered and not is_cooldown(alert): send_notification(alert, latest) mark_triggered(alert)这里有一个非常重要的细节冷却时间。如果某条提醒触发后没有冷却机制那么股价在阈值附近反复穿越时短信或者微信通知会爆炸式发送。我设置的是同一规则在 60 分钟内只触发一次通知触发后记录last_triggered_at下一次判断时对比时间戳决定是否继续推送。这个设计在实战中非常有必要。通知渠道我做了三个邮件、Server酱推送到微信、自定义 Webhook。配置都放在config.py里如果不想接第三方本地也可以只是把提醒写入日志文件。多渠道的好处是灵活关键提醒走微信普通信号走日志不会让渠道成为一个瓶颈。5. 部署与日常维护5.1 使用 systemd 托管服务部署环节我一开始用的是nohup后台跑进程但服务器重启后进程不会自动恢复日志管理也不方便。后来切换到 systemd 托管一百行左右的服务配置就解决了问题。在/etc/systemd/system/openstock.service下新建服务文件[Unit] DescriptionOpenStock Service Afternetwork.target [Service] WorkingDirectory/opt/openstock ExecStart/opt/openstock/venv/bin/uvicorn app:app --host 0.0.0.0 --port 8000 Restartalways RestartSec5 [Install] WantedBymulti-user.targetRestartalways的含义是进程退出后自动重启RestartSec5是重启前等待 5 秒。这样可以保证偶尔的网络问题导致进程退出时服务能自动恢复。启动命令sudo systemctl enable openstock sudo systemctl start openstock sudo systemctl status openstock同时采集任务并不依赖 Web 服务。如果采集模块跑挂了系统只会出现数据不更新的问题页面仍然能访问。我单独把采集任务写成了一个脚本run_collector.py也用一个 systemd 服务来管每天在固定时间触发。这样 Web 服务和采集服务互不阻塞更加稳定。5.2 日志、备份与可观测性日志是排查问题最重要的工具。我在代码里统一使用 Python 的logging模块输出到两个地方控制台和文件。控制台输出方便调试时实时观察文件输出方便事后排查。日志文件会自动按天滚动import logging from logging.handlers import TimedRotatingFileHandler handler TimedRotatingFileHandler( logs/openstock.log, whenmidnight, backupCount30 ) logging.basicConfig( levellogging.INFO, handlers[handler, logging.StreamHandler()] )backupCount30保留最近 30 天的日志超过的自动清理避免日志文件无限膨胀。备份方面SQLite 数据库文件很方便我写了三个备份策略每日定时拷贝数据库文件到备份目录每周把备份文件上传到对象存储或另一台机器保留最近 30 天的备份。因为 SQLite 是单文件备份就是一条cp命令的事非常省事。5.3 跑稳之后还可以扩展什么项目跑起来以后如果你还有余力有几个方向非常值得扩展。第一是策略回测模块。把指标计算的代码与历史数据结合起来模拟买入卖出计算收益率、回撤等指标。复用daily_bars里的历史数据这个过程相对顺滑。第二是板块热力图或者多股票对比图用 ECharts 的 heatmap 系列把自选股池的当日涨跌映射成颜色一屏就能看完整个市场表现。第三是接入更多的数据源比如分钟级数据或者实时快照这需要更换数据源或者对接更高级的行情接口但项目的数据层设计已经预留了扩展空间。还有一点如果部署在公网服务器上务必设置好访问认证最简单的做法是在 Web 服务前加一层 HTTP Basic Auth 或者 Token 校验避免行情数据和自选股信息被陌生人扫描抓取。这个属于基本的安全习惯别等到出事再补。6. 常见问题与排查实录6.1 数据延迟很大怎么对齐交易日有段时间我发现自己页面上的最新数据总比行情软件慢一天。排查后发现问题出在最新交易日的判断逻辑上。我用的是自然日但股市有休市日比如周末和节假日。如果今天是周日数据库里没有新数据页面却会去取周五的数据看起来就像是延迟了。解决方法是在数据表里建一个trade_calendar交易日历表记录每年所有交易日。查询最新数据时先查日历获得最近的一个交易日再去daily_bars里取数。这样即使节假日期间打开系统显示的还是正确的最后交易日数据。6.2 数据库文件越来越大怎么办daily_bars表如果做全市场的数据累积几年下来数据量增长还是可观的。优化办法有几招只保留自选股池和持仓股票的明细数据其它股票只保留最近两年的日线定期清理超过一定年限的旧数据如果需要长期保留历史可以把旧数据导出为 CSV 或者 parquet 文件做冷存储数据库只保留热数据。我实际跑下来的经验是自选股池几十只股票的日线数据加十年历史都不到 10M完全不需要特别处理。但如果做全市场回测建议把回测数据与实时监控数据分离避免回测时的大查询拖慢 Web 响应。6.3 提醒为什么没有触发排查提醒问题我建议按这个顺序来先检查数据库里的alerts表是否启用了对应规则再检查最新一条行情数据是否真实存在然后手动执行一次check_alerts()脚本观察日志输出。最常见的原因有两个一是行情数据里最新日期不是今天提醒进程读取到的是过期数据条件自然判断不出来二是冷却时间未到规则触发过了但还在冷却期内日志里会有cooldown记录。确认这两个地方都没有问题后再往通知渠道上排查比如邮件是否被放到垃圾箱、Webhook 地址是否失效。6.4 数据源接口失效怎么办公开接口说挂就挂这件事我遇到不止一次。我的应对方法是做接口健康检查采集任务启动时先请求一个轻量的接口作为探活如果请求失败自动切换到备用数据源。备用数据源的数据格式可能略有差异所以代码里做一层适配器把不同数据源的返回统一转换成内部的数据结构。如果所有外部接口都不可用系统也可以进入降级模式页面仍能展示数据库里最后一批数据并在页面顶部显示数据更新于 xx 分钟前的提示。这样至少系统不会黑屏你也能第一时间知道数据源出了状况。6.5 前端图表不显示或者报错前端图表不显示我先看浏览器开发者工具的 Network 面板确认/api/kline/xxx这个请求是否返回了正常数据。如果接口 500通常是 SQL 查询有问题去后端日志看具体报错。如果是格式问题K 线不显示或者显示异常重点检查返回数据里的字段顺序和是否有null值。曾经遇到过一个隐蔽的问题因为数据库里某只股票的某天数据缺失了low字段ECharts 对空值处理不佳导致整条 K 线都渲染不出来。后来我在后端接口里加了数据清洗逻辑对空值做前值填充用上一天的收盘价填充这个问题才彻底解决。凡是外部数据进入系统都必须假定它可能不完整这是数据处理的基本觉悟。从最初的一台树莓派跑一个简单的抓取脚本到现在一个完整的行情监控面板OpenStock 前前后后改了很多个版本。最深的感受是个人项目最怕的不是功能少而是把架构搞得太大、各模块耦合太紧最后改一个功能就得动全身。OpenStock 这种数据采集、存储、展示、提醒分层明确的小系统反而是最适合长期维护的。搭建的过程中那些接口抖动、数据缺失、指标计算偏差的问题每一个都让我多踩了一层坑但踩完也就明白了。如果你也想搭一套自己的行情系统我的建议是别一上来就追求全市场数据先把自选股的日线跑通把页面画出来再慢慢往上加东西。系统的生命力不在于一开始有多完整而在于你能不能持续地改它、用它。
