要说最近在金融数据这个圈子里有什么值得自己动手玩一玩的开源项目OpenStock绝对算一个。简单来说OpenStock是一套开源的股票行情数据采集、存储与展示系统它把A股行情源、数据库、API服务和前端展示整个链路的代码全部开放出来任何人都能在自己的服务器上把它跑起来。这个项目解决的最大痛点是个人投资者和量化爱好者经常面临的数据“三难”问题免费数据源不稳定、历史数据不完整、想二次加工又要自己造轮子。我花了三个晚上把它完整搭了一遍过程中踩了不少坑也把整套流程摸透了。这篇文章就是一份手把手的搭建笔记从架构设计、环境准备、核心模块实现到启动验证和问题排查全部讲清楚。如果你手里有一台能跑Docker的Linux服务器或者哪怕只是一台本地电脑跟着这篇文章走完就能拥有一个完全属于自己的股票行情数据系统。1. 项目整体设计与数据链路拆解很多人拿到OpenStock第一反应是“直接跑起来再说”但我的建议是先花十分钟把它的整体设计看明白。因为这套系统的架构思路非常典型理解了它后面遇到问题你自己就能定位而不是到处问人。1.1 OpenStock到底解决什么问题个人做量化或者做数据复盘最先遇到的就是数据获取的问题。传统做法是手动从行情软件里导出Excel然后再写一堆脚本去清洗整个过程极其痛苦。OpenStock的核心目标就是把“行情数据从采集到入库再到对外提供服务”这条链路自动化让你每天一睁眼昨天全市场几千只股票的日线数据已经整整齐齐躺在数据库里了。它的数据范围覆盖A股主板、创业板、科创板也包含指数、ETF等常见品种。字段方面包括开高低收、成交量、成交额、换手率这些基础行情数据部分接口还支持前复权价格计算。这对于做回测、算因子、画K线图来说已经非常够用了。1.2 架构选型为什么是Python MySQL FastAPI我看了一下OpenStock的源码它的技术栈选得非常务实。采集端用Python这是整个量化生态里最成熟的数据源接口封装多社区案例丰富。存储端用的是MySQL虽然很多人觉得上ClickHouse或者TimescaleDB更“高级”但个人项目最怕的是复杂度不可控。MySQL装起来简单运维成本低几千万行的日线数据配合索引也完全跑得动。对外服务用了FastAPI这也是现在Python社区里很流行的异步框架。我实测它的响应速度比Flask快不少而且自带Swagger文档调试接口的时候省了很多事。整个项目没有把前后端混在一起而是把数据采集、数据存储、API服务三层拆开每一层可以独立部署和扩展。这种设计虽然初期稍微多花点时间但后续想加功能或者换数据源都是动一个模块的事不会牵一发动全身。1.3 数据链路的完整走向我画了一张流程图用文字描述数据源接口 → 采集脚本 → 数据清洗 → MySQL数据库 → FastAPI接口 → 前端展示/量化策略调用。核心数据的流向是一个单向管道。采集脚本定时去数据源拉取行情拿到的是JSON或者DataFrame格式的数据。这里有个关键的细节行情源返回的数据不是拿来就能入库的比如复权因子、停牌状态、涨跌幅异常这些都需要先处理。OpenStock在采集和入库之间加了一个清洗层专门处理缺失值、去重和类型转换。数据落库之后所有读取都走FastAPI接口前端页面和量化策略都不直接连数据库这样既安全又统一。理解了这条链路你再看后面的部署步骤就不会晕了。所有操作其实都是往这条管道的不同节点上灌东西。2. 环境准备与基础组件部署搭建OpenStock的环境部分我按照“最省心”的原则说。不要一上来就搞什么K8s集群先用最简单的方式跑通之后再考虑扩展。2.1 Python虚拟环境与版本选择OpenStock要求Python 3.10以上我建议直接装3.10或者3.11太新的3.12在某些依赖库上可能还没适配完全。我的服务器是Ubuntu 22.04系统自带的Python是3.10刚好满足。sudo apt update sudo apt install -y python3.10 python3.10-venv python3-pip这里我强烈建议用虚拟环境不要图省事直接pip install到系统环境里。不同的Python项目依赖经常冲突而且系统升级的时候容易把环境搞坏。mkdir -p /opt/openstock cd /opt/openstock python3 -m venv venv source venv/bin/activate进入虚拟环境之后你会看到命令行前面多了一个(venv)前缀这就对了。后面所有Python相关的操作都在这个环境里执行包括pip安装依赖和启动脚本。2.2 MySQL 8.0安装与连接配置MySQL建议装8.0版本因为OpenStock建表语句里用到了一些8.0才支持的语法特性。如果你的机器上还没有安装执行sudo apt install -y mysql-server sudo systemctl start mysql sudo systemctl enable mysql安装完成后创建数据库和专用账号。我强烈建议不要用root连接应用安全是一方面更重要的是后续排查问题时权限边界清晰你要知道问题出在应用还是数据库配置上。mysql -uroot -p进入MySQL命令行后执行CREATE DATABASE openstock DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; CREATE USER openstocklocalhost IDENTIFIED BY your_strong_password; GRANT ALL PRIVILEGES ON openstock.* TO openstocklocalhost; FLUSH PRIVILEGES;这里有个关键点数据库字符集一定要用utf8mb4而不是默认的latin1。因为行情数据里可能包含股票名称等中文内容utf8mb4才能完整存储。我一开始用默认字符集建的库结果写入中文股票名的时候直接报错这个坑大家尽量别踩。2.3 目录结构与配置文件规划OpenStock的目录结构经过我梳理后如下并不复杂/opt/openstock/ ├── openstock/ │ ├── collector/ # 数据采集模块 │ ├── storage/ # 数据持久化模块 │ ├── api/ # FastAPI接口服务 │ ├── models/ # 数据模型定义 │ └── utils/ # 公共工具函数 ├── config/ │ └── config.yaml # 全局配置 ├── scripts/ │ └── init_db.sh # 数据库初始化脚本 └── requirements.txt # Python依赖清单配置文件是整个系统的中枢我把关键内容贴出来你根据自己环境改database: host: localhost port: 3306 user: openstock password: your_strong_password name: openstock collector: source: akshare symbol_file: config/symbols.txt interval: daily retry_times: 3 request_timeout: 10 api: host: 0.0.0.0 port: 8000 debug: false注意symbol_file这个配置它是股票代码列表文件OpenStock就是依次遍历这个列表里的代码去拉数据的。你可以通过修改它来控制采集范围不必全市场都拉。依赖安装一步到位pip install pandas numpy pymysql sqlalchemy fastapi uvicorn akshare matplotlib tqdm pyyaml3. 核心功能模块的实现环境准备好了接下来说OpenStock核心模块的实现。我按照数据从源头到服务的顺序把每一个环节的关键代码和设计思路拆开讲。3.1 数据采集层把行情拉下来采集层是整个系统最底层的输入它做的事情就是遍历股票代码列表调用数据源接口获取每一个股票的历史行情数据。OpenStock默认使用的数据源是AKShare这是一个开源财经数据接口库免费且不需要token覆盖面广对个人开发者很友好。核心采集逻辑如下# openstock/collector/collector.py import akshare as ak import pandas as pd from time import sleep def fetch_daily_history(symbol: str, start_date: str, end_date: str) - pd.DataFrame: 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 df.rename(columns{ 日期: trade_date, 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume, 成交额: amount, 换手率: turnover }) df[symbol] symbol df[trade_date] pd.to_datetime(df[trade_date]).dt.strftime(%Y-%m-%d) return df[[symbol, trade_date, open, high, low, close, volume, amount, turnover]]重点解释几个选择第一个是adjustqfq这个参数代表前复权。股票发生过送转股、分红之后历史价格会出现跳空如果不复权直接用原始价格做回测或画图趋势线会失真。前复权是把历史价格调整到当前的价格基准保持价格的连续性。如果你做的是分红再投入策略也可以取后复权数据看实际情况。第二个是字段改名。AKShare返回的字段是中文的直接入库没问题但后续写SQL和接口的时候中文字段名会遇到各种编码和兼容问题。所以统一改成英文字段这是从业者的习惯做法。第三个是请求节奏控制。我写了一个重试和限速机制每个请求之间间隔0.3秒避免触发数据源的反爬限制。实测全市场5000多只股票大约需要25到30分钟能拉完这个速度可以接受。3.2 数据存储层表结构设计与入库采集层拿到的是规整的DataFrame下一步就是入库。OpenStock的数据表结构设计得比较精简主要就是一张日线行情表。CREATE TABLE IF NOT EXISTS daily_quote ( id BIGINT AUTO_INCREMENT PRIMARY KEY, symbol VARCHAR(20) NOT NULL, trade_date DATE NOT NULL, open DECIMAL(10, 2), high DECIMAL(10, 2), low DECIMAL(10, 2), close DECIMAL(10, 2), volume BIGINT, amount DECIMAL(16, 2), turnover DECIMAL(10, 4), updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_symbol_date (symbol, trade_date) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;这里最关键的是UNIQUE KEY uk_symbol_date (symbol, trade_date)它为每只股票和每个交易日的组合设置了唯一约束。这个设计配合INSERT ... ON DUPLICATE KEY UPDATE语句可以实现幂等写入重复跑同一天的采集脚本不会产生重复数据只会更新已存在的记录。入库的Python代码如下# openstock/storage/db_engine.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker def get_engine(config): db_config config[database] conn_str ( fmysqlpymysql://{db_config[user]}:{db_config[password]} f{db_config[host]}:{db_config[port]}/{db_config[name]} f?charsetutf8mb4 ) return create_engine(conn_str, pool_size5, pool_recycle3600, echoFalse) def write_daily_quote(engine, df): df.to_sql(namedaily_quote, conengine, if_existsappend, indexFalse)写入时还有一个必须注意的问题pandas的to_sql默认用的是INSERT语句如果直接运行存量数据初始化脚本遇到重复行会直接报错。我这里提供了一个完整的初始化脚本在批量写入历史数据时做去重先清空表再全量写入或者用临时表过渡。日常增量更新则建议用INSERT IGNORE把已经存在的数据自动跳过只在缺失日期上补录。我推荐增量用这种做法from sqlalchemy.dialects.mysql import insert def upsert_daily_quote(engine, df): with engine.begin() as conn: stmt insert(daily_quote_table).values(df.to_dict(records)) update_cols {c.name: c for c in stmt.inserted if c.name not in (symbol, trade_date, id)} on_dup_stmt stmt.on_duplicate_key_update(update_cols) conn.execute(on_dup_stmt)3.3 API服务层对外提供数据接口有了数据接下来要做的事情把它们变成可以方便调用的接口。OpenStock的API层基于FastAPI实现它提供了几个重点接口下面是我的实现示例同时也是最常用的访问入口# openstock/api/main.py from fastapi import FastAPI, Query from sqlalchemy.orm import Session from fastapi.middleware.cors import CORSMiddleware app FastAPI() app.add_middleware( CORSMiddleware, allow_origins[*], allow_methods[*], allow_headers[*], ) app.get(/api/stock/) def get_stock_daily(symbol: str Query(...), limit: int 10): with Session(engine) as session: rows session.execute(text( SELECT * FROM daily_quote WHERE symbol :symbol ORDER BY trade_date DESC LIMIT :limit ), {symbol: symbol, limit: limit}).fetchall() return [dict(r) for r in rows] app.get(/api/stock/range) def get_stock_range(symbol: str, start: str, end: str): with Session(engine) as session: rows session.execute(text( SELECT * FROM daily_quote WHERE symbol :symbol AND trade_date BETWEEN :start AND :end ORDER BY trade_date ASC ), {symbol: symbol, start: start, end: end}).fetchall() return [dict(r) for r in rows]我个人在实际使用中最常用的是/api/stock/range接口做回测的时候拉取某只股票某段时间的数据一行请求就直接拿到JSON。启动API服务也很简单uvicorn openstock.api.main:app --host 0.0.0.0 --port 8000启动后浏览器打开http://你的服务器IP:8000/docs会看到自动生成的接口文档页面。你可以直接在页面上试接口非常方便。3.4 定时任务与增量更新行情数据不是一次性拉完就结束的每个交易日收盘后都需要增量更新。OpenStock用APScheduler实现了定时任务调度。# openstock/scheduler.py from apscheduler.schedulers.blocking import BlockingScheduler from openstock.collector.collector import fetch_daily_history from openstock.storage.db_engine import upsert_daily_quote def daily_update_job(): symbols load_symbol_list() engine get_engine(config) for symbol in symbols: try: df fetch_daily_history(symbol, start_dateNone, end_dateNone) if df is not None and not df.empty: # 只保留最近两个交易日的数据 df df.tail(2) upsert_daily_quote(engine, df) print(fsymbol {symbol} updated, rows: {len(df)}) except Exception as e: print(fsymbol {symbol} error: {e}) sleep(0.3) scheduler BlockingScheduler() scheduler.add_job(daily_update_job, cron, day_of_weekmon-fri, hour18, minute30) scheduler.start()这里的时间选择有讲究。A股下午3点收盘行情数据在收盘后需要一段时间才能在所有数据源上完整更新完。我测试过AKShare的日线数据在下午5点半之后基本就稳定了所以定时任务设置在傍晚6点半确保数据完整。如果你用其他数据源需要自己测试数据更新完成的时间窗口。我日常的增量更新策略是每个交易日收盘后自动拉取当天数据写入库中。对于已经入库的历史数据每周末跑一次全量“体检”把缺失日期补上即可。4. 手把手启动与验证所有代码和配置都准备好之后下面是完整的启动流程。我尽量把每一步都写出来包括我实际操作时看到的输出让你心里有数。4.1 初始化数据库进入项目根目录确保虚拟环境已经激活。然后执行数据库初始化脚本它会把建表语句跑一遍并在数据库里插入一个stock_symbols的股票代码表作为采集任务的分发清单。cd /opt/openstock mysql -uopenstock -pyour_strong_password openstock scripts/init_db.sql执行完成后可以用MySQL命令行检查表结构是否创建成功mysql -uopenstock -pyour_strong_password openstock -e SHOW TABLES;你会看到类似下面的输出---------------------- | Tables_in_openstock | ---------------------- | daily_quote | | stock_symbols | ----------------------4.2 启动API服务数据库初始化完成后先把API服务跑起来这样后续测试数据拉取时可以直接通过接口观察结果。nohup uvicorn openstock.api.main:app --host 0.0.0.0 --port 8000 /var/log/openstock_api.log 21 用nohup让它在后台持续运行日志输出到指定文件。启动后用curl简单验证服务是否正常curl http://localhost:8000/api/health正常情况下会返回类似{status: ok}的JSON响应。4.3 用实际请求验证数据完整性等API服务跑起来以后我建议做一次小范围的采集测试先把1到2只股票拉下来验证全链路是否通。# test_verify.py from openstock.collector.collector import fetch_daily_history from openstock.storage.db_engine import get_engine from openstock.storage.db_engine import upsert_daily_quote import yaml config yaml.safe_load(open(config/config.yaml)) engine get_engine(config) df fetch_daily_history(000001, start_date20240101, end_date20240401) print(df.head()) upsert_daily_quote(engine, df) print(f写入数据行数: {len(df)})执行这个测试脚本如果一切正常你会看到类似下面的输出symbol trade_date open high low close volume amount turnover 0 000001 2024-01-02 9.50 9.66 9.40 9.58 724423 693524.0 0.5300 1 000001 2024-01-03 9.60 9.70 9.50 9.62 605321 580122.0 0.4400 ... 写入数据行数: 58然后通过API验证数据是否已经可以被查询到curl http://localhost:8000/api/stock/?symbol000001limit3接口返回的JSON数组里应该有刚刚写入的记录。到这一步整套系统就已经跑通了。5. 常见问题与排查技巧实录搭建过程中我踩了不少坑这里直接整理成一份速查表你照着排查就行。5.1 数据源限流全量采集报错采集全市场股票的时候最容易遇到的就是数据源接口返回错误或是超时。AKShare虽然对个人用户挺友好但访问频率太密一样会触发限制。错误现象可能原因解决办法大量symbol采集失败提示缺少字段或直接返回空请求频率太高触发数据源限流把请求间隔从0.3秒提高到0.8秒以上并增加随机抖动单只股票反复失败该股票可能停牌太久或代码已退市失败重试3次后跳过记录到日志文件最后统一处理数据只有最近几个交易日被接口默认的最近行情窗口限制了明确传入start_date参数避免使用默认值这类问题确实是“等一等就好”的类型但这种等待时间非常浪费。后来我干脆写了一个失败重试的独立队列把失败任务先放到Redis队列里等全部跑完之后再统一补拉。这个方案后来也一直沿用了下来。实操心得采集任务的核心原则是“先跑通再跑全”。第一次上线不要试图一口气拉全市场先选20只股票把链路跑通再放开全量。这样排查问题时可观测性高得多你不会在几千只股票里找那几只失败的。5.2 时区与复权数据容易搞混如果你发现数据库里的交易日期总是差一天或者是回测数据和别人对不上八成是时区或者复权方式的问题。时区问题AKShare返回的日期字符串本身就是不含时区的交易日字符串不会因为服务器时区不同而偏移。但如果你的服务器时区设置有问题用Python的datetime.now()生成默认日期就会把“当前时间”当成“当前交易日”导致判断“是否交易日”出错。解决办法是落地的时间相关逻辑统一以数据库中已存在的MAX(trade_date)为准不要依赖服务器本地时间的推测。复权问题前复权数据会随时间推移而改变。你会发现昨天拉”2023年1月1日以来的前复权日线“跟今天拉同一段区间昨天的价格数值可能会变。原因是复权因子计算时会用到最新价格今天的除权除息导致历史价格被重新计算。所以如果你要做严谨的回测建议一次性把历史全量数据拉完然后固定下来平时只追加新交易日的数据。如果你每次都用前复权补全数据就会一直变动。5.3 数据库连接慢和日常运维为了省事很多人一开始用SQLite做存储跑通了但很快就发现两个问题并发读取端口很弱库存数据稍微多几百万行后就明显慢。如果你决定用MySQL下面几个坑值得留意。连接崩溃应用长时间运行后MySQL连接会超时导致新请求报错。这就是为什么我在get_engine中加了pool_recycle3600让连接池每小时刷新一次避开MySQL的wait_timeout。查询慢当daily_quote表的数据量超过500万行后基本的select可能变得卡顿。解决办法是按交易日期做分区比如按年份做RANGE分区查询某一段日期时只扫描对应分区速度可以提升好几倍。再有就是确保查询条件里始终带着symbol和trade_date避免全表扫描。备份策略我在每周日凌晨用crontab自动执行mysqldump备份整个openstock库保留最近30天的备份文件。个人项目数据库被误操作破坏是最常见但也最可惜的事故。5.4 内存与性能优化全市场采集时Python进程的内存占用很容易飙升。AKShare返回整个DataFrame后pandas会把所有数据一次性装入内存。如果你的服务器配置比较低建议按板块分批处理比如先拉上证主板再拉深证主板每批处理完立即入库并释放内存。def chunked(seq, size): for i in range(0, len(seq), size): yield seq[i:isize] symbols load_symbol_list() for batch in chunked(symbols, 200): process_batch(batch) gc.collect()还有一个很容易忽略的地方to_sql批量入库时如果用默认的methodmulti极大的数据块会一次性拼成一条超长SQL可能超出MySQL max_allowed_packet的上限报错时给了个莫名其妙的“Packet too large”提示。解决办法是适当调小批大小或者把max_allowed_packet调大我通常用chunksize500。最后说一个大家都关心的接口性能和并发。FastAPI本身是异步框架但我的SQLAlchemy连接池是同步的高并发时会有一定的阻塞风险。如果你后续要面向多个策略服务提供数据建议把API层再套一层Redis缓存热点请求完全不走数据库响应时间可以从几十毫秒降到几毫秒。我自己跑起来之后第一步就是把日线查询用Redis缓存了效果立竿见影。我个人在实际操作中的体会是OpenStock这套系统的价值不在于代码本身有多复杂而在于它把“从零开始搭一套行情数据服务”这件事变成了“只需要照着步骤配置就能跑通”。它省下的是最没有技术含量、又最耗费精力的数据基建部分把时间留给真正有价值的数据分析和策略研究。如果你搭完基础版本之后想继续扩展我的建议是按这个顺序来先去配置里把要监控的股票池调整成自己关心的行业然后给API层加一个Redis缓存最后再根据自己的回测框架对接数据接口。等到你开始写策略的时候会发现有一个稳定的数据底座是所有工作里最值的一笔投入。