KAN 的可逆混合 - RMoK

​

1. 辅助函数

​

1.1 WaveKAN

​

WaveKANLayer

 WaveKANLayer (in_features, out_features, wavelet_type='mexican_hat',
               with_bn=True, device='cpu')

*这是论文模拟的示例代码：Bozorgasl, Zavareh 和 Chen, Hao, Wav-KAN: Wavelet Kolmogorov-Arnold Networks (2024 年 5 月)

https://arxiv.org/abs/2405.12832 也可以在这里获取：https://papers.ssrn.com/sol3/papers.cfm?abstract_id=4835325 我们使用了高效的 KAN 表示法和部分代码：+*

​

1.2 TaylorKAN

​

TaylorKANLayer

 TaylorKANLayer (input_dim, out_dim, order, addbias=True)

https://github.com/Muyuzhierchengse/TaylorKAN/

​

1.3. JacobiKAN

​

JacobiKANLayer

 JacobiKANLayer (input_dim, output_dim, degree, a=1.0, b=1.0)

https://github.com/SpaceLearner/JacobiKAN/blob/main/JacobiKANLayer.py

​

2. 模型

​

RMoK

 RMoK (h, input_size, n_series:int, futr_exog_list=None,
       hist_exog_list=None, stat_exog_list=None, taylor_order:int=3,
       jacobi_degree:int=6, wavelet_function:str='mexican_hat',
       dropout:float=0.1, revin_affine:bool=True, loss=MAE(),
       valid_loss=None, max_steps:int=1000, learning_rate:float=0.001,
       num_lr_decays:int=-1, early_stop_patience_steps:int=-1,
       val_check_steps:int=100, batch_size:int=32,
       valid_batch_size:Optional[int]=None, windows_batch_size=32,
       inference_windows_batch_size=32, start_padding_enabled=False,
       step_size:int=1, scaler_type:str='identity', random_seed:int=1,
       drop_last_loader:bool=False, alias:Optional[str]=None,
       optimizer=None, optimizer_kwargs=None, lr_scheduler=None,
       lr_scheduler_kwargs=None, dataloader_kwargs=None, **trainer_kwargs)

*KAN 的可逆混合

参数
h: int，预测范围。
input_size: int，自回归输入大小，y=[1,2,3,4] input_size=2 -> y_[t-2:t]=[1,2]。
n_series: int，时间序列数量。
futr_exog_list: str 列表，未来外部变量列。
hist_exog_list: str 列表，历史外部变量列。
stat_exog_list: str 列表，静态外部变量列。
taylor_order: int，泰勒多项式阶数。
jacobi_degree: int，雅可比多项式次数。
wavelet_function: str，在 WaveKAN 中使用的小波函数。从 [“mexican_hat”, “morlet”, “dog”, “meyer”, “shannon”] 中选择
dropout: float，dropout 率。
revin_affine: bool=False，在 RevIn 中使用 affine 的布尔值。
loss: PyTorch 模块，从 损失集合 中实例化的训练损失类。
valid_loss: PyTorch 模块=loss，从 损失集合 中实例化的验证损失类。
max_steps: int=1000，最大训练步数。
learning_rate: float=1e-3，学习率，范围 (0, 1)。
num_lr_decays: int=-1，学习率衰减次数，平均分布在最大步数中。
early_stop_patience_steps: int=-1，早停前验证迭代次数。
val_check_steps: int=100，每次验证损失检查之间的训练步数。
batch_size: int=32，每个批次中不同序列的数量。
valid_batch_size: int=None，每个验证和测试批次中不同序列的数量，如果为 None，则使用 batch_size。
windows_batch_size: int=32，每个训练批次中采样的窗口数量，默认使用所有窗口。
inference_windows_batch_size: int=32，每个推理批次中采样的窗口数量，-1 使用所有窗口。
start_padding_enabled: bool=False，如果为 True，模型将根据输入大小在时间序列开头用零填充。
step_size: int=1，每个时间数据窗口之间的步长。
scaler_type: str=‘identity’，用于时间输入归一化的缩放器类型，参见 时间缩放器。
random_seed: int=1，用于 PyTorch 初始化器和 NumPy 生成器的随机种子。
drop_last_loader: bool=False，如果为 True，TimeSeriesDataLoader 将丢弃最后一个非完整批次。
alias: str，可选，模型的自定义名称。
optimizer: ‘torch.optim.Optimizer’ 的子类，可选，用户指定的优化器，而非默认选择 (Adam)。
optimizer_kwargs: dict，可选，用户指定的 optimizer 使用的参数列表。
lr_scheduler: ‘torch.optim.lr_scheduler.LRScheduler’ 的子类，可选，用户指定的 lr_scheduler，而非默认选择 (StepLR)。
lr_scheduler_kwargs: dict，可选，用户指定的 lr_scheduler 使用的参数列表。
dataloader_kwargs: dict，可选，由 TimeSeriesDataLoader 传递给 PyTorch Lightning dataloader 的参数列表。
**trainer_kwargs: int，从 PyTorch Lighning 的 trainer 继承的关键字 trainer 参数。

