基于非线性树模型的时间序列预测框架：全生命周期深度研究报告

import numpy as np  
import pandas as pd  
import lightgbm as lgb  
import mlflow  
import joblib  
from sklearn.multioutput import MultiOutputRegressor  
from sklearn.model_selection import TimeSeriesSplit  
from sklearn.metrics import mean_squared_error  
from datetime import timedelta

class ProductionTSForecaster:  
    """  
    支持递归(Recursive)与多输出(MIMO)策略的工业级时序框架  
    """  
    def __init__(self, strategy='recursive', horizon=7):  
        self.strategy = strategy  
        self.horizon = horizon  
        self.model = None  
        self.feature_cols =

    def engineer_features(self, df, target_col, lags=[1, 2, 3]):  
        """  
        原理：构建滞后特征。注意在训练集中必须删除包含未来信息的行。  
        """  
        data = df.copy()  
        # 1. 滞后特征 (Memory)  
        for lag in lags:  
            data[f'lag_{lag}'] = data[target_col].shift(lag)  
          
        # 2. 滚动统计 (Trend/Volatility)  
        data['rolling_mean_7'] = data[target_col].shift(1).rolling(7).mean()  
        data['rolling_std_7'] = data[target_col].shift(1).rolling(7).std()  
          
        # 3. 周期性编码 (Cycles)  
        data['dow_sin'] = np.sin(2 * np.pi * data.index.dayofweek / 7)  
        data['dow_cos'] = np.cos(2 * np.pi * data.index.dayofweek / 7)

        if self.strategy == 'mimo':  
            # 为MIMO准备多目标Label  
            for h in range(1, self.horizon + 1):  
                data[f'target_h{h}'] = data[target_col].shift(-h)  
          
        self.feature_cols = [c for c in data.columns if 'lag' in c or 'rolling' in c or 'dow' in c]  
        return data.dropna()

    def train_pipeline(self, df, target_col):  
        """  
        集成 MLflow 的训练流水线  
        """  
        data = self.engineer_features(df, target_col)  
        X = data[self.feature_cols]  
          
        mlflow.set_experiment("TS_Production_Project")  
        with mlflow.start_run():  
            tscv = TimeSeriesSplit(n_splits=3)  
            fold_errors =

            for fold, (train_idx, val_idx) in enumerate(tscv.split(X)):  
                X_train, X_val = X.iloc[train_idx], X.iloc[val_idx]  
                  
                if self.strategy == 'recursive':  
                    y_train, y_val = data[target_col].iloc[train_idx], data[target_col].iloc[val_idx]  
                    self.model = lgb.LGBMRegressor(n_estimators=200, importance_type='gain')  
                    self.model.fit(X_train, y_train, eval_set=[(X_val, y_val)], callbacks=[lgb.early_stopping(20)])  
                else:  
                    # MIMO 模式：训练多输出回归器  
                    target_list = [f'target_h{h}' for h in range(1, self.horizon+1)]  
                    y_train = data[target_list].iloc[train_idx]  
                    self.model = MultiOutputRegressor(lgb.LGBMRegressor(n_estimators=200))  
                    self.model.fit(X_train, y_train)

            # 保存模型与特征列表（生产部署必备）  
            artifacts = {'model': self.model, 'features': self.feature_cols}  
            joblib.dump(artifacts, "ts_model_pack.pkl")  
            mlflow.log_artifact("ts_model_pack.pkl")

    def predict(self, history_df):  
        """  
        推理逻辑：区分递归与直接输出  
        """  
        if self.strategy == 'mimo':  
            # 原理：一步输出整个向量，无累积误差  
            X_latest = self.engineer_features(history_df, 'sales').tail(1)[self.feature_cols]  
            return self.model.predict(X_latest).flatten()  
          
        else:  
            # 递归预测：原理是将t+1的预测值作为t+2的特征输入  
            current_data = history_df.copy()  
            preds =  
            for _ in range(self.horizon):  
                # 重新构建最后一行的特征  
                feat_df = self.engineer_features(current_data, 'sales')  
                X_input = feat_df.tail(1)[self.feature_cols]  
                p = self.model.predict(X_input)  
                preds.append(p)  
                  
                # 更新历史序列以进行下一步预测  
                new_date = current_data.index.max() + timedelta(days=1)  
                current_data.loc[new_date, 'sales'] = p  
            return np.array(preds)

# ==========================================  
# 模拟运行与测试  
# ==========================================  
if __name__ == "__main__":  
    # 生成带趋势和周季节性的数据  
    dates = pd.date_range('2024-01-01', periods=200)  
    y = np.linspace(0, 10, 200) + 5 * np.sin(np.arange(200) * (2*np.pi/7)) + np.random.normal(0, 1, 200)  
    df = pd.DataFrame({'sales': y}, index=dates)

    # 1. 运行递归策略  
    forecaster = ProductionTSForecaster(strategy='recursive', horizon=7)  
    forecaster.train_pipeline(df, 'sales')  
    res = forecaster.predict(df.tail(30))  
    print(f"未来7天递归预测结果: {res.round(2)}")