1{
2  "start_line": 5,
3  "content": "import torch\nimport torch.nn as nn\nfrom layers.Autoformer_EncDec import series_decomp\n\nclass Model(nn.Module):\n    \"\"\"\n    Fourier-Channel-Mixer Reversible Linear Model (FCM-RLinear)\n    Combines independent time-domain linear forecasting with \n    low-frequency cross-channel mixing in the frequency domain.\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.channels = getattr(configs, 'enc_in', 7)\n        \n        # Decompose input series into trend and seasonal components\n        moving_avg = getattr(configs, 'moving_avg', 25)\n        self.decomp = series_decomp(moving_avg)\n        \n        # Time-domain forecasting layers (per-channel)\n        self.linear_seasonal = nn.Linear(self.seq_len, self.pred_len)\n        self.linear_trend = nn.Linear(self.seq_len, self.pred_len)\n        \n        # Hyperparameters for frequency mixing\n        # We only interact channels in the top-k lowest frequencies (dominantly periodic structures)\n        # This acts as a principled low-pass noise filter and regularization\n        self.k = getattr(configs, 'fcm_k', min(self.seq_len // 2 + 1, 12)) \n        \n        # Complex matrices for cross-variable modeling per frequency mode\n        # Initialized with small magnitude for robust residual learning from DLinear\n        self.freq_mix_seasonal = nn.Parameter(\n            torch.randn(self.k, self.channels, self.channels, dtype=torch.cfloat) * 0.01\n        )\n        self.freq_mix_trend = nn.Parameter(\n            torch.randn(self.k, self.channels, self.channels, dtype=torch.cfloat) * 0.01\n        )\n        \n    def forecast(self, x_enc, x_mark_enc, x_dec, x_mark_dec):\n        # x_enc: [B, L, C]\n\n        # Reversible Instance Normalization to address distribution shift\n        means = x_enc.mean(1, keepdim=True).detach()\n        x = x_enc - means\n        stdev = torch.sqrt(torch.var(x, dim=1, keepdim=True, unbiased=False) + 1e-5)\n        x = x / stdev\n        \n        # Temporal Decomposition\n        seasonal, trend = self.decomp(x)\n        \n        # --- Frequency-Domain Cross-Channel Mixing (Seasonal) ---\n        # 1. Transform to frequency domain\n        s_fft = torch.fft.rfft(seasonal, dim=1)\n        # 2. Extract top-k dominant low frequencies\n        s_fft_top = s_fft[:, :self.k, :] \n        # 3. Dense cross-channel mixing purely along frequency bands\n        s_mix = torch.einsum('bkc,kcd->bkd', s_fft_top, self.freq_mix_seasonal)\n        # 4. Residual addition\n        s_fft_new = s_fft.clone()\n        s_fft_new[:, :self.k, :] = s_fft_new[:, :self.k, :] + s_mix\n        # 5. Inverse transform to time domain\n        seasonal_mixed = torch.fft.irfft(s_fft_new, n=self.seq_len, dim=1)\n        \n        # --- Frequency-Domain Cross-Channel Mixing (Trend) ---\n        t_fft = torch.fft.rfft(trend, dim=1)\n        t_fft_top = t_fft[:, :self.k, :] \n        t_mix = torch.einsum('bkc,kcd->bkd', t_fft_top, self.freq_mix_trend)\n        t_fft_new = t_fft.clone()\n        t_fft_new[:, :self.k, :] = t_fft_new[:, :self.k, :] + t_mix\n        trend_mixed = torch.fft.irfft(t_fft_new, n=self.seq_len, dim=1)\n        \n        # --- Time-Domain Independent Forecasting ---\n        seasonal_out = self.linear_seasonal(seasonal_mixed.transpose(1, 2)).transpose(1, 2)\n        trend_out = self.linear_trend(trend_mixed.transpose(1, 2)).transpose(1, 2)\n        \n        dec_out = seasonal_out + trend_out\n        \n        # Re-apply De-Normalization\n        dec_out = dec_out * stdev[:, 0, :].unsqueeze(1) + means[:, 0, :].unsqueeze(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 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",
4  "op": "replace",
5  "filename": "Time-Series-Library/models/Custom.py",
6  "end_line": 43
7}

