如何为市场数据搭建增量更新工作流-迪斯科星球

如何为市场数据搭建增量更新工作流

【免费下载链接】machine-learning-for-tradingCode for Machine Learning for Trading, 3rd edition — from data sourcing to live execution.项目地址: https://gitcode.com/GitHub_Trending/ma/machine-learning-for-trading

在 machine-learning-for-trading（ML4T，第 3 版）仓库里做完第一次数据下载后，市场数据就会开始过期。全量刷新会重新拉取某个标的从上市以来的每一行数据，而增量更新只拉取最后一次存储时间之后的交易日——在几百个标的、十年历史的量级上，前者每天是百万行，后者只有几百行。这篇文章覆盖的是「初始加载 → 每日增量更新 → 完整性校验」这条连续路径：先用免费数据源完成初始加载，再用仓库自带的download_all.py --update或 ETF 单数据集脚本把数据延伸到今天，最后用GapDetector和OHLCVValidator确认更新没有留下空洞，并把整个流程挂到 cron 上。

准备条件

环境已就绪：Docker 镜像ml4t或本地uv环境二选一，且verify_installation.py全部 PASS（两种路径见安装指南）。
仓库根目录下执行cp .env.example .env，默认值即可开始；本场景用到的免费数据源（Yahoo Finance）不需要 API key。
数据路径遵循固定优先级：CLI 参数--data-path> 环境变量ML4T_DATA_PATH> 仓库自带的data/目录（见 utils/downloading.py 中的resolve_data_dir）。数据默认落在<repo>/data/，所有命令都要从仓库根目录执行。
增量更新需要访问 Yahoo Finance 的实时网络。

本地uv路径下命令带uv run前缀；Docker 路径在 Jupyter Lab 终端（File → New → Terminal）里直接去掉uv run执行同样的命令。

第一步：初始加载（全量历史）

增量更新的前提是存储里已经有历史数据。最小路径是只加载 ETF 数据集（Yahoo Finance，无需 API key，约 30 秒）：

uv run python data/etfs/market/download.py

该脚本读取 data/etfs/market/config.yaml，其中声明provider: yahoo、start: '2006-01-01'、end: '2025-12-31'、frequency: daily和 100 个 ETF 分组。执行结束时打印一段 SUMMARY，其中的action是判断依据：首次运行显示downloaded，之后的运行显示updated，同时给出rows、symbols和date_range。

如果后续章节需要更多数据集，用批量入口一次性下载免费数据（约 75 MB，跳过 4 GB 的 firm-characteristics 数据集）：

uv run python data/download_all.py --free-only --skip-firm-characteristics

加载成功与否可以直接用仓库的 loader 验证——所有 loader 返回 Polars DataFrame，且列名统一为symbol和timestamp（见 data/README.md）：

from data import load_etfs df = load_etfs(symbols=["SPY", "QQQ"], start_date="2020-01-01") df["timestamp"].max()

如果数据缺失，loader 会抛出DataNotFoundError并附带对应的下载命令。

第二步：增量更新（每日 delta）

批量入口：`download_all.py --update`

这是书中定义的「生产接口」，同一条命令适用于 notebook、CI 和 cron：

uv run python data/download_all.py --update

它把所有可更新的数据集从各自配置的end日期（例如 ETF 配置的2025-12-31）直接延伸到当天，不需要修改任何配置文件。运行前会先打印本次将更新的 7 个数据集：ETF Universe、Crypto Premium、Macro（FRED）、FX Pairs、Fama-French Factors、AQR Factors、CFTC Commitment of Traders；同时声明被跳过的冻结数据集——US Equities（数据止于 2018 年）和 AlgoSeek（授权快照）。

注意这是批量入口：一次运行会更新全部 7 个可更新数据集，而不是某一个。只想更新 ETF 一个数据集时，用下面的单数据集命令。

运行结束的验证方式是UPDATE SUMMARY：每个数据集一行[OK]或[FAIL]，最后给出Updated: N/7 datasets。进程只在全部成功时以 0 退出，任何一个失败就非零退出——对 CI 来说这比读日志可靠。日志无法取回的场景（例如 CI job）可以加参数把结果落盘：

