Data Pipeline 規格文檔

概述

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

每日／批次排程入口：investment-agent/backend/scripts/scheduler.py — docker-compose 的 scheduler 服務（container cortex-scheduler， command python scripts/scheduler.py）。此服務不註冊即時抓價工作。
即時抓價入口：investment-agent/backend/scripts/realtime_quote_scheduler.py — realtime-quote-scheduler 服務（container cortex-realtime-quote-scheduler），只輪詢台股即時報價，可獨立啟停。
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 小時（起跑時刻、重試預算與由此得出的 21:30 截止時點定義於 services/daily_update_window.py；僅供排程與重試使用，投資績效不再以每日 checkpoint 判定計算完整性）。
每週五 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. 即時抓價：`scripts/realtime_quote_scheduler.py`

僅註冊 realtime_quote_poll，不匯入每日排程，也不執行 startup recovery／catch-up。
間隔與交易日／時段沿用 config/core.yaml 的 realtime_quote 設定及官方日曆。非交易日或設定時段外不呼叫任一來源，也不開啟寫入交易。
寫入 taiwan.realtime_quote_snapshots；provider deadline、交易 timeout、去重與前日資料保留規則不變。SIGTERM／SIGINT 停止新工作、取消進行中的輪詢並清理 DB engine。
預設 data_hub，需 DATABASE_URL、HUB_API_URL、HUB_SERVICE_TOKEN、Python path 與台北時區；REALTIME_QUOTE_SOURCE=finmind 時改用 FINMIND_TOKEN。兩來源共用一個 job，不自動 fallback。完整來源／freshness／操作契約見即時來源文件。不依賴 FRED、Redis 或 LLM。以 backend 目錄為工作目錄並配置 Python path 後執行：

python scripts/realtime_quote_scheduler.py

即時報價 REST／SSE 使用原有資料表與契約。SIT 的獨立容器為 cortex-sit-realtime-quote-scheduler。

3. `data_pipeline/` 函式庫

目錄: investment-agent/backend/data_pipeline/

模組	內容	主要使用者
`indicator_contract.py`	V2 `CONTRACT` — 78 欄指標的凍結 spec（first_valid/mature/zero_denominator/文獻出處，issue #1356）	`indicator_engine_v2.py`、oracle harness、schema-parity 測試
`indicator_primitives.py`	V2 共用數值 primitives（`recursive_ema`/`wilder_rma`/`true_range`/輸入驗證函數）	V2 `indicator_formulas_*.py`、`indicator_engine_v2.py`
`indicator_formulas_*.py`	V2 個別指標公式實作（channels/core/custom/trend/volatility/volume_flow/wave 七個分片）	`indicator_engine_v2.py`
`indicator_engine_v2.py`	`compute_all_v2` — 組裝 78 欄 V2 輸出；公式行為凍結（PR1）	`indicator_adapter_v2.py`
`indicator_adapter_v2.py`	`compute_indicator_frame` — V2 production adapter（issue #1356 PR2 Commit 1）	未來 snapshot writer（PR2 Commit 3 起）
`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`	指標輔助函數	snapshot / backfill 流程（`daily_analysis_snapshot.py`、`ta_index_payload.py`、`signal_scanner.py` 等）
`safe_cast.py`	`safe_float` / `safe_int` / `safe_str` / `safe_bool`	各處

indicators.py（IndicatorCalculator，V1 技術指標計算引擎）與 ta_compat.py（pandas-ta ARM 相容 patch）已於 #1356 決策 5 移除：V1 引擎全面停用，原本的呼叫端（scripts/daily_analysis_snapshot.py、 services/periodic_technical_raw.py、api/routes/stocks.py、 scripts/run_periodic_recap.py）全部改用 V2 production adapter data_pipeline/indicator_adapter_v2.compute_indicator_frame； ta_compat.py 只服務 V1 的 pandas-ta ARM 相依，隨之一併移除，V2 引擎不需要這個 patch。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 起全歷史單引擎）。

4. 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
  realtime-quote-scheduler:
    build:
      context: ..
      dockerfile: investment-agent/backend/Dockerfile
    container_name: cortex-realtime-quote-scheduler
    working_dir: /app/backend
    command: python scripts/realtime_quote_scheduler.py
    environment:
      - DATABASE_URL=postgresql+asyncpg://...
      - PYTHONPATH=/app/backend:/app:/app/core
      - TZ=Asia/Taipei
      - REALTIME_QUOTE_SOURCE=${REALTIME_QUOTE_SOURCE:-data_hub}
      - HUB_API_URL=${HUB_API_URL:-http://host.docker.internal:58080}
      - HUB_SERVICE_TOKEN=${HUB_SERVICE_TOKEN:-}
      - FINMIND_TOKEN=${FINMIND_TOKEN:-}
    depends_on:
      postgres:
        condition: service_healthy
    restart: unless-stopped

兩個服務都不會 hot-reload；套用變更時依實際受影響的入口更新／重啟服務。首次拆分必須先讓原 scheduler 載入已移除 quote job 的版本，再啟動新 worker，避免同時存在兩個輪詢器。既有資料重建或 DB 維護尚在執行時，先完成其停止寫入安排，再切換服務。

需要停止所有 DB writers 的維護流程必須包含 API、每日 scheduler 與 quote scheduler。SIT 部署、rollback、DB rebase 與 dev cutover rehearsal 已納入此 inventory；操作步驟見 docs/runbooks/sit-deploy.md。舊版本沒有獨立 worker 時， rollback 只恢復舊 API／scheduler，保持新 worker 停止。

5. 測試

tests/test_commodity_crawler.py — commodity 爬蟲（BaseCrawler 行為）
tests/test_ohlc_market_column.py — OHLC 寫入路徑的 market 欄位（source inspection）
tests/test_realtime_quote_scheduler_isolation.py — 入口分離、單一 job 與停止清理
tests/test_scheduler_realtime_quote_job.py — quote trigger 與設定
tests/integration/test_scheduler_realtime_quote_pg.py — 真實 PostgreSQL 輪詢／timeout／保留資料

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

Data Pipeline 規格文檔

概述

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

2. 即時抓價：scripts/realtime_quote_scheduler.py

3. data_pipeline/ 函式庫

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

4. Docker 配置

5. 測試

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

2. 即時抓價：`scripts/realtime_quote_scheduler.py`

3. `data_pipeline/` 函式庫