AI时序分析折腾手记

时序分析时间序列预测

作者: 醉月思发布日期: 2021-10-18 分类: 技术实践

AI时序分析相关的坑，多半出在边界条件上。

结果数据一加载，我就傻眼了：

半年前接手了一个监控系统数据的项目。

背景和需求

半年前接手了一个监控系统数据的项目。业务方说：“你看，我们系统每秒产生几千条指标数据，能不能预测一下未来1小时的负载趋势？这样我们可以提前扩容。”

我第一反应是：“这不是很简单吗，用 ARIMA 或 Prophet 不就搞定了吗？”

结果数据一加载，我就傻眼了：

日志文件压缩后有 50GB，解压后超过 500GB
时间戳混乱，有些是 UTC，有些是本地时间
数据缺失严重，有些时间段直接断层
还有多条记录同一秒的重复数据

看起来"简单"的时序分析，实际落地时全是坑。这篇文章记录我从传统方法到 AI 驱动方案的完整实践过程，以及踩过的那些坑。

为什么写这篇文章

时序分析的资料很多，但大多数是：

理论讲得很深，但落地少
只关注算法本身，不讲数据清洗
假设数据完美，不考虑真实场景的限制

我想写一篇"接地气"的，从实际工程角度出发的文章。希望能帮到那些真正要在生产环境落地时序分析的同学。

数据清洗的真实痛点

先说数据清洗。这是时序分析中最耗时但最少被提及的部分。

数据缺失处理

数据缺失是时序分析的常见问题。我们的数据缺失有几种情况：

时间戳连续，但中间某段时间没有数据
时间戳本身不连续，需要先对齐
某些指标完全缺失

我试过几种方法：

# 方法1：线性插值
df['value'] = df['value'].interpolate(method='linear')

# 方法2：前向填充（简单粗暴）
df['value'] = df['value'].fillna(method='ffill')

# 方法3：使用同周期历史数据（考虑季节性）
# 周一14:00的数据缺失，用最近几个周一14:00的平均值

最后发现，对于监控数据，线性插值 + 限制填充范围效果最好：

# 只插值不超过10分钟的缺失
df['value'] = df['value'].interpolate(method='linear', limit=60)  # 假设1分钟粒度

# 超过10分钟的缺失保持NaN，后续再处理

异常值检测

时序数据的异常值很棘手，因为"异常"可能只是业务正常波动。

我试过统计方法（3σ原则、IQR），结果把正常的高负载流量都过滤了。

后来用了更智能的方法：

from statsmodels.tsa.seasonal import seasonal_decompose

# 分解趋势、季节性和残差
decomposition = seasonal_decompose(df['value'], model='additive', period=1440)
residual = decomposition.resid

# 对残差做异常检测
threshold = residual.std() * 3
anomalies = residual.abs() > threshold

这样只检测"不符合季节性规律的异常"，而不是简单的"离群值"。

时间戳对齐

这是最头疼的问题。数据来源多，时间标准不一致。

我们的做法：

# 统一转换为 UTC
df['timestamp'] = pd.to_datetime(df['timestamp'], utc=True)

# 重采样到固定间隔（比如1分钟）
df = df.set_index('timestamp').resample('1T').mean()

# 处理重复时间戳
df = df.groupby('timestamp').agg({
    'value': 'mean'  # 同一时间的多个数据点取平均值
})

从传统方法到 AI 的演进

尝试传统方法：ARIMA 和 Prophet

先试了经典方法，毕竟资料多，社区支持好。

ARIMA 的实践

from statsmodels.tsa.arima.model import ARIMA

# 自动定阶（这个步骤很慢）
from pmdarima import auto_arima
model = auto_arima(train_data, seasonal=True, m=1440)

# 训练和预测
model = ARIMA(train_data, order=model.order)
fitted = model.fit()
forecast = fitted.forecast(steps=60)  # 预测60个点

ARIMA 的问题：

假设数据是平稳的，但我们的数据有明显的季节性
计算成本高，自动定阶在历史数据上跑了好几个小时
对最近的变化反应慢

Prophet 的尝试

from prophet import Prophet

# Prophet 需要特定格式
df_prophet = df.reset_index()
df_prophet.columns = ['ds', 'y']

# 添加季节性
model = Prophet(
    yearly_seasonality=True,
    weekly_seasonality=True,
    daily_seasonality=True,
    changepoint_prior_scale=0.05  # 控制趋势变化的敏感度
)
model.fit(df_prophet)

# 预测
future = model.make_future_dataframe(periods=60, freq='1T')
forecast = model.predict(future)

Prophet 的优点：

