learn/量化工程优化/2026-05-26-数据缓存分离工程优化/README.md

📄 文档learn/量化工程优化/2026-05-26-数据缓存分离工程优化/README.md

🌱 创建于2026-07-08 02:17

✏️ 修改于2026-07-08 02:17

👁️ 浏览量—

🔗 短链接/~learn/s/ece383d9

🔐 可见性

数据缓存分离工程优化

文档目标

本文档涵盖需求背景、设计方案、架构图和实现细节，一次性回答「做了什么、为什么、怎么做的」。

一句话结论

将数据拉取和策略推送拆分为两个独立进程，推送时直接读本地 CSV 缓存而不再实时调 API，从根本上消除「API 不稳定导致推送失败」的系统性风险。

改造前后对比

Mermaid Live View

Mermaid Live Edit: https://mermaid.live/edit#pako:eNp9k11rE0EUhv_KMFcKTaFtopILYdOmIChKKl7oiozZSbawH2E_FGkKTcEmMdmmommwWm3BtrnQraKWagv5L5Kd3Vz1L3imY6KxqXN1duY57znzntkFnDUVipM4p5mPsyqxHHQ9IxsIlu0-zFukoKJUevZmJn1Pxuzl997SVlD1To43Aq8ZdV6H7VrwtM2-lSCQ8X2RyFdqAvjA32Ctg2hvN2g8h1MUi11FqUk4yFEnqz5QiEMuXESB_xYoVqsGjfVhjUmeUZRuXUPdQy9sfwG9IkpN8U5O6eD95-jrzqgcVlkLnr0DOA5w-HE9bO5E_nbot4bhKdFSAqCfm3XEVtu9pdIo1bgAL_HSBxVW2hfoAKKGIhv_2CbN3k5n_ri2tspda9R_u1ZZCXePhopIE31nqIWEdQNTTstL3LoZMG2aAHODGCRPrXGbPKJoeu7OsBY3qeDaf0n1GxZS8VFSmkmUs1Li7lLif05KCQGda5C4IIqNi4Gy-nL3x0oRSZchodvZDJuvWGsPSp8c17uHNfZhu1f2ov3mmU54PmBRp8zewHylK7xg9YXtEI2yrXLkf-IKR57o8fwpiTctmuZzwmNYp5ZO5hX4FxY4I2NHpTqVcRJCheaIqzkylo1FQInrmHNPjCxOOpZLx7BbgLdMZ-YJTF7HyRzRbNi1TDevDr4KxLhrmno_JW_xUjxe_AXJE2kB

flowchart LR subgraph BEFORE["改造前：单进程全流程"] B1["定时触发"] --> B2["fetch_data() 实时拉取"] B2 -->|API 不稳定| B3["拉取失败"] B2 -->|成功| B4["策略计算"] B3 --> B5["推送失败"] B4 --> B6["消息推送"] end subgraph AFTER["改造后：双进程分离"] A1["fetcher 定时拉取"] --> A2["DataCacheManager.save CSV"] A3["pusher 定时推送"] --> A4["DataCacheManager.load CSV"] A4 --> A5["策略计算"] A5 --> A6["消息推送"] A1 -.->|API挂了| A7["保留旧CSV，下次重试"] A4 -.->|CSV过期| A8["打stale标记，仍计算推送"] end BEFORE --> AFTER

架构总览

Mermaid Live View

Mermaid Live Edit: https://mermaid.live/edit#pako:eNqNU0Fr2zAY_SvCpxTaFPvowyA4KRtZmZmyHTYPo9qyHWbLQZbbhSTQ3kphZZe0sB22MQpjsKSH0cNG92_iZP9in-w2jcIo88V6fu_Tp_f0eaB5qU81Uwvi9MCLCBeo03QYgifL90JOehGy9ZeOZvPUo1mGdBPxnLk-EcQNqPAiyuu9vqO9qorkg6U-A8bPY4pmk_fF-dXs97f56VSR7UhZuYUrupTrtQ2VNhTaWKMb9qMlLw9T20Db6A66ewSWKzWU-Q5bN2asGDNMlEXpgZsQ_rr0hGrF6dc_h0dqX2z8lzmrDTKgivFx8eWw-HhhopAKN-A0ixi0W3PzuCnl48vi7WR28mlxfW2iOCX-mgpboJp_P5uPLxaTz_PJucLaWO5xdVwcTauD3-vdalgPW1BQhuURMLStGsDPgR3I6Ri5ul_3sn2F3211GoogoYLUxRtxb1f85NlTq7X0Wvx8p2zakRZEnsEcUrhOIg8H737QZYR59F9bYx1tbT2AYVKhUcGdCsKw3GBDxbAoP3SwgodNaGzJUHYJIyGMQ0b26VCGouqKDz-Ks8thmcbyRFUPq11Bq10qF9NfaJnRbUF19Xc8hLzSpKoEwaoQhuCmjVVCG2ubWkJ5Qro-_MYDSUKGEU0gLhOWPg1IHsO1OGwEUpKLFPeZp5mC53RTy3sQMm12CVxQopkBiTP4ytM8jJaoR9iLNE1uS0IuW8n16C-_I2cB

