
作者 | 崔皓
審校 | 重樓
預測模型的困境
在 IT 運維的預測場景中,NeuralProphet 憑借出色的自回歸能力證明了價值 —— 只需依托 CPU 利用率自身的歷史數據,它就能精準預測未來時段的指標波動,為運維工程師提前規避故障風險提供了有力支撐。這種基于目標指標自身規律的預測方式,在真實運維場景中展現出了極高的實用性。
不過,若深入剖析 NeuralProphet 的預測邏輯便會發現,它的核心預測能力來源于趨勢(T (t))、季節性(S (t))、事件(E (t))、未來回歸量(F (t))、自回歸(A (t))與滯后回歸量(L (t))六大組件的協同作用。

組件符號 | 簡單描述 |
T(t) | 趨勢 (Trend):數據隨時間變化的長期基礎走向(例如,增長或下降)。 |
S(t) | 季節性 (Seasonality):在固定周期內重復出現的模式(例如,每周、每年的周期)。 |
E(t) | 事件 (Events):特殊日期(如節假日)對數據產生的短期影響。 |
F(t) | 未來回歸量 (Future Regressors):其未來值已知的外部變量(例如,已計劃的營銷活動)。 |
A(t) | 自回歸 (Auto-Regression):近期歷史觀測值對未來值的直接影響。 |
L(t) | 滯后回歸量 (Lagged Regressors):其未來值未知的外部變量(例如,昨天的天氣)。 |
而進一步觀察各組件的特性會發現:
- T(t)趨勢組件反映的是數據長期走向,受業務發展等客觀因素制約,基本無法人為調整;
- S(t)季節性組件捕捉的是每日、每周的周期性規律,是系統運行中形成的固定模式,同樣難以改變;
- E(t)事件組件針對的是節假日等特殊日期的影響,這類日期是既定事實,不存在修改空間;
- F(t)未來回歸量依賴已知的未來外部變量,如周末這類固定時間節點,也屬于客觀存在的信息;
- A(t)自回歸組件則完全依托目標指標自身的歷史數據,數據的展現形式由過往系統運行狀態決定,可調性極低。
在所有組件中,唯有L(t)滯后回歸量顯得格外特殊:它依賴與預測值相關的外部指標歷史數據,就像機器學習中的 “特征” 一樣,具有極強的可選擇性。舉個生活中的例子,判斷一個西瓜是否甘甜,會涉及瓜的紋路、敲瓜的聲音、瓜蒂的形狀、產地、日照時間、品種等眾多特征,可選擇的維度極為豐富;回到 IT 運維場景,預測 CPU 使用率時,磁盤使用率、內存使用率、網卡流入量、網卡流出量等外部指標,都可能與 CPU 使用率存在關聯,都可作為滯后回歸量的備選 “特征”。
但在實際預測過程中,推理時間和數據量都是有限的,顯然無法將所有潛在的外部指標都納入滯后回歸量的范疇。這就帶來了一個關鍵問題:在眾多與目標指標相關的外部指標中,哪些對預測結果的影響最大?以 CPU 使用率預測為例,僅磁盤使用率、內存使用率、網卡流入量、網卡流出量這四項指標,就可衍生出 16 種不同的滯后回歸量組合,若后續增加更多備選指標,組合數量還會呈指數級增長。

因此,我們必須通過科學的方法評估每種滯后回歸量組合對預測結果的影響 —— 具體來說,就是計算每種組合下預測值的 MAE(平均絕對誤差),并將其與不引入任何外部指標的基準 MAE 進行對比,最終選擇 MAE 最低且低于基準值的組合,作為預測時使用的滯后回歸量組合。
如下圖所示,黃色的五角星就是基準的 MAE,在黃色五角星下面的綠色長方形包裹的藍點就是滿足條件的特征組合,這些藍點中最低的那一個點,我們把它標注為紅色,這個點就是最小 MAE 的組合點。換句話說,這個紅色點的特征作為滯后回歸量放到 NeuralProphet 的模型中進行預測時,其預測值與真實值的誤差是最小的。

