你是不是也有过这种经历上午还能跑的行情脚本下午接口就报错了好不容易把全市场数据抓到本地算历史回撤时发现日期字段对不上想找一个免费工具把自选股的K线、均线、成交额排行放到一个页面里翻来翻去总有不满意的地方。OpenStock就是在这种背景下开始折腾的一个开源项目目标很朴素用自己的代码跑通一条完整的行情数据链路——从数据采集、存储、指标计算再到Web可视化展示。这篇文章会按我实际搭建的过程来讲从架构设计到每一段关键代码再到运行中踩过的坑。如果你正在做个人投资分析、准备入门量化或者想找一个Python后端练手项目这套思路可以直接参考。1. OpenStock到底解决什么问题1.1 免费接口的痛点和自建系统的价值用现成行情软件和免费接口的时候最让人头疼的一点是“数据不自由”。行情软件展示的指标是别人定义好的你想把自定义的均线周期、换手率分位数、各行业涨跌统计放在同一个视图里几乎做不到。网页接口呢通常只提供一段时间的日线历史分钟线要么收费要么需要积分字段还经常变动。靠手动导出Excel维护数据时间一长基本都会放弃。自己搭OpenStock这一类系统核心价值在于三点。第一数据是资产的观念采集到的历史行情存放在自己的数据库里格式由自己定义以后做任何分析都不会被上游限制第二数据链路是透明的从抓取、清洗、落库到计算展示每一步都可以追溯出了问题可以直接查日志和数据表第三扩展性是无限逼近自己需求的想加北向资金、龙虎榜、行业板块只需要扩展采集模块不需要迁就别人的产品设计。1.2 整体架构和四个核心模块OpenStock最简版本可以拆成四个模块对应一条完整的数据流数据采集层从开源数据源获取股票列表、日线行情、实时报价数据存储层把采集结果写入数据库支持增量更新业务计算层计算技术指标、价格排名、涨跌分布等Web展示层提供REST API和可交互的前端页面我强烈建议第一次搭建时不要一上来就搞微服务、消息队列、分布式任务调度。以“能跑通全链路”为第一目标最简形态就是单机上的Python进程加SQLite文件加一个FastAPI服务。跑通之后再在某些模块上做优化比如把SQLite换成PostgreSQL、把定时任务拆成独立worker这样心里有底知道每一步优化的动机是什么。1.3 技术选型背后的理由OpenStack选择Python作为主语言可以说没有任何悬念pandas和numpy让行情数据处理非常顺手AKShare/Tushare这类数据接口也是Python生态。Web框架我选了FastAPI而不是Flask或Django主要看中三点原生异步支持让WebSocket推送实时行情非常方便自带Swagger文档调试接口很直观以及基于Pydantic的请求校验让接口健壮性有了保障。存储层面日线数据用SQLite足够应付个人使用的体量。全市场5000多只股票按每天5000行、每年约250个交易日计算一年的日线数据也就120万行左右SQLite完全扛得住。只有当计划存储全市场的分钟级数据时才需要认真考虑PostgreSQL或时序数据库。前端K线图推荐Lightweight Charts样式接近专业行情软件滚动流畅排行榜和统计图则用ECharts配置灵活、图表类型丰富。2. 环境初始化先把项目骨架和数据库建好2.1 项目目录结构与虚拟环境配置一个清晰的目录结构能省掉很多后期维护的麻烦。我的OpenStock项目目录是这样的openstock/ ├── app/ │ ├── api/ # FastAPI路由 │ ├── core/ # 配置项、数据库连接 │ ├── models/ # ORM模型 │ ├── services/ # 业务逻辑 │ ├── collectors/ # 数据采集模块 │ └── indicators/ # 技术指标计算 ├── scripts/ # 初始化脚本、一键更新脚本 ├── data/ # SQLite文件、日志文件 ├── requirements.txt └── docker-compose.yml用Python虚拟环境隔离依赖是做这个项目的第一件正事。Python 3.10以上版本都自带venv模块操作很简单mkdir openstock cd openstock python -m venv .venv source .venv/bin/activate pip install fastapi uvicorn sqlalchemy pandas akshare tushare apscheduler为什么一定要虚拟环境因为AKShare和Tushare这类数据源库更新很频繁依赖的pandas、requests版本经常会变动。如果直接用系统Python一段时间后很可能出现“升级了一个包结果另一个项目跑不起来”的连锁反应。requirements.txt建议把版本号也打上我在实际工作中吃过亏某次AKShare新版本调整了一个接口的参数名结果整个采集任务在凌晨定时执行时静默失败第二天早上看到空数据才发现。锁定版本至少能保证上线环境可控升级时再做针对性的回归测试。2.2 数据库表结构日线数据与股票基础信息数据库是OpenStock的核心资产表设计得好不好直接决定后续分析的效率。最开始只需要两张表股票基础信息表和每日行情表。先看股票基础信息表的设计思路from sqlalchemy import Column, Integer, String, Date, Float, BigInteger, UniqueConstraint from sqlalchemy.orm import declarative_base Base declarative_base() class StockInfo(Base): __tablename__ stock_info id Column(Integer, primary_keyTrue, autoincrementTrue) symbol Column(String(16), uniqueTrue, indexTrue, nullableFalse) name Column(String(32), nullableFalse) exchange Column(String(8), default) list_date Column(Date, nullableTrue)每日行情表是数据量最大的表字段设计上要把常用查询场景想清楚class StockDaily(Base): __tablename__ stock_daily id Column(Integer, primary_keyTrue, autoincrementTrue) symbol Column(String(16), nullableFalse, indexTrue) trade_date Column(Date, nullableFalse, indexTrue) open Column(Float) high Column(Float) low Column(Float) close Column(Float) volume Column(BigInteger) amount Column(Float) __table_args__ ( UniqueConstraint(symbol, trade_date, nameuq_symbol_trade_date), )这里有两个细节值得展开。一是symbol统一用纯数字字符串不带交易所前缀只在展示层根据需要拼接成“sh600000”或“600000.SH”这类格式。二是增加symbol和trade_date的联合唯一索引这个约束能防止采集任务重复执行时写入重复数据配合“插入时先删后插”或“INSERT OR REPLACE”的策略可以保证历史数据始终干净。2.3 数据源选型AKShare与Tushare对比数据源是OpenStock的命脉。免费可用的方案里AKShare和Tushare是最常用的两个选择。简单对比一下对比维度AKShareTushare免费程度完全免费无需Token积分制部分接口需要积分接口稳定性接口和字段变化频繁相对稳定数据字段中文为主直观但需转换英文字段规范统一学习成本低文档示例多中等需要看文档和积分说明适用场景快速原型、个人分析规范化项目、稳定数据需求数据源这块我个人的建议是初版用AKShare因为它零门槛拿到就能跑但要在代码里做一个“数据源适配层”把所有对AKShare的调用封装到collectors模块里不在业务逻辑中散落。这样哪天想切换到Tushare只要改适配层的实现上层API和指标计算完全不受影响。配置文件我用一个简单的config.py管理方便在不同环境里调整参数# config.py DATABASE_URL sqlite:///data/openstock.db DATA_SOURCE akshare REQUEST_INTERVAL 1.0 # 采集请求间隔单位秒 REQUEST_TIMEOUT 15 STOCK_LIST_CACHE_TTL 3600REQUEST_INTERVAL这个参数很关键。AKShare虽然免费但免费的东西往往有隐性频率限制。请求太快容易被远端限制建议抓日线时每只股票之间至少间隔0.5到1秒实时行情推送则不要直接发HTTP请求而是用WebSocket长连接接收数据这会在后面的章节展开。3. 核心模块实现采集、计算与实时推送3.1 行情采集全量历史抓取与增量更新策略OpenStock的采集任务分两个层次全量历史抓取和每日增量更新。第一次跑的时候需要全量抓取之后每天只要做增量就好。全量抓取的第一步是获取股票列表。AKShare提供现成接口返回的是DataFrameimport akshare as ak import pandas as pd def fetch_stock_list(): df ak.stock_info_a_code_name() df df.rename(columns{code: symbol, name: name}) return df[[symbol, name]]抓日线数据时建议用前复权方式获取因为计算均线、MACD这类技术指标时如果遇到分红送股而不复权K线上会出现断崖式的跳空指标会严重失真。def fetch_daily_bars(symbol, start_date, end_date): df ak.stock_zh_a_hist( symbolsymbol, perioddaily, start_datestart_date, end_dateend_date, adjustqfq, ) if df is None or df.empty: return None df[symbol] symbol df[trade_date] pd.to_datetime(df[日期]).dt.date df df.rename(columns{ 开盘: open, 最高: high, 最低: low, 收盘: close, 成交量: volume, 成交额: amount, }) return df[[symbol, trade_date, open, high, low, close, volume, amount]]增量更新的逻辑是每次先查数据库里每只股票的最大交易日期再从这个日期加一天开始抓。A股市场有休市日所以不能简单地“今天减一天”而要看数据源返回了几天数据、最新日期是不是上一个交易日交给数据源判断就行。这里要特别提一个复权基准的坑。前复权价格会随着每次新数据加入而变化举个例子一只股票在2023年6月分红除权你在2023年7月抓到的2020年历史价格和2024年1月抓到的同一历史日期价格数值可能不一样。如果只是用来展示K线问题不大但如果用这些数据做策略回测历史价格随时变化会导致回测结果无法复现。更稳的做法是存储原始不复权价格再把复权因子也存下来计算时自行决定用哪种复权。这样数据是稳定的公式是透明的。3.2 技术指标计算MA、MACD的实现细节有了干净的日线数据指标计算就是顺理成章的事。我习惯把指标计算做成独立的服务函数输入DataFrame输出带指标列的DataFrame方便随时全量重算。均线是最基础的指标pandas的rolling一行就能算def add_ma(df, windows(5, 10, 20, 60)): for w in windows: df[fma{w}] df[close].rolling(windoww).mean() return dfMACD的实现稍微复杂一点但理解了原理代码也不长。MACD由三部分组成DIF是快线EMA12与慢线EMA26的差DEA是DIF的9日EMAMACD柱状图等于DIF与DEA差值的两倍def add_macd(df, fast12, slow26, signal9): df[ema_fast] df[close].ewm(spanfast, adjustFalse).mean() df[ema_slow] df[close].ewm(spanslow, adjustFalse).mean() df[dif] df[ema_fast] - df[ema_slow] df[dea] df[dif].ewm(spansignal, adjustFalse).mean() df[macd] (df[dif] - df[dea]) * 2 return df为什么不直接用现成的技术指标库库用起来虽然省事但遇到数据缺失和停牌空值时库内部的处理方式未必符合你的预期。手写rolling和ewm能让你对每一步的输入输出都有掌控。等后续确实需要布林带、KDJ、RSI时再引入talib或pandas-ta也不迟。指标计算环节还有一个容易忽略的点计算指标前要把数据按trade_date升序排列并且要处理停牌缺口。A股停牌期间数据源可能直接不返回记录直接用rolling计算会导致停牌前后被当作相邻交易日MA和MACD的值都会偏差。严格的做法是先补全交易日历把缺失交易日记成NaN再计算指标最后展示时再决定是隐藏NaN还是画断线。3.3 实时行情推送从轮询到WebSocket长连接OpenStock的实时行情展示最初我用的方案是前端每3秒调用一次后端HTTP接口后端每3秒主动去数据源抓一次最新价格。这个方案实现最简单但有两个问题前端轮询压力大时后端响应变慢而且每次都重新抓数据源会触发限流。后来改成WebSocket推送体验好了很多。FastAPI的WebSocket支持非常简洁from fastapi import FastAPI, WebSocket app FastAPI() app.websocket(/ws/quote) async def websocket_endpoint(websocket: WebSocket): await websocket.accept() while True: quote await latest_quote_queue.get() await websocket.send_json(quote)关键在于后端维护一个asyncio.Queue后台任务定时从数据源获取整个自选股列表的最新行情然后推送到队列里所有已连接的WebSocket客户端都从这个队列取数据。这样的设计有几个好处不重复请求数据源节省接口配额推送的内容对所有客户端一致实现简单代码可读性高。有个细节要注意WebSocket连接需要处理心跳。有些反向代理或网关会在一段时间没有数据传输后主动断开连接所以每隔15秒左右主动发一个ping或者一个象征性的数据帧能有效避免断线。另外断线重连的前端逻辑也别忘了浏览器端在onclose事件里重新调用connect并且加一个指数退避避免刷新风暴。4. Web展示层让数据变成能看的K线与榜单4.1 API接口设计与统一返回结构OpenStock的后端API建议设计成扁平、单一职责的风格前端调用容易理解接口方法说明/api/stocksGET股票搜索与列表支持 keyword 参数/api/stocks/{symbol}/klineGET返回指定股票的K线数据支持 period、limit/api/stocks/{symbol}/realtimeGET返回单只股票的最新行情/api/rankGET按成交额、涨跌幅等维度排名/ws/quoteWebSocket实时行情推送响应结构统一用这样的格式{ code: 0, message: ok, data: { symbol: 600000, name: 浦发银行, kline: [] } }这么做的好处是前端可以统一处理错误和加载状态后端错误信息也能通过message字段传得比较清楚。4.2 K线图组件选型与前端渲染K线图是行情系统最核心的展示组件。我对比过Lightweight Charts和ECharts两者各有优势对比维度Lightweight ChartsECharts设计目标专门做金融图表通用数据可视化性能与流畅度轻量大量K线下滚动流畅数据量大时稍有卡顿API复杂度简单上手快配置项多学习成本高扩展能力K线相关为主图表类型丰富推荐场景主K线图排行榜、行业分布等统计图所以我最后采用的是组合方案主K线图用Lightweight Charts页面上的成交额排行榜、涨跌分布直方图用ECharts。两者的CDN引入方式都很简单用ESModule的方式引入import { createChart } from lightweight-charts; const chart createChart(document.getElementById(kline), { width: 800, height: 450, }); const series chart.addCandlestickSeries(); series.setData(klineData);前端页面不需要做得很复杂第一版能展示行情列表、K线详情、排行榜三块就够了。更多精力建议放在数据质量和后端稳定性上这也是我踩过坑之后最大的体会——页面好看救不了底层数据混乱。4.3 定时任务与数据刷新时间窗行情数据更新有自己的节奏盘中要高频盘后要完整。OpenStock用APScheduler来管理定时任务。from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger scheduler BackgroundScheduler() def start_scheduler(): scheduler.add_job( run_daily_update, CronTrigger(timezoneAsia/Shanghai, day_of_weekmon-fri, hour16, minute5), iddaily_update, max_instances1, replace_existingTrue, ) scheduler.start()收盘后为什么要等到16:05再跑A股15:00收盘行情数据源不会在收盘那一刻马上把所有数据整理完毕通常需要半小时左右。如果设置15:05跑大概率抓不到完整数据或者抓到的是未完成结算的临时数据。多等一分钟换个数据完整度非常划算。盘中如果需要实时刷新可以再加一个小时级的任务每个交易日10:30、13:30各更新一次当日分钟线并入当日汇总。但这部分会让数据量快速增长建议在OpenStock运行稳定后再加。5. OpenStock实测踩坑这些坑我替你先踩了5.1 数据源接口频繁变动重试与兜底缺一不可AKShare接口变动频繁这件事用过的朋友都知道。接口名、参数、返回字段都可能突然变更而且常常是周六周日改完周一开盘才发现。所以我写了统一的调用包装函数所有对数据源的访问都通过这个函数走import time import logging logger logging.getLogger(openstock.collector) def safe_call(func, *args, **kwargs): for attempt in range(4): try: return func(*args, **kwargs) except Exception as e: logger.warning(第%s次调用失败: %s, attempt 1, e) time.sleep(2 attempt * 3) raise RuntimeError(f数据源调用最终失败: {func.__name__})这里用了线性退避而不是固定等待。第一次失败等2秒第二次5秒第三次8秒给对方服务一点恢复时间也避免自己撞上限流。还有一点很重要每次采集任务结束后把成功和失败的股票数写入日志。如果某次任务失败率异常高说明上游接口可能变了这时候最好触发一个告警通知而不是任由任务“静默失败”。5.2 股票代码格式与时区错乱A股有很多种代码格式AKShare返回的上海股票是600000纯数字不带前缀有些数据源返回的是sh600000带交易所前缀还有的用600000.SH。如果OpenStock采集模块存了一种格式后来接入另一个数据源又存了另一种格式前端关联时就会出问题。我的建议是统一规定数据库里只存纯数字字符串代码里用exchange字段区分交易所展示层再拼接前缀。时区问题则是另一个容易被忽视的坑。如果服务器时区设置为UTCPython的datetime.now()取出来的是UTC时间在和交易日的日期做比较时可能差8小时导致增量更新的日期范围判断出错某些交易日的行情被漏掉。OpenStock全项目统一用北京时间from datetime import datetime import pytz CN_TZ pytz.timezone(Asia/Shanghai) def now_cn() - datetime: return datetime.now(CN_TZ)所有涉及“今天”“最近交易日”的判断都走now_cn()数据库里存储的trade_date用date类型而不是datetime天然避免时区混淆。5.3 停牌缺失与分钟级数据膨胀A股股票停牌期间日线接口通常不会返回记录。这带来两个问题一是K线图上会断开二是计算指标时停牌前后被连成相邻数据。前者是展示问题前端可以用前向填充的方式补全显示后者是数据质量问题更严谨的做法是维护一份交易日历表用交易日做左连接缺失的日子填NaN再算指标。分钟级数据要特别注意膨胀问题。全市场5000多只股票每分钟一条数据一天交易240分钟理论上每天会生成约120万行记录一年接近3亿行。这还没算上高频的快照数据SQLite根本处理不了。我的方案是分层次保留最近5个交易日的分钟线完整保留用于短线复盘更早的分钟线只保留聚合后的15分钟、60分钟K线再往前的自动归档或删除。这个策略可以在“数据可用性”和“存储成本”之间找到平衡点动手采集分钟数据之前一定要先把这部分设计好不然跑一个月就会面临数据库膨胀问题。6. 进阶从本地跑通到长期稳定运行6.1 Docker Compose一键部署OpenStock本地跑通之后下一步是部署到一台常年开机的服务器上让定时任务稳定运行。Docker Compose是比较轻量的部署方案一个文件搞定服务编排version: 3.8 services: db: image: postgres:15-alpine environment: POSTGRES_USER: openstock POSTGRES_PASSWORD: openstock POSTGRES_DB: openstock volumes: - pgdata:/var/lib/postgresql/data healthcheck: test: [CMD-SHELL, pg_isready -U openstock] interval: 10s timeout: 5s retries: 5 web: build: . ports: - 8000:8000 depends_on: db: condition: service_healthy environment: DATABASE_URL: postgresql://openstock:openstockdb:5432/openstock volumes: pgdata:容器化部署的好处是环境依赖全部固化换机器部署时只要装了Docker基本就是一条docker compose up -d的事。从SQLite迁移到PostgreSQL时SQLAlchemy层几乎不用改代码只改一个DATABASE_URL即可。要注意的是pandas的read_sql和DataFrame的to_sql在两者之间有一些日期和浮点精度的差异迁移后要跑一遍完整的增量任务做验证。6.2 日志、健康检查与失败告警长期稳定运行的关键不是“写更多代码”而是“能及时发现问题”。OpenStock里我加了三层保障第一层是结构化日志。用loguru替换标准logging输出格式带上时间、模块、任务ID排查问题的时候能快速定位是哪只股票、哪次任务、哪个环节出了问题。第二层是健康检查接口app.get(/healthz) def healthz(): db_ok check_database() last_update get_last_update_time() return { status: ok if db_ok else error, last_update: last_update, stale_minutes: minutes_since(last_update), }这个接口不仅能被Docker的healthcheck用来判断容器状态还能接外部监控平台做周期性探测。只要数据超过某个阈值没有更新就说明定时任务可能挂了。第三层是失败告警。采集任务失败后除了记日志还可以通过简单的方式发一个通知。在Python里调用Webhook发送告警非常简单几十行代码就能接上钉钉、企微或者普通邮件。这样即使凌晨定时任务失败第二天早上也能在手机上看到消息而不是等打开页面才发现数据是空的。6.3 后续扩展思路回测、因子与多市场OpenStock的基础版本稳定运行之后可以沿着几个方向继续扩展。回测方向是最自然的一步。日线数据和技术指标已经齐全接上backtrader或vectorbt这类回测框架就能验证交易策略。数据格式要做一层适配把OpenStock的字段映射成回测框架的OHLCV格式基本就能跑起来。因子选股方向也很有意思。在基础行情数据之上增加财务数据、行业板块、北向持仓等字段就可以计算估值因子、动量因子、质量因子每天对全市场股票打分排序。这个方向的难点不在计算而在于数据源的质量和历史覆盖度。多市场覆盖方面港股和美股在AKShare里也有对应的免费接口。多市场数据的关键问题是交易日历不同、汇率换算、代码格式差异更大。如果打算覆盖多市场建议在数据模型上提前加一个market字段隔离各市场的配置和数据表不要混在同一个表里。写在最后我自己在实际搭建OpenStock的过程中最大的体会是真正有价值的不是跑通的那一刻而是后续不断修修补补、让系统稳定运行的过程。数据源接口会变、服务器会宕机、数据库会长大所有这些都是在真实项目中才会遇到的问题也是最好的学习机会。最后再分享一个小技巧把数据采集和指标计算彻底解耦采集只负责把原始数据写进库指标计算随时可以从库里重新读取重算。这样即使你改了指标公式也不需要回补历史行情数据一条命令就能重算所有结果。这套架构风格同样适用于很多数据类项目希望这篇记录能帮你少踩几个坑。