处理季节性好
内置了节假日支持（我们的业务确实需要）
对数据缺失容忍度高

Prophet 的问题：

预测区间过大，业务方不接受"可能涨也可能跌"的结果
对突发流量反应不足
内存占用大，500GB 数据根本跑不动

深度学习方案：LSTM

传统方法在数据量大时表现不佳，转向深度学习是必然。

数据预处理

深度学习对数据格式要求严格：

import numpy as np
from sklearn.preprocessing import MinMaxScaler

# 归一化
scaler = MinMaxScaler()
scaled_data = scaler.fit_transform(df[['value']])

# 创建滑动窗口（用过去N分钟预测未来M分钟）
def create_sequences(data, seq_length, pred_length):
    X, y = [], []
    for i in range(len(data) - seq_length - pred_length):
        X.append(data[i:i+seq_length])
        y.append(data[i+seq_length:i+seq_length+pred_length])
    return np.array(X), np.array(y)

seq_length = 1440  # 用过去1天的数据
pred_length = 60   # 预测未来1小时
X, y = create_sequences(scaled_data, seq_length, pred_length)

LSTM 模型

from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout

model = Sequential([
    LSTM(128, return_sequences=True, input_shape=(seq_length, 1)),
    Dropout(0.2),
    LSTM(64, return_sequences=False),
    Dropout(0.2),
    Dense(32, activation='relu'),
    Dense(pred_length)
])

model.compile(optimizer='adam', loss='mse', metrics=['mae'])

训练和预测

# 训练
history = model.fit(
    X_train, y_train,
    epochs=50,
    batch_size=256,
    validation_data=(X_val, y_val),
    callbacks=[
        EarlyStopping(patience=10, restore_best_weights=True),
        ReduceLROnPlateau(factor=0.5, patience=5)
    ]
)

# 预测
y_pred_scaled = model.predict(X_test[-1:])
y_pred = scaler.inverse_transform(y_pred_scaled.reshape(-1, 1))

LSTM 的效果：

对短期预测（1小时内）准确度明显提升
对突发流量有更好的反应
训练完成后预测速度快

LSTM 的问题：

训练慢，单 GPU 跑了一整晚
容易过拟合，需要仔细调参
对超长期预测（超过4小时）效果下降明显

工程落地的关键踩坑

踩坑1：实时性问题

第一个坑是实时性。业务方要求"实时预测"，但我们的模型训练要几小时。

解决方案：训练-预测分离架构

graph TD A[实时数据流] --> B[特征提取] B --> C[模型推理] C --> D[预测结果] E[历史数据] --> F[定期模型训练] F --> G[模型评估] G --> H[模型更新] H --> C

每天凌晨用完整历史数据重训练模型
白天用训练好的模型做实时推理
如果预测误差超过阈值，触发紧急重训练

踩坑2：模型退化

运行几个月后发现，预测准确率慢慢下降。

原因分析：

业务模式变化（比如新功能上线）
数据分布漂移
季节性变化（比如618大促）

解决方案：在线学习 + 模型监控

# 模拟在线学习（简化版）
def online_learning(model, new_data, learning_rate=0.001):
    # 只对最近的数据微调
    model.fit(new_data, epochs=1, learning_rate=learning_rate, verbose=0)
    return model

# 每小时收集实际值和预测值，计算误差
def calculate_accuracy(predictions, actuals):
    mae = np.mean(np.abs(predictions - actuals))
    mape = np.mean(np.abs((actuals - predictions) / actuals)) * 100
    return mae, mape

踩坑3：置信区间问题

业务方问：“你这个预测95是多少，那有没有可能到120？”

LSTM 只给点预测，没有置信区间。

解决方案：集成方法 + Bootstrap

# 训练多个模型
models = []
for i in range(10):
    model = create_lstm_model()
    # 用不同的数据子集训练
    bootstrap_idx = np.random.choice(len(X_train), size=len(X_train), replace=True)
    model.fit(X_train[bootstrap_idx], y_train[bootstrap_idx], epochs=50)
    models.append(model)

# 预测时取多个结果的分位数
predictions = [model.predict(X_test) for model in models]
lower_bound = np.percentile(predictions, 5, axis=0)
upper_bound = np.percentile(predictions, 95, axis=0)

踩坑4：特征工程不足

最初只用了历史值作为输入，预测结果忽好忽坏。

后来加了更多特征：

def extract_features(timestamp):
    return {
        'hour': timestamp.hour,
        'day_of_week': timestamp.dayofweek,
        'is_weekend': timestamp.dayofweek >= 5,
        'is_business_hour': 9 <= timestamp.hour <= 18,
        'quarter': timestamp.quarter,
        'is_month_start': timestamp.is_month_start,
        'is_month_end': timestamp.is_month_end,
    }

