' 销售预测主模块Option Explicit' 预测方法枚举Enum ForecastMethod    SimpleMovingAverage    WeightedMovingAverage    ExponentialSmoothing    LinearRegression    SeasonalDecompositionEnd Enum' 预测结果结构Type ForecastResult    Period As String    Actual As Double    Forecast As Double    Error As Double    ErrorPercent As Double    ConfidenceLower As Double    ConfidenceUpper As DoubleEnd Type' 主预测函数Sub RunSalesForecast()    Dim startTime As Double    startTime = Timer    On Error GoTo ErrorHandler    Application.ScreenUpdating = False    Application.Calculation = xlCalculationManual    ' 1. 加载历史数据    Dim historicalData As Collection    Set historicalData = LoadHistoricalData("Sheet1")    If historicalData.Count = 0 Then        MsgBox "没有历史数据！", vbExclamation        Exit Sub    End If    LogMessage "加载历史数据完成，共" & historicalData.Count & "条记录"    ' 2. 数据预处理    Dim cleanedData As Collection    Set cleanedData = PreprocessData(historicalData)    LogMessage "数据预处理完成，有效数据" & cleanedData.Count & "条"    ' 3. 选择预测方法    Dim method As ForecastMethod    method = SelectForecastMethod(cleanedData)    ' 4. 执行预测    Dim forecastPeriods As Long    forecastPeriods = InputBox("请输入预测期数（例如预测未来3个月输入3）：", "预测期数", 3)    If forecastPeriods <= 0 Then Exit Sub    Dim forecastResults As Collection    Set forecastResults = ExecuteForecast(cleanedData, method, forecastPeriods)    ' 5. 评估预测结果    Dim evaluation As Dictionary    Set evaluation = EvaluateForecast(forecastResults)    ' 6. 生成预测报告    GenerateForecastReport forecastResults, evaluation, method    ' 7. 可视化结果    CreateForecastCharts forecastResults    Application.ScreenUpdating = True    Application.Calculation = xlCalculationAutomatic    Dim endTime As Double    endTime = Timer    MsgBox "预测完成！" & vbCrLf & _           "预测方法：" & GetMethodName(method) & vbCrLf & _           "预测期数：" & forecastPeriods & vbCrLf & _           "平均误差：" & Format(evaluation("MAE"), "0.00") & vbCrLf & _           "MAPE：" & Format(evaluation("MAPE"), "0.00%") & vbCrLf & _           "耗时：" & Format(endTime - startTime, "0.0") & "秒", _           vbInformation, "预测完成"    Exit SubErrorHandler:    Application.ScreenUpdating = True    Application.Calculation = xlCalculationAutomatic    MsgBox "预测过程中发生错误：" & vbCrLf & _           Err.Description, vbCritical, "错误"End Sub' 简单移动平均法Function SimpleMovingAverage(data As Collection, periods As Long) As Collection    Dim results As New Collection    If data.Count < periods Then        Set SimpleMovingAverage = results        Exit Function    End If    Dim i As Long, j As Long    Dim sum As Double    For i = periods To data.Count        sum = 0        For j = i - periods + 1 To i            sum = sum + data(j)("Sales")        Next j        Dim result As Dictionary        Set result = New Dictionary        result("Period") = data(i)("Period")        result("Actual") = data(i)("Sales")        result("Forecast") = sum / periods        result("Error") = data(i)("Sales") - result("Forecast")        If data(i)("Sales") > 0 Then            result("ErrorPercent") = Abs(result("Error") / data(i)("Sales"))        Else            result("ErrorPercent") = 0        End If        results.Add result    Next i    Set SimpleMovingAverage = resultsEnd Function' 加权移动平均法Function WeightedMovingAverage(data As Collection, periods As Long) As Collection    Dim results As New Collection    If data.Count < periods Then        Set WeightedMovingAverage = results        Exit Function    End If    ' 计算权重（最近的数据权重更大）    Dim weights() As Double    ReDim weights(1 To periods)    Dim totalWeight As Double    Dim k As Long    For k = 1 To periods        weights(k) = k  ' 线性权重        totalWeight = totalWeight + weights(k)    Next k    ' 归一化权重    For k = 1 To periods        weights(k) = weights(k) / totalWeight    Next k    Dim i As Long, j As Long    Dim weightedSum As Double    For i = periods To data.Count        weightedSum = 0        Dim weightIndex As Long        weightIndex = 1        For j = i - periods + 1 To i            weightedSum = weightedSum + data(j)("Sales") * weights(weightIndex)            weightIndex = weightIndex + 1        Next j        Dim result As Dictionary        Set result = New Dictionary        result("Period") = data(i)("Period")        result("Actual") = data(i)("Sales")        result("Forecast") = weightedSum        result("Error") = data(i)("Sales") - result("Forecast")        If data(i)("Sales") > 0 Then            result("ErrorPercent") = Abs(result("Error") / data(i)("Sales"))        Else            result("ErrorPercent") = 0        End If        results.Add result    Next i    Set WeightedMovingAverage = resultsEnd Function' 指数平滑法（单指数）Function ExponentialSmoothing(data As Collection, alpha As Double) As Collection    Dim results As New Collection    If data.Count < 2 Then        Set ExponentialSmoothing = results        Exit Function    End If    ' 初始化：第一个预测值等于第一个实际值    Dim previousForecast As Double    previousForecast = data(1)("Sales")    Dim i As Long    For i = 2 To data.Count        Dim forecast As Double        forecast = alpha * data(i)("Sales") + (1 - alpha) * previousForecast        Dim result As Dictionary        Set result = New Dictionary        result("Period") = data(i)("Period")        result("Actual") = data(i)("Sales")        result("Forecast") = forecast        result("Error") = data(i)("Sales") - forecast        If data(i)("Sales") > 0 Then            result("ErrorPercent") = Abs(result("Error") / data(i)("Sales"))        Else            result("ErrorPercent") = 0        End If        results.Add result        previousForecast = forecast    Next i    Set ExponentialSmoothing = resultsEnd Function' 线性回归预测Function LinearRegressionForecast(data As Collection, forecastPeriods As Long) As Collection    Dim results As New Collection    If data.Count < 2 Then        Set LinearRegressionForecast = results        Exit Function    End If    ' 计算回归系数 y = a + bx    Dim sumX As Double, sumY As Double, sumXY As Double, sumX2 As Double    Dim n As Long    n = data.Count    Dim i As Long    For i = 1 To n        sumX = sumX + i        sumY = sumY + data(i)("Sales")        sumXY = sumXY + i * data(i)("Sales")        sumX2 = sumX2 + i * i    Next i    Dim b As Double, a As Double    b = (n * sumXY - sumX * sumY) / (n * sumX2 - sumX * sumX)    a = (sumY - b * sumX) / n    ' 计算R平方    Dim ssTotal As Double, ssResidual As Double    Dim meanY As Double    meanY = sumY / n    For i = 1 To n        Dim yPred As Double        yPred = a + b * i        ssTotal = ssTotal + (data(i)("Sales") - meanY) ^ 2        ssResidual = ssResidual + (data(i)("Sales") - yPred) ^ 2    Next i    Dim rSquared As Double    If ssTotal > 0 Then        rSquared = 1 - ssResidual / ssTotal    Else        rSquared = 0    End If    ' 生成预测    For i = 1 To n + forecastPeriods        Dim forecastValue As Double        forecastValue = a + b * i        Dim result As Dictionary        Set result = New Dictionary        result("Period") = IIf(i <= n, data(i)("Period"), "F" & (i - n))        If i <= n Then            result("Actual") = data(i)("Sales")            result("Forecast") = forecastValue            result("Error") = data(i)("Sales") - forecastValue            If data(i)("Sales") > 0 Then                result("ErrorPercent") = Abs(result("Error") / data(i)("Sales"))            Else                result("ErrorPercent") = 0            End If        Else            result("Actual") = 0            result("Forecast") = forecastValue            result("Error") = 0            result("ErrorPercent") = 0        End If        results.Add result    Next i    ' 添加模型信息    Dim modelInfo As Dictionary    Set modelInfo = New Dictionary    modelInfo("Intercept") = a    modelInfo("Slope") = b    modelInfo("RSquared") = rSquared    modelInfo("Equation") = "y = " & Format(a, "0.00") & " + " & Format(b, "0.00") & "x"    results.Add modelInfo, "ModelInfo"    Set LinearRegressionForecast = resultsEnd Function' 季节分解法（基于移动平均）Function SeasonalDecomposition(data As Collection, seasonalPeriod As Long) As Collection    Dim results As New Collection    If data.Count < seasonalPeriod * 2 Then        Set SeasonalDecomposition = results        Exit Function    End If    ' 步骤1：计算移动平均（去季节化）    Dim movingAverages As Collection    Set movingAverages = New Collection    Dim i As Long, j As Long    Dim sum As Double    For i = seasonalPeriod To data.Count - seasonalPeriod + 1        sum = 0        For j = i - Int(seasonalPeriod / 2) To i + Int(seasonalPeriod / 2) - 1            sum = sum + data(j)("Sales")        Next j        Dim maValue As Double        maValue = sum / seasonalPeriod        movingAverages.Add maValue    Next i    ' 步骤2：计算季节指数    Dim seasonalIndices() As Double    ReDim seasonalIndices(1 To seasonalPeriod)    Dim indexCounts() As Long    ReDim indexCounts(1 To seasonalPeriod)    Dim maIndex As Long    maIndex = 1    For i = Int(seasonalPeriod / 2) + 1 To data.Count - Int(seasonalPeriod / 2)        Dim seasonalIndex As Long        seasonalIndex = ((i - 1) Mod seasonalPeriod) + 1        If data(i)("Sales") > 0 And movingAverages(maIndex) > 0 Then            seasonalIndices(seasonalIndex) = seasonalIndices(seasonalIndex) + data(i)("Sales") / movingAverages(maIndex)            indexCounts(seasonalIndex) = indexCounts(seasonalIndex) + 1        End If        maIndex = maIndex + 1    Next i    ' 平均季节指数    For i = 1 To seasonalPeriod        If indexCounts(i) > 0 Then            seasonalIndices(i) = seasonalIndices(i) / indexCounts(i)        End If    Next i    ' 步骤3：去季节化数据    Dim deseasonalized As Collection    Set deseasonalized = New Collection    For i = 1 To data.Count        Dim seasonalIndex2 As Long        seasonalIndex2 = ((i - 1) Mod seasonalPeriod) + 1        Dim deseasonValue As Double        If seasonalIndices(seasonalIndex2) > 0 Then            deseasonValue = data(i)("Sales") / seasonalIndices(seasonalIndex2)        Else            deseasonValue = data(i)("Sales")        End If        deseasonalized.Add deseasonValue    Next i    ' 步骤4：对去季节化数据进行趋势预测（使用线性回归）    Dim trendForecast As Collection    Set trendForecast = LinearRegressionForecast(CreateCollectionFromArray(deseasonalized), 0)    ' 步骤5：重新加入季节因素    For i = 1 To data.Count        Dim forecastValue As Double        forecastValue = trendForecast(i)("Forecast")        Dim seasonalIndex3 As Long        seasonalIndex3 = ((i - 1) Mod seasonalPeriod) + 1        forecastValue = forecastValue * seasonalIndices(seasonalIndex3)        Dim result As Dictionary        Set result = New Dictionary        result("Period") = data(i)("Period")        result("Actual") = data(i)("Sales")        result("Forecast") = forecastValue        result("Error") = data(i)("Sales") - forecastValue        If data(i)("Sales") > 0 Then            result("ErrorPercent") = Abs(result("Error") / data(i)("Sales"))        Else            result("ErrorPercent") = 0        End If        results.Add result    Next i    Set SeasonalDecomposition = resultsEnd Function' 预测评估Function EvaluateForecast(forecastResults As Collection) As Dictionary    Dim evaluation As New Dictionary    If forecastResults.Count = 0 Then        Set EvaluateForecast = evaluation        Exit Function    End If    Dim mae As Double, mse As Double, mape As Double    Dim errorCount As Long    errorCount = 0    Dim i As Long    For i = 1 To forecastResults.Count        If forecastResults(i).Exists("Error") Then            Dim [error] As Double            [error] = forecastResults(i)("Error")            Dim absError As Double            absError = Abs([error])            mae = mae + absError            mse = mse + [error] ^ 2            If forecastResults(i)("Actual") > 0 Then                mape = mape + (absError / forecastResults(i)("Actual"))            End If            errorCount = errorCount + 1        End If    Next i    If errorCount > 0 Then        mae = mae / errorCount        mse = mse / errorCount        mape = mape / errorCount    End If    evaluation("MAE") = mae    evaluation("MSE") = mse    evaluation("RMSE") = Sqr(mse)    evaluation("MAPE") = mape    evaluation("Count") = errorCount    Set EvaluateForecast = evaluationEnd Function

