先说个大实话:行情软件装得越多,越觉得数据不掌握在自己手里。广告弹窗、自选股同步慢、指标动不动就收费,折腾一圈下来,你会发现最舒服的方式还是自己搭一套。我最近一直用的这套开源项目 OpenStock,就是专门干这个的——它把行情采集、数据存储、API 服务和可视化看板串成了一条完整的链路,花一个晚上就能从零跑起来。这篇文章就按“手把手教你搭建 OpenStock”的思路,把完整过程、踩坑记录和部署方案一次讲清楚,适合有一定 Python 基础、想自建行情看板的技术爱好者参考。
OpenStock 本身的定位不复杂:它不是一个像 OpenStack 那样的云计算平台,而是一个面向个人投资者的开源数据展示系统,核心能力就是把公开行情数据聚合成自己的数据库,再通过网页看板展示出来。整个项目没有用重型框架,所有模块加起来也就几十个文件,个人完全能维护。下面我从设计思路开始讲,把每个环节为什么这么做、怎么做、遇到问题怎么处理都摊开来说。
1. 先盘清楚:OpenStock 的定位与整体设计思路
1.1 它到底解决了什么问题
做这个项目之前,我每天打开行情软件至少三五次,但越想越别扭:你看盘软件很多底层数据接口都是封闭的,你用它的客户端,数据就在它手里,想导出来做二次分析基本没门。更麻烦的是,我手里有几个自选股组合,想同时看 A 股、港股、指数的实时价格,不同软件之间数据口径还不一致,这边涨幅 3.2%,那边显示 3.17%,看着就心烦。
OpenStock 的目标就是把这些痛点一次解决掉。它把数据源统一收敛到后端,所有展示都基于自己存的数据库,自选股、K 线、涨跌幅、成交量这些指标想看什么自己定义。从长期来看,这套系统的价值不是省几十块钱会员费,而是真正把数据资产握在手里。行情数据通过公开接口拿回来,清洗后落地存储,之后无论做统计分析、写监控告警,还是接钉钉、企业微信推送,都是自己说了算。
很多人问,直接装个 akshare 或者 tushare 本地跑不就行了吗?说实话,只做数据分析确实够用,但你要的是“能看、能存、能对外提供接口、能和前端图表联动”的一整套服务。OpenStock 多出来的价值,恰恰是后端服务化和可视化这部分。
1.2 技术选型背后的考量
这个项目我选的是 Python 做后端,前端用 Vue3 生态里最轻量的一种写法,图表用 ECharts。选 Python 不纠结,因为股票数据这块的现成库基本都在 Python 这边,像 akshare、pandas 处理数据非常顺。FastAPI 则是最近几年个人项目里用起来最舒服的框架,自带 Swagger 文档,写接口的时候不用额外花时间维护文档,数据校验用 Pydantic 也顺手。
数据库没有上 MySQL 或者 PostgreSQL,而是直接落在 SQLite。为什么?个人使用场景下,数据量一天撑死也就几万条增量,SQLite 完全扛得住,而且零运维、单文件备份,整个系统复制到另一台机器上,拷个 db 文件就完成了迁移。等哪天真觉得性能不够了,再迁移到 PostgreSQL 也不迟,ORM 层换一下就行。
前端没有采用前后端分离的重工程结构,而是用一个 FastAPI 静态文件目录把 HTML、JS、CSS 直接挂出去。这个选择在个人项目里非常务实:省去了 Node 构建步骤,改完前端代码刷新浏览器就能看到效果,部署的时候也不用同时维护两个进程。图表用 ECharts,因为它的折线图、K 线图交互很成熟,缩放、十字光标、数据标签这些功能开箱即用,社区例子也多,遇到不会配的图表,复制一个案例改改就能跑。
1.3 整体架构与数据流
整个系统从逻辑上分成四层,我画了一个简化的调用关系:
- 采集层:定时从公开行情源抓取实时报价和历史 K 线,负责把原始数据变成结构化的 DataFrame
- 存储层:SQLite 数据库,负责保存自选股列表、实时报价快照、日 K 历史数据
- API 层:FastAPI 提供统一接口,供前端页面和外部脚本调用
- 展示层:浏览器页面,通过 ECharts 渲染行情走势和自选股表格
数据流是这样的:采集服务启动后,先读取自选股配置文件,然后请求公开行情接口拿到实时数据,解析清洗后写入 SQLite;历史 K 线则在每天收盘后通过增量更新的方式写入,避免全量重复抓取。前端页面启动时调用 API 接口加载一次全量数据,之后每隔若干秒只请求最新的实时报价,通过图表和表格无刷新更新。
这个设计的好处是层级解耦,每一层都能单独改。比如你嫌腾讯行情源的字段不够多,直接把采集层换成别的数据源,API 层和展示层都不用动;再比如你想加一个均线指标,只需要在 API 层做一次聚合计算,前端加一个 ECharts series 就行。
2. 动手前的准备:环境、依赖与项目骨架
2.1 基础环境安装
我默认你用的是 Linux 服务器或者本地 Linux/Mac 环境,Windows 也能跑,只是数据库路径和 systemd 部分需要相应调整。先确认几个基础工具是否就位:
python3 --version git --version建议 Python 版本不低于 3.10,因为 FastAPI 和 Pydantic 的新特性在低版本上会有兼容问题。没有 Python 的话,Ubuntu/Debian 系统直接用 apt 安装,注意别用系统自带的旧版本 Python 裸跑,最好用虚拟环境隔离。
接下来创建项目目录和虚拟环境:
mkdir -p /opt/openstock cd /opt/openstock python3 -m venv venv source venv/bin/activate虚拟环境这一步千万别省。我有一次图省事,把依赖直接装在系统 Python 里,后来另一个项目要装不同版本的 requests,直接互相污染,排查了半个小时才找到原因。个人项目虽然规模小,但环境隔离的习惯从一开始就养好,后面省很多事。
2.2 数据源评估与合规提醒
做行情采集,第一个要解决的问题就是数据从哪来。目前国内公开的免费行情源主要有两个:新浪财经和腾讯财经的 HTTP 接口。腾讯的qt.gtimg.cn接口返回实时报价,新浪的hq.sinajs.cn接口也是老牌方案。两个接口都支持批量请求,一次请求可以带多只股票代码,只是返回格式略有差异。
实时报价我推荐腾讯接口,字段用~分隔,解析起来稳定性比较好,数据更新速度也能到秒级。历史 K 线则直接用 akshare 封装好的接口,它对不同市场的股票做了很多兼容处理,省得自己给每个市场写解析逻辑。
这里必须多说一句合规问题:以上数据源都是公开的行情展示接口,仅限个人学习研究和少量请求使用。不要拿去做商业分发,也不要用高并发脚本反复抓取,否则很容易被限流甚至封 IP。做个人看板没问题,但要有节制,合理设置请求频率,这就是我在后面会反复强调的“克制请求”原则。
依赖安装直接写进requirements.txt:
fastapi>=0.110.0 uvicorn[standard]>=0.29.0 requests>=2.31.0 pandas>=2.2.0 akshare>=1.12.0 apscheduler>=3.10.0装依赖就一句命令:
pip install -r requirements.txtakshare 安装的时候会带很多依赖,如果在国内网络环境下有些包下载慢,可以换国内 pip 镜像源。装完最好验证一下版本,确认没有依赖冲突。
2.3 项目目录规划与配置设计
项目结构我按功能模块划分,不追求过于复杂的包结构,但也不能把代码全堆在一个文件里。下面是我实际使用的目录结构:
/opt/openstock/ ├── requirements.txt ├── config.py # 全局配置 ├── main.py # FastAPI 入口 ├── collector/ │ ├── __init__.py │ ├── realtime.py # 实时行情采集 │ └── kline.py # 历史 K 线采集与更新 ├── database/ │ ├── __init__.py │ └── db.py # SQLite 连接与建表 ├── static/ │ ├── index.html # 看板页面 │ └── app.js # 前端图表逻辑 └── stock.db # SQLite 数据库文件配置文件单独放的好处是,后续改端口、改自选股、改刷新频率都集中在一个文件里,不需要翻代码。我习惯把配置项的注释写详细一点,毕竟时间一长,你很难记得每一个参数是干嘛用的:
# config.py STOCKS = ["600000", "000001", "601318", "300750"] QUERY_INTERVAL = 5 # 前端轮询接口的间隔(秒) REFRESH_INTERVAL = 15 # 采集层更新实时报价的间隔(秒) DB_PATH = "stock.db" KLINE_START_DATE = "20240101" # K 线起始日期配置里我把自选股写成了一个列表,前端展示和后端采集都从这份配置读取。后面如果你想把自选股改成数据库里维护,也只需要动采集层和管理页面,整体改动范围非常小。
3. 核心模块实操:从抓数据到画图表
3.1 实时行情采集模块
实时行情采集是整个系统的心脏,它决定看板上的价格和数据是否及时。我用腾讯接口做实时报价采集,先看一下原始返回长什么样。请求下面的地址:
https://qt.gtimg.cn/q=sh600000返回内容是 GBK 编码的字符串,类似这样:
v_sh600000="1~浦发银行~600000~7.88~7.90~7.85~123456~..."不同字段用~分隔,常用的字段有:1 表示市场标识,2 是股票名称,3 是代码,4 是当前价,5 是昨收,6 是今开,后面的还有成交量、成交额、买一卖一等等。采集模块要做的第一件事就是注意编码:腾讯接口默认是 GBK,直接用 requests 的 text 属性会得到乱码,必须显式设置resp.encoding = "gbk"。
实时采集代码我这样写:
# collector/realtime.py import time import requests import pandas as pd def fetch_quotes(codes): symbols = [] for code in codes: if code.startswith("6"): symbols.append("sh" + code) else: symbols.append("sz" + code) url = "https://qt.gtimg.cn/q=" + ",".join(symbols) resp = requests.get(url, timeout=5) resp.encoding = "gbk" lines = resp.text.strip().split(";") rows = [] for line in lines: if "=" not in line: continue value = line.split("=", 1)[1].strip().strip('"') fields = value.split("~") if len(fields) < 10: continue rows.append({ "code": fields[2], "name": fields[1], "price": float(fields[3]), "pre_close": float(fields[4]), "open": float(fields[5]), "volume": int(fields[6]) if fields[6] else 0, "amount": float(fields[37]) if fields[37] else 0, "update_time": time.strftime("%Y-%m-%d %H:%M:%S") }) return pd.DataFrame(rows)有几个细节容易踩坑:一是代码前缀的判断不能只判断是否以 6 开头,北交所、科创板代码规则不同,个人自选股如果包含这些板块,需要额外扩充判断逻辑;二是 fields 的长度要先做校验,避免某只停牌股返回的字段不足导致数组越界;三是批量请求时代码数量不要太多,我一般控制在 30 只以内,超过就分批请求,避免返回内容过大被服务端拒绝。
3.2 历史 K 线采集与增量更新
看板只有实时价格不够,K 线趋势是判断走势的重要依据。历史 K 线用 akshare 拉取,一行代码就能拿到 DataFrame:
# collector/kline.py import akshare as ak def fetch_daily_kline(code, start_date, end_date): df = ak.stock_zh_a_hist( symbol=code, period="daily", start_date=start_date, end_date=end_date, adjust="qfq" ) return df这里有个关键选择:adjust参数要不要做复权。如果不复权,遇到除权除息的日子,K 线图上会出现价格断崖,均线指标也会失真;如果用前复权,历史价格会按最新价格调整,适合看趋势和技术指标。我做个人看板用的是qfq前复权,这样和大多数行情软件的习惯一致。
数据写入 SQLite 时要注意增量更新。全量更新每天跑一次虽然简单,但数据量会越来越大,接口请求次数也浪费。我的做法是:先查表里最新的一条日期,下次更新只抓取这个日期之后的数据,然后追加到表里。逻辑不复杂,但能有效减少重复请求:
def update_kline(code): conn = get_conn() last_date = conn.execute( "SELECT MAX(日期) FROM kline WHERE code=?", (code,) ).fetchone()[0] start = last_date if last_date else "20240101" today = time.strftime("%Y%m%d") if start >= today: return df = fetch_daily_kline(code, start, today) if df.empty: return df["code"] = code df.to_sql("kline", conn, if_exists="append", index=False) conn.close()akshare 返回的列名是中文,直接写入数据库没问题,但 API 返回时最好转成英文字段,避免前端 JS 读数据时还要处理中文 key。
3.3 FastAPI 接口设计
后端接口是整个系统的连接器。前端要看什么,接口就提供什么,我设计了三个核心接口:/api/quotes返回自选股实时行情,/api/kline返回单只股票的 K 线数据,/api/status返回系统基本状态。
用 FastAPI 实现起来很简洁:
# main.py from fastapi import FastAPI, Query from fastapi.staticfiles import StaticFiles import sqlite3 import json app = FastAPI(title="OpenStock API") app.mount("/static", StaticFiles(directory="static"), name="static") def get_conn(): conn = sqlite3.connect("stock.db") conn.row_factory = sqlite3.Row return conn @app.get("/api/quotes") def get_quotes(): conn = get_conn() rows = conn.execute("SELECT * FROM realtime").fetchall() conn.close() return {"data": [dict(r) for r in rows]} @app.get("/api/kline") def get_kline(code: str = Query(..., description="股票代码"), limit: int = 120): conn = get_conn() rows = conn.execute( "SELECT * FROM kline WHERE code=? ORDER BY 日期 DESC LIMIT ?", (code, limit) ).fetchall() conn.close() data = [dict(r) for r in rows] data.reverse() return {"code": code, "data": data}注意get_conn里面设置了row_factory = sqlite3.Row,这样查询结果才能直接转 dict,否则前端拿到的是元组,字段名要靠猜。FastAPI 的 Query 参数自动帮你完成了代码校验,接口文档在/docs页面直接能看,这也是我选 FastAPI 的一个重要原因。
3.4 前端看板页面与图表渲染
前端页面我用最朴素的方式实现:一个 HTML 文件负责布局,一个 JS 文件负责数据请求和图表渲染。页面结构分三块:顶部是总览信息和刷新时间,中间是自选股行情表格,下方是当前选中股票的 K 线图。
ECharts 的 K 线图需要的数据格式比较特殊,是[开, 收, 低, 高]这种顺序。很多人在这一步容易搞乱,前端死活画不出正常的蜡烛图,往往就是数据顺序不对。代码如下:
// static/app.js const chart = echarts.init(document.getElementById("kline")); async function loadKline(code) { const resp = await fetch(`/api/kline?code=${code}&limit=120`); const json = await resp.json(); const dates = json.data.map(d => d.日期); const values = json.data.map(d => [ d.开盘, d.收盘, d.最低, d.最高 ]); chart.setOption({ tooltip: { trigger: "axis" }, xAxis: { type: "category", data: dates }, yAxis: { scale: true }, series: [{ type: "candlestick", data: values }] }); }实时表格的刷新则用setInterval定时调用/api/quotes,每次拿到数据后局部更新表头和价格列。刷新间隔建议至少 5 秒以上,太频繁会给后端和行情源增加不必要的压力。技术上完全支持 1 秒刷新,但个人看板没必要,反而容易被限流。
整个系统跑起来以后,你在浏览器打开http://服务器IP:8000/static/index.html,就能看到行情看板了。默认端口是 8000,如果想直接访问根路径,可以在 FastAPI 里加一个根路由,把 index.html 返回。
4. 部署运行与日常维护
4.1 本地验证启动
把所有模块写完,先在本地跑一遍,确保流程通顺。启动顺序有讲究:先手动执行一次采集程序,确认数据库里有了数据,再启动 FastAPI 服务。否则会出现“接口能用但没数据”的空转状态,前端拿到空数组,图表一片空白。
我用一个简单的启动脚本把两步合并:
#!/bin/bash source /opt/openstock/venv/bin/activate python -m collector.realtime python -m collector.kline uvicorn main:app --host 0.0.0.0 --port 8000先同步一次数据,再启动服务,这样第一次打开页面就有完整内容。
4.2 用 systemd 守护进程稳定后台运行
本地验证通过后,部署到服务器或者长期运行的机器上,我推荐用 systemd 管理进程,而不是nohup或者screen。systemd 的好处很多:开机自启、进程崩溃自动重启、日志统一管理,这几个特性对常驻服务来说太重要了。
新建一个服务文件/etc/systemd/system/openstock.service:
[Unit] Description=OpenStock API Service After=network.target [Service] User=www WorkingDirectory=/opt/openstock ExecStart=/opt/openstock/venv/bin/uvicorn main:app --host 0.0.0.0 --port 8000 Restart=always RestartSec=5 Environment=PYTHONUNBUFFERED=1 [Install] WantedBy=multi-user.target然后执行:
sudo systemctl daemon-reload sudo systemctl enable openstock sudo systemctl start openstock这里踩过一个坑:服务文件里的 User 字段,一开始我用 root 运行,结果 Python 代码里所有文件权限都是 root 的,后来想用普通用户改文件内容,发现根本没有写权限。建议单独建一个www用户来跑服务,目录权限适当收紧,安全性好也方便管理。
4.3 定时任务与数据更新策略
OpenStock 的更新策略分两部分:实时行情是常驻服务里用循环加间隔控制的,历史 K 线则适合用定时任务在收盘后跑一次。A 股收盘时间是 15:00,我设定每天 16:30 更新一次日 K,这时候当天的数据基本已经稳定,再早容易拿到被修正前的数据。
定时任务我直接用 APScheduler,集成在 FastAPI 的启动事件里,省了一个 crontab 的管理成本:
# main.py from apscheduler.schedulers.background import BackgroundScheduler from collector.kline import update_all_kline scheduler = BackgroundScheduler() scheduler.add_job(update_all_kline, "cron", hour=16, minute=30) scheduler.start()这样整个系统就一个进程搞定所有事情:前端静态页面、API 接口、定时 K 线更新都在 uvicorn 里跑,运维成本降到最低。实时行情采集没必要用定时任务,因为它是持续循环的,我用一个后台任务配合asyncio.sleep来实现,代码里可以写成独立线程。
要注意的一点是,服务器时区最好设置成Asia/Shanghai,否则 APScheduler 的定时时间会按 UTC 计算,16:30 会变成第二天凌晨执行,K 线数据就差了一天。很多开发环境默认是 UTC,这个问题防不胜防。
5. 常见问题与排查技巧实录
自己搭系统,遇到的问题比你想象的多,而且很多问题在官方文档里根本不会写。我把这段时间踩过的坑按频率排了个序,整理成速查表,方便你按图索骥。
| 现象 | 可能原因 | 排查与解决 |
|---|---|---|
| 前端图表空白 | K 线数据为空或字段名对不上 | 先访问/api/kline?code=600000看返回 JSON,确认数据库有数据再查前端字段 |
| 行情数据全是 0 | 腾讯接口字段位置解析错误 | 手动请求接口,把返回字符串按~逐段拆分,核对字段索引 |
| 中文乱码 | 接口编码不是 UTF-8 | 在 requests 中显式设置resp.encoding = "gbk" |
| 请求多了就被封 | 请求频率太高触发限流 | 降低刷新频率,采集层加缓存,批量抓取时用time.sleep(1)间隔 |
| SQLite 报 database is locked | 多线程同时写入 | 打开数据库时设置check_same_thread=False,写入加锁或使用连接串timeout=10 |
| 定时任务不执行 | 服务器时区不对 | 检查系统时区,timedatectl 设置为 Asia/Shanghai |
| uvicorn 进程时不时退出 | 内存不足或代码抛异常 | 查看journalctl -u openstock日志,确认异常堆栈 |
逐个说几个典型的。
第一个是乱码问题。用腾讯接口时,如果忘记把 requests 的编码设置为 gbk,你抓回来的字符串全是非法字符,字段数量都对不上,解析时很可能直接报错。这个问题好排查,因为报错信息很明显,但新手容易在 requests 的text和content之间反复纠结。其实不用,记住一条经验:北方的行情接口很多是 GBK,遇到乱码优先设置编码,而不是换请求库。
第二个是 SQLite 并发写问题。FastAPI 是一个多线程模型,前端定时刷新会并发读接口,后台定时任务在更新 K 线时又要写数据库,读写并发的时候 SQLite 很容易报database is locked。我的处理方案是:读操作走默认连接,写操作改成串行,用scheduler单线程执行。最省事的办法是给数据库连接加一个参数:
conn = sqlite3.connect("stock.db", check_same_thread=False)但这样只是治标。真正稳妥的做法是每次操作都新建连接、用完立即关闭,不让连接跨线程复用。个人项目的数据量不大,这种方式性能完全够用。
第三个是限流问题。很多人一看到行情接口就控制不住想 1 秒刷一次,结果跑不了十分钟就被服务端限制。我自己的策略是:实时报价 15 秒刷新一次,前端页面 5 秒查询一次 API,后端查询数据库不会打到行情源,所以压力其实很小。换句话说,请求行情源的是采集层,前端页面只是读自己的数据库,中间隔了一层缓存,这个设计天然避免了对数据源的频繁请求。
第四个是部署环境时区。这个真的特别隐蔽,APScheduler 一直没反应,我还以为是服务没启动。后来date一看,服务器时间是 UTC,比北京时间慢了 8 个小时,任务当然不会在预期时间触发。解决办法也别偷懒,直接改系统时区:
sudo timedatectl set-timezone Asia/Shanghai改了之后重启服务,问题立刻消失。
我自己的操作习惯是,每加一个功能就先把日志打出来看一遍。OpenStock 的日志我用的是 Python 自带的logging,分两个文件:一个记录采集日志,一个记录 API 访问日志。遇到问题先翻日志,能少走很多弯路。比如行情采集偶尔会拿到空的 DataFrame,这时候如果日志里记录了原始返回内容,排查起来会快很多。
再分享一个进阶技巧。如果觉得 ECharts 的 K 线图不够直观,可以叠加成交量柱状图,用两个grid和两个xAxis实现,一个在上面画蜡烛图,一个在下面画成交量。这个改造很多人想加,其实原理不复杂,就是 ECharts 的多坐标系配置。把成交量数据用第二个 yAxis 绑定,缩放下面的 grid,就能得到一个非常像专业行情软件的双栏图。
数据存储方面,SQLite 单文件的好处我前面说了,但也要注意定期备份。我写了一个简单的备份脚本,每天凌晨用cp把stock.db复制一份带上日期后缀,再定期清理 30 天前的备份。数据是辛苦抓回来的,宁可备份用不上,也不能哪天真丢了再后悔。
整个 OpenStock 搭下来,我最大的体会是:真正的工作量不在代码,而在数据链路和各种边角情况的处理上。接口返回格式变了、某天数据源超时、除权除息后 K 线价格跳变,这些才是日常维护里真正会反复遇到的问题。但从零把它搭起来,再看着看板在自己写的系统里跑起来,那种掌控感是直接用现成软件完全体会不到的。后面我打算给 OpenStock 加上简单的均线指标计算和异动提醒,再考虑接入更多的数据源做交叉验证,让这套个人看板越来越顺手。