自回归基础模型#

class pytorch_forecasting.models.base_model.AutoRegressiveBaseModel(dataset_parameters: Dict[str, Any] = None, log_interval: int | float = -1, log_val_interval: int | float = None, learning_rate: float | List[float] = 0.001, log_gradient_flow: bool = False, loss: 指标 = SMAPE(), logging_metrics: ModuleList = ModuleList(), reduce_on_plateau_patience: int = 1000, reduce_on_plateau_reduction: float = 2.0, reduce_on_plateau_min_lr: float = 1e-05, weight_decay: float = 0.0, optimizer_params: Dict[str, Any] = None, monotone_constraints: Dict[str, int] = {}, output_transformer: Callable = None, optimizer='adam')[来源]#

基础: BaseModel

具有自回归模型附加方法的模型。

特别添加了decode_autoregressive()方法用于进行自回归预测。

假设以下超参数：

Parameters:

target (str) – 目标变量的名称
target_lags (Dict[str, Dict[str, int]]) – 目标名称的字典，每个目标名称映射到一个相应的滞后变量及其滞后期的字典。滞后期对于指示季节性对模型是有用的。如果你知道数据的季节性，添加至少目标变量及其相应的滞后期可以提高性能。默认为没有滞后期，即一个空字典。

用于继承的时间序列预测的基础模型

Parameters:

log_interval (Union[int, float], optional) – 预测将被记录的批次数。如果 < 1.0，则每个批次会记录多个条目。默认值为 -1。
log_val_interval (Union[int, float], optional) – 预测验证结果记录的批次数。默认为 None/log_interval。
learning_rate (float, 可选) – 学习率。默认为 1e-3。
log_gradient_flow (bool) – 如果要记录梯度流，这会耗费时间，应仅用于诊断训练失败。默认为 False。
损失 (指标, 可选) – 要优化的指标，也可以是指标的列表。默认为 SMAPE()。
logging_metrics (nn.ModuleList[MultiHorizonMetric]) – 在训练期间记录的度量列表。默认为[]。
reduce_on_plateau_patience (int) – 在此之后学习率减少10倍的耐心。默认为1000
reduce_on_plateau_reduction (float) – 当遇到平台期时，学习率的减少。默认值为 2.0。
reduce_on_plateau_min_lr (float) – 降至平台学习率调度器的最小学习率。默认值为 1e-5
weight_decay (float) – 权重衰减。默认为 0.0。
optimizer_params (字典[字符串, 任意类型]) – 优化器的附加参数。默认为 {}.
monotone_constraints (Dict[str, int]) – 用于连续解码器变量的单调性约束字典将位置映射到约束（例如，"0"表示第一个位置）(-1表示负约束，+1表示正约束，较大的数字对约束相对于损失增加更多的权重，但通常是不必要的)。这个约束显著减慢训练速度。默认值为 {}。
output_transformer (可调用对象) – 转换器，将网络输出转换为预测空间。默认为 None，相当于 lambda out: out["prediction"]。
optimizer (str) – 优化器，“ranger”、“sgd”、“adam”、“adamw”或torch.optim中的优化器类名或pytorch_optimizer。可以传递一个类或函数，该类或函数将参数作为第一个参数，并可选地带有lr参数（也可以有weight_decay）。默认为“adam”。

方法

`decode_autoregressive`(decode_one, ...[, ...])	以自回归方式进行预测。
`from_dataset`(数据集, **kwargs)	从数据集中创建模型。
`output_to_prediction`(...[, n_samples])	将网络输出转换为重新缩放和归一化的预测。
`plot_prediction`(x, out[, idx, ...])	绘制预测值与实际值的对比图

decode_autoregressive(decode_one: Callable, first_target: List[Tensor] | Tensor, first_hidden_state: Any, target_scale: List[Tensor] | Tensor, n_decoder_steps: int, n_samples: int = 1, **kwargs) → List[Tensor] | Tensor[来源]#

以自回归方式进行预测。

仅支持连续目标。

Parameters:

decode_one (Callable) –
一个需要至少以下参数的函数：
- idx (int)：解码步骤的索引（从 0 到 n_decoder_steps-1）
- lagged_targets (List[torch.Tensor])：归一化目标的列表。列表长度为 idx + 1，最近的条目在末尾，即 previous_target = lagged_targets[-1] 和一般情况下 lagged_targets[-lag]。
- hidden_state (Any)：进行预测所需的当前隐藏状态。键是变量名称。仅包含大于 idx 的滞后。
- 额外参数不是动态的，但可以通过 **kwargs 参数传递
并返回 (未重缩放的) 网络预测输出和下一个自回归步骤的隐藏状态的元组。
first_target (联合[列表[torch.Tensor], torch.Tensor]) – 解码时使用的第一个目标值
first_hidden_state (Any) – 解码时使用的第一个隐藏状态
target_scale (联合[列表[torch.Tensor], torch.Tensor]) – 与 x 中的目标尺度相同
n_decoder_steps (int) – 解码/预测步骤的数量
n_samples (int) – 从分布中抽取的独立样本数量 - 仅与多变量模型相关。默认为1。
**kwargs – 传递给 decode_one 函数的附加参数。