本次實驗就是要教大家如何在眾多特征中找到最適合的組合,從而讓NeuralProphet 預測的數值更加準確。
創建數據集
在理解了整個實踐要完成的任務之后,我們來準備實驗所需要的數據,先確定數據結構,如下圖所示。

數據集完全貼合 NeuralProphet 的輸入要求,核心字段分為三類:
- 必需字段:
a.Ds:時間戳,記錄每個數據點的采集時間(如2025-08-01 00:00:00),是時間序列的 “錨點”。
b.y:目標預測值,這里指CPU 使用率(百分比),是我們最終要精準預測的指標。
- 未來回歸量字段:
is_weekend:標記是否為周末(0 = 工作日,1 = 周末)。這類字段的 “未來值是已知的”,比如預測下周一的 CPU 使用率,提前就知道下周一不是周末,因此能作為 “已知未來的外部變量” 輔助預測。
- 滯后回歸量字段:
a.Memory:內存利用率(百分比)。
b.Disk:磁盤利用率(百分比)。
c.Net_in:網絡流入量(單位可自定義,如 Mbps)。
d.Net_out:網絡流出量(單位可自定義,如 Mbps)。
這四類字段是我們本次實驗的核心變量:它們是 “未來值未知、但歷史值已知的外部指標”,將作為滯后回歸量的候選,用于測試不同組合對 CPU 預測精度的影響。
接下來我們需要生成這些測試數據,這里會利用 Python 來完成。在生成數據之前先整理一下思路,實驗的目的是要看上面四個特征的組合與預測結果的相關程度,那么在創建數據的時候,就需要預先建立他們之間的相關性。
需要注意的是,這里只是為了實驗才會創建類似的數據,在生產環境中我們只需要獲取真實數據就可以了。這里為了體現實驗效果,會按照“CPU 強/弱”的方式分別對不同的特征進行設置。
- memory,與CPU強相關
- disk,與CPU弱相關、趨勢增長
- net_in 與CPU強相關
- net_out 與CPU中等相關
在 python 腳本中我們會分別利用不同的函數生成對應的特征數據,我們以一個函數來觀察數據是如何生成的,該函數用來生成 network_in 的特征,代碼如下:
def create_network_in(cpu_series):
# 設置滯后周期為1,表示網絡輸入流量受前一個時間點的CPU影響
lag = 1
# 將CPU序列向后滾動lag個位置,獲取滯后數據
lagged_cpu = np.roll(cpu_series, lag)
# 處理滾動后的前lag個數據點,用平均值填充
lagged_cpu[:lag] = np.mean(cpu_series)
# 基礎流量:1 + 日周期模式(振幅5,相位偏移0.8)
base = 1 + daily_pattern(minutes_from_start, amplitude=5, phase_shift=0.8)
# CPU效應:標準化滯后CPU數據,乘以系數0.8和縮放因子4
cpu_effect = 0.8 * (lagged_cpu - np.mean(lagged_cpu)) / np.std(lagged_cpu) * 4
# 計算基礎網絡流量 = 基礎流量 + CPU效應
net = base + cpu_effect
# 工作日/周末效應:工作日流量×1.2,周末流量×0.9
net *= np.where(weekday < 5, 1.2, 0.9)
# 創建午餐時間掩碼(11點至13點)
lunch_mask = ((hours >= 11) & (hours <= 13))
# 午餐時間流量增加30%
net[lunch_mask] *= 1.3
# 尖峰掩碼:滯后CPU尖峰位置或隨機生成新尖峰(概率0.2%)
net_spike_mask = np.roll(spike_mask, lag) | (np.random.rand(periods) < 0.002)
# 在尖峰位置添加20-150的隨機流量增量
net[net_spike_mask] += np.random.uniform(20, 150, size=np.sum(net_spike_mask))
# 確保網絡流量不為負值
return np.maximum(net, 0)上面代碼都有對應的注釋,理解起來應該不難,下面對幾個重要的部分進行解釋。
CPU 滯后效應:捕捉 “CPU→網絡流入” 的關聯
- 先設置滯后周期 lag=1,表示 “網絡流入量受前一個時間點的 CPU 使用率影響”。
- 通過 np.roll 將 CPU 序列向后滾動 1 個位置,得到lagged_cpu(前 1 個時間點的 CPU 數據),并用 CPU 序列的平均值填充前 1 個缺失值。
- 對lagged_cpu做標準化處理(減去均值、除以標準差),再乘以系數和縮放因子,最終得到 “CPU 對網絡流入的影響量(cpu_effect)”。
基礎流量與周期模式:模擬時間規律
- 基礎流量:由 “固定值 1 + 日周期模式(daily_pattern)” 構成,daily_pattern 是一個自定義函數(代碼中未展示,推測是生成日維度的流量波動,如早高峰、晚高峰),振幅為 5,相位偏移 0.8。
- 工作日 / 周末效應:通過 np.where 判斷是否為工作日(weekday < 5),工作日流量乘以 1.2,周末乘以 0.9,模擬 “工作日業務請求多、網絡流入高;周末請求少、流量低” 的場景。
- 午餐時間效應:創建lunch_mask掩碼(11 點 - 13 點),在此時間段內流量額外增加 30%,模擬 “午餐時段業務交互(如員工操作系統)帶來的流量小高峰”。
突發尖峰:模擬異常流量場景
- 創建net_spike_mask掩碼:一方面繼承 CPU 尖峰的滯后影響(np.roll(spike_mask, lag)),另一方面以 0.2% 的概率隨機生成新尖峰,模擬 “網絡突發流量(如 DDOS 攻擊、突發業務請求)”。
- 在尖峰位置添加 20-150 的隨機增量,讓流量波動更貼近真實運維中 “突發流量沖垮帶寬” 的場景。
按照上述的方式對 4 個特征數據進行生成操作,最終將數據寫入到 dataset.xlsx 的文件中,由于代碼內容比較多,就不在復制粘貼了。如下圖所示,將代碼命名為 gen_feature_data.py 文件。

