使用飛鏢進行需求預測：教程 | 作者：Sandra E.G. | 2025年1月

你收集到所有相關數據了嗎？

假設你的公司提供了一個交易數據庫，裡面有不同產品和不同銷售地點的銷售數據。這些數據稱為面板數據，這意味著你將同時處理多個時間序列。

這個交易數據庫可能會有以下格式：銷售日期、銷售地點識別碼、產品識別碼、數量，還有可能的金錢成本。根據這些數據的收集方式，它們會以不同的方式進行聚合，按時間（日、週、月）和按組（按客戶或按地點和產品）。

但這些數據就足夠用來預測需求了嗎？有時是，有時不是。當然，你可以使用這些數據進行一些預測，如果系列之間的關係不複雜，簡單的模型可能會有效。但如果你正在閱讀這個教程，你可能對在數據不那麼簡單的情況下預測需求感興趣。在這種情況下，還有其他信息可以成為遊戲規則改變者，如果你能夠獲得它：

• 歷史庫存數據：了解缺貨何時發生是至關重要的，因為當銷售數據未反映出來時，需求可能仍然很高。
• 促銷數據：折扣和促銷也會影響需求，因為它們會影響顧客的購物行為。
• 事件數據：如後面所討論的，可以從日期索引中提取時間特徵。然而，假期數據或特殊日期也可以影響消費。
• 其他領域數據：任何其他可能影響你所處理的產品需求的數據都可能與這項任務相關。

讓我們開始編碼吧！

在這個教程中，我們將使用按產品和銷售地點聚合的每月銷售數據。這個示例數據集來自 Stallion Kaggle Competition，記錄了通過批發商（代理商）分發給零售商的啤酒產品（SKU）。第一步是格式化數據集，並選擇我們想用來訓練模型的列。正如你在代碼片段中看到的，我們將所有事件列合併為一個名為「特殊日子」的列，以簡化操作。如前所述，這個數據集缺少庫存數據，因此如果發生缺貨，我們可能會誤解實際需求。