flowchart TD subgraph P1["Process 1: run_data_fetcher.py"] S1["schedule 定时循环"] F1["fetch_tier1()"] F2["fetch_tier2()"] API["adata API 直连"] end subgraph P2["Process 2: show_marker.py (推送)"] S2["schedule 定时循环"] CK["时效检查: get_freshness()"] LD["数据加载: load()"] SC["策略计算"] PS["消息推送"] end subgraph CACHE["data_cache/"] CSV["{code}_1d.csv"] META["{code}_1d.meta.txt"] end S1 --> F1 S1 --> F2 F1 --> API F2 --> API API -->|DataCacheManager.save| CSV API -->|更新| META S2 --> CK CK -->|读 .meta.txt| META LD -->|读 .csv| CSV CK --> LD LD --> SC SC --> PS

CSV 缓存格式

文件命名

{code}_{k_type_suffix}.csv，例如 399102.SZ_1d.csv

k_type_suffix：1d / 1wk / 1mo / 5m / 15m / 1m

保存行为

全量覆盖，不是追加。每次拉取 1990-01-01 ~ 今天，返回全量 DataFrame，完整覆写整个 CSV。好处是数据始终完整一致，坏处是每次重复拉全量历史（约 3000~8000 行/品种）。

后续可优化为增量更新：只拉最近 N 天，与已有 CSV 合并后重新写入。

元数据文件

.meta.txt 格式：

fetched_at=2026-05-26T09:35:12
latest_trade_date=2026-05-25
rows=3847

推送时的时效检查只需读这 2 行文本，不需要加载整个 CSV。

防误覆盖机制

DataCacheManager.save() 三道防线：

输入 df 为空或 None → 跳过，保留旧缓存
新数据 latest_trade_date 比旧缓存更旧 → 跳过
写入时用 .tmp + os.rename 原子替换，防止断电写坏

品种分层与定时排程

Tier 1（高频，每日 6 次）

择时策略代码池（show_marker.py 的 codes_to_process） + 截面 WATCHLIST，去重后约 66 个品种。定时：08:00（全量）、09:30、12:00、14:00、20:00、02:00。

注意拉取时用 _quick_fetch_one() 直接调 adata API，跳过 tushare（后者不稳定且容易并发超限）。非 ETF 代码（HSI、GLD、QQQ、980092.CNI 等）仍走 fetch_data()。

Tier 2（全量，每日 1 次，08:00）

全市场 ETF 池（CURATED_ETF_POOL），约 76 个品种。

文件清单

文件	操作	说明
------	------	------
data_cache.py	新建	`DataCacheManager`：save/load/get_freshness/find_stale_codes
run_data_fetcher.py	新建	独立拉取进程，schedule 定时循环
run_fetch.sh	新建	CLI 入口脚本
data_cache/	新建目录	CSV 缓存，已 gitignore
show_marker.py	修改	时效检查→`get_freshness`；回测数据→`load`；截面引擎→`fetch_all_data_from_cache`
engine.py	修改	新增 `fetch_all_data_from_cache()`
.gitignore	修改	排除 data_cache

不改的文件：df_loader.py、allocation_utils.py、run_push.sh、所有策略文件。

启动方式

# 启动数据拉取定时循环（长期运行，建议 nohup 后台）
nohup ./run_fetch.sh > fetch.log 2>&1 &

# 或手动触发单次
./run_fetch.sh -f       # 全量（Tier1 + Tier2），跑完退出
./run_fetch.sh -t       # 仅 Tier1，跑完退出

# 策略推送定时循环（另一个终端）
nohup .venv/bin/python -m unittest \
    adata.aohan.quant.chart_test.show_marker.MarketDataPlotly.test_schedule_push -v \
    > push.log 2>&1 &

错误处理

fetcher 崩溃 → pusher 不受影响，基于旧 CSV 继续工作
pusher 在 CSV 缺失时 → 跳过该品种，标记「无缓存」，不崩溃
API 超时 → 单 code 60s 超时自动跳过，失败品种等待 30s 后重试一轮
推送消息时效标签改为反映 CSV 缓存新鲜度

TODO

后续可优化为增量更新：只拉最近数据，合并到已有 CSV，减少网络开销
Excel 备份逻辑（df_loader.py 中 export_mode='replace'）后续可清理
可考虑给 fetcher 加 launchd 系统守护，崩溃后自动重启
未来若 CSV 文件过大，可考虑按年份分文件，当前 load 的日期截断已支持

▶ 📜 变更历史（1 次提交）

01005c22026-07-08 02:17chore: 更新择时图表缓存、推送消息、数据拉取、推送历史

🍃quant 知识库

by 伊甸园和枫