#!/usr/bin/env python3# -*- coding: utf-8 -*-"""智能销售预测系统支持多种预测算法、特征工程、模型评估"""import osimport sysimport jsonimport loggingimport warningsimport numpy as npimport pandas as pdfrom datetime import datetime, timedeltafrom typing import List, Dict, Any, Optional, Tuple, Unionfrom dataclasses import dataclass, asdict, fieldfrom decimal import Decimalfrom pathlib import Pathimport pickleimport hashlibimport tracebackfrom enum import Enumimport copyimport itertoolsfrom collections import defaultdict# 数据可视化import matplotlib.pyplot as pltimport seaborn as snsfrom matplotlib import rcParams# 统计检验from scipy import statsfrom scipy.stats import boxcox, yeojohnsonfrom scipy.signal import periodogram# 时间序列分析import statsmodels.api as smfrom statsmodels.tsa.stattools import adfuller, kpss, acf, pacffrom statsmodels.tsa.seasonal import seasonal_decompose, STLfrom statsmodels.tsa.holtwinters import ExponentialSmoothing, SimpleExpSmoothingfrom statsmodels.tsa.arima.model import ARIMAfrom statsmodels.tsa.statespace.sarimax import SARIMAXfrom statsmodels.tsa.vector_ar.var_model import VARfrom statsmodels.graphics.tsaplots import plot_acf, plot_pacffrom statsmodels.tsa.forecasting.stl import STLForecast# 机器学习from sklearn.preprocessing import StandardScaler, MinMaxScaler, RobustScalerfrom sklearn.linear_model import LinearRegression, Ridge, Lasso, ElasticNetfrom sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressorfrom sklearn.svm import SVRfrom sklearn.neural_network import MLPRegressorfrom sklearn.model_selection import TimeSeriesSplit, cross_val_score, GridSearchCVfrom sklearn.metrics import mean_absolute_error, mean_squared_error, r2_scorefrom sklearn.pipeline import Pipelinefrom sklearn.compose import ColumnTransformerfrom sklearn.feature_selection import RFE, SelectKBest, f_regression# 深度学习try:    import tensorflow as tf    from tensorflow.keras.models import Sequential, Model    from tensorflow.keras.layers import Dense, LSTM, GRU, Dropout, Input, Conv1D, MaxPooling1D, Flatten    from tensorflow.keras.callbacks import EarlyStopping, ModelCheckpoint, ReduceLROnPlateau    TF_AVAILABLE = Trueexcept ImportError:    TF_AVAILABLE = False    logging.warning("TensorFlow not available, deep learning models disabled")# 预测评估from pmdarima import auto_arimafrom prophet import Prophetimport xgboost as xgbimport lightgbm as lgbimport optuna# 数据库import sqlalchemyfrom sqlalchemy import create_engine, text, MetaData, Table, selectfrom sqlalchemy.orm import sessionmaker, Session# 配置import yamlfrom pydantic import BaseModel, Field, validatorimport redis# 日志配置logging.basicConfig(    level=logging.INFO,    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',    handlers=[        logging.FileHandler('sales_forecast.log'),        logging.StreamHandler()    ])logger = logging.getLogger(__name__)warnings.filterwarnings('ignore')# 设置中文字体plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei']plt.rcParams['axes.unicode_minus'] = False# 数据模型class ForecastMethod(Enum):    """预测方法枚举"""    SIMPLE_MA = "simple_moving_average"    EXPONENTIAL_SMOOTHING = "exponential_smoothing"    LINEAR_REGRESSION = "linear_regression"    ARIMA = "arima"    SARIMA = "sarima"    PROPHET = "prophet"    RANDOM_FOREST = "random_forest"    XGBOOST = "xgboost"    LSTM = "lstm"    ENSEMBLE = "ensemble"@dataclassclass ForecastConfig:    """预测配置"""    # 数据配置    data_source: str    target_column: str = "sales"    date_column: str = "date"    frequency: str = "D"  # 频率：D日，W周，M月，Q季，Y年    # 预测配置    forecast_horizon: int = 30    test_size: float = 0.2    cv_folds: int = 5    # 特征工程配置    lag_features: List[int] = field(default_factory=lambda: [1, 7, 14, 30])    rolling_features: List[int] = field(default_factory=lambda: [3, 7, 14, 30])    seasonal_features: bool = True    external_features: List[str] = field(default_factory=list)    # 模型配置    primary_method: ForecastMethod = ForecastMethod.SARIMA    ensemble_methods: List[ForecastMethod] = field(default_factory=list)    use_automl: bool = False    # 评估配置    metrics: List[str] = field(default_factory=lambda: ["MAE", "RMSE", "MAPE", "R2"])    confidence_level: float = 0.95    class Config:        arbitrary_types_allowed = True@dataclassclass ForecastResult:    """预测结果"""    method: ForecastMethod    forecasts: pd.Series    lower_bounds: pd.Series    upper_bounds: pd.Series    metrics: Dict[str, float]    model: Any    feature_importance: Optional[pd.DataFrame] = None    residuals: Optional[pd.Series] = None    def to_dict(self) -> Dict[str, Any]:        """转换为字典"""        return {            "method": self.method.value,            "forecasts": self.forecasts.tolist(),            "lower_bounds": self.lower_bounds.tolist(),            "upper_bounds": self.upper_bounds.tolist(),            "metrics": self.metrics,            "feature_importance": self.feature_importance.to_dict() if self.feature_importance is not None else None        }class SalesForecastSystem:    """销售预测系统"""    def __init__(self, config_path: str = "config/forecast_config.yaml"):        """初始化"""        self.config = self._load_config(config_path)        self.data_loader = DataLoader(self.config)        self.feature_engineer = FeatureEngineer(self.config)        self.model_factory = ModelFactory(self.config)        self.evaluator = ForecastEvaluator(self.config)        self.visualizer = ForecastVisualizer()        logger.info(f"销售预测系统初始化完成，主方法：{self.config.primary_method.value}")    def _load_config(self, config_path: str) -> ForecastConfig:        """加载配置"""        with open(config_path, 'r', encoding='utf-8') as f:            config_data = yaml.safe_load(f)        return ForecastConfig(**config_data)    def run_forecast_pipeline(self) -> Dict[str, Any]:        """运行完整的预测流程"""        logger.info("开始销售预测流程")        start_time = datetime.now()        try:            # 1. 数据加载            logger.info("步骤1: 加载数据")            data = self.data_loader.load_data()            if data.empty:                raise ValueError("数据加载失败或数据为空")            logger.info(f"数据加载完成，共{len(data)}条记录")            # 2. 数据探索            logger.info("步骤2: 数据探索分析")            exploration_report = self._explore_data(data)            # 3. 特征工程            logger.info("步骤3: 特征工程")            features, target = self.feature_engineer.transform(data)            # 4. 划分训练测试集            logger.info("步骤4: 划分数据集")            X_train, X_test, y_train, y_test = self._split_time_series(features, target)            # 5. 模型训练与预测            logger.info("步骤5: 模型训练")            forecast_results = []            # 主模型预测            primary_result = self._train_and_predict(                self.config.primary_method,                 X_train, X_test, y_train, y_test            )            forecast_results.append(primary_result)            # 集成模型预测            for method in self.config.ensemble_methods:                try:                    result = self._train_and_predict(method, X_train, X_test, y_train, y_test)                    forecast_results.append(result)                except Exception as e:                    logger.warning(f"模型{method.value}训练失败: {str(e)}")            # 6. 模型评估            logger.info("步骤6: 模型评估")            evaluation_report = self.evaluator.evaluate_all(forecast_results, y_test)            # 7. 生成预测报告            logger.info("步骤7: 生成预测报告")            final_forecast = self._generate_final_forecast(forecast_results, evaluation_report)            report = self._generate_forecast_report(                data, features, forecast_results,                 evaluation_report, final_forecast, exploration_report            )            # 8. 可视化            logger.info("步骤8: 生成可视化")            self.visualizer.generate_all_plots(                data, features, forecast_results,                 evaluation_report, final_forecast, report["output_dir"]            )            end_time = datetime.now()            execution_time = (end_time - start_time).total_seconds()            logger.info(f"预测流程完成，总耗时: {execution_time:.2f}秒")            return {                "success": True,                "execution_time": execution_time,                "best_model": evaluation_report["best_model"],                "best_score": evaluation_report["best_score"],                "report_path": report["report_path"],                "forecast": final_forecast.to_dict()            }        except Exception as e:            logger.error(f"预测流程失败: {str(e)}")            logger.error(traceback.format_exc())            return {                "success": False,                "error": str(e),                "traceback": traceback.format_exc()            }    def _explore_data(self, data: pd.DataFrame) -> Dict[str, Any]:        """数据探索分析"""        report = {            "summary": {},            "stationarity": {},            "seasonality": {},            "trend": {},            "correlation": {}        }        # 基本统计        report["summary"] = {            "total_records": len(data),            "date_range": f"{data[self.config.date_column].min()} 到 {data[self.config.date_column].max()}",            "mean": data[self.config.target_column].mean(),            "std": data[self.config.target_column].std(),            "min": data[self.config.target_column].min(),            "max": data[self.config.target_column].max(),            "missing_values": data[self.config.target_column].isnull().sum(),            "zero_values": (data[self.config.target_column] == 0).sum()        }        # 平稳性检验        try:            adf_result = adfuller(data[self.config.target_column].dropna())            kpss_result = kpss(data[self.config.target_column].dropna())            report["stationarity"] = {                "adf_statistic": adf_result[0],                "adf_pvalue": adf_result[1],                "adf_stationary": adf_result[1] < 0.05,                "kpss_statistic": kpss_result[0],                "kpss_pvalue": kpss_result[1],                "kpss_stationary": kpss_result[1] > 0.05            }        except Exception as e:            logger.warning(f"平稳性检验失败: {str(e)}")        # 季节性分解        try:            if len(data) >= 2 * 365:  # 至少两年数据                decomposition = seasonal_decompose(                    data[self.config.target_column],                     model='additive',                     period=365                )                seasonal_strength = 1 - (decomposition.resid.var() / decomposition.observed.var())                trend_strength = 1 - (decomposition.resid.var() / decomposition.trend.var())                report["seasonality"] = {                    "seasonal_strength": seasonal_strength,                    "trend_strength": trend_strength,                    "has_seasonality": seasonal_strength > 0.5                }        except Exception as e:            logger.warning(f"季节性分解失败: {str(e)}")        return report    def _split_time_series(self, features: pd.DataFrame, target: pd.Series) -> tuple:        """划分时间序列数据集"""        test_size = int(len(features) * self.config.test_size)        X_train = features.iloc[:-test_size]        X_test = features.iloc[-test_size:]        y_train = target.iloc[:-test_size]        y_test = target.iloc[-test_size:]        logger.info(f"数据集划分: 训练集{len(X_train)}条, 测试集{len(X_test)}条")        return X_train, X_test, y_train, y_test    def _train_and_predict(        self,         method: ForecastMethod,        X_train: pd.DataFrame,        X_test: pd.DataFrame,        y_train: pd.Series,        y_test: pd.Series    ) -> ForecastResult:        """训练模型并预测"""        logger.info(f"训练模型: {method.value}")        # 获取模型        model = self.model_factory.get_model(method)        # 训练模型        if method in [ForecastMethod.ARIMA, ForecastMethod.SARIMA, ForecastMethod.PROPHET]:            # 时间序列模型            model.fit(y_train)            # 预测            forecast = model.predict(steps=len(X_test))            forecasts = pd.Series(forecast, index=X_test.index)            # 置信区间            if hasattr(model, 'get_prediction'):                pred_results = model.get_prediction(start=len(y_train), end=len(y_train) + len(y_test) - 1)                lower_bounds = pred_results.conf_int(alpha=1-self.config.confidence_level).iloc[:, 0]                upper_bounds = pred_results.conf_int(alpha=1-self.config.confidence_level).iloc[:, 1]            else:                # 简单置信区间                std = forecasts.std()                z_score = stats.norm.ppf((1 + self.config.confidence_level) / 2)                margin = z_score * std                lower_bounds = forecasts - margin                upper_bounds = forecasts + margin        else:            # 机器学习模型            model.fit(X_train, y_train)            # 预测            forecasts = pd.Series(model.predict(X_test), index=X_test.index)            # 特征重要性            feature_importance = None            if hasattr(model, 'feature_importances_'):                importance = model.feature_importances_                feature_importance = pd.DataFrame({                    'feature': X_train.columns,                    'importance': importance                }).sort_values('importance', ascending=False)            # 置信区间（使用交叉验证）            tscv = TimeSeriesSplit(n_splits=self.config.cv_folds)            cv_scores = cross_val_score(model, X_train, y_train, cv=tscv, scoring='neg_mean_squared_error')            cv_std = np.sqrt(-cv_scores).std()            z_score = stats.norm.ppf((1 + self.config.confidence_level) / 2)            margin = z_score * cv_std            lower_bounds = forecasts - margin            upper_bounds = forecasts + margin        # 计算残差        residuals = y_test - forecasts        # 评估指标        metrics = self.evaluator.calculate_metrics(y_test, forecasts)        return ForecastResult(            method=method,            forecasts=forecasts,            lower_bounds=lower_bounds,            upper_bounds=upper_bounds,            metrics=metrics,            model=model,            feature_importance=feature_importance if 'feature_importance' in locals() else None,            residuals=residuals        )    def _generate_final_forecast(        self,         results: List[ForecastResult],         evaluation_report: Dict[str, Any]    ) -> ForecastResult:        """生成最终预测（选择最佳模型或集成）"""        if not results:            raise ValueError("没有可用的预测结果")        # 如果只有一个结果，直接返回        if len(results) == 1:            return results[0]        # 获取最佳模型        best_method = evaluation_report["best_model"]        best_result = next((r for r in results if r.method == best_method), results[0])        # 如果需要集成，计算加权平均        if self.config.ensemble_methods:            weights = {}            total_weight = 0            for result in results:                # 根据MAPE分配权重（误差越小权重越大）                mape = result.metrics.get("MAPE", 1.0)                weight = 1.0 / (mape + 0.0001)  # 避免除零                weights[result.method] = weight                total_weight += weight            # 归一化权重            for method in weights:                weights[method] /= total_weight            # 加权平均预测            weighted_forecasts = pd.Series(0.0, index=results[0].forecasts.index)            weighted_lower = pd.Series(0.0, index=results[0].lower_bounds.index)            weighted_upper = pd.Series(0.0, index=results[0].upper_bounds.index)            for result in results:                weight = weights[result.method]                weighted_forecasts += result.forecasts * weight                weighted_lower += result.lower_bounds * weight                weighted_upper += result.upper_bounds * weight            # 计算集成模型的指标            # 这里需要实际值来计算，但在最终预测中我们没有实际值            # 所以使用最佳模型的指标            ensemble_result = ForecastResult(                method=ForecastMethod.ENSEMBLE,                forecasts=weighted_forecasts,                lower_bounds=weighted_lower,                upper_bounds=weighted_upper,                metrics=best_result.metrics,  # 使用最佳模型的指标                model=None,                feature_importance=None            )            return ensemble_result        return best_result    def _generate_forecast_report(        self,        data: pd.DataFrame,        features: pd.DataFrame,        results: List[ForecastResult],        evaluation_report: Dict[str, Any],        final_forecast: ForecastResult,        exploration_report: Dict[str, Any]    ) -> Dict[str, Any]:        """生成预测报告"""        # 创建输出目录        output_dir = Path(f"output/forecast/{datetime.now().strftime('%Y%m%d_%H%M%S')}")        output_dir.mkdir(parents=True, exist_ok=True)        # 1. 生成文本报告        report_lines = [            "=" * 80,            "销售预测分析报告",            "=" * 80,            f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",            f"数据范围: {exploration_report['summary']['date_range']}",            f"总记录数: {exploration_report['summary']['total_records']}",            "",            "一、数据概览",            "-" * 40,            f"平均值: {exploration_report['summary']['mean']:.2f}",            f"标准差: {exploration_report['summary']['std']:.2f}",            f"最小值: {exploration_report['summary']['min']:.2f}",            f"最大值: {exploration_report['summary']['max']:.2f}",            f"缺失值: {exploration_report['summary']['missing_values']}",            "",            "二、平稳性分析",            "-" * 40,        ]        if exploration_report["stationarity"]:            report_lines.extend([                f"ADF检验p值: {exploration_report['stationarity']['adf_pvalue']:.4f}",                f"平稳性结论: {'平稳'if exploration_report['stationarity']['adf_stationary'] else'不平稳'}",                f"KPSS检验p值: {exploration_report['stationarity']['kpss_pvalue']:.4f}",            ])        report_lines.extend([            "",            "三、模型评估结果",            "-" * 40,            f"最佳模型: {evaluation_report['best_model'].value}",            f"最佳MAPE: {evaluation_report['best_score']:.2%}",            "",            "各模型表现:"        ])        for method, scores in evaluation_report["model_scores"].items():            report_lines.append(f"  {method}:")            for metric, value in scores.items():                if metric == "MAPE":                    report_lines.append(f"    {metric}: {value:.2%}")                else:                    report_lines.append(f"    {metric}: {value:.2f}")        report_lines.extend([            "",            "四、最终预测结果",            "-" * 40,            f"预测方法: {final_forecast.method.value}",            f"预测期数: {self.config.forecast_horizon}",            f"置信水平: {self.config.confidence_level:.0%}",            "",            "预测值:"        ])        # 添加预测值        forecast_dates = pd.date_range(            start=data[self.config.date_column].max() + timedelta(days=1),            periods=self.config.forecast_horizon,            freq=self.config.frequency        )        for i, (date, value) in enumerate(zip(forecast_dates, final_forecast.forecasts)):            lower = final_forecast.lower_bounds.iloc[i] if i < len(final_forecast.lower_bounds) else value            upper = final_forecast.upper_bounds.iloc[i] if i < len(final_forecast.upper_bounds) else value            report_lines.append(                f"  {date.strftime('%Y-%m-%d')}: {value:.2f} "                f"[{lower:.2f}, {upper:.2f}]"            )        report_lines.extend([            "",            "五、业务建议",            "-" * 40        ])        # 添加业务建议        avg_forecast = final_forecast.forecasts.mean()        last_actual = data[self.config.target_column].iloc[-1]        if avg_forecast > last_actual * 1.2:            report_lines.append("1. 预计销售将显著增长，建议增加库存和生产计划")        elif avg_forecast < last_actual * 0.8:            report_lines.append("1. 预计销售将下降，建议控制库存，加大促销力度")        else:            report_lines.append("1. 预计销售保持平稳，建议维持当前运营策略")        if exploration_report.get("seasonality", {}).get("has_seasonality", False):            report_lines.append("2. 数据存在明显季节性，建议制定季节性营销策略")        report_lines.extend([            "",            "=" * 80,            "报告结束",            "=" * 80        ])        # 保存报告        report_path = output_dir / "forecast_report.txt"        with open(report_path, 'w', encoding='utf-8') as f:            f.write('\n'.join(report_lines))        # 2. 保存预测数据        forecast_df = pd.DataFrame({            'date': forecast_dates,            'forecast': final_forecast.forecasts,            'lower_bound': final_forecast.lower_bounds,            'upper_bound': final_forecast.upper_bounds        })        forecast_csv_path = output_dir / "forecast_results.csv"        forecast_df.to_csv(forecast_csv_path, index=False, encoding='utf-8')        # 3. 保存模型        if final_forecast.model is not None:            model_path = output_dir / "forecast_model.pkl"            with open(model_path, 'wb') as f:                pickle.dump(final_forecast.model, f)        logger.info(f"预测报告已生成: {report_path}")        return {            "report_path": str(report_path),            "output_dir": str(output_dir),            "forecast_csv": str(forecast_csv_path)        }class DataLoader:    """数据加载器"""    def __init__(self, config: ForecastConfig):        self.config = config    def load_data(self) -> pd.DataFrame:        """加载数据"""        if self.config.data_source.endswith('.csv'):            return self._load_csv()        elif self.config.data_source.endswith('.xlsx'):            return self._load_excel()        elif self.config.data_source.startswith('sql://'):            return self._load_database()        else:            raise ValueError(f"不支持的数据源: {self.config.data_source}")    def _load_csv(self) -> pd.DataFrame:        """加载CSV文件"""        df = pd.read_csv(self.config.data_source, parse_dates=[self.config.date_column])        # 确保日期列正确        if self.config.date_column in df.columns:            df[self.config.date_column] = pd.to_datetime(df[self.config.date_column])            df = df.set_index(self.config.date_column).sort_index()        return df    def _load_excel(self) -> pd.DataFrame:        """加载Excel文件"""        df = pd.read_excel(self.config.data_source, parse_dates=[self.config.date_column])        if self.config.date_column in df.columns:            df[self.config.date_column] = pd.to_datetime(df[self.config.date_column])            df = df.set_index(self.config.date_column).sort_index()        return df    def _load_database(self) -> pd.DataFrame:        """从数据库加载"""        # 这里简化处理，实际应根据连接字符串解析        engine = create_engine(self.config.data_source.replace('sql://', ''))        query = f"""        SELECT {self.config.date_column}, {self.config.target_column}        FROM sales_data        ORDER BY {self.config.date_column}        """        df = pd.read_sql(query, engine, parse_dates=[self.config.date_column])        df = df.set_index(self.config.date_column)        return dfclass FeatureEngineer:    """特征工程"""    def __init__(self, config: ForecastConfig):        self.config = config    def transform(self, data: pd.DataFrame) -> Tuple[pd.DataFrame, pd.Series]:        """特征转换"""        # 复制数据避免修改原始数据        df = data.copy()        # 确保是时间序列        if not isinstance(df.index, pd.DatetimeIndex):            raise ValueError("数据索引必须是日期时间类型")        # 处理缺失值        df = self._handle_missing_values(df)        # 创建滞后特征        df = self._create_lag_features(df)        # 创建滚动特征        df = self._create_rolling_features(df)        # 创建时间特征        df = self._create_time_features(df)        # 创建季节性特征        if self.config.seasonal_features:            df = self._create_seasonal_features(df)        # 分离特征和目标        features = df.drop(columns=[self.config.target_column])        target = df[self.config.target_column]        # 删除包含NaN的行        valid_idx = features.notna().all(axis=1) & target.notna()        features = features[valid_idx]        target = target[valid_idx]        logger.info(f"特征工程完成，特征数: {len(features.columns)}，样本数: {len(features)}")        return features, target    def _handle_missing_values(self, df: pd.DataFrame) -> pd.DataFrame:        """处理缺失值"""        # 前向填充        df[self.config.target_column] = df[self.config.target_column].ffill()        # 如果还有缺失，用均值填充        if df[self.config.target_column].isnull().any():            mean_value = df[self.config.target_column].mean()            df[self.config.target_column] = df[self.config.target_column].fillna(mean_value)        return df    def _create_lag_features(self, df: pd.DataFrame) -> pd.DataFrame:        """创建滞后特征"""        for lag in self.config.lag_features:            df[f'lag_{lag}'] = df[self.config.target_column].shift(lag)        return df    def _create_rolling_features(self, df: pd.DataFrame) -> pd.DataFrame:        """创建滚动特征"""        for window in self.config.rolling_features:            df[f'rolling_mean_{window}'] = df[self.config.target_column].rolling(window=window).mean()            df[f'rolling_std_{window}'] = df[self.config.target_column].rolling(window=window).std()            df[f'rolling_min_{window}'] = df[self.config.target_column].rolling(window=window).min()            df[f'rolling_max_{window}'] = df[self.config.target_column].rolling(window=window).max()        return df    def _create_time_features(self, df: pd.DataFrame) -> pd.DataFrame:        """创建时间特征"""        df['day_of_week'] = df.index.dayofweek        df['day_of_month'] = df.index.day        df['week_of_year'] = df.index.isocalendar().week        df['month'] = df.index.month        df['quarter'] = df.index.quarter        df['year'] = df.index.year        df['is_weekend'] = df['day_of_week'].isin([5, 6]).astype(int)        return df    def _create_seasonal_features(self, df: pd.DataFrame) -> pd.DataFrame:        """创建季节性特征"""        # 使用正弦/余弦编码季节性        df['sin_day'] = np.sin(2 * np.pi * df.index.dayofyear / 365)        df['cos_day'] = np.cos(2 * np.pi * df.index.dayofyear / 365)        df['sin_week'] = np.sin(2 * np.pi * df.index.dayofweek / 7)        df['cos_week'] = np.cos(2 * np.pi * df.index.dayofweek / 7)        df['sin_month'] = np.sin(2 * np.pi * df.index.month / 12)        df['cos_month'] = np.cos(2 * np.pi * df.index.month / 12)        return dfclass ModelFactory:    """模型工厂"""    def __init__(self, config: ForecastConfig):        self.config = config    def get_model(self, method: ForecastMethod) -> Any:        """获取模型"""        if method == ForecastMethod.SIMPLE_MA:            return SimpleMovingAverageModel(window=7)        elif method == ForecastMethod.EXPONENTIAL_SMOOTHING:            return ExponentialSmoothingModel()        elif method == ForecastMethod.LINEAR_REGRESSION:            return LinearRegressionModel()        elif method == ForecastMethod.ARIMA:            return ARIMAModel()        elif method == ForecastMethod.SARIMA:            return SARIMAModel()        elif method == ForecastMethod.PROPHET:            return ProphetModel()        elif method == ForecastMethod.RANDOM_FOREST:            return RandomForestModel()        elif method == ForecastMethod.XGBOOST:            return XGBoostModel()        elif method == ForecastMethod.LSTM and TF_AVAILABLE:            return LSTMModel()        else:            raise ValueError(f"不支持的模型: {method}")# 模型实现（简化的示例）class SimpleMovingAverageModel:    """简单移动平均模型"""    def __init__(self, window=7):        self.window = window    def fit(self, y):        self.y = y        return self    def predict(self, steps):        # 使用最后window个值的平均作为预测        last_values = self.y.iloc[-self.window:].values        return np.full(steps, last_values.mean())# 其他模型实现类似，为节省篇幅省略# 实际应用中应使用完整的模型实现# 主程序if __name__ == "__main__":    # 使用示例    forecast_system = SalesForecastSystem("config/forecast_config.yaml")    result = forecast_system.run_forecast_pipeline()    if result["success"]:        print("预测成功！")        print(f"最佳模型: {result['best_model']}")        print(f"最佳MAPE: {result['best_score']:.2%}")        print(f"报告路径: {result['report_path']}")    else:        print(f"预测失败: {result['error']}")