1{
2  "op": "replace",
3  "filename": "Time-Series-Library/models/Custom.py",
4  "start_line": 1,
5  "end_line": 41,
6  "content": "import torch\nimport torch.nn as nn\nimport torch.nn.functional as F\n\n\nclass Model(nn.Module):\n    \"\"\"\n    Spectral-Guided Multi-Scale Decomposition Linear (SGMSDLinear)\n\n    Novel forecasting model combining:\n    1. Multi-scale trend-seasonal decomposition (kernels 3, 7, 13, 25)\n    2. Frequency-domain filtered prediction branch with learnable spectral mask\n    3. Spectral energy-guided adaptive scale weighting — the optimal decomposition\n       scale is selected per-instance based on its frequency profile\n\n    Key insight: Different time series require different decomposition granularity.\n    High-frequency dominated series benefit from small kernels (fine decomposition),\n    while smooth/trending series benefit from large kernels. This model learns\n    to route each input to the most appropriate decomposition scale using its\n    spectral energy distribution as a routing signal.\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 (odd sizes for symmetric padding)\n        self.kernels = [3, 7, 13, 25]\n        self.n_scales = len(self.kernels)\n        n_branches = self.n_scales + 1  # +1 for frequency branch\n\n        # Per-scale seasonal and trend linear predictors (DLinear-style init)\n        self.seasonal_linears = nn.ModuleList()\n        self.trend_linears = nn.ModuleList()\n        for _ in range(self.n_scales):\n            s_lin = nn.Linear(self.seq_len, self.pred_len)\n            t_lin = nn.Linear(self.seq_len, self.pred_len)\n            s_lin.weight = nn.Parameter(\n                (1.0 / self.seq_len) * torch.ones(self.pred_len, self.seq_len))\n            t_lin.weight = nn.Parameter(\n                (1.0 / self.seq_len) * torch.ones(self.pred_len, self.seq_len))\n            self.seasonal_linears.append(s_lin)\n            self.trend_linears.append(t_lin)\n\n        # Frequency-domain branch: learnable spectral filter + linear projection\n        n_freq = self.seq_len // 2 + 1\n        self.freq_filter = nn.Parameter(torch.ones(n_freq) * 3.0)  # sigmoid(3)≈0.95\n        self.freq_linear = nn.Linear(self.seq_len, self.pred_len)\n        self.freq_linear.weight = nn.Parameter(\n            (1.0 / self.seq_len) * torch.ones(self.pred_len, self.seq_len))\n\n        # Spectral energy-guided scale selector network\n        # Maps normalized spectral energy distribution -> branch weights\n        self.scale_selector = nn.Sequential(\n            nn.Linear(n_freq, 32),\n            nn.GELU(),\n            nn.Linear(32, n_branches),\n        )\n        # Initialize to produce roughly uniform weights initially\n        nn.init.zeros_(self.scale_selector[2].weight)\n        nn.init.zeros_(self.scale_selector[2].bias)\n\n    def _decompose(self, x, kernel_size):\n        \"\"\"Trend-seasonal decomposition via moving average with boundary padding.\"\"\"\n        half = (kernel_size - 1) // 2\n        front = x[:, :1, :].repeat(1, half, 1)\n        end = x[:, -1:, :].repeat(1, half, 1)\n        x_padded = torch.cat([front, x, end], dim=1)\n        trend = F.avg_pool1d(\n            x_padded.permute(0, 2, 1), kernel_size, stride=1\n        ).permute(0, 2, 1)\n        seasonal = x - trend\n        return seasonal, trend\n\n    def forecast(self, x_enc, x_mark_enc, x_dec, x_mark_dec):\n        # Instance normalization (Non-stationary Transformer style)\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, T, C = x_enc.shape\n\n        # --- Spectral analysis for adaptive scale selection ---\n        x_freq = torch.fft.rfft(x_enc, dim=1)  # [B, n_freq, C]\n        spectral_energy = (x_freq.abs() ** 2).mean(dim=-1)  # [B, n_freq]\n        spectral_energy = spectral_energy / (spectral_energy.sum(-1, keepdim=True) + 1e-8)\n\n        # Compute per-instance adaptive branch weights\n        scale_weights = F.softmax(\n            self.scale_selector(spectral_energy), dim=-1)  # [B, n_branches]\n\n        branch_outputs = []\n\n        # --- Multi-scale temporal decomposition branches ---\n        for i, k in enumerate(self.kernels):\n            seasonal, trend = self._decompose(x_enc, k)\n            s_pred = self.seasonal_linears[i](\n                seasonal.permute(0, 2, 1)).permute(0, 2, 1)\n            t_pred = self.trend_linears[i](\n                trend.permute(0, 2, 1)).permute(0, 2, 1)\n            branch_outputs.append(s_pred + t_pred)\n\n        # --- Frequency-domain filtered branch ---\n        freq_mask = torch.sigmoid(self.freq_filter).unsqueeze(0).unsqueeze(-1)\n        x_filtered = torch.fft.irfft(x_freq * freq_mask, n=T, dim=1)\n        f_pred = self.freq_linear(x_filtered.permute(0, 2, 1)).permute(0, 2, 1)\n        branch_outputs.append(f_pred)\n\n        # --- Spectral-guided weighted combination ---\n        stacked = torch.stack(branch_outputs, dim=-1)  # [B, pred_len, C, n_branches]\n        w = scale_weights.unsqueeze(1).unsqueeze(2)     # [B, 1, 1, n_branches]\n        output = (stacked * w).sum(dim=-1)               # [B, pred_len, C]\n\n        # De-normalization\n        output = output * stdev[:, 0, :].unsqueeze(1).repeat(1, self.pred_len, 1)\n        output = output + means[:, 0, :].unsqueeze(1).repeat(1, self.pred_len, 1)\n\n        return output\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 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": 130,
6  "content": "import torch\nimport torch.nn as nn\nimport torch.nn.functional as F\n\n\nclass Model(nn.Module):\n    \"\"\"\n    Multi-Scale Slope-Aware Decomposition Linear (MSSADLinear)\n\n    Novel forecasting model combining three key contributions:\n\n    1. Multi-scale trend-seasonal decomposition: Instead of a single fixed\n       moving average kernel (DLinear), decompose at multiple scales (3,7,13,25)\n       to capture patterns at different granularities.\n\n    2. Slope-aware trend extrapolation: For each decomposition scale, blend\n       a learned linear forecast with an explicit local slope continuation.\n       The blending weight (alpha) is learned per-scale. This provides a strong\n       inductive bias: the future trend should continue the recent trajectory.\n       Different scales use different slope windows proportional to their kernel\n       size, ensuring the slope is estimated at the appropriate time scale.\n\n    3. Frequency-domain filtered branch: Apply a learnable spectral mask to\n       selectively pass or suppress specific frequency components, then predict\n       via linear projection. This captures periodic patterns missed by\n       the trend-seasonal decomposition.\n\n    All branches are combined via learned softmax weights.\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.kernels = [3, 7, 13, 25]\n        self.n_scales = len(self.kernels)\n        n_branches = self.n_scales + 1  # +1 for frequency branch\n\n        # Per-scale slope estimation windows (proportional to kernel size)\n        self.slope_windows = [max(3, k // 3) for k in self.kernels]\n\n        # Per-scale seasonal and trend linear predictors (DLinear-style init)\n        self.seasonal_linears = nn.ModuleList()\n        self.trend_linears = nn.ModuleList()\n        for _ in range(self.n_scales):\n            s_lin = nn.Linear(self.seq_len, self.pred_len)\n            t_lin = nn.Linear(self.seq_len, self.pred_len)\n            s_lin.weight = nn.Parameter(\n                (1.0 / self.seq_len) * torch.ones(self.pred_len, self.seq_len))\n            t_lin.weight = nn.Parameter(\n                (1.0 / self.seq_len) * torch.ones(self.pred_len, self.seq_len))\n            self.seasonal_linears.append(s_lin)\n            self.trend_linears.append(t_lin)\n\n        # Learnable slope blending weights (one per scale)\n        # sigmoid(0) = 0.5 -> equal blend initially\n        self.slope_alpha = nn.Parameter(torch.zeros(self.n_scales))\n\n        # Frequency-domain filtered branch\n        n_freq = self.seq_len // 2 + 1\n        self.freq_filter = nn.Parameter(torch.ones(n_freq) * 3.0)\n        self.freq_linear = nn.Linear(self.seq_len, self.pred_len)\n        self.freq_linear.weight = nn.Parameter(\n            (1.0 / self.seq_len) * torch.ones(self.pred_len, self.seq_len))\n\n        # Learned combination weights (initialized uniform via softmax of zeros)\n        self.combine_weights = nn.Parameter(torch.zeros(n_branches))\n\n    def _decompose(self, x, kernel_size):\n        \"\"\"Trend-seasonal decomposition via moving average with boundary padding.\"\"\"\n        half = (kernel_size - 1) // 2\n        front = x[:, :1, :].repeat(1, half, 1)\n        end = x[:, -1:, :].repeat(1, half, 1)\n        x_padded = torch.cat([front, x, end], dim=1)\n        trend = F.avg_pool1d(\n            x_padded.permute(0, 2, 1), kernel_size, stride=1\n        ).permute(0, 2, 1)\n        seasonal = x - trend\n        return seasonal, trend\n\n    def _slope_extrapolation(self, trend, slope_window):\n        \"\"\"\n        Compute local slope at the end of the trend and extrapolate forward.\n        This provides an explicit inductive bias for trend continuation.\n\n        Args:\n            trend: [B, T, C] - trend component\n            slope_window: int - number of recent points for slope estimation\n        Returns:\n            [B, pred_len, C] - slope-based trend extrapolation\n        \"\"\"\n        end_val = trend[:, -1:, :]  # [B, 1, C]\n        recent_mean = trend[:, -slope_window:, :].mean(dim=1, keepdim=True)\n        # Slope: rate of change per time step\n        slope = (end_val - recent_mean) / (slope_window / 2.0)\n        # Extrapolate forward\n        steps = torch.arange(1, self.pred_len + 1, device=trend.device).float()\n        return end_val + slope * steps.view(1, -1, 1)\n\n    def forecast(self, x_enc, x_mark_enc, x_dec, x_mark_dec):\n        # Instance normalization (Non-stationary Transformer style)\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, T, C = x_enc.shape\n        branch_outputs = []\n\n        # --- Multi-scale decomposition with slope-aware trend ---\n        for i, k in enumerate(self.kernels):\n            seasonal, trend = self._decompose(x_enc, k)\n\n            # Seasonal prediction: learned linear mapping\n            s_pred = self.seasonal_linears[i](\n                seasonal.permute(0, 2, 1)).permute(0, 2, 1)\n\n            # Trend prediction: blend of learned linear and slope extrapolation\n            t_linear = self.trend_linears[i](\n                trend.permute(0, 2, 1)).permute(0, 2, 1)\n            t_slope = self._slope_extrapolation(trend, self.slope_windows[i])\n            alpha = torch.sigmoid(self.slope_alpha[i])\n            t_pred = alpha * t_slope + (1.0 - alpha) * t_linear\n\n            branch_outputs.append(s_pred + t_pred)\n\n        # --- Frequency-domain filtered branch ---\n        x_freq = torch.fft.rfft(x_enc, dim=1)\n        freq_mask = torch.sigmoid(self.freq_filter).unsqueeze(0).unsqueeze(-1)\n        x_filtered = torch.fft.irfft(x_freq * freq_mask, n=T, dim=1)\n        f_pred = self.freq_linear(x_filtered.permute(0, 2, 1)).permute(0, 2, 1)\n        branch_outputs.append(f_pred)\n\n        # --- Learned weighted combination ---\n        weights = F.softmax(self.combine_weights, dim=0)\n        output = sum(w * o for w, o in zip(weights, branch_outputs))\n\n        # De-normalization\n        output = output * stdev[:, 0, :].unsqueeze(1).repeat(1, self.pred_len, 1)\n        output = output + means[:, 0, :].unsqueeze(1).repeat(1, self.pred_len, 1)\n\n        return output\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 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": 1
3}