uv run python data/download_all.py --update --report /tmp/ml4t-update.json

写出的 JSON 包含每个数据集的成功状态、失败名单和completed/total计数。若部分失败，脚本会提示可能缺少的 key：FRED_API_KEY（Macro 必需）和OANDA_API_KEY（FX 用，可选，缺省时回退 Yahoo）。

单数据集入口：ETF 下载脚本的`--update`

uv run python data/etfs/market/download.py --update

--update把配置的end日期改为当天再执行增量追加。该脚本内部逻辑是：输出文件etf_universe.parquet已存在且未指定--force时走manager.update()（增量），否则走全量download_all(force=...)。所以即使不加--update，重复运行该脚本本身也是增量的——区别只是要不要把end日期推到今天。

API 级更新与两个坑

第 2 章 notebook 19_incremental_updates 用DataManager展示了同一套机制的单标的版本，也解释了为什么不能盲目调用DataManager.update()。以下代码取自该 notebook（从仓库根目录执行，demo_storage是本例的本地存储目录）：

from ml4t.data import DataManager from ml4t.data.storage import HiveStorage from ml4t.data.storage.backend import StorageConfig from utils.downloading import update_through_last_complete_bar config = StorageConfig(base_path="demo_storage/updates_demo", compression="zstd") storage = HiveStorage(config=config) dm = DataManager(storage=storage) # 初始加载：2023 全年到 2024 年底 dm.load("AAPL", "2023-01-01", "2024-12-31", provider="yahoo") # 增量更新：只取新数据 + 7 天重叠 rows = update_through_last_complete_bar( dm, storage, "AAPL", provider="yahoo", lookback_days=7 ) print(rows)

更新读取存储中的最后时间戳，重新拉取lookback_days（这里 7 天）的重叠窗口并按timestamp去重合并，因此供应商事后修订过的 bar 会替换存储中的旧行，而不是重复。notebook 中 AAPL 初始加载为 502 行（文档示例数值，你的运行日期不同则行数不同），更新后行数等于 502 加上新增交易日——没有合成出来的周末/假日行。

为什么用update_through_last_complete_bar而不是dm.update()：

当前交易日的占位行。DataManager.update()取数到datetime.now()，每天都会向 Yahoo 索要「正在进行中」的那个交易日。Yahoo 会把当前交易所日期发布成一行只有累计成交量、open/high/low/close 为空的数据，provider 直接拒绝：DataValidationError: yahoo: Column 'open' contains 1 null values。update_through_last_complete_bar改用的做法是「寻找窗口终点而不是计算窗口终点」：从当前交易所日期（按America/New_York时区，不是 UTC）的前一天开始，只要 provider 拒绝该窗口就往前退一天，最多退 5 天（覆盖长周末加假日）。如果日志里出现一行Failed to fetch AAPL: yahoo: Column 'open' contains 1 null values而后一行打印了行数，那是退让逻辑在工作，不是失败。
fill_gaps=True默认值对 OHLCV 不安全。盲目调用update()时，默认的日历无感知 gap 检测会把每个周末和美国假日都当成「缺失的交易日」并前向填充，每个标的会多出几百根幻影 bar。OHLCV 数据的正确做法是保留非交易日的缺失，用日历感知的完整性检查在下游验证——就是下一步要做的GapDetector。

也可以直接跑整个 notebook 看端到端演示（会实时访问 Yahoo Finance，并在开头清空自己的 demo 存储目录DEMO_DIR以保证可复现，注意不要在目录里存放其他数据）：

uv run python 02_financial_data_universe/19_incremental_updates.py

ml4t-data 底层支持四种更新策略，选型参考 notebook 给出的对照表：INCREMENTAL（取最后存储时间戳之后的数据，每日更新默认，最快）、APPEND_ONLY（只加不改，适合审计档案）、FULL_REFRESH（重新下载并替换全部数据，用于数据损坏后的恢复）、BACKFILL（补齐既有范围内缺失的时段）。大多数场景用DataManager.update()走的INCREMENTAL就够，FULL_REFRESH是恢复路径——验证发现回归时用，不要试图就地修补损坏的 parquet。

第三步：完整性与数据健康验证

