在Darts项目中实现基于历史预测的未来协变量训练数据集

背景介绍

在时间序列预测领域，使用未来协变量（future covariates）是提高模型预测精度的重要手段。然而，当这些未来协变量本身也是预测值（如天气预报）并且会随着时间推移不断更新时，如何正确构建训练数据集就成为一个技术挑战。

问题本质

传统的时间序列预测模型在处理未来协变量时，通常假设这些协变量在预测时是已知的。但在实际应用中，很多未来协变量（如天气预报）本身就是预测结果，并且会随着时间推移不断更新。这就需要在模型训练时，能够使用历史上对未来协变量的预测值，而不是最终的真实值。

技术解决方案

在Darts项目中，我们可以通过自定义数据集类来实现这一需求。具体需要以下步骤：

1. 继承GenericShiftedDataset类

首先需要创建一个自定义的GenericShiftedDataset子类，重写__getitem__方法以实现对未来协变量的特殊处理：

class CustomGSD(GenericShiftedDataset):
    def __getitem__(self, idx):
        # 原有逻辑...
        
        # 新增的未来协变量处理逻辑
        if covariate_series is not None and self.covariate_type != CovariateType.PAST:
            fc_idx = idx % self.max_samples_per_ts
            covariate_series = covariate_series[fc_idx]
            if not covariate_series.has_same_time_as(target_series):
                raise ValueError("未来协变量必须与目标序列具有相同的时间索引")
        
        # 继续原有逻辑...

2. 创建混合协变量数据集

根据模型类型（如TiDE、TFT等），创建对应的混合协变量数据集类：

class CustomMCSD(MixedCovariatesSequentialDataset):
    def __init__(self, target_series, past_covariates=None, future_covariates=None, 
                 input_chunk_length=12, output_chunk_length=1, output_chunk_shift=0,
                 max_samples_per_ts=None, use_static_covariates=True, sample_weight=None):
        
        # 初始化三个数据集实例
        self.ds_past = CustomGSD(...)  # 处理过去协变量
        self.ds_historic_future = CustomGSD(...)  # 处理历史未来协变量
        self.ds_future = CustomGSD(...)  # 处理未来协变量
        
    def __getitem__(self, idx):
        # 从三个数据集中分别获取数据
        past_target, past_covariate, static_covariate, sample_weight, future_target = self.ds_past[idx]
        _, historic_future_covariate, _, _, _ = self.ds_historic_future[idx]
        _, future_covariate, _, _, _ = self.ds_future[idx]
        
        return (past_target, past_covariate, historic_future_covariate, 
                future_covariate, static_covariate, sample_weight, future_target)

3. 数据集使用示例

创建并验证自定义数据集：

# 定义输入输出长度
input_length, output_length = 5, 2

# 创建目标序列和未来协变量
series = linear_timeseries(length=input_length + output_length + 1).astype(np.float32)
fc1 = series + 1  # 第一个未来协变量版本
fc2 = series + 2  # 第二个未来协变量版本

# 创建自定义数据集
dataset = CustomMCSD(
    target_series=[series],
    future_covariates=[[fc1, fc2]],  # 注意这里是列表的列表
    input_chunk_length=input_length,
    output_chunk_length=output_length
)

# 验证数据集
assert len(dataset) == 2
assert (dataset[0][2] == dataset[0][0] + 1).all()  # 验证第一个样本的协变量
assert (dataset[1][2] == dataset[1][0] + 2).all()  # 验证第二个样本的协变量

4. 模型训练

最后，使用自定义数据集训练模型：

from darts.models import TiDEModel

model = TiDEModel(input_length, output_length)
model.fit_from_dataset(dataset)

技术要点

1. 未来协变量处理：关键在于识别每个训练样本对应的历史预测版本，确保使用正确的未来协变量值。

2. 数据集设计：采用分层设计，分别处理过去协变量、历史未来协变量和未来协变量。

3. 时间索引一致性：要求所有未来协变量序列必须与目标序列具有相同的时间索引，这是实现方案的前提条件。

4. 灵活性：该方案可以适配Darts中支持混合协变量的各种模型，如TiDE、TFT等。

应用场景

这种技术方案特别适用于以下场景：

1. 天气预报作为协变量的预测问题
2. 经济指标预测中使用的专家预测数据
3. 任何协变量本身也是预测结果的应用场景

总结

通过自定义Darts中的数据集类，我们实现了在模型训练时使用历史上对未来协变量的预测值，而不是最终的真实值。这种方法更加贴近实际应用场景，能够提高模型在真实环境中的预测性能。实现的关键在于正确识别每个训练样本对应的协变量版本，并确保时间索引的一致性。