標籤: 時間序列預測

  • LLM × sktime craft 打造 AutoForecast

    LLM × sktime craft 打造 AutoForecast

    📌 本文重點

    • sktime + craft() 讓 LLM 設計可訓練的預測 pipeline
    • 以 LLM 當 search policy,降低傳統 AutoML 搜尋成本
    • 把業務限制寫進 prompt,兼顧準確度、延遲與維運

    傳統做時間序列 AutoML,多半靠 grid search / Bayesian search 把模型和超參數「全掃一輪」,成本高、速度慢,而且一換業務場景就得重調。這篇要介紹的是:用 sktime 的 pipeline + craft() 介面,讓 LLM 充當預測藍圖設計師,自動組裝可訓練、可回測的 forecasting pipeline,實際解決:

    • 算力被 AutoML 搜尋吃光的問題
    • 每個產品線都要單獨手調模型的維護地獄
    • 新資料來時,預測流程很難「可重現、自動化演進」的痛點

    重點說明

    1. sktime pipeline + craft():把「模型設計」變成文字介面

    sktime 提供了時間序列專用的 pipeline / composer 抽象,可以把:

    • 前處理(差分、假日特徵、滯後特徵)
    • 模型本體(ARIMA、Gradient Boosting、機器樹、深度模型包裝器)
    • 聚合、多變量等

    組成一個可訓練的 forecaster 物件

    craft() 的角色:

    • 接收一段以文字描述的「藍圖」或結構化 blueprint
    • 解析成真正的 sktime pipeline 物件
    • 後續你可以直接 .fit() / .predict() / 做交叉驗證

    💡 關鍵: craft() 讓「文字藍圖」直接變成可訓練的 sktime pipeline,是把 LLM 接到 AutoForecast 流程的關鍵樞紐。

    這讓 LLM 可以只負責「寫藍圖」,而不是直接產生一大段 Python 亂碼。


    2. 為什麼用 LLM 當 search policy,而不是再做一個 AutoML

    傳統 AutoML:

    • 先定義 search space(例如 10 種模型 × 10 個超參數 × 若干取值)
    • Grid/Bayesian 搜尋會盲目探索大量組合

    LLM 驅動 AutoForecast 的思路:

    • 你提供:資料描述、目標、限制條件、白名單元件
    • LLM 輸出:一個「專家風格」的 pipeline blueprint
    • sktime craft():把 blueprint 變成真正的 estimator
    • 之後用交叉驗證 + 回測,打分這個藍圖好不好

    好處:

    • 搜尋空間更有結構:LLM 預先排除很多不合理組合
    • 成本可控:每次只訓練少數幾個「有合理解釋」的藍圖
    • 你可以 固化得分高的藍圖,變成穩定的產線預測器

    💡 關鍵: 相較於暴力掃描大量組合,讓 LLM 先縮小「合理藍圖集合」,能在相同算力下探索更有價值的模型設計。


    3. 把業務限制寫進 prompt:準確度只是其中一個目標

    在實務專案中,像房地產、銷售預測一樣,除了誤差小,你還會在乎:

    • 推理時間(每天要跑上千條 SKU / 房源)
    • 部署複雜度(不要依賴罕見套件或 GPU)
    • 解釋性(要能向業務說明為什麼預測變化)

    這些都可以在 prompt 中顯式告訴 LLM,例如:

    • 限定只能用某些 estimator:NaiveForecaster, ExponentialSmoothing, ElasticEnsemble, LightGBM-based forecaster...
    • 給出目標指標:sMAPEMAE,並設計多目標(準確度 + latency)

    實作範例:銷售時間序列 LLM AutoForecast

    以下示範一個簡化版流程:

    • 場景:預測未來 3 個月每月團隊銷售額
    • 資料:月度歷史 revenue、房源數量、成交率、利率、季節 dummy 等

    假設你已經載入 sktime 與一個 LLM SDK(例如 Anthropic、OpenAI 等)。以下程式碼偏 pseudo,但結構可直接套入專案。

    1. 描述資料與目標,建立 LLM prompt

    schema_description = {
        "frequency": "M",  # 月度資料
        "target": "team_gci",  # 團隊佣金收入
        "horizon": 3,  # 預測 3 個月
        "exogenous_features": [
            "active_listings",     # 有效掛牌數
            "pending_transactions", # 待成交案件
            "mortgage_rate",      # 抵押貸款利率
            "seasonality_flags"   # 季節性特徵
        ],
        "constraints": {
            "max_pipeline_depth": 4,
            "allowed_estimators": [
                "NaiveForecaster",
                "ExponentialSmoothing",
                "ThetaForecaster",
                "LightGBMForecaster"
            ],
            "allowed_transformers": [
                "Detrender",
                "STLTransformer",
                "Lag",
                "DateTimeFeatures"
            ],
            "primary_metric": "sMAPE",
            "secondary_metric": "latency_ms",
            "max_fit_time_minutes": 15
        }
    }
    
    system_prompt = """
    你是一位時間序列預測專家,負責設計 sktime 預測 pipeline。
    
    要求:
    1. 只使用以下白名單中的 estimator 和 transformer。
    2. 避免過度複雜的 pipeline,深度不超過 4 層。
    3. 避免使用不存在的類別和參數,嚴格遵守 sktime API。
    4. 針對月度房地產團隊銷售數據,考慮趨勢與季節性。
    5. primary metric 是 sMAPE,次要考慮推理延遲,盡量用輕量模型。
    
    輸出格式:只輸出 JSON,欄位為 `pipeline_spec`,不可包含其他文字。
    `pipeline_spec` 要能被 sktime.craft() 解析。
    """
    
    user_prompt = f"資料與限制如下:\n{schema_description}\n請產生 pipeline_spec。"
    

    2. 呼叫 LLM,取得 pipeline blueprint

    from some_llm_client import LLM
    
    llm = LLM(api_key="...")
    
    response = llm.chat(
        system=system_prompt,
        user=user_prompt
    )
    
    blueprint = response["pipeline_spec"]  # 假設已解析 JSON
    print(blueprint)
    

    例:LLM 可能輸出類似(簡化)

    {
      "type": "TransformedTargetForecaster",
      "steps": [
        {"name": "detrend", "class": "Detrender", "params": {"forecaster": "NaiveForecaster"}},
        {"name": "stl", "class": "STLTransformer", "params": {"seasonal": 7}},
        {"name": "lag", "class": "Lag", "params": {"lags": [1, 2, 3, 6, 12]}},
        {"name": "model", "class": "LightGBMForecaster", "params": {"num_leaves": 31, "learning_rate": 0.05}}
      ]
    }
    

    💡 關鍵: 藉由 max_pipeline_depth = 4、白名單與 max_fit_time_minutes = 15 等約束,LLM 被強迫產出既合理又可在時限內完成訓練的藍圖。

    3. 用 craft() 轉成可執行 pipeline

    from sktime.craft import craft
    
    # 這裡的 blueprint 就是上一步 LLM 回傳的 JSON
    forecaster = craft(blueprint)
    
    print(type(forecaster))
    # e.g. <class 'sktime.forecasting.compose._pipeline.TransformedTargetForecaster'>
    

    如果 LLM 有亂給不存在的 class/參數,這一步會直接爆掉,所以建議外面包一層驗證:

    def safe_craft(blueprint):
        try:
            return craft(blueprint)
        except Exception as e:
            # 記錄錯誤,丟回給 LLM 做自我修正或直接丟棄該藍圖
            print("Invalid blueprint:", e)
            return None
    
    forecaster = safe_craft(blueprint)
    if forecaster is None:
        # 重新請 LLM 生成,或 fallback 到手寫 baseline
        ...
    

    4. 做時間序列交叉驗證與回測(避免 leakage)

    時間序列不能隨機 shuffle,必須用 滾動時間窗

    from sktime.forecasting.model_selection import ExpandingWindowSplitter
    from sktime.performance_metrics.forecasting import mean_absolute_percentage_error
    
    cv = ExpandingWindowSplitter(
        initial_window=36,  # 例如先用 3 年訓練
        step_length=3,      # 每次往前滾 3 個月
        fh=[1, 2, 3]        # 評估 1-3 個月 horizon
    )
    
    CV_scores = []
    for train_idx, test_idx in cv.split(y):
        y_train, y_test = y.iloc[train_idx], y.iloc[test_idx]
        X_train, X_test = X.iloc[train_idx], X.iloc[test_idx]
    
        forecaster.fit(y_train, X=X_train)
        y_pred = forecaster.predict(fh=cv.fh, X=X_test)
    
        score = mean_absolute_percentage_error(y_test, y_pred)
        CV_scores.append(score)
    
    print("CV MAPE:", sum(CV_scores) / len(CV_scores))
    

    這個流程可以放在一個 LLMBlueprintForecaster 類別裡,作為你的 AutoForecast 前端代理:

    1. 給資料描述與限制
    2. LLM 產生藍圖
    3. craft() 轉成 pipeline
    4. 用時間窗交叉驗證打分
    5. 挑最佳藍圖,固化成產線模型

    建議與注意事項

    1. LLM 亂組 class / 參數:一定要有 validator

    常見問題:

    • 寫出不存在的 class 名稱(如 XGBoostForecaster 明明沒這個)
    • 傳錯 參數名稱 或型別

    實務上建議:

    • 自行維護一份 白名單 registryALLOWED_ESTIMATORS, ALLOWED_TRANSFORMERS,包含合法 class 與參數 schema
    • LLM 輸出後先做 schema validation,不合法就直接丟棄或要求 LLM 修正

    2. 避免過度複雜 pipeline:限制深度與組合數

    LLM 很容易產生「看起來很專業」的 pipeline:一堆 transformer 疊來疊去,訓練時間爆炸還容易 overfit。

    做法:

    • 在 prompt 裡明寫:max_pipeline_depth、禁止嵌套某些昂貴 transformer
    • 在 validator 裡硬限制步數,例如 len(steps) <= 4
    • 將 fit time / memory 也當成 約束條件,訓練時加上 timeout + 監控

    3. 嚴格避免 leakage:時間切割一律「只看過去」

    坑點:

    • 把整段資料做標準化 / 滯後特徵時,無意間用到未來資訊

    避免方式:

    • 一律用 sktime 的 transformer + forecaster pipeline,讓 transform 在 fit 只看到 train window
    • cross-validation 必須用 ExpandingWindowSplitter / SlidingWindowSplitter 類型
    • 在 prompt 裡提醒:不得使用未來的統計量(例如整體均值)來處理訓練資料

    4. 在專案中落地:把 LLM 當 search policy,而不是 oracle

    建議實務流程:

    1. Search policy:LLM 只負責提案藍圖,不直接上產線
    2. 離線評估:用固定的 backtest 配置(split、metric、timeout)評估每個藍圖
    3. 固化最佳藍圖:將 blueprint JSON 連同該版本資料 schema 一起存入 repo
    4. CI 自動回歸
    5. 新資料 schema / 分布變化時,自動對舊藍圖重新訓練 + 打分
    6. 可以定期讓 LLM 在新 constraint 下重新產生藍圖,與舊版本對比
    7. 可重現性:所有 LLM 輸出(prompt + response)都要 versioning(例如存到 S3 / Git LFS),確保每個産線模型的來歷可追溯

    關鍵結論:

    • craft() + LLM = 可控的 AutoForecast 工具鏈,你掌控 search space 與評估邏輯
    • 把 LLM 當作「有經驗的建模同事」,而不是神諭;所有藍圖都要在 sktime 的嚴格回測與 CI 下過關,才能進產線

    這樣,在實際銷售 / 房地產等時間序列場景中,你可以以相對低成本,不斷迭代更好的預測流程,同時維持可重現與可維運的工程品質。


    🚀 你現在可以做的事

    • 在專案中安裝並載入 sktime,試著手動呼叫 craft() 建一個簡單 pipeline
    • 依照文中範例,實作一份包含 max_pipeline_depth 與白名單的 LLM prompt,讓 LLM 先產出一版 pipeline_spec
    • 為產出的 blueprint 加上 safe_craft() + 時間序列交叉驗證,建立一個最小可用的 LLMBlueprintForecaster 原型