把 OpenStock 这套开源股票行情分析系统从零到一搭起来前后花了我一个多星期。它本质上是一套自建的行情数据聚合与策略验证平台通过公开数据源拉取日K、分钟K线清洗后落库再用 Web 界面画 K 线、算技术指标、跑简单的策略回测。如果你一直觉得市面看盘软件太封闭、想把交易想法用代码快速验证一遍或者想培养自己的数据分析和量化基本功这套 OpenStock 会非常适合你。我会把整个搭建过程、踩过的坑、以及最终能跑通的完整方案都写清楚建议收藏后照着操作。1. OpenStock 项目定位与整体设计1.1 这个项目要解决什么问题我在之前做交易记录和复盘时最大的痛点是数据太散。今天用券商软件看一眼日K明天用网页工具查一下基本面真正想做一个“过去三年某策略表现如何”的回测时根本找不到足够长、足够干净的历史数据。手工从Excel整理一天能坚持一个月就能把人逼疯。所以 OpenStock 的出发点很朴素把“数据采集、存储、展示、回测”这条链路做成一个开源单机系统所有数据都落在自己手里想看什么指标自己写想验证什么策略直接跑不再受第三方看盘软件的功能限制。它适合四类人想入门量化分析的开发者、需要做投研数据归档的爱好者、Freelancer 接了金融数据展示类项目的前后端工程师以及单纯想搞一套 K 线可视化工具练手的学生。1.2 技术选型Python FastAPI SQLite ECharts技术栈选型我犹豫过最终确定为 Python FastAPI SQLite ECharts。Python 生态做金融数据分析最成熟pandas 处理K线数据几乎就是标准答案所以采集和计算都用 Python。后端接口选择了 FastAPI 而不是 Flask原因是它自带 OpenAPI 文档调接口时可以直接在浏览器里看返回结构对 K 线图联调非常有帮助性能也足够个人使用。数据库选了 SQLite 而不是 PostgreSQL主要是考虑部署成本和个人数据量级。一套日K线数据全市场五千多只股票如果存三年大约不到一千万行SQLite 完全扛得住而且备份就是拷一个文件非常省事。等数据量真正大起来再把连接串换成 PostgreSQL代码改动量很小。前端可视化用 ECharts它的 K 线图组件自带蜡烛图、均线、成交量联动缩放也顺手不用自己造轮子。组件选型理由数据采集akshare免费、无需Token、支持A股日K/分钟K/实时行情数据存储SQLite单文件、零配置、适合个人级数据量后端服务FastAPI自带API文档、异步支持好、与pandas配合方便前端图表EChartsK线图组件成熟、缩放交互流畅部署方式Docker Compose一条命令启动全部服务、环境隔离数据源这块我对比过 akshare、baostock、tushare 三套方案。tushare 的积分机制对新手不友好很多高质量接口需要高积分才能用baostock 代码稳定但接口纬度偏窄分钟线力度不足akshare 接口每天在更新覆盖面最广从K线到行业板块、资金流向都有虽然偶尔会因目标网站改版而失效但整体维护节奏快更适合做个人项目的数据底座。2. 环境准备与数据库表结构设计2.1 本地环境搭建与项目初始化正式开始前先安排环境。我用的是 Python 3.10推荐大家至少用 3.9 以上版本否则一些 pandas 和 FastAPI 的新特性会报错。项目目录我习惯这样组织openstock/ ├── app/ │ ├── main.py # FastAPI 入口 │ ├── database.py # 数据库连接与建表 │ ├── models.py # ORM 模型 │ ├── routers/ │ │ ├── kline.py # K线数据接口 │ │ └── indicator.py # 指标接口 │ ├── services/ │ │ ├── fetcher.py # 数据抓取 │ │ ├── cleaner.py # 数据清洗 │ │ ├── indicator.py # 指标计算 │ │ └── backtest.py # 策略回测 │ └── static/ # 前端页面资源 ├── scripts/ │ ├── init_db.py # 初始化数据库 │ └── sync_data.py # 每日增量同步 ├── docker-compose.yml └── requirements.txt创建虚拟环境并安装依赖我用的是 uv 工具比 pip 快很多如果你还在用 pip 也完全没问题但注意不要全局安装避免污染系统 Pythonuv venv uv pip install fastapi uvicorn pandas akshare sqlalchemy pydantic docker-compose安装完成后顺手验证一下 akshare 能不能正常拉数据这一步值得提前做不然后面写一堆代码才发现数据源不通会很痛苦。测试命令很简单python -c import akshare as ak; df ak.stock_zh_a_hist(symbol000001, perioddaily, start_date20240101, end_date20240201); print(df.head())能输出一段表格式数据就说明环境没问题。实际测试中如果遇到网络超时多半是目标数据源接口限流稍等重试即可。2.2 数据库表结构为行情数据量身设计数据库用于存个股基础信息和K线数据我个人不建议把所有东西塞进一张大表。行情数据按股票代码和时间两个维度查询最频繁所以主表stock_kline必须建联合索引否则数据量上来后查询会越来越慢。核心表结构如下CREATE TABLE IF NOT EXISTS stock_basic ( code TEXT PRIMARY KEY, name TEXT NOT NULL, market TEXT, industry TEXT, list_date TEXT ); CREATE TABLE IF NOT EXISTS stock_kline ( code TEXT NOT NULL, date TEXT NOT NULL, open REAL NOT NULL, high REAL NOT NULL, low REAL NOT NULL, close REAL NOT NULL, volume REAL NOT NULL, amount REAL, adjust_flag TEXT DEFAULT qfq, PRIMARY KEY (code, date, adjust_flag) ); CREATE INDEX IF NOT EXISTS idx_kline_code_date ON stock_kline(code, date);有几个设计细节需要特别说清楚。date字段我统一用字符串YYYY-MM-DD格式而不是时间戳原因是日K线天然不需要时区换算字符串排序和比较都符合时间顺序查询也直观。adjust_flag这个字段非常关键它标识K线的复权类型可以是none不复权、qfq前复权、hfq后复权同一只股票在不同复权方式下的价格不同必须作为联合主键的一部分否则复权数据会互相覆盖。主键设计成(code, date, adjust_flag)还有一个好处就是做增量更新时可以直接用INSERT OR REPLACE省去先查重再插入的麻烦。建表我没有直接连 SQLite 命令行而是用 SQLAlchemy 在应用启动时自动执行这样后续改动表结构会更方便。init_db.py里放建表语句每次启动 FastAPI 前执行一次保证环境一致性。3. 行情数据采集与清洗落库流程3.1 数据抓取用 akshare 替代手工下载写数据采集模块时我踩过不少坑第一个就是“接口返回的字段名会变”。akshare 是爬虫类库它抓取的目标是公开网页接口目标网站一改版返回的列名就可能从日期变成date或者中间多出一列。所以我把抓取和清洗拆成两个独立的 service抓取只负责拿原始数据清洗统一处理列名和类型这样即使接口变化改一个函数就行而不用动核心逻辑。抓取日K线的核心函数import akshare as ak import pandas as pd from datetime import datetime, timedelta def fetch_daily_kline(code: str, start_date: str, end_date: str, adjust: str qfq): df ak.stock_zh_a_hist( symbolcode, perioddaily, start_datestart_date.replace(-, ), end_dateend_date.replace(-, ), adjustadjust, ) if df is None or df.empty: return pd.DataFrame() df.columns [date, open, close, high, low, volume, amount, amplitude, pct_change, change, turnover] df df[[date, open, close, high, low, volume, amount]].copy() df[date] pd.to_datetime(df[date]).dt.strftime(%Y-%m-%d) df[volume] pd.to_numeric(df[volume], errorscoerce) df[amount] pd.to_numeric(df[amount], errorscoerce) return df这里我把原始数据的英文列名映射到自己定义的标准列名上即使 akshare 内部把中文列名改成英文只要数量对得上重新映射一次就行。列顺序我特意调整成date, open, close, high, low而不是常规的date, open, high, low, close这是根据 akshare 实际返回顺序记忆的你也可以直接按索引取列更省事。3.2 清洗与增量更新避免重复数据和脏数据数据抓回来后不能直接入库必须过一遍清洗。清洗逻辑重点处理三件事类型转换、去重排序、增量过滤。类型转换是第一个大坑。akshare 返回的成交量和成交额有时候是字符串或者其他类型直接影响后续指标计算。我会用pd.to_numeric强制转换转换不了的值变成NaN再统一填充或剔除。成交量在SQLite里我用 REAL 类型存储因为部分接口返回的成交量是手数或股数单位不一致用 REAL 比 INTEGER 更安全哪怕后续换数据源也不会因为整数溢出丢数据。增量更新的做法是每次同步前先把库里的最大日期查出来只从那个日期开始往后拉避免全量重跑浪费时间。比如今天是2025年6月10日库里已经有6月9日的数据那增量同步区间就是6月9日到今天目标网站会返回6月9日和6月10日两天数据通过主键冲突覆盖来保证当天的数据是最新的因为盘中或盘后数据可能修正过。def sync_stock(code: str, start_date: str None): conn get_connection() if start_date is None: row conn.execute( SELECT MAX(date) FROM stock_kline WHERE code ?, (code,) ).fetchone() start_date row[0] if row[0] else 20000101 end_date datetime.now().strftime(%Y%m%d) df fetch_daily_kline(code, start_date, end_date, adjustqfq) if df.empty: return 0 records df.to_dict(records) conn.executemany( INSERT OR REPLACE INTO stock_kline (code, date, open, high, low, close, volume, amount, adjust_flag) VALUES (?, ?, ?, ?, ?, ?, ?, ?, qfq) , [(code, r[date], r[open], r[high], r[low], r[close], r[volume], r[amount]) for r in records], ) conn.commit() return len(records)这段代码每次调用都会执行INSERT OR REPLACE所以即使重复拉取数据也不会产生重复记录。增量更新一个人每天跑一遍就够了写个循环把所有股票代码都过一遍大概几分钟可以完成全市场同步。4. 指标计算与 K 线可视化4.1 常用技术指标计算MA、MACD、RSI 的实现细节有了干净的K线数据就可以算技术指标了。技术指标看似简单实际实现时有很多细节会影响结果。我实现的第一个指标是 MA 均线用的是 pandas 的滚动窗口这个最简单def calc_ma(df: pd.DataFrame, windows[5, 10, 20, 60]): for w in windows: df[fma{w}] df[close].rolling(windoww, min_periodsw).mean() return dfMACD 的计算就要小心了它依赖 EMA 指数移动平均pandas 的ewm方法默认参数和同花顺、通达信这类行情软件存在细微差异直接导致指标数值对不上。我遇到过最典型的问题是ewm的adjust参数默认是 True而行情软件用的是adjustFalse这种递推模式。必须显式设置adjustFalse并且用alpha2/(span1)来匹配。这样计算出来的 DIF、DEA、MACD 柱状图才跟主流看盘软件一致。def calc_macd(df: pd.DataFrame, fast12, slow26, signal9): ema_fast df[close].ewm(spanfast, adjustFalse).mean() ema_slow df[close].ewm(spanslow, adjustFalse).mean() df[dif] ema_fast - ema_slow df[dea] df[dif].ewm(spansignal, adjustFalse).mean() df[macd] 2 * (df[dif] - df[dea]) return dfRSI 这类摆动指标更考验对公式的理解。RSI 的计算基础是平均涨幅和平均跌幅直接用 rolling 对每天的涨跌幅做平均虽然简单但和主流行情软件常用的 Wilder 平滑方法会有数值差异。为了让指标尽量接近大家熟悉的软件我用了 Wilder 递推先算首个14日的平均涨跌幅之后每一天用前一天的值乘以13再加上当日涨跌幅除以14这种方式算出来的RSI曲线更平滑金叉死叉信号也更稳定。4.2 后端接口与 ECharts K 线图对接指标算完后通过 FastAPI 暴露给前端。我设计了一个统一的 K 线接口返回的数据结构包括日期、开高低收、成交量和已经算好的技术指标前端拿过来直接画图不需要再做二次计算。app.get(/api/stock/{code}/kline) def get_kline(code: str, limit: int 250): df load_kline_from_db(code, limitlimit) df calc_ma(df) df calc_macd(df) return { code: code, data: df[[date, open, high, low, close, volume, ma5, ma20, dif, dea, macd]].to_dict(records), }前端页面我用了一个极大的简化方案一个 HTML 页面引入 ECharts 的 CDN 文件然后用 fetch 请求上面的接口把数据填充到series里。K 线图的关键在于要把data数组按[open, close, low, high]的顺序传入不少新手会在这里搞混导致K线画出来是颠倒的。成交量图用柱状图覆盖在同一个坐标系下再把 MA 和 MACD 作为独立的折线叠加整体效果就比较接近专业看盘软件了。async function loadKline(code) { const res await fetch(/api/stock/${code}/kline?limit250); const json await res.json(); const baseData json.data.map(d [d.date, d.open, d.close, d.low, d.high]); chart.setOption({ xAxis: { type: category, data: json.data.map(d d.date) }, yAxis: { scale: true }, series: [ { type: candlestick, data: baseData }, { type: line, data: json.data.map(d d.ma5), smooth: true }, { type: line, data: json.data.map(d d.ma20), smooth: true } ] }); }这里有个体验细节是缩放。ECharts 的dataZoom组件一定要开启不然当你看一只股票一年的日K时550根K线挤在一起根本看不清。dataZoom的inside支持鼠标滚轮缩放slider可以拖到底部缩放两个一起用最好。5. 策略引擎与回测功能实现5.1 双均线策略为什么选它作为第一个策略OpenStock 的核心不只是看盘我还加入了策略回测能力。第一个实现的策略是双均线策略当短期均线上穿长期均线时买入下穿时卖出。双均线逻辑足够简单适合验证整个回测框架的正确性等框架稳了再往里面加更复杂的策略逻辑。为什么选择双均线而不是更复杂的机器学习模型因为回测框架最容易出错的地方是“未来函数”也就是不小心用了当天的数据去决策当天的操作这在现实中根本做不到。双均线策略逻辑直观便于检查每一个交易信号是否严格发生在收盘后。等你的回测框架经过双均线验证没问题后再上其他策略才有底气不然回测收益再高也是假的。def backtest_double_ma(df: pd.DataFrame, fast: int 5, slow: int 20): df calc_ma(df, windows[fast, slow]) df[position] 0 df.loc[df[fma{fast}] df[fma{slow}], position] 1 df[position] df[position].shift(1).fillna(0) df[signal] df[position].diff() return df第四行shift(1)是整个回测里最关键的代码。它把持仓信号整体向后移动一格意思是今天收盘后根据均线关系算出信号明天开盘才执行交易避免在“今天收盘的同时用收盘信号去计算今天的收益”这种容易出错的逻辑。5.2 回测流程交易信号、持仓管理与收益统计有了持仓信号之后就可以模拟完整的交易过程了。我会计算每天的仓位价值、买入卖出点、手续费和最终收益。def run_backtest(df: pd.DataFrame, init_cash: float 100000.0, fee_rate: float 0.0003): df[position] df[position].shift(1).fillna(0) df[next_open] df[open].shift(-1) df[cash] float(init_cash) df[holding] 0.0 df[equity] 0.0 cash float(init_cash) holding 0.0 for i in range(len(df)): signal df.iloc[i][signal] if signal 1 and holding 0: buy_price df.iloc[i][open] * 1.0011 holding cash / buy_price cash 0.0 elif signal -1 and holding 0: sell_price df.iloc[i][open] * 0.9987 cash holding * sell_price holding 0.0 df.loc[df.index[i], cash] cash df.loc[df.index[i], holding] holding df.loc[df.index[i], equity] cash holding * df.iloc[i][close] return df这里我给自己预留了手续费和滑点空间。买入价在开盘价基础上上浮千分之1.1卖出价在开盘价基础上下调千分之1.3这是A股双边手续费加上冲击成本的粗略估算。实际交易中还会遇到涨停买不进、跌停卖不出的情况但个人级别回测先不引入太复杂的约束重点是评估策略本身的趋势捕捉能力。回测结果我会输出总收益率、年化收益、最大回撤、交易次数这几个核心指标用来判断策略是否值得进一步优化。6. 常见问题与排查技巧实录6.1 数据源相关接口失效、字段变化、限流用 akshare 这类爬虫类数据源最怕就是接口突然失效。我遇到过几次点开网页看源代码发现目标网站把字段从日期改成了date或者新增了一列没用的数据。我的应对思路是不在抓取层做太多假设数据进来以后统一通过列名映射清洗列名对不上就报警告绝不静默入库。同时给同步脚本加了重试机制遇到网络超时或返回空数据等待30秒重试三次实测可以解决大部分临时性限流。6.2 与行情软件对不上除权除息和复权锚点问题这是最容易让人怀疑人生的坑。同一只股票用 OpenStock 算出来的 MACD 和同花顺显示的不一致先别怀疑计算逻辑大概率是复权方式不一致。我默认选择了前复权但前复权的价格锚点是最新价历史区间的价格会随着每次除权而改变。这意味着如果你用了区间数据做回测每次有新的除权事件发生历史价格就会整体变化策略结果也会变。解决方式是历史回测统一用后复权这样历史价格固定策略可复现性更强实时看盘用前复权贴合日常使用习惯。6.3 数据库和查询性能数据多了怎么办当我把数据量扩展到全市场后SQLite 的瓶颈开始显现。一开始没有在stock_kline表上建索引查询某一年的数据要扫全表慢得让人崩溃。后来我补了(code, date)联合索引单股查询迅速降到毫秒级。另外SQLite 的写入锁是全局的如果同步脚本和 Web 服务同时访问数据库偶尔会报database is locked。我的处理办法是同步脚本用 WAL 模式连接数据库Web 服务只读有效减少锁冲突。问题现象解决方案数据源接口失效返回空表或列名变化升级akshare版本列名映射做防御性处理指标值对不上MACD/RSI与行情软件不一致检查EMA的adjust参数改用Wilder平滑除权除息跳变价格和成交量出现断崖回测用后复权看盘用前复权查询缓慢接口响应超过2秒建联合索引避免全表扫描database is locked同步时Web偶发报错开启SQLite WAL模式读写分离回测结果失真收益虚高检查是否用了shift(1)计入手续费滑点6.4 部署与上线Docker Compose 一键启动为了部署方便我把 OpenStock 打包成了 Docker 服务。后端用 FastAPI Uvicorn数据库挂载到宿主机数据卷页面沿用单页HTML。这样在任何一台有 Docker 的机器上一条命令就能启动整个系统。version: 3.8 services: web: build: . ports: - 8000:8000 volumes: - ./data:/app/data environment: - DATABASE_PATH/app/data/openstock.db restart: unless-stopped部署时最容易踩的坑是端口冲突。如果机器上已经跑着别的服务占用了8000端口容器会启动失败排查时先docker compose logs -f查看日志再用lsof -i :8000检查端口占用。另一个坑是数据卷权限容器内运行的进程可能需要写宿主机挂载的目录如果宿主机目录权限不对数据库文件写入会报错直接chmod -R 755挂载目录即可解决。6.5 回测陷阱未来函数、幸存者偏差、过拟合回测系统上线后最大的陷阱不是代码 bug而是数据偏差。未来函数是最需要注意的我刚开始写回测时忘记shift(1)导致策略收益虚高差点以为自己找到了圣杯。后来每次写策略都会专门检查这个信号用的是哪一天的数据这个价格是当天的收盘价还是次日的开盘价只要信号和成交价格有任何一天重叠结果就会有水分。幸存者偏差是另一个隐蔽问题。如果你只选当前还在上市的股票做回测那些退市的股票就已经被剔除了策略实际表现会比回测结果差很多。处理方案是回测时使用历史时点的成分股列表或者至少纳入退市股票的数据才更接近真实情况。至于过拟合我的经验是参数越复杂的策略越容易过拟合双均线的参数选择20和60还是5和20差异并不大但如果同时优化十几个参数回测结果往往不可靠实盘时大概率翻车。7. 后续还能如何扩展 OpenStock搭建完成之后OpenStock 的框架已经可以用但离一个完整的产品还有一段距离。我个人下一步最想做三件事第一是增加定时任务每天盘后自动同步数据并通过邮件或 Server 酱发送当日的策略金叉死叉信号第二是加入分钟级K线支持让短周期策略也能跑起来但这就需要在数据库字段和存储容量上做一些调整第三是增加策略库管理把不同策略的代码注册成模块可以随时对比多个策略在同一段时间的表现。另外前端展示也可以继续丰富比如加入资金曲线、回撤区域、个股财务数据面板等。不过这些都属于锦上添花核心的数据链路和回测框架要足够可靠才不会在后续迭代时需要返工。搭建 OpenStock 这套系统对我最大的启发是技术指标、回测框架这些代码难题反而是最好解决的真正需要花心思的是数据理解和策略验证的思路。你在抄作业的过程中如果遇到什么问题欢迎在评论区留言我会尽量给出具体的排查建议。