1{
2  "content": "import torch\nimport torch.nn as nn\nfrom layers.Autoformer_EncDec import series_decomp\n\nclass Model(nn.Module):",
3  "op": "replace",
4  "filename": "Time-Series-Library/models/Custom.py",
5  "start_line": 1,
6  "end_line": 9
7}

1{
2  "filename": "Time-Series-Library/models/Custom.py",
3  "start_line": 22,
4  "op": "replace",
5  "end_line": 38,
6  "content": "        # Time-domain forecasting layers (per-channel)\n        self.linear_seasonal = nn.Linear(self.seq_len, self.pred_len)\n        self.linear_trend = nn.Linear(self.seq_len, self.pred_len)\n        \n        # Proper DLinear initialization for temporal layers (crucial for stability)\n        self.linear_seasonal.weight = nn.Parameter(\n            (1 / self.seq_len) * torch.ones([self.pred_len, self.seq_len]))\n        self.linear_trend.weight = nn.Parameter(\n            (1 / self.seq_len) * torch.ones([self.pred_len, self.seq_len]))\n        \n        # Hyperparameters for frequency mixing\n        self.k = getattr(configs, 'fcm_k', min(self.seq_len // 2 + 1, 16)) \n        \n        # Complex matrices for cross-variable modeling per frequency mode\n        # Initialized with zero for exact DLinear behavior at initialization\n        self.freq_mix_seasonal = nn.Parameter(\n            torch.zeros(self.k, self.channels, self.channels, dtype=torch.cfloat)\n        )\n        self.freq_mix_trend = nn.Parameter(\n            torch.zeros(self.k, self.channels, self.channels, dtype=torch.cfloat)\n        )"
7}

1{
2  "end_line": 81,
3  "start_line": 26,
4  "filename": "Time-Series-Library/models/Custom.py",
5  "op": "replace",
6  "content": "        # Proper DLinear initialization for temporal layers (crucial for stability)\n        self.linear_seasonal.weight = nn.Parameter(\n            (1 / self.seq_len) * torch.ones([self.pred_len, self.seq_len]))\n        self.linear_trend.weight = nn.Parameter(\n            (1 / self.seq_len) * torch.ones([self.pred_len, self.seq_len]))\n        \n        # Hyperparameters for frequency mixing\n        self.k = getattr(configs, 'fcm_k', min(self.seq_len // 2 + 1, 16)) \n        \n        # Complex matrices for cross-variable modeling per frequency mode (only for seasonal part)\n        # Initialized with zero for exact DLinear behavior at initialization\n        self.freq_mix_seasonal = nn.Parameter(\n            torch.zeros(self.k, self.channels, self.channels, dtype=torch.cfloat)\n        )\n        \n    def forecast(self, x_enc, x_mark_enc, x_dec, x_mark_dec):\n        # x_enc: [B, L, C]\n\n        # Reversible Instance Normalization to address distribution shift\n        means = x_enc.mean(1, keepdim=True).detach()\n        x = x_enc - means\n        stdev = torch.sqrt(torch.var(x, dim=1, keepdim=True, unbiased=False) + 1e-5)\n        x = x / stdev\n        \n        # Temporal Decomposition\n        seasonal, trend = self.decomp(x)\n        \n        # --- Frequency-Domain Cross-Channel Mixing (Seasonal Component Only) ---\n        s_fft = torch.fft.rfft(seasonal, dim=1)\n        s_fft_top = s_fft[:, :self.k, :] \n        s_mix = torch.einsum('bkc,kcd->bkd', s_fft_top, self.freq_mix_seasonal)\n        \n        s_fft_new = s_fft.clone()\n        s_fft_new[:, :self.k, :] = s_fft_new[:, :self.k, :] + s_mix\n        seasonal_mixed = torch.fft.irfft(s_fft_new, n=self.seq_len, dim=1)\n        \n        # --- Time-Domain Independent Forecasting ---\n        # Note: trend is unchanged, we only cross-channel mix the seasonal elements\n        seasonal_out = self.linear_seasonal(seasonal_mixed.transpose(1, 2)).transpose(1, 2)\n        trend_out = self.linear_trend(trend.transpose(1, 2)).transpose(1, 2)\n        \n        dec_out = seasonal_out + trend_out"
7}