# 載入數據
sales_data = pd.read_csv(f'{local_path}/price_sales_promotion.csv’)
volume_data = pd.read_csv(f'{local_path}/historical_volume.csv’)
events_data = pd.read_csv(f'{local_path}/event_calendar.csv’)

# 合併所有數據
dataset = pd.merge(volume_data, sales_data, on=[‘Agency’,’SKU’,’YearMonth’], how=’left’)
dataset = pd.merge(dataset, events_data, on=’YearMonth’, how=’left’)

# 日期時間處理
dataset.rename(columns={‘YearMonth’: ‘Date’, ‘SKU’: ‘Product’}, inplace=True)
dataset[‘Date’] = pd.to_datetime(dataset[‘Date’], format=’%Y%m’)

# 格式化折扣
dataset[‘Discount’] = dataset[‘Promotions’]/dataset[‘Price’]
dataset = dataset.drop(columns=[‘Promotions’,’Sales’])

# 格式化事件
special_days_columns = [‘Easter Day’,’Good Friday’,’New Year’,’Christmas’,’Labor Day’,’Independence Day’,’Revolution Day Memorial’,’Regional Games ‘,’FIFA U-17 World Cup’,’Football Gold Cup’,’Beer Capital’,’Music Fest’]
dataset[‘Special_days’] = dataset[special_days_columns].max(axis=1)
dataset = dataset.drop(columns=special_days_columns)

你檢查過錯誤值嗎？

雖然這部分比較明顯，但仍然值得一提，因為這可以避免將錯誤數據輸入到我們的模型中。在交易數據中，尋找零價格交易、銷售量大於剩餘庫存、停產產品的交易等。

你是在預測銷售還是需求？

這是我們在預測需求時需要做出的關鍵區分，因為目標是預見產品的需求以優化補貨。如果我們只看銷售而不觀察庫存數據，當發生缺貨時，我們可能會低估需求，從而引入偏差到我們的模型中。在這種情況下，我們可以忽略缺貨後的交易，或者試著正確填補這些值，例如用需求的移動平均。

讓我們開始編碼吧！

在這個教程中選擇的數據集的預處理相當簡單，因為我們沒有庫存數據。我們需要通過填補正確的值來修正零價格交易，並填補折扣列的缺失值。

# 填補價格
dataset.Price = np.where(dataset.Price==0, np.nan, dataset.Price)
dataset.Price = dataset.groupby([‘Agency’, ‘Product’])[‘Price’].ffill()
dataset.Price = dataset.groupby([‘Agency’, ‘Product’])[‘Price’].bfill()

# 填補折扣
dataset.Discount = dataset.Discount.fillna(0)

# 排序
dataset = dataset.sort_values(by=[‘Agency’,’Product’,’Date’]).reset_index(drop=True

# 你需要預測所有產品嗎？
根據一些條件，如預算、成本節約和你使用的模型，你可能不想預測整個產品目錄。假設在實驗後，你決定使用神經網絡。這些通常需要昂貴的訓練，並且需要更多的時間和資源。如果你選擇訓練和預測完整的產品集，解決方案的成本將增加，甚至可能使其對你的公司來說不值得投資。在這種情況下，一個好的替代方案是根據特定標準對產品進行細分，例如使用你的模型僅預測產生最高收入的產品。其餘產品的需求可以使用更簡單且便宜的模型進行預測。

你能提取更多相關信息嗎？

特徵提取可以應用於任何時間序列任務，因為你可以從日期索引中提取一些有趣的變量。特別是在需求預測任務中，這些特徵是有趣的，因為一些消費習慣可能是季節性的。提取星期幾、月份的第幾週或一年的月份可能對幫助你的模型識別這些模式很有幫助。正確編碼這些特徵是關鍵，我建議你閱讀有關循環編碼的資料，因為在某些情況下，這對時間特徵可能更合適。

讓我們開始編碼吧！

在這個教程中，我們首先對產品進行細分，只保留高周轉的產品。在進行特徵提取之前執行這一步可以幫助減少性能成本，當你有太多低周轉系列時，你不會使用這些系列。為了計算周轉，我們只使用訓練數據。為此，我們事先定義數據的拆分。請注意，我們有兩個日期用於驗證集，VAL_DATE_IN表示那些也屬於訓練集的日期，但可以用作驗證集的輸入，而VAL_DATE_OUT表示從哪個時間點開始將使用時間步驟來評估模型的輸出。在這種情況下，我們將所有銷售75%時間的系列標記為高周轉，但你可以根據源代碼中的實現函數進行調整。之後，我們進行第二次細分，以確保我們有足夠的歷史數據來訓練模型。

# 拆分日期
TEST_DATE = pd.Timestamp(‘2017-07-01’)
VAL_DATE_OUT = pd.Timestamp(‘2017-01-01’)
VAL_DATE_IN = pd.Timestamp(‘2016-01-01’)
MIN_TRAIN_DATE = pd.Timestamp(‘2015-06-01’)

# 周轉
rotation_values = rotation_tags(dataset[dataset.Datedataset = dataset.merge(rotation_values, on=[‘Agency’,’Product’], how=’left’)
dataset = dataset[dataset.Rotation==’high’].reset_index(drop=True)
dataset = dataset.drop(columns=[‘Rotation’])

# 歷史
first_transactions = dataset[dataset.Volume!=0].groupby([‘Agency’,’Product’], as_index=False).agg(First_transaction = (‘Date’, ‘min’),)
dataset = dataset.merge(first_transactions, on=[‘Agency’,’Product’], how=’left’)
dataset = dataset[dataset.Date>=dataset.First_transaction]
dataset = dataset[MIN_TRAIN_DATE>=dataset.First_transaction].reset_index(drop=True)
dataset = dataset.drop(columns=[‘First_transaction’])

由於我們正在處理每月聚合的數據，因此沒有太多時間特徵可以提取。在這種情況下，我們包括位置，這只是系列順序的數字索引。時間特徵可以在訓練時間內計算，通過編碼器告訴 Darts。此外，我們還計算了前四個月的移動平均和指數移動平均。

dataset[‘EMA_4’] = dataset.groupby([‘Agency’,’Product’], group_keys=False).apply(lambda group: group.Volume.ewm(span=4, adjust=False).mean())
dataset[‘MA_4’] = dataset.groupby([‘Agency’,’Product’], group_keys=False).apply(lambda group: group.Volume.rolling(window=4, min_periods=1).mean())

# Darts 的編碼器
encoders = {“position”: {“past”: [“relative”], “future”: [“relative”]},”transformer”: Scaler(),}

你是否定義了一組基準預測？

與其他用例一樣，在訓練任何高級模型之前，你需要建立一個基準，這個基準是你想要超越的。通常，當選擇基準模型時，你應該尋求一些簡單的東西，幾乎沒有任何成本。在這個領域的一個常見做法是使用需求的移動平均作為基準。這個基準可以在不需要任何模型的情況下計算，但為了簡化代碼，在這個教程中，我們將使用 Darts 的基準模型 NaiveMovingAverage。

你的模型是本地還是全局的？

你正在處理多個時間序列。現在，你可以選擇為每個時間序列訓練一個本地模型，或者為所有序列訓練一個全局模型。沒有「正確」的答案，兩者都根據你的數據而有效。如果你有數據，你認為在按商店、產品類型或其他類別特徵分組時行為相似，你可能會從全局模型中受益。此外，如果你有大量系列，並且想使用訓練後更昂貴的模型來存儲，你也可能更喜歡全局模型。然而，如果在分析數據後，你認為系列之間沒有共同模式，系列的數量是可管理的，或者你不使用複雜模型，選擇本地模型可能是最好的。

你選擇了哪些庫和模型？

有很多選擇可以處理時間序列。在這個教程中，我建議使用 Darts。假設你正在使用 Python，這個預測庫非常易於使用。它提供了管理時間序列數據、拆分數據、管理分組時間序列和執行不同分析的工具。它提供了各種全局和本地模型，因此你可以在不切換庫的情況下進行實驗。可用選項的例子包括基準模型、統計模型如 ARIMA 或 Prophet、基於 Scikit-learn 的模型、基於 Pytorch 的模型和集成模型。有趣的選擇包括 Temporal Fusion Transformer (TFT) 或 Time Series Deep Encoder (TiDE) 模型，這些模型可以學習分組系列之間的模式，支持類別協變量。

讓我們開始編碼吧！

開始使用不同的 Darts 模型的第一步是將 Pandas 數據框轉換為時間序列 Darts 對象並正確拆分。為此，我實現了兩個不同的函數，使用 Darts 的功能執行這些操作。價格、折扣和事件的特徵在預測時將是已知的，而計算的特徵我們只會知道過去的值。

# Darts 格式
series_raw, series, past_cov, future_cov = to_darts_time_series_group(dataset=dataset,target=’Volume’,time_col=’Date’,group_cols=[‘Agency’,’Product’],past_cols=[‘EMA_4′,’MA_4’],future_cols=[‘Price’,’Discount’,’Special_days’],freq=’MS’, # 每個月的第一天
encode_static_cov=True, # 使模型可以使用類別變量（代理商和產品））

# 拆分
train_val, test = split_grouped_darts_time_series(series=series,split_date=TEST_DATE)

train, _ = split_grouped_darts_time_series(series=train_val,split_date=VAL_DATE_OUT)

_, val = split_grouped_darts_time_series(series=train_val,split_date=VAL_DATE_IN)

# 我們將使用的第一個模型是 NaiveMovingAverage 基準模型，我們將其與其餘模型進行比較。這個模型非常快速，因為它不學習任何模式，只是根據輸入和輸出維度執行移動平均預測。

maes_baseline, time_baseline, preds_baseline = eval_local_model(train_val, test, NaiveMovingAverage, mae, prediction_horizon=6, input_chunk_length=12)

通常，在深入學習之前，你會嘗試使用更簡單和成本更低的模型，但在這個教程中，我想專注於兩個特別的深度學習模型，這些模型對我來說效果很好。我使用這兩個模型預測了數百種產品在多個商店的需求，使用每日聚合的銷售數據和不同的靜態和連續協變量，以及庫存數據。值得注意的是，這些模型在長期預測中特別表現更好。

第一個模型是 Temporal Fusion Transformer。這個模型允許你同時處理大量時間序列（即它是一個全局模型），並且在協變量方面非常靈活。它可以處理靜態、過去（僅在過去已知的值）和未來（過去和未來都已知的值）協變量。它能夠學習複雜的模式，並支持概率預測。唯一的缺點是，雖然它經過良好優化，但調整和訓練的成本可能很高。根據我的經驗，它可以給出非常好的結果，但調整超參數的過程如果資源有限會花費太多時間。在這個教程中，我們將使用大多數默認參數來訓練 TFT，並使用與基準模型相同的輸入和輸出窗口。

# PyTorch Lightning 訓練器參數
early_stopping_args = {“monitor”: “val_loss”,”patience”: 50,”min_delta”: 1e-3,”mode”: “min”,}
pl_trainer_kwargs = {“max_epochs”: 200,#”accelerator”: “gpu”, # 若要使用 GPU，請取消註解
“callbacks”: [EarlyStopping(**early_stopping_args)],”enable_progress_bar”:True}

common_model_args = {“output_chunk_length”: 6,”input_chunk_length”: 12,”pl_trainer_kwargs”: pl_trainer_kwargs,”save_checkpoints”: True, # 檢查點以檢索最佳模型狀態
“force_reset”: True,”batch_size”: 128,”random_state”: 42,}

# TFT 參數
best_hp = {‘optimizer_kwargs’: {‘lr’:0.0001},’loss_fn’: MAELoss(),’use_reversible_instance_norm’: True,’add_encoders’:encoders,}

# 訓練
start = time.time()## 註釋以加載預訓練模型
fit_mixed_covariates_model(model_cls = TFTModel,common_model_args = common_model_args,specific_model_args = best_hp,model_name = ‘TFT_model’,past_cov = past_cov,future_cov = future_cov,train_series = train,val_series = val,)
time_tft = time.time() – start

# 預測
best_tft = TFTModel.load_from_checkpoint(model_name=’TFT_model’, best=True)
preds_tft = best_tft.predict(series = train_val,past_covariates = past_cov,future_covariates = future_cov,n = 6)

第二個模型是 Time Series Deep Encoder。這個模型比 TFT 更新一些，並且是用密集層而不是 LSTM 層構建的，這使得模型的訓練時間大大縮短。Darts 的實現也支持所有類型的協變量和概率預測，以及多個時間序列。這個模型的論文顯示，它可以在預測基準上與基於變壓器的模型相匹配或超越。在我的情況下，由於調整成本較低，我在相同或更短的時間內獲得了比 TFT 模型更好的結果。再次強調，在這個教程中，我們僅進行第一次運行，主要使用默認參數。請注意，對於 TiDE，所需的訓練輪次通常少於 TFT。

# PyTorch Lightning 訓練器參數
early_stopping_args = {“monitor”: “val_loss”,”patience”: 10,”min_delta”: 1e-3,”mode”: “min”,}
pl_trainer_kwargs = {“max_epochs”: 50,#”accelerator”: “gpu”, # 若要使用 GPU，請取消註解
“callbacks”: [EarlyStopping(**early_stopping_args)],”enable_progress_bar”:True}

common_model_args = {“output_chunk_length”: 6,”input_chunk_length”: 12,”pl_trainer_kwargs”: pl_trainer_kwargs,”save_checkpoints”: True, # 檢查點以檢索最佳模型狀態
“force_reset”: True,”batch_size”: 128,”random_state”: 42,}

# TiDE 參數
best_hp = {‘optimizer_kwargs’: {‘lr’:0.0001},’loss_fn’: MAELoss(),’use_layer_norm’: True,’use_reversible_instance_norm’: True,’add_encoders’:encoders,}

# 訓練
start = time.time()## 註釋以加載預訓練模型
fit_mixed_covariates_model(model_cls = TiDEModel,common_model_args = common_model_args,specific_model_args = best_hp,model_name = ‘TiDE_model’,past_cov = past_cov,future_cov = future_cov,train_series = train,val_series = val,)
time_tide = time.time() – start

# 預測
best_tide = TiDEModel.load_from_checkpoint(model_name=’TiDE_model’, best=True)
preds_tide = best_tide.predict(series = train_val,past_covariates = past_cov,future_covariates = future_cov,n = 6)

你如何評估模型的性能？

雖然典型的時間序列指標對評估模型的預測能力很有用，但建議更進一步。首先，在評估測試集時，你應該丟棄所有有缺貨的系列，因為你不會將預測與實際數據進行比較。其次，將領域知識或關鍵績效指標（KPI）納入評估也很有趣。一個關鍵指標可能是你用模型賺了多少錢，避免缺貨。另一個關鍵指標可能是你通過避免過量庫存短保質期產品節省了多少錢。根據價格的穩定性，你甚至可以用自定義損失函數訓練模型，例如價格加權的平均絕對誤差（MAE）損失。

你的模型預測會隨著時間惡化嗎？

將數據分為訓練、驗證和測試拆分並不足以評估可能進入生產的模型的性能。僅僅在測試集上評估短時間窗口，模型選擇會受到特定預測窗口的影響。Darts 提供了一個易於使用的回測實現，允許你通過預測移動時間窗口來模擬模型隨時間的性能。通過回測，你還可以模擬每 N 步驟重新訓練模型。

讓我們開始編碼吧！

如果我們查看所有系列的 MAE 結果，可以看到明顯的贏家是 TiDE，因為它能夠在保持時間成本相對較低的情況下，最大程度地減少基準的誤差。然而，假設我們的啤酒公司的最佳利益是平等地減少缺貨和過量庫存的金錢成本。在這種情況下，我們可以使用價格加權的 MAE 來評估預測。

在計算所有系列的價格加權 MAE 後，TiDE 仍然是最佳模型，儘管可能會有所不同。如果我們計算使用 TiDE 相對於基準模型的改進，MAE 的改善為 6.11%，但在金錢成本方面，改善略微增加。反過來，當查看使用 TFT 的改進時，當僅考慮銷售量而不是價格計算時，改善會更大。

對於這個數據集，我們沒有使用回測來比較預測，因為數據量有限，因為它是每月聚合的。然而，我鼓勵你在可能的情況下對你的項目進行回測。在源代碼中，我包括這個函數，以便輕鬆地使用 Darts 進行回測：

def backtesting(model, series, past_cov, future_cov, start_date, horizon, stride):
historical_backtest = model.historical_forecasts(series, past_cov, future_cov,start=start_date,forecast_horizon=horizon,stride=stride, # 每 N 個月預測
retrain=False, # 保持模型固定（不重新訓練）
overlap_end=False, last_points_only=False )
maes = model.backtest(series, historical_forecasts=historical_backtest, metric=mae)
return np.mean(maes)

你將如何提供預測？

在這個教程中，假設你已經在使用預定的預測時間範圍和頻率。如果這沒有提供，這也是一個單獨的用例，應考慮交付或供應商的交貨時間。知道你的模型預測需要多頻繁是很重要的，因為這可能需要不同的自動化水平。如果你的公司每兩個月需要預測，也許在這項任務的自動化上投入時間、金錢和資源就沒有必要。然而，如果你的公司每週需要預測兩次，而你的模型需要更長的時間來進行這些預測，自動化過程可以節省未來的努力。

你會將模型部署到公司的雲服務中嗎？

根據之前的建議，如果你和你的公司決定部署模型並將其投入生產，遵循 MLOps 原則是個好主意。這將允許任何人在未來輕鬆進行更改，而不會干擾整個系統。此外，一旦投入生產，監控模型的性能也很重要，因為概念漂移或數據漂移可能會發生。如今，許多雲服務提供管理機器學習模型的開發、部署和監控的工具。例如 Azure Machine Learning 和 Amazon Web Services。