从零搭建OpenStock开源行情数据服务:半小时跑通量化数据管道
2026/9/24 12:26:43 网站建设 项目流程

先说结论:OpenStock 这类开源行情数据服务,不是给你看 K 线的,它真正解决的是"数据分散、接口不稳定、策略逻辑没法沉淀成代码"这一连串问题。我花了一个周末把整套系统从零搭到上线,照着下面这套流程走,你也能半小时跑起来,把行情数据变成自己的。

1. 为什么我放着现成行情软件不用,非要自己搭一套

1.1 现成工具的四个"不顺手"

做投资研究、写量化策略或者单纯想盯自选股的人,大概率都用过这么几种工具:手机上的行情 App、网页版财经平台、某些付费数据终端。它们不是不能用,但用久了你会发现四个很别扭的地方。

第一,数据不闭环。行情 App 上看一眼涨跌幅没问题,但你想把每分钟的数据导出来做回测、算个均线金叉,它不给你这个接口。第二,自选股逻辑是"死"的。你想按"最近 20 天涨幅超过 15% 且换手率小于 10%"这种条件筛股票,绝大多数 App 做不到,或者要开会员。第三,历史数据拿不全。很多平台只给你最近几年的日线,你一做长周期回测就抓瞎。第四,也是最关键的,策略逻辑不能沉淀。你在软件里手动翻股票、做判断,这一套操作没法保存成规则,下一次还得重新来。

所以我把目光转向了自托管方案——也就是自己搭建一套代码完全可控的行情数据服务。OpenStock 这类项目就是干这个的:从公开的行情接口抓数据、存到自己的数据库里、通过 API 暴露给前端看板或者你自己的策略脚本。数据在你手里,规则在你手里,不存在"平台改版导致功能下架"的问题。

1.2 OpenStock 解决的真正问题:数据归我所有

OpenStock 的定位,说白了就一句话:把行情数据从"看完就忘"变成"可存储、可查询、可计算"的资产。它不生产数据,它只做搬运、整理和存储。搬运的是交易所公开的行情快照,整理是把不同数据源的字段统一成自己的标准格式,存储是落进数据库里随时可以查。

这个定位意味着两件事。第一,它不涉及任何投资建议,纯粹是技术基础设施——就像你不会说 PostgreSQL 是炒股软件一样,OpenStock 只是一套数据管道。第二,它的上层应用完全开放:你可以接一个 Telegram Bot 做提醒,可以接一个 Jupyter Notebook 做策略研究,可以接一个 Web 看板做可视化,也可以什么都不接,就当一个数据库用。这种"平台化"的思维方式,是现成工具给不了你的。

1.3 这套系统最终做了什么

我具体用 OpenStock 搭出来的东西,包含这么几个模块:

  • 每日全市场行情快照采集(A 股日线级别,覆盖全部标的)
  • 分钟级增量更新任务(交易时段内每 5 分钟同步一次)
  • 一套 RESTful API,支持查询 K 线、实时行情、涨跌统计、板块聚合
  • 一个轻量级前端看板,自选股列表 + 分时走势 + 简单技术指标
  • 所有数据存放在本地数据库,完全私有化

老实说,这套东西放在 2015 年,没有专业开发能力是搞不定的。但 OpenStock 把最繁琐的采集、清洗、存储逻辑都封装好了,你只需要跟着配置跑起来就行。下面我按从零开始的顺序,把完整的搭建过程拆给你看。

2. 搭建前的架构选型:每个组件为什么这么选

2.1 整体架构一览

动手之前,先搞清楚 OpenStock 由哪些模块组成。它不是一个单体应用,而是四层结构:数据采集层、数据存储层、API 服务层、前端展示层。

数据采集层负责定时从公开行情接口拉取数据,这是整个系统的心脏。数据存储层承接采集层写入的数据,最常用的是 SQLite(个人够用)或者 PostgreSQL(数据量大或者多端访问时选用)。API 服务层把数据库里的数据封装成 HTTP 接口,方便前端和脚本调用。前端展示层就是个网页,通过 API 拿数据画图、列表。

