TimesNet Layer0 接入界面

1. 在父层中的位置

父层是 [[00-总览]] 的 TFB 调用链。本层回答：

TFB 怎样补齐 config
-> 怎样实例化 TimesNet(config)
-> TransformerAdapter._process 怎样调用 TimesNet.forward

2. I/O 接口定义

TimesNet 的底层入口：

def forward(self, x_enc, x_mark_enc, x_dec, x_mark_dec, mask=None):

全局 toy 参数下：

参数	shape	含义
x_enc	(3,8,4)	encoder 历史窗口，seq_len=8，变量数 C=4
x_mark_enc	(3,8,4)	encoder 时间特征，小时频率 timeF 下通常是 4 个时间特征
x_dec	(3,9,4)	adapter 构造的 decoder 输入，长度 label_len + horizon = 4 + 5；TimesNet forecast 主线不依赖它
x_mark_dec	(3,9,4)	decoder 时间特征；TimesNet forecast 主线不依赖它
forecast() 输出	(3,13,4)	完整长度输出，seq_len + pred_len = 8 + 5
forward() 输出	(3,5,4)	dec_out[:, -pred_len:, :] 截取后的最终预测窗口

3. 顺序图（具体层）

4. 语义分组图（索引层）

5. 逐步精读

5.1 config 自动补齐

位置：ts_benchmark/baselines/deep_forecasting_model_base.py

