TimesFM动态协变量实战指南：如何将预测精度提升20%以上

TimesFM动态协变量实战指南：如何将预测精度提升20%以上

【免费下载链接】timesfmTimesFM (Time Series Foundation Model) is a pretrained time-series foundation model developed by Google Research for time-series forecasting.项目地址: https://gitcode.com/GitHub_Trending/ti/timesfm

TimesFM作为Google Research开发的时间序列基础模型，其2.5版本重新引入了强大的协变量支持功能，为复杂业务场景下的预测问题提供了全新的解决方案。本文将深入探讨TimesFM动态协变量的核心技术原理、实际应用场景和最佳实践，帮助开发者充分利用这一功能提升预测精度。

核心关键词：TimesFM时间序列预测、动态协变量、外生变量、预测精度提升
长尾关键词：时间序列协变量处理、外生变量预测模型、多变量时间序列预测、零售销售预测优化、电力负荷预测协变量

为什么传统时间序列预测在复杂场景下表现不佳？

在实际业务中，时间序列数据往往受到多种外部因素的影响。以零售销售预测为例，销售额不仅受历史趋势影响，还受到价格变动、促销活动、节假日、天气等多种因素的综合作用。传统的时间序列模型如ARIMA、Prophet等在处理这些复杂关系时存在明显局限：

1. 单变量限制：大多数传统模型只能处理单变量时间序列
2. 外生变量整合困难：难以有效整合多种类型的协变量
3. 非线性关系建模不足：对协变量与目标变量之间的复杂关系建模能力有限

TimesFM协变量可视化分析：展示价格、促销、节假日等协变量对零售销售的影响

TimesFM协变量架构解析

TimesFM的协变量支持通过XReg模块实现，该模块位于src/timesfm/utils/xreg_lib.py，提供了四种协变量类型的完整支持：

1. 协变量类型定义

# TimesFM支持的四类协变量 dynamic_numerical_covariates = { "temperature": [31.0, 24.3, 19.4, 26.2, 24.6, 30.0, 31.1, 32.4, 30.9, 26.0] } # 动态数值协变量：随时间变化的连续值 dynamic_categorical_covariates = { "weekday": ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun", "Mon", "Tue"] } # 动态分类协变量：随时间变化的分类值 static_numerical_covariates = { "base_price": [12.0, 10.0, 7.5] } # 静态数值协变量：每个序列的固定数值 static_categorical_covariates = { "store_type": ["premium", "standard", "discount"] } # 静态分类协变量：每个序列的固定分类

2. 核心API设计

TimesFM的协变量预测API位于src/timesfm/timesfm_2p5/timesfm_2p5_base.py的forecast_with_covariates方法：

def forecast_with_covariates( self, inputs: list[Sequence[float]], dynamic_numerical_covariates: dict[str, Sequence[Sequence[float]]] | None = None, dynamic_categorical_covariates: dict[str, Sequence[Sequence[Category]]] | None = None, static_numerical_covariates: dict[str, Sequence[float]] | None = None, static_categorical_covariates: dict[str, Sequence[Category]] | None = None, xreg_mode: XRegMode = "xreg + timesfm", normalize_xreg_target_per_input: bool = True, ridge: float = 0.0, max_rows_per_col: int = 0, force_on_cpu: bool = False, ):

两种协变量处理模式深度对比

TimesFM提供了两种协变量处理策略，选择哪种模式取决于业务场景和数据特性：

"xreg + timesfm"模式（默认）

# 模式1：先回归后预测 model.forecast_with_covariates( xreg_mode="xreg + timesfm", # 其他参数... )

工作流程：

1. 首先使用线性回归模型拟合协变量与目标变量的关系
2. TimesFM预测回归残差
3. 最终预测 = 回归预测 + TimesFM残差预测

适用场景：

• 协变量能解释大部分目标变量变化
• 如：温度对电力负荷的影响、价格对销量的影响
• 当协变量与目标变量有强线性关系时效果最佳

"timesfm + xreg"模式

# 模式2：先预测后回归 model.forecast_with_covariates( xreg_mode="timesfm + xreg", # 其他参数... )

工作流程：

1. TimesFM先进行基线预测
2. 线性回归模型拟合预测残差与协变量的关系
3. 最终预测 = TimesFM基线预测 + 回归调整

适用场景：

• 协变量主要解释预测残差
• 如：促销活动对销售异常波动的解释
• 当TimesFM能捕捉主要趋势，协变量解释剩余变异时

TimesFM与其他主流时间序列模型在多个数据集上的性能对比，显示其在协变量支持下的优势

实战案例：零售销售预测系统构建

场景分析

某零售连锁企业需要预测未来12周的销售额，已知影响因素包括：

• 历史销售数据（目标变量）
• 价格变动（动态数值协变量）
• 促销活动（动态分类协变量）
• 节假日（动态分类协变量）
• 商店类型（静态分类协变量）
• 区域位置（静态分类协变量）