Returns:

重新缩放的预测

Return type:

并集[列表[torch.Tensor], torch.Tensor]

示例

LSTM/GRU 解码器

def decode(self, x, hidden_state):
    # create input vector
    input_vector = x["decoder_cont"].clone()
    input_vector[..., self.target_positions] = torch.roll(
        input_vector[..., self.target_positions],
        shifts=1,
        dims=1,
    )
    # but this time fill in missing target from encoder_cont at the first time step instead of
    # throwing it away
    last_encoder_target = x["encoder_cont"][
        torch.arange(x["encoder_cont"].size(0), device=x["encoder_cont"].device),
        x["encoder_lengths"] - 1,
        self.target_positions.unsqueeze(-1)
    ].T.contiguous()
    input_vector[:, 0, self.target_positions] = last_encoder_target

    if self.training:  # training mode
        decoder_output, _ = self.rnn(
            x,
            hidden_state,
            lengths=x["decoder_lengths"],
            enforce_sorted=False,
        )

        # from hidden state size to outputs
        if isinstance(self.hparams.target, str):  # single target
            output = self.distribution_projector(decoder_output)
        else:
            output = [projector(decoder_output) for projector in self.distribution_projector]

        # predictions are not yet rescaled -> so rescale now
        return self.transform_output(output, target_scale=target_scale)

    else:  # prediction mode
        target_pos = self.target_positions

        def decode_one(idx, lagged_targets, hidden_state):
            x = input_vector[:, [idx]]
            x[:, 0, target_pos] = lagged_targets[-1]  # overwrite at target positions

            # overwrite at lagged targets positions
            for lag, lag_positions in lagged_target_positions.items():
                if idx > lag:  # only overwrite if target has been generated
                    x[:, 0, lag_positions] = lagged_targets[-lag]

            decoder_output, hidden_state = self.rnn(x, hidden_state)
            decoder_output = decoder_output[:, 0]  # take first timestep
            # from hidden state size to outputs
            if isinstance(self.hparams.target, str):  # single target
                output = self.distribution_projector(decoder_output)
            else:
                output = [projector(decoder_output) for projector in self.distribution_projector]
            return output, hidden_state

        # make predictions which are fed into next step
        output = self.decode_autoregressive(
            decode_one,
            first_target=input_vector[:, 0, target_pos],
            first_hidden_state=hidden_state,
            target_scale=x["target_scale"],
            n_decoder_steps=input_vector.size(1),
        )

        # predictions are already rescaled
        return output

classmethod from_dataset(dataset: 时间序列数据集, **kwargs) → LightningModule[来源]#

从数据集中创建模型。

Parameters:

dataset – 时间序列数据集
**kwargs – 额外的参数，例如模型的超参数（见 __init__()）

Returns:

Lightning模块

output_to_prediction(normalized_prediction_parameters: Tensor, target_scale: List[Tensor] | Tensor, n_samples: int = 1, **kwargs) → Tuple[List[Tensor] | Tensor, Tensor][来源]#

将网络输出转换为重新缩放和归一化的预测。

这个函数通常不是直接调用，而是通过 decode_autoregressive() 调用。

Parameters:

normalized_prediction_parameters (torch.Tensor) – 网络预测输出
target_scale (Union[List[torch.Tensor], torch.Tensor]) – 目标缩放，用于重新缩放网络输出
n_samples (int, 可选) – 独立抽取的样本数量。默认为 1。
**kwargs – 传递给 transform_output() 方法的字典的额外参数。

Returns:

重缩预测的元组和: 标准化预测（例如，用于下一个自回归步骤的输入）

Return type:

元组[联合[列表[torch.Tensor], torch.Tensor], torch.Tensor]

plot_prediction(x: Dict[str, Tensor], out: Dict[str, Tensor], idx: int = 0, add_loss_to_title: 指标 | Tensor | bool = False, show_future_observed: bool = True, ax=None, quantiles_kwargs: Dict[str, Any] | None = None, prediction_kwargs: Dict[str, Any] | None = None)[来源]#

绘制预测值与实际值的对比图

Parameters:

x – 网络输入
out – 网络输出
idx – 要绘制的预测索引
add_loss_to_title – 是否将损失添加到标题或计算损失函数。可以是指标，布尔值指示是否使用损失指标，或包含所有样本损失的张量。计算的损失在没有权重的情况下确定。默认值为 False。
show_future_observed – 是否显示未来的实际值。默认为 True。
ax – 用于绘图的matplotlib坐标轴
quantiles_kwargs (字典[字符串, 任意]) – to_quantiles()损失度量的参数。
prediction_kwargs (Dict[str, Any]) – 用于 to_prediction() 的损失指标的参数。

Returns:

matplotlib 图形

property lagged_target_positions: Dict[int, LongTensor]#

滞后目标变量在协变量中的位置。

Returns:: 字典将整数滞后映射到变量位置的张量。
Return type:: 字典[int, torch.LongTensor]

property target_positions: LongTensor#

协变量中目标变量的位置。

Returns:: 位置的张量。
Return type:: torch.LongTensor

自回归基础模型#

此页面