名称: 时间序列分析师 描述: 分析时间序列数据以识别模式、趋势、季节性和异常,使用统计和机器学习方法进行预测。 许可证: MIT
时间序列分析师
此技能提供分析时间数据、识别模式和构建预测模型的指导。
核心能力
- 分解: 趋势、季节性、残差分析
- 统计方法: ARIMA、SARIMA、指数平滑法
- 机器学习方法: Prophet、LSTM、基于Transformer的模型
- 异常检测: 统计和机器学习方法
时间序列基础
数据特性
分析前评估:
| 属性 | 问题 | 影响 |
|---|---|---|
| 平稳性 | 均值/方差是否恒定? | 方法选择 |
| 季节性 | 是否有重复模式? | 模型组件 |
| 趋势 | 是否有长期方向? | 差分需求 |
| 频率 | 采样率是多少? | 聚合选择 |
| 缺失值 | 是否有间隙? | 插补需求 |
平稳性检验
from statsmodels.tsa.stattools import adfuller, kpss
# 增广迪基-富勒检验(原假设:非平稳)
adf_result = adfuller(series)
print(f"ADF统计量: {adf_result[0]:.4f}")
print(f"p值: {adf_result[1]:.4f}")
# p < 0.05 表明平稳性
# KPSS检验(原假设:平稳)
kpss_result = kpss(series, regression='c')
print(f"KPSS统计量: {kpss_result[0]:.4f}")
print(f"p值: {kpss_result[1]:.4f}")
# p > 0.05 表明平稳性
使序列平稳
# 差分处理趋势
diff_1 = series.diff().dropna()
# 季节性差分
seasonal_diff = series.diff(periods=12).dropna() # 月度季节性
# 对数变换处理变化方差
log_series = np.log(series)
# Box-Cox用于最优变换
from scipy.stats import boxcox
transformed, lambda_param = boxcox(series)
时间序列分解
经典分解
原始 = 趋势 + 季节性 + 残差 (加法模型)
原始 = 趋势 × 季节性 × 残差 (乘法模型)
from statsmodels.tsa.seasonal import seasonal_decompose, STL
# 经典分解
decomposition = seasonal_decompose(
series,
model='additive', # 或 'multiplicative'
period=12
)
# STL(更鲁棒)
stl = STL(series, period=12, robust=True)
result = stl.fit()
# 访问组件
trend = result.trend
seasonal = result.seasonal
residual = result.resid
可视化模式
┌────────────────────────────────────────┐
│ 原始序列 │ ← 原始数据
├────────────────────────────────────────┤
│ 趋势组件 │ ← 长期方向
├────────────────────────────────────────┤
│ 季节性组件 │ ← 重复模式
├────────────────────────────────────────┤
│ 残差组件 │ ← 随机噪声
└────────────────────────────────────────┘
统计预测方法
ARIMA模型选择
ARIMA(p, d, q):
- p: 自回归阶数(ACF/PACF)
- d: 差分阶数(平稳性)
- q: 移动平均阶数(ACF)
from statsmodels.tsa.arima.model import ARIMA
from pmdarima import auto_arima
# 自动选择
auto_model = auto_arima(
series,
start_p=0, max_p=5,
start_q=0, max_q=5,
d=None, # 自动检测差分
seasonal=False,
information_criterion='aic',
trace=True
)
print(auto_model.summary())
# 手动ARIMA
model = ARIMA(series, order=(2, 1, 2))
fitted = model.fit()
forecast = fitted.forecast(steps=30)
季节性数据的SARIMA
SARIMA(p, d, q)(P, D, Q, m):
- 小写: 非季节性组件
- 大写: 季节性组件
- m: 季节性周期
from statsmodels.tsa.statespace.sarimax import SARIMAX
model = SARIMAX(
series,
order=(1, 1, 1),
seasonal_order=(1, 1, 1, 12), # 月度季节性
enforce_stationarity=False
)
fitted = model.fit()
# 带置信区间的预测
forecast = fitted.get_forecast(steps=24)
mean = forecast.predicted_mean
ci = forecast.conf_int(alpha=0.05)
指数平滑法
from statsmodels.tsa.holtwinters import ExponentialSmoothing
# Holt-Winters(趋势 + 季节性)
model = ExponentialSmoothing(
series,
trend='add', # 或 'mul', None
seasonal='add', # 或 'mul', None
seasonal_periods=12
)
fitted = model.fit()
forecast = fitted.forecast(24)
基于机器学习的预测
Facebook Prophet
from prophet import Prophet
# 准备数据(必须有'ds'和'y'列)
df = pd.DataFrame({'ds': dates, 'y': values})
# 基础模型
model = Prophet(
yearly_seasonality=True,
weekly_seasonality=True,
daily_seasonality=False
)
# 添加自定义季节性
model.add_seasonality(
name='monthly',
period=30.5,
fourier_order=5
)
# 添加回归变量
model.add_regressor('holiday_flag')
model.add_regressor('promotion')
model.fit(df)
# 预测
future = model.make_future_dataframe(periods=365)
forecast = model.predict(future)
用于序列的LSTM
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
# 准备序列
def create_sequences(data, seq_length):
X, y = [], []
for i in range(len(data) - seq_length):
X.append(data[i:i+seq_length])
y.append(data[i+seq_length])
return np.array(X), np.array(y)
X, y = create_sequences(scaled_data, seq_length=60)
X = X.reshape((X.shape[0], X.shape[1], 1))
# 构建模型
model = Sequential([
LSTM(50, return_sequences=True, input_shape=(60, 1)),
Dropout(0.2),
LSTM(50, return_sequences=False),
Dropout(0.2),
Dense(1)
])
model.compile(optimizer='adam', loss='mse')
model.fit(X_train, y_train, epochs=50, batch_size=32, validation_split=0.1)
异常检测
统计方法
# Z-分数方法
def zscore_anomalies(series, threshold=3):
mean, std = series.mean(), series.std()
z_scores = (series - mean) / std
return abs(z_scores) > threshold
# IQR方法
def iqr_anomalies(series, multiplier=1.5):
Q1, Q3 = series.quantile(0.25), series.quantile(0.75)
IQR = Q3 - Q1
lower = Q1 - multiplier * IQR
upper = Q3 + multiplier * IQR
return (series < lower) | (series > upper)
# 滚动统计方法
def rolling_anomalies(series, window=30, threshold=2):
rolling_mean = series.rolling(window).mean()
rolling_std = series.rolling(window).std()
lower = rolling_mean - threshold * rolling_std
upper = rolling_mean + threshold * rolling_std
return (series < lower) | (series > upper)
孤立森林
from sklearn.ensemble import IsolationForest
# 时间序列的特征工程
features = pd.DataFrame({
'value': series,
'hour': series.index.hour,
'dayofweek': series.index.dayofweek,
'rolling_mean': series.rolling(24).mean(),
'rolling_std': series.rolling(24).std()
}).dropna()
model = IsolationForest(contamination=0.01, random_state=42)
anomalies = model.fit_predict(features)
# -1 = 异常, 1 = 正常
模型评估
指标
| 指标 | 公式 | 使用场景 |
|---|---|---|
| MAE | mean(|实际 - 预测|) | 可解释误差 |
| RMSE | sqrt(mean((实际 - 预测)²)) | 惩罚大误差 |
| MAPE | mean(|实际 - 预测| / 实际) | 百分比误差 |
| SMAPE | 对称MAPE | 处理零值更好 |
时间序列交叉验证
from sklearn.model_selection import TimeSeriesSplit
tscv = TimeSeriesSplit(n_splits=5)
for train_idx, test_idx in tscv.split(series):
train, test = series.iloc[train_idx], series.iloc[test_idx]
# 在训练集上拟合,在测试集上评估
回测模式
├──────────────────────────────────────────────────────▶ 时间
│
│ ┌─────────────────┬─────┐
│ │ 训练 │ 测试│ 折叠1
│ └─────────────────┴─────┘
│ ┌───────────────────────┬─────┐
│ │ 训练 │ 测试│ 折叠2
│ └───────────────────────┴─────┘
│ ┌─────────────────────────────┬─────┐
│ │ 训练 │ 测试│ 折叠3
│ └─────────────────────────────┴─────┘
参考
references/arima-guide.md- ARIMA模型选择和诊断references/prophet-tuning.md- Prophet配置和自定义季节性references/anomaly-patterns.md- 异常检测技术和阈值