OpenStock开源项目：手把手搭建A股行情数据采集与展示系统

要说最近在金融数据这个圈子里有什么值得自己动手玩一玩的开源项目，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 'openstock'@'localhost' IDENTIFIED BY 'your_strong_password'; GRANT ALL PRIVILEGES ON openstock.* TO 'openstock'@'localhost'; 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 pyyaml

3. 核心功能模块的实现

环境准备好了，接下来说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(symbol=symbol, period="daily", start_date=start_date, end_date=end_date, adjust="qfq") 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"]]

重点解释几个选择：

第一个是adjust="qfq"，这个参数代表前复权。股票发生过送转股、分红之后，历史价格会出现跳空，如果不复权直接用原始价格做回测或画图，趋势线会失真。前复权是把历史价格调整到当前的价格基准，保持价格的连续性。如果你做的是分红再投入策略，也可以取后复权数据，看实际情况。

第二个是字段改名。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) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

这里最关键的是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 = ( f"mysql+pymysql://{db_config['user']}:{db_config['password']}" f"@{db_config['host']}:{db_config['port']}/{db_config['name']}" f"?charset=utf8mb4" ) return create_engine(conn_str, pool_size=5, pool_recycle=3600, echo=False) def write_daily_quote(engine, df): df.to_sql(name="daily_quote", con=engine, if_exists="append", index=False)

写入时还有一个必须注意的问题：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_date=None, end_date=None) if df is not None and not df.empty: # 只保留最近两个交易日的数据 df = df.tail(2) upsert_daily_quote(engine, df) print(f"symbol {symbol} updated, rows: {len(df)}") except Exception as e: print(f"symbol {symbol} error: {e}") sleep(0.3) scheduler = BlockingScheduler() scheduler.add_job(daily_update_job, "cron", day_of_week="mon-fri", hour=18, minute=30) scheduler.start()

这里的时间选择有讲究。A股下午3点收盘，行情数据在收盘后需要一段时间才能在所有数据源上完整更新完。我测试过，AKShare的日线数据在下午5点半之后基本就稳定了，所以定时任务设置在傍晚6点半，确保数据完整。如果你用其他数据源，需要自己测试数据更新完成的时间窗口。

我日常的增量更新策略是：每个交易日收盘后自动拉取当天数据写入库中。对于已经入库的历史数据，每周末跑一次全量“体检”，把缺失日期补上即可。

4. 手把手启动与验证

所有代码和配置都准备好之后，下面是完整的启动流程。我尽量把每一步都写出来，包括我实际操作时看到的输出，让你心里有数。

4.1 初始化数据库

进入项目根目录，确保虚拟环境已经激活。然后执行数据库初始化脚本，它会把建表语句跑一遍，并在数据库里插入一个stock_symbols的股票代码表作为采集任务的分发清单。

cd /opt/openstock mysql -uopenstock -p'your_strong_password' openstock < scripts/init_db.sql

执行完成后，可以用MySQL命令行检查表结构是否创建成功：

mysql -uopenstock -p'your_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 2>&1 &

用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_date="20240101", end_date="20240401") 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/?symbol=000001&limit=3"

接口返回的JSON数组里应该有刚刚写入的记录。到这一步，整套系统就已经跑通了。

5. 常见问题与排查技巧实录

搭建过程中我踩了不少坑，这里直接整理成一份速查表，你照着排查就行。

5.1 数据源限流：全量采集报错

采集全市场股票的时候，最容易遇到的就是数据源接口返回错误或是超时。AKShare虽然对个人用户挺友好，但访问频率太密一样会触发限制。

这类问题确实是“等一等就好”的类型，但这种等待时间非常浪费。后来我干脆写了一个失败重试的独立队列，把失败任务先放到Redis队列里，等全部跑完之后再统一补拉。这个方案后来也一直沿用了下来。

实操心得：采集任务的核心原则是“先跑通，再跑全”。第一次上线不要试图一口气拉全市场，先选20只股票把链路跑通，再放开全量。这样排查问题时可观测性高得多，你不会在几千只股票里找那几只失败的。

5.2 时区与复权数据容易搞混

如果你发现数据库里的交易日期总是差一天，或者是回测数据和别人对不上，八成是时区或者复权方式的问题。

• 时区问题：AKShare返回的日期字符串本身就是不含时区的交易日字符串，不会因为服务器时区不同而偏移。但如果你的服务器时区设置有问题，用Python的datetime.now()生成默认日期，就会把“当前时间”当成“当前交易日”，导致判断“是否交易日”出错。解决办法是落地的时间相关逻辑统一以数据库中已存在的MAX(trade_date)为准，不要依赖服务器本地时间的推测。
• 复权问题：前复权数据会随时间推移而改变。你会发现，昨天拉”2023年1月1日以来的前复权日线“，跟今天拉同一段区间，昨天的价格数值可能会变。原因是复权因子计算时会用到最新价格，今天的除权除息导致历史价格被重新计算。所以如果你要做严谨的回测，建议一次性把历史全量数据拉完，然后固定下来，平时只追加新交易日的数据。如果你每次都用前复权补全，数据就会一直变动。

5.3 数据库连接慢和日常运维

为了省事，很多人一开始用SQLite做存储跑通了，但很快就发现两个问题：并发读取端口很弱，库存数据稍微多几百万行后就明显慢。如果你决定用MySQL，下面几个坑值得留意。

• 连接崩溃：应用长时间运行后，MySQL连接会超时，导致新请求报错。这就是为什么我在get_engine中加了pool_recycle=3600，让连接池每小时刷新一次，避开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:i+size] symbols = load_symbol_list() for batch in chunked(symbols, 200): process_batch(batch) gc.collect()

还有一个很容易忽略的地方：to_sql批量入库时，如果用默认的method="multi"，极大的数据块会一次性拼成一条超长SQL，可能超出MySQL max_allowed_packet的上限，报错时给了个莫名其妙的“Packet too large”提示。解决办法是适当调小批大小，或者把max_allowed_packet调大，我通常用chunksize=500。

最后说一个大家都关心的：接口性能和并发。FastAPI本身是异步框架，但我的SQLAlchemy连接池是同步的，高并发时会有一定的阻塞风险。如果你后续要面向多个策略服务提供数据，建议把API层再套一层Redis缓存，热点请求完全不走数据库，响应时间可以从几十毫秒降到几毫秒。我自己跑起来之后，第一步就是把日线查询用Redis缓存了，效果立竿见影。

我个人在实际操作中的体会是，OpenStock这套系统的价值不在于代码本身有多复杂，而在于它把“从零开始搭一套行情数据服务”这件事变成了“只需要照着步骤配置就能跑通”。它省下的是最没有技术含量、又最耗费精力的数据基建部分，把时间留给真正有价值的数据分析和策略研究。

如果你搭完基础版本之后想继续扩展，我的建议是按这个顺序来：先去配置里把要监控的股票池调整成自己关心的行业，然后给API层加一个Redis缓存，最后再根据自己的回测框架对接数据接口。等到你开始写策略的时候，会发现有一个稳定的数据底座，是所有工作里最值的一笔投入。