效果明显提升，特别是对周期性规律明显的指标。

最终方案架构

经过多次迭代，最终的方案：

graph TB A[原始数据采集] --> B[数据清洗] B --> C[特征工程] C --> D[多模型预测] D --> E[LSTM] D --> F[XGBoost] D --> G[Prophet] E --> H[集成预测] F --> H G --> H H --> I[后处理] I --> J[业务决策] K[监控告警] --> L[模型重训练] J --> L

多模型集成

from xgboost import XGBRegressor

# XGBoost 用于捕捉非线性关系
xgb_model = XGBRegressor(
    n_estimators=1000,
    max_depth=8,
    learning_rate=0.01
)
xgb_model.fit(X_train.reshape(X_train.shape[0], -1), y_train)

# LSTM 模型（代码同上）
lstm_model = create_lstm_model()
lstm_model.fit(X_train, y_train)

# 集成预测
def ensemble_predict(X):
    lstm_pred = lstm_model.predict(X)
    xgb_pred = xgb_model.predict(X.reshape(X.shape[0], -1))
    # 简单平均
    return (lstm_pred + xgb_pred) / 2

后处理

# 确保预测值合理
def post_process(predictions, historical_min, historical_max):
    predictions = np.clip(predictions, historical_min * 0.8, historical_max * 1.2)
    predictions = np.maximum(predictions, 0)  # 负载不能为负
    return predictions

结果和收益

准确度对比

方法	MAE	MAPE	训练时间	推理时间
ARIMA	12.5	15.3%	4h	10ms
Prophet	10.2	12.8%	2h	50ms
LSTM	8.3	10.5%	12h	5ms
集成方案	7.6	9.2%	15h	15ms

业务价值

提前扩容：成功避免了3次因流量突增导致的系统宕机
成本优化：减少了不必要的过度扩容，节省30%服务器成本
响应速度：预测结果平均延迟从分钟级降到毫秒级

总结和建议

通过这次实践，我有几点心得：

数据质量 > 模型复杂度：花在数据清洗上的时间，绝对值得
没有银弹：不同场景适合不同方法，多模型集成往往更稳定
监控很重要：模型效果会退化，要持续监控和更新
业务理解不可少：时序分析不只是算法，更要理解业务含义

如果你也想做时序分析，我的建议是：

先小步快跑：用简单方法快速验证可行性
关注数据质量：80%的时间应该花在数据上
建立监控体系：模型上线后要持续观察效果
留足冗余：预测不是算命，要考虑误差和异常情况

时序分析是一个持续优化的过程，没有终点，只有更好的预测精度。希望你在这条路上也能少踩些坑。

如果这篇文章对你有帮助，欢迎分享给你的团队。有任何问题，欢迎在评论区交流讨论。

版权声明: 本文首发于指尖魔法屋-AI时序分析折腾手记（https://blog.thinkmoon.cn/post/324-ai-time-series-history-future-practice/）转载或引用必须申明原指尖魔法屋来源及源地址！

利用 Cursor ACP 模式搭建飞书助手机器人，无需任何大模型接口

用 lark-cli 收飞书消息，Python 桥接 Cursor ACP（agent acp），项目代码里不集成任何大模型 API；适合已有 Cursor 订阅、想在本机跑 Agent 的场景。

📁 技术实践

📅 2026-07-07

用 GolemBot 把 Codex 接到飞书机器人

这篇文章记录一种少写代码的方案：用 GolemBot 接飞书机器人，用 Codex 做 Agent，再通过飞书 CLI Skills 读取文档、需求清单和 bug 清单。

📁 技术实践

📅 2026-06-09

调试 OpenAI Responses API 的 Web Search 踩坑记录

本文记录一次在投资研究 Agent 中调试 OpenAI Responses API Web Search 工具的过程，重点梳理请求格式、流式 SSE 解析、工具名兼容性和结构化数据源设计。

📁 技术实践

📅 2026-04-30

AI时序分析折腾手记

背景和需求

为什么写这篇文章

数据清洗的真实痛点

数据缺失处理

异常值检测

时间戳对齐

从传统方法到 AI 的演进

尝试传统方法：ARIMA 和 Prophet

ARIMA 的实践

Prophet 的尝试

深度学习方案：LSTM

数据预处理

LSTM 模型

训练和预测

工程落地的关键踩坑

踩坑1：实时性问题

踩坑2：模型退化

踩坑3：置信区间问题

踩坑4：特征工程不足

最终方案架构

多模型集成

后处理

结果和收益

准确度对比

业务价值

总结和建议

评论

相关阅读