未记录事实核验 · 状态：维护中
记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。
warning 适用边界与维护说明
本文适用于规则日频合成销售序列上的ARIMA选参、Prophet建模、离线异常检测及预测报告流程学习，不提供真实电商业务效果证据，也不是可直接部署的生产系统。性能评估仅针对ARIMA的一次末段留出，不构成多模型比较或滚动回测；ARIMA函数返回点预测，预测区间需另行获取。异常检测的居中窗口包含未来观测，不能直接用于实时监控，且窗口不足的首尾位置未被有效检测，其False标记不应理解为已确认正常。“异常检测与变点检测”一节实际未实现变点检测；报告标题固定为未来30天，调用其他预测天数时标题不会同步改变。模型选择表仅宜作为初步学习指引，不能替代具体数据上的验证。
Python时间序列分析与预测
abstract 学习概览
- 核心内容：时间序列基础理论、ARIMA模型、Prophet框架、异常检测
- 前置技能：机器学习、Pandas数据清洗
- 应用场景：销售预测、异常监控、趋势分析、业务预测
时间序列基础理论
核心概念与定义
时间序列是一系列在时间上有序排列的数据点，通常表示为 X(t)，其中t代表时间索引。时间序列分析的关键在于理解数据的时间依赖性和预测未来趋势。
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')
# 创建时间序列数据示例
np.random.seed(42)
dates = pd.date_range('2024-01-01', periods=365, freq='D')
# 模拟销售数据：基础趋势 + 季节性 + 噪声
base_trend = 1000 + 2 * np.arange(365)
seasonal = 200 * np.sin(2 * np.pi * np.arange(365) / 365) + 100 * np.sin(2 * np.pi * np.arange(365) / 7)
noise = np.random.normal(0, 80, 365)
sales = base_trend + seasonal + noise
sales_ts = pd.Series(sales, index=dates)
print("时间序列的基本特征:")
print(f"数据点数量: {len(sales_ts)}")
print(f"时间范围: {sales_ts.index[0]} 至 {sales_ts.index[-1]}")
print(f"平均值: {sales_ts.mean():.2f}")
print(f"波动性: {sales_ts.std():.2f}")
平稳性检验
平稳性是时间序列分析的核心概念，指序列的统计特性不随时间变化。平稳序列具有：
- 常数均值
- 常数方差
- 自协方差只依赖于滞后期
from statsmodels.tsa.stattools import adfuller
def check_stationarity(ts, title=""):
"""检验时间序列的平稳性"""
# ADF检验
result = adfuller(ts, autolag='AIC')
print(f'{title} ADF平稳性检验结果:')
print(f'ADF统计量: {result[0]:.6f}')
print(f'p值: {result[1]:.6f}')
print('临界值:')
for key, value in result[4].items():
print(f'\t{key}: {value:.3f}')
if result[1] <= 0.05:
print("结论: 在5%显著性水平下拒绝单位根原假设；这不等同于证明序列满足所有平稳性条件")
else:
print("结论: 在5%显著性水平下未拒绝单位根原假设；这不等同于证明序列不平稳")
return result[1] <= 0.05
# 检验原序列平稳性
is_stationary = check_stationarity(sales_ts, "原序列")
# 如果不平稳，进行差分
if not is_stationary:
diff_ts = sales_ts.diff().dropna()
check_stationarity(diff_ts, "一阶差分序列")
自相关与偏自相关
自相关函数(ACF)衡量序列与自身滞后版本的相关性，偏自相关函数(PACF)衡量剔除中间滞后期影响后的相关性。
from statsmodels.tsa.stattools import acf, pacf
from statsmodels.graphics.tsaplots import plot_acf, plot_pacf
def analyze_correlation(ts, lags=40):
"""分析自相关和偏自相关"""
# 计算自相关和偏自相关系数
acf_values = acf(ts, nlags=lags)
pacf_values = pacf(ts, nlags=lags)
# 显著性检验
significant_acf = []
significant_pacf = []
# 95%置信区间
confidence_level = 1.96 / np.sqrt(len(ts))
for i in range(1, lags+1):
if abs(acf_values[i]) > confidence_level:
significant_acf.append(i)
if abs(pacf_values[i]) > confidence_level:
significant_pacf.append(i)
print(f"显著自相关滞后期: {significant_acf[:10]}") # 显示前10个
print(f"显著偏自相关滞后期: {significant_pacf[:10]}")
return acf_values, pacf_values
# 对平稳化后的序列进行相关性分析
if 'diff_ts' in locals():
acf_vals, pacf_vals = analyze_correlation(diff_ts)
ARIMA模型详解
ARIMA(p,d,q)模型是时间序列分析的经典方法：
- AR(p): 自回归项，使用p个历史观测值
- I(d): 差分次数，用于平稳化
- MA(q): 移动平均项，使用q个历史误差项
模型参数选择
from statsmodels.tsa.arima.model import ARIMA
from itertools import product
def find_best_arima(ts, p_range=(0, 3), d_range=(0, 2), q_range=(0, 3)):
"""使用网格搜索找到最佳ARIMA参数"""
best_aic = float('inf')
best_params = None
best_model = None
results = []
for p, d, q in product(range(p_range[0], p_range[1]+1),
range(d_range[0], d_range[1]+1),
range(q_range[0], q_range[1]+1)):
try:
model = ARIMA(ts, order=(p, d, q))
fitted_model = model.fit()
aic = fitted_model.aic
results.append({'params': (p, d, q), 'aic': aic})
if aic < best_aic:
best_aic = aic
best_params = (p, d, q)
best_model = fitted_model
except Exception as e:
continue
print(f"最佳ARIMA参数: {best_params}")
print(f"最佳AIC值: {best_aic:.2f}")
return best_model, best_params, results
# 寻找最佳ARIMA参数
best_arima, best_params, arima_results = find_best_arima(sales_ts)
# 模型诊断
print("\n模型诊断:")
print(best_arima.summary())
# 残差分析
residuals = best_arima.resid
print(f"\n残差统计:")
print(f"残差均值: {residuals.mean():.4f}")
print(f"残差标准差: {residuals.std():.4f}")
# 检验残差是否为白噪声
from statsmodels.stats.diagnostic import acorr_ljungbox
ljung_box = acorr_ljungbox(residuals, lags=10, return_df=True)
print(f"Ljung-Box检验p值: {ljung_box['lb_pvalue'].iloc[0]:.4f}")
ARIMA预测实现
def arima_forecast(model, steps=30):
"""预测具有规则日频DatetimeIndex的序列，保留模型生成的预测索引"""
forecast_result = model.get_forecast(steps=steps)
forecast_series = forecast_result.predicted_mean
# 本函数返回点预测；如需95%预测区间，使用forecast_result.conf_int(alpha=0.05)
return forecast_series
# 进行30天预测
forecast = arima_forecast(best_arima, steps=30)
print("未来30天销售预测:")
for i, (date, value) in enumerate(forecast.head(10).items()):
print(f"{date.strftime('%Y-%m-%d')}: {value:.2f}")
print(f"\n预测总结:")
print(f"未来30天预期总销售: {forecast.sum():,.0f}")
print(f"日均预期销售: {forecast.mean():.0f}")
print(f"与历史均值对比: {(forecast.mean() / sales_ts.mean() - 1) * 100:+.1f}%")
Prophet框架应用
Prophet是Facebook开发的时间序列预测工具，特别适用于具有明显季节性模式的业务数据。
# 需要安装: pip install prophet
try:
from prophet import Prophet
# 准备Prophet格式数据
prophet_data = pd.DataFrame({
'ds': sales_ts.index,
'y': sales_ts.values
})
# 创建Prophet模型
prophet_model = Prophet(
yearly_seasonality=True,
weekly_seasonality=True,
daily_seasonality=False,
changepoint_prior_scale=0.05, # 趋势变化点敏感度
seasonality_prior_scale=10.0 # 季节性强度
)
# 训练模型
prophet_model.fit(prophet_data)
# 创建未来时间框架
future = prophet_model.make_future_dataframe(periods=30)
# 进行预测
prophet_forecast = prophet_model.predict(future)
print("Prophet模型预测结果:")
future_prophet = prophet_forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].tail(10)
for _, row in future_prophet.iterrows():
print(f"{row['ds'].strftime('%Y-%m-%d')}: {row['yhat']:.2f} "
f"[{row['yhat_lower']:.2f}, {row['yhat_upper']:.2f}]")
# 分析成分
print(f"\n趋势成分变化: {prophet_forecast['trend'].iloc[-1] - prophet_forecast['trend'].iloc[0]:.2f}")
print(f"季节性影响范围: {prophet_forecast['yearly'].max() - prophet_forecast['yearly'].min():.2f}")
except ImportError:
print("Prophet库未安装，跳过Prophet示例")
异常检测与变点检测
时间序列中的异常值检测对于数据质量和预测准确性至关重要。
def detect_anomalies(ts, method='iqr', window=30, threshold=3):
"""检测时间序列中的异常值"""
anomalies = pd.DataFrame(index=ts.index)
anomalies['value'] = ts
anomalies['is_anomaly'] = False
if method == 'iqr':
# 基于滚动IQR的离线异常检测；居中窗口包含未来观测
rolling_q1 = ts.rolling(window=window, center=True).quantile(0.25)
rolling_q3 = ts.rolling(window=window, center=True).quantile(0.75)
rolling_iqr = rolling_q3 - rolling_q1
upper_bound = rolling_q3 + threshold * rolling_iqr
lower_bound = rolling_q1 - threshold * rolling_iqr
anomalies['is_anomaly'] = (ts > upper_bound) | (ts < lower_bound)
elif method == 'zscore':
# 基于Z-score的异常检测
rolling_mean = ts.rolling(window=window, center=True).mean()
rolling_std = ts.rolling(window=window, center=True).std()
z_scores = np.abs((ts - rolling_mean) / rolling_std)
anomalies['is_anomaly'] = z_scores > threshold
# 统计异常值
anomaly_count = anomalies['is_anomaly'].sum()
anomaly_rate = anomaly_count / len(ts) * 100
print(f"异常检测结果 ({method}方法):")
print(f"检测到异常点: {anomaly_count}个")
print(f"异常率: {anomaly_rate:.2f}%")
if anomaly_count > 0:
print("异常点详情:")
anomaly_points = anomalies[anomalies['is_anomaly']]
for i, (date, row) in enumerate(anomaly_points.head(10).iterrows()):
print(f" {date.strftime('%Y-%m-%d')}: {row['value']:.2f}")
if anomaly_count > 10:
print(f" ... 还有{anomaly_count-10}个异常点")
return anomalies
# 检测异常值
anomalies_iqr = detect_anomalies(sales_ts, method='iqr')
anomalies_zscore = detect_anomalies(sales_ts, method='zscore')
完整的销售预测案例
基于前文生成的模拟日频销售数据，演示假设电商销售场景下的完整预测分析流程；本例不使用真实电商平台数据，不能据此推断实际业务预测效果：
class SalesPredictor:
"""销售预测系统"""
def __init__(self):
self.models = {}
self.data = None
self.forecasts = {}
def load_data(self, sales_data):
"""加载销售数据"""
self.data = sales_data.copy()
print(f"数据加载完成: {len(self.data)}个数据点")
def prepare_data(self):
"""数据预处理"""
# 检查缺失值
missing_count = self.data.isnull().sum()
if missing_count > 0:
print(f"发现{missing_count}个缺失值，使用前向填充")
self.data = self.data.fillna(method='ffill')
# 检查异常值
anomalies = detect_anomalies(self.data, method='iqr')
anomaly_count = anomalies['is_anomaly'].sum()
if anomaly_count > 0:
print(f"发现{anomaly_count}个异常值")
# 可选择是否处理异常值
def train_models(self):
"""训练多个预测模型"""
print("训练ARIMA模型...")
# ARIMA模型
best_arima, params, _ = find_best_arima(self.data)
self.models['arima'] = best_arima
try:
from prophet import Prophet
print("训练Prophet模型...")
# Prophet模型
prophet_data = pd.DataFrame({
'ds': self.data.index,
'y': self.data.values
})
prophet_model = Prophet(
yearly_seasonality=True,
weekly_seasonality=True,
daily_seasonality=False
)
prophet_model.fit(prophet_data)
self.models['prophet'] = prophet_model
except ImportError:
print("Prophet库未安装，跳过Prophet模型")
def predict(self, days=30):
"""进行预测"""
predictions = {}
# ARIMA预测
if 'arima' in self.models:
arima_pred = arima_forecast(self.models['arima'], days)
predictions['arima'] = arima_pred
# Prophet预测
if 'prophet' in self.models:
future = self.models['prophet'].make_future_dataframe(periods=days)
prophet_pred = self.models['prophet'].predict(future)
future_dates = pd.date_range(
start=self.data.index[-1] + pd.Timedelta(days=1),
periods=days, freq='D'
)
predictions['prophet'] = pd.Series(
prophet_pred['yhat'].tail(days).values,
index=future_dates
)
self.forecasts = predictions
return predictions
def evaluate_performance(self, test_days=30):
"""评估模型性能"""
# 分割数据
train_data = self.data[:-test_days]
test_data = self.data[-test_days:]
# 临时训练模型
temp_arima, _, _ = find_best_arima(train_data)
arima_pred = arima_forecast(temp_arima, test_days)
# 计算误差指标
mae = np.mean(np.abs(test_data.values - arima_pred.values))
mape = np.mean(np.abs((test_data.values - arima_pred.values) / test_data.values)) * 100
rmse = np.sqrt(np.mean((test_data.values - arima_pred.values) ** 2))
print(f"模型性能评估 (最近{test_days}天):")
print(f"平均绝对误差 (MAE): {mae:.2f}")
print(f"平均绝对百分比误差 (MAPE): {mape:.2f}%")
print(f"均方根误差 (RMSE): {rmse:.2f}")
return {'mae': mae, 'mape': mape, 'rmse': rmse}
def generate_report(self):
"""生成预测报告"""
if not self.forecasts:
print("请先进行预测")
return
print("\n" + "="*50)
print("销售预测报告")
print("="*50)
# 历史数据统计
print(f"历史数据统计:")
print(f" 数据时间范围: {self.data.index[0]} 至 {self.data.index[-1]}")
print(f" 历史日均销售: {self.data.mean():.2f}")
print(f" 最近30天均值: {self.data.tail(30).mean():.2f}")
# 预测结果
print(f"\n未来30天预测:")
for model_name, forecast in self.forecasts.items():
print(f" {model_name.upper()}模型:")
print(f" 日均预测: {forecast.mean():.2f}")
print(f" 总量预测: {forecast.sum():,.0f}")
print(f" 与历史对比: {(forecast.mean()/self.data.mean()-1)*100:+.1f}%")
# 演示完整预测流程
print("\n" + "="*60)
print("电商销售预测系统演示")
print("="*60)
# 创建预测器实例
predictor = SalesPredictor()
# 加载数据
predictor.load_data(sales_ts)
# 数据预处理
predictor.prepare_data()
# 训练模型
predictor.train_models()
# 性能评估
performance = predictor.evaluate_performance(test_days=30)
# 进行预测
predictions = predictor.predict(days=30)
# 生成报告
predictor.generate_report()
实用技术要点
模型选择指南
| 数据特征 | 推荐模型 | 适用场景 |
|---|---|---|
| 线性趋势 | ARIMA | 传统业务数据 |
| 强季节性 | Prophet | 电商、零售 |
| 复杂模式 | 机器学习 | 多因子影响 |
| 短期预测 | 简单模型 | 库存管理 |
预测准确性提升技巧
- 数据质量优先：处理缺失值和异常值
- 模型组合：ensemble方法提高稳定性
- 定期更新：使用最新数据重训练
- 业务理解：结合领域知识调整参数
- 多维验证：使用多种评估指标
知识连接
前置学习：
- Pandas数据清洗 - 时间序列数据预处理
- 机器学习 - 预测模型理论基础
后续进阶：
实际应用：
- 销售预测与需求规划
- 金融风险管理
- 用户增长预测
- 运营指标监控
创建时间：2024-06-29
最后更新：2025-09-02
预计学习时间：90分钟
#时间序列 #ARIMA #Prophet #预测分析 #异常检测 #Python