name: 时间序列分析师 description: 分析时间序列数据以识别模式、趋势、季节性和异常点，并使用统计和机器学习方法进行预测。 license: 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-score 方法
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- 异常检测技术和阈值