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 SpectralBottleneckBlock(nn.Module):\n    \"\"\"\n    Novel block: Low-rank spectral projection bottleneck combined with\n    depthwise separable temporal convolution feed-forward.\n    \n    The bottleneck compresses n_freq spectral coefficients to k < n_freq\n    via learned linear projections (with GELU nonlinearity), creating a\n    genuine information bottleneck in the frequency domain. Normal data\n    patterns with consistent spectral signatures are well-reconstructed,\n    while anomalous patterns with unusual frequency characteristics\n    suffer higher reconstruction error.\n    \"\"\"\n    def __init__(self, d_model, d_ff, seq_len, bottleneck_ratio=0.5, dropout=0.1):\n        super().__init__()\n        n_freq = seq_len // 2 + 1\n        k = max(int(n_freq * bottleneck_ratio), 4)\n\n        # Low-rank spectral bottleneck: n_freq -> k -> n_freq\n        # Separate paths for real and imaginary parts\n        self.enc_r = nn.Linear(n_freq, k)\n        self.enc_i = nn.Linear(n_freq, k)\n        self.dec_r = nn.Linear(k, n_freq)\n        self.dec_i = nn.Linear(k, n_freq)\n        self.norm1 = nn.LayerNorm(d_model)\n\n        # Depthwise separable temporal convolution feed-forward\n        self.ff = nn.Sequential(\n            nn.Conv1d(d_model, d_ff, 1),\n            nn.GELU(),\n            nn.Conv1d(d_ff, d_ff, 3, padding=1, groups=d_ff),\n            nn.GELU(),\n            nn.Conv1d(d_ff, d_model, 1),\n        )\n        self.norm2 = nn.LayerNorm(d_model)\n        self.drop = nn.Dropout(dropout)\n\n    def forward(self, x):\n        B, T, D = x.shape\n\n        # --- Spectral Bottleneck (no skip - hard bottleneck) ---\n        X = torch.fft.rfft(x, dim=1)  # [B, n_freq, D]\n\n        # Compress: [B, D, n_freq] -> [B, D, k]\n        r = F.gelu(self.enc_r(X.real.transpose(1, 2)))\n        i = F.gelu(self.enc_i(X.imag.transpose(1, 2)))\n\n        # Decompress: [B, D, k] -> [B, D, n_freq]\n        r = self.dec_r(r).transpose(1, 2)  # [B, n_freq, D]\n        i = self.dec_i(i).transpose(1, 2)  # [B, n_freq, D]\n\n        # Reconstruct time domain signal\n        x = torch.fft.irfft(torch.complex(r, i), n=T, dim=1)  # [B, T, D]\n        x = self.norm1(x)\n\n        # --- Temporal Feed-Forward with residual ---\n        res = x\n        h = self.ff(x.transpose(1, 2)).transpose(1, 2)\n        x = self.norm2(res + self.drop(h))\n\n        return x\n\n\nclass Model(nn.Module):\n    \"\"\"\n    Spectral Bottleneck Reconstruction Network (SBRNet) for Anomaly Detection.\n\n    Novel algorithmic contribution: Low-rank spectral projection bottleneck.\n    \n    Key ideas:\n    1. Transform input to frequency domain via FFT\n    2. Compress spectral coefficients through a learned low-rank projection\n       (n_freq -> k -> n_freq where k < n_freq), with GELU nonlinearity\n       in the latent space\n    3. Reconstruct via IFFT, then refine with temporal convolutions\n    \n    This creates a frequency-domain information bottleneck that:\n    - During training: learns the most important spectral features for\n      reconstructing normal time series patterns\n    - During testing: anomalous data with unusual spectral characteristics\n      cannot be well-reconstructed through the bottleneck, yielding higher\n      reconstruction error for better anomaly detection\n    \n    Unlike TimesNet (which uses FFT for period discovery) or GFNet (which\n    uses frequency for token mixing), this approach uses frequency-domain\n    compression as a reconstruction bottleneck.\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.seq_len\n        self.enc_in = configs.enc_in\n        self.c_out = configs.c_out\n\n        d_model = configs.d_model\n        d_ff = configs.d_ff\n        e_layers = configs.e_layers\n        dropout = configs.dropout\n\n        # Token embedding via circular convolution (same as standard)\n        self.embed = nn.Conv1d(\n            self.enc_in, d_model, kernel_size=3, padding=1,\n            padding_mode='circular', bias=False\n        )\n        nn.init.kaiming_normal_(\n            self.embed.weight, mode='fan_in', nonlinearity='leaky_relu'\n        )\n\n        # Learnable positional embedding\n        self.pos = nn.Parameter(torch.zeros(1, self.seq_len, d_model))\n        nn.init.normal_(self.pos, std=0.02)\n        self.embed_drop = nn.Dropout(dropout)\n\n        # Stack of spectral bottleneck blocks\n        self.blocks = nn.ModuleList([\n            SpectralBottleneckBlock(\n                d_model, d_ff, self.seq_len,\n                bottleneck_ratio=0.5, dropout=dropout\n            )\n            for _ in range(e_layers)\n        ])\n\n        # Output projection\n        self.norm = nn.LayerNorm(d_model)\n        self.proj = nn.Linear(d_model, self.c_out)\n\n    def anomaly_detection(self, x_enc):\n        \"\"\"\n        Reconstruct input via spectral bottleneck network.\n        Input: x_enc [batch, seq_len, enc_in]\n        Output: [batch, seq_len, c_out]\n        \"\"\"\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        # Embedding\n        x = self.embed(x_enc.transpose(1, 2)).transpose(1, 2)  # [B, T, d_model]\n        x = self.embed_drop(x + self.pos)\n\n        # Process through spectral bottleneck blocks\n        for blk in self.blocks:\n            x = blk(x)\n\n        # Output projection\n        dec_out = self.proj(self.norm(x))\n\n        # De-normalization\n        dec_out = dec_out * stdev[:, 0, :].unsqueeze(1).repeat(\n            1, self.seq_len, 1)\n        dec_out = dec_out + means[:, 0, :].unsqueeze(1).repeat(\n            1, self.seq_len, 1)\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 == 'anomaly_detection':\n            dec_out = self.anomaly_detection(x_enc)\n            return dec_out\n        return None\n"
7}

