1{
2  "op": "replace",
3  "filename": "Time-Series-Library/models/Custom.py",
4  "start_line": 1,
5  "end_line": 43,
6  "content": "import torch\nimport torch.nn as nn\nfrom layers.Autoformer_EncDec import series_decomp\n\n\nclass Model(nn.Module):\n    \"\"\"\n    Multi-Scale Spectral-Gated Decomposition Linear (MS-SpectralDLinear).\n\n    Novel algorithmic contribution: Extends DLinear's single-scale decomposition\n    with multi-scale processing and learnable frequency-domain gating.\n\n    Key ideas:\n    1. Multi-scale decomposition: Uses multiple moving average kernels (3, 7, 13, 25)\n       to capture patterns at different temporal resolutions.\n    2. Spectral gating: For each scale's seasonal component, applies a learnable\n       sigmoid gate in the frequency domain. This allows the model to adaptively\n       suppress or amplify specific frequencies at each scale, acting as learned\n       bandpass filters.\n    3. Adaptive scale combination: Learns softmax weights to combine predictions\n       from different scales.\n\n    The motivation is that different temporal patterns (e.g., daily, weekly cycles)\n    are best captured at different decomposition scales, and the spectral gate at\n    each scale learns to focus on the relevant frequency band for that scale.\n    \"\"\"\n\n    def __init__(self, configs):\n        super(Model, self).__init__()\n        self.task_name = configs.task_name\n        self.seq_len = configs.seq_len\n        self.pred_len = configs.pred_len\n        self.enc_in = configs.enc_in\n        self.c_out = configs.c_out\n\n        # Multi-scale decomposition kernels\n        self.kernel_sizes = [3, 7, 13, 25]\n        self.n_scales = len(self.kernel_sizes)\n        self.decomps = nn.ModuleList(\n            [series_decomp(k) for k in self.kernel_sizes]\n        )\n\n        # Frequency domain: number of frequency bins for rfft\n        n_freq = self.seq_len // 2 + 1\n\n        # Learnable spectral gates per scale for seasonal components\n        # Initialized at 0 -> sigmoid(0) = 0.5 (pass half the signal initially)\n        self.spectral_gates = nn.ParameterList([\n            nn.Parameter(torch.zeros(n_freq)) for _ in range(self.n_scales)\n        ])\n\n        # Per-scale linear projections for seasonal and trend\n        self.seasonal_projs = nn.ModuleList()\n        self.trend_projs = nn.ModuleList()\n        for _ in range(self.n_scales):\n            s_proj = nn.Linear(self.seq_len, self.pred_len)\n            s_proj.weight = nn.Parameter(\n                (1.0 / self.seq_len) * torch.ones(self.pred_len, self.seq_len))\n            self.seasonal_projs.append(s_proj)\n\n            t_proj = nn.Linear(self.seq_len, self.pred_len)\n            t_proj.weight = nn.Parameter(\n                (1.0 / self.seq_len) * torch.ones(self.pred_len, self.seq_len))\n            self.trend_projs.append(t_proj)\n\n        # Learnable scale combination weights (softmax applied during forward)\n        self.scale_logits = nn.Parameter(torch.zeros(self.n_scales))\n\n    def forecast(self, x_enc, x_mark_enc, x_dec, x_mark_dec):\n        # RevIN: Instance normalization\n        means = x_enc.mean(1, keepdim=True).detach()\n        x_enc = x_enc - means\n        stdev = torch.sqrt(\n            torch.var(x_enc, dim=1, keepdim=True, unbiased=False) + 1e-5)\n        x_enc = x_enc / stdev\n\n        B, L, C = x_enc.shape\n\n        # Compute scale combination weights\n        scale_w = torch.softmax(self.scale_logits, dim=0)\n\n        # Multi-scale processing with spectral gating\n        dec_out = torch.zeros(\n            B, C, self.pred_len, device=x_enc.device, dtype=x_enc.dtype)\n\n        for i in range(self.n_scales):\n            # Decompose at this scale\n            seasonal, trend = self.decomps[i](x_enc)  # [B, L, C]\n\n            # Spectral gating on seasonal component\n            s = seasonal.permute(0, 2, 1)  # [B, C, L]\n            s_freq = torch.fft.rfft(s, dim=-1)  # [B, C, n_freq]\n            gate = torch.sigmoid(self.spectral_gates[i])  # [n_freq]\n            s_freq = s_freq * gate.unsqueeze(0).unsqueeze(0)\n            s = torch.fft.irfft(s_freq, n=L, dim=-1)  # [B, C, L]\n\n            # Linear projections (channel-independent)\n            s_out = self.seasonal_projs[i](s)  # [B, C, pred_len]\n            t_out = self.trend_projs[i](\n                trend.permute(0, 2, 1))  # [B, C, pred_len]\n\n            # Weighted accumulation\n            dec_out = dec_out + scale_w[i] * (s_out + t_out)\n\n        dec_out = dec_out.permute(0, 2, 1)  # [B, pred_len, C]\n\n        # De-normalization\n        dec_out = dec_out * stdev[:, 0, :].unsqueeze(1).repeat(\n            1, self.pred_len, 1)\n        dec_out = dec_out + means[:, 0, :].unsqueeze(1).repeat(\n            1, self.pred_len, 1)\n\n        return dec_out\n\n    def forward(self, x_enc, x_mark_enc, x_dec, x_mark_dec, mask=None):\n        if (self.task_name == 'long_term_forecast' or\n                self.task_name == 'short_term_forecast'):\n            dec_out = self.forecast(x_enc, x_mark_enc, x_dec, x_mark_dec)\n            return dec_out[:, -self.pred_len:, :]\n        return None\n"
7}

