这件事我琢磨了很久最后决定写一篇关于 OpenStock 的完整搭建记录。OpenStock 是一套完全开源、可自托管的股票行情数据与提醒系统它能自动抓取行情、存入数据库、提供查询接口还能在价格触发条件时主动通知你。很多朋友问我为什么放着现成App不用非要自己折腾一套这篇文章就从我的真实动机讲起把架构、环境、代码、部署、踩坑整个过程都拆开讲适合有一定开发基础、想拥有自己行情数据管道的读者。1. 为什么我不愿意继续用现成行情 App非要自建 OpenStock1.1 现成工具的痛点我以前手机里装了至少三四个行情软件每个都有自己的自选股列表却没有一个能完全满足我的需求。最常见的问题是数据不连续某天没打开App当天的分钟数据就丢了后面想做回测或者复盘历史K线缺一段整个分析就没法看。刷新也完全是手动的想盯某只股票的分时异动要么一直盯着屏幕要么频繁下拉刷新体验非常差。更让我难受的是自定义提醒太不灵活。现成App的提醒通常只能设置价格涨到XX元这种简单条件我想要的是连续三个交易日放量上涨股价突破20日均线这类组合条件App根本实现不了。后来我意识到问题是这些工具的定位是给人看的不是给数据用的所有逻辑都被封闭在别人的App里我拿不到原始数据自然也写不了自己的策略。1.2 OpenStock 到底解决什么问题OpenStock 的思路和别人不一样它把行情数据当成一种基础设施来对待。系统把行情数据定时抓下来落到自己的数据库里然后通过API对外提供查询能力。这样做有几个直接的好处数据所有权在自己手里不会因为某个App改版、下线而丢失历史数据。采集、存储、展示、提醒每个环节都解耦我可以单独替换或升级任何一层。所有规则都是代码我可以用Python写任意复杂的筛选逻辑和提醒策略。数据可以作为回测、统计、机器学习的底料不再只停留在肉眼盯盘层面。我搭好之后日常使用频率最高的功能有三个自动记录每个交易日的收盘数据、通过网页看板快速扫一眼自选股表现、在价格或指标触发条件时收到通知。这套系统不需要我人工干预只要服务器在运行它就会像钟表一样持续工作。1.3 它适合哪些人如果你是下面这几类人OpenStock 会很对胃口有一定编程基础想用自己的代码处理行情数据。对现成行情工具的提醒机制不满意希望自定义规则。正在做量化分析、个人回测或者数据可视化需要稳定的历史数据来源。单纯想通过一个真实项目学习数据采集、定时任务、API设计和前端展示。如果你只是偶尔看两眼股价完全不想碰代码那我建议你别折腾了直接用App就好。自建系统有维护成本硬盘会满、接口会变、程序会崩你必须有动手解决问题的心理准备。2. 搭建之前的架构思考OpenStock 不是一个程序而是一套流水线2.1 我把系统拆成了四块刚开始我以为 OpenStock 就是一个 Python 脚本加上一个网页就完事了。真正动手后才发现如果不想让系统变成一个跑两天就废的玩具必须从一开始就把架构理清楚。我最终拆成了四个部分数据采集层负责按固定周期从行情数据源拉取报价、K线、成交量等信息。存储层把采集到的数据统一写入数据库同时维护自选股列表、任务日志等元数据。服务层对外提供 REST API支持查询股票列表、最新报价、历史K线和告警规则。展示与通知层一个网页看板展示数据一套规则引擎在满足条件时推送告警。这样拆分之后每个环节都可以独立测试。比如我不用打开网页先单独跑采集脚本看看数据库里是否有数据不用等前端做好先用 curl 请求API确认接口返回正确。2.2 数据源怎么选稳定比免费更重要这是整个系统里最重要的决定。数据源选不好后面的工作全白搭。我先后试过几类数据源数据源类型优点缺点我的评价公共HTTP行情接口免费、无需鉴权、延迟较低接口随时可能变动、有频率限制适合个人学习和小流量场景Python开源数据库封装好、字段统一依赖维护者更新可能突然失效上手最快推荐新手商业行情服务稳定、字段丰富、有技术支持需要付费按调用量计费长期自动运行建议考虑我最终选择了开源数据库为主公共HTTP接口为备用的方案。主数据源负责常规的日线、分钟线抓取备用数据源用于主数据源异常时的补充验证。两个数据源不能同时挂掉否则整个系统就成了摆设。选数据源还有一个原则先确认它在你的部署环境下是否可访问、是否合法合规。在国内服务器上部署就选国内可达的公开数据源在海外服务器上部署就选当地合规的服务。别装一个不可达的依赖然后到处问为什么超时。2.3 技术栈选择为什么用 Python FastAPI SQLite Redis技术栈我没有追求新潮只选了自己最熟悉且社区生态最稳的组合采集脚本用 Python股票相关的开源库绝大多数都是Python写的处理数据也方便。API服务用 FastAPI自带Swagger文档调试接口不用额外工具性能对个人项目绰绰有余。数据库先用 SQLite数据量大了再换 PostgreSQL个人自用单机并发不高SQLite 零配置、备份方便一个文件就能打包。等历史数据超过几千万条我会迁移到 PostgreSQL但前期完全没必要给自己增加运维负担。Redis 用来做缓存和轻量任务队列热点API加缓存可以显著提高响应速度告警任务也可以丢到队列里异步处理。这个组合的好处是部署简单依赖少遇到问题网上资料多。Python 采集端和 FastAPI 服务端可以共享同一个数据库代码维护起来也不会太分裂。3. 手把手搭建 OpenStock从环境准备到跑通第一个页面3.1 环境准备我以 Ubuntu 22.04 服务器为例操作在本地电脑一样可以完成只是地址改成 localhost 就行。建议至少 2核4G 内存磁盘预留 20G 以上因为几年分钟线数据积累下来体积会超出你的预期。需要提前装好Docker 与 Docker Compose用于一键启动依赖组件。Python 3.10 及以上版本用于运行采集脚本和 FastAPI 应用。Node.js 18 以上如果要用前端脚手架构建前端也可以直接用静态HTML。我建议把项目放在/opt/openstock下避免权限问题和路径混乱。3.2 项目目录结构参考我习惯把代码分成采集、服务、前端三个目录。一个干净的项目结构大概是这样的openstock/ ├── collector/ # 数据采集与定时任务 │ ├── main.py │ ├── jobs.py │ └── sources/ ├── server/ # FastAPI 服务 │ ├── main.py │ ├── models.py │ └── routers/ ├── web/ # 前端看板 │ ├── index.html │ └── static/ ├── data/ # SQLite 数据库文件目录 ├── docker-compose.yml └── .env先创建这个结构然后一步一步填充。3.3 数据库初始化我直接用 SQLite建表时就把唯一约束设好避免重复数据。拿自选股表和K线表举例CREATE TABLE IF NOT EXISTS stocks ( symbol TEXT PRIMARY KEY, name TEXT NOT NULL, market TEXT NOT NULL, enabled INTEGER DEFAULT 1, created_at TEXT DEFAULT (datetime(now)) ); CREATE TABLE IF NOT EXISTS daily_bars ( symbol TEXT NOT NULL, date TEXT NOT NULL, open REAL NOT NULL, high REAL NOT NULL, low REAL NOT NULL, close REAL NOT NULL, volume REAL NOT NULL, PRIMARY KEY (symbol, date) );daily_bars的主键是(symbol, date)意思是同一只股票同一天只能有一条记录重复写入会直接冲突正好用来做去重。你不需要在一个表里存所有数据日线、分钟线、公告、资金流分开表后面查询和清理都方便。3.4 配置数据源在.env里配置数据源相关参数。拿公开数据源举例DB_PATH/opt/openstock/data/openstock.db FETCH_INTERVAL60 KLINE_PERIODdaily DEFAULT_STOCKS000001.SZ,600519.SH,00700.HK REQUEST_TIMEOUT10我坚持把配置和代码分离因为数据源地址、更新频率、自选股列表这些内容会经常变没必要每次改配置都重新部署代码。3.5 用 Docker Compose 启动如果只是采集和API裸跑 Python 够了。但我把 Redis 也纳入了管理用 Docker Compose 统一启动version: 3.8 services: redis: image: redis:7-alpine restart: always ports: - 6379:6379 api: build: ./server restart: always ports: - 8000:8000 volumes: - /opt/openstock/data:/app/data environment: - DB_PATH/app/data/openstock.db - REDIS_URLredis://redis:6379/0 depends_on: - redis执行cd /opt/openstock docker compose up -d第一次启动会自动拉镜像之后每次更新代码只需要docker compose restart api对个人项目足够用了。3.6 首次启动验证打开浏览器访问http://服务器IP:8000/docs你会看到 FastAPI 自带的Swagger接口文档。先调用一个简单的健康检查接口curl http://127.0.0.1:8000/health如果返回类似{status:ok}说明服务已经跑起来了。注意这时候数据库里还是空的需要先执行采集任务才能在前端看到数据。4. 行情数据入库与定时任务让系统自己长期跑起来4.1 我只信调度任务不靠手动刷数据采集最重要的是节奏感。A股开盘时间是 9:30 到 11:30、13:00 到 15:00如果每秒钟都去拉一次数据不仅浪费资源还容易被数据源限制。我的做法是分成两个节奏盘中使用短周期增量采集比如每 1 分钟同步一次最新价和成交量。收盘后做一次全量日线更新把当天最终数据补齐。定时任务我直接用 Python 的APScheduler因为它支持cron表达式能精确控制工作日执行。from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger scheduler BackgroundScheduler() scheduler.add_job( fetch_intraday, CronTrigger(day_of_weekmon-fri, hour9-11,13-14, minute*/1), idintraday, max_instances2, coalesceTrue, ) scheduler.add_job( fetch_daily, CronTrigger(day_of_weekmon-fri, hour15, minute5), iddaily, max_instances1, ) scheduler.start()这里最重要的是coalesceTrue。如果服务器在交易时间宕机重启错过多个执行周期它不会把漏掉的周期全部补跑而是合并到最近一次避免任务堆积。4.2 抓取、落库、去重三步走一次完整采集流程我总结为三步第一从自选股表读取需要采集的股票列表。不要在每个函数里硬编码股票代码否则以后换股票要改代码。def get_watchlist(): rows db.execute(SELECT symbol FROM stocks WHERE enabled 1).fetchall() return [r[symbol] for r in rows]第二调用数据源接口获取实时行情。返回结果通常是一个包含symbol, price, open, high, low, volume, timestamp的词典列表。拿到后不要直接入库先做字段校验和类型转换因为不同数据源的字段命名差异很大。第三写入数据库。SQLite 支持INSERT OR IGNORE加上前面设置过的唯一约束天然防止重复db.executemany( INSERT OR IGNORE INTO daily_bars (symbol, date, open, high, low, close, volume) VALUES (:symbol, :date, :open, :high, :low, :close, :volume) , records, )这样即使任务重复执行数据也不会翻倍。4.3 交易日历与停牌问题我最早踩过一个大坑周末和非交易日定时任务还在跑结果拉回来的全是上一交易日的旧数据导致我误以为实时行情没更新。后来我加了一个交易日历判断只有当天是交易日才执行盘中任务。def is_trading_day(dt): if dt.weekday() 5: return False # 这里可以接入每年的节假日安排数据源 return True停牌股票也要特别处理。很多数据源在股票停牌时返回的成交量为 0价格保持停牌前收盘价。如果你用价格没变就跳过入库的逻辑就会漏掉正常数据。我的做法是直接入库但标记volume0查询时再过滤。5. 查询接口与前端看板把库存数据变成真正能看的界面5.1 后端 API 不搞花活够用就行我用 FastAPI 写了几个核心接口分别是GET /api/stocks获取自选股列表。GET /api/quotes获取多只股票的最新行情。GET /api/kline?symbolXXXperioddaylimit120获取某只股票的历史K线。GET /api/alerts获取当前告警规则。接口返回统一使用 JSON前端不需要关心数据源差异。以K线接口为例简单写就是app.get(/api/kline) def get_kline(symbol: str, limit: int 120): rows db.execute( SELECT date, open, high, low, close, volume FROM daily_bars WHERE symbol :symbol ORDER BY date DESC LIMIT :limit , {symbol: symbol, limit: limit}, ).fetchall() return {symbol: symbol, items: rows}我加了 Redis 缓存对limit比较小的热门查询直接从缓存返回数据库压力会小很多。5.2 前端看板我不追求炫酷只求一眼看到重点前端我用了一个很轻量的方案一个index.html配合原生 JavaScript 和一个小型绘图库比如 ECharts。页面分三块区域顶部自选股列表显示最新价、涨跌幅、成交量并按涨跌幅排序。中间是K线图点击股票列表时自动切换。右侧是告警信息显示最近触发的提醒。为什么不引入重型前端框架因为这套系统实际使用场景多数是我自己的 PC 或手机浏览器不需要复杂状态管理原生 JS 足够。前端定期轮询/api/quotes比如每 30 秒刷新一次实时性完全够用。如果你想做成 WebSocket 推送后面的扩展空间也很大因为数据层已经和展示层解耦了。5.3 一个让我意外的心态转变搭完前端后我发现数据可视化带来的价值比想象中大。以前用App看到的是单只股票的孤立数字现在打开自己的看板所有自选股涨跌排序一目了然板块变化也能从成交量的柱状图里看出端倪。数据一旦在数据库里你就可以用任意工具去分析它这是第三方App永远给不了的自由。6. 告警提醒让 OpenStock 从显示屏变成哨兵6.1 把提醒条件写成规则告警是 OpenStock 最有价值的模块。我实现了一个简单的规则引擎规则以 JSON 形式存在数据库里{ id: 1, symbol: 600519.SH, condition: price_above, threshold: 1500.0 }支持的规则类型包括但不限于price_above价格高于、price_below价格低于、change_percent_above涨跌幅高于、volume_above成交量高于。规则求值放在每次采集完成之后def check_alerts(bars): for rule in get_rules(): quote bars.get(rule[symbol]) if not quote: continue if rule[condition] price_above and quote[close] rule[threshold]: send_notification(rule[symbol], f价格突破 {rule[threshold]})关键点在于每次规则触发后至少标记为已通知一段时间内不重复轰炸。我用 Redis 存了一个last_notify:{rule_id}键设置 60 秒过期防止同一条件下一条消息刷屏。6.2 通知渠道我优先推荐 Server酱和邮件通知渠道我接了三类邮件最通用自己搭一个SMTP发送即可。Server酱通过微信收到推送个人使用免费接入非常容易。企业微信/钉钉机器人适合想推送给自己或者一个小团队的场景webhook 地址配置一下就能用。以 Server酱为例发送消息其实就是发一个 HTTP 请求curl https://sctapi.ftqq.com/你的SendKey.send?titleOpenStock告警desp600519.SH已突破1500元这种方式稳定可靠不用自己维护App推送通道。我实际使用的时候把重点提醒放在 Server酱把每天收盘的汇总数据发到邮箱两个渠道互补。6.3 组合条件才是进阶玩法简单阈值满足不了我所以我后来实现了组合条件。比如当日涨幅超过 5% 并且成交量大于过去 5 日均量的 2 倍这类条件需要先算出一个结果再判断是否触发def check_volume_surge(symbol): bars get_recent_bars(symbol, 6) avg_vol sum(b[volume] for b in bars[:-1]) / 5 today_vol bars[-1][volume] return today_vol avg_vol * 2这种逻辑在现成App里几乎不可能自定义但在 OpenStock 里只是多写几行代码的事。告警模块一旦跑起来整个系统就从被动的数据展示工具变成了主动盯盘助手这是让我觉得所有折腾都值得的关键功能。7. 搭建过程中我真实踩过的坑以及最后的一些提醒7.1 数据源限流比想象中来得更快第一次跑全量采集时我一次性把上千只股票的数据都去请求结果几秒钟之后数据源直接把我的IP限制了一段时间等了差不多半小时才恢复。从那以后我学乖了批量任务一定要加限速每请求一只股票之后至少间隔 0.1 到 0.5 秒。代码层面可以用一个简单的sleep但更好看的是用并发控制在合理范围from concurrent.futures import ThreadPoolExecutor import time executor ThreadPoolExecutor(max_workers5) def fetch_with_interval(symbol): data fetch_one(symbol) time.sleep(0.2) return data限制最大并发数配合固定间隔数据源就不会再把你当恶意访问者处理。7.2 时区、复权和停牌三个隐藏的坑时区问题最容易让人困惑。很多公开数据源返回的时间戳是 UTC 时间A股数据如果直接按 UTC 存取凌晨 0 点会被算成前一天 16 点。我统一在写入前把时间转换为北京时间数据库里所有时间字段显式标注时区比如2025-01-10 15:00:0008:00。查询的时候再按需转换避免前后端各转一次导致偏差。复权问题是做历史K线必须考虑的。前复权、后复权、不复权的价格完全不同。我建议数据库里保留不复权原始价在需要计算指标时通过接口传参获取复权价格。这样数据不会失真也能应对数据源偶尔更新除权信息的情况。停牌股票的成交量是 0某些数据源的close可能返回null不注意就会在计算涨跌幅时把整条数据变成NaN进而污染指标。我在采集端统一处理成交量缺失填 0价格缺失用最后一条有效价格填充。7.3 SQLite 并发写入的体感SQLite 在高并发写入场景下会出现database is locked错误。我一开始没当回事结果采集脚本和告警任务同时写表时一天能撞上五六次。解决方案有两个把所有写入操作放到同一个进程里串行执行等待队列由任务调度器保证。如果多个进程都需要写把 SQLite 换成 PostgreSQL。我在采集脚本里用了一个threading.Lock保护写函数基本就解决了。如果你的自选股数量和刷新频率比我高很多建议直接上 PostgreSQL不要再和 SQLite 较劲。7.4 合规与安全提醒最后想认真提醒一句OpenStock 适合用来做个人技术学习、数据研究和自动化实践不要拿它去做任何非法数据抓取也不要商用未经授权的行情数据。不同数据源有不同的服务条款你用之前最好逐条看清尤其是能否存储、能否再分发。我自己部署的时候做了两件事一是 API 端口不直接暴露到公网用 Nginx 加一层反向代理并开启简单的访问认证二是给系统所在目录做好定期备份每天备份一次 SQLite 文件到另一个磁盘。数据是自己的丢了就真的没了。搭建 OpenStock 这件事我从一开始只是想省掉打开App的麻烦到最后却把数据采集、存储、API、前端、通知全链路都练了一遍。整个过程里最大的体会是不要求一步到位先把最小闭环跑通再慢慢加功能。你不需要在所有设计上都符合最佳实践只需要让系统在你自己的使用场景下稳定可靠。把采集、入库、展示、提醒这条流水线跑通之后后面任何新想法都只是往里加模块的问题。