数据准备关键点

import numpy as np import timesfm # 1. 数据长度必须匹配 context_length = 24 # 历史数据周数 horizon_length = 12 # 预测周数 total_length = context_length + horizon_length # 动态协变量必须覆盖完整时间段（历史+未来） dynamic_covariates = { "price": np.random.uniform(7.5, 12.0, (3, total_length)), # 3家商店，36周价格 "promotion": np.random.choice([0, 1], (3, total_length), p=[0.8, 0.2]), "holiday": np.zeros((3, total_length)), } # 设置节假日（第0、11、23、35周） for store_idx in range(3): for holiday_week in [0, 11, 23, 35]: if holiday_week < total_length: dynamic_covariates["holiday"][store_idx, holiday_week] = 1 # 静态协变量每个序列一个值 static_covariates = { "store_type": ["premium", "standard", "discount"], "region": ["urban", "suburban", "rural"] }

模型配置与预测

# 加载TimesFM 2.5模型 model = timesfm.TimesFM_2p5_200M_torch.from_pretrained( "google/timesfm-2.5-200m-pytorch" ) # 配置预测参数（必须设置return_backcast=True） model.compile( timesfm.ForecastConfig( max_context=1024, max_horizon=256, normalize_inputs=True, return_backcast=True, # XReg必需 use_continuous_quantile_head=True, force_flip_invariance=True, infer_is_positive=True, fix_quantile_crossing=True, ) ) # 生成模拟销售数据 historical_sales = [] for i in range(3): base_sales = [1000, 750, 500][i] trend = base_sales * (1 + 0.005 * np.arange(context_length)) seasonality = 80 * np.sin(2 * np.pi * np.arange(context_length) / 52) noise = np.random.normal(0, 40, context_length) sales = trend + seasonality + noise historical_sales.append(sales) # 执行协变量预测 cov_forecast, ols_forecast = model.forecast_with_covariates( inputs=historical_sales, dynamic_numerical_covariates={ "price": dynamic_covariates["price"][:, :context_length] }, dynamic_categorical_covariates={ "promotion": dynamic_covariates["promotion"][:, :context_length].astype(int), "holiday": dynamic_covariates["holiday"][:, :context_length].astype(int) }, static_categorical_covariates=static_covariates, xreg_mode="xreg + timesfm", normalize_xreg_target_per_input=True ) print(f"协变量预测形状: {cov_forecast.shape}") # (3, 12) print(f"OLS预测形状: {ols_forecast.shape}") # (3, 12)

TimesFM在长预测周期（96-336步）上的性能表现，显示其在长序列预测中的优势

性能优化与最佳实践

1. 协变量编码优化

问题：字符串分类协变量会显著降低推理速度解决方案：使用数值编码代替字符串

# 不推荐：字符串编码 dynamic_categorical_covariates = { "weekday": ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"] } # 推荐：数值编码 weekday_mapping = {"Mon": 0, "Tue": 1, "Wed": 2, "Thu": 3, "Fri": 4, "Sat": 5, "Sun": 6} dynamic_categorical_covariates = { "weekday": [0, 1, 2, 3, 4, 5, 6] }

2. 内存与计算优化

# 批量处理优化 batch_size = 32 # 根据GPU内存调整 per_core_batch_size = 8 # 多GPU时调整 # 使用混合精度训练加速 model.compile( timesfm.ForecastConfig( max_context=1024, max_horizon=256, normalize_inputs=True, use_mixed_precision=True # 如果支持 ) )

3. 协变量选择策略

必须包含的协变量：

• 已知未来值的变量（如计划价格、预定促销）
• 对目标变量有明确因果关系的变量
• 在训练和预测期间都存在的变量

应该避免的协变量：

• 未来值未知的实时指标（如系统负载）
• 与目标变量相关性不明确的变量
• 存在数据泄露风险的变量

常见问题与解决方案

Q1: 协变量数据长度不匹配

错误信息：ValueError: Forecast horizon length inferred from the dynamic covariates is longer than the max_horizon解决方案：

# 确保动态协变量长度 = 历史长度 + 预测长度 context_len = len(historical_data[0]) horizon_len = 12 total_len = context_len + horizon_len dynamic_covariates = { "temperature": temperature_data[:total_len] # 必须包含完整时间段 }

Q2: 缺少训练或测试协变量

错误信息：ValueError: If train dynamic covariates are provided, test dynamic covariates must also be provided解决方案：

# 必须成对提供训练和测试协变量 train_covariates = temperature_data[:context_len] test_covariates = temperature_data[context_len:total_len] dynamic_numerical_covariates = { "temperature": [train_covariates + test_covariates] }

Q3: 分类协变量编码错误

解决方案：

# 使用sklearn的LabelEncoder进行编码 from sklearn.preprocessing import LabelEncoder encoder = LabelEncoder() encoded_categories = encoder.fit_transform(category_list) dynamic_categorical_covariates = { "category_feature": [encoded_categories.tolist()] }