层与层之间相互独立,意味着你可以随时替换其中任何一层。比如你不想用自带的前端,就只部署 API 服务,用自己的代码对接。你不想用自带的采集器,可以单独跑自己的脚本往数据库里插数据。这种解耦设计是我很看重的一点——它降低了对"集成度"的依赖,后期维护成本低很多。

2.2 技术栈选型理由

OpenStock 选择的技术栈,逻辑上是围绕"快速实现 + 生态丰富 + 易于二次开发"这三个原则来定的。

后端用 Python + FastAPI。Python 在数据处理领域的地位不用多说,Pandas、NumPy 这些库让行情数据的清洗和计算变得非常顺手。FastAPI 提供了很舒服的异步支持,正好匹配行情采集场景下大量的 HTTP 请求——你不能一个一个串行去拉几百个股票的行情,那太慢了,必须用异步并发。

存储层靠 SQLAlchemy ORM 做了一层隔离。当你用 SQLite 跑了一段时间觉得性能不够,想换 MySQL 或 PostgreSQL,只需要改一行连接字符串。很多项目不愿意做这种抽象,但 OpenStock 在这里做得很克制,没有引入过于复杂的中间件,SQLite 默认就能满足绝大部分个人场景。

前端看板则是 Vue 3 + ECharts。为什么不是 React?不是 React 不好,而是 Vue 的单文件组件语法更贴近"快速搭一个内部工具"的需求,学习曲线更平缓。ECharts 做金融图表是公认的好用,蜡烛图、成交量图开箱即用,省掉了很多造轮子的时间。

2.3 为什么不用现成的量化平台

可能有人会问:现在不是有很多现成的量化交易平台吗?为什么还要自己搭?我的回答是,平台和基础设施是两码事。

量化平台给你一副完整的"拐杖"——回测框架、策略编辑器、模拟交易都给你配好了,你只管写策略。但问题在于,这些平台的数据通常不开放,你没法导出完整的原始数据去做深度研究。而 OpenStock 是"基础设施",它只负责把数据给你存好、管好,至于你拿这些数据去做什么,它不管。这就像带着自己的食材去下厨,而不是只能吃别人配好的套餐。

另外,很多量化平台的免费额度非常有限,每分钟调用次数、历史数据深度都卡得很死。自托管之后,数据是自己的,想怎么折腾都行,一个月还能省下好几顿饭钱。

3. 环境准备与项目初始化:先把地基打牢

3.1 运行环境要求

OpenStock 对硬件的要求真的不高。我自己跑的时候用的是学校宿舍里一台装了 Ubuntu 22.04 的老笔记本,4 核 8G 内存,500G 机械硬盘,跑全市场日线级别的采集完全不卡。如果是个人玩玩,2 核 4G 的云服务器也足够了。注意一点,硬盘空间决定你存多久的历史数据——A股日线全市场一年大概也就几百兆,按这个量级规划即可。

软件层面的依赖有这些:

  • Python 3.9 及以上版本(推荐 3.10 或 3.11)
  • Node.js 16 及以上版本(构建前端用)
  • Git
  • 包管理器 pip 和 npm

如果系统里已经装了 Anaconda 那更省事,直接用 conda 建一个 Python 3.10 的虚拟环境最稳妥。

3.2 获取项目源码与初始化

拿到 OpenStock 源码的方式就是标准的 Git clone:

git clone https://github.com/your-repo/openstock.git cd openstock

这里要说一下为什么推荐 clone 而不是下载 zip:Git 仓库保留了完整的提交历史,后期想回退版本、对比改动非常方便。而且你可以直接 fork 一份到自己名下再 clone,这样你可以加一些自己的 commit,后续上游更新了也可以随时 merge。

接下来创建一个虚拟环境,避免依赖冲突:

