Data Pipeline 規格文檔

概述

資料排程與資料處理由兩個部分組成：

排程入口：investment-agent/backend/scripts/scheduler.py — docker-compose 的 scheduler 服務（container cortex-scheduler， command python scripts/scheduler.py）。這是唯一的排程服務入口。
data_pipeline/ 函式庫：investment-agent/backend/data_pipeline/ — 共用函式庫（指標計算、commodity 爬蟲），不是獨立服務，沒有自己的 daemon 或 container。

歷史備註：data_pipeline/scheduler.py 過去有一個 DataPipelineScheduler 類別（獨立的 per-market 股票排程器），已於 #601 移除 — 它沒有任何 live caller。該檔案現在只保留 run_commodity_pipeline_standalone。

1. 排程入口：`scripts/scheduler.py`

檔案: investment-agent/backend/scripts/scheduler.py

使用 APScheduler（AsyncIOScheduler），時區 Asia/Taipei。
每日更新於台灣時間 15:30（收盤 13:30 後）執行；資料未到齊時每 30 分鐘重試，最多重試 6 小時（MAX_RETRY_HOURS = 6）。
每週五 20:00 執行 weekly full update（weekly_full_update_job）。
每週日 02:00 執行 weekly strategy discovery（_run_weekly_discovery，TA/CT/IR/COM）。
Commodity pipeline：cron 由 config/commodities.yaml 的 scheduler 設定決定，呼叫 data_pipeline.scheduler.run_commodity_pipeline_standalone；另有 catch-up 路徑（commodity_ohlc 落後台股 OHLC 時補跑）。
台股 OHLC 抓取由 scripts/taiwan_stocks.py 執行：FinMind 為主要來源（#409），yfinance 為備援（#396）— 不經過 data_pipeline/ 的爬蟲。

CLI:

python scripts/scheduler.py            # 啟動排程器（daemon）
python scripts/scheduler.py --run-now  # 先立即執行一次每日更新，再啟動排程器
python scripts/scheduler.py --setup    # 執行初始 setup（下載全部股票）後結束

需要 DATABASE_URL 環境變數，未設定時直接退出。其餘必要環境變數（FRED_API_KEY、FINMIND_TOKEN 等）見 investment-agent/docker-compose.yml 的 scheduler 服務定義。

2. `data_pipeline/` 函式庫

目錄: investment-agent/backend/data_pipeline/

模組	內容	主要使用者
`indicators.py`	`IndicatorCalculator` — 技術指標計算引擎	`scripts/daily_analysis_snapshot.py`、`services/periodic_technical_raw.py`、`api/routes/stocks.py`、`scripts/run_periodic_recap.py`
`base_crawler.py`	`BaseCrawler` + `CrawlerError` — 共用 retry / rate-limit 基礎	`FredCrawler`、`YFinanceCommodityCrawler` 繼承
`fred_crawler.py`	`FredCrawler` — FRED commodity 資料	commodity pipeline
`commodity_crawler.py`	`YFinanceCommodityCrawler`	commodity pipeline
`commodity_orchestrator.py`	`CommodityCrawlerOrchestrator`	commodity pipeline
`commodity_indicators.py`	`CommodityIndicatorCalculator`	commodity pipeline、`scripts/backfill_commodity_history.py`
`commodity_validation.py`	`CommoditySourceValidator`	commodity pipeline
`commodity_daily_write.py`	commodity OHLC / snapshot 寫入函數	commodity pipeline
`scheduler.py`	`run_commodity_pipeline_standalone`（唯一保留的函數）	`scripts/scheduler.py`（排程 job + catch-up）
`config_loader.py`	`load_core_yaml` / `load_markets_yaml`	`scripts/scheduler.py` 等
`ma_helpers.py` / `volume_helpers.py`	指標輔助函數	`indicators.py`、snapshot 流程
`safe_cast.py`	`safe_float` / `safe_int` / `safe_str` / `safe_bool`	各處
`ta_compat.py`	pandas-ta ARM 相容 patch	`indicators.py`

data_pipeline/__init__.py 匯出 IndicatorCalculator。

crawler.py（StockCrawler，yfinance 股票爬蟲）已隨 #620 移除：它最後的 production caller 是 #601 刪除的 DataPipelineScheduler，之後只剩測試與文檔引用。台股 OHLC 由 scripts/taiwan_stocks.py 負責。

processor.py（DataProcessor，OHLC upsert）已隨 #628 移除：沒有任何 production caller，只剩自身行為的測試與 __init__ re-export。 OHLC 寫入的 live 路徑是 database/crud.py 的 save_ohlc_batch 與各 script（scripts/taiwan_stocks.py 等）。

download_history.py（yfinance 5 年歷史 OHLC 批次回補）已隨 #855 移除：它呼叫 yf.Ticker().history() 未帶 auto_adjust=False（寫入還原價，違反 raw-traded-price 紀律，#396），且繞過 #856 的 no-trade placeholder 過濾（非交易日會插入捏造的零量 bar，#847）。沒有任何 production caller，台股歷史回補由 scripts/backfill_taiwan_ohlc_raw.py（FinMind 官方 raw 價）負責。

ema_helpers.py（compute_ema_series，EMA 輔助函數）已隨 #678 移除：其最後的 production caller scripts/backfill_real_ema_periods.py 在

661 退役後即孤兒化，只剩自身測試。EMA 計算的 live 路徑是

indicators.py 的 IndicatorCalculator（#637 起全歷史單引擎）。

3. Docker 配置

實際的服務定義在 investment-agent/docker-compose.yml（節錄）：

services:
  scheduler:
    build:
      context: ..
      dockerfile: investment-agent/backend/Dockerfile
    container_name: cortex-scheduler
    command: python scripts/scheduler.py
    environment:
      - DATABASE_URL=postgresql+asyncpg://...
      - PYTHONPATH=/app:/app/core
      - TZ=Asia/Taipei
      - FRED_API_KEY=${FRED_API_KEY:?...}
      - FINMIND_TOKEN=${FINMIND_TOKEN:?...}
    depends_on:
      postgres:
        condition: service_healthy
    restart: unless-stopped

修改 scheduler 所 import 的模組後，merge 完必須 docker restart cortex-scheduler（該服務不會 hot-reload）。

4. 測試

tests/test_commodity_crawler.py — commodity 爬蟲（BaseCrawler 行為）
tests/test_ohlc_market_column.py — OHLC 寫入路徑的 market 欄位（source inspection）
scripts/test_indicators.py — 手動指標驗證腳本（python scripts/test_indicators.py [ticker]）

執行方式見 repo 根目錄 CLAUDE.md 的 Build & Test Commands。

Data Pipeline 規格文檔

概述

1. 排程入口：scripts/scheduler.py

2. data_pipeline/ 函式庫

661 退役後即孤兒化，只剩自身測試。EMA 計算的 live 路徑是

3. Docker 配置

4. 測試

1. 排程入口：`scripts/scheduler.py`

2. `data_pipeline/` 函式庫