1{
2  "op": "replace",
3  "filename": "Time-Series-Library/models/Custom.py",
4  "start_line": 1,
5  "end_line": 120,
6  "content": "import torch\nimport torch.nn as nn\nfrom layers.Autoformer_EncDec import series_decomp\n\n\nclass Model(nn.Module):\n    \"\"\"\n    Frequency-Corrected Decomposition Linear (FC-DLinear).\n\n    Novel algorithmic contribution: Augments DLinear with a non-linear\n    frequency-domain residual correction that captures cross-frequency\n    interactions the linear time-domain projections miss.\n\n    Architecture:\n    1. RevIN (instance normalization) for distribution shift handling\n    2. DLinear backbone: series decomposition (moving average) into\n       trend + seasonal, each processed by independent linear projections\n    3. Frequency correction path: FFT of input -> 2-layer MLP with GELU\n       on concatenated real/imaginary parts -> IFFT to time-domain correction\n\n    Key design choices:\n    - The correction MLP output layer is zero-initialized, so the model\n      starts exactly as DLinear+RevIN and gradually learns non-linear\n      frequency adjustments. This provides a safe training curriculum.\n    - The MLP's non-linearity (GELU) enables modeling of frequency coupling\n      effects (e.g., harmonic interactions) that pure linear projections\n      cannot capture.\n    - Channel-independent processing for scalability to high-dimensional\n      datasets (e.g., ECL with 321 variables).\n    \"\"\"\n\n    def __init__(self, configs):\n        super(Model, self).__init__()\n        self.task_name = configs.task_name\n        self.seq_len = configs.seq_len\n        self.pred_len = configs.pred_len\n        self.enc_in = configs.enc_in\n        self.c_out = configs.c_out\n\n        # ---------- DLinear backbone ----------\n        # Series decomposition with moving average\n        kernel_size = getattr(configs, 'moving_avg', 25)\n        self.decomp = series_decomp(kernel_size)\n\n        # Linear projections for seasonal and trend (DLinear-style init)\n        init_w = (1.0 / self.seq_len) * torch.ones(self.pred_len, self.seq_len)\n\n        self.seasonal_proj = nn.Linear(self.seq_len, self.pred_len)\n        self.seasonal_proj.weight = nn.Parameter(init_w.clone())\n\n        self.trend_proj = nn.Linear(self.seq_len, self.pred_len)\n        self.trend_proj.weight = nn.Parameter(init_w.clone())\n\n        # ---------- Frequency correction path ----------\n        n_freq_in = self.seq_len // 2 + 1    # FFT bins for input\n        n_freq_out = self.pred_len // 2 + 1   # FFT bins for output\n        hidden_dim = n_freq_in  # moderate hidden size\n\n        self.freq_correction = nn.Sequential(\n            nn.Linear(2 * n_freq_in, hidden_dim),\n            nn.GELU(),\n            nn.Linear(hidden_dim, 2 * n_freq_out)\n        )\n        # Zero-init output layer: model starts as pure DLinear\n        nn.init.zeros_(self.freq_correction[-1].weight)\n        nn.init.zeros_(self.freq_correction[-1].bias)\n        self.n_freq_out = n_freq_out\n\n    def forecast(self, x_enc, x_mark_enc, x_dec, x_mark_dec):\n        # ---- RevIN: Instance normalization ----\n        means = x_enc.mean(1, keepdim=True).detach()\n        x_enc = x_enc - means\n        stdev = torch.sqrt(\n            torch.var(x_enc, dim=1, keepdim=True, unbiased=False) + 1e-5)\n        x_enc = x_enc / stdev\n\n        B, L, C = x_enc.shape\n\n        # ---- DLinear path ----\n        seasonal, trend = self.decomp(x_enc)  # each [B, L, C]\n        # Channel-independent linear projections (permute to [B, C, L])\n        s_out = self.seasonal_proj(seasonal.permute(0, 2, 1))  # [B, C, pred_len]\n        t_out = self.trend_proj(trend.permute(0, 2, 1))        # [B, C, pred_len]\n        base_pred = s_out + t_out  # [B, C, pred_len]\n\n        # ---- Frequency correction path ----\n        # Transform to frequency domain\n        x_freq = torch.fft.rfft(\n            x_enc.permute(0, 2, 1), dim=-1)  # [B, C, n_freq_in]\n        # Concatenate real and imaginary parts\n        x_ri = torch.cat(\n            [x_freq.real, x_freq.imag], dim=-1)  # [B, C, 2*n_freq_in]\n        # Non-linear frequency processing\n        y_ri = self.freq_correction(x_ri)  # [B, C, 2*n_freq_out]\n        # Reconstruct complex frequency representation\n        y_freq = torch.complex(\n            y_ri[:, :, :self.n_freq_out],\n            y_ri[:, :, self.n_freq_out:])\n        # Transform back to time domain\n        correction = torch.fft.irfft(\n            y_freq, n=self.pred_len, dim=-1)  # [B, C, pred_len]\n\n        # ---- Combine ----\n        dec_out = (base_pred + correction).permute(0, 2, 1)  # [B, pred_len, C]\n\n        # ---- De-normalization ----\n        dec_out = dec_out * stdev[:, 0, :].unsqueeze(1).repeat(\n            1, self.pred_len, 1)\n        dec_out = dec_out + means[:, 0, :].unsqueeze(1).repeat(\n            1, self.pred_len, 1)\n\n        return dec_out\n\n    def forward(self, x_enc, x_mark_enc, x_dec, x_mark_dec, mask=None):\n        if (self.task_name == 'long_term_forecast' or\n                self.task_name == 'short_term_forecast'):\n            dec_out = self.forecast(x_enc, x_mark_enc, x_dec, x_mark_dec)\n            return dec_out[:, -self.pred_len:, :]\n        return None\n"
7}

1{
2  "n": 2
3}