我一直觉得，做投资研究最烦的不是没有想法，而是数据太散。今天想看看自选股的资金流，明天想复盘一下某只票的历史走势，后天又想把不同股票放在一个面板上对比——每一个需求都要单独开网站、单独查数据，时间全花在切换工具上了。后来我在开源社区看到一个叫 OpenStock 的解决方案，本质上是把行情数据抓取、存储、展示全部打通，自己本地部署一套个人股票行情工作台。这篇文章就是把我从零搭建 OpenStock 的完整过程写下来，包括架构思路、数据源选型、数据库设计、代码实现和一路踩过的坑，希望能帮到同样想自己动手折腾一套投研工具的朋友。
OpenStock 适合谁？如果你会一点 Python 基础语法，想拥有一个完全属于自己的股票数据面板，不想被任何平台的数据接口限制住；或者你单纯想练手，把“爬行情数据 + 定时任务 + 可视化展示”这条链路跑通，那这篇文章就是给你准备的。它不是那种复杂到看不懂的企业级量化系统，而是一个结构清晰、可以按需扩展的开源项目骨架。我会尽量把每一步的原理讲透，包括为什么选这个数据源、为什么这么设计数据库、为什么定时任务要这样写，方便你看完之后不只是抄代码，而是能自己改出想要的功能。
1. 内容整体设计与思路拆解
1.1 核心需求解析：为什么自己搭一套行情系统
在动手之前，我得先想明白一件事：市面上炒股软件那么多，自选股、K线、资金流、公告啥都有，为什么还要自己搭？回答这个问题，直接决定了 OpenStock 该怎么设计。
我的真实痛点是三个。第一，数据分散。同花顺看盘、东财查资金、韭研公社看研报、雪球看讨论，每个平台的数据口径还不一样，经常出现同一只股票在两个平台显示的涨跌幅对不上。第二，历史数据不可控。免费软件通常只能看最近几年的日K，想拉一只票过去十年的复权数据做回测，要么手动复制粘贴，要么付费开会员。第三，自己写的分析脚本没有统一的数据入口。我平时会用 Python 写一些选股策略，每次写策略都要重新现找数据接口、重新清洗数据，重复劳动非常严重。
OpenStock 的核心价值就在于把这三件事一次性解决：用统一的接口抓数据、用统一的数据库存数据、用统一的页面展示数据。之后我写任何策略脚本，只需要从自己的数据库里取数，数据格式固定、字段统一，时间成本能省下一大半。说白了，它不是一个给人推荐股票的“智能投顾”，而是一条让你自己的数据和代码跑起来的“基础设施”。
1.2 整体架构设计：四层结构各司其职
OpenStock 的整体架构我分成了四层，每一层解决一类问题，层与层之间用清晰的接口隔开。这样做的好处非常明显：以后想换数据源、换数据库、换前端展示，都只需要动对应的一层，其他层完全不受影响。
第一层是数据采集层，负责从公开数据源抓取股票列表、日线行情、实时盘口、资金流向等原始数据。第二层是数据存储层，把抓到的数据统一整理后写入数据库。第三层是业务逻辑层，提供数据查询接口，同时负责处理均线计算、涨跌幅计算、收益率统计等衍生指标。第四层是展示层，把数据用网页仪表盘的方式呈现出来，可以看自选股列表、K线形态、账户资产曲线。
这个分层思路，其实很像我以前写后端接口时的习惯。你要是一上来就把采集、存储、展示全塞到一个脚本里，表面上看跑通很容易，但一旦行情数据量变大，或者你想增加一个回测功能，整个脚本就会变得牵一发动全身，改起来想哭。所以哪怕 OpenStock 一开始只是个个人工具，我也建议你按这个分层结构来写，后面能省掉大量重构的成本。
注意：层与层之间尽量用函数调用或 HTTP 接口通信，不要直接共享全局变量。比如采集层只负责写数据库，展示层只负责读数据库，两者通过数据库这个“中间人”交互。
2. 核心细节解析与实操要点
2.1 数据源选型：谁才是最适合个人项目的行情接口
数据源是 OpenStock 最重要的地基，地基不稳，上面全白搭。我在选型时对比了五六种方案，最终留下两个主力数据源：AKShare 和 Tushare Pro。
AKShare 的优势是完全免费、不需要 token、接口非常多，从股票、基金、期货到宏观数据都有，而且接口命名很人性化，比如stock_zh_a_hist一看就知道是 A 股历史行情。它的缺点是接口返回的数据格式偶尔会变，毕竟它本质上是解析各财经网站的公开接口，网站改版它就要跟着适配，所以版本升级时经常要留意字段变化。
Tushare Pro 的优势是数据质量更稳定、字段更规范，而且有积分机制，积分越高能调用的接口越全。缺点是注册后默认积分很低，很多高级接口要积累积分才能用，对于只想本地存数据的新手来说门槛稍微高了一点。
我最终的选型策略是：日常抓日线行情用 AKShare，因为免费、灵活、覆盖广；如果后续要做更严肃的量化回测，再考虑用 Tushare Pro 补充更规范的基本面数据。这种“双数据源”的思路，能让你在某个源出问题时不至于完全抓瞎。
另外提醒一句，无论选哪个数据源，都不要在代码里硬编码 API 密钥或 token。我在 OpenStock 项目里统一用环境变量来管理，写到.env文件里，用python-dotenv加载，这样代码传到 GitHub 上也不会泄露敏感信息。
2.2 数据库设计：一张表搞定日线行情
存储层我选了 SQLite，没有上 MySQL 或 PostgreSQL。原因很简单：个人项目的数据量远没到需要独立数据库服务的程度，A 股五千多只股票，每只每天一条日线记录，一年下来也就一百多万行，SQLite 完全扛得住。而且 SQLite 是单文件数据库，备份、迁移都特别方便，拷个文件就能带走。
日线行情表是最核心的一张表，我的建表语句是这样的：
CREATE TABLE IF NOT EXISTS daily_price ( ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, open REAL, high REAL, low REAL, close REAL, pre_close REAL, change REAL, pct_chg REAL, vol REAL, amount REAL, PRIMARY KEY (ts_code, trade_date) );这里有几个设计点可以说一下。第一，为什么用ts_code + trade_date做复合主键？因为同一只股票同一天只能有一条日线数据，复合主键可以天然防止数据重复插入。第二，为什么 trade_date 用 TEXT 而不是 DATE 类型？因为数据源返回的日期格式基本都是YYYYMMDD或YYYY-MM-DD这种字符串，直接用 TEXT 存储可以避免类型转换出错，查询排序时字符串顺序也恰好是时间顺序，一举两得。第三，字段名称我尽量和数据源保持一致，像ts_code、pct_chg这种命名虽然看起来不够“通俗”，但好处是写抓取代码时可以直接拿数据源的字典键名映射，少写一堆rename代码。
除了日线表，我还设计了自选股表watchlist、账户持仓表positions和交易记录表trades。自选股表很简单，就是存股票代码和添加时间；持仓表记录每只股票当前持有数量和成本价；交易记录表则用来记录每一笔买卖操作，方便后续算收益率。这三张表的关联逻辑不复杂，等讲到实操环节我再展开。
2.3 抓取策略：别再循环里 sleep 了，请用限速器和重试
自己写爬虫抓行情，最容易犯的错误就是一上来for循环猛拉几千只股票，结果被对方服务器封 IP，或者被数据源限流。AKShare 虽然免费，但是我实测下来，短时间高频请求同样会被封，而且封了之后整个 IP 段都会受影响，非常麻烦。
我的做法是封装一个带限速和重试机制的函数。核心思路很简单：每两次请求之间至少间隔一个固定时间，比如 0.5 秒；请求失败时自动重试最多 3 次，重试前随机 sleep 一段时间，避免每次都撞在同一个时间点上。如果你要抓全市场五千只股票的日线数据，按 0.5 秒一个请求来算，大概是 40 多分钟，完全可以接受。千万不要为了省时间把间隔调到 0.1 秒以下，得不偿失。
为了实现这个机制，我没有直接用 requests，而是选用了 httpx（一个支持异步但也能同步用的 HTTP 库），结合tenacity库做重试。这样代码可读性更高，而且可以把重试策略统一配置。关于为什么用 httpx 而不是 requests，纯粹是因为 httpx 的 API 更现代，后续如果要改造成异步抓取，一行代码的改动量就能切换过去。
3. 实操过程与核心环节实现
3.1 环境准备：20 分钟从零搭好 Python 环境
开始写代码前，先把环境准备好。我用的是 Python 3.10，操作系统是 Windows 11，但下面的操作在 macOS 和 Linux 上完全通用。
第一步，创建项目目录和虚拟环境：
mkdir openstock cd openstock python -m venv venvWindows 激活虚拟环境用venv\Scripts\activate，macOS / Linux 用source venv/bin/activate。激活后命令行提示符前面会出现(venv)，就说明环境切换成功了。
第二步，安装依赖库。我按功能分了组，方便你按需安装：
pip install akshare pip install httpx pip install tenacity pip install pandas pip install streamlit pip install plotly pip install python-dotenv pip install scheduleAKShare 是数据源，httpx 是 HTTP 客户端，tenacity 做重试，pandas 做数据处理，streamlit 做前端面板，plotly 做交互图表，python-dotenv 读环境变量，schedule 做定时任务。这些库都是开源社区的成熟方案，组合起来完全不冲突。
第三步，把环境依赖导出到requirements.txt，方便以后换机器时一键安装：
pip freeze > requirements.txt3.2 数据采集模块：封装一个万能的行情抓取器
数据采集模块是整个 OpenStock 的入口，我把它拆成了collector.py和fetcher.py两个文件，职责分离。fetcher.py负责调用 AKShare 接口拿到原始数据，collector.py负责把原始数据转换成统一的字典格式并写入数据库。
先看fetcher.py的核心代码：
import akshare as ak import pandas as pd from tenacity import retry, stop_after_attempt, wait_random_exponential @retry(stop=stop_after_attempt(3), wait=wait_random_exponential(min=1, max=10)) def fetch_daily_history(ts_code: str, start_date: str, end_date: str) -> pd.DataFrame: """ 获取单只股票的日线历史行情。 ts_code: 股票代码，如 "000001" 或 "000001.SZ" """ symbol = ts_code.split(".")[0] 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: raise ValueError(f"No data for {ts_code}") return df这里有个非常重要的小细节：adjust参数我填的是"qfq"，也就是前复权。前复权的意思是以当前价格为基准，把历史价格按分红送股做调整，这样看 K 线时不会出现因为除权导致的跳空大坑。做技术分析时强烈建议用前复权数据，否则均线、MACD 这些指标全会被除权导致的假缺口干扰。如果你做的是分红策略，想单独看原始价格，那就用""不填复权。
tenacity装饰器的含义是：这个函数最多尝试 3 次，失败后等待一个随机的指数退避时间再重试，避免被数据源误判为攻击请求。我在实际运行中发现，AKShare 偶尔会因为目标网站反爬而抛异常，加上这个重试机制后，抓取全市场数据基本一次跑通。
再看collector.py的核心代码：
import sqlite3 import pandas as pd from fetcher import fetch_daily_history INSERT_SQL = """ INSERT OR REPLACE INTO daily_price (ts_code, trade_date, open, high, low, close, pre_close, change, pct_chg, vol, amount) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """ def save_daily_history(db_path: str, ts_code: str, start_date: str, end_date: str): df = fetch_daily_history(ts_code, start_date, end_date) if df is None or df.empty: return 0 # AKShare 返回的列名是中文，需要映射成统一的英文字段 df = df.rename(columns={ "日期": "trade_date", "开盘": "open", "收盘": "close", "最高": "high", "最低": "low", "涨跌幅": "pct_chg", "涨跌额": "change", "成交量": "vol", "成交额": "amount", "振幅": "amplitude", "换手率": "turnover_rate" }) rows = [ (ts_code, row["trade_date"], row["open"], row["high"], row["low"], row["close"], row.get("pre_close"), row["change"], row["pct_chg"], row["vol"], row["amount"]) for _, row in df.iterrows() ] with sqlite3.connect(db_path) as conn: conn.executemany(INSERT_SQL, rows) return len(rows)这里INSERT OR REPLACE是关键。由于daily_price表的主键是(ts_code, trade_date)，如果当天已经抓过某只股票的数据，再次抓取时会自动覆盖旧数据，保证数据库里永远是最新的。这就是为什么前面强调要建复合主键——没有主键约束的话，重复抓取会导致数据越堆越多，最终出现大量重复记录。
3.3 定时任务与全市场更新：把“每天自动更新”变成现实
手动跑数据采集脚本只能解决“一次性建库”的问题，想让 OpenStock 真正像日报一样每天自动更新，就得靠定时任务。我在项目里用了schedule这个轻量级库，它比操作系统的 cron 好在跨平台且能嵌入 Python 进程。
我的调度脚本长这样：
import schedule import time from collector import update_all_stocks, update_realtime_quotes # 工作日每天 17:30 更新日线数据 schedule.every().monday.at("17:30").do(update_all_stocks) schedule.every().tuesday.at("17:30").do(update_all_stocks) schedule.every().wednesday.at("17:30").do(update_all_stocks) schedule.every().thursday.at("17:30").do(update_all_stocks) schedule.every().friday.at("17:30").do(update_all_stocks) # 开盘时间每 5 分钟更新一次实时行情 schedule.every(5).minutes.do(update_realtime_quotes) while True: schedule.run_pending() time.sleep(30)为什么选择 17:30 更新日线？因为 A 股 15:00 收盘后，各大数据源通常需要半小时左右完成数据清洗和入库，15:30 就能拿到当天完整的日线数据。我留到 17:30 再拉，既能确保数据齐全，又不会太晚影响我晚上复盘。
不过schedule库有个小毛病，它只适合单进程的轻量任务，如果你的定时任务数量多了，或者担心进程崩溃导致错过更新，我建议直接用系统自带的定时器（Windows 的任务计划程序 / macOS 和 Linux 的 crontab）来调用你的 Python 脚本。比如在 crontab 里写这样一行就够了：
30 17 * * 1-5 cd /path/to/openstock && /path/to/venv/bin/python run_update.py这样即使电脑重启，只要系统时间正确，脚本到点照样执行。个人项目里，这个简单方案往往比引入 Celery 这种重量级任务队列靠谱得多。
3.4 展示层：用 Streamlit 拼出一个可视化仪表盘
数据存好了，下一步就是让它“看得见”。展示层我选了 Streamlit，原因特别简单：它是目前开源社区里对小白最友好的 Python Web 框架，几十行代码就能出一个带图表、带下拉框、带指标卡的交互式面板，不用写一行前端代码。
我在展示层做了三个页面分支：
- 自选股行情页：展示自选股列表、最新价、涨跌幅、5 日走势图。
- 个股深度分析页：输入股票代码后展示日 K 线图、成交量图、60 日均线和 120 日均线。
- 账户资产页：展示每笔交易记录、持仓成本和当前收益率。
核心代码片段如下：
import sqlite3 import pandas as pd import streamlit as st import plotly.graph_objects as go st.set_page_config(page_title="OpenStock", layout="wide") @st.cache_data(ttl=300) def load_daily_data(ts_code: str) -> pd.DataFrame: conn = sqlite3.connect("openstock.db") df = pd.read_sql_query( "SELECT * FROM daily_price WHERE ts_code = ? ORDER BY trade_date", conn, params=(ts_code,) ) conn.close() return df # 侧边栏：选择股票 ts_code = st.sidebar.selectbox("选择股票", watchlist_codes()) df = load_daily_data(ts_code) if len(df) > 20: df["ma5"] = df["close"].rolling(5).mean() df["ma20"] = df["close"].rolling(20).mean() df["ma60"] = df["close"].rolling(60).mean() fig = go.Figure() fig.add_trace(go.Candlestick( x=df["trade_date"], open=df["open"], high=df["high"], low=df["low"], close=df["close"], name="K线" )) fig.add_trace(go.Scatter(x=df["trade_date"], y=df["ma20"], mode="lines", name="MA20")) fig.add_trace(go.Scatter(x=df["trade_date"], y=df["ma60"], mode="lines", name="MA60")) st.plotly_chart(fig, use_container_width=True)这里st.cache_data(ttl=300)是性能优化的关键。Streamlit 每次界面交互都会重新执行整个脚本，如果每次都查数据库，页面会卡到怀疑人生。加上ttl=300后，同一个股票代码的数据在 5 分钟内只查一次数据库，后续滚动、缩放图表都会走缓存，体感顺畅很多。
K 线图我用了 Plotly 的Candlestick，可以在同一个图里叠加均线。Plotly 图表天然支持缩放、平移、悬浮提示，鼠标一放上去就能看到 OHLC（开高低收）数据，复盘体验比静态图片强太多了。
3.5 账户管理与收益率计算：让系统不只是看客
OpenStock 不能只做行情展示，不然它跟纯看盘软件没区别。我在系统里加入了持仓管理模块，用来记录每次模拟买入和卖出的操作，然后自动算当前持仓市值和累计收益率。
交易记录表的结构如下：
CREATE TABLE IF NOT EXISTS trades ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, side TEXT NOT NULL CHECK (side IN ('BUY', 'SELL')), price REAL NOT NULL, shares REAL NOT NULL, fee REAL DEFAULT 0 );每次买入时往trades表插入一条side = 'BUY'的记录，卖出时插入一条side = 'SELL'。持仓数量就是所有买入数量减去所有卖出数量。这里要注意手续费的处理，A 股佣金一般是万 2.5 到千 3 不等，买入卖出都要扣，卖出时还要加印花税。虽然单笔看起来不多，但长期交易下来手续费会明显扭曲收益率，所以我干脆把每笔手续费也记录下来，在算收益率时直接扣除。
计算收益率的逻辑不复杂，就是先算当前市值（最新价乘以持仓数量），减去总买入成本（含费），再加上总卖出收入（扣费），最后除以总投入本金得到收益率。这部分的 SQL 比较长，我直接写在 Python 里用 pandas 汇总，写完读起来比一大串 SQL 清晰得多。
4. 常见问题与排查技巧实录
4.1 数据抓取报错：网络超时与数据源字段变动
我在搭建 OpenStock 的过程里遇到最多的坑，几乎都集中在数据采集这一层。第一个经典问题是ConnectTimeout或者ReadTimeout，尤其下午收盘后半小时内抓数据，数据源服务器负载特别高，经常超时。解决思路就是前面说的 tenacity 重试机制，超时后等几秒再重试，基本都能成功。
第二个经典问题是 AKShare 版本升级后字段名变了。某次我执行抓取脚本，突然报KeyError: '日期'，排查半天才发现是库升级后返回的DataFrame列名从“日期”变成了“时间”。这种问题很难提前预防，我的处理方法是两层保险：第一层是在requirements.txt里固定 AKShare 的版本号，不要随意升级；第二层是在 rename 之前先打印一下df.columns，方便出问题时快速定位。如果你把 AKShare 当成生产依赖来用，建议把抓出来的数据先做一层 schema 校验，字段对不上就抛异常并跳过去，避免脏数据污染数据库。
4.2 主键冲突与数据重复：为什么用 INSERT OR REPLACE
如果你刚开始没建复合主键，跑了几次全市场更新后再去查数据库，大概率会发现daily_price表里同一只股票同一天有多条记录。这就是没加主键约束导致的。
解决方法是把表删了重建，再加上复合主键。如果你已经在表里积累了很多数据不想丢弃，可以先查重再删除：
DELETE FROM daily_price WHERE rowid NOT IN ( SELECT MIN(rowid) FROM daily_price GROUP BY ts_code, trade_date );然后再补上唯一索引：
CREATE UNIQUE INDEX idx_unique_price ON daily_price(ts_code, trade_date);不过说实话，删表重建更干脆。因为行情数据本来就是从公开源抓来的，删了重新抓一遍也就几十分钟的事情，没必要保留有问题的历史数据。
4.3 SQLite 数据库锁：并发读写时的 Performing Recovery 问题
SQLite 适合单进程访问，但如果你一边跑定时抓取脚本，一边开着 Streamlit 展示页面，就可能会出现database is locked的报错。原因是 SQLite 同一时间只允许一个进程写数据，抓取脚本正在批量写入时，展示页面的读请求会被暂时阻塞。
我踩过这个坑后的解决方案有两条。第一，把数据库连接改为每次操作都重新打开关闭，不用长连接。SQLite 的连接开销非常低，每次短连接能最大程度避免锁冲突。第二，把写操作放在事务里批量执行，不要一条一条 commit。比如前面代码里的executemany，一次性插入几百上千行，远比循环里execute+commit快得多，占锁时间也短得多。
如果你的并发需求真的很强，比如要同时跑多个抓取进程，那就别硬撑 SQLite 了，尽早切换到 PostgreSQL，OpenStock 的存储层代码我当初就是按 SQL 标准语法写的，切换成本很低。
4.4 复权数据不一致：前复权数据会随着时间变化而变化
最后分享一个非常容易踩但很多人没意识到的问题：前复权数据是“会变”的。因为前复权是以当前最新价格作为基准去倒推历史价格，如果某只股票今天除权除息，那么它过去所有的历史价格都会被调整一遍，以前拉到的历史数据就“过时”了。
这个特性会导致一个潜在 bug：你今天建的数据库里存了某只股票过去三年的前复权数据，下周它分红除权了，你再增量更新最新的日线数据，会发现今天这天的数据跟数据库里已有的历史数据之间出现价格断层，甚至指标计算都会出错。要彻底解决这个问题，只有定期对所有股票做一次全量重抓，覆盖掉旧的复权数据。我的做法是每周末把全市场股票的日线数据重刷一遍，大概耗时一个小时，换取数据一致性，非常划算。
5. 进阶扩展：从展示工具到策略回测平台
OpenStock 跑到这一步，已经是个可以日常使用的个人行情面板了。但说实话，如果只是复制粘贴到这就结束，它跟普通看盘软件的区别还没完全体现出来。我的下一步计划，也是我认为这个项目最有价值的方向，是把 OpenStock 扩展成自己的策略验证平台。
具体来说有两个可以马上动手的方向。第一个是均线策略回测。因为数据库里已经有完整的历史日线数据，写一个简单的双均线策略回测器只需要几十行代码：金叉买入、死叉卖出，算一下年化收益率和最大回撤。这样你不用再等行情软件慢慢回放，自己就能批量验证不同参数在不同股票上的表现，而且所有取数逻辑都基于本地数据库，速度非常快。
第二个方向是接入企业微信或钉钉机器人做异动提醒。当你自定义的监控条件满足时，比如某只自选股单日涨跌幅超过 5%、成交量放大到 5 日均量的 3 倍、均线金叉等，OpenStock 可以自动推送消息到手机上。这个需求其实是把定时任务、条件判断和消息推送三条链路串起来，技术难度不高，但能极大提升日常盯盘的效率。
我个人在实际操作中的体会是：OpenStock 这个项目最值钱的不是某个单独的代码功能，而是它把数据获取、存储、分析、展示这一整条链路打通了。以前我写策略卡在数据上，现在数据都在自己手里，想怎么算都行。最后再分享一个小技巧：如果你也想长期维护这个项目，一定要养成为每个模块写if __name__ == "__main__":测试入口的习惯，改完一个函数就跑一下，这比任何高级架构设计都更能帮你防止代码腐烂。