python3 -m venv venv source venv/bin/activate pip install -r requirements.txt

requirements.txt 里基本涵盖了 FastAPI、Uvicorn、SQLAlchemy、Pandas、httpx、APScheduler 这些核心依赖。安装过程一般不会出问题,如果你在国内服务器上装得慢,可以换用镜像源:

pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

前端部分也在这个仓库里,安装依赖的命令是:

cd frontend npm install npm run build

如果你不需要前端看板,这步可以完全跳过。API 服务本身提供了所有数据访问能力,前端只是锦上添花。

3.3 配置文件的骨架与含义

项目根目录下会有一个 config.example.yaml,第一次部署时要把它复制成 config.yaml 再改:

cp config.example.yaml config.yaml

打开这个文件,你会看到核心配置项。我建议一个字段一个字段理解,而不是直接照抄。

database: url: "sqlite:///./data/openstock.db" collector: interval_seconds: 300 symbol_list: - "000001" - "600000" api: host: "0.0.0.0" port: 8000 enable_cors: true frontend: build_dir: "./frontend/dist"

database.url 是数据库连接串,默认连到 data 目录下的 SQLite 文件。collector.interval_seconds 是采集频率,默认 300 秒,也就是每 5 分钟采集一次。symbol_list 是你要跟踪的股票代码列表,注意这里没有交易所前缀,是因为该项目内部默认了 A 股市场。api.host 和 api.port 决定了你通过什么地址访问 API 服务。frontend.build_dir 指向打包后的前端静态文件目录。

对于第一次部署,我建议先只用 2~3 只股票跑通全流程,确认没有问题后再把 symbol_list 换成全量。你用全量在第一次跑的时候,采集任务可能要花很久,一旦配置有误排查起来也麻烦。

4. 核心模块详解与手把手配置

4.1 数据采集模块:怎么把行情拉下来

采集模块从公开行情接口抓数据,具体实现原理是这样的:用 httpx 的异步客户端同时向行情服务器发送请求,拿到证券列表的快照数据后,解析出股票名称、最新价、涨跌幅、成交量、成交额等字段,再按设定的时间间隔循环执行。

我在这里截取采集模块的逻辑核心,方便你做二次开发时候参考:

import httpx import asyncio from datetime import datetime async def fetch_quotes(symbols: list[str]): url = "https://example-quote-api.example.com" params = {"symbols": ",".join(symbols)} async with httpx.AsyncClient() as client: resp = await client.get(url, params=params) return parse_quote_resp(resp.json())

这个 parse_quote_resp 函数做了一件事:把不同数据源返回的字段统一成 OpenStock 内部的标准格式。因为行情接口返回的 JSON 可能字段名很随意,比如最新价有叫pricelasttrade的,涨跌幅有叫pct_chgchange_percent的,不统一处理,后面数据库和 API 就乱套了。

采集完成后,数据会经过一个清洗步骤再写入数据库。清洗的逻辑包括:把字符串数字转成 float、把空值填成 NaN、过滤掉停牌股票(成交量大于 0 但最新价不变的极端情况)。这一步很多人容易忽略,但恰恰是最能体现代码质量的环节——没有清洗的数据,后面所有计算都会出问题。

4.2 数据存储与增量更新策略

存储层有两种表:股票基础信息表和日线行情表。基础信息表存股票代码、名称、所属行业、上市日期这些静态数据,行情表存时间戳、开盘价、收盘价、最高价、最低价、成交量、成交额。

增量更新的关键,在于主键的设计。OpenStock 的行情表主键是代码 + 时间戳的组合,这意味着同一只股票同一时刻的数据只有一条。采集任务运行时,如果发现该股票的时间戳已经有数据,就会跳过,而不是覆盖——避免在高频采集时把盘中的高点/低点值覆盖掉。当然,你也可以改成覆盖模式,适合收盘后重新核对数据的场景,配置项里有一行update_mode: skip,改成overwrite就是覆盖模式。