1{
2  "op": "replace",
3  "filename": "Time-Series-Library/models/Custom.py",
4  "start_line": 1,
5  "end_line": 168,
6  "content": "import torch\nimport torch.nn as nn\nimport torch.nn.functional as F\n\n\nclass SpectralAttentionBlock(nn.Module):\n    \"\"\"\n    Novel block: Spectral Bottleneck + Adaptive Gating + Temporal Self-Attention.\n    \n    Three-stage processing:\n    1. Spectral Bottleneck: FFT -> low-rank compress -> decompress -> IFFT\n       Creates a frequency-domain information bottleneck (no skip connection).\n       Normal patterns with consistent spectral signatures are preserved;\n       anomalous patterns with unusual frequency content are distorted.\n    2. Adaptive Spectral Gating: Input-dependent soft mask over frequency\n       components, making the bottleneck responsive to the input's spectrum.\n    3. Temporal Self-Attention + Feed-Forward: Captures temporal dependencies\n       in the bottlenecked representation for better reconstruction.\n    \"\"\"\n    def __init__(self, d_model, d_ff, seq_len, n_heads,\n                 bottleneck_ratio=0.5, dropout=0.1):\n        super().__init__()\n        n_freq = seq_len // 2 + 1\n        k = max(int(n_freq * bottleneck_ratio), 4)\n\n        # Low-rank spectral bottleneck: n_freq -> k -> n_freq\n        self.enc_r = nn.Linear(n_freq, k)\n        self.enc_i = nn.Linear(n_freq, k)\n        self.dec_r = nn.Linear(k, n_freq)\n        self.dec_i = nn.Linear(k, n_freq)\n\n        # Adaptive spectral gating: input-dependent frequency mask\n        self.gate = nn.Sequential(\n            nn.Linear(n_freq, k),\n            nn.GELU(),\n            nn.Linear(k, n_freq),\n            nn.Sigmoid()\n        )\n        self.norm_spec = nn.LayerNorm(d_model)\n\n        # Temporal self-attention for dependency modeling\n        self.attn = nn.MultiheadAttention(\n            d_model, n_heads, dropout=dropout, batch_first=True\n        )\n        self.norm_attn = nn.LayerNorm(d_model)\n\n        # Feed-forward network\n        self.ff = nn.Sequential(\n            nn.Linear(d_model, d_ff),\n            nn.GELU(),\n            nn.Linear(d_ff, d_model),\n        )\n        self.norm_ff = nn.LayerNorm(d_model)\n        self.drop = nn.Dropout(dropout)\n\n    def forward(self, x):\n        B, T, D = x.shape\n\n        # === Stage 1+2: Spectral Bottleneck with Adaptive Gating ===\n        X = torch.fft.rfft(x, dim=1)  # [B, n_freq, D]\n\n        # Low-rank compress: [B, D, n_freq] -> [B, D, k]\n        r = F.gelu(self.enc_r(X.real.transpose(1, 2)))\n        i = F.gelu(self.enc_i(X.imag.transpose(1, 2)))\n        # Decompress: [B, D, k] -> [B, D, n_freq]\n        r = self.dec_r(r).transpose(1, 2)\n        i = self.dec_i(i).transpose(1, 2)\n\n        # Input-adaptive frequency gating\n        amp = X.abs().mean(-1)  # [B, n_freq] - spectral signature\n        g = self.gate(amp).unsqueeze(-1)  # [B, n_freq, 1]\n        X_out = torch.complex(r, i) * g\n\n        # Back to time domain (no skip - hard bottleneck)\n        x = torch.fft.irfft(X_out, n=T, dim=1)\n        x = self.norm_spec(x)\n\n        # === Stage 3: Temporal Self-Attention (with skip) ===\n        attn_out, _ = self.attn(x, x, x)\n        x = self.norm_attn(x + self.drop(attn_out))\n\n        # === Feed-Forward (with skip) ===\n        res = x\n        x = self.norm_ff(res + self.drop(self.ff(x)))\n\n        return x\n\n\nclass Model(nn.Module):\n    \"\"\"\n    Spectral Bottleneck Attention Network (SBANet) for Anomaly Detection.\n\n    Novel algorithmic contribution: Combines three mechanisms for\n    reconstruction-based anomaly detection:\n    \n    1. Low-rank Spectral Projection Bottleneck - Compresses n_freq spectral\n       coefficients to k < n_freq via learned projections with GELU\n       nonlinearity. This forces the model to learn the most important\n       frequency-domain features for reconstructing normal patterns.\n       \n    2. Input-Adaptive Spectral Gating - A learned network that takes the\n       input's amplitude spectrum and produces a soft frequency mask.\n       This makes the bottleneck responsive to the input's spectral\n       characteristics, enabling better discrimination between normal\n       and anomalous patterns at test time.\n       \n    3. Temporal Self-Attention - After the spectral bottleneck, self-attention\n       captures temporal dependencies in the compressed representation,\n       improving reconstruction quality for normal patterns while maintaining\n       the bottleneck's discriminative power.\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.seq_len\n        self.enc_in = configs.enc_in\n        self.c_out = configs.c_out\n\n        d_model = configs.d_model\n        d_ff = configs.d_ff\n        e_layers = configs.e_layers\n        n_heads = configs.n_heads\n        dropout = configs.dropout\n\n        # Token embedding via circular convolution\n        self.embed = nn.Conv1d(\n            self.enc_in, d_model, kernel_size=3, padding=1,\n            padding_mode='circular', bias=False\n        )\n        nn.init.kaiming_normal_(\n            self.embed.weight, mode='fan_in', nonlinearity='leaky_relu'\n        )\n\n        # Learnable positional embedding\n        self.pos = nn.Parameter(torch.zeros(1, self.seq_len, d_model))\n        nn.init.normal_(self.pos, std=0.02)\n        self.embed_drop = nn.Dropout(dropout)\n\n        # Stack of spectral bottleneck + attention blocks\n        self.blocks = nn.ModuleList([\n            SpectralAttentionBlock(\n                d_model, d_ff, self.seq_len, n_heads,\n                bottleneck_ratio=0.4, dropout=dropout\n            )\n            for _ in range(e_layers)\n        ])\n\n        # Output projection\n        self.norm = nn.LayerNorm(d_model)\n        self.proj = nn.Linear(d_model, self.c_out)\n\n    def anomaly_detection(self, x_enc):\n        \"\"\"\n        Reconstruct input via spectral bottleneck + attention network.\n        \"\"\"\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        # Embedding\n        x = self.embed(x_enc.transpose(1, 2)).transpose(1, 2)\n        x = self.embed_drop(x + self.pos)\n\n        # Process through spectral bottleneck + attention blocks\n        for blk in self.blocks:\n            x = blk(x)\n\n        # Output projection\n        dec_out = self.proj(self.norm(x))\n\n        # De-normalization\n        dec_out = dec_out * stdev[:, 0, :].unsqueeze(1).repeat(\n            1, self.seq_len, 1)\n        dec_out = dec_out + means[:, 0, :].unsqueeze(1).repeat(\n            1, self.seq_len, 1)\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 == 'anomaly_detection':\n            dec_out = self.anomaly_detection(x_enc)\n            return dec_out\n        return None\n"
7}

1{
2  "n": 2
3}