我最初关注到 OpenStock 这个项目是因为长期盯盘用的行情软件越来越臃肿广告、弹窗、资讯推送全往脸上招呼。我就想能不能自己搭一个轻量、开源的股票监控平台只看自己关心的自选股行情顺带把每日复盘数据自动汇总。折腾了大概两个晚上我把整个架子跑通了从数据获取、存储、后端接口到前端展示全部自己掌控这个过程中踩了不少坑也总结了一些关键经验。这篇就手把手带你把 OpenStock 搭起来聊清楚每一步为什么这么做以及哪些地方容易翻车。这篇内容适合有一定 Python 基础、对股票数据感兴趣但不想被商业软件绑定的开发者来参考。它不是一个完整的量化交易系统而是一个能自托管、可扩展的行情监控与自选股管理平台。如果你只是想快速看盘可能用现成的 App 更省事如果你想搞清楚股票数据怎么来、怎么存、怎么展示甚至后续想往量化方向走那 OpenStock 这个项目绝对值得折腾一遍。1. 项目定位与技术选型1.1 OpenStock 到底解决什么问题市面上的行情软件大多围绕“让用户多停留”设计首页塞满热点头条、投顾直播、广告位真正对你有用的可能就是那几行价格数据。OpenStock 的思路反过来它只做三件事拉取行情数据、管理自选股列表、展示 K 线和关键指标。这个项目不是要做一个替代同花顺、雪球的商业产品而是把“看行情”这个需求精简到极致。你部署在自己的服务器上数据是自己的代码是自己的想改逻辑随时改。因为代码量不大整个项目结构非常透明后续想加自己的技术指标、回测功能或者接入更多数据源都比较容易扩展。为了让你对项目结构有个直观印象我用一张目录结构图来说明这是搭建前必须搞清楚的骨架openstock/ ├── app.py # FastAPI 主入口 ├── database.py # SQLite 连接与建表 ├── data_fetcher.py # 数据采集模块封装 AKShare 接口 ├── scheduler.py # 定时任务模块 ├── static/ │ └── index.html # 单页前端 └── requirements.txt # 依赖清单1.2 技术栈选型为什么是这些不是那些语言选了 Python理由是股票数据相关的开源库基本都集中在 Python 生态尤其是 AKShare 这个财经数据接口库免费、无需 token、覆盖 A 股、港股、美股和期货数据非常适合个人项目快速起步。爬虫直接抓网页太脆弱K 线数据、实时行情这种需要稳定结构和字段的数据用成熟的数据接口库比自己写解析器靠谱得多。后端框架用 FastAPI 而不是 Flask主要看中它的异步性能和自动生成 API 文档。虽然这个项目的数据量很小但 FastAPI 的 /docs 页面能让你在调试接口时省很多时间浏览器打开就能直接测试每个接口、看返回参数结构。数据库选用 SQLite因为当前项目体量没有并发写入压力一台个人服务器存几千只股票的日线数据SQLite 绰绰有余而且备份就是拷一个文件省去安装配置 MySQL 的麻烦。前端没引入 Vue 或 React就用原生 HTML 加 ECharts。因为整个页面只有一个 K 线图加一个自选股列表框架带来的复杂度远大于收益。ECharts 自带股票 K 线图、成交量图、缩放组件写起来是最短的路径。2. 数据获取与存储设计2.1 数据源选择AKShare 的实战用法AKShare 是一个纯开源接口库它把东方财富等公开财经网站的接口做了统一封装调用方式非常简单。以获取 A 股实时行情为例import akshare as ak # 获取全部 A 股实时行情 df ak.stock_zh_a_spot_em() print(df.head())返回的是一个 pandas DataFrame里面包含股票代码、名称、最新价、涨跌幅、成交量、成交额、换手率等字段。单次调用返回约 5000 行数据用来做全市场扫描足够了但注意别频繁调用。我实测下来AKShare 的免费接口不是无限量供应的短时间大量请求会被源网站临时封 IP需要做限速和缓存。历史 K 线数据用另一个接口# 获取平安银行000001的日线数据 df ak.stock_zh_a_hist( symbol000001, perioddaily, # 周期daily/weekly/monthly adjustqfq # 复权方式qfq前复权/不复权 )这里有三个容易踩坑的点需要重点提醒。第一股票代码必须是六位字符串比如平安银行是000001如果直接传 1 会报错需要用str(code).zfill(6)补零。第二adjust参数建议用qfq前复权因为你会看长期 K 线前复权数据能正确反映当前价格相对于历史价格的涨跌幅度避免除权除息导致的跳空。第三这个接口默认只返回有交易的数据停牌日不会出现在结果里处理时间序列时要留意日期缺口。2.2 SQLite 表结构设计OpenStock 的数据库设计追求极简但也不是随便建几张表就行。我的方案是三张表加一张预警表-- 自选股清单 CREATE TABLE IF NOT EXISTS watchlist ( id INTEGER PRIMARY KEY AUTOINCREMENT, symbol TEXT NOT NULL UNIQUE, -- 股票代码 name TEXT NOT NULL, -- 股票名称 added_at TEXT DEFAULT CURRENT_TIMESTAMP ); -- 日K线数据 CREATE TABLE IF NOT EXISTS daily_kline ( symbol TEXT NOT NULL, date TEXT NOT NULL, -- 交易日 open REAL, high REAL, low REAL, close REAL, volume INTEGER, PRIMARY KEY (symbol, date) ); -- 行情快照最近一次拉取的全市场快照 CREATE TABLE IF NOT EXISTS market_snapshot ( symbol TEXT PRIMARY KEY, name TEXT, price REAL, change_percent REAL, volume REAL, updated_at TEXT ); -- 触发预警的股票记录 CREATE TABLE IF NOT EXISTS price_alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, symbol TEXT NOT NULL, price REAL NOT NULL, type TEXT NOT NULL, -- up/down triggered_at TEXT DEFAULT CURRENT_TIMESTAMP );daily_kline表用联合主键(symbol, date)防止重复数据。因为同一个指标的 K 线数据是周期拉取的如果当天重复同步必须用INSERT OR REPLACE或先查后插。market_snapshot这表很重要如果把全市场 5000 只股票的数据存在一张表里每次查询都会全表扫描而快照表只保留最新一版查询效率高得多。2.3 数据同步脚本需要处理的关键细节写数据同步脚本时最烦的是增量与全量的问题。我的处理办法是全量 K 线只在首次部署时跑一次之后每个交易日收盘后跑一次增量更新。增量更新怎么做不是每次把一年的数据全部拉下来而是只拉最近 5 个交易日的然后用INSERT OR REPLACE写库import akshare as ak import sqlite3 def sync_kline_latest(conn, symbol): 增量同步最近5个交易日K线 df ak.stock_zh_a_hist(symbolsymbol, perioddaily, adjustqfq, start_datedelta_date) if df is None or df.empty: return 0 df df[[日期, 开盘, 最高, 最低, 收盘, 成交量]] df.columns [date, open, high, low, close, volume] data [(symbol, row.date, row.open, row.high, row.low, row.close, row.volume) for row in df.itertuples()] conn.executemany( INSERT OR REPLACE INTO daily_kline (symbol, date, open, high, low, close, volume) VALUES (?, ?, ?, ?, ?, ?, ?) , data) conn.commit() return len(data)那个str(symbol).zfill(6)的坑一定要提到前面AKShare 要求代码为字符串且不足六位要补零。你可以封装一个normalize_symbol函数所有入口和数据校验都过这个函数而不是每次手动处理这样做的好处是当后面添加基金、指数代码时统一逻辑不会乱。3. 后端接口与预警模块实现3.1 FastAPI 接口设计与核心代码后端是整个项目的中枢负责把 SQLite 里的数据通过 JSON 接口暴露给前端。最核心的接口有三个自选股列表、K 线数据、添加/删除自选股。from fastapi import FastAPI, HTTPException, Query from fastapi.middleware.cors import CORSMiddleware import sqlite3, json app FastAPI() app.add_middleware(CORSMiddleware, allow_origins[*], allow_methods[*]) def get_db(): conn sqlite3.connect(openstock.db, check_same_threadFalse) conn.row_factory sqlite3.Row return conn app.get(/api/watchlist) def get_watchlist(): conn get_db() rows conn.execute( SELECT w.symbol, w.name, s.price, s.change_percent FROM watchlist w LEFT JOIN market_snapshot s ON w.symbol s.symbol ORDER BY w.added_at ).fetchall() return [dict(r) for r in rows] app.get(/api/kline/{symbol}) def get_kline(symbol: str, limit: int 250): conn get_db() rows conn.execute( SELECT date, open, high, low, close, volume FROM daily_kline WHERE symbol ? ORDER BY date DESC LIMIT ? , (symbol, limit)).fetchall() # 需要反转成时间升序ECharts 才方便画 rows list(reversed(rows)) return [dict(r) for r in rows] app.post(/api/watchlist) def add_watchlist(symbol: str, name: str Query()): ...这里有个小细节sqlite3.connect加check_same_threadFalse是因为 FastAPI 在多线程模式下可能从不同线程访问连接不加会报错。但要注意这只是绕过了安全检查不代表连接是线程安全的实际操作中我对写操作加了锁读操作可以放开。响应数据的时间顺序也是个容易忽略的坑。数据库里存的是时间升序但如果用ORDER BY date DESC LIMIT 250取最近的数据拿到的是倒序。画 K 线图必须保证数组从旧到新所以读取后再reversed一下。这种细节一旦亲自跑一遍就会记住因为图表数据反过来画的 K 线完全是乱的。3.2 预警模块基于 APScheduler 的定时监控预警模块是让这个项目从“被动看盘”变成“主动盯盘”的关键。我用 APScheduler 写了一个后台任务每 5 分钟扫一次自选股的最新价如果当日涨跌幅超过阈值就把记录写到price_alerts表里同时打印日志。from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.interval import IntervalTrigger def check_alerts(): conn get_db() stocks conn.execute(SELECT symbol, name FROM watchlist).fetchall() for stock in stocks: df ak.stock_zh_a_spot_em() # 频繁调用风险高注意优化 row df[df[代码] stock[symbol]] if row.empty: continue price float(row.iloc[0][最新价]) pct float(row.iloc[0][涨跌幅]) if pct 5 or pct -5: conn.execute( INSERT INTO price_alerts (symbol, price, type) VALUES (?, ?, ?) , (stock[symbol], price, up if pct 0 else down)) conn.commit() scheduler BackgroundScheduler() scheduler.add_job(check_alerts, triggerIntervalTrigger(minutes5)) scheduler.start()这段代码能跑通但把stock_zh_a_spot_em()放在循环里效率很低。自选股有几十只每扫一次就全市场拉一次数据接口很容易被限制。优化方案是每轮循环只拉一次全市场快照在内存里做过滤然后统一写库。所以我在实际项目里改成了这样一次性拉取之后用字典快速查找def check_alerts_optimized(): conn get_db() stocks conn.execute(SELECT symbol FROM watchlist).fetchall() symbols [s[symbol] for s in stocks] df ak.stock_zh_a_spot_em() df df[df[代码].isin(symbols)] for row in df.itertuples(): symbol row.代码 pct float(row.涨跌幅) if pct 5 or pct -5: # 记录预警 ...预警只是日志记录不会主动推送消息。如果你有钉钉或企业微信的机器人可以把告警信息通过 webhook 发出去这里不展开了但结构上已经留好了扩展位。3.3 启动入口与运行注意事项整个项目通过app.py启动核心是同时拉起 FastAPI 和后端定时任务import uvicorn from fastapi import FastAPI from scheduler import scheduler app FastAPI() app.on_event(startup) def on_startup(): scheduler.start() scheduler_jobs scheduler.get_jobs() print(fScheduler started, {len(scheduler_jobs)} jobs running) if __name__ __main__: uvicorn.run(app:app, host0.0.0.0, port8000, reloadTrue)启动项目时需要注意如果你用的是reloadTrue模式HTTP 服务代码发生修改会触发整个进程重启可能出现重复任务或端口被占用。建议开发期用reloadTrue部署时改成reloadFalse。另外强烈建议把项目放进虚拟环境里跑避免和系统 Python 的包版本冲突。4. 前端页面K 线图与自选股面板4.1 页面布局与交互逻辑前端我用的是一个 static 目录下的 index.html单页搞定所有交互。页面布局主要分左、中、右三个区域左侧是自选股列表显示名称、代码、当前价、涨跌幅中间主区域是 K 线图右侧是每日复盘数据表包括当日涨停数量、跌停数量、成交量对比等指标。这个布局是基于实际操作习惯定的。我盯盘时最关心的其实是两件事自选股当前什么情况以及大盘/自己想看的股票的 K 线形态。所以左侧列表必须始终可见中间图表跟随点击切换股票右侧统计信息放在辅助位置。4.2 ECharts K 线图的正确接入方式ECharts 股票图使用candlestick系列类型要求数据格式是一个二维数组每个元素是[开盘, 收盘, 最低, 最高]。注意这个顺序不是常见的[开、高、低、收]我在第一次接入时就栽在这里导致 K 线阴影方向完全反了。正确的数据组装代码const klineData data.map(item ([ item.open, item.close, item.low, item.high ])); const dates data.map(item item.date); const option { xAxis: { type: category, data: dates }, yAxis: { type: value, scale: true }, series: [{ type: candlestick, data: klineData, itemStyle: { color: #ef232a, // 阳线颜色红涨 color0: #14b143, // 阴线颜色绿跌 borderColor: #ef232a, borderColor0: #14b143 } }] }; chart.setOption(option);成交量和均线可以先不加但如果你想加 MA5/MA10 均线只需要自己计算后作为折线数据叠加function calcMA(data, days) { const result []; for (let i 0; i data.length; i) { if (i days - 1) { result.push(-); continue; } let sum 0; for (let j i - days 1; j i; j) { sum data[j].close; } result.push((sum / days).toFixed(2)); } return result; }自定义接口时注意前端拿到 K 线数据后后端返回的字段名要和前端解析的字段名对齐。我在开发时后端把字段名设成open、close、high、low前端直接用item.open取值但实际跑的时候发现 SQLite 列名是区分大小写的HIGH和high不是一回事。统一规范后这个问题就不会再出现。4.3 自选股管理的交互细节添加自选股我设计了一个简单的搜索框输入代码或名称后从全市场快照表里查返回匹配列表点击确认再写入 watchlist 表。前端表单提交用 fetch请求后刷新左侧列表async function addStock(symbol, name) { const resp await fetch(/api/watchlist?symbol${symbol}name${name}, { method: POST }); if (resp.ok) { await loadWatchlist(); // 刷新自选股列表 } }这里前端只管发请求真正的校验逻辑都在后端。例如重复添加同一只股票数据库的 UNIQUE 约束会抛异常我在后端捕获后返回一个友好的错误信息而不是把堆栈信息直接抛给前端。这个习惯很重要因为你是自己用调试起来无所谓但如果之后分享给别人用接口返回信息不友好会让人一头雾水。5. 排坑实录与部署建议5.1 实际操作中的高频报错与解决办法搭建过程中我整理了遇到过的典型问题这里直接给出一张速查表省得你一个个去翻报错问题现象可能原因解决办法AKShare 请求报data mismatch源站接口字段变动或返回数据量过大升级 AKShare 版本或加time.sleep(1)降低频率sqlite3.ProgrammingError跨线程使用同一个 SQLite 连接每线程单独创建连接或用check_same_threadFalseCannot operate on a closed database连接在后台任务中未正常管理用with sqlite3.connect(...)上下文管理器确保用完即关股票代码变成整数丢失前导零数据结构定义未设为字符串在外部数据源入库前统一str(code).zfill(6)ECharts 为空白但接口返回正常JS 数组格式或字段名不匹配在浏览器 DevTools 里打印数据比对后端返回 JSON 的关键字段端口 8000 被占用uvicorn 非正常退出导致进程常驻用lsof -i:8000查找 PIDkill -9后重启5.2 部署到服务器的几个关键决定项目在本地跑通后部署到服务器上会遇到一些新问题。我最后的方案是既不用 Docker 也不用 Nginx 反向代理就保持最简单的uvicorn app:app --host 0.0.0.0 --port 8000后台运行因为这是个人项目低并发简单的才是最好维护的。如果你需要在服务器上长期跑建议用systemd创建一个服务文件让它在开机自启的同事具备崩溃重启能力。核心配置只需一个.service文件搞定[Unit] DescriptionOpenStock Service Afternetwork.target [Service] WorkingDirectory/path/to/openstock ExecStart/path/to/venv/bin/python -m uvicorn app:app --host 0.0.0.0 --port 8000 Restartalways [Install] WantedBymulti-user.target5.3 合规安全与数据使用提醒最后必须强调一下数据合规的问题。AKShare 拉取的数据来自公开财经网站仅供个人学习研究使用不能用于商业目的。我在代码里加了一行声明表明数据来源和用途限制提示本项目仅用于个人学习与技术实践所有行情数据来自公开接口不构成任何投资建议。不要把抓取频率调得过高那既是被限制的根源也容易给别人带来服务器压力。6. 后续扩展方向让 OpenStock 更有价值跑通基础版本后我给自己列了几个扩展方向。一个是每日收盘后自动生成当天复盘报告统计自选股中涨幅超过 5% 的股票以及主要指数涨跌情况推送到指定的 webhook 上。另一个是接入深度学习模型或简单均线策略在测试集上验证效果虽然个人做量化上限有限但技术路线能打通。还有一个小众但实用的功能是导出数据。把daily_kline表导出成 CSV 或者 Excel方便后续用 Excel 做进一步分析。这个功能我做了两行代码就实现了简单到很多人懒得做但真到用数据写报告时方便得不得了。最后我确实觉得 OpenStock 这类自建工具不要停在一个静态版本上它最爽的使用状态是周一到周五收盘后自动跑完更新打开页面瞄一眼自己关注的股票今天如何不用刷任何商业软件。如果你也是工具控或用数据说话的人这个项目值得花两个晚上折腾搭建过程中你会把数据采集、存储、接口、前端展示这一整条链路完整打通。