我用 SQLAlchemy 定义这张核心表,示意如下:

from sqlalchemy import Column, String, Float, BigInteger, DateTime from sqlalchemy.dialects.sqlite import insert from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() class Quote(Base): __tablename__ = "quotes" symbol = Column(String(16), primary_key=True) timestamp = Column(DateTime, primary_key=True) open = Column(Float) high = Column(Float) low = Column(Float) close = Column(Float) volume = Column(BigInteger) amount = Column(Float)

注意看主键是 symbol + timestamp 的复合主键。数据库层面的约束保证了不会出现"时间戳和代码都相同但数据却不同"的脏记录,这种设计对增量抓取场景非常友好。

4.3 API 服务:给前端和策略喂数据

API 服务层是数据和外部世界之间的桥梁。OpenStock 提供的 API 端点不算多,但覆盖了基本场景:查询实时行情、查询历史 K 线、按涨跌幅排序、按行业筛选。最常用的应该是查 K 线的接口:

@app.get("/api/v1/quotes/{symbol}") def get_quotes(symbol: str, start: str = None, end: str = None, limit: int = 500): query = db.query(Quote).filter(Quote.symbol == symbol) if start: query = query.filter(Quote.timestamp >= start) if end: query = query.filter(Quote.timestamp <= end) rows = query.order_by(Quote.timestamp.asc()).limit(limit).all() return [ { "timestamp": r.timestamp, "open": r.open, "high": r.high, "low": r.low, "close": r.close, "volume": r.volume, "amount": r.amount, } for r in rows ]

这个接口设计遵循了一个原则:参数越少越好,只保留 start、end、limit 三个核心参数。为什么不做分页?因为行情数据是高度按时间线组织的,按时间范围切片比页码更适合金融数据场景。如果你有自己的策略脚本,完全可以绕过前端,直接请求这个接口拉数据,在本地用 Pandas 做计算。

4.4 前端看板:自选股监控页面

前端看板是一个单页应用,核心功能就是"列表 + 图表"。列表展示自选股的实时行情,图表展示点击某只股票后的分时/日 K 线走势。实现上是用 Fetch 定期轮询 API 接口拿到最新数据,刷新视图。ECharts 的蜡烛图配置不算复杂,但要记得开启dataZoom组件,否则数据量一多,图表会挤成一团看不清。

前端开发中最烦的是跨域问题。如果你用开发模式(npm run dev)跑前端,然后 API 在 8000 端口,前端在 5173 端口,就存在跨域请求。解决方式很简单:在 FastAPI 中加入 CORSMiddleware,把前端地址加进允许列表。这个配置项在 config.yaml 里已经有enable_cors: true这个开关,你只需要确保它开着就行。生产环境里我建议用 Nginx 做反向代理,把前端和 API 放在同一个域名下,这样根本不存在跨域问题,后面讲部署时会详细说。

5. 部署运行与验证:从命令行到浏览器

5.1 本地启动全流程

在本地把整条链路跑通,我建议按"先后端、后采集、再前端"的顺序来。

第一步,开启 API 服务:

cd ~/openstock source venv/bin/activate python -m uvicorn main:app --host 0.0.0.0 --port 8000

看到Uvicorn running on http://0.0.0.0:8000就说明 API 起来了。你可以先在浏览器访问http://localhost:8000/docs,FastAPI 自带 Swagger 交互式文档,这是调试阶段非常好用的功能,所有接口都列在那里,可以直接在线调用试一下。

第二步,另开一个终端,启动采集任务:

python -m openstock.collector start

采集任务启动后,你会看到日志文件在输出正在抓取的股票代码。第一次跑建议直接在终端前台运行,Ctrl+C 就能停,方便观察输出。确认能抓到数据后,再结合 systemd 后台常驻。

第三步,如果没有自己构建前端,可以直接用 OpenStock 自带的静态文件服务访问看板。浏览器打开http://localhost:8000应该就能看到页面。如果显示空白,十有八九是前端静态文件的路径配置不对,检查一下 frontend.build_dir 是否指向了正确的目录。