def multi_forecasting_hyper_param_tune(self, train_data: pd.DataFrame):
    freq = pd.infer_freq(train_data.index)
    if freq == None:
        raise ValueError("Irregular time intervals")
    elif freq[0].lower() not in ["m", "w", "b", "d", "h", "t", "s"]:
        self.config.freq = "s"
    else:
        self.config.freq = freq[0].lower()

    column_num = train_data.shape[1]
    self.config.enc_in = column_num
    self.config.dec_in = column_num
    self.config.c_out = column_num

    if self.model_name == "MICN":
        setattr(self.config, "label_len", self.config.seq_len)
    else:
        setattr(self.config, "label_len", self.config.seq_len // 2)

toy 是多变量 forecasting，column_num=4，所以：

enc_in = dec_in = c_out = 4
label_len = seq_len // 2 = 8 // 2 = 4
freq = 由训练数据 index 推断，例如小时频率得到 "h"

单变量 forecasting 会走 single_forecasting_hyper_param_tune()，此时 label_len = horizon。因此真实调试小数据集如果只有 1 个变量，label_len 会和本文档 toy 不同。

5.2 TimesNet.init 参数注册

位置：ts_benchmark/baselines/time_series_library/models/TimesNet.py

def __init__(self, configs):
    super(TimesNet, self).__init__()
    self.configs = configs
    self.task_name = configs.task_name
    self.seq_len = configs.seq_len
    self.label_len = configs.label_len
    self.pred_len = configs.pred_len
    self.model = nn.ModuleList(
        [TimesBlock(configs) for _ in range(configs.e_layers)]
    )
    self.enc_embedding = DataEmbedding(
        configs.enc_in,
        configs.d_model,
        configs.embed,
        configs.freq,
        configs.dropout,
    )
    self.layer = configs.e_layers
    self.layer_norm = nn.LayerNorm(configs.d_model)
    if (
        self.task_name == "long_term_forecast"
        or self.task_name == "short_term_forecast"
    ):
        self.predict_linear = nn.Linear(self.seq_len, self.pred_len + self.seq_len)
        self.projection = nn.Linear(configs.d_model, configs.c_out, bias=True)
    if self.task_name == "imputation" or self.task_name == "anomaly_detection":
        self.projection = nn.Linear(configs.d_model, configs.c_out, bias=True)
    if self.task_name == "classification":
        self.act = F.gelu
        self.dropout = nn.Dropout(configs.dropout)
        self.projection = nn.Linear(
            configs.d_model * configs.seq_len, configs.num_class
        )

参数来源表：

参数	toy 值	来源	用到的位置
task_name	short_term_forecast	adapter 默认 MODEL_HYPER_PARAMS	决定 forward() 走 forecast 分支
seq_len	8	命令行或默认配置	predict_linear 输入长度、TimesBlock.seq_len
pred_len	5	horizon 映射而来	predict_linear 输出长度、forward() 切片
label_len	4	TFB 根据数据自动补齐	adapter 构造 dec_input，TimesNet forecast 不直接使用
enc_in	4	数据列数覆盖	DataEmbedding(c_in=4)
c_out	4	数据列数覆盖	projection(d_model -> c_out)
d_model	6	命令行或默认配置	hidden 维度、LayerNorm 维度
d_ff	7	命令行或默认配置	TimesBlock 中 Inception 中间通道
top_k	2	命令行或默认配置	FFT_for_Period(x, self.k)
num_kernels	3	命令行或默认配置	Inception 卷积核数量
e_layers	1	命令行或默认配置	ModuleList([TimesBlock] * e_layers)
embed	timeF	adapter 默认	选择 TimeFeatureEmbedding
freq	h	数据 index 推断或默认	决定时间特征维度

5.3 TransformerAdapter._process

位置：ts_benchmark/baselines/time_series_library/adapters_for_transformers.py

def _process(self, input, target, input_mark, target_mark):
    # Level 8：统一训练/预测接口先在 adapter 这里做一次输入翻译。
    # 这里把训练循环里的四元组，整理成底层模型 forward 需要的四个输入。
    # decoder 输入 = 历史尾部 label_len + 未来 horizon 的零占位
    dec_input = torch.zeros_like(target[:, -self.config.horizon :, :]).float()
    dec_input = (
        torch.cat([target[:, : self.config.label_len, :], dec_input], dim=1)
        .float()
        .to(input.device)
    )
    # 进入底层模型的 forward 方法
    output = self.model(input, input_mark, dec_input, target_mark)

    return {"output": output}

在 toy 参数中，target 长度是 label_len + horizon。多变量 forecasting 下 label_len=4，horizon=5：

target: (3,9,4)
target[:, :label_len, :] -> (3,4,4)
zeros_like(target[:, -horizon:, :]) -> (3,5,4)
cat(dim=1) -> dec_input: (3,9,4)

数值追踪只看 batch=0, channel=0：

target[0,:,0] = [101,102,103,104,201,202,203,204,205]

历史段 = [101,102,103,104]
未来占位 = [0,0,0,0,0]
dec_input[0,:,0] = [101,102,103,104,0,0,0,0,0]

5.4 TimesNet.forward

位置：ts_benchmark/baselines/time_series_library/models/TimesNet.py

def forward(self, x_enc, x_mark_enc, x_dec, x_mark_dec, mask=None):
    if (
        self.task_name == "long_term_forecast"
        or self.task_name == "short_term_forecast"
    ):
        dec_out = self.forecast(x_enc, x_mark_enc, x_dec, x_mark_dec)
        return dec_out[:, -self.pred_len :, :]  # [B, L, D]
    if self.task_name == "imputation":
        dec_out = self.imputation(x_enc, x_mark_enc, x_dec, x_mark_dec, mask)
        return dec_out  # [B, L, D]
    if self.task_name == "anomaly_detection":
        dec_out = self.anomaly_detection(x_enc)
        return dec_out  # [B, L, D]
    if self.task_name == "classification":
        dec_out = self.classification(x_enc, x_mark_enc)
        return dec_out  # [B, N]
    return None

在本 benchmark 调试主线中：

TimesNet.forecast 返回 dec_out: (3,13,4)
dec_out[:, -5:, :] -> output: (3,5,4)

x_dec 和 x_mark_dec 是统一 adapter 接口传入，但 TimesNet forecast 不用它们参与主计算。TimesNet 真正用于建模的是 x_enc 和 x_mark_enc。

6. 下钻子组件

子组件	职责	下层文档
TimesNet.forecast	TimesNet 预测主链：归一化、embedding、时间扩展、TimesBlock、projection、反归一化	[[02-Layer1-forecast主链]]