行情软件导不出数据、现成量化平台策略代码改不动、开源项目东拼西凑还要自己缝缝补补……这套“想验证一个选股想法就得手动翻K线”的流程我忍了很多年。OpenStock 就是在这个背景下开始折腾的它是一套开源股票数据分析系统你可以把它简单理解成“数据采集 指标计算 策略回测 可视化看板”的一体化工具箱搭好之后输入股票代码就能拉数据、算指标、跑策略、出报告。这篇文章会把我在搭建 OpenStock 过程中的选型思路、踩坑记录和最终落地结构完整写出来适合想入门量化分析、想摆脱手工复盘、或者正在自建投研工具的开发者参考照着抄即可跑通一版可用的系统。1. 为什么自己折腾 OpenStock现有工具到底缺在哪1.1 行情软件和现成平台都有的三个痛点先说行情软件。绝大多数行情软件都把“看盘”和“导出数据”当成两件事更准确地说导出数据这件事被刻意做得很繁琐。日线数据手动导出到 Excel 还要选格式、选周期、选复权方式一次只能导一只股票导完之后字段命名还不统一。当你想批量化验证一个“市值小于某个阈值且近期放量”的选股条件靠行情软件逐个翻股票基本不现实。现成的量化平台则是另一个极端。策略代码必须跑在别人的服务器上平台升级导致接口变动、函数废弃是家常便饭本地数据无法导出策略逻辑被锁定在特定框架里。更麻烦的是黑盒回测部分平台的撮合逻辑不透明手续费和滑点设置粗糙回测结果和实盘之间的差距大到让人怀疑人生。OpenStock 的出发点就是把数据掌握在自己手里。行情数据落在本地数据库策略逻辑用标准 Python 写回测引擎自己控制撮合细节指标计算可以随时改参数重跑。整个过程可追溯、可复现、可二次开发。1.2 五层结构OpenStock 的整体架构OpenStock 不是一个大而全的单一程序而是拆成五个清晰的模块后续维护和替换成本会低很多模块职责落地方式采集层从公开数据源拉取行情、财务数据Python 定时任务存储层保存原始数据和计算中间结果SQLite / PostgreSQL计算层指标计算、选股信号生成pandas 向量化计算回测层策略回测、绩效统计自研事件驱动引擎展示层数据看板、策略结果可视化Streamlit这五层之间通过数据库表结构解耦。采集层只负责写库计算层只读库并写计算结果到新表回测层只依赖计算层产出的信号表展示层则完全不关心数据从哪来。好处很明显某一天想换数据源只改采集层想换回测框架只动回测层其余部分基本不受影响。1.3 明确边界OpenStock 不做什么在自己动手之前必须划清边界。OpenStock 的目标不是做一个全自动交易系统它不做实时 tick 级数据推送不做自动下单也不会给你推荐“明天买什么”。它的工作边界是把数据整理干净、把分析逻辑跑通、把策略绩效算准确最后的决策还是人来做。这一点想清楚之后整个项目就不会跑偏。很多人一开始就想接入实盘、接入券商接口、做分钟级策略结果数据、撮合、风控、运维全堆在一起项目很快烂尾。从日线级别的分析系统起步是性价比最高的路径。2. 搭建前的技术选型数据源、数据库和项目布局2.1 数据源怎么选免费接口与商业数据服务对比搭建系统的第一步是选数据源这一步决定了后续数据质量和维护成本。我对比过几类常见方案数据源类型优点缺点适用场景免费公开接口如 AkShare零成本数据覆盖面广社区活跃接口偶尔变动数据口径需自行核验个人学习、策略验证商业数据服务如 Tushare Pro 积分接口稳定性好字段规范有维护保障需要积分或付费部分高频字段有门槛长期维护、工作级使用自研爬虫完全可控可定制字段开发量大反爬风险高维护成本极高不建议作为首选我个人推荐“免费接口起步 商业接口兜底”的组合。先用 AkShare 把整个流程跑通积累一段时间后发现数据缺失或者字段不满足需求再平滑切换到商业数据源因为 OpenStock 的采集层做了统一的数据结构封装换数据源只需要重写一个 adapter。很多人会忽略一个关键问题数据源的字段口径必须提前确认。比如“涨跌幅”是相对昨收还是今开“成交量”是什么单位复权因子是否提供这些细节到写回测引擎时会直接影响结果宁可一开始多花时间看数据文档也不要等回测完才发现数据口径错误。2.2 数据库选择从 SQLite 起步按需迁移存储层我建议从 SQLite 开始不建议一上来就上 PostgreSQL。原因很简单个人分析场景的数据量通常在百万到千万行级别SQLite 单文件、零配置、备份方便完全够用。当你发现写入并发成为瓶颈、或者需要多客户端同时访问时再迁移 PostgreSQL 也不迟。OpenStock 的数据库设计分四张核心表stock_basic股票基础信息表包含代码、名称、上市日期、退市日期、行业。daily_price日线行情表包含交易日期、开高低收、成交量、成交额、复权因子。daily_indicator指标结果表保存计算好的 MA、MACD、RSI 等指标值。strategy_signal策略信号表包含股票代码、信号日期、信号方向、策略名称。字段命名统一使用小写下划线风格日期字段统一使用YYYY-MM-DD文本格式。有人会用时间戳或者整数格式存储日期回测排序和区间筛选时还要来回转换属于自找麻烦。2.3 目录结构与依赖管理OpenStock 采用模块化目录布局每个人都可以按自己习惯微调但核心逻辑是“入口脚本只负责启动业务逻辑全部分布到模块”openstock/ ├── collector/ # 数据采集模块 │ ├── base.py # 数据源抽象类 │ └── ak_adapter.py # AkShare 适配器 ├── storage/ # 数据库操作封装 │ ├── database.py # 连接与建表 │ └── repository.py # 读写服务 ├── analysis/ # 指标计算与选股 │ ├── indicators.py # 技术指标 │ └── signals.py # 选股信号 ├── backtest/ # 回测模块 │ ├── engine.py # 回测引擎 │ └── metrics.py # 绩效指标 ├── web/ # 可视化层 │ └── app.py # Streamlit 应用 ├── tasks/ # 定时任务 ├── requirements.txt └── config.py # 全局配置依赖管理方面我推荐使用poetry或者uv避免直接用pip install装一堆包到全局环境。pandas、numpy、akshare、streamlit、plotly、sqlalchemy 是核心依赖建议在requirements.txt中锁版本。尤其是 akshare 这类更新频繁的库锁版本可以避免某天接口变动导致整个采集任务崩溃。3. 数据采集与清洗决定系统成败的地基3.1 初始化全量数据与每日增量同步数据采集分两个阶段全量初始化和增量更新。初始化阶段拉取指定股票列表的历史日线数据比如过去五年的日线增量阶段则每天收盘后拉取最新的行情并追加写入。全量初始化要考虑接口限流。很多免费接口对单次请求量有隐性限制一次性拉取五千只股票很容易触发限制。我的做法是先按股票代码分批每批 200 只批次之间暂停 1 到 2 秒同时记录每一只股票的拉取状态失败的自动重试重试三次仍失败的写入日志文件等全部跑完再统一排查。增量更新的核心是“只拉最新数据”避免重复请求。实现上可以读取数据库里每只股票的最大交易日期只拉取该日期之后的数据再配合INSERT OR REPLACE写入天然保证幂等性。这样即使某天定时任务重复执行也不会产生脏数据。def sync_daily_price(stock_code: str, start_date: str | None None): latest repo.get_latest_trade_date(stock_code) if latest and not start_date: start_date latest # 增量更新起点 df data_source.fetch_daily(stock_code, start_date) cleaned clean_daily_data(df) repo.upsert_daily_price(stock_code, cleaned)3.2 复权因子前复权、后复权与不复权的正确用法这是整个项目里最容易踩坑的地方值得单独拿出来说。股票分红送转后价格会跳空比如一只 100 元的股票实施 10 送 10除权日价格直接变成 50 元。如果不做复权处理技术指标会在除权日出现虚假的暴跌信号回测也会得出完全错误的结果。不复权数据真实的历史成交价格适合计算真实涨跌幅和成交量。前复权数据以当前价格为基准调整历史价格适合看 K 线图和技术指标形态。后复权数据以最早价格为基准调整后续价格适合长期收益计算。OpenStock 的做法是原始行情表里统一保存“不复权价格 复权因子”计算指标时按需将价格转换为前复权或后复权。复权因子一般数据源会直接提供如果没有可以根据除权除息公告自行计算。注意同一只股票在数据库里永远只保存一份原始数据和一份复权因子不要分别存储前复权和后复权两份表。后复权价格会随新除权事件变化前复权价格也会随最新价格变化如果直接存复权价格历史数据会持续变动极难排查。3.3 财务数据对齐与缺失值处理财务数据比行情数据更复杂因为财报的发布日期和报告期不是一回事。比如一家公司 2024 年年报可能在 2025 年 4 月才发布如果你直接用报告期去对齐交易数据就会犯“未来函数”的错误——在 2025 年 1 月的回测里用了 4 月才发布的财报数据。解决思路是维护一个“公告日期announcement_date”字段策略里使用财务数据时必须按公告日期进行asof对齐也就是只能使用“这一交易日之前已经公告”的数据。我在分析层封装了一个merge_with_financial(stock_code, trade_date)函数内部通过公告日期与交易日期匹配确保不会前视。缺失值的处理也有一套固定原则。停牌导致的缺失行情保留空行但不填充因为停牌期间价格没有变化回测时这笔持仓在停牌期内不能卖出这是重要的流动性约束。指标计算中出现的 NaN前段缺失不填充中间缺失用前值填充尾部缺失直接丢弃。这套原则要写进采集清洗模块统一执行避免不同模块各填各的。4. 技术指标与选股信号把分析逻辑变成代码4.1 常见技术指标的计算实现技术指标在 OpenStock 中统一采用 pandas 向量化计算尽量避免用 for 循环逐行处理。以 MACD 为例核心计算就是 EMA指数移动平均的差值def calc_macd(close: pd.Series, fast: int 12, slow: int 26, signal: int 9) - pd.DataFrame: ema_fast close.ewm(spanfast, adjustFalse).mean() ema_slow close.ewm(spanslow, adjustFalse).mean() dif ema_fast - ema_slow dea dif.ewm(spansignal, adjustFalse).mean() hist (dif - dea) * 2 return pd.DataFrame({dif: dif, dea: dea, macd_hist: hist})其他指标如 RSI相对强弱指标、KDJ随机指标也都是几行 pandas 代码的事关键是参数要可以配置。OpenStock 里把指标参数集中放在config.py的字典里不同策略可以引用不同参数组合。比如短中期趋势策略用MACD(12, 26, 9)长线趋势策略要用MACD(26, 52, 15)参数和策略代码分离改参数无需改逻辑。4.2 未来函数陷阱为什么回测业绩总是虚高实盘交易中当根 K 线收盘后你才知道收盘价但很多指标计算习惯会把“包含当前未完成 K 线的数据”纳入计算。还有一类问题是信号发出当天就以收盘价成交这其实已经隐含了“你能在收盘那一刻立即下单”的假设。OpenStock 处理信号的原则是使用第 T 日及之前的数据计算信号最早在第 T1 日开盘价成交。这虽然让回测结果稍微保守但更贴近真实交易。我在信号表里同时记录signal_date和trade_date两个字段本质上就是“信号日”和“可执行日”分离。未来函数最隐蔽的形式出现在财务数据和停牌处理上。财务数据的公告日期陷阱前面已经说过停牌处理也一样如果信号表没有剔除停牌日策略可能在停牌期间“偷偷”成交这显然不符合实际。所以回测引擎处理订单之前会统一检查目标股票在订单执行日是否可交易。4.3 一个可运行的选股信号组合搭建完成指标层之后我写了一个非常基础的选股信号作为验证当 20 日均线向上穿越 60 日均线金叉且成交量是前 5 日均量的 1.5 倍以上时产生买入信号当 5 日均线下穿 20 日均线时产生卖出信号。这个信号不复杂但足以验证整条链路。它需要的数据很简单日线行情的收盘价和成交量。回测跑完之后再逐步叠加条件比如增加 RSI 过滤、市盈率过滤、行业过滤等。这样一点点把逻辑加进去每一步都能精确知道是哪个条件提升了收益还是损害了收益。def generate_ma_signal(df: pd.DataFrame) - pd.Series: ma5 df[close].rolling(5).mean() ma20 df[close].rolling(20).mean() ma60 df[close].rolling(60).mean() volume_ma5 df[volume].rolling(5).mean() buy (ma20 ma60) (ma20.shift(1) ma60.shift(1)) (df[volume] 1.5 * volume_ma5) sell (ma5 ma20) (ma5.shift(1) ma20.shift(1)) return pd.Series(np.where(buy, 1, np.where(sell, -1, 0)), indexdf.index)5. 回测引擎与可视化看板验证策略是否真的有效5.1 最小回测引擎信号、持仓与撮合自研回测引擎的核心是处理信号到实际持仓的变化。OpenStock 的回测引擎按“日级别 bar”推进在每天开盘前处理信号和订单在每天收盘后计算持仓市值和盈亏。同样是第 T 日产生的信号第 T1 日开盘价撮合撮合时扣除手续费和滑点。引擎内部维护一个Position对象记录每只股票的持仓数量和可用现金。收到买入信号后按仓位管理规则计算可买股数这里默认使用“单只股票不超过总资产 20%”的仓位限制收到卖出信号则按全部持仓卖出。回测结束时输出每日资产净值序列 DataFrame交给绩效模块计算指标。整个过程没有引入复杂的撮合队列因为日线级别策略不需要逐笔撮合只要约束好开盘价成交、停牌不能交易、涨跌停不能买入这三条规则就已经比大多数傻瓜式回测工具严谨很多。5.2 绩效指标怎么算年化、夏普和最大回撤绩效指标计算是回测结果可信度的核心OpenStock 的metrics.py集中实现了几个常用指标指标计算公式要点说明累计收益率期末净值 / 期初净值 - 1最直观的收益表现年化收益率(1 累计收益率)^(250/交易天数) - 1用 250 个交易日近似一年最大回撤max(1 - 净值/历史最高净值)衡量最坏情况下的亏损幅度夏普比率(年化收益率 - 无风险利率) / 年化波动率单位风险对应的超额收益最大回撤的算法需要特别注意用“历史最高点”和“当前净值”的比值去计算峰谷落差按日滚动更新。很多新手直接用(cummax - cummax.shift())之类的简化方式会漏掉回撤区间内的最新低点导致最大回撤被低估。5.3 用 Streamlit 搭建本地看板看板选 Streamlit 是性价比最高的方案它不需要写前端用纯 Python 就能生成交互页面。我将看板拆成三个页签行情总览、股票详情、策略回测。行情总览页用一个数据表展示所有关注股票的最新价格、涨跌幅和当日成交额支持排序和筛选股票详情页输入股票代码后展示 K 线图、成交量副图、MACD 指标叠加以及最近一周的财务摘要策略回测页选择策略和股票池点击运行后展示资金曲线、回撤曲线和绩效指标表。import streamlit as st import plotly.graph_objects as go def render_stock_detail(stock_code: str): df repo.get_daily_with_indicator(stock_code) fig go.Figure(data[go.Candlestick( xdf[trade_date], opendf[open], highdf[high], lowdf[low], closedf[close], nameK线)]) st.plotly_chart(fig, use_container_widthTrue)看板本身不承担数据分析工作只是把数据库里的结果渲染出来。这样设计的好处是策略在后台跑批更新数据看板打开就是最新结果不需要在交互页面里做繁重计算响应速度很快。6. 上线运行后的坑与优化来自实战的教训6.1 除权除息导致的历史数据“失真”系统跑了一个多月后我发现一个奇怪的现象某只股票的历史 K 线图上三年多以前出现了一个从未见过的巨大跌幅单日下跌接近 50%但当天并没有任何利空消息。排查了半天问题出在除权除息处理上——这只股票在几天前实施了大比例送转而我只更新了最新数据没有重算历史复权因子导致前复权价格整体变化后历史 K 线和指标全部错位。这类问题的根源是复权因子变更后没有触发历史数据重算。OpStock 的解决方案是每次增量更新时检测该股票的复权因子是否发生变化如果变化则重新拉取并覆盖该股票全部历史行情。说白了除权除息不是简单的新增一行数据而是一次针对该股票的全量修正。6.2 查询性能优化索引、缓存和分区当数据库积累到三年以上数据、几千只股票之后不同股票的指标计算和查询开始变慢。最初的慢是由于指标计算层每次从数据库读取某只股票的三年日线数据再做滚动窗口计算。读了几百次之后I/O 瓶颈明显。优化的第一步是加索引。daily_price表在(stock_code, trade_date)字段上建联合唯一索引strategy_signal表在(strategy_name, signal_date)上建普通索引。这一步让按股票代码和日期区间的查询从全表扫描变成索引查询速度提升非常明显。第二步是引入结果缓存。技术指标的计算结果在数据没有更新之前是确定的所以我将指标计算的结果存进daily_indicator表计算时先检查这张表是否已有对应股票和日期范围的数据避免重复计算。Streamlit 看板也加了st.cache_data缓存相同参数查询直接走缓存不再查数据库。6.3 定时任务与数据质量监控数据采集不能只靠手动执行。我用系统的定时任务来触发每日数据更新每个交易日收盘后 1 小时执行增量同步之后计算指标和信号最后重跑一遍重点股票池的回测。整体耗时控制在几分钟内全部跑完之后发送一份任务摘要和日志到本地文件。定时任务运行稳定只是第一步数据质量监控同样不能缺。OpenStock 里有一个简单的数据质量检查脚本每天运行任务后自动检查各股票最新交易日期是否达到预期、近五日行情是否有异常空值、指标计算结果是否存在无穷值、回测净值曲线是否出现异常跳变。一旦发现问题会输出告警信息到固定的alerts.log文件。我个人使用中发现行情数据源偶发返回空数据是最常见的故障。某一天任务执行成功但同步数量为 0脚本不会报错数据却悄悄缺失了。后来我在数据质量检查里增加了“当日同步股票数量”的阈值校验不同步数量低于某个范围就标记为异常这个问题才算真正被监控住。6.4 维护成本与后续扩展方向OpenStock 跑通之后日常维护成本其实很低。数据源是免费接口数据库是单文件 SQLite看板在本地运行整个系统不依赖任何外部服务唯一要关注的就是接口变动和数据质量。对我个人而言它确实解决了“数据在手、策略可验”的核心问题省下了大量手工复盘的时间。如果你也想自己搭一套我的建议是不要一开始就追求功能全面。先跑通日线数据采集和回测再逐步加财务数据、加指标、加看板。每个模块独立验证后再接入总体流程排错的时候会轻松很多。后续我还打算在 OpenStock 里加入基本面因子模块和定期自动生成分析报告的功能让数据从“能看”进一步变成“能用”。如果你也有类似需求不妨从今天的数据采集层开始先搭一个最小闭环出来剩下的交给持续迭代。