5.2 功能验证清单

系统跑起来之后,我按以下清单逐项验证,避免"看起来没问题但数据是脏的"这种尴尬情况:

  • 查询最近一次的行情快照,看时间戳是否是当前交易时段(如果是交易日且是盘中,时间应该相差不到 5 分钟)
  • 查询某只股票最近 5 天的日 K 线,比对一下收盘价和行情软件是否一致
  • 临时停掉采集服务 10 分钟再启动,看增量数据是否正常续上,有没有缺口
  • 在数据库里直接运行SELECT COUNT(*) FROM quotes WHERE symbol='000001',确认数据量在持续增长
  • 用浏览器访问前端看板,把自选股添加一列,确认图表能正确渲染

这五项是"最低验收标准"。你不需要做到多完美,但至少走完这五步,系统是真正可用的。

5.3 部署到服务器的小抄:systemd + Nginx

本地跑通了,就要部署到服务器上。这里我给出两份可以直接抄的配置。

第一份是 systemd 服务文件(/etc/systemd/system/openstock.service),负责让 API 和采集任务开机自启、崩溃自拉起:

[Unit] Description=OpenStock Service After=network.target [Service] Type=simple User=www-data WorkingDirectory=/opt/openstock ExecStart=/opt/openstock/venv/bin/python -m uvicorn main:app --host 0.0.0.0 --port 8000 Restart=always RestartSec=5 [Install] WantedBy=multi-user.target

注意这里我用/opt/openstock作为安装目录,你可以改成自己的实际路径。而采集任务建议单独再写一个 service 文件,这样你可以只重启某一个服务而不影响另一个。如果数据采集和 API 共用同一个 service,一旦采集任务启动失败,API 也一并挂掉。

第二份是 Nginx 反向代理配置(/etc/nginx/sites-available/openstock):

server { listen 80; server_name your-domain.com; location / { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; } }

这份配置把 80 端口的请求全部转发给 8000 端口上的 OpenStock API,同时带上了客户端真实 IP。这样用户访问http://your-domain.com就能打开看板,而不需要在 URL 里带端口号。如果以后想加 HTTPS,用 certbot 申请证书后改改配置就能搞定。

配置好后,执行:

sudo systemctl enable openstock sudo systemctl start openstock sudo systemctl status openstock

看到 status 是 active (running),就说明服务已经常驻了。

6. 我踩过的坑与解决记录

6.1 字段含义的坑:复权因子到底怎么算

搭建过程中最大的一个坑,就是复权数据。第一次跑完采集,我把某只股票的日 K 线和行情软件做对比,发现价格有明显的"断崖"——后来才反应过来是除权导致的。

这里解释一下:股票分红或送股之后,股价会有一个跳空缺口,不复权的数据看起来就像"跌了一大半"。技术指标计算如果基于不复权数据,会发生很多误导性的金叉死叉信号。OpenStock 的采集模块默认只抓了不复权数据,要拿到前复权数据,需要额外调用复权因子的接口,在存储前把每个价格乘以复权因子。

具体做法是在采集配置里开启adjust: "qfq"(前复权)选项。但要注意,复权因子是随着时间变化的——上市公司每次除权除息,之前所有历史价格的复权因子都会变。这意味着你不仅需要在采集时计算,还需要在每次除权日之后重算历史数据。这也是为什么我建议把原始不复权数据存一份,再单独存一份复权后的数据表,两边各司其职:原始数据用于留存备查,复权数据用于指标计算。这里我补一行配置示意:

collector: adjust: "qfq" store_raw: true

6.2 SQLite 锁与写入冲突

当采集任务和 API 服务同时访问 SQLite 数据库时,偶尔会出现database is locked的报错。这是因为 SQLite 的默认事务处理在并发写入场景下有局限——采集线程高频率写入,API 查询也在频繁读取,一旦写入事务和读取事务发生冲突,SQLite 会把较晚的那个请求判死。