高级应用场景

场景1：多变量时间序列预测

# 将次要变量作为协变量处理 main_series = sales_data # 主要预测目标 covariate_series = temperature_data # 作为协变量 # 使用协变量模式进行预测 forecast = model.forecast_with_covariates( inputs=main_series, dynamic_numerical_covariates={"temperature": covariate_series}, xreg_mode="xreg + timesfm" )

场景2：层级时间序列预测

# 地区层级预测 region_sales = { "north": north_sales, "south": south_sales, "east": east_sales, "west": west_sales } # 使用地区作为静态分类协变量 static_categorical_covariates = { "region": list(region_sales.keys()) } # 合并所有地区数据进行预测 all_sales = list(region_sales.values()) forecast = model.forecast_with_covariates( inputs=all_sales, static_categorical_covariates=static_categorical_covariates )

场景3：实时预测系统

class RealTimeForecaster: def __init__(self, model_path="google/timesfm-2.5-200m-pytorch"): self.model = timesfm.TimesFM_2p5_200M_torch.from_pretrained(model_path) self.model.compile(timesfm.ForecastConfig(max_context=1024, max_horizon=256)) self.covariate_cache = {} def update_covariates(self, new_data: dict): """更新协变量缓存""" for key, values in new_data.items(): if key not in self.covariate_cache: self.covariate_cache[key] = [] self.covariate_cache[key].extend(values) def forecast_with_latest(self, horizon: int, use_covariates: bool = True): """使用最新数据预测""" latest_data = self.get_latest_data() if use_covariates and self.covariate_cache: return self.model.forecast_with_covariates( inputs=latest_data, dynamic_numerical_covariates=self.covariate_cache, horizon=horizon ) else: return self.model.forecast(inputs=latest_data, horizon=horizon)

TimesFM在多个基准测试数据集上的综合表现，显示其在不同场景下的稳定性

部署与生产建议

1. 模型版本管理

# 使用Hugging Face模型仓库 model = timesfm.TimesFM_2p5_200M_torch.from_pretrained( "google/timesfm-2.5-200m-pytorch", revision="main" # 指定版本 )

2. 监控与评估

def evaluate_forecast_quality(actual, forecast, covariates=None): """评估预测质量""" metrics = { "MAE": np.mean(np.abs(actual - forecast)), "RMSE": np.sqrt(np.mean((actual - forecast) ** 2)), "MAPE": np.mean(np.abs((actual - forecast) / actual)) * 100 } if covariates is not None: # 计算协变量贡献度 cov_contribution = analyze_covariate_contribution(actual, forecast, covariates) metrics["covariate_contribution"] = cov_contribution return metrics

3. A/B测试框架

class ForecastABTest: def __init__(self): self.baseline_model = load_baseline_model() self.timesfm_model = timesfm.TimesFM_2p5_200M_torch.from_pretrained( "google/timesfm-2.5-200m-pytorch" ) def run_test(self, test_data, covariates=None): """运行A/B测试""" baseline_forecast = self.baseline_model.predict(test_data) if covariates: timesfm_forecast = self.timesfm_model.forecast_with_covariates( inputs=test_data, dynamic_numerical_covariates=covariates ) else: timesfm_forecast = self.timesfm_model.forecast(test_data) # 计算改进百分比 improvement = calculate_improvement(baseline_forecast, timesfm_forecast) return improvement

总结

TimesFM的动态协变量功能为复杂时间序列预测问题提供了强大的解决方案。通过合理使用四种协变量类型、选择合适的处理模式、遵循最佳实践，开发者可以显著提升预测精度。关键要点包括：

1. 数据准备是关键：确保协变量数据完整、长度匹配、编码正确
2. 模式选择需谨慎：根据业务场景选择"xreg + timesfm"或"timesfm + xreg"
3. 性能优化不可忽视：使用数值编码、批量处理、混合精度等技术
4. 监控评估要持续：建立完整的评估体系，持续优化模型表现

通过本文的实战指南，您应该能够充分利用TimesFM的协变量功能，在零售、金融、能源等多个领域构建高性能的时间序列预测系统。记住，成功的预测系统不仅依赖于强大的模型，更需要合理的数据准备、正确的参数配置和持续的优化迭代。

立即开始：克隆TimesFM仓库并尝试协变量示例

git clone https://gitcode.com/GitHub_Trending/ti/timesfm cd timesfm uv pip install -e .[xreg] python timesfm-forecasting/examples/covariates-forecasting/demo_covariates.py

通过实际运行示例代码，您将更深入地理解TimesFM协变量功能的工作原理和应用方法，为您的业务预测系统带来显著的精度提升。

【免费下载链接】timesfmTimesFM (Time Series Foundation Model) is a pretrained time-series foundation model developed by Google Research for time-series forecasting.项目地址: https://gitcode.com/GitHub_Trending/ti/timesfm