先说结论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.11Node.js 16 及以上版本构建前端用Git包管理器 pip 和 npm如果系统里已经装了 Anaconda 那更省事直接用 conda 建一个 Python 3.10 的虚拟环境最稳妥。3.2 获取项目源码与初始化拿到 OpenStock 源码的方式就是标准的 Git clonegit clone https://github.com/your-repo/openstock.git cd openstock这里要说一下为什么推荐 clone 而不是下载 zipGit 仓库保留了完整的提交历史后期想回退版本、对比改动非常方便。而且你可以直接 fork 一份到自己名下再 clone这样你可以加一些自己的 commit后续上游更新了也可以随时 merge。接下来创建一个虚拟环境避免依赖冲突python3 -m venv venv source venv/bin/activate pip install -r requirements.txtrequirements.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/distdatabase.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, paramsparams) return parse_quote_resp(resp.json())这个 parse_quote_resp 函数做了一件事把不同数据源返回的字段统一成 OpenStock 内部的标准格式。因为行情接口返回的 JSON 可能字段名很随意比如最新价有叫price、last、trade的涨跌幅有叫pct_chg、change_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_keyTrue) timestamp Column(DateTime, primary_keyTrue) 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/docsFastAPI 自带 Swagger 交互式文档这是调试阶段非常好用的功能所有接口都列在那里可以直接在线调用试一下。第二步另开一个终端启动采集任务python -m openstock.collector start采集任务启动后你会看到日志文件在输出正在抓取的股票代码。第一次跑建议直接在终端前台运行CtrlC 就能停方便观察输出。确认能抓到数据后再结合 systemd 后台常驻。第三步如果没有自己构建前端可以直接用 OpenStock 自带的静态文件服务访问看板。浏览器打开http://localhost:8000应该就能看到页面。如果显示空白十有八九是前端静态文件的路径配置不对检查一下 frontend.build_dir 是否指向了正确的目录。5.2 功能验证清单系统跑起来之后我按以下清单逐项验证避免看起来没问题但数据是脏的这种尴尬情况查询最近一次的行情快照看时间戳是否是当前交易时段如果是交易日且是盘中时间应该相差不到 5 分钟查询某只股票最近 5 天的日 K 线比对一下收盘价和行情软件是否一致临时停掉采集服务 10 分钟再启动看增量数据是否正常续上有没有缺口在数据库里直接运行SELECT COUNT(*) FROM quotes WHERE symbol000001确认数据量在持续增长用浏览器访问前端看板把自选股添加一列确认图表能正确渲染这五项是最低验收标准。你不需要做到多完美但至少走完这五步系统是真正可用的。5.3 部署到服务器的小抄systemd Nginx本地跑通了就要部署到服务器上。这里我给出两份可以直接抄的配置。第一份是 systemd 服务文件/etc/systemd/system/openstock.service负责让 API 和采集任务开机自启、崩溃自拉起[Unit] DescriptionOpenStock Service Afternetwork.target [Service] Typesimple Userwww-data WorkingDirectory/opt/openstock ExecStart/opt/openstock/venv/bin/python -m uvicorn main:app --host 0.0.0.0 --port 8000 Restartalways RestartSec5 [Install] WantedBymulti-user.target注意这里我用/opt/openstock作为安装目录你可以改成自己的实际路径。而采集任务建议单独再写一个 service 文件这样你可以只重启某一个服务而不影响另一个。如果数据采集和 API 共用同一个 service一旦采集任务启动失败API 也一并挂掉。第二份是 Nginx 反向代理配置/etc/nginx/sites-available/openstockserver { 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: true6.2 SQLite 锁与写入冲突当采集任务和 API 服务同时访问 SQLite 数据库时偶尔会出现database is locked的报错。这是因为 SQLite 的默认事务处理在并发写入场景下有局限——采集线程高频率写入API 查询也在频繁读取一旦写入事务和读取事务发生冲突SQLite 会把较晚的那个请求判死。解决方法其实很简单在初始化数据库引擎时把 WALWrite-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:00ECharts 直接拿这个当 x 轴是会报错的。你需要把它转成时间戳或者标准的格式化时间字符串。这不算什么复杂问题但前端页面空白的时候排查起来挺绕——你会怀疑是图表配置的问题其实是数据格式的问题。还一个常见问题是蜡烛图必须按时间升序传入数据。如果后端排序是降序最新的排前面ECharts 画出来的图会把单根 K 线倒过来画看着非常诡异。处理方法是在 API 查询层就固定order_by(Quote.timestamp.asc())这样不管前端怎么处理都不会错。写在搭建完之后的几点建议OpenStock 这套系统搭好之后我个人的体会是它最值钱的地方不是它本身而是它让你第一次有了属于自己的、完整可控的行情数据环境。你不再需要每次写策略都到处找数据、导数据、清洗数据而是把时间真正花在研究和实现上。最后分享一个小实操把前端看板加上一个导出 CSV按钮点一下就能把当前股票的历史数据导出成表格文件。这个功能实现起来很简单但它会让你的工作流顺滑很多——每次要做研究、写报告、做演示都不需要再打开数据库敲命令了。我试过之后发现这已经是整个系统里使用频率最高的功能之一了。