解决方法其实很简单:在初始化数据库引擎时,把 WAL(Write-Ahead Logging)模式和 busy_timeout 打开:

engine = create_engine( "sqlite:///./data/openstock.db", connect_args={"timeout": 30, "isolation_level": None} )

WAL 模式下,读和写可以并发执行而不互相阻塞,这是 SQLite 实现高并发最推荐的做法。实测开 WAL 之后,一个采集进程加一个 API 服务进程同时跑,再也没见过锁错误。如果未来数据量继续膨胀,就考虑迁移到 PostgreSQL,连接串换一下即可,SQLAlchemy 抽象让切换成本很低。

6.3 时区与交易时段的判断

部署到云服务器后,我遇到了一个非常隐蔽的问题:采集任务在凌晨也在抓数据,白白浪费流量,而且产生了很多无意义的数据。查下来是时区问题——服务器默认 UTC 时间,凌晨 4 点对应北京时间中午 12 点,正好是交易时段,我按照服务器时间设的采集窗口自然就全错了。

行情系统里,所有与交易时段相关的判断必须统一使用北京时间。OpenStock 提供了一个全局配置项:

collector: timezone: "Asia/Shanghai" trading_days_file: "./data/trading_days.json"

还有一个补充配置:交易日历文件。为什么需要它?因为单单判断周一到周五是不够的,法定节假日也得排除,否则节假日也会去采集,白干活。你可以在每年最后一个交易日从交易所官方网站下载下一年的交易日历,放到 data 目录下。采集模块启动时会读取这个文件,不在交易日列表中的日期直接跳过。

6.4 数据源限流与重试策略

行情数据接口虽然公开,但也有限流保护机制。我一开始没考虑这一点,用 10 个并发线程全量拉数据,拉了一百多只股票后,请求全部超时。说明源端针对高频请求做了封禁。

解决办法是引入重试策略和请求间隔。OpenStock 的采集模块中设置了最多 3 次重试,每次重试间隔递增(2 秒、5 秒、15 秒),同时把全量抓取改成分批执行,每批之间间隔 1~2 秒。灵活一点,用指数退避算法而不是固定间隔,因为瞬时流量高峰时连续重试仍然会被限流,退避算法会让请求错峰分布。

还有一点:如果接口返回的数据明显异常,比如大量字段是 NaN、成交量为 0,我建议不要立刻重试,而是把原始返回内容保存到日志里排查。因为拉取接口本身没报错,重试也只会得到同样的坏数据。

6.5 前端 ECharts 数据格式的适配

后端 API 返回的时间戳是 ISO8601 字符串,比如2024-12-04T09:30:00,ECharts 直接拿这个当 x 轴是会报错的。你需要把它转成时间戳或者标准的格式化时间字符串。这不算什么复杂问题,但前端页面空白的时候排查起来挺绕——你会怀疑是图表配置的问题,其实是数据格式的问题。

还一个常见问题是蜡烛图必须按时间升序传入数据。如果后端排序是降序(最新的排前面),ECharts 画出来的图会把单根 K 线倒过来画,看着非常诡异。处理方法是在 API 查询层就固定order_by(Quote.timestamp.asc()),这样不管前端怎么处理都不会错。

写在搭建完之后的几点建议

OpenStock 这套系统搭好之后,我个人的体会是:它最值钱的地方不是它本身,而是它让你第一次有了属于自己的、完整可控的行情数据环境。你不再需要每次写策略都到处找数据、导数据、清洗数据,而是把时间真正花在研究和实现上。

最后分享一个小实操:把前端看板加上一个"导出 CSV"按钮,点一下就能把当前股票的历史数据导出成表格文件。这个功能实现起来很简单,但它会让你的工作流顺滑很多——每次要做研究、写报告、做演示,都不需要再打开数据库敲命令了。我试过之后,发现这已经是整个系统里使用频率最高的功能之一了。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询