参考文献
- Xiao Han, Xinfeng Zhang, Yiling Wu, Zhenduo Zhang, Zhe Wu.“KAN4TSF: KAN 及基于 KAN 的模型对于时间序列预测有效吗？”。arXiv.
*

​

RMoK.fit

 RMoK.fit (dataset, val_size=0, test_size=0, random_seed=None,
           distributed_config=None)

*拟合。

fit 方法使用初始化参数（learning_rate、windows_batch_size 等）和初始化时定义的 loss 函数来优化神经网络的权重。在 fit 中，我们使用继承初始化 self.trainer_kwargs 的 PyTorch Lightning Trainer 来定制其输入，参见 PL 的 trainer 参数。

该方法设计用于兼容类似 SKLearn 的类，特别是与 StatsForecast 库兼容。

默认情况下，model 不保存训练检查点以保护磁盘内存，要保存检查点，请在 __init__ 中将 enable_checkpointing 更改为 True。

参数
dataset: NeuralForecast 的 TimeSeriesDataset，参见 文档。
val_size: int，用于时间交叉验证的验证集大小。
random_seed: int=None，用于 PyTorch 初始化器和 NumPy 生成器的随机种子，将覆盖 model.__init__ 中的设置。
test_size: int，用于时间交叉验证的测试集大小。
*

​

RMoK.predict

 RMoK.predict (dataset, test_size=None, step_size=1, random_seed=None,
               quantiles=None, **data_module_kwargs)

*预测。

使用 PL 的 Trainer 执行 predict_step 进行神经网络预测。

参数
dataset: NeuralForecast 的 TimeSeriesDataset，参见 文档。
test_size: int=None，用于时间交叉验证的测试集大小。
step_size: int=1，每个窗口之间的步长。
random_seed: int=None，用于 PyTorch 初始化器和 NumPy 生成器的随机种子，将覆盖 model.__init__ 中的设置。
quantiles: float 列表，可选（默认=None），要预测的目标分位数。
**data_module_kwargs: PL 的 TimeSeriesDataModule 参数，参见 文档。*

​

3. 使用示例

import pandas as pd
import matplotlib.pyplot as plt

from neuralforecast import NeuralForecast
from neuralforecast.models import RMoK
from neuralforecast.utils import AirPassengersPanel, AirPassengersStatic
from neuralforecast.losses.pytorch import MSE

Y_train_df = AirPassengersPanel[AirPassengersPanel.ds<AirPassengersPanel['ds'].values[-12]].reset_index(drop=True) # 132 train
Y_test_df = AirPassengersPanel[AirPassengersPanel.ds>=AirPassengersPanel['ds'].values[-12]].reset_index(drop=True) # 12 test

model = RMoK(h=12,
             input_size=24,
             n_series=2,
             taylor_order=3,
             jacobi_degree=6,
             wavelet_function='mexican_hat',
             dropout=0.1,
             revin_affine=True,
             loss=MSE(),
             valid_loss=MAE(),
             early_stop_patience_steps=3,
             batch_size=32)

fcst = NeuralForecast(models=[model], freq='ME')
fcst.fit(df=Y_train_df, static_df=AirPassengersStatic, val_size=12)
forecasts = fcst.predict(futr_df=Y_test_df)

# Plot predictions
fig, ax = plt.subplots(1, 1, figsize = (20, 7))
Y_hat_df = forecasts.reset_index(drop=False).drop(columns=['unique_id','ds'])
plot_df = pd.concat([Y_test_df, Y_hat_df], axis=1)
plot_df = pd.concat([Y_train_df, plot_df])

plot_df = plot_df[plot_df.unique_id=='Airline1'].drop('unique_id', axis=1)
plt.plot(plot_df['ds'], plot_df['y'], c='black', label='True')
plt.plot(plot_df['ds'], plot_df['RMoK'], c='blue', label='Forecast')
ax.set_title('AirPassengers Forecast', fontsize=22)
ax.set_ylabel('Monthly Passengers', fontsize=20)
ax.set_xlabel('Year', fontsize=20)
ax.legend(prop={'size': 15})
ax.grid()