更新跑完后，在跑任何回测之前验证存储是否完整。第 2 章 notebook 的做法是对每个标的做 gap 检测：

from ml4t.data.update_manager import GapDetector gap_detector = GapDetector(exclude_weekends=True) df = storage.read("equities/daily/AAPL").collect() gaps = gap_detector.detect_gaps(df, frequency="daily")

输出要么是complete (no gaps)，要么是逐条 gap（起止日期加天数）。必须知道一个限制：GapDetector没有交易所日历，exclude_weekends=True只过滤周六周日，美国市场假日（MLK Day、Good Friday、Thanksgiving 等）仍会各登记为一个 1 天的 gap。对例行更新健康检查这没问题；但在把数据用于回测面板之前，要配合日历感知的完整性检查。

notebook 第 4 节给出了单标的健康报告的完整写法，每个标的产出一行：存储行数、最后日期、距今天数（days_stale）、gap 数、验证问题数，并用阈值把状态标为fresh或stale：

from datetime import datetime from ml4t.data.update_manager import GapDetector from ml4t.data.validation import OHLCVValidator FRESH_DAYS = 3 # 落后不超过 3 天算新鲜 STALE_DAYS = 7 # 超过 7 天算失效 validator = OHLCVValidator(max_return_threshold=0.5) detector = GapDetector(exclude_weekends=True) df = storage.read("equities/daily/AAPL").collect() last_date = df["timestamp"].max().replace(tzinfo=None) days_stale = (datetime.now() - last_date).days gaps = detector.detect_gaps(df, frequency="daily") result = validator.validate(df) print({ "status": "stale" if days_stale > STALE_DAYS else "fresh", "rows": len(df), "last_date": last_date.date(), "days_stale": days_stale, "gaps": len(gaps), "issues": result.error_count if not result.passed else 0, })

这两个阈值是 notebook 声明的判断参数：日频股票数据落后超过一个周末算滞后，超过一周算失效。OHLCVValidator的max_return_threshold=0.5是极端收益截断阈值，建议对每次加载都做验证（见 13_data_quality_framework）。

第四步：把更新挂到计划任务上

notebook 的 Production Checklist 给出四步：初始下载（一次性，约 10 分钟）、把download_all.py --update加入 cron（书中建议每天 6 PM）、回测前检查新鲜度/gap/验证结果、每次加载用OHLCVValidator验证。按这个清单，一个示例 cron 条目（/path/to/machine-learning-for-trading换成你的克隆位置）：

0 18 * * * cd /path/to/machine-learning-for-trading && uv run python data/download_all.py --update --report /tmp/ml4t-update.json

这个组合就是文档强调的「cron + 幂等 CLI」：同一条命令在 notebook、CI 和定时任务里行为一致，重复运行不会重复拉取历史，--report让无法读日志的调用方也能拿到每个数据集的结果。

边界与已知限制

--update不碰冻结数据集。US Equities 止于 2018 年、AlgoSeek 是授权快照，再频繁运行更新它们也不会变新。
失败定位看退出码和 FAIL 名单。单源故障（供应商宕机）和整体故障（网络出口问题）在日志缺失时外观相同，--report写出的 JSON 就是用来区分两者的。
lookback_days是运行参数而非标的列表。notebook 的结论是把它一次性设到能覆盖供应商的修订窗口即可（文档给出 Yahoo 约回溯调整 5 个交易日、CRSP 约 30 个），策略就能跨品种通用。
gap 检测针对的是存储数据而不是数据流。每天收尾时跑一遍detect_gaps()，让漏掉的交易日在任何回测读到过期分区之前被发现。
更新模式不修改配置。标的列表 YAML 用「今天」作默认 end，--update把数据延伸过配置里的end字段而不需要每年改文件。

下一步

单标的的DataManager批量接口（fetch、batch、Universe、storage）在 18_data_management 中展开；更新隐含的 parquet 写路径性能对比见 20_storage_benchmark_file 和 21_storage_benchmark_database（后者需要docker compose --profile benchmark up -d起数据库服务）。数据侧的完整目录结构、loader 列表和 API key 说明在 data/README.md。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

企业官网建设流程全解析