如果你每天的工作流还是“打开几个网页手动看行情、复制粘贴到表格里做复盘、想验证一个思路只能靠猜”那这个项目确实能帮你省下不少时间。OpenStock 是我最近从零开始搭建的一个开源股票数据采集与分析工具核心目标很朴素把行情数据自动抓下来、清洗后存进数据库、批量计算技术指标最后用一套 Web 页面统一展示和筛选。整套东西不依赖昂贵的商业软件一台普通 Linux 机器就能跑数据规模在几千只标的、多年日线的量级内单机资源完全够用。这篇文章不是项目文档的复述而是我从踩坑到跑通全过程的还原。如果你对数据工程、自动化和量化分析感兴趣或者只是受够了反复手动整理数据可以直接按文章里的方案搭一套。我会把数据怎么拿、表怎么建、指标怎么算、服务怎么部署这些关键环节都拆开讲明白还会把实际运行中遇到的典型问题和排查思路一并整理出来。1. 整体设计与思路拆解1.1 为什么非要自己搭一套市面上现成的行情软件和网站并不少但它们解决的问题和我的需求不太匹配。我需要的是一个能“批量处理、持续积累、可自己扩展”的数据底座而不是看盘界面。举个例子我想验证某种技术指标的筛选效果需要把全市场几千只标的过去几年的数据全部拉下来再统一按同一套规则计算。这种需求在普通软件里基本没法低成本实现自己搭一套反而更直接。另外数据只有沉淀在自己手里才有长期价值。行情数据是典型的时间序列数据每天增量更新一点点长期积累下来就是一份宝贵的本地数据集。用自己的库想加字段、改算法、做回测、做统计都完全没有限制。这个过程本身也是很好的工程实践能锻炼数据采集、存储、计算、Web 展示这一整套链条。1.2 技术选型为什么是这一套组合最终选定的技术栈是 Python pandas SQLite FastAPI ECharts Docker。选型时我也有过犹豫比如数据库要不要上 PostgreSQL、前端要不要用重型框架但实际跑下来这组方案在个人项目里足够顺手。Python生态最全处理数据的 pandas 几乎没法绕开。SQLite零配置、单文件几千只标的几年的日线数据存下来也就几百 MBSQLite 完全扛得住。先不引入 PostgreSQL可以省掉非常多运维成本。FastAPI用来提供查询接口性能足够代码量少。ECharts前端画 K 线、走势图非常方便。Docker部署时把环境固化换机器不折腾。后端语言和前端框架都有替代品但 Python SQLite 这个组合在个人项目里性价比确实最高。如果后续数据量真的大到 SQLite 不够用迁移到 PostgreSQL 也不困难因为表结构设计从一开始就是按关系型数据库规范来的。1.3 模块划分与数据流整个系统的数据流非常清晰采集模块从数据源拉取原始行情经过清洗和去重后写入数据库分析模块从库里读取数据计算指标并回写结果API 层提供查询服务前端拉取 API 数据做可视化展示。另外有一个调度器负责每天定时触发增量采集任务。拆成独立模块的好处是每个环节都能单独调试。采集挂了不会影响查询指标算法改版了只需要重算结果表前端变了也不用动数据库。这种分层思路虽然简单但很实用。2. 行情数据采集与存储2.1 行情数据从哪里拿数据源是做这类项目第一个要解决的问题。市面上有几个公开的行情接口可以免费调用数据源不必只用一家可以主力用一家、再用另一家做交叉校验。开源社区里也有封装好的库比如 AKShare接口丰富省去了自己解析各种返回格式的麻烦。如果你的网络环境里访问外部接口不稳定也可以在项目里保留一个“本地模拟数据”模式用随机数生成一批具有基本 OHLC 结构的示例数据先把整个流程跑通。等外部数据源可用后再切换过去。我在开发初期就是这么干的先把管道搭好后面换真实数据只是改一行配置的事。2.2 数据库表结构设计数据库表是整套系统的心脏。我的表设计并不复杂核心就是一张行情主表加一张指标结果表。这里最关键的设计决策是行情数据按“标的代码 日期”做唯一约束这样重复跑任务也不会产生脏数据。建表语句示例SQLiteCREATE TABLE IF NOT EXISTS daily_bar ( symbol TEXT NOT NULL, trade_date TEXT NOT NULL, -- 统一用 YYYY-MM-DD open REAL NOT NULL, high REAL NOT NULL, low REAL NOT NULL, close REAL NOT NULL, volume INTEGER NOT NULL, amount REAL, source TEXT, PRIMARY KEY (symbol, trade_date) ); CREATE TABLE IF NOT EXISTS indicator_cache ( symbol TEXT NOT NULL, trade_date TEXT NOT NULL, ma5 REAL, ma10 REAL, ma20 REAL, macd_dif REAL, macd_dea REAL, macd_hist REAL, rsi14 REAL, PRIMARY KEY (symbol, trade_date) ); CREATE INDEX IF NOT EXISTS idx_daily_date ON daily_bar (trade_date);表结构看似简单但我实际踩过不少坑。比如日期字段统一用YYYY-MM-DD文本格式而不是时间戳。原因很简单日线级别的时间精度不需要到时分秒文本格式反而便于阅读、比较和索引。再比如主键用symbol trade_date天然规避了重复插入的问题配合INSERT OR REPLACE就能实现幂等写入。2.3 采集脚本的幂等与重试增量采集的通用逻辑是查询本地已有数据的最大日期然后只拉取这个日期之后的数据。这个逻辑本身不复杂但真正执行时要考虑网络波动、接口限流、数据源返回空值等各种意外。我的做法是加一个简单的重试机制单只标的失败最多重试三次连续失败超过一定数量就暂停任务并记录日志。持仓过程中我发现数据采集不是一个“跑一次就结束”的任务而是一个需要每天稳定运行的小型 ETL 系统。因此任务必须可重跑、可中断、可恢复。基于这个思路每条原始数据都记录source字段出问题时能溯源写入时使用INSERT OR REPLACE重跑不会产生重复记录每个标的独立抓取一只失败不影响其他标的入库。一个简单的采集函数示例import time import random import requests import pandas as pd def fetch_daily_bar(symbol: str, start_date: str, end_date: str) - pd.DataFrame: 拉取某标的在指定日期区间的日线数据。 这里以公开 HTTP 接口为例具体参数以实际数据源为准。 如果接口不可用会进入异常处理逻辑。 params { symbol: symbol, start: start_date, end: end_date, adjust: qfq, # 前复权 } for attempt in range(3): try: resp requests.get(https://example.com/api/daily, paramsparams, timeout10) resp.raise_for_status() data resp.json() df pd.DataFrame(data) return df except Exception as e: print(f[{symbol}] 第 {attempt 1} 次请求失败: {e}) time.sleep(1 random.random()) return pd.DataFrame()注意这里的 URL 只是演示占位实际使用请替换成你在用的数据源地址。数据源字段经常变化建议单独封装一层数据适配器不要在上层业务代码里直接拼 HTTP 请求。3. 指标计算与策略筛选逻辑3.1 常见技术指标的计算方式数据入库只是第一步OpenStock 真正有意义的功能是把原始行情转换成可理解的指标。我实现了 MA、MACD、RSI 这几个最常用的指标。它们的算法并不神秘只是很多人在 Excel 里手算太麻烦MA均线过去 N 个交易日收盘价的平均值。MACD先算 EMA12 和 EMA26两者的差是 DIF 线DIF 再算 EMA9 得到 DEA 线最后用 DIF 减去 DEA 得到柱状图值。RSI基于一定周期内涨跌幅平均值计算相对强弱指数。以 MACD 为例很多人以为它很复杂拆开之后就是把 EMA 算好几遍。用 pandas 实现非常简洁import pandas as pd def add_ma(df: pd.DataFrame, windows(5, 10, 20)) - pd.DataFrame: for w in windows: df[fma{w}] df[close].rolling(windoww).mean() return df def add_macd(df: pd.DataFrame, fast12, slow26, signal9) - pd.DataFrame: ema_fast df[close].ewm(spanfast, adjustFalse).mean() ema_slow df[close].ewm(spanslow, adjustFalse).mean() df[macd_dif] ema_fast - ema_slow df[macd_dea] df[macd_dif].ewm(spansignal, adjustFalse).mean() df[macd_hist] df[macd_dif] - df[macd_dea] return df def add_rsi(df: pd.DataFrame, period14) - pd.DataFrame: delta df[close].diff() gain delta.clip(lower0) loss -delta.clip(upper0) avg_gain gain.ewm(alpha1 / period, adjustFalse).mean() avg_loss loss.ewm(alpha1 / period, adjustFalse).mean() rs avg_gain / avg_loss.replace(0, float(nan)) df[rsi14] 100 - (100 / (1 rs)) return df这里容易踩坑的地方是ewm的adjust参数。如果不设置adjustFalse计算结果会和很多行情软件里的值对不上。因为我习惯和主流行情软件对比验证所以统一用adjustFalse这种方式。RSI 公式里avg_loss可能为 0要提前用replace(0, nan)避免除零错误。3.2 批量计算的使用方式单只标的一分钟能跑完几千只批量计算时IO 和内存就要控制好。我采用按标的逐只处理的方式每只标的从daily_bar读取数据算完指标写入indicator_cache。这样每一轮的 DataFrame 都不大内存峰值很低。真正耗时间的不是计算而是数据库写入所以写入时用批量事务而不是逐条 commit。为了提高计算效率可以把所有指标计算函数封装成一个 pipelinedef compute_all(symbol: str) - pd.DataFrame: df load_from_db(symbol) df add_ma(df) df add_macd(df) df add_rsi(df) return df3.3 什么叫做“技术面筛选”有了指标结果表筛选就变成简单的 SQL 查询。比如我想找出“20 日均线向上且 MACD 指标第一天翻红”的标的一条 SQL 就能搞定。结合前端表格我能在十几秒内扫一遍全市场。这种能力并不高深但自己动手做过之后你对数据的敏感度和对指标的理解会完全不一样。如果只看现成软件的指标你只能被动接受它给你的结果自己算了之后你会知道每一个值背后经历了什么运算也更容易发现哪些指标有效、哪些指标只是心理安慰。4. Web 展示与自动化部署4.1 API 层设计我选择 FastAPI 作为 API 层接口设计得很简单一个接口返回标的最新指标数据一个接口返回指定的历史行情序列一个接口返回筛选结果。查询逻辑直接走 SQL配合索引单次请求基本在几十毫秒级别。一个典型接口示例from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware import sqlite3 import pandas as pd app FastAPI() app.add_middleware( CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*], ) DB_PATH /data/stock.db app.get(/api/screener) def screen(min_close: float 0, limit: int 50): conn sqlite3.connect(DB_PATH) sql SELECT i.symbol, d.close, i.ma5, i.ma20, i.macd_dif, i.macd_dea FROM indicator_cache i JOIN daily_bar d ON i.symbol d.symbol AND i.trade_date d.trade_date WHERE i.trade_date (SELECT MAX(trade_date) FROM daily_bar) AND d.close ? LIMIT ? rows conn.execute(sql, (min_close, limit)).fetchall() conn.close() return {items: rows}开发阶段可以完全不用考虑认证因为只是内网或本地访问。但如果部署到公网至少加一层简单的 Token 校验否则你的数据接口就是公开的。这个一定要记得。4.2 前端仪表盘前端我用的是 ECharts 加一个简单的 HTML 页面没有上 Node 构建工具。原因很简单项目核心是数据和计算不是前端工程化。一套静态页面直接由 FastAPI 托管零构建成本。页面上主要两块内容左边是筛选出来的标的列表点击任意标的右侧就会展示 K 线图和均线图。K 线图对于判断形态很有帮助配合 MA 和成交量基本能看清行情走势。如果觉得默认配色不好看ECharts 的visualMap和颜色配置都可以慢慢调。4.3 Docker 部署与定时任务部署上我选择了 Docker Compose包含两个服务一个是 API 服务一个是定时任务服务。实际跑的时候发现定时任务不要放在 API 进程里否则会阻塞请求线程。最稳妥的方式是用独立的 cron 容器或者直接在宿主机上写 crontab。docker-compose.yml 核心内容version: 3.8 services: api: build: . ports: - 8000:8000 volumes: - ./data:/data environment: - DB_PATH/data/stock.db restart: unless-stopped scheduler: build: . command: python -m scripts.run_scheduler volumes: - ./data:/data environment: - DB_PATH/data/stock.db restart: unless-stopped数据目录通过 volume 挂载出来这样备份和迁移都方便。定时任务我用的是一个非常轻量的 Python 循环加time.sleep而不是 Cron 表达式因为这样在 Docker 里更容易控制。4.4 数据质量校验自动化跑久了最怕的是数据悄悄变脏。我加了一个每日数据校验任务检查三件事每天的记录数是否在合理范围内、最新交易日期是否正常推进、是否存在明显异常值比如负成交量、最高价低于最低价。任何一条校验不通过就标记异常并推送通知而不是等到分析时才发现数据有问题。这步看起来很常规但实际价值很大。有一次数据源调整了字段命名采集任务静默返回空值如果不是校验任务报警我可能到做分析时才发现数据断了一周。5. 常见问题与排查技巧实录5.1 数据源接口限流与字段变化最典型的故障是请求频率过高被封。第一次跑全量采集时我单线程循环请求跑了一千多个标的就开始报错。解决办法是加限速、加随机延时、分段执行全量数据不要一次性拉完而是分批次在非交易时段执行。数据源字段变化也是很隐蔽的问题。某些接口偶尔会把volume从整数变成字符串或者单位发生变化。我的解决思路是采集之后强制做一次类型校验和范围校验不满足要求的记录直接丢弃并计数而不是强行入库。5.2 复权问题做历史行情分析时前复权和后复权必须想清楚。我的策略是全部统一使用前复权数据因为这是主流行情软件的默认展示方式对比起来最不容易出错。但这带来的副作用是随着每次除权除息历史价格会被重算所以指标结果缓存必须定期更新。我的做法是每周重新计算一遍全量指标保证缓存不陈旧。5.3 时区问题日期边界问题一开始被忽略了后来发现某些标的的数据会提前在凌晨刷新导致“今天”的数据记录到了“昨天”。我的处理方式是以数据源返回的交易日期为准而不是本地时间。同时定时任务安排在收盘后一小时再跑给数据源留出足够的稳定时间。5.4 指标结果和行情软件对不上如果你用同一套数据在别处对比发现 MACD 数值有差异先检查三件事数据源是否一致、复权方式是否一致、EMA 是否使用了adjustFalse。绝大多数不一致都是这三类原因。不用追求和任何一家软件完全一致但至少要保证自己内部逻辑稳定。5.5 常见问题速查表针对实际运行中遇到的典型问题我整理了一张速查表遇到问题时可以先对照排查基本能覆盖大部分故障场景。问题现象可能原因排查方法解决建议某只标的数据持续为空代码变动或请求参数错误手动请求一次接口检查返回结构更新适配器或修正代码某天数据量骤减限流或网络抖动查看采集日志中的失败列表增加重试与暂停时间指标值全为 NULL数据不足或除零检查标的上市日期和周期长度过滤上市时间过短的标的MACD 与行情软件不一致EMA 算法或复权不同核对参数统一使用 adjustFalse 和前复权采集任务偶发重复无幂等约束检查主键设计使用 INSERT OR REPLACEAPI 查询越来越慢缺少索引或全表扫描查看 explain按日期和代码建立复合索引Docker 容器内存过高单次读取大量数据查看日志中的 DataFrame 规模按标的逐只分批处理重启后数据丢失容器内路径未挂载检查 volume 配置将数据目录挂载到宿主机这套系统跑下来我最大的感受是百分之八十的时间都在和数据质量较劲真正写业务逻辑的时间反而没那么多。但也正是这些折腾的过程让我对数据从生成到落库再到展示的每一个环节都有清楚的认识这也是我觉得这个项目最值得回味的价值。如果你的需求也是“批量拉历史数据、算指标、做筛选”OpenStock 这种以数据为中心的架构思路可以直接复用。后续我还在考虑扩展两个方向一是把日线频率降到分钟级挖掘更多日内特征二是接入简单的规则回测模块让历史信号可以自动验证。先把数据底座打好后面很多想法自然就能长出来。