然后,執行如下命令,生成 xlsx 的數據集文件。
python gen_feature_data.py執行命令之后,會在 py 文件相同的目錄下創建 xlsx 文件。
組合滯后回歸量判斷誤差
準備好測試數據之后,接著需要編寫代碼對不同組合進行評估。
這里我們會通過 Python 代碼實現整體功能,在執行代碼執行前,通過下圖,整理思路。

評估組合的核心邏輯:
1. 入口層(main):作為程序總控,依次調用 “參數解析”“數據加載”“核心分析” 三大模塊,確保流程按序執行。
2. 準備層(parse_arguments + load_dataset):前者處理用戶輸入的命令行參數,明確數據來源和輸出要求;后者負責讀取數據并處理異常,為后續分析提供合規的原始數據。
3. 核心層(run_analysis):是整個程序的 “中樞”,一方面調用generate_combinations生成所有滯后回歸量組合、process_data預處理數據,另一方面內部完成模型訓練評估,最后調用highlight_no_variable_combination優化結果輸出格式。
核心代碼如下:
import pandas as pd
import warnings
import matplotlib.pyplot as plt
import itertools
import numpy as np
import traceback
import sys
import argparse
from neuralprophet import NeuralProphet, set_log_level
from pathlib import Path
from openpyxl import load_workbook
from openpyxl.styles import PatternFill
# 設置中文顯示
plt.rcParams['font.family'] = 'STHeiti'
plt.rcParams['axes.unicode_minus'] = False
# 忽略警告
warnings.filterwarnings('ignore')
def load_dataset(file_path):
"""
加載數據集,支持CSV和Excel格式
參數:
file_path (str): 數據集文件路徑
返回:
pandas.DataFrame: 加載后的數據集
異常:
ValueError: 當文件格式不支持時拋出
FileNotFoundError: 當文件不存在時拋出
Exception: 其他加載錯誤
"""
# 檢查文件是否存在
if not Path(file_path).exists():
raise FileNotFoundError(f"數據集文件不存在: {file_path}")
# 根據文件擴展名選擇合適的讀取方法
file_ext = Path(file_path).suffix.lower()
try:
if file_ext == '.csv':
return pd.read_csv(file_path)
elif file_ext in ['.xlsx', '.xls']:
return pd.read_excel(file_path)
else:
raise ValueError(f"不支持的文件格式: {file_ext}。請使用CSV(.csv)或Excel(.xlsx, .xls)格式。")
except Exception as e:
raise Exception(f"加載數據集失敗: {str(e)}")
def process_data(df, future_regressors):
"""
數據預處理函數:重命名列名并轉換日期格式
參數:
df (pandas.DataFrame): 原始數據集
future_regressors (list): 未來回歸量列表
返回:
pandas.DataFrame: 預處理后的數據集
"""
# 重命名必要的列(NeuralProphet要求的標準列名)
df = df.rename(columns={"日期": "ds", "實際值": "y"})
# 確保日期列是datetime類型
df['ds'] = pd.to_datetime(df['ds'], errors='coerce')
# 檢查日期轉換是否有問題
if df['ds'].isna().any():
print("警告: 部分日期轉換失敗,已設置為NaT")
return df
def generate_combinations(other_variables):
"""
生成所有可能的變量組合(從0個到所有變量)
參數:
other_variables (list): 待組合的變量列表
返回:
list: 所有可能的變量組合,每個組合以元組形式表示
"""
variable_combinations = []
for i in range(0, len(other_variables) + 1):
variable_combinations.extend(list(itertools.combinations(other_variables, i)))
return variable_combinations
def print_combinations(variable_combinations):
"""
打印變量組合信息,顯示前5個和后5個組合(避免過多輸出)
參數:
variable_combinations (list): 變量組合列表
"""
total = len(variable_combinations)
print(f"生成的所有可能組合數量: {total}")
# 只顯示前5個和后5個組合
display_count = 5
# 顯示前5個組合
for combo in variable_combinations[:display_count]:
print(f" 組合示例: {combo}")
# 如果組合數量超過10個,顯示省略號
if total > 2 * display_count:
print(" ...")
# 顯示后5個組合(如果有足夠的組合)
if total > display_count:
start_idx = max(total - display_count, display_count)
for combo in variable_combinations[start_idx:]:
print(f" 組合示例: {combo}")
def highlight_no_variable_combination(output_file):
"""
高亮顯示Excel中"無變量"組合的行
參數:
output_file (str): Excel文件路徑
"""
# 加載Excel文件
wb = load_workbook(output_file)
ws = wb.active
# 定義紅色填充樣式
red_fill = PatternFill(start_color="FFC7CE", end_color="FFC7CE", fill_type="solid")
# 查找"無變量"的行并標紅
for row in ws.iter_rows(min_row=2): # 從第2行開始,跳過表頭
if row[0].value == "無變量": # 第一列是組合名稱
for cell in row:
cell.fill = red_fill
# 保存修改
wb.save(output_file)
def run_analysis(df, future_regressors=None, output_file="變量組合預測結果.xlsx"):
"""
運行主分析流程:生成變量組合、訓練模型并評估性能
參數:
df (pandas.DataFrame): 預處理后的數據集
future_regressors (list, 可選): 未來回歸量列表,默認為空列表
output_file (str, 可選): 結果輸出文件路徑,默認為"變量組合預測結果.xlsx"
返回:
list: 每個變量組合的評估結果,包含組合名稱、變量數量、MAE和MAPE
"""
# 設置默認的未來回歸量為空列表
if future_regressors is None:
future_regressors = []
# 顯示當前使用的未來回歸量
if not future_regressors:
print("未使用任何未來回歸量")
else:
print(f"使用的未來回歸量: {future_regressors}")
# 識別其他變量(潛在的滯后回歸量)
# 排除日期列、目標列和未來回歸量
other_variables = [col for col in df.columns if col not in ["ds", "y"] + future_regressors]
print(f"潛在的滯后回歸量: {other_variables}")
print(f"滯后回歸量總數: {len(other_variables)}")
# 生成所有可能的變量組合
variable_combinations = generate_combinations(other_variables)
# 打印組合信息
print_combinations(variable_combinations)
# 只顯示錯誤日志
set_log_level("ERROR")
# 存儲每個組合的結果
results = []
# 遍歷所有變量組合
for i, combo in enumerate(variable_combinations, 1):
# 處理空組合的顯示
combo_desc = "無變量" if len(combo) == 0 else combo
print(f"\n正在處理組合 {i}/{len(variable_combinations)}: {combo_desc}")
# 創建NeuralProphet模型實例
m = NeuralProphet(
changepoints_range=0.8,
trend_reg=1,
yearly_seasonality=False,
weekly_seasonality=5,
daily_seasonality=False,
seasonality_reg=0.5,
n_lags=7,
ar_reg=0.7,
collect_metrics={ "MAE": "MeanAbsoluteError", "MAPE": "MeanAbsolutePercentageError" },
)
# 添加未來回歸量(如果有)
for var in future_regressors:
if var in df.columns:
# 確保數據是數值型
if not pd.api.types.is_numeric_dtype(df[var]):
print(f"轉換 {var} 為數值型")
df[var] = pd.to_numeric(df[var], errors='coerce')
m.add_future_regressor(var)
else:
print(f"警告: 未來回歸量 '{var}' 不在數據集中,已跳過")
# 添加滯后回歸量(處理空組合情況)
if len(combo) > 0:
for var in combo:
if var in df.columns:
# 確保數據是數值型
if not pd.api.types.is_numeric_dtype(df[var]):
print(f"轉換 {var} 為數值型")
df[var] = pd.to_numeric(df[var], errors='coerce')
m.add_lagged_regressor(var, n_lags=1)
else:
print(f"警告: 滯后回歸量 '{var}' 不在數據集中,已跳過")
else:
print("此組合不包含任何滯后回歸量")
# 設置繪圖后端
m.set_plotting_backend("plotly")
# 準備數據 - 只保留需要的列
required_columns = ["ds", "y"] + future_regressors + list(combo)
available_columns = [col for col in required_columns if col in df.columns]
df_filtered = df[available_columns].copy()
# 處理缺失值
df_filtered = df_filtered.fillna(df_filtered.mean(numeric_only=True))
# 確保所有數值列都是float32類型
numeric_cols = df_filtered.select_dtypes(include=['number']).columns
df_filtered[numeric_cols] = df_filtered[numeric_cols].astype(np.float32)
# 分割訓練集和驗證集
df_train, df_val = m.split_df(df_filtered, valid_p=0.2)
# 擬合模型
try:
# 訓練模型
metrics = m.fit(
df=df_train,
validation_df=df_val,
freq="auto"
)
# 預測階段
df_pred = df_val[available_columns].copy()
predictions = m.predict(df_pred)
# 提取驗證集的最終指標
final_mae = metrics['MAE_val'].iloc[-1]
final_mape = metrics['MAPE_val'].iloc[-1] * 100 # 轉換為百分比
print(f"組合 {combo_desc} 的驗證集MAE: {final_mae:.2f}")
print(f"組合 {combo_desc} 的驗證集MAPE: {final_mape:.2f}%")
# 保存結果
results.append({
'combination': '無變量' if len(combo) == 0 else ', '.join(combo),
'variables_count': len(combo),
'MAE': final_mae,
'MAPE_percent': final_mape
})
except Exception as e:
print(f"處理組合 {combo_desc} 時出錯: {str(e)}")
# 打印詳細的錯誤跟蹤信息以便調試
print("詳細錯誤信息:")
traceback.print_exc()
continue
# 處理并展示結果
if results:
results_df = pd.DataFrame(results)
# 按MAPE升序排序(最優的結果排在前面)
results_df = results_df.sort_values(by='MAPE_percent')
# 保存結果到Excel(尚未標紅)
results_df.to_excel(output_file, index=False)
# 高亮顯示"無變量"組合的行
highlight_no_variable_combination(output_file)
print(f"\n所有組合的評估結果已保存到 '{output_file}'")
print("注意:Excel中已將'無變量'組合的行標為紅色")
# 顯示前10個最佳組合
print("\n表現最佳的前10個變量組合:")
print(results_df.head(10).to_string(index=False))
# 可視化結果
plt.figure(figsize=(12, 8))
# 獲取無變量組合的性能指標
no_var_row = results_df[results_df['combination'] == '無變量']
if not no_var_row.empty:
no_var_mae = no_var_row['MAE'].values[0]
no_var_mape = no_var_row['MAPE_percent'].values[0]
# 繪制優于無變量結果的區域(MAE更低且MAPE更低)
# 確定繪圖邊界
max_mae = results_df['MAE'].max()
max_mape = results_df['MAPE_percent'].max()
# 填充區域:MAE < 無變量MAE 且 MAPE < 無變量MAPE
plt.fill_between(
[0, no_var_mae],
[0, 0],
[no_var_mape, no_var_mape],
color='lightgreen',
alpha=0.3,
label='優于無變量組合的區域'
)
# 繪制所有組合
plt.scatter(results_df['MAE'], results_df['MAPE_percent'], alpha=0.6, label='所有變量組合')
# 如果存在無變量組合,單獨標記
if not no_var_row.empty:
plt.scatter(
no_var_mae,
no_var_mape,
color='orange',
s=120,
marker='*',
label=f'無變量組合\n(MAE: {no_var_mae:.2f}, MAPE: {no_var_mape:.2f}%)'
)
# 標記最佳組合
best_idx = results_df['MAPE_percent'].idxmin()
plt.scatter(
results_df.loc[best_idx, 'MAE'],
results_df.loc[best_idx, 'MAPE_percent'],
color='red', s=100, label=f'最佳組合: {results_df.loc[best_idx, "combination"]}'
)
# 圖表設置
plt.xlabel('MAE (平均絕對誤差)')
plt.ylabel('MAPE (%)')
plt.title('不同變量組合的預測性能比較')
plt.grid(True, linestyle='--', alpha=0.7)
plt.legend()
plt.tight_layout()
plt.show()
else:
print("\n沒有成功處理任何變量組合")
return results
def parse_arguments():
"""
解析命令行參數
返回:
argparse.Namespace: 解析后的參數對象
"""
# 設置中文幫助信息
parser = argparse.ArgumentParser(
description='NeuralProphet變量組合測試工具',
formatter_class=argparse.RawTextHelpFormatter,
usage='''np_lagged_regressors.py [選項] 數據集路徑
示例:
np_lagged_regressors.py 數據.csv
np_lagged_regressors.py --regressors 是否有定時任務 數據.xlsx
np_lagged_regressors.py --output 結果.xlsx 數據.csv'''
)
parser.add_argument('dataset', help='數據集文件路徑(支持.csv, .xlsx, .xls格式)')
parser.add_argument('--regressors', nargs='*', default=[],
help='未來回歸量列表,多個變量用空格分隔\n例如:--regressors 最高溫度 最低溫度')
parser.add_argument('--output', default='變量組合預測結果.xlsx',
help='結果輸出文件路徑,默認為"變量組合預測結果.xlsx"')
return parser
def main():
"""主函數:解析參數并執行分析"""
parser = parse_arguments()
try:
args = parser.parse_args()
except argparse.ArgumentError as e:
print(f"錯誤: {e}\n")
parser.print_help()
sys.exit(1)
# 檢查是否提供了數據集路徑(argparse會自動檢查,這里做額外友好提示)
if len(sys.argv) == 1:
print("錯誤:必須提供數據集文件路徑!\n")
parser.print_help()
sys.exit(1)
# 讀取數據集
try:
df = load_dataset(args.dataset)
print(f"成功加載數據集: {args.dataset}")
except Exception as e:
print(f"錯誤: {str(e)}")
sys.exit(1)
# 運行分析,傳入從命令行獲取的未來回歸量
run_analysis(df, future_regressors=args.regressors, output_file=args.output)
if __name__ == "__main__":
"""程序入口點"""
main()
上面代碼很長,不過我們只需盯著run_analysis 函數,它是組合評估的核心函數。該函數接收預處理后的數據集、未來回歸量列表和輸出文件路徑,通過枚舉所有可能的滯后回歸量組合,逐一訓練 NeuralProphet 模型并計算預測誤差(MAE、MAPE),最終篩選出最優組合并以表格和圖表形式輸出結果。其本質是通過 “控制變量法” 對比不同外部指標組合對預測精度的影響,為滯后回歸量的選擇提供數據支撐。
該函數大致分為如下幾個步驟:
初始化與變量識別
- 未來回歸量處理:若未指定未來回歸量(如is_weekend),則默認為空列表;若有指定,打印當前使用的未來回歸量(如 “使用的未來回歸量: ['is_weekend']”)。
- 滯后回歸量識別:從數據集中排除必需列(ds時間、y目標值)和未來回歸量,剩余列被標記為 “潛在的滯后回歸量”(如memory、disk等),并統計其數量。
生成滯后回歸量組合
調用generate_combinations函數生成所有可能的組合,包括 “0 個變量(無滯后回歸量)” 到 “所有變量全選” 的所有情況(例如 4 個變量時生成 15 種組合),并打印部分組合示例(前 5 個 + 后 5 個),方便用戶了解測試范圍。
遍歷組合訓練模型(核心循環)
對每一種滯后回歸量組合,執行以下操作:
- 模型初始化:創建 NeuralProphet 模型實例,配置趨勢(changepoints_range=0.8)、季節性(weekly_seasonality=5)、自回歸(n_lags=7)等參數,并指定需收集的評估指標(MAE、MAPE)。
- 添加回歸量:
a.未來回歸量:若列表中有值(如is_weekend),將其添加到模型中(需確保為數值型,非數值則轉換)。
b.滯后回歸量:根據當前組合添加對應的外部指標(如memory+net_in),同樣確保數據為數值型。
- 數據預處理:
a.篩選當前組合所需的列(ds、y、未來回歸量、當前滯后回歸量組合),排除無關列以減少冗余。
b.填充缺失值(用均值填充),并將所有數值列轉換為float32類型,提升模型訓練效率。
- 訓練與評估:
a.按 8:2 比例分割訓練集和驗證集。
b.用訓練集擬合模型,用驗證集評估性能,提取最終的 MAE(平均絕對誤差)和 MAPE(平均絕對百分比誤差,轉換為百分比)。
c.若訓練出錯(如數據格式問題),打印錯誤信息并跳過當前組合,繼續下一組測試。
結果處理與可視化
- 結果整理:將所有組合的 MAE、MAPE 及變量數量整理成 DataFrame,按 MAPE 升序排序(值越小表示預測越精準)。
- 輸出與標記:
a.將結果保存到 Excel,并用highlight_no_variable_combination函數將 “無變量組合”(基準組)標紅,方便對比。
b.打印表現最佳的前 10 個組合,直觀展示最優選項。
- 可視化:
a.繪制散點圖,橫軸為 MAE、縱軸為 MAPE,每個點代表一個組合。
b.用綠色區域標記 “優于基準組(無變量)” 的區域(MAE 和 MAPE 均更低),用橙色星標標記基準組,用紅色點標記最優組合,讓結果差異一目了然。
測試組合評估
完成代碼編寫之后,將其保存到np_combinate.py 中,通過執行如下指令進行測試:
python np_combinate.py dataset.xlsx --regressors is_weekend在指令中傳入之前生成的測試數據集 dataset.xlsx,設置歷史回歸量 is_weekend。
執行之后,如下圖所示,會進行多輪組合的評估。

這個步驟比較花費時間,完成評估之后會自動彈出如下圖的圖標。從圖中可以看出,黃色的五角星是基準的 MAE,在黃色五角星下面的綠色長方形包裹的藍點就是滿足條件的特征組合,這些藍點中最低的那一個點,我們把它標注為紅色,這個點就是最小 MAE 的組合點。換句話說,這個紅色點的特征作為滯后回歸量放到 NeuralProphet 的模型中進行預測時,其預測值與真實值的誤差是最小的。

此時,在命令行輸出如下信息,列舉出 10 個最優的組合。(MAE 越小說明預測得越準確)

最后,程序會在同級目錄下生成“變量組合預測結果.xlsx” 文件,大家文件可以看到如下圖所示內容。

從圖中可以看出,無變量就是不考慮任何“滯后回歸量”時的 MAE 值,所有小于這個值的特征組合用紅色框體包括起來,在“無變量”的上方都是相對優秀的特征組合。
作者介紹
崔皓,51CTO社區編輯,資深架構師,擁有18年的軟件開發和架構經驗,10年分布式架構經驗。

























