核心类

Source code in src/pytorch_tabular/tabular_model.py

class TabularModel:
    def __init__(
        self,
        config: Optional[DictConfig] = None,
        data_config: Optional[Union[DataConfig, str]] = None,
        model_config: Optional[Union[ModelConfig, str]] = None,
        optimizer_config: Optional[Union[OptimizerConfig, str]] = None,
        trainer_config: Optional[Union[TrainerConfig, str]] = None,
        experiment_config: Optional[Union[ExperimentConfig, str]] = None,
        model_callable: Optional[Callable] = None,
        model_state_dict_path: Optional[Union[str, Path]] = None,
        verbose: bool = True,
        suppress_lightning_logger: bool = False,
    ) -> None:
        """核心模型,负责协调从初始化数据模块、模型、训练器等所有内容.

Parameters:
    config (Optional[Union[DictConfig, str]], 可选): 单个OmegaConf DictConfig对象或包含所有配置参数的yaml文件路径.默认为None.

    data_config (Optional[Union[DataConfig, str]], 可选): DataConfig对象或yaml文件路径.默认为None.

    model_config (Optional[Union[ModelConfig, str]], 可选): ModelConfig的子类或yaml文件路径.
        根据配置类型确定运行哪个模型.默认为None.

    optimizer_config (Optional[Union[OptimizerConfig, str]], 可选): OptimizerConfig对象或yaml文件路径.默认为None.

    trainer_config (Optional[Union[TrainerConfig, str]], 可选): TrainerConfig对象或yaml文件路径.默认为None.

    experiment_config (Optional[Union[ExperimentConfig, str]], 可选): ExperimentConfig对象或yaml文件路径.
        如果提供,将配置实验跟踪.默认为None.

    model_callable (Optional[Callable], 可选): 如果提供,将覆盖从配置加载的模型可调用对象.
        通常在提供自定义模型时使用.

    model_state_dict_path (Optional[Union[str, Path]], 可选): 如果提供,将在从配置初始化模型后加载状态字典.

    verbose (bool): 控制日志记录的开关.默认为True.

    suppress_lightning_logger (bool): 如果为True,将抑制PyTorch Lightning的默认日志记录.默认为False.
"""
        super().__init__()
        if suppress_lightning_logger:
            suppress_lightning_logs()
        self.verbose = verbose
        self.exp_manager = ExperimentRunManager()
        if config is None:
            assert any(c is not None for c in (data_config, model_config, optimizer_config, trainer_config)), (
                "If `config` is None, `data_config`, `model_config`,"
                " `trainer_config`, and `optimizer_config` cannot be None"
            )
            data_config = self._read_parse_config(data_config, DataConfig)
            model_config = self._read_parse_config(model_config, ModelConfig)
            trainer_config = self._read_parse_config(trainer_config, TrainerConfig)
            optimizer_config = self._read_parse_config(optimizer_config, OptimizerConfig)
            if model_config.task != "ssl":
                assert data_config.target is not None, (
                    "`target` in data_config should not be None for" f" {model_config.task} task"
                )
            if experiment_config is None:
                if self.verbose:
                    logger.info("Experiment Tracking is turned off")
                self.track_experiment = False
                self.config = OmegaConf.merge(
                    OmegaConf.to_container(data_config),
                    OmegaConf.to_container(model_config),
                    OmegaConf.to_container(trainer_config),
                    OmegaConf.to_container(optimizer_config),
                )
            else:
                experiment_config = self._read_parse_config(experiment_config, ExperimentConfig)
                self.track_experiment = True
                self.config = OmegaConf.merge(
                    OmegaConf.to_container(data_config),
                    OmegaConf.to_container(model_config),
                    OmegaConf.to_container(trainer_config),
                    OmegaConf.to_container(experiment_config),
                    OmegaConf.to_container(optimizer_config),
                )
        else:
            self.config = config
            if hasattr(config, "log_target") and (config.log_target is not None):
                # experiment_config = OmegaConf.structured(experiment_config)
                self.track_experiment = True
            else:
                if self.verbose:
                    logger.info("Experiment Tracking is turned off")
                self.track_experiment = False

        self.run_name, self.uid = self._get_run_name_uid()
        if self.track_experiment:
            self._setup_experiment_tracking()
        else:
            self.logger = None

        self.exp_manager = ExperimentRunManager()
        if model_callable is None:
            self.model_callable = getattr_nested(self.config._module_src, self.config._model_name)
            self.custom_model = False
        else:
            self.model_callable = model_callable
            self.custom_model = True
        self.model_state_dict_path = model_state_dict_path
        self._is_config_updated_with_data = False
        self._run_validation()
        self._is_fitted = False

    @property
    def has_datamodule(self):
        if hasattr(self, "datamodule") and self.datamodule is not None:
            return True
        else:
            return False

    @property
    def has_model(self):
        if hasattr(self, "model") and self.model is not None:
            return True
        else:
            return False

    @property
    def is_fitted(self):
        return self._is_fitted

    @property
    def name(self):
        if self.has_model:
            return self.model.__class__.__name__
        else:
            return self.config._model_name

    @property
    def num_params(self):
        if self.has_model:
            return count_parameters(self.model)

    def _run_validation(self):
        """验证配置参数,如果有问题则抛出错误."""
        if self.config.task == "regression":
            if self.config.target_range is not None:
                if (
                    (len(self.config.target_range) != len(self.config.target))
                    or any(len(range_) != 2 for range_ in self.config.target_range)
                    or any(range_[0] > range_[1] for range_ in self.config.target_range)
                ):
                    raise ValueError(
                        "Targe Range, if defined, should be list tuples of length"
                        " two(min,max). The length of the list should be equal to hte"
                        " length of target columns"
                    )

    def _read_parse_config(self, config, cls):
        if isinstance(config, str):
            if os.path.exists(config):
                _config = OmegaConf.load(config)
                if cls == ModelConfig:
                    cls = getattr_nested(_config._module_src, _config._config_name)
                config = cls(
                    **{
                        k: v
                        for k, v in _config.items()
                        if (k in cls.__dataclass_fields__.keys()) and (cls.__dataclass_fields__[k].init)
                    }
                )
            else:
                raise ValueError(f"{config} is not a valid path")
        config = OmegaConf.structured(config)
        return config

    def _get_run_name_uid(self) -> Tuple[str, int]:
        """获取实验名称并将版本号加1.

Returns:
    tuple[str, int]: 返回名称和版本号
"""
        if hasattr(self.config, "run_name") and self.config.run_name is not None:
            name = self.config.run_name
        elif hasattr(self.config, "checkpoints_name") and self.config.checkpoints_name is not None:
            name = self.config.checkpoints_name
        else:
            name = self.config.task
        uid = self.exp_manager.update_versions(name)
        return name, uid

    def _setup_experiment_tracking(self):
        """根据在实验配置中所做的选择,设置实验跟踪框架."""
        if self.config.log_target == "tensorboard":
            self.logger = pl.loggers.TensorBoardLogger(
                name=self.run_name, save_dir=self.config.project_name, version=self.uid
            )
        elif self.config.log_target == "wandb":
            self.logger = pl.loggers.WandbLogger(
                name=f"{self.run_name}_{self.uid}",
                project=self.config.project_name,
                offline=False,
            )
        else:
            raise NotImplementedError(
                f"{self.config.log_target} is not implemented. Try one of [wandb," " tensorboard]"
            )

    def _prepare_callbacks(self, callbacks=None) -> List:
        """    根据配置准备训练器所需的回调.

Returns:
    List: 回调列表
"""
        callbacks = [] if callbacks is None else callbacks
        if self.config.early_stopping is not None:
            early_stop_callback = pl.callbacks.early_stopping.EarlyStopping(
                monitor=self.config.early_stopping,
                min_delta=self.config.early_stopping_min_delta,
                patience=self.config.early_stopping_patience,
                mode=self.config.early_stopping_mode,
                **self.config.early_stopping_kwargs,
            )
            callbacks.append(early_stop_callback)
        if self.config.checkpoints:
            ckpt_name = f"{self.run_name}-{self.uid}"
            ckpt_name = ckpt_name.replace(" ", "_") + "_{epoch}-{valid_loss:.2f}"
            model_checkpoint = pl.callbacks.ModelCheckpoint(
                monitor=self.config.checkpoints,
                dirpath=self.config.checkpoints_path,
                filename=ckpt_name,
                save_top_k=self.config.checkpoints_save_top_k,
                mode=self.config.checkpoints_mode,
                every_n_epochs=self.config.checkpoints_every_n_epochs,
                **self.config.checkpoints_kwargs,
            )
            callbacks.append(model_checkpoint)
            self.config.enable_checkpointing = True
        else:
            self.config.enable_checkpointing = False
        if self.config.progress_bar == "rich" and self.config.trainer_kwargs.get("enable_progress_bar", True):
            callbacks.append(RichProgressBar())
        if self.verbose:
            logger.debug(f"Callbacks used: {callbacks}")
        return callbacks

    def _prepare_trainer(self, callbacks: List, max_epochs: int = None, min_epochs: int = None) -> pl.Trainer:
        """Prepares the Trainer object.

        Args:
            callbacks (List): A list of callbacks to be used
            max_epochs (int, optional): Maximum number of epochs to train for. Defaults to None.
            min_epochs (int, optional): Minimum number of epochs to train for. Defaults to None.

        Returns:
            pl.Trainer: A PyTorch Lightning Trainer object

        """
        if self.verbose:
            logger.info("Preparing the Trainer")
        if max_epochs is not None:
            self.config.max_epochs = max_epochs
        if min_epochs is not None:
            self.config.min_epochs = min_epochs
        # Getting Trainer Arguments from the init signature
        trainer_sig = inspect.signature(pl.Trainer.__init__)
        trainer_args = [p for p in trainer_sig.parameters.keys() if p != "self"]
        trainer_args_config = {k: v for k, v in self.config.items() if k in trainer_args}
        # For some weird reason, checkpoint_callback is not appearing in the Trainer vars
        trainer_args_config["enable_checkpointing"] = self.config.enable_checkpointing
        # turn off progress bar if progress_bar=='none'
        trainer_args_config["enable_progress_bar"] = self.config.progress_bar != "none"
        # Adding trainer_kwargs from config to trainer_args
        trainer_args_config.update(self.config.trainer_kwargs)
        if trainer_args_config["devices"] == -1:
            # Setting devices to auto if -1 so that lightning will use all available GPUs/CPUs
            trainer_args_config["devices"] = "auto"
        return pl.Trainer(
            logger=self.logger,
            callbacks=callbacks,
            **trainer_args_config,
        )

    def _check_and_set_target_transform(self, target_transform):
        if target_transform is not None:
            if isinstance(target_transform, Iterable):
                assert len(target_transform) == 2, (
                    "If `target_transform` is a tuple, it should have and only have"
                    " forward and backward transformations"
                )
            elif isinstance(target_transform, TransformerMixin):
                pass
            else:
                raise ValueError(
                    "`target_transform` should wither be an sklearn Transformer or a" " tuple of callables."
                )
        if self.config.task == "classification" and target_transform is not None:
            logger.warning("For classification task, target transform is not used. Ignoring the" " parameter")
            target_transform = None
        return target_transform

    def _prepare_for_training(self, model, datamodule, callbacks=None, max_epochs=None, min_epochs=None):
        self.callbacks = self._prepare_callbacks(callbacks)
        self.trainer = self._prepare_trainer(self.callbacks, max_epochs, min_epochs)
        self.model = model
        self.datamodule = datamodule

    @classmethod
    def _load_weights(cls, model, path: Union[str, Path]) -> None:
        """Loads the model weights in the specified directory.

        Args:
            path (str): The path to the file to load the model from

        Returns:
            None

        """
        ckpt = pl_load(path, map_location=lambda storage, loc: storage)
        model.load_state_dict(ckpt.get("state_dict") or ckpt)

    @classmethod
    def load_model(cls, dir: str, map_location=None, strict=True):
        """加载保存在目录中的模型.

Parameters:
    dir (str): 保存模型的目录,包含检查点
    map_location (Union[Dict[str, str], str, device, int, Callable, None]) : 如果你的检查点保存了一个GPU模型,而你现在在CPU上或不同数量的GPU上加载,使用这个参数来映射到新的设置.行为与torch.load()中的相同
    strict (bool) : 是否严格要求checkpoint_path中的键与该模块的状态字典返回的键完全匹配.默认值: True.

Returns:
    TabularModel (TabularModel): 保存的TabularModel
"""
        config = OmegaConf.load(os.path.join(dir, "config.yml"))
        datamodule = joblib.load(os.path.join(dir, "datamodule.sav"))
        if (
            hasattr(config, "log_target")
            and (config.log_target is not None)
            and os.path.exists(os.path.join(dir, "exp_logger.sav"))
        ):
            logger = joblib.load(os.path.join(dir, "exp_logger.sav"))
        else:
            logger = None
        if os.path.exists(os.path.join(dir, "callbacks.sav")):
            callbacks = joblib.load(os.path.join(dir, "callbacks.sav"))
            # Excluding Gradient Accumulation Scheduler Callback as we are creating
            # a new one in trainer
            callbacks = [c for c in callbacks if not isinstance(c, GradientAccumulationScheduler)]
        else:
            callbacks = []
        if os.path.exists(os.path.join(dir, "custom_model_callable.sav")):
            model_callable = joblib.load(os.path.join(dir, "custom_model_callable.sav"))
            custom_model = True
        else:
            model_callable = getattr_nested(config._module_src, config._model_name)
            # model_callable = getattr(
            #     getattr(models, config._module_src), config._model_name
            # )
            custom_model = False
        inferred_config = datamodule.update_config(config)
        inferred_config = OmegaConf.structured(inferred_config)
        model_args = {
            "config": config,
            "inferred_config": inferred_config,
        }
        custom_params = joblib.load(os.path.join(dir, "custom_params.sav"))
        if custom_params.get("custom_loss") is not None:
            model_args["loss"] = "MSELoss"  # For compatibility. Not Used
        if custom_params.get("custom_metrics") is not None:
            model_args["metrics"] = ["mean_squared_error"]  # For compatibility. Not Used
            model_args["metrics_params"] = [{}]  # For compatibility. Not Used
            model_args["metrics_prob_inputs"] = [False]  # For compatibility. Not Used
        if custom_params.get("custom_optimizer") is not None:
            model_args["optimizer"] = "Adam"  # For compatibility. Not Used
        if custom_params.get("custom_optimizer_params") is not None:
            model_args["optimizer_params"] = {}  # For compatibility. Not Used

        # Initializing with default metrics, losses, and optimizers. Will revert once initialized
        try:
            model = model_callable.load_from_checkpoint(
                checkpoint_path=os.path.join(dir, "model.ckpt"),
                map_location=map_location,
                strict=strict,
                **model_args,
            )
        except RuntimeError as e:
            if (
                "Unexpected key(s) in state_dict" in str(e)
                and "loss.weight" in str(e)
                and "custom_loss.weight" in str(e)
            ):
                # Custom loss will be loaded after the model is initialized
                # continuing with strict=False
                model = model_callable.load_from_checkpoint(
                    checkpoint_path=os.path.join(dir, "model.ckpt"),
                    map_location=map_location,
                    strict=False,
                    **model_args,
                )
            else:
                raise e
        if custom_params.get("custom_optimizer") is not None:
            model.custom_optimizer = custom_params["custom_optimizer"]
        if custom_params.get("custom_optimizer_params") is not None:
            model.custom_optimizer_params = custom_params["custom_optimizer_params"]
        if custom_params.get("custom_loss") is not None:
            model.loss = custom_params["custom_loss"]
        if custom_params.get("custom_metrics") is not None:
            model.custom_metrics = custom_params.get("custom_metrics")
            model.hparams.metrics = [m.__name__ for m in custom_params.get("custom_metrics")]
            model.hparams.metrics_params = [{}]
            model.hparams.metrics_prob_input = custom_params.get("custom_metrics_prob_inputs")
        model._setup_loss()
        model._setup_metrics()
        tabular_model = cls(config=config, model_callable=model_callable)
        tabular_model.model = model
        tabular_model.custom_model = custom_model
        tabular_model.datamodule = datamodule
        tabular_model.callbacks = callbacks
        tabular_model.trainer = tabular_model._prepare_trainer(callbacks=callbacks)
        # tabular_model.trainer.model = model
        tabular_model.logger = logger
        return tabular_model

    def prepare_dataloader(
        self,
        train: DataFrame,
        validation: Optional[DataFrame] = None,
        train_sampler: Optional[torch.utils.data.Sampler] = None,
        target_transform: Optional[Union[TransformerMixin, Tuple]] = None,
        seed: Optional[int] = 42,
        cache_data: str = "memory",
    ) -> TabularDatamodule:
        """准备用于训练和验证的数据加载器.

Parameters:
    train (DataFrame): 训练数据框

    validation (Optional[DataFrame], optional):
        如果提供,将在训练过程中使用此数据框作为验证集.
        用于早停和日志记录.如果未提供,将使用训练数据的20%作为验证集.
        默认为 None.

    train_sampler (Optional[torch.utils.data.Sampler], optional):
        自定义的 PyTorch 批次采样器,将传递给 DataLoaders.
        适用于处理不平衡数据和其他自定义批次策略.

    target_transform (Optional[Union[TransformerMixin, Tuple(Callable)]], optional):
        如果提供,在模型训练前对目标应用此变换,并在预测时应用逆变换.
        参数可以是具有 inverse_transform 方法的 sklearn Transformer,或
        由可调用对象组成的元组 (transform_func, inverse_transform_func).

    seed (Optional[int], optional): 用于可重复性的随机种子.默认为 42.

    cache_data (str): 决定如何在数据加载器中缓存数据.如果设置为
        "memory",将在内存中缓存.如果设置为有效路径,将在该路径中缓存.默认为 "memory".

Returns:
    TabularDatamodule: 准备好的数据模块
"""
        if self.verbose:
            logger.info("Preparing the DataLoaders")
        target_transform = self._check_and_set_target_transform(target_transform)

        datamodule = TabularDatamodule(
            train=train,
            validation=validation,
            config=self.config,
            target_transform=target_transform,
            train_sampler=train_sampler,
            seed=seed,
            cache_data=cache_data,
            verbose=self.verbose,
        )
        datamodule.prepare_data()
        datamodule.setup("fit")
        return datamodule

    def prepare_model(
        self,
        datamodule: TabularDatamodule,
        loss: Optional[torch.nn.Module] = None,
        metrics: Optional[List[Callable]] = None,
        metrics_prob_inputs: Optional[List[bool]] = None,
        optimizer: Optional[torch.optim.Optimizer] = None,
        optimizer_params: Dict = None,
    ) -> BaseModel:
        """准备模型以进行训练.

Parameters:
    datamodule (TabularDatamodule): 数据模块

    loss (Optional[torch.nn.Module], 可选): 自定义损失函数,不在标准 PyTorch 库中

    metrics (Optional[List[Callable]], 可选): 自定义度量函数（可调用对象）,具有 metric_fn(y_hat, y) 签名并作用于 torch 张量输入

    metrics_prob_inputs (Optional[List[bool]], 可选): 这是分类度量的必填参数.如果度量函数需要概率作为输入,请设置为 True.
        列表的长度应等于度量函数的数量.默认为 None.

    optimizer (Optional[torch.optim.Optimizer], 可选):
        自定义优化器,是标准 PyTorch 优化器的直接替代品.
        这应该是类,而不是初始化的对象

    optimizer_params (Optional[Dict], 可选): 用于初始化自定义优化器的参数.

Returns:
    BaseModel: 准备好的模型
"""
        if self.verbose:
            logger.info(f"Preparing the Model: {self.config._model_name}")
        # Fetching the config as some data specific configs have been added in the datamodule
        self.inferred_config = self._read_parse_config(datamodule.update_config(self.config), InferredConfig)
        model = self.model_callable(
            self.config,
            custom_loss=loss,  # Unused in SSL tasks
            custom_metrics=metrics,  # Unused in SSL tasks
            custom_metrics_prob_inputs=metrics_prob_inputs,  # Unused in SSL tasks
            custom_optimizer=optimizer,
            custom_optimizer_params=optimizer_params or {},
            inferred_config=self.inferred_config,
        )
        # Data Aware Initialization(for the models that need it)
        model.data_aware_initialization(datamodule)
        if self.model_state_dict_path is not None:
            self._load_weights(model, self.model_state_dict_path)
        if self.track_experiment and self.config.log_target == "wandb":
            self.logger.watch(model, log=self.config.exp_watch, log_freq=self.config.exp_log_freq)
        return model

    def train(
        self,
        model: pl.LightningModule,
        datamodule: TabularDatamodule,
        callbacks: Optional[List[pl.Callback]] = None,
        max_epochs: int = None,
        min_epochs: int = None,
        handle_oom: bool = True,
    ) -> pl.Trainer:
        """    训练模型.

Parameters:
    model (pl.LightningModule): 要训练的PyTorch Lightning模型.

    datamodule (TabularDatamodule): 数据模块

    callbacks (Optional[List[pl.Callback]], optional):
        训练期间使用的回调函数列表.默认为None.

    max_epochs (Optional[int]): 覆盖要运行的最大epoch数.默认为None.

    min_epochs (Optional[int]): 覆盖要运行的最小epoch数.默认为None.

    handle_oom (bool): 如果为True,将尝试优雅地处理OOM错误.默认为True.

Returns:
    pl.Trainer: PyTorch Lightning Trainer实例
"""
        self._prepare_for_training(model, datamodule, callbacks, max_epochs, min_epochs)
        train_loader, val_loader = (
            self.datamodule.train_dataloader(),
            self.datamodule.val_dataloader(),
        )
        self.model.train()
        if self.config.auto_lr_find and (not self.config.fast_dev_run):
            if self.verbose:
                logger.info("Auto LR Find Started")
            with OutOfMemoryHandler(handle_oom=handle_oom) as oom_handler:
                result = Tuner(self.trainer).lr_find(
                    self.model,
                    train_dataloaders=train_loader,
                    val_dataloaders=val_loader,
                )
            if oom_handler.oom_triggered:
                raise OOMException(
                    "OOM detected during LR Find. Try reducing your batch_size or the"
                    " model parameters." + "/n" + "Original Error: " + oom_handler.oom_msg
                )
            if self.verbose:
                logger.info(
                    f"Suggested LR: {result.suggestion()}. For plot and detailed"
                    " analysis, use `find_learning_rate` method."
                )
            self.model.reset_weights()
            # Parameters in models needs to be initialized again after LR find
            self.model.data_aware_initialization(self.datamodule)
        self.model.train()
        if self.verbose:
            logger.info("Training Started")
        with OutOfMemoryHandler(handle_oom=handle_oom) as oom_handler:
            self.trainer.fit(self.model, train_loader, val_loader)
        if oom_handler.oom_triggered:
            raise OOMException(
                "OOM detected during Training. Try reducing your batch_size or the"
                " model parameters."
                "/n" + "Original Error: " + oom_handler.oom_msg
            )
        self._is_fitted = True
        if self.verbose:
            logger.info("Training the model completed")
        if self.config.load_best:
            self.load_best_model()
        return self.trainer

    def fit(
        self,
        train: Optional[DataFrame],
        validation: Optional[DataFrame] = None,
        loss: Optional[torch.nn.Module] = None,
        metrics: Optional[List[Callable]] = None,
        metrics_prob_inputs: Optional[List[bool]] = None,
        optimizer: Optional[torch.optim.Optimizer] = None,
        optimizer_params: Dict = None,
        train_sampler: Optional[torch.utils.data.Sampler] = None,
        target_transform: Optional[Union[TransformerMixin, Tuple]] = None,
        max_epochs: Optional[int] = None,
        min_epochs: Optional[int] = None,
        seed: Optional[int] = 42,
        callbacks: Optional[List[pl.Callback]] = None,
        datamodule: Optional[TabularDatamodule] = None,
        cache_data: str = "memory",
        handle_oom: bool = True,
    ) -> pl.Trainer:
        """    fit方法,接收数据并触发训练.

Parameters:
    train (DataFrame): 训练数据框

    validation (Optional[DataFrame], optional):
        如果提供,将在训练过程中使用此数据框作为验证集.
        用于早停和日志记录.如果未提供,将使用20%的训练数据作为验证集.
        默认为None.

    loss (Optional[torch.nn.Module], optional): 自定义损失函数,不在标准PyTorch库中

    metrics (Optional[List[Callable]], optional): 自定义度量函数（可调用对象）,具有
        签名metric_fn(y_hat, y),并适用于torch张量输入.对于分类任务,y_hat预期形状为
        (batch_size, num_classes),对于回归任务,y_hat预期形状为(batch_size, 1),y预期形状为
        (batch_size, 1)

    metrics_prob_inputs (Optional[List[bool]], optional): 这是分类度量的强制参数.
        如果度量函数需要概率作为输入,请设置为True.
        列表的长度应等于度量函数的数量.默认为None.

    optimizer (Optional[torch.optim.Optimizer], optional):
        自定义优化器,是标准PyTorch优化器的替代品.
        这应该是类,而不是初始化的对象

    optimizer_params (Optional[Dict], optional): 用于初始化自定义优化器的参数.

    train_sampler (Optional[torch.utils.data.Sampler], optional):
        自定义PyTorch批次采样器,将传递给DataLoaders.
        对于处理不平衡数据和其他自定义批次策略很有用

    target_transform (Optional[Union[TransformerMixin, Tuple(Callable)]], optional):
        如果提供,在模型训练前对目标应用变换,在预测时应用逆变换.
        参数可以是具有inverse_transform方法的sklearn Transformer,
        或由可调用对象组成的元组(transform_func, inverse_transform_func)

    max_epochs (Optional[int]): 覆盖要运行的最大轮数.默认为None.

    min_epochs (Optional[int]): 覆盖要运行的最小轮数.默认为None.

    seed: (int): 用于可重复性的随机种子.默认为42.

    callbacks (Optional[List[pl.Callback]], optional):
        训练期间使用的回调列表.默认为None.

    datamodule (Optional[TabularDatamodule], optional): 数据模块.
        如果提供,将忽略其他参数如train、test等,并使用数据模块.
        默认为None.

    cache_data (str): 决定如何在数据加载器中缓存数据.如果设置为
        "memory",将在内存中缓存.如果设置为有效路径,将在该路径中缓存.默认为"memory".

    handle_oom (bool): 如果为True,将尝试优雅地处理OOM错误.默认为True.

Returns:
    pl.Trainer: PyTorch Lightning Trainer实例
"""
        assert self.config.task != "ssl", (
            "`fit` is not valid for SSL task. Please use `pretrain` for" " semi-supervised learning"
        )
        if metrics is not None:
            assert len(metrics) == len(
                metrics_prob_inputs or []
            ), "The length of `metrics` and `metrics_prob_inputs` should be equal"
        seed = seed or self.config.seed
        if seed:
            seed_everything(seed)
        if datamodule is None:
            datamodule = self.prepare_dataloader(
                train,
                validation,
                train_sampler,
                target_transform,
                seed,
                cache_data,
            )
        else:
            if train is not None:
                warnings.warn(
                    "train data and datamodule is provided."
                    " Ignoring the train data and using the datamodule."
                    " Set either one of them to None to avoid this warning."
                )
        model = self.prepare_model(
            datamodule,
            loss,
            metrics,
            metrics_prob_inputs,
            optimizer,
            optimizer_params or {},
        )

        return self.train(model, datamodule, callbacks, max_epochs, min_epochs, handle_oom)

    def pretrain(
        self,
        train: Optional[DataFrame],
        validation: Optional[DataFrame] = None,
        optimizer: Optional[torch.optim.Optimizer] = None,
        optimizer_params: Dict = None,
        # train_sampler: Optional[torch.utils.data.Sampler] = None,
        max_epochs: Optional[int] = None,
        min_epochs: Optional[int] = None,
        seed: Optional[int] = 42,
        callbacks: Optional[List[pl.Callback]] = None,
        datamodule: Optional[TabularDatamodule] = None,
        cache_data: str = "memory",
    ) -> pl.Trainer:
        """    预训练方法,接收数据并触发训练.

Parameters:
    train (DataFrame): 训练数据框

    validation (Optional[DataFrame], optional): 如果提供,将在训练过程中使用此数据框作为验证集.
        用于早停和日志记录.如果未提供,将使用训练数据的20%作为验证集.默认为None.

    optimizer (Optional[torch.optim.Optimizer], optional): 自定义优化器,可作为标准PyTorch优化器的替代品.
        应为类,而非初始化对象.

    optimizer_params (Optional[Dict], optional): 用于初始化自定义优化器的参数.

    max_epochs (Optional[int]): 覆盖要运行的最大周期数.默认为None.

    min_epochs (Optional[int]): 覆盖要运行的最小周期数.默认为None.

    seed: (int): 随机种子,用于可重复性.默认为42.

    callbacks (Optional[List[pl.Callback]], optional): 训练过程中使用的回调列表.
        默认为None.

    datamodule (Optional[TabularDatamodule], optional): 数据模块.如果提供,将忽略其他参数如train、test等,
        并使用数据模块.默认为None.

    cache_data (str): 决定如何在数据加载器中缓存数据.如果设置为"memory",将在内存中缓存.
        如果设置为有效路径,将在该路径中缓存.默认为"memory".
Returns:
    pl.Trainer: PyTorch Lightning Trainer实例
"""
        assert self.config.task == "ssl", (
            f"`pretrain` is not valid for {self.config.task} task. Please use `fit`" " instead."
        )
        seed = seed or self.config.seed
        if seed:
            seed_everything(seed)
        if datamodule is None:
            datamodule = self.prepare_dataloader(
                train,
                validation,
                train_sampler=None,
                target_transform=None,
                seed=seed,
                cache_data=cache_data,
            )
        else:
            if train is not None:
                warnings.warn(
                    "train data and datamodule is provided."
                    " Ignoring the train data and using the datamodule."
                    " Set either one of them to None to avoid this warning."
                )
        model = self.prepare_model(
            datamodule,
            optimizer,
            optimizer_params or {},
        )

        return self.train(model, datamodule, callbacks, max_epochs, min_epochs)

    def create_finetune_model(
        self,
        task: str,
        head: str,
        head_config: Dict,
        train: DataFrame,
        validation: Optional[DataFrame] = None,
        train_sampler: Optional[torch.utils.data.Sampler] = None,
        target_transform: Optional[Union[TransformerMixin, Tuple]] = None,
        target: Optional[str] = None,
        optimizer_config: Optional[OptimizerConfig] = None,
        trainer_config: Optional[TrainerConfig] = None,
        experiment_config: Optional[ExperimentConfig] = None,
        loss: Optional[torch.nn.Module] = None,
        metrics: Optional[List[Union[Callable, str]]] = None,
        metrics_prob_input: Optional[List[bool]] = None,
        metrics_params: Optional[Dict] = None,
        optimizer: Optional[torch.optim.Optimizer] = None,
        optimizer_params: Dict = None,
        learning_rate: Optional[float] = None,
        target_range: Optional[Tuple[float, float]] = None,
        seed: Optional[int] = 42,
    ):
        """创建一个新的TabularModel模型,使用预训练权重以及新的任务和头部.

Parameters:
    task (str): 要执行的任务.可以是 "regression" 或 "classification" 之一.

    head (str): 用于模型的头部.应为 `pytorch_tabular.models.common.heads` 中定义的头部之一.默认为 LinearHead.可选值为:
        [`None`,`LinearHead`,`MixtureDensityHead`].

    head_config (Dict): 定义头部的配置字典.如果留空,将初始化为默认的线性头部.

    train (DataFrame): 带有标签的训练数据.

    validation (Optional[DataFrame], optional): 带有标签的验证数据.默认为 None.

    train_sampler (Optional[torch.utils.data.Sampler], optional): 如果提供,将用作训练的批次采样器.默认为 None.

    target_transform (Optional[Union[TransformerMixin, Tuple]], optional): 如果提供,将在训练前用于转换目标,并在预测后进行逆转换.

    target (Optional[str], optional): 如果未在初始预训练阶段提供,则为目标列名称.默认为 None.

    optimizer_config (Optional[OptimizerConfig], optional):
        如果提供,将重新定义微调阶段的优化器.默认为 None.

    trainer_config (Optional[TrainerConfig], optional):
        如果提供,将重新定义微调阶段的训练器.默认为 None.

    experiment_config (Optional[ExperimentConfig], optional):
        如果提供,将重新定义微调阶段的实验配置.默认为 None.

    loss (Optional[torch.nn.Module], optional):
        如果提供,将用作微调阶段的损失函数.默认情况下,回归任务为 MSELoss,分类任务为 CrossEntropyLoss.

    metrics (Optional[List[Callable]], optional): 用于微调阶段的指标列表（可以是可调用对象或字符串）.如果是字符串,应为 ``torchmetrics.functional`` 中实现的功能性指标之一.默认为 None.

    metrics_prob_input (Optional[List[bool]], optional): 分类指标的强制参数.
        这定义了指标函数的输入是概率还是类别.长度应与指标数量相同.默认为 None.

    metrics_params (Optional[Dict], optional): 与指标顺序相同的指标参数.
        例如,多类别的 f1_score 需要参数 `average` 来完全定义指标.默认为 None.

    optimizer (Optional[torch.optim.Optimizer], optional):
        自定义优化器,是标准 PyTorch 优化器的替代品.如果提供,将忽略 OptimizerConfig.默认为 None.

    optimizer_params (Dict, optional): 优化器的参数.默认为 {}.

    learning_rate (Optional[float], optional): 要使用的学习率.默认为 1e-3.

    target_range (Optional[Tuple[float, float]], optional): 回归任务的目标范围.分类任务中忽略.默认为 None.

    seed (Optional[int], optional): 随机种子,用于可重复性.默认为 42.

Returns:
    TabularModel (TabularModel): 用于微调的新 TabularModel 模型
"""
        config = self.config
        optimizer_params = optimizer_params or {}
        if target is None:
            assert (
                hasattr(config, "target") and config.target is not None
            ), "`target` cannot be None if it was not set in the initial `DataConfig`"
        else:
            assert isinstance(target, list), "`target` should be a list of strings"
            config.target = target
        config.task = task
        # Add code to update configs with newly provided ones
        if optimizer_config is not None:
            for key, value in optimizer_config.__dict__.items():
                config[key] = value
            if len(optimizer_params) > 0:
                config.optimizer_params = optimizer_params
            else:
                config.optimizer_params = {}
        if trainer_config is not None:
            for key, value in trainer_config.__dict__.items():
                config[key] = value
        if experiment_config is not None:
            for key, value in experiment_config.__dict__.items():
                config[key] = value
        else:
            if self.track_experiment:
                # Renaming the experiment run so that a different log is created for finetuning
                if self.verbose:
                    logger.info("Renaming the experiment run for finetuning as" f" {config['run_name'] + '_finetuned'}")
                config["run_name"] = config["run_name"] + "_finetuned"

        datamodule = self.datamodule.copy(
            train=train,
            validation=validation,
            target_transform=target_transform,
            train_sampler=train_sampler,
            seed=seed,
            config_override={"target": target} if target is not None else {},
        )
        model_callable = _GenericModel
        inferred_config = OmegaConf.structured(datamodule._inferred_config)
        # Adding dummy attributes for compatibility. Not used because custom metrics are provided
        if not hasattr(config, "metrics"):
            config.metrics = "dummy"
        if not hasattr(config, "metrics_params"):
            config.metrics_params = {}
        if not hasattr(config, "metrics_prob_input"):
            config.metrics_prob_input = metrics_prob_input or [False]
        if metrics is not None:
            assert len(metrics) == len(metrics_params), "Number of metrics and metrics_params should be same"
            assert len(metrics) == len(metrics_prob_input), "Number of metrics and metrics_prob_input should be same"
            metrics = [getattr(torchmetrics.functional, m) if isinstance(m, str) else m for m in metrics]
        if task == "regression":
            loss = loss or torch.nn.MSELoss()
            if metrics is None:
                metrics = [torchmetrics.functional.mean_squared_error]
                metrics_params = [{}]
        elif task == "classification":
            loss = loss or torch.nn.CrossEntropyLoss()
            if metrics is None:
                metrics = [torchmetrics.functional.accuracy]
                metrics_params = [
                    {
                        "task": "multiclass",
                        "num_classes": inferred_config.output_dim,
                        "top_k": 1,
                    }
                ]
                metrics_prob_input = [False]
            else:
                for i, mp in enumerate(metrics_params):
                    # For classification task, output_dim == number of classses
                    metrics_params[i]["task"] = mp.get("task", "multiclass")
                    metrics_params[i]["num_classes"] = mp.get("num_classes", inferred_config.output_dim)
                    metrics_params[i]["top_k"] = mp.get("top_k", 1)
        else:
            raise ValueError(f"Task {task} not supported")
        # Forming partial callables using metrics and metric params
        metrics = [partial(m, **mp) for m, mp in zip(metrics, metrics_params)]
        self.model.mode = "finetune"
        if learning_rate is not None:
            config.learning_rate = learning_rate
        config.target_range = target_range
        model_args = {
            "backbone": self.model,
            "head": head,
            "head_config": head_config,
            "config": config,
            "inferred_config": inferred_config,
            "custom_loss": loss,
            "custom_metrics": metrics,
            "custom_metrics_prob_inputs": metrics_prob_input,
            "custom_optimizer": optimizer,
            "custom_optimizer_params": optimizer_params,
        }
        # Initializing with default metrics, losses, and optimizers. Will revert once initialized
        model = model_callable(
            **model_args,
        )
        tabular_model = TabularModel(config=config, verbose=self.verbose)
        tabular_model.model = model
        tabular_model.datamodule = datamodule
        # Setting a flag to identify this as a fine-tune model
        tabular_model._is_finetune_model = True
        return tabular_model

    def finetune(
        self,
        max_epochs: Optional[int] = None,
        min_epochs: Optional[int] = None,
        callbacks: Optional[List[pl.Callback]] = None,
        freeze_backbone: bool = False,
    ) -> pl.Trainer:
        """微调模型于提供的数据上.

Parameters:
    max_epochs (Optional[int], 可选): 训练的最大周期数.默认为 None.

    min_epochs (Optional[int], 可选): 训练的最小周期数.默认为 None.

    callbacks (Optional[List[pl.Callback]], 可选): 如果提供,将添加到 Trainer 的回调中.
        默认为 None.

    freeze_backbone (bool, 可选): 如果为 True,将通过关闭梯度来冻结主干网络.
        默认为 False,这意味着预训练的权重在微调期间也会进一步调整.

Returns:
    pl.Trainer: Trainer 对象
"""
        assert self._is_finetune_model, (
            "finetune() can only be called on a finetune model created using" " `TabularModel.create_finetune_model()`"
        )
        seed_everything(self.config.seed)
        if freeze_backbone:
            for param in self.model.backbone.parameters():
                param.requires_grad = False
        return self.train(
            self.model,
            self.datamodule,
            callbacks=callbacks,
            max_epochs=max_epochs,
            min_epochs=min_epochs,
        )

    def find_learning_rate(
        self,
        model: pl.LightningModule,
        datamodule: TabularDatamodule,
        min_lr: float = 1e-8,
        max_lr: float = 1,
        num_training: int = 100,
        mode: str = "exponential",
        early_stop_threshold: Optional[float] = 4.0,
        plot: bool = True,
        callbacks: Optional[List] = None,
    ) -> Tuple[float, DataFrame]:
        """    允许用户进行一系列良好的初始学习率测试,以减少选择合适起始学习率的猜测工作.

Parameters:
    model (pl.LightningModule): 要训练的PyTorch Lightning模型.

    datamodule (TabularDatamodule): 数据模块

    min_lr (Optional[float], optional): 要调查的最小学习率

    max_lr (Optional[float], optional): 要调查的最大学习率

    num_training (Optional[int], optional): 要测试的学习率数量

    mode (Optional[str], optional): 搜索策略,可以是'linear'或'exponential'.如果设置为
        'linear',学习率将通过在每个批次后线性增加来搜索.如果设置为'exponential',将指数增加学习率.

    early_stop_threshold (Optional[float], optional): 停止搜索的阈值.如果在任何时候损失大于
        early_stop_threshold*best_loss,则停止搜索.要禁用,请设置为None.

    plot (bool, optional): 如果为真,将使用matplotlib绘图

    callbacks (Optional[List], optional): 如果提供,将添加到Trainer的回调中.

Returns:
    建议的学习率和学习率查找器的结果
"""
        self._prepare_for_training(model, datamodule, callbacks, max_epochs=None, min_epochs=None)
        train_loader, _ = datamodule.train_dataloader(), datamodule.val_dataloader()
        lr_finder = Tuner(self.trainer).lr_find(
            model=self.model,
            train_dataloaders=train_loader,
            val_dataloaders=None,
            min_lr=min_lr,
            max_lr=max_lr,
            num_training=num_training,
            mode=mode,
            early_stop_threshold=early_stop_threshold,
        )
        if plot:
            fig = lr_finder.plot(suggest=True)
            fig.show()
        new_lr = lr_finder.suggestion()
        # cancelling the model and trainer that was loaded
        self.model = None
        self.trainer = None
        self.datamodule = None
        self.callbacks = None
        return new_lr, DataFrame(lr_finder.results)

    def evaluate(
        self,
        test: Optional[DataFrame] = None,
        test_loader: Optional[torch.utils.data.DataLoader] = None,
        ckpt_path: Optional[Union[str, Path]] = None,
        verbose: bool = True,
    ) -> Union[dict, list]:
        """    使用配置中已设置的损失和指标对数据框进行评估.

Parameters:
    test (可选[DataFrame]): 要评估的数据框.如果未提供,将尝试使用拟合期间提供的测试数据.如果两者均未提供,将返回一个空字典.

    test_loader (可选[torch.utils.data.DataLoader], 可选): 用于评估的数据加载器.如果提供,将使用该数据加载器而不是测试数据框或拟合期间提供的测试数据.默认为None.

    ckpt_path (可选[Union[str, Path]], 可选): 要加载的检查点路径.如果未提供,将尝试使用训练期间的最佳检查点.

    verbose (bool, 可选): 如果为真,将打印结果.默认为True.
Returns:
    最终的测试结果字典.
"""
        assert not (test_loader is None and test is None), (
            "Either `test_loader` or `test` should be provided."
            " If `test_loader` is not provided, `test` should be provided."
        )
        if test_loader is None:
            test_loader = self.datamodule.prepare_inference_dataloader(test)
        result = self.trainer.test(
            model=self.model,
            dataloaders=test_loader,
            ckpt_path=ckpt_path,
            verbose=verbose,
        )
        return result

    def _generate_predictions(
        self,
        model,
        inference_dataloader,
        quantiles,
        n_samples,
        ret_logits,
        progress_bar,
        is_probabilistic,
    ):
        point_predictions = []
        quantile_predictions = []
        logits_predictions = defaultdict(list)
        for batch in progress_bar(inference_dataloader):
            for k, v in batch.items():
                if isinstance(v, list) and (len(v) == 0):
                    continue  # Skipping empty list
                batch[k] = v.to(model.device)
            if is_probabilistic:
                samples, ret_value = model.sample(batch, n_samples, ret_model_output=True)
                y_hat = torch.mean(samples, dim=-1)
                quantile_preds = []
                for q in quantiles:
                    quantile_preds.append(torch.quantile(samples, q=q, dim=-1).unsqueeze(1))
            else:
                y_hat, ret_value = model.predict(batch, ret_model_output=True)
            if ret_logits:
                for k, v in ret_value.items():
                    logits_predictions[k].append(v.detach().cpu())
            point_predictions.append(y_hat.detach().cpu())
            if is_probabilistic:
                quantile_predictions.append(torch.cat(quantile_preds, dim=-1).detach().cpu())
        point_predictions = torch.cat(point_predictions, dim=0)
        if point_predictions.ndim == 1:
            point_predictions = point_predictions.unsqueeze(-1)
        if is_probabilistic:
            quantile_predictions = torch.cat(quantile_predictions, dim=0).unsqueeze(-1)
            if quantile_predictions.ndim == 2:
                quantile_predictions = quantile_predictions.unsqueeze(-1)
        return point_predictions, quantile_predictions, logits_predictions

    def _format_predicitons(
        self,
        test,
        point_predictions,
        quantile_predictions,
        logits_predictions,
        quantiles,
        ret_logits,
        include_input_features,
        is_probabilistic,
    ):
        pred_df = test.copy() if include_input_features else DataFrame(index=test.index)
        if self.config.task == "regression":
            point_predictions = point_predictions.numpy()
            # Probabilistic Models are only implemented for Regression
            if is_probabilistic:
                quantile_predictions = quantile_predictions.numpy()
            for i, target_col in enumerate(self.config.target):
                if self.datamodule.do_target_transform:
                    if self.config.target[i] in pred_df.columns:
                        pred_df[self.config.target[i]] = self.datamodule.target_transforms[i].inverse_transform(
                            pred_df[self.config.target[i]].values.reshape(-1, 1)
                        )
                    pred_df[f"{target_col}_prediction"] = self.datamodule.target_transforms[i].inverse_transform(
                        point_predictions[:, i].reshape(-1, 1)
                    )
                    if is_probabilistic:
                        for j, q in enumerate(quantiles):
                            col_ = f"{target_col}_q{int(q*100)}"
                            pred_df[col_] = self.datamodule.target_transforms[i].inverse_transform(
                                quantile_predictions[:, j, i].reshape(-1, 1)
                            )
                else:
                    pred_df[f"{target_col}_prediction"] = point_predictions[:, i]
                    if is_probabilistic:
                        for j, q in enumerate(quantiles):
                            pred_df[f"{target_col}_q{int(q*100)}"] = quantile_predictions[:, j, i].reshape(-1, 1)

        elif self.config.task == "classification":
            start_index = 0
            for i, target_col in enumerate(self.config.target):
                end_index = start_index + self.datamodule._inferred_config.output_cardinality[i]
                prob_prediction = nn.Softmax(dim=-1)(point_predictions[:, start_index:end_index]).numpy()
                start_index = end_index
                for j, class_ in enumerate(self.datamodule.label_encoder[i].classes_):
                    pred_df[f"{target_col}_{class_}_probability"] = prob_prediction[:, j]
                pred_df[f"{target_col}_prediction"] = self.datamodule.label_encoder[i].inverse_transform(
                    np.argmax(prob_prediction, axis=1)
                )
            warnings.warn(
                "Classification prediction column will be renamed to"
                " `{target_col}_prediction` in the next release to maintain"
                " consistency with regression.",
                DeprecationWarning,
            )
        if ret_logits:
            for k, v in logits_predictions.items():
                v = torch.cat(v, dim=0).numpy()
                if v.ndim == 1:
                    v = v.reshape(-1, 1)
                for i in range(v.shape[-1]):
                    if v.shape[-1] > 1:
                        pred_df[f"{k}_{i}"] = v[:, i]
                    else:
                        pred_df[f"{k}"] = v[:, i]
        return pred_df

    def _predict(
        self,
        test: DataFrame,
        quantiles: Optional[List] = [0.25, 0.5, 0.75],
        n_samples: Optional[int] = 100,
        ret_logits=False,
        include_input_features: bool = False,
        device: Optional[torch.device] = None,
        progress_bar: Optional[str] = None,
    ) -> DataFrame:
        """使用训练好的模型对新数据进行预测,并以数据框形式返回结果.

Parameters:
    test (DataFrame): 包含训练期间定义的特征的新数据框
    quantiles (可选[List]): 对于像混合密度网络这样的概率模型,这指定了除了`central_tendency`之外要提取的不同分位数,并将其添加到数据框中.对于其他模型,此参数被忽略.默认为 [0.25, 0.5, 0.75]
    n_samples (可选[int]): 从后验中抽取的样本数量,用于估计分位数.对于非概率模型,此参数被忽略.默认为 100
    ret_logits (bool): 标志,用于返回原始模型输出/logits（除了骨干特征）以及数据框.默认为 False
    include_input_features (bool): 已弃用: 标志,用于在返回的数据框中包含输入特征.默认为 True
    progress_bar: 选择用于跟踪进度的进度条."rich" 或 "tqdm" 将设置相应的进度条.如果为 None,则不会显示进度条.

Returns:
    DataFrame: 返回一个包含预测结果和特征（如果 `include_input_features=True`）的数据框.如果是分类任务,则返回概率和最终预测结果
"""
        assert all(q <= 1 and q >= 0 for q in quantiles), "Quantiles should be a decimal between 0 and 1"
        model = self.model  # default
        if device is not None:
            if isinstance(device, str):
                device = torch.device(device)
            if self.model.device != device:
                model = self.model.to(device)
        model.eval()
        inference_dataloader = self.datamodule.prepare_inference_dataloader(test)
        is_probabilistic = hasattr(model.hparams, "_probabilistic") and model.hparams._probabilistic

        if progress_bar == "rich":
            from rich.progress import track

            progress_bar = partial(track, description="Generating Predictions...")
        elif progress_bar == "tqdm":
            from tqdm.auto import tqdm

            progress_bar = partial(tqdm, description="Generating Predictions...")
        else:
            progress_bar = lambda it: it  # E731
        point_predictions, quantile_predictions, logits_predictions = self._generate_predictions(
            model,
            inference_dataloader,
            quantiles,
            n_samples,
            ret_logits,
            progress_bar,
            is_probabilistic,
        )
        pred_df = self._format_predicitons(
            test,
            point_predictions,
            quantile_predictions,
            logits_predictions,
            quantiles,
            ret_logits,
            include_input_features,
            is_probabilistic,
        )
        return pred_df

    def predict(
        self,
        test: DataFrame,
        quantiles: Optional[List] = [0.25, 0.5, 0.75],
        n_samples: Optional[int] = 100,
        ret_logits=False,
        include_input_features: bool = False,
        device: Optional[torch.device] = None,
        progress_bar: Optional[str] = None,
        test_time_augmentation: Optional[bool] = False,
        num_tta: Optional[float] = 5,
        alpha_tta: Optional[float] = 0.1,
        aggregate_tta: Optional[str] = "mean",
        tta_seed: Optional[int] = 42,
    ) -> DataFrame:
        """使用训练好的模型对新数据进行预测,并以数据框形式返回结果.

Parameters:
    test (DataFrame): 包含训练期间定义的特征的新数据框

    quantiles (可选[List]): 对于概率模型（如混合密度网络）,这指定了除了`central_tendency`之外要提取的不同分位数并添加到数据框中.
        对于其他模型,此参数被忽略.默认为 [0.25, 0.5, 0.75]

    n_samples (可选[int]): 从后验分布中抽取的样本数量,用于估计分位数.
        对于非概率模型,此参数被忽略.默认为 100

    ret_logits (bool): 标志,用于返回原始模型输出/logits（除了骨干特征）以及数据框.默认为 False

    include_input_features (bool): 已弃用: 标志,用于在返回的数据框中包含输入特征.默认为 True

    progress_bar: 选择用于跟踪进度的进度条."rich" 或 "tqdm" 将设置相应的进度条.如果为 None,则不显示进度条.

    test_time_augmentation (bool): 如果为 True,将使用测试时增强来生成预测.
        该方法与[此处](https://kozodoi.me/blog/20210908/tta-tabular)描述的方法非常相似,
        但我们还在嵌入输入中添加噪声以处理分类特征.                \(x_{aug} = x_{orig} + \alpha * \epsilon\) 其中 \(\epsilon \sim \mathcal{N}(0, 1)\)
        默认为 False
    num_tta (float): 为 TTA 运行的增强次数.默认为 0.0

    alpha_tta (float): 要添加到输入特征的高斯噪声的标准差

    aggregate_tta (Union[str, Callable], 可选): 用于聚合每次增强预测的函数.如果为 str,应为 "mean", "median", "min", 或 "max" 之一
        用于回归.对于分类,前面的选项应用于置信度分数（软投票）,然后转换为最终预测.分类还提供了一个额外的选项
        "hard_voting".
        如果为可调用对象,应为一个函数,该函数接收一个包含 3D 数组（num_samples, num_cv, num_targets）的列表,并返回一个 2D 数组
        的最终概率（num_samples, num_targets）.默认为 "mean".

    tta_seed (int): 用于 TTA 中添加噪声的随机种子.默认为 42.

Returns:
    DataFrame: 返回一个包含预测和特征（如果 `include_input_features=True`）的数据框.
        如果是分类,则返回概率和最终预测
"""
        warnings.warn(
            "`include_input_features` will be deprecated in the next release."
            " Please add index columns to the test dataframe if you want to"
            " retain some features like the key or id",
            DeprecationWarning,
        )
        if test_time_augmentation:
            assert num_tta > 0, "num_tta should be greater than 0"
            assert alpha_tta > 0, "alpha_tta should be greater than 0"
            assert include_input_features is False, "include_input_features cannot be True for TTA."
            if not callable(aggregate_tta):
                assert aggregate_tta in [
                    "mean",
                    "median",
                    "min",
                    "max",
                    "hard_voting",
                ], "aggregate should be one of 'mean', 'median', 'min', 'max', or" " 'hard_voting'"
            if self.config.task == "regression":
                assert aggregate_tta != "hard_voting", "hard_voting is only available for classification"

            torch.manual_seed(tta_seed)

            def add_noise(module, input, output):
                return output + alpha_tta * torch.randn_like(output, memory_format=torch.contiguous_format)

            # Register the hook to the embedding_layer
            handle = self.model.embedding_layer.register_forward_hook(add_noise)
            pred_prob_l = []
            for _ in range(num_tta):
                pred_df = self._predict(
                    test,
                    quantiles,
                    n_samples,
                    ret_logits,
                    include_input_features=False,
                    device=device,
                    progress_bar=progress_bar or "None",
                )
                pred_idx = pred_df.index
                if self.config.task == "classification":
                    pred_prob_l.append(pred_df.values[:, : -len(self.config.target)])
                elif self.config.task == "regression":
                    pred_prob_l.append(pred_df.values)
            pred_df = self._combine_predictions(pred_prob_l, pred_idx, aggregate_tta, None)
            # Remove the hook
            handle.remove()
        else:
            pred_df = self._predict(
                test,
                quantiles,
                n_samples,
                ret_logits,
                include_input_features,
                device,
                progress_bar,
            )
        return pred_df

    def load_best_model(self) -> None:
        """在训练完成后加载最佳模型."""
        if self.trainer.checkpoint_callback is not None:
            if self.verbose:
                logger.info("Loading the best model")
            ckpt_path = self.trainer.checkpoint_callback.best_model_path
            if ckpt_path != "":
                if self.verbose:
                    logger.debug(f"Model Checkpoint: {ckpt_path}")
                ckpt = pl_load(ckpt_path, map_location=lambda storage, loc: storage)
                self.model.load_state_dict(ckpt["state_dict"])
            else:
                logger.warning("No best model available to load. Did you run it more than 1" " epoch?...")
        else:
            logger.warning(
                "No best model available to load. Checkpoint Callback needs to be" " enabled for this to work"
            )

    def save_datamodule(self, dir: str, inference_only: bool = False) -> None:
        """    Saves the datamodule in the specified directory.

Args:
    dir (str): 保存datamodule的目录路径
    inference_only (bool): 如果为True,将仅保存不带数据的推理datamodule.
        这不能用于进一步训练,但可用于推理.默认为False.
"""
        if inference_only:
            dm = self.datamodule.inference_only_copy()
        else:
            dm = self.datamodule

        joblib.dump(dm, os.path.join(dir, "datamodule.sav"))

    def save_config(self, dir: str) -> None:
        """将配置保存到指定目录."""
        with open(os.path.join(dir, "config.yml"), "w") as fp:
            OmegaConf.save(self.config, fp, resolve=True)

    def save_model(self, dir: str, inference_only: bool = False) -> None:
        """    保存模型和检查点在指定目录中.

Parameters:
    dir (str): 保存模型的目录路径
    inference_only (bool): 如果为True,将仅保存数据模块的推理版本
"""
        if os.path.exists(dir) and (os.listdir(dir)):
            logger.warning("Directory is not empty. Overwriting the contents.")
            for f in os.listdir(dir):
                os.remove(os.path.join(dir, f))
        os.makedirs(dir, exist_ok=True)
        self.save_config(dir)
        self.save_datamodule(dir, inference_only=inference_only)
        if hasattr(self.config, "log_target") and self.config.log_target is not None:
            joblib.dump(self.logger, os.path.join(dir, "exp_logger.sav"))
        if hasattr(self, "callbacks"):
            joblib.dump(self.callbacks, os.path.join(dir, "callbacks.sav"))
        self.trainer.save_checkpoint(os.path.join(dir, "model.ckpt"))
        custom_params = {}
        custom_params["custom_loss"] = getattr(self.model, "custom_loss", None)
        custom_params["custom_metrics"] = getattr(self.model, "custom_metrics", None)
        custom_params["custom_metrics_prob_inputs"] = getattr(self.model, "custom_metrics_prob_inputs", None)
        custom_params["custom_optimizer"] = getattr(self.model, "custom_optimizer", None)
        custom_params["custom_optimizer_params"] = getattr(self.model, "custom_optimizer_params", None)
        joblib.dump(custom_params, os.path.join(dir, "custom_params.sav"))
        if self.custom_model:
            joblib.dump(self.model_callable, os.path.join(dir, "custom_model_callable.sav"))

    def save_weights(self, path: Union[str, Path]) -> None:
        """保存模型权重到指定目录.

Parameters:
    path (str): 保存模型的文件路径
"""
        torch.save(self.model.state_dict(), path)

    def load_weights(self, path: Union[str, Path]) -> None:
        """加载指定目录中的模型权重.

Parameters:
    path (str): 要从中加载模型的文件路径

Returns:
    None
"""
        self._load_weights(self.model, path)

    # TODO Need to test ONNX export
    def save_model_for_inference(
        self,
        path: Union[str, Path],
        kind: str = "pytorch",
        onnx_export_params: Dict = {"opset_version": 12},
    ) -> bool:
        """保存模型以供推理.

Parameters:
    path (Union[str, Path]): 保存模型的路径
    kind (str): "pytorch" 或 "onnx"（实验性）
    onnx_export_params (Dict): 传递给 torch.onnx.export 的 ONNX 导出参数

Returns:
    bool: 如果模型成功保存则为 True
"""
        if kind == "pytorch":
            torch.save(self.model, str(path))
            return True
        elif kind == "onnx":
            # Export the model
            onnx_export_params["input_names"] = ["categorical", "continuous"]
            onnx_export_params["output_names"] = onnx_export_params.get("output_names", ["output"])
            onnx_export_params["dynamic_axes"] = {
                onnx_export_params["input_names"][0]: {0: "batch_size"},
                onnx_export_params["output_names"][0]: {0: "batch_size"},
            }
            cat = torch.zeros(
                self.config.batch_size,
                len(self.config.categorical_cols),
                dtype=torch.int,
            )
            cont = torch.randn(
                self.config.batch_size,
                len(self.config.continuous_cols),
                requires_grad=True,
            )
            x = {"continuous": cont, "categorical": cat}
            torch.onnx.export(self.model, x, str(path), **onnx_export_params)
            return True
        else:
            raise ValueError("`kind` must be either pytorch or onnx")

    def summary(self, model=None, max_depth: int = -1) -> None:
        """    打印模型的摘要.

Parameters:
    max_depth (int): 遍历模块并显示在摘要中的最大深度.
        默认为 -1,表示将显示所有模块.
"""
        if model is not None:
            print(summarize(model, max_depth=max_depth))
        elif self.has_model:
            print(summarize(self.model, max_depth=max_depth))
        else:
            rich_print(f"[bold green]{self.__class__.__name__}[/bold green]")
            rich_print("-" * 100)
            rich_print("[bold yellow]Config[/bold yellow]")
            rich_print("-" * 100)
            pprint(self.config.__dict__["_content"])
            rich_print(
                ":triangular_flag:[bold red]Full Model Summary once model has "
                "been initialized or passed in as an argument[/bold red]"
            )

    def __str__(self) -> str:
        return self.summary()

    def feature_importance(self) -> DataFrame:
        """返回模型的特征重要性,格式为pandas DataFrame."""
        return self.model.feature_importance()

    def _prepare_input_for_captum(self, test_dl: torch.utils.data.DataLoader) -> Dict:
        tensor_inp = []
        tensor_tgt = []
        for x in test_dl:
            tensor_inp.append(self.model.embed_input(x))
            tensor_tgt.append(x["target"].squeeze(1))
        tensor_inp = torch.cat(tensor_inp, dim=0)
        tensor_tgt = torch.cat(tensor_tgt, dim=0)
        return tensor_inp, tensor_tgt

    def _prepare_baselines_captum(
        self,
        baselines: Union[float, torch.tensor, str],
        test_dl: torch.utils.data.DataLoader,
        do_baselines: bool,
        is_full_baselines: bool,
    ):
        if do_baselines and baselines is not None and isinstance(baselines, str):
            if baselines.startswith("b|"):
                num_samples = int(baselines.split("|")[1])
                tensor_inp_tr = []
                # tensor_tgt_tr = []
                count = 0
                for x in self.datamodule.train_dataloader():
                    tensor_inp_tr.append(self.model.embed_input(x))
                    # tensor_tgt_tr.append(x["target"])
                    count += x["target"].shape[0]
                    if count >= num_samples:
                        break
                tensor_inp_tr = torch.cat(tensor_inp_tr, dim=0)
                # tensor_tgt_tr = torch.cat(tensor_tgt_tr, dim=0)
                baselines = tensor_inp_tr[:num_samples]
                if is_full_baselines:
                    pass
                else:
                    baselines = baselines.mean(dim=0, keepdim=True)
            else:
                raise ValueError(
                    "Invalid value for `baselines`. Please refer to the documentation" " for more details."
                )
        return baselines

    def _handle_categorical_embeddings_attributions(
        self,
        attributions: torch.tensor,
        is_embedding1d: bool,
        is_embedding2d: bool,
        is_embbeding_dims: bool,
    ):
        # post processing to get attributions for categorical features
        if is_embedding1d and is_embbeding_dims:
            if self.model.hparams.categorical_dim > 0:
                cat_attributions = []
                index_counter = self.model.hparams.continuous_dim
                for _, embed_dim in self.model.hparams.embedding_dims:
                    cat_attributions.append(attributions[:, index_counter : index_counter + embed_dim].sum(dim=1))
                    index_counter += embed_dim
                cat_attributions = torch.stack(cat_attributions, dim=1)
                attributions = torch.cat(
                    [
                        attributions[:, : self.model.hparams.continuous_dim],
                        cat_attributions,
                    ],
                    dim=1,
                )
        elif is_embedding2d:
            attributions = attributions.mean(dim=-1)
        return attributions

    def explain(
        self,
        data: DataFrame,
        method: str = "GradientShap",
        method_args: Optional[Dict] = {},
        baselines: Union[float, torch.tensor, str] = None,
        **kwargs,
    ) -> DataFrame:
        """返回模型的特征归因/解释,以pandas DataFrame的形式呈现.返回的数据框形状为(样本数量, 特征数量)

Parameters:
    data (DataFrame): 需要解释的数据框
    method (str): 用于解释模型的方法.
        应为以下默认值之一:"GradientShap".
        更多详情,请参考 https://captum.ai/api/attribution.html
    method_args (Optional[Dict], optional): 传递给Captum方法初始化的参数.
    baselines (Union[float, torch.tensor, str]): 用于解释的基线.
        如果提供标量,将使用该值作为所有特征的基线.
        如果提供张量,将使用该张量作为所有特征的基线.
        如果提供类似`b|<num_samples>`的字符串,将使用训练数据中的那么多样本.
        不推荐使用整个训练数据作为基线,因为它可能计算量很大.默认情况下,PyTorch Tabular使用训练数据中的10000个样本作为基线.你可以通过传递一个特殊字符串"b|<num_samples>"来配置,其中<num_samples>是要用作基线的样本数量.例如,"b|1000"将使用1000个样本.
        如果为None,将使用captum中的默认设置（这取决于方法）.对于`GradientShap`,它是训练数据.
        默认为None.

    **kwargs: 传递给Captum方法`attribute`函数的额外关键字参数.

Returns:
    DataFrame: 包含特征重要性的数据框
"""
        assert CAPTUM_INSTALLED, "Captum not installed. Please install using `pip install captum` or "
        "install PyTorch Tabular using `pip install pytorch-tabular[extra]`"
        ALLOWED_METHODS = [
            "GradientShap",
            "IntegratedGradients",
            "DeepLift",
            "DeepLiftShap",
            "InputXGradient",
            "FeaturePermutation",
            "FeatureAblation",
            "KernelShap",
        ]
        assert method in ALLOWED_METHODS, f"method should be one of {ALLOWED_METHODS}"
        if isinstance(data, pd.Series):
            data = data.to_frame().T
        if method in ["DeepLiftShap", "KernelShap"]:
            warnings.warn(
                f"{method} is computationally expensive and will take some time. For"
                " faster results, try usingsome other methods like GradientShap,"
                " IntegratedGradients etc."
            )
        if method in ["FeaturePermutation", "FeatureAblation"]:
            assert data.shape[0] > 1, f"{method} only works when the number of samples is greater than 1"
            if len(data) <= 100:
                warnings.warn(
                    f"{method} gives better results when the number of samples is"
                    " large. For better results, try using more samples or some other"
                    " methods like GradientShap which works well on single examples."
                )
        is_full_baselines = method in ["GradientShap", "DeepLiftShap"]
        is_not_supported = self.model._get_name() in [
            "TabNetModel",
            "MDNModel",
            "TabTransformerModel",
        ]
        do_baselines = method not in [
            "Saliency",
            "InputXGradient",
            "FeaturePermutation",
            "LRP",
        ]
        if is_full_baselines and (baselines is None or isinstance(baselines, (float, int))):
            raise ValueError(
                f"baselines cannot be a scalar or None for {method}. Please "
                "provide a tensor or a string like `b|<num_samples>`"
            )
        if is_not_supported:
            raise NotImplementedError(f"Attributions are not implemented for {self.model._get_name()}")

        is_embedding1d = isinstance(self.model.embedding_layer, (Embedding1dLayer, PreEncoded1dLayer))
        is_embedding2d = isinstance(self.model.embedding_layer, Embedding2dLayer)
        # Models like NODE may have no embedding dims (doing leaveOneOut encoding) even if categorical_dim > 0
        is_embbeding_dims = (
            hasattr(self.model.hparams, "embedding_dims") and self.model.hparams.embedding_dims is not None
        )
        if (not is_embedding1d) and (not is_embedding2d):
            raise NotImplementedError(
                "Attributions are not implemented for models with this type of" " embedding layer"
            )
        test_dl = self.datamodule.prepare_inference_dataloader(data)
        self.model.eval()
        # prepare import for Captum
        tensor_inp, tensor_tgt = self._prepare_input_for_captum(test_dl)
        baselines = self._prepare_baselines_captum(baselines, test_dl, do_baselines, is_full_baselines)
        # prepare model for Captum
        try:
            interp_model = _CaptumModel(self.model)
            captum_interp_cls = getattr(captum.attr, method)(interp_model, **method_args)
            if do_baselines:
                attributions = captum_interp_cls.attribute(
                    tensor_inp,
                    baselines=baselines,
                    target=(tensor_tgt if self.config.task == "classification" else None),
                    **kwargs,
                )
            else:
                attributions = captum_interp_cls.attribute(
                    tensor_inp,
                    target=(tensor_tgt if self.config.task == "classification" else None),
                    **kwargs,
                )
            attributions = self._handle_categorical_embeddings_attributions(
                attributions, is_embedding1d, is_embedding2d, is_embbeding_dims
            )
        finally:
            self.model.train()
        assert attributions.shape[1] == self.model.hparams.continuous_dim + self.model.hparams.categorical_dim, (
            "Something went wrong. The number of features in the attributions"
            f" ({attributions.shape[1]}) does not match the number of features in"
            " the model"
            f" ({self.model.hparams.continuous_dim+self.model.hparams.categorical_dim})"
        )
        return pd.DataFrame(
            attributions.detach().cpu().numpy(),
            columns=self.config.continuous_cols + self.config.categorical_cols,
        )

    def _check_cv(self, cv):
        cv = 5 if cv is None else cv
        if isinstance(cv, int):
            if self.config.task == "classification":
                return StratifiedKFold(cv)
            else:
                return KFold(cv)
        elif isinstance(cv, Iterable) and not isinstance(cv, str):
            # An iterable yielding (train, test) splits as arrays of indices.
            return cv
        elif isinstance(cv, BaseCrossValidator):
            return cv
        else:
            raise ValueError("cv must be int, iterable or scikit-learn splitter")

    def _split_kwargs(self, kwargs):
        prep_dl_kwargs = {}
        prep_model_kwargs = {}
        train_kwargs = {}
        # using the defined args in self.prepare_dataloder, self.prepare_model, and self.train
        # to split the kwargs
        for k, v in kwargs.items():
            if k in self.prepare_dataloader.__code__.co_varnames:
                prep_dl_kwargs[k] = v
            elif k in self.prepare_model.__code__.co_varnames:
                prep_model_kwargs[k] = v
            elif k in self.train.__code__.co_varnames:
                train_kwargs[k] = v
            else:
                raise ValueError(f"Invalid keyword argument: {k}")
        return prep_dl_kwargs, prep_model_kwargs, train_kwargs

    def cross_validate(
        self,
        cv: Optional[Union[int, Iterable, BaseCrossValidator]],
        train: DataFrame,
        metric: Optional[Union[str, Callable]] = None,
        return_oof: bool = False,
        groups: Optional[Union[str, np.ndarray]] = None,
        verbose: bool = True,
        reset_datamodule: bool = True,
        handle_oom: bool = True,
        **kwargs,
    ):
        """交叉验证模型.

Parameters:
    cv (可选[Union[int, Iterable, BaseCrossValidator]]): 确定交叉验证的分割策略.
        可能的输入包括:

        - None,使用默认的5折交叉验证（回归问题使用KFold,分类问题使用StratifiedKFold）,
        - 整数,指定(Stratified)KFold中的折数,
        - 一个可迭代对象,生成(train, test)索引数组的分割.
        - 一个scikit-learn的CV分割器.

    train (DataFrame): 带有标签的训练数据

    metric (可选[Union[str, Callable]], 可选): 用于评估的指标.
        如果为None,将使用配置中的第一个指标.如果提供字符串,将使用定义的该指标.如果提供可调用对象,将使用该函数作为指标.我们期望可调用对象的形式为`metric(y_true, y_pred)`.对于分类问题,`y_pred`是一个包含每个类别的概率（<class>_probability）和最终预测（prediction）的数据框.对于回归问题,它是一个包含最终预测（<target>_prediction）的数据框.
        默认为None.

    return_oof (bool, 可选): 如果为True,将返回交叉验证结果以及折叠外的预测.
        默认为False.

    groups (可选[Union[str, np.ndarray]], 可选): 用于分割样本的组标签.如果提供,将作为交叉验证器`split`方法的`groups`参数.
        如果输入为字符串,将使用输入数据框中该名称的列作为组标签.如果输入为类数组对象,将使用该组标签.唯一的约束是组标签的大小应与输入数据框的行数相同.
        默认为None.

    verbose (bool, 可选): 如果为True,将记录结果.
        默认为True.

    reset_datamodule (bool, 可选): 如果为True,将在每次迭代时重置datamodule.
        这将更慢,因为我们将为每个折叠拟合变换.如果为False,我们采用一种近似方法,即一旦变换在第一个折叠上拟合,它们将对所有其他折叠有效.
        默认为True.

    handle_oom (bool, 可选): 如果为True,将优雅地处理内存不足错误.
    **kwargs: 传递给模型`fit`方法的其他关键字参数.

Returns:
    DataFrame: 包含交叉验证结果的数据框
"""
        cv = self._check_cv(cv)
        prep_dl_kwargs, prep_model_kwargs, train_kwargs = self._split_kwargs(kwargs)
        is_callable_metric = False
        if metric is None:
            metric = "test_" + self.config.metrics[0]
        elif isinstance(metric, str):
            metric = metric if metric.startswith("test_") else "test_" + metric
        elif callable(metric):
            is_callable_metric = True

        if isinstance(cv, BaseCrossValidator):
            it = enumerate(cv.split(train, y=train[self.config.target], groups=groups))
        else:
            # when iterable is directly passed
            it = enumerate(cv)
        cv_metrics = []
        datamodule = None
        model = None
        oof_preds = []
        for fold, (train_idx, val_idx) in it:
            if verbose:
                logger.info(f"Running Fold {fold+1}/{cv.get_n_splits()}")
            # train_fold = train.iloc[train_idx]
            # val_fold = train.iloc[val_idx]
            if reset_datamodule:
                datamodule = None
            if datamodule is None:
                # Initialize datamodule and model in the first fold
                # uses train data from this fold to fit all transformers
                datamodule = self.prepare_dataloader(
                    train=train.iloc[train_idx], validation=train.iloc[val_idx], seed=42, **prep_dl_kwargs
                )
                model = self.prepare_model(datamodule, **prep_model_kwargs)
            else:
                # Preprocess the current fold data using the fitted transformers and save in datamodule
                datamodule.train, _ = datamodule.preprocess_data(train.iloc[train_idx], stage="inference")
                datamodule.validation, _ = datamodule.preprocess_data(train.iloc[val_idx], stage="inference")

            # Train the model
            handle_oom = train_kwargs.pop("handle_oom", handle_oom)
            self.train(model, datamodule, handle_oom=handle_oom, **train_kwargs)
            if return_oof or is_callable_metric:
                preds = self.predict(train.iloc[val_idx], include_input_features=False)
                oof_preds.append(preds)
            if is_callable_metric:
                cv_metrics.append(metric(train.iloc[val_idx][self.config.target], preds))
            else:
                result = self.evaluate(train.iloc[val_idx], verbose=False)
                cv_metrics.append(result[0][metric])
            if verbose:
                logger.info(f"Fold {fold+1}/{cv.get_n_splits()} score: {cv_metrics[-1]}")
            self.model.reset_weights()
        return cv_metrics, oof_preds

    def _combine_predictions(
        self,
        pred_prob_l: List[DataFrame],
        pred_idx: Union[pd.Index, List],
        aggregate: Union[str, Callable],
        weights: Optional[List[float]] = None,
    ):
        if aggregate == "mean":
            bagged_pred = np.average(pred_prob_l, axis=0, weights=weights)
        elif aggregate == "median":
            bagged_pred = np.median(pred_prob_l, axis=0)
        elif aggregate == "min":
            bagged_pred = np.min(pred_prob_l, axis=0)
        elif aggregate == "max":
            bagged_pred = np.max(pred_prob_l, axis=0)
        elif aggregate == "hard_voting" and self.config.task == "classification":
            pred_l = [np.argmax(p, axis=1) for p in pred_prob_l]
            final_pred = np.apply_along_axis(
                lambda x: np.argmax(np.bincount(x)),
                axis=0,
                arr=pred_l,
            )
        elif callable(aggregate):
            bagged_pred = aggregate(pred_prob_l)
        if self.config.task == "classification":
            # FIXME need to iterate .label_encoder[x]
            classes = self.datamodule.label_encoder[0].classes_
            if aggregate == "hard_voting":
                pred_df = pd.DataFrame(
                    np.concatenate(pred_prob_l, axis=1),
                    columns=[f"{c}_probability_fold_{i}" for i in range(len(pred_prob_l)) for c in classes],
                    index=pred_idx,
                )
                pred_df["prediction"] = classes[final_pred]
            else:
                final_pred = classes[np.argmax(bagged_pred, axis=1)]
                pred_df = pd.DataFrame(
                    bagged_pred,
                    # FIXME
                    columns=[f"{c}_probability" for c in self.datamodule.label_encoder[0].classes_],
                    index=pred_idx,
                )
                pred_df["prediction"] = final_pred
        elif self.config.task == "regression":
            pred_df = pd.DataFrame(bagged_pred, columns=self.config.target, index=pred_idx)
        else:
            raise NotImplementedError(f"Task {self.config.task} not supported for bagging")
        return pred_df

    def bagging_predict(
        self,
        cv: Optional[Union[int, Iterable, BaseCrossValidator]],
        train: DataFrame,
        test: DataFrame,
        groups: Optional[Union[str, np.ndarray]] = None,
        verbose: bool = True,
        reset_datamodule: bool = True,
        return_raw_predictions: bool = False,
        aggregate: Union[str, Callable] = "mean",
        weights: Optional[List[float]] = None,
        handle_oom: bool = True,
        **kwargs,
    ):
        """    Bagging 预测测试数据.

Parameters:
    cv (可选[Union[int, Iterable, BaseCrossValidator]]): 确定交叉验证的分割策略.
        可能的输入包括:

        - None,使用默认的5折交叉验证（回归使用KFold,分类使用StratifiedKFold）,
        - 整数,指定(Stratified)KFold中的折数,
        - 可迭代对象,生成(训练, 测试)索引对作为数组.
        - scikit-learn的CV分割器.

    train (DataFrame): 带有标签的训练数据

    test (DataFrame): 需要预测的测试数据

    groups (可选[Union[str, np.ndarray]], 可选): 在分割时使用的样本组标签.如果提供,将作为交叉验证器`split`方法的`groups`参数.
        如果输入是字符串,将使用输入数据框中该名称的列作为组标签.如果输入是类数组对象,将使用该组标签.
        唯一的约束是组标签的大小应与输入数据框的行数相同.默认为None.

    verbose (bool, 可选): 如果为True,将记录结果.默认为True.

    reset_datamodule (bool, 可选): 如果为True,将在每次迭代时重置datamodule.
        由于每次折叠都会拟合变换,因此速度会较慢.如果为False,我们采用一种近似方法,即一旦在第一次折叠上拟合了变换,
        它们将对所有其他折叠有效.默认为True.

    return_raw_predictions (bool, 可选): 如果为True,将返回每次折叠的原始预测.默认为False.

    aggregate (Union[str, Callable], 可选): 用于聚合每次折叠预测的函数.如果为字符串,应为"mean"、"median"、"min"或"max"之一,
        用于回归.对于分类,前面的选项应用于置信度分数（软投票）,然后转换为最终预测.分类还提供额外的选项"hard_voting".
        如果为可调用对象,应为接受3D数组列表（样本数, 交叉验证数, 目标数）并返回最终概率2D数组（样本数, 目标数）的函数.默认为"mean".

    weights (可选[List[float]], 可选): 用于聚合每次折叠预测的权重.如果为None,将使用相等的权重.仅在`aggregate`为"mean"时使用.
        默认为None.

    handle_oom (bool, 可选): 如果为True,将优雅地处理内存不足错误.

    **kwargs: 传递给模型`fit`方法的其他关键字参数.

Returns:
    DataFrame: 包含集成预测的数据框.
"""
        if weights is not None:
            assert len(weights) == cv.n_splits, "Number of weights should be equal to the number of folds"
        assert self.config.task in [
            "classification",
            "regression",
        ], "Bagging is only available for classification and regression"
        if not callable(aggregate):
            assert aggregate in ["mean", "median", "min", "max", "hard_voting"], (
                "aggregate should be one of 'mean', 'median', 'min', 'max', or" " 'hard_voting'"
            )
        if self.config.task == "regression":
            assert aggregate != "hard_voting", "hard_voting is only available for classification"
        cv = self._check_cv(cv)
        prep_dl_kwargs, prep_model_kwargs, train_kwargs = self._split_kwargs(kwargs)
        pred_prob_l = []
        datamodule = None
        model = None
        for fold, (train_idx, val_idx) in enumerate(cv.split(train, y=train[self.config.target], groups=groups)):
            if verbose:
                logger.info(f"Running Fold {fold+1}/{cv.get_n_splits()}")
            train_fold = train.iloc[train_idx]
            val_fold = train.iloc[val_idx]
            if reset_datamodule:
                datamodule = None
            if datamodule is None:
                # Initialize datamodule and model in the first fold
                # uses train data from this fold to fit all transformers
                datamodule = self.prepare_dataloader(train=train_fold, validation=val_fold, seed=42, **prep_dl_kwargs)
                model = self.prepare_model(datamodule, **prep_model_kwargs)
            else:
                # Preprocess the current fold data using the fitted transformers and save in datamodule
                datamodule.train, _ = datamodule.preprocess_data(train_fold, stage="inference")
                datamodule.validation, _ = datamodule.preprocess_data(val_fold, stage="inference")

            # Train the model
            handle_oom = train_kwargs.pop("handle_oom", handle_oom)
            self.train(model, datamodule, handle_oom=handle_oom, **train_kwargs)
            fold_preds = self.predict(test, include_input_features=False)
            pred_idx = fold_preds.index
            if self.config.task == "classification":
                pred_prob_l.append(fold_preds.values[:, : -len(self.config.target)])
            elif self.config.task == "regression":
                pred_prob_l.append(fold_preds.values)
            if verbose:
                logger.info(f"Fold {fold+1}/{cv.get_n_splits()} prediction done")
            self.model.reset_weights()
        pred_df = self._combine_predictions(pred_prob_l, pred_idx, aggregate, weights)
        if return_raw_predictions:
            return pred_df, pred_prob_l
        else:
            return pred_df

__init__(config=None, data_config=None, model_config=None, optimizer_config=None, trainer_config=None, experiment_config=None, model_callable=None, model_state_dict_path=None, verbose=True, suppress_lightning_logger=False)

核心模型,负责协调从初始化数据模块、模型、训练器等所有内容.

Parameters:

Name	Type	Description	Default
config	(Optional[Union[DictConfig, str]], 可选)	单个OmegaConf DictConfig对象或包含所有配置参数的yaml文件路径.默认为None.	None
data_config	(Optional[Union[DataConfig, str]], 可选)	DataConfig对象或yaml文件路径.默认为None.	None
model_config	(Optional[Union[ModelConfig, str]], 可选)	ModelConfig的子类或yaml文件路径. 根据配置类型确定运行哪个模型.默认为None.	None
optimizer_config	(Optional[Union[OptimizerConfig, str]], 可选)	OptimizerConfig对象或yaml文件路径.默认为None.	None
trainer_config	(Optional[Union[TrainerConfig, str]], 可选)	TrainerConfig对象或yaml文件路径.默认为None.	None
experiment_config	(Optional[Union[ExperimentConfig, str]], 可选)	ExperimentConfig对象或yaml文件路径. 如果提供,将配置实验跟踪.默认为None.	None
model_callable	(Optional[Callable], 可选)	如果提供,将覆盖从配置加载的模型可调用对象. 通常在提供自定义模型时使用.	None
model_state_dict_path	(Optional[Union[str, Path]], 可选)	如果提供,将在从配置初始化模型后加载状态字典.	None
verbose	bool	控制日志记录的开关.默认为True.	True
suppress_lightning_logger	bool	如果为True,将抑制PyTorch Lightning的默认日志记录.默认为False.	False

Source code in src/pytorch_tabular/tabular_model.py

    def __init__(
        self,
        config: Optional[DictConfig] = None,
        data_config: Optional[Union[DataConfig, str]] = None,
        model_config: Optional[Union[ModelConfig, str]] = None,
        optimizer_config: Optional[Union[OptimizerConfig, str]] = None,
        trainer_config: Optional[Union[TrainerConfig, str]] = None,
        experiment_config: Optional[Union[ExperimentConfig, str]] = None,
        model_callable: Optional[Callable] = None,
        model_state_dict_path: Optional[Union[str, Path]] = None,
        verbose: bool = True,
        suppress_lightning_logger: bool = False,
    ) -> None:
        """核心模型,负责协调从初始化数据模块、模型、训练器等所有内容.

Parameters:
    config (Optional[Union[DictConfig, str]], 可选): 单个OmegaConf DictConfig对象或包含所有配置参数的yaml文件路径.默认为None.

    data_config (Optional[Union[DataConfig, str]], 可选): DataConfig对象或yaml文件路径.默认为None.

    model_config (Optional[Union[ModelConfig, str]], 可选): ModelConfig的子类或yaml文件路径.
        根据配置类型确定运行哪个模型.默认为None.

    optimizer_config (Optional[Union[OptimizerConfig, str]], 可选): OptimizerConfig对象或yaml文件路径.默认为None.

    trainer_config (Optional[Union[TrainerConfig, str]], 可选): TrainerConfig对象或yaml文件路径.默认为None.

    experiment_config (Optional[Union[ExperimentConfig, str]], 可选): ExperimentConfig对象或yaml文件路径.
        如果提供,将配置实验跟踪.默认为None.

    model_callable (Optional[Callable], 可选): 如果提供,将覆盖从配置加载的模型可调用对象.
        通常在提供自定义模型时使用.

    model_state_dict_path (Optional[Union[str, Path]], 可选): 如果提供,将在从配置初始化模型后加载状态字典.

    verbose (bool): 控制日志记录的开关.默认为True.

    suppress_lightning_logger (bool): 如果为True,将抑制PyTorch Lightning的默认日志记录.默认为False.
"""
        super().__init__()
        if suppress_lightning_logger:
            suppress_lightning_logs()
        self.verbose = verbose
        self.exp_manager = ExperimentRunManager()
        if config is None:
            assert any(c is not None for c in (data_config, model_config, optimizer_config, trainer_config)), (
                "If `config` is None, `data_config`, `model_config`,"
                " `trainer_config`, and `optimizer_config` cannot be None"
            )
            data_config = self._read_parse_config(data_config, DataConfig)
            model_config = self._read_parse_config(model_config, ModelConfig)
            trainer_config = self._read_parse_config(trainer_config, TrainerConfig)
            optimizer_config = self._read_parse_config(optimizer_config, OptimizerConfig)
            if model_config.task != "ssl":
                assert data_config.target is not None, (
                    "`target` in data_config should not be None for" f" {model_config.task} task"
                )
            if experiment_config is None:
                if self.verbose:
                    logger.info("Experiment Tracking is turned off")
                self.track_experiment = False
                self.config = OmegaConf.merge(
                    OmegaConf.to_container(data_config),
                    OmegaConf.to_container(model_config),
                    OmegaConf.to_container(trainer_config),
                    OmegaConf.to_container(optimizer_config),
                )
            else:
                experiment_config = self._read_parse_config(experiment_config, ExperimentConfig)
                self.track_experiment = True
                self.config = OmegaConf.merge(
                    OmegaConf.to_container(data_config),
                    OmegaConf.to_container(model_config),
                    OmegaConf.to_container(trainer_config),
                    OmegaConf.to_container(experiment_config),
                    OmegaConf.to_container(optimizer_config),
                )
        else:
            self.config = config
            if hasattr(config, "log_target") and (config.log_target is not None):
                # experiment_config = OmegaConf.structured(experiment_config)
                self.track_experiment = True
            else:
                if self.verbose:
                    logger.info("Experiment Tracking is turned off")
                self.track_experiment = False

        self.run_name, self.uid = self._get_run_name_uid()
        if self.track_experiment:
            self._setup_experiment_tracking()
        else:
            self.logger = None

        self.exp_manager = ExperimentRunManager()
        if model_callable is None:
            self.model_callable = getattr_nested(self.config._module_src, self.config._model_name)
            self.custom_model = False
        else:
            self.model_callable = model_callable
            self.custom_model = True
        self.model_state_dict_path = model_state_dict_path
        self._is_config_updated_with_data = False
        self._run_validation()
        self._is_fitted = False

bagging_predict(cv, train, test, groups=None, verbose=True, reset_datamodule=True, return_raw_predictions=False, aggregate='mean', weights=None, handle_oom=True, **kwargs)

Bagging 预测测试数据.

Parameters:

Name	Type	Description	Default
cv	可选[Union[int, Iterable, BaseCrossValidator]]	确定交叉验证的分割策略. 可能的输入包括: None,使用默认的5折交叉验证（回归使用KFold,分类使用StratifiedKFold）, 整数,指定(Stratified)KFold中的折数, 可迭代对象,生成(训练, 测试)索引对作为数组. scikit-learn的CV分割器.	required
train	DataFrame	带有标签的训练数据	required
test	DataFrame	需要预测的测试数据	required
groups	(可选[Union[str, ndarray]], 可选)	在分割时使用的样本组标签.如果提供,将作为交叉验证器split方法的groups参数. 如果输入是字符串,将使用输入数据框中该名称的列作为组标签.如果输入是类数组对象,将使用该组标签. 唯一的约束是组标签的大小应与输入数据框的行数相同.默认为None.	None
verbose	(bool, 可选)	如果为True,将记录结果.默认为True.	True
reset_datamodule	(bool, 可选)	如果为True,将在每次迭代时重置datamodule. 由于每次折叠都会拟合变换,因此速度会较慢.如果为False,我们采用一种近似方法,即一旦在第一次折叠上拟合了变换, 它们将对所有其他折叠有效.默认为True.	True
return_raw_predictions	(bool, 可选)	如果为True,将返回每次折叠的原始预测.默认为False.	False
aggregate	(Union[str, Callable], 可选)	用于聚合每次折叠预测的函数.如果为字符串,应为"mean"、"median"、"min"或"max"之一, 用于回归.对于分类,前面的选项应用于置信度分数（软投票）,然后转换为最终预测.分类还提供额外的选项"hard_voting". 如果为可调用对象,应为接受3D数组列表（样本数, 交叉验证数, 目标数）并返回最终概率2D数组（样本数, 目标数）的函数.默认为"mean".	'mean'
weights	(可选[List[float]], 可选)	用于聚合每次折叠预测的权重.如果为None,将使用相等的权重.仅在aggregate为"mean"时使用. 默认为None.	None
handle_oom	(bool, 可选)	如果为True,将优雅地处理内存不足错误.	True
**kwargs		传递给模型fit方法的其他关键字参数.	{}

Returns:

Name	Type	Description
DataFrame		包含集成预测的数据框.

Source code in src/pytorch_tabular/tabular_model.py

    def bagging_predict(
        self,
        cv: Optional[Union[int, Iterable, BaseCrossValidator]],
        train: DataFrame,
        test: DataFrame,
        groups: Optional[Union[str, np.ndarray]] = None,
        verbose: bool = True,
        reset_datamodule: bool = True,
        return_raw_predictions: bool = False,
        aggregate: Union[str, Callable] = "mean",
        weights: Optional[List[float]] = None,
        handle_oom: bool = True,
        **kwargs,
    ):
        """    Bagging 预测测试数据.

Parameters:
    cv (可选[Union[int, Iterable, BaseCrossValidator]]): 确定交叉验证的分割策略.
        可能的输入包括:

        - None,使用默认的5折交叉验证（回归使用KFold,分类使用StratifiedKFold）,
        - 整数,指定(Stratified)KFold中的折数,
        - 可迭代对象,生成(训练, 测试)索引对作为数组.
        - scikit-learn的CV分割器.

    train (DataFrame): 带有标签的训练数据

    test (DataFrame): 需要预测的测试数据

    groups (可选[Union[str, np.ndarray]], 可选): 在分割时使用的样本组标签.如果提供,将作为交叉验证器`split`方法的`groups`参数.
        如果输入是字符串,将使用输入数据框中该名称的列作为组标签.如果输入是类数组对象,将使用该组标签.
        唯一的约束是组标签的大小应与输入数据框的行数相同.默认为None.

    verbose (bool, 可选): 如果为True,将记录结果.默认为True.

    reset_datamodule (bool, 可选): 如果为True,将在每次迭代时重置datamodule.
        由于每次折叠都会拟合变换,因此速度会较慢.如果为False,我们采用一种近似方法,即一旦在第一次折叠上拟合了变换,
        它们将对所有其他折叠有效.默认为True.

    return_raw_predictions (bool, 可选): 如果为True,将返回每次折叠的原始预测.默认为False.

    aggregate (Union[str, Callable], 可选): 用于聚合每次折叠预测的函数.如果为字符串,应为"mean"、"median"、"min"或"max"之一,
        用于回归.对于分类,前面的选项应用于置信度分数（软投票）,然后转换为最终预测.分类还提供额外的选项"hard_voting".
        如果为可调用对象,应为接受3D数组列表（样本数, 交叉验证数, 目标数）并返回最终概率2D数组（样本数, 目标数）的函数.默认为"mean".

    weights (可选[List[float]], 可选): 用于聚合每次折叠预测的权重.如果为None,将使用相等的权重.仅在`aggregate`为"mean"时使用.
        默认为None.

    handle_oom (bool, 可选): 如果为True,将优雅地处理内存不足错误.

    **kwargs: 传递给模型`fit`方法的其他关键字参数.

Returns:
    DataFrame: 包含集成预测的数据框.
"""
        if weights is not None:
            assert len(weights) == cv.n_splits, "Number of weights should be equal to the number of folds"
        assert self.config.task in [
            "classification",
            "regression",
        ], "Bagging is only available for classification and regression"
        if not callable(aggregate):
            assert aggregate in ["mean", "median", "min", "max", "hard_voting"], (
                "aggregate should be one of 'mean', 'median', 'min', 'max', or" " 'hard_voting'"
            )
        if self.config.task == "regression":
            assert aggregate != "hard_voting", "hard_voting is only available for classification"
        cv = self._check_cv(cv)
        prep_dl_kwargs, prep_model_kwargs, train_kwargs = self._split_kwargs(kwargs)
        pred_prob_l = []
        datamodule = None
        model = None
        for fold, (train_idx, val_idx) in enumerate(cv.split(train, y=train[self.config.target], groups=groups)):
            if verbose:
                logger.info(f"Running Fold {fold+1}/{cv.get_n_splits()}")
            train_fold = train.iloc[train_idx]
            val_fold = train.iloc[val_idx]
            if reset_datamodule:
                datamodule = None
            if datamodule is None:
                # Initialize datamodule and model in the first fold
                # uses train data from this fold to fit all transformers
                datamodule = self.prepare_dataloader(train=train_fold, validation=val_fold, seed=42, **prep_dl_kwargs)
                model = self.prepare_model(datamodule, **prep_model_kwargs)
            else:
                # Preprocess the current fold data using the fitted transformers and save in datamodule
                datamodule.train, _ = datamodule.preprocess_data(train_fold, stage="inference")
                datamodule.validation, _ = datamodule.preprocess_data(val_fold, stage="inference")

            # Train the model
            handle_oom = train_kwargs.pop("handle_oom", handle_oom)
            self.train(model, datamodule, handle_oom=handle_oom, **train_kwargs)
            fold_preds = self.predict(test, include_input_features=False)
            pred_idx = fold_preds.index
            if self.config.task == "classification":
                pred_prob_l.append(fold_preds.values[:, : -len(self.config.target)])
            elif self.config.task == "regression":
                pred_prob_l.append(fold_preds.values)
            if verbose:
                logger.info(f"Fold {fold+1}/{cv.get_n_splits()} prediction done")
            self.model.reset_weights()
        pred_df = self._combine_predictions(pred_prob_l, pred_idx, aggregate, weights)
        if return_raw_predictions:
            return pred_df, pred_prob_l
        else:
            return pred_df

create_finetune_model(task, head, head_config, train, validation=None, train_sampler=None, target_transform=None, target=None, optimizer_config=None, trainer_config=None, experiment_config=None, loss=None, metrics=None, metrics_prob_input=None, metrics_params=None, optimizer=None, optimizer_params=None, learning_rate=None, target_range=None, seed=42)

创建一个新的TabularModel模型,使用预训练权重以及新的任务和头部.

Parameters:

Name	Type	Description	Default
task	str	要执行的任务.可以是 "regression" 或 "classification" 之一.	required
head	str	用于模型的头部.应为 pytorch_tabular.models.common.heads 中定义的头部之一.默认为 LinearHead.可选值为: [None,LinearHead,MixtureDensityHead].	required
head_config	Dict	定义头部的配置字典.如果留空,将初始化为默认的线性头部.	required
train	DataFrame	带有标签的训练数据.	required
validation	Optional[DataFrame]	带有标签的验证数据.默认为 None.	None
train_sampler	Optional[Sampler]	如果提供,将用作训练的批次采样器.默认为 None.	None
target_transform	Optional[Union[TransformerMixin, Tuple]]	如果提供,将在训练前用于转换目标,并在预测后进行逆转换.	None
target	Optional[str]	如果未在初始预训练阶段提供,则为目标列名称.默认为 None.	None
optimizer_config	Optional[OptimizerConfig]	如果提供,将重新定义微调阶段的优化器.默认为 None.	None
trainer_config	Optional[TrainerConfig]	如果提供,将重新定义微调阶段的训练器.默认为 None.	None
experiment_config	Optional[ExperimentConfig]	如果提供,将重新定义微调阶段的实验配置.默认为 None.	None
loss	Optional[Module]	如果提供,将用作微调阶段的损失函数.默认情况下,回归任务为 MSELoss,分类任务为 CrossEntropyLoss.	None
metrics	Optional[List[Callable]]	用于微调阶段的指标列表（可以是可调用对象或字符串）.如果是字符串,应为 torchmetrics.functional 中实现的功能性指标之一.默认为 None.	None
metrics_prob_input	Optional[List[bool]]	分类指标的强制参数. 这定义了指标函数的输入是概率还是类别.长度应与指标数量相同.默认为 None.	None
metrics_params	Optional[Dict]	与指标顺序相同的指标参数. 例如,多类别的 f1_score 需要参数 average 来完全定义指标.默认为 None.	None
optimizer	Optional[Optimizer]	自定义优化器,是标准 PyTorch 优化器的替代品.如果提供,将忽略 OptimizerConfig.默认为 None.	None
optimizer_params	Dict	优化器的参数.默认为 {}.	None
learning_rate	Optional[float]	要使用的学习率.默认为 1e-3.	None
target_range	Optional[Tuple[float, float]]	回归任务的目标范围.分类任务中忽略.默认为 None.	None
seed	Optional[int]	随机种子,用于可重复性.默认为 42.	42

Returns:

Name	Type	Description
TabularModel	TabularModel	用于微调的新 TabularModel 模型

Source code in src/pytorch_tabular/tabular_model.py

    def create_finetune_model(
        self,
        task: str,
        head: str,
        head_config: Dict,
        train: DataFrame,
        validation: Optional[DataFrame] = None,
        train_sampler: Optional[torch.utils.data.Sampler] = None,
        target_transform: Optional[Union[TransformerMixin, Tuple]] = None,
        target: Optional[str] = None,
        optimizer_config: Optional[OptimizerConfig] = None,
        trainer_config: Optional[TrainerConfig] = None,
        experiment_config: Optional[ExperimentConfig] = None,
        loss: Optional[torch.nn.Module] = None,
        metrics: Optional[List[Union[Callable, str]]] = None,
        metrics_prob_input: Optional[List[bool]] = None,
        metrics_params: Optional[Dict] = None,
        optimizer: Optional[torch.optim.Optimizer] = None,
        optimizer_params: Dict = None,
        learning_rate: Optional[float] = None,
        target_range: Optional[Tuple[float, float]] = None,
        seed: Optional[int] = 42,
    ):
        """创建一个新的TabularModel模型,使用预训练权重以及新的任务和头部.

Parameters:
    task (str): 要执行的任务.可以是 "regression" 或 "classification" 之一.

    head (str): 用于模型的头部.应为 `pytorch_tabular.models.common.heads` 中定义的头部之一.默认为 LinearHead.可选值为:
        [`None`,`LinearHead`,`MixtureDensityHead`].

    head_config (Dict): 定义头部的配置字典.如果留空,将初始化为默认的线性头部.

    train (DataFrame): 带有标签的训练数据.

    validation (Optional[DataFrame], optional): 带有标签的验证数据.默认为 None.

    train_sampler (Optional[torch.utils.data.Sampler], optional): 如果提供,将用作训练的批次采样器.默认为 None.

    target_transform (Optional[Union[TransformerMixin, Tuple]], optional): 如果提供,将在训练前用于转换目标,并在预测后进行逆转换.

    target (Optional[str], optional): 如果未在初始预训练阶段提供,则为目标列名称.默认为 None.

    optimizer_config (Optional[OptimizerConfig], optional):
        如果提供,将重新定义微调阶段的优化器.默认为 None.

    trainer_config (Optional[TrainerConfig], optional):
        如果提供,将重新定义微调阶段的训练器.默认为 None.

    experiment_config (Optional[ExperimentConfig], optional):
        如果提供,将重新定义微调阶段的实验配置.默认为 None.

    loss (Optional[torch.nn.Module], optional):
        如果提供,将用作微调阶段的损失函数.默认情况下,回归任务为 MSELoss,分类任务为 CrossEntropyLoss.

    metrics (Optional[List[Callable]], optional): 用于微调阶段的指标列表（可以是可调用对象或字符串）.如果是字符串,应为 ``torchmetrics.functional`` 中实现的功能性指标之一.默认为 None.

    metrics_prob_input (Optional[List[bool]], optional): 分类指标的强制参数.
        这定义了指标函数的输入是概率还是类别.长度应与指标数量相同.默认为 None.

    metrics_params (Optional[Dict], optional): 与指标顺序相同的指标参数.
        例如,多类别的 f1_score 需要参数 `average` 来完全定义指标.默认为 None.

    optimizer (Optional[torch.optim.Optimizer], optional):
        自定义优化器,是标准 PyTorch 优化器的替代品.如果提供,将忽略 OptimizerConfig.默认为 None.

    optimizer_params (Dict, optional): 优化器的参数.默认为 {}.

    learning_rate (Optional[float], optional): 要使用的学习率.默认为 1e-3.

    target_range (Optional[Tuple[float, float]], optional): 回归任务的目标范围.分类任务中忽略.默认为 None.

    seed (Optional[int], optional): 随机种子,用于可重复性.默认为 42.

Returns:
    TabularModel (TabularModel): 用于微调的新 TabularModel 模型
"""
        config = self.config
        optimizer_params = optimizer_params or {}
        if target is None:
            assert (
                hasattr(config, "target") and config.target is not None
            ), "`target` cannot be None if it was not set in the initial `DataConfig`"
        else:
            assert isinstance(target, list), "`target` should be a list of strings"
            config.target = target
        config.task = task
        # Add code to update configs with newly provided ones
        if optimizer_config is not None:
            for key, value in optimizer_config.__dict__.items():
                config[key] = value
            if len(optimizer_params) > 0:
                config.optimizer_params = optimizer_params
            else:
                config.optimizer_params = {}
        if trainer_config is not None:
            for key, value in trainer_config.__dict__.items():
                config[key] = value
        if experiment_config is not None:
            for key, value in experiment_config.__dict__.items():
                config[key] = value
        else:
            if self.track_experiment:
                # Renaming the experiment run so that a different log is created for finetuning
                if self.verbose:
                    logger.info("Renaming the experiment run for finetuning as" f" {config['run_name'] + '_finetuned'}")
                config["run_name"] = config["run_name"] + "_finetuned"

        datamodule = self.datamodule.copy(
            train=train,
            validation=validation,
            target_transform=target_transform,
            train_sampler=train_sampler,
            seed=seed,
            config_override={"target": target} if target is not None else {},
        )
        model_callable = _GenericModel
        inferred_config = OmegaConf.structured(datamodule._inferred_config)
        # Adding dummy attributes for compatibility. Not used because custom metrics are provided
        if not hasattr(config, "metrics"):
            config.metrics = "dummy"
        if not hasattr(config, "metrics_params"):
            config.metrics_params = {}
        if not hasattr(config, "metrics_prob_input"):
            config.metrics_prob_input = metrics_prob_input or [False]
        if metrics is not None:
            assert len(metrics) == len(metrics_params), "Number of metrics and metrics_params should be same"
            assert len(metrics) == len(metrics_prob_input), "Number of metrics and metrics_prob_input should be same"
            metrics = [getattr(torchmetrics.functional, m) if isinstance(m, str) else m for m in metrics]
        if task == "regression":
            loss = loss or torch.nn.MSELoss()
            if metrics is None:
                metrics = [torchmetrics.functional.mean_squared_error]
                metrics_params = [{}]
        elif task == "classification":
            loss = loss or torch.nn.CrossEntropyLoss()
            if metrics is None:
                metrics = [torchmetrics.functional.accuracy]
                metrics_params = [
                    {
                        "task": "multiclass",
                        "num_classes": inferred_config.output_dim,
                        "top_k": 1,
                    }
                ]
                metrics_prob_input = [False]
            else:
                for i, mp in enumerate(metrics_params):
                    # For classification task, output_dim == number of classses
                    metrics_params[i]["task"] = mp.get("task", "multiclass")
                    metrics_params[i]["num_classes"] = mp.get("num_classes", inferred_config.output_dim)
                    metrics_params[i]["top_k"] = mp.get("top_k", 1)
        else:
            raise ValueError(f"Task {task} not supported")
        # Forming partial callables using metrics and metric params
        metrics = [partial(m, **mp) for m, mp in zip(metrics, metrics_params)]
        self.model.mode = "finetune"
        if learning_rate is not None:
            config.learning_rate = learning_rate
        config.target_range = target_range
        model_args = {
            "backbone": self.model,
            "head": head,
            "head_config": head_config,
            "config": config,
            "inferred_config": inferred_config,
            "custom_loss": loss,
            "custom_metrics": metrics,
            "custom_metrics_prob_inputs": metrics_prob_input,
            "custom_optimizer": optimizer,
            "custom_optimizer_params": optimizer_params,
        }
        # Initializing with default metrics, losses, and optimizers. Will revert once initialized
        model = model_callable(
            **model_args,
        )
        tabular_model = TabularModel(config=config, verbose=self.verbose)
        tabular_model.model = model
        tabular_model.datamodule = datamodule
        # Setting a flag to identify this as a fine-tune model
        tabular_model._is_finetune_model = True
        return tabular_model

cross_validate(cv, train, metric=None, return_oof=False, groups=None, verbose=True, reset_datamodule=True, handle_oom=True, **kwargs)

交叉验证模型.

Parameters:

Name	Type	Description	Default
cv	可选[Union[int, Iterable, BaseCrossValidator]]	确定交叉验证的分割策略. 可能的输入包括: None,使用默认的5折交叉验证（回归问题使用KFold,分类问题使用StratifiedKFold）, 整数,指定(Stratified)KFold中的折数, 一个可迭代对象,生成(train, test)索引数组的分割. 一个scikit-learn的CV分割器.	required
train	DataFrame	带有标签的训练数据	required
metric	(可选[Union[str, Callable]], 可选)	用于评估的指标. 如果为None,将使用配置中的第一个指标.如果提供字符串,将使用定义的该指标.如果提供可调用对象,将使用该函数作为指标.我们期望可调用对象的形式为metric(y_true, y_pred).对于分类问题,y_pred是一个包含每个类别的概率（_probability）和最终预测（prediction）的数据框.对于回归问题,它是一个包含最终预测（_prediction）的数据框. 默认为None.	None
return_oof	(bool, 可选)	如果为True,将返回交叉验证结果以及折叠外的预测. 默认为False.	False
groups	(可选[Union[str, ndarray]], 可选)	用于分割样本的组标签.如果提供,将作为交叉验证器split方法的groups参数. 如果输入为字符串,将使用输入数据框中该名称的列作为组标签.如果输入为类数组对象,将使用该组标签.唯一的约束是组标签的大小应与输入数据框的行数相同. 默认为None.	None
verbose	(bool, 可选)	如果为True,将记录结果. 默认为True.	True
reset_datamodule	(bool, 可选)	如果为True,将在每次迭代时重置datamodule. 这将更慢,因为我们将为每个折叠拟合变换.如果为False,我们采用一种近似方法,即一旦变换在第一个折叠上拟合,它们将对所有其他折叠有效. 默认为True.	True
handle_oom	(bool, 可选)	如果为True,将优雅地处理内存不足错误.	True
**kwargs		传递给模型fit方法的其他关键字参数.	{}

Returns:

Name	Type	Description
DataFrame		包含交叉验证结果的数据框

Source code in src/pytorch_tabular/tabular_model.py

    def cross_validate(
        self,
        cv: Optional[Union[int, Iterable, BaseCrossValidator]],
        train: DataFrame,
        metric: Optional[Union[str, Callable]] = None,
        return_oof: bool = False,
        groups: Optional[Union[str, np.ndarray]] = None,
        verbose: bool = True,
        reset_datamodule: bool = True,
        handle_oom: bool = True,
        **kwargs,
    ):
        """交叉验证模型.

Parameters:
    cv (可选[Union[int, Iterable, BaseCrossValidator]]): 确定交叉验证的分割策略.
        可能的输入包括:

        - None,使用默认的5折交叉验证（回归问题使用KFold,分类问题使用StratifiedKFold）,
        - 整数,指定(Stratified)KFold中的折数,
        - 一个可迭代对象,生成(train, test)索引数组的分割.
        - 一个scikit-learn的CV分割器.

    train (DataFrame): 带有标签的训练数据

    metric (可选[Union[str, Callable]], 可选): 用于评估的指标.
        如果为None,将使用配置中的第一个指标.如果提供字符串,将使用定义的该指标.如果提供可调用对象,将使用该函数作为指标.我们期望可调用对象的形式为`metric(y_true, y_pred)`.对于分类问题,`y_pred`是一个包含每个类别的概率（<class>_probability）和最终预测（prediction）的数据框.对于回归问题,它是一个包含最终预测（<target>_prediction）的数据框.
        默认为None.

    return_oof (bool, 可选): 如果为True,将返回交叉验证结果以及折叠外的预测.
        默认为False.

    groups (可选[Union[str, np.ndarray]], 可选): 用于分割样本的组标签.如果提供,将作为交叉验证器`split`方法的`groups`参数.
        如果输入为字符串,将使用输入数据框中该名称的列作为组标签.如果输入为类数组对象,将使用该组标签.唯一的约束是组标签的大小应与输入数据框的行数相同.
        默认为None.

    verbose (bool, 可选): 如果为True,将记录结果.
        默认为True.

    reset_datamodule (bool, 可选): 如果为True,将在每次迭代时重置datamodule.
        这将更慢,因为我们将为每个折叠拟合变换.如果为False,我们采用一种近似方法,即一旦变换在第一个折叠上拟合,它们将对所有其他折叠有效.
        默认为True.

    handle_oom (bool, 可选): 如果为True,将优雅地处理内存不足错误.
    **kwargs: 传递给模型`fit`方法的其他关键字参数.

Returns:
    DataFrame: 包含交叉验证结果的数据框
"""
        cv = self._check_cv(cv)
        prep_dl_kwargs, prep_model_kwargs, train_kwargs = self._split_kwargs(kwargs)
        is_callable_metric = False
        if metric is None:
            metric = "test_" + self.config.metrics[0]
        elif isinstance(metric, str):
            metric = metric if metric.startswith("test_") else "test_" + metric
        elif callable(metric):
            is_callable_metric = True

        if isinstance(cv, BaseCrossValidator):
            it = enumerate(cv.split(train, y=train[self.config.target], groups=groups))
        else:
            # when iterable is directly passed
            it = enumerate(cv)
        cv_metrics = []
        datamodule = None
        model = None
        oof_preds = []
        for fold, (train_idx, val_idx) in it:
            if verbose:
                logger.info(f"Running Fold {fold+1}/{cv.get_n_splits()}")
            # train_fold = train.iloc[train_idx]
            # val_fold = train.iloc[val_idx]
            if reset_datamodule:
                datamodule = None
            if datamodule is None:
                # Initialize datamodule and model in the first fold
                # uses train data from this fold to fit all transformers
                datamodule = self.prepare_dataloader(
                    train=train.iloc[train_idx], validation=train.iloc[val_idx], seed=42, **prep_dl_kwargs
                )
                model = self.prepare_model(datamodule, **prep_model_kwargs)
            else:
                # Preprocess the current fold data using the fitted transformers and save in datamodule
                datamodule.train, _ = datamodule.preprocess_data(train.iloc[train_idx], stage="inference")
                datamodule.validation, _ = datamodule.preprocess_data(train.iloc[val_idx], stage="inference")

            # Train the model
            handle_oom = train_kwargs.pop("handle_oom", handle_oom)
            self.train(model, datamodule, handle_oom=handle_oom, **train_kwargs)
            if return_oof or is_callable_metric:
                preds = self.predict(train.iloc[val_idx], include_input_features=False)
                oof_preds.append(preds)
            if is_callable_metric:
                cv_metrics.append(metric(train.iloc[val_idx][self.config.target], preds))
            else:
                result = self.evaluate(train.iloc[val_idx], verbose=False)
                cv_metrics.append(result[0][metric])
            if verbose:
                logger.info(f"Fold {fold+1}/{cv.get_n_splits()} score: {cv_metrics[-1]}")
            self.model.reset_weights()
        return cv_metrics, oof_preds

evaluate(test=None, test_loader=None, ckpt_path=None, verbose=True)

使用配置中已设置的损失和指标对数据框进行评估.

Parameters:

Name	Type	Description	Default
test	可选[DataFrame]	要评估的数据框.如果未提供,将尝试使用拟合期间提供的测试数据.如果两者均未提供,将返回一个空字典.	None
test_loader	(可选[DataLoader], 可选)	用于评估的数据加载器.如果提供,将使用该数据加载器而不是测试数据框或拟合期间提供的测试数据.默认为None.	None
ckpt_path	(可选[Union[str, Path]], 可选)	要加载的检查点路径.如果未提供,将尝试使用训练期间的最佳检查点.	None
verbose	(bool, 可选)	如果为真,将打印结果.默认为True.	True

Returns: 最终的测试结果字典.

Source code in src/pytorch_tabular/tabular_model.py

    def evaluate(
        self,
        test: Optional[DataFrame] = None,
        test_loader: Optional[torch.utils.data.DataLoader] = None,
        ckpt_path: Optional[Union[str, Path]] = None,
        verbose: bool = True,
    ) -> Union[dict, list]:
        """    使用配置中已设置的损失和指标对数据框进行评估.

Parameters:
    test (可选[DataFrame]): 要评估的数据框.如果未提供,将尝试使用拟合期间提供的测试数据.如果两者均未提供,将返回一个空字典.

    test_loader (可选[torch.utils.data.DataLoader], 可选): 用于评估的数据加载器.如果提供,将使用该数据加载器而不是测试数据框或拟合期间提供的测试数据.默认为None.

    ckpt_path (可选[Union[str, Path]], 可选): 要加载的检查点路径.如果未提供,将尝试使用训练期间的最佳检查点.

    verbose (bool, 可选): 如果为真,将打印结果.默认为True.
Returns:
    最终的测试结果字典.
"""
        assert not (test_loader is None and test is None), (
            "Either `test_loader` or `test` should be provided."
            " If `test_loader` is not provided, `test` should be provided."
        )
        if test_loader is None:
            test_loader = self.datamodule.prepare_inference_dataloader(test)
        result = self.trainer.test(
            model=self.model,
            dataloaders=test_loader,
            ckpt_path=ckpt_path,
            verbose=verbose,
        )
        return result

explain(data, method='GradientShap', method_args={}, baselines=None, **kwargs)

返回模型的特征归因/解释,以pandas DataFrame的形式呈现.返回的数据框形状为(样本数量, 特征数量)

Parameters:

Name	Type	Description	Default
data	DataFrame	需要解释的数据框	required
method	str	用于解释模型的方法. 应为以下默认值之一:"GradientShap". 更多详情,请参考 https://captum.ai/api/attribution.html	'GradientShap'
method_args	Optional[Dict]	传递给Captum方法初始化的参数.	{}
baselines	Union[float, tensor, str]	用于解释的基线. 如果提供标量,将使用该值作为所有特征的基线. 如果提供张量,将使用该张量作为所有特征的基线. 如果提供类似b|<num_samples>的字符串,将使用训练数据中的那么多样本. 不推荐使用整个训练数据作为基线,因为它可能计算量很大.默认情况下,PyTorch Tabular使用训练数据中的10000个样本作为基线.你可以通过传递一个特殊字符串"b|"来配置,其中是要用作基线的样本数量.例如,"b|1000"将使用1000个样本. 如果为None,将使用captum中的默认设置（这取决于方法）.对于GradientShap,它是训练数据. 默认为None.	None
**kwargs		传递给Captum方法attribute函数的额外关键字参数.	{}

Returns:

Name	Type	Description
DataFrame	DataFrame	包含特征重要性的数据框

Source code in src/pytorch_tabular/tabular_model.py

    def explain(
        self,
        data: DataFrame,
        method: str = "GradientShap",
        method_args: Optional[Dict] = {},
        baselines: Union[float, torch.tensor, str] = None,
        **kwargs,
    ) -> DataFrame:
        """返回模型的特征归因/解释,以pandas DataFrame的形式呈现.返回的数据框形状为(样本数量, 特征数量)

Parameters:
    data (DataFrame): 需要解释的数据框
    method (str): 用于解释模型的方法.
        应为以下默认值之一:"GradientShap".
        更多详情,请参考 https://captum.ai/api/attribution.html
    method_args (Optional[Dict], optional): 传递给Captum方法初始化的参数.
    baselines (Union[float, torch.tensor, str]): 用于解释的基线.
        如果提供标量,将使用该值作为所有特征的基线.
        如果提供张量,将使用该张量作为所有特征的基线.
        如果提供类似`b|<num_samples>`的字符串,将使用训练数据中的那么多样本.
        不推荐使用整个训练数据作为基线,因为它可能计算量很大.默认情况下,PyTorch Tabular使用训练数据中的10000个样本作为基线.你可以通过传递一个特殊字符串"b|<num_samples>"来配置,其中<num_samples>是要用作基线的样本数量.例如,"b|1000"将使用1000个样本.
        如果为None,将使用captum中的默认设置（这取决于方法）.对于`GradientShap`,它是训练数据.
        默认为None.

    **kwargs: 传递给Captum方法`attribute`函数的额外关键字参数.

Returns:
    DataFrame: 包含特征重要性的数据框
"""
        assert CAPTUM_INSTALLED, "Captum not installed. Please install using `pip install captum` or "
        "install PyTorch Tabular using `pip install pytorch-tabular[extra]`"
        ALLOWED_METHODS = [
            "GradientShap",
            "IntegratedGradients",
            "DeepLift",
            "DeepLiftShap",
            "InputXGradient",
            "FeaturePermutation",
            "FeatureAblation",
            "KernelShap",
        ]
        assert method in ALLOWED_METHODS, f"method should be one of {ALLOWED_METHODS}"
        if isinstance(data, pd.Series):
            data = data.to_frame().T
        if method in ["DeepLiftShap", "KernelShap"]:
            warnings.warn(
                f"{method} is computationally expensive and will take some time. For"
                " faster results, try usingsome other methods like GradientShap,"
                " IntegratedGradients etc."
            )
        if method in ["FeaturePermutation", "FeatureAblation"]:
            assert data.shape[0] > 1, f"{method} only works when the number of samples is greater than 1"
            if len(data) <= 100:
                warnings.warn(
                    f"{method} gives better results when the number of samples is"
                    " large. For better results, try using more samples or some other"
                    " methods like GradientShap which works well on single examples."
                )
        is_full_baselines = method in ["GradientShap", "DeepLiftShap"]
        is_not_supported = self.model._get_name() in [
            "TabNetModel",
            "MDNModel",
            "TabTransformerModel",
        ]
        do_baselines = method not in [
            "Saliency",
            "InputXGradient",
            "FeaturePermutation",
            "LRP",
        ]
        if is_full_baselines and (baselines is None or isinstance(baselines, (float, int))):
            raise ValueError(
                f"baselines cannot be a scalar or None for {method}. Please "
                "provide a tensor or a string like `b|<num_samples>`"
            )
        if is_not_supported:
            raise NotImplementedError(f"Attributions are not implemented for {self.model._get_name()}")

        is_embedding1d = isinstance(self.model.embedding_layer, (Embedding1dLayer, PreEncoded1dLayer))
        is_embedding2d = isinstance(self.model.embedding_layer, Embedding2dLayer)
        # Models like NODE may have no embedding dims (doing leaveOneOut encoding) even if categorical_dim > 0
        is_embbeding_dims = (
            hasattr(self.model.hparams, "embedding_dims") and self.model.hparams.embedding_dims is not None
        )
        if (not is_embedding1d) and (not is_embedding2d):
            raise NotImplementedError(
                "Attributions are not implemented for models with this type of" " embedding layer"
            )
        test_dl = self.datamodule.prepare_inference_dataloader(data)
        self.model.eval()
        # prepare import for Captum
        tensor_inp, tensor_tgt = self._prepare_input_for_captum(test_dl)
        baselines = self._prepare_baselines_captum(baselines, test_dl, do_baselines, is_full_baselines)
        # prepare model for Captum
        try:
            interp_model = _CaptumModel(self.model)
            captum_interp_cls = getattr(captum.attr, method)(interp_model, **method_args)
            if do_baselines:
                attributions = captum_interp_cls.attribute(
                    tensor_inp,
                    baselines=baselines,
                    target=(tensor_tgt if self.config.task == "classification" else None),
                    **kwargs,
                )
            else:
                attributions = captum_interp_cls.attribute(
                    tensor_inp,
                    target=(tensor_tgt if self.config.task == "classification" else None),
                    **kwargs,
                )
            attributions = self._handle_categorical_embeddings_attributions(
                attributions, is_embedding1d, is_embedding2d, is_embbeding_dims
            )
        finally:
            self.model.train()
        assert attributions.shape[1] == self.model.hparams.continuous_dim + self.model.hparams.categorical_dim, (
            "Something went wrong. The number of features in the attributions"
            f" ({attributions.shape[1]}) does not match the number of features in"
            " the model"
            f" ({self.model.hparams.continuous_dim+self.model.hparams.categorical_dim})"
        )
        return pd.DataFrame(
            attributions.detach().cpu().numpy(),
            columns=self.config.continuous_cols + self.config.categorical_cols,
        )

feature_importance()

返回模型的特征重要性,格式为pandas DataFrame.

Source code in src/pytorch_tabular/tabular_model.py

def feature_importance(self) -> DataFrame:
    """返回模型的特征重要性,格式为pandas DataFrame."""
    return self.model.feature_importance()

find_learning_rate(model, datamodule, min_lr=1e-08, max_lr=1, num_training=100, mode='exponential', early_stop_threshold=4.0, plot=True, callbacks=None)

允许用户进行一系列良好的初始学习率测试,以减少选择合适起始学习率的猜测工作.

Parameters:

Name	Type	Description	Default
model	LightningModule	要训练的PyTorch Lightning模型.	required
datamodule	TabularDatamodule	数据模块	required
min_lr	Optional[float]	要调查的最小学习率	1e-08
max_lr	Optional[float]	要调查的最大学习率	1
num_training	Optional[int]	要测试的学习率数量	100
mode	Optional[str]	搜索策略,可以是'linear'或'exponential'.如果设置为 'linear',学习率将通过在每个批次后线性增加来搜索.如果设置为'exponential',将指数增加学习率.	'exponential'
early_stop_threshold	Optional[float]	停止搜索的阈值.如果在任何时候损失大于 early_stop_threshold*best_loss,则停止搜索.要禁用,请设置为None.	4.0
plot	bool	如果为真,将使用matplotlib绘图	True
callbacks	Optional[List]	如果提供,将添加到Trainer的回调中.	None

Returns:

Type	Description
Tuple[float, DataFrame]	建议的学习率和学习率查找器的结果

Source code in src/pytorch_tabular/tabular_model.py

    def find_learning_rate(
        self,
        model: pl.LightningModule,
        datamodule: TabularDatamodule,
        min_lr: float = 1e-8,
        max_lr: float = 1,
        num_training: int = 100,
        mode: str = "exponential",
        early_stop_threshold: Optional[float] = 4.0,
        plot: bool = True,
        callbacks: Optional[List] = None,
    ) -> Tuple[float, DataFrame]:
        """    允许用户进行一系列良好的初始学习率测试,以减少选择合适起始学习率的猜测工作.

Parameters:
    model (pl.LightningModule): 要训练的PyTorch Lightning模型.

    datamodule (TabularDatamodule): 数据模块

    min_lr (Optional[float], optional): 要调查的最小学习率

    max_lr (Optional[float], optional): 要调查的最大学习率

    num_training (Optional[int], optional): 要测试的学习率数量

    mode (Optional[str], optional): 搜索策略,可以是'linear'或'exponential'.如果设置为
        'linear',学习率将通过在每个批次后线性增加来搜索.如果设置为'exponential',将指数增加学习率.

    early_stop_threshold (Optional[float], optional): 停止搜索的阈值.如果在任何时候损失大于
        early_stop_threshold*best_loss,则停止搜索.要禁用,请设置为None.

    plot (bool, optional): 如果为真,将使用matplotlib绘图

    callbacks (Optional[List], optional): 如果提供,将添加到Trainer的回调中.

Returns:
    建议的学习率和学习率查找器的结果
"""
        self._prepare_for_training(model, datamodule, callbacks, max_epochs=None, min_epochs=None)
        train_loader, _ = datamodule.train_dataloader(), datamodule.val_dataloader()
        lr_finder = Tuner(self.trainer).lr_find(
            model=self.model,
            train_dataloaders=train_loader,
            val_dataloaders=None,
            min_lr=min_lr,
            max_lr=max_lr,
            num_training=num_training,
            mode=mode,
            early_stop_threshold=early_stop_threshold,
        )
        if plot:
            fig = lr_finder.plot(suggest=True)
            fig.show()
        new_lr = lr_finder.suggestion()
        # cancelling the model and trainer that was loaded
        self.model = None
        self.trainer = None
        self.datamodule = None
        self.callbacks = None
        return new_lr, DataFrame(lr_finder.results)

finetune(max_epochs=None, min_epochs=None, callbacks=None, freeze_backbone=False)

微调模型于提供的数据上.

Parameters:

Name	Type	Description	Default
max_epochs	(Optional[int], 可选)	训练的最大周期数.默认为 None.	None
min_epochs	(Optional[int], 可选)	训练的最小周期数.默认为 None.	None
callbacks	(Optional[List[Callback]], 可选)	如果提供,将添加到 Trainer 的回调中. 默认为 None.	None
freeze_backbone	(bool, 可选)	如果为 True,将通过关闭梯度来冻结主干网络. 默认为 False,这意味着预训练的权重在微调期间也会进一步调整.	False

Returns:

Type	Description
Trainer	pl.Trainer: Trainer 对象

Source code in src/pytorch_tabular/tabular_model.py

    def finetune(
        self,
        max_epochs: Optional[int] = None,
        min_epochs: Optional[int] = None,
        callbacks: Optional[List[pl.Callback]] = None,
        freeze_backbone: bool = False,
    ) -> pl.Trainer:
        """微调模型于提供的数据上.

Parameters:
    max_epochs (Optional[int], 可选): 训练的最大周期数.默认为 None.

    min_epochs (Optional[int], 可选): 训练的最小周期数.默认为 None.

    callbacks (Optional[List[pl.Callback]], 可选): 如果提供,将添加到 Trainer 的回调中.
        默认为 None.

    freeze_backbone (bool, 可选): 如果为 True,将通过关闭梯度来冻结主干网络.
        默认为 False,这意味着预训练的权重在微调期间也会进一步调整.

Returns:
    pl.Trainer: Trainer 对象
"""
        assert self._is_finetune_model, (
            "finetune() can only be called on a finetune model created using" " `TabularModel.create_finetune_model()`"
        )
        seed_everything(self.config.seed)
        if freeze_backbone:
            for param in self.model.backbone.parameters():
                param.requires_grad = False
        return self.train(
            self.model,
            self.datamodule,
            callbacks=callbacks,
            max_epochs=max_epochs,
            min_epochs=min_epochs,
        )

fit(train, validation=None, loss=None, metrics=None, metrics_prob_inputs=None, optimizer=None, optimizer_params=None, train_sampler=None, target_transform=None, max_epochs=None, min_epochs=None, seed=42, callbacks=None, datamodule=None, cache_data='memory', handle_oom=True)

fit方法,接收数据并触发训练.

Parameters:

Name	Type	Description	Default
train	DataFrame	训练数据框	required
validation	Optional[DataFrame]	如果提供,将在训练过程中使用此数据框作为验证集. 用于早停和日志记录.如果未提供,将使用20%的训练数据作为验证集. 默认为None.	None
loss	Optional[Module]	自定义损失函数,不在标准PyTorch库中	None
metrics	Optional[List[Callable]]	自定义度量函数（可调用对象）,具有 签名metric_fn(y_hat, y),并适用于torch张量输入.对于分类任务,y_hat预期形状为 (batch_size, num_classes),对于回归任务,y_hat预期形状为(batch_size, 1),y预期形状为 (batch_size, 1)	None
metrics_prob_inputs	Optional[List[bool]]	这是分类度量的强制参数. 如果度量函数需要概率作为输入,请设置为True. 列表的长度应等于度量函数的数量.默认为None.	None
optimizer	Optional[Optimizer]	自定义优化器,是标准PyTorch优化器的替代品. 这应该是类,而不是初始化的对象	None
optimizer_params	Optional[Dict]	用于初始化自定义优化器的参数.	None
train_sampler	Optional[Sampler]	自定义PyTorch批次采样器,将传递给DataLoaders. 对于处理不平衡数据和其他自定义批次策略很有用	None
target_transform	Optional[Union[TransformerMixin, Tuple(Callable)]]	如果提供,在模型训练前对目标应用变换,在预测时应用逆变换. 参数可以是具有inverse_transform方法的sklearn Transformer, 或由可调用对象组成的元组(transform_func, inverse_transform_func)	None
max_epochs	Optional[int]	覆盖要运行的最大轮数.默认为None.	None
min_epochs	Optional[int]	覆盖要运行的最小轮数.默认为None.	None
seed	Optional[int]	(int): 用于可重复性的随机种子.默认为42.	42
callbacks	Optional[List[Callback]]	训练期间使用的回调列表.默认为None.	None
datamodule	Optional[TabularDatamodule]	数据模块. 如果提供,将忽略其他参数如train、test等,并使用数据模块. 默认为None.	None
cache_data	str	决定如何在数据加载器中缓存数据.如果设置为 "memory",将在内存中缓存.如果设置为有效路径,将在该路径中缓存.默认为"memory".	'memory'
handle_oom	bool	如果为True,将尝试优雅地处理OOM错误.默认为True.	True

Returns:

Type	Description
Trainer	pl.Trainer: PyTorch Lightning Trainer实例

Source code in src/pytorch_tabular/tabular_model.py

    def fit(
        self,
        train: Optional[DataFrame],
        validation: Optional[DataFrame] = None,
        loss: Optional[torch.nn.Module] = None,
        metrics: Optional[List[Callable]] = None,
        metrics_prob_inputs: Optional[List[bool]] = None,
        optimizer: Optional[torch.optim.Optimizer] = None,
        optimizer_params: Dict = None,
        train_sampler: Optional[torch.utils.data.Sampler] = None,
        target_transform: Optional[Union[TransformerMixin, Tuple]] = None,
        max_epochs: Optional[int] = None,
        min_epochs: Optional[int] = None,
        seed: Optional[int] = 42,
        callbacks: Optional[List[pl.Callback]] = None,
        datamodule: Optional[TabularDatamodule] = None,
        cache_data: str = "memory",
        handle_oom: bool = True,
    ) -> pl.Trainer:
        """    fit方法,接收数据并触发训练.

Parameters:
    train (DataFrame): 训练数据框

    validation (Optional[DataFrame], optional):
        如果提供,将在训练过程中使用此数据框作为验证集.
        用于早停和日志记录.如果未提供,将使用20%的训练数据作为验证集.
        默认为None.

    loss (Optional[torch.nn.Module], optional): 自定义损失函数,不在标准PyTorch库中

    metrics (Optional[List[Callable]], optional): 自定义度量函数（可调用对象）,具有
        签名metric_fn(y_hat, y),并适用于torch张量输入.对于分类任务,y_hat预期形状为
        (batch_size, num_classes),对于回归任务,y_hat预期形状为(batch_size, 1),y预期形状为
        (batch_size, 1)

    metrics_prob_inputs (Optional[List[bool]], optional): 这是分类度量的强制参数.
        如果度量函数需要概率作为输入,请设置为True.
        列表的长度应等于度量函数的数量.默认为None.

    optimizer (Optional[torch.optim.Optimizer], optional):
        自定义优化器,是标准PyTorch优化器的替代品.
        这应该是类,而不是初始化的对象

    optimizer_params (Optional[Dict], optional): 用于初始化自定义优化器的参数.

    train_sampler (Optional[torch.utils.data.Sampler], optional):
        自定义PyTorch批次采样器,将传递给DataLoaders.
        对于处理不平衡数据和其他自定义批次策略很有用

    target_transform (Optional[Union[TransformerMixin, Tuple(Callable)]], optional):
        如果提供,在模型训练前对目标应用变换,在预测时应用逆变换.
        参数可以是具有inverse_transform方法的sklearn Transformer,
        或由可调用对象组成的元组(transform_func, inverse_transform_func)

    max_epochs (Optional[int]): 覆盖要运行的最大轮数.默认为None.

    min_epochs (Optional[int]): 覆盖要运行的最小轮数.默认为None.

    seed: (int): 用于可重复性的随机种子.默认为42.

    callbacks (Optional[List[pl.Callback]], optional):
        训练期间使用的回调列表.默认为None.

    datamodule (Optional[TabularDatamodule], optional): 数据模块.
        如果提供,将忽略其他参数如train、test等,并使用数据模块.
        默认为None.

    cache_data (str): 决定如何在数据加载器中缓存数据.如果设置为
        "memory",将在内存中缓存.如果设置为有效路径,将在该路径中缓存.默认为"memory".

    handle_oom (bool): 如果为True,将尝试优雅地处理OOM错误.默认为True.

Returns:
    pl.Trainer: PyTorch Lightning Trainer实例
"""
        assert self.config.task != "ssl", (
            "`fit` is not valid for SSL task. Please use `pretrain` for" " semi-supervised learning"
        )
        if metrics is not None:
            assert len(metrics) == len(
                metrics_prob_inputs or []
            ), "The length of `metrics` and `metrics_prob_inputs` should be equal"
        seed = seed or self.config.seed
        if seed:
            seed_everything(seed)
        if datamodule is None:
            datamodule = self.prepare_dataloader(
                train,
                validation,
                train_sampler,
                target_transform,
                seed,
                cache_data,
            )
        else:
            if train is not None:
                warnings.warn(
                    "train data and datamodule is provided."
                    " Ignoring the train data and using the datamodule."
                    " Set either one of them to None to avoid this warning."
                )
        model = self.prepare_model(
            datamodule,
            loss,
            metrics,
            metrics_prob_inputs,
            optimizer,
            optimizer_params or {},
        )

        return self.train(model, datamodule, callbacks, max_epochs, min_epochs, handle_oom)

load_best_model()

在训练完成后加载最佳模型.

Source code in src/pytorch_tabular/tabular_model.py

def load_best_model(self) -> None:
    """在训练完成后加载最佳模型."""
    if self.trainer.checkpoint_callback is not None:
        if self.verbose:
            logger.info("Loading the best model")
        ckpt_path = self.trainer.checkpoint_callback.best_model_path
        if ckpt_path != "":
            if self.verbose:
                logger.debug(f"Model Checkpoint: {ckpt_path}")
            ckpt = pl_load(ckpt_path, map_location=lambda storage, loc: storage)
            self.model.load_state_dict(ckpt["state_dict"])
        else:
            logger.warning("No best model available to load. Did you run it more than 1" " epoch?...")
    else:
        logger.warning(
            "No best model available to load. Checkpoint Callback needs to be" " enabled for this to work"
        )

load_model(dir, map_location=None, strict=True) classmethod

加载保存在目录中的模型.

Parameters:

Name	Type	Description	Default
dir	str	保存模型的目录,包含检查点	required
map_location	Union[Dict[str, str], str, device, int, Callable, None])	如果你的检查点保存了一个GPU模型,而你现在在CPU上或不同数量的GPU上加载,使用这个参数来映射到新的设置.行为与torch.load()中的相同	None
strict	bool)	是否严格要求checkpoint_path中的键与该模块的状态字典返回的键完全匹配.默认值: True.	True

Returns:

Name	Type	Description
TabularModel	TabularModel	保存的TabularModel

Source code in src/pytorch_tabular/tabular_model.py

    @classmethod
    def load_model(cls, dir: str, map_location=None, strict=True):
        """加载保存在目录中的模型.

Parameters:
    dir (str): 保存模型的目录,包含检查点
    map_location (Union[Dict[str, str], str, device, int, Callable, None]) : 如果你的检查点保存了一个GPU模型,而你现在在CPU上或不同数量的GPU上加载,使用这个参数来映射到新的设置.行为与torch.load()中的相同
    strict (bool) : 是否严格要求checkpoint_path中的键与该模块的状态字典返回的键完全匹配.默认值: True.

Returns:
    TabularModel (TabularModel): 保存的TabularModel
"""
        config = OmegaConf.load(os.path.join(dir, "config.yml"))
        datamodule = joblib.load(os.path.join(dir, "datamodule.sav"))
        if (
            hasattr(config, "log_target")
            and (config.log_target is not None)
            and os.path.exists(os.path.join(dir, "exp_logger.sav"))
        ):
            logger = joblib.load(os.path.join(dir, "exp_logger.sav"))
        else:
            logger = None
        if os.path.exists(os.path.join(dir, "callbacks.sav")):
            callbacks = joblib.load(os.path.join(dir, "callbacks.sav"))
            # Excluding Gradient Accumulation Scheduler Callback as we are creating
            # a new one in trainer
            callbacks = [c for c in callbacks if not isinstance(c, GradientAccumulationScheduler)]
        else:
            callbacks = []
        if os.path.exists(os.path.join(dir, "custom_model_callable.sav")):
            model_callable = joblib.load(os.path.join(dir, "custom_model_callable.sav"))
            custom_model = True
        else:
            model_callable = getattr_nested(config._module_src, config._model_name)
            # model_callable = getattr(
            #     getattr(models, config._module_src), config._model_name
            # )
            custom_model = False
        inferred_config = datamodule.update_config(config)
        inferred_config = OmegaConf.structured(inferred_config)
        model_args = {
            "config": config,
            "inferred_config": inferred_config,
        }
        custom_params = joblib.load(os.path.join(dir, "custom_params.sav"))
        if custom_params.get("custom_loss") is not None:
            model_args["loss"] = "MSELoss"  # For compatibility. Not Used
        if custom_params.get("custom_metrics") is not None:
            model_args["metrics"] = ["mean_squared_error"]  # For compatibility. Not Used
            model_args["metrics_params"] = [{}]  # For compatibility. Not Used
            model_args["metrics_prob_inputs"] = [False]  # For compatibility. Not Used
        if custom_params.get("custom_optimizer") is not None:
            model_args["optimizer"] = "Adam"  # For compatibility. Not Used
        if custom_params.get("custom_optimizer_params") is not None:
            model_args["optimizer_params"] = {}  # For compatibility. Not Used

        # Initializing with default metrics, losses, and optimizers. Will revert once initialized
        try:
            model = model_callable.load_from_checkpoint(
                checkpoint_path=os.path.join(dir, "model.ckpt"),
                map_location=map_location,
                strict=strict,
                **model_args,
            )
        except RuntimeError as e:
            if (
                "Unexpected key(s) in state_dict" in str(e)
                and "loss.weight" in str(e)
                and "custom_loss.weight" in str(e)
            ):
                # Custom loss will be loaded after the model is initialized
                # continuing with strict=False
                model = model_callable.load_from_checkpoint(
                    checkpoint_path=os.path.join(dir, "model.ckpt"),
                    map_location=map_location,
                    strict=False,
                    **model_args,
                )
            else:
                raise e
        if custom_params.get("custom_optimizer") is not None:
            model.custom_optimizer = custom_params["custom_optimizer"]
        if custom_params.get("custom_optimizer_params") is not None:
            model.custom_optimizer_params = custom_params["custom_optimizer_params"]
        if custom_params.get("custom_loss") is not None:
            model.loss = custom_params["custom_loss"]
        if custom_params.get("custom_metrics") is not None:
            model.custom_metrics = custom_params.get("custom_metrics")
            model.hparams.metrics = [m.__name__ for m in custom_params.get("custom_metrics")]
            model.hparams.metrics_params = [{}]
            model.hparams.metrics_prob_input = custom_params.get("custom_metrics_prob_inputs")
        model._setup_loss()
        model._setup_metrics()
        tabular_model = cls(config=config, model_callable=model_callable)
        tabular_model.model = model
        tabular_model.custom_model = custom_model
        tabular_model.datamodule = datamodule
        tabular_model.callbacks = callbacks
        tabular_model.trainer = tabular_model._prepare_trainer(callbacks=callbacks)
        # tabular_model.trainer.model = model
        tabular_model.logger = logger
        return tabular_model

load_weights(path)

加载指定目录中的模型权重.

Parameters:

Name	Type	Description	Default
path	str	要从中加载模型的文件路径	required

Returns:

Type	Description
None	None

Source code in src/pytorch_tabular/tabular_model.py

    def load_weights(self, path: Union[str, Path]) -> None:
        """加载指定目录中的模型权重.

Parameters:
    path (str): 要从中加载模型的文件路径

Returns:
    None
"""
        self._load_weights(self.model, path)

predict(test, quantiles=[0.25, 0.5, 0.75], n_samples=100, ret_logits=False, include_input_features=False, device=None, progress_bar=None, test_time_augmentation=False, num_tta=5, alpha_tta=0.1, aggregate_tta='mean', tta_seed=42)

使用训练好的模型对新数据进行预测,并以数据框形式返回结果.

Parameters:

Name	Type	Description	Default
test	DataFrame	包含训练期间定义的特征的新数据框	required
quantiles	可选[List]	对于概率模型（如混合密度网络）,这指定了除了central_tendency之外要提取的不同分位数并添加到数据框中. 对于其他模型,此参数被忽略.默认为 [0.25, 0.5, 0.75]	[0.25, 0.5, 0.75]
n_samples	可选[int]	从后验分布中抽取的样本数量,用于估计分位数. 对于非概率模型,此参数被忽略.默认为 100	100
ret_logits	bool	标志,用于返回原始模型输出/logits（除了骨干特征）以及数据框.默认为 False	False
include_input_features	bool	已弃用: 标志,用于在返回的数据框中包含输入特征.默认为 True	False
progress_bar	Optional[str]	选择用于跟踪进度的进度条."rich" 或 "tqdm" 将设置相应的进度条.如果为 None,则不显示进度条.	None
test_time_augmentation	bool	如果为 True,将使用测试时增强来生成预测. 该方法与此处描述的方法非常相似, 但我们还在嵌入输入中添加噪声以处理分类特征. (x_{aug} = x_{orig} + lpha * \epsilon) 其中 (\epsilon \sim \mathcal{N}(0, 1)) 默认为 False	False
num_tta	float	为 TTA 运行的增强次数.默认为 0.0	5
alpha_tta	float	要添加到输入特征的高斯噪声的标准差	0.1
aggregate_tta	(Union[str, Callable], 可选)	用于聚合每次增强预测的函数.如果为 str,应为 "mean", "median", "min", 或 "max" 之一 用于回归.对于分类,前面的选项应用于置信度分数（软投票）,然后转换为最终预测.分类还提供了一个额外的选项 "hard_voting". 如果为可调用对象,应为一个函数,该函数接收一个包含 3D 数组（num_samples, num_cv, num_targets）的列表,并返回一个 2D 数组 的最终概率（num_samples, num_targets）.默认为 "mean".	'mean'
tta_seed	int	用于 TTA 中添加噪声的随机种子.默认为 42.	42

Returns:

Name	Type	Description
DataFrame	DataFrame	返回一个包含预测和特征（如果 include_input_features=True）的数据框. 如果是分类,则返回概率和最终预测

Source code in src/pytorch_tabular/tabular_model.py

    def predict(
        self,
        test: DataFrame,
        quantiles: Optional[List] = [0.25, 0.5, 0.75],
        n_samples: Optional[int] = 100,
        ret_logits=False,
        include_input_features: bool = False,
        device: Optional[torch.device] = None,
        progress_bar: Optional[str] = None,
        test_time_augmentation: Optional[bool] = False,
        num_tta: Optional[float] = 5,
        alpha_tta: Optional[float] = 0.1,
        aggregate_tta: Optional[str] = "mean",
        tta_seed: Optional[int] = 42,
    ) -> DataFrame:
        """使用训练好的模型对新数据进行预测,并以数据框形式返回结果.

Parameters:
    test (DataFrame): 包含训练期间定义的特征的新数据框

    quantiles (可选[List]): 对于概率模型（如混合密度网络）,这指定了除了`central_tendency`之外要提取的不同分位数并添加到数据框中.
        对于其他模型,此参数被忽略.默认为 [0.25, 0.5, 0.75]

    n_samples (可选[int]): 从后验分布中抽取的样本数量,用于估计分位数.
        对于非概率模型,此参数被忽略.默认为 100

    ret_logits (bool): 标志,用于返回原始模型输出/logits（除了骨干特征）以及数据框.默认为 False

    include_input_features (bool): 已弃用: 标志,用于在返回的数据框中包含输入特征.默认为 True

    progress_bar: 选择用于跟踪进度的进度条."rich" 或 "tqdm" 将设置相应的进度条.如果为 None,则不显示进度条.

    test_time_augmentation (bool): 如果为 True,将使用测试时增强来生成预测.
        该方法与[此处](https://kozodoi.me/blog/20210908/tta-tabular)描述的方法非常相似,
        但我们还在嵌入输入中添加噪声以处理分类特征.                \(x_{aug} = x_{orig} + \alpha * \epsilon\) 其中 \(\epsilon \sim \mathcal{N}(0, 1)\)
        默认为 False
    num_tta (float): 为 TTA 运行的增强次数.默认为 0.0

    alpha_tta (float): 要添加到输入特征的高斯噪声的标准差

    aggregate_tta (Union[str, Callable], 可选): 用于聚合每次增强预测的函数.如果为 str,应为 "mean", "median", "min", 或 "max" 之一
        用于回归.对于分类,前面的选项应用于置信度分数（软投票）,然后转换为最终预测.分类还提供了一个额外的选项
        "hard_voting".
        如果为可调用对象,应为一个函数,该函数接收一个包含 3D 数组（num_samples, num_cv, num_targets）的列表,并返回一个 2D 数组
        的最终概率（num_samples, num_targets）.默认为 "mean".

    tta_seed (int): 用于 TTA 中添加噪声的随机种子.默认为 42.

Returns:
    DataFrame: 返回一个包含预测和特征（如果 `include_input_features=True`）的数据框.
        如果是分类,则返回概率和最终预测
"""
        warnings.warn(
            "`include_input_features` will be deprecated in the next release."
            " Please add index columns to the test dataframe if you want to"
            " retain some features like the key or id",
            DeprecationWarning,
        )
        if test_time_augmentation:
            assert num_tta > 0, "num_tta should be greater than 0"
            assert alpha_tta > 0, "alpha_tta should be greater than 0"
            assert include_input_features is False, "include_input_features cannot be True for TTA."
            if not callable(aggregate_tta):
                assert aggregate_tta in [
                    "mean",
                    "median",
                    "min",
                    "max",
                    "hard_voting",
                ], "aggregate should be one of 'mean', 'median', 'min', 'max', or" " 'hard_voting'"
            if self.config.task == "regression":
                assert aggregate_tta != "hard_voting", "hard_voting is only available for classification"

            torch.manual_seed(tta_seed)

            def add_noise(module, input, output):
                return output + alpha_tta * torch.randn_like(output, memory_format=torch.contiguous_format)

            # Register the hook to the embedding_layer
            handle = self.model.embedding_layer.register_forward_hook(add_noise)
            pred_prob_l = []
            for _ in range(num_tta):
                pred_df = self._predict(
                    test,
                    quantiles,
                    n_samples,
                    ret_logits,
                    include_input_features=False,
                    device=device,
                    progress_bar=progress_bar or "None",
                )
                pred_idx = pred_df.index
                if self.config.task == "classification":
                    pred_prob_l.append(pred_df.values[:, : -len(self.config.target)])
                elif self.config.task == "regression":
                    pred_prob_l.append(pred_df.values)
            pred_df = self._combine_predictions(pred_prob_l, pred_idx, aggregate_tta, None)
            # Remove the hook
            handle.remove()
        else:
            pred_df = self._predict(
                test,
                quantiles,
                n_samples,
                ret_logits,
                include_input_features,
                device,
                progress_bar,
            )
        return pred_df

prepare_dataloader(train, validation=None, train_sampler=None, target_transform=None, seed=42, cache_data='memory')

准备用于训练和验证的数据加载器.

Parameters:

Name	Type	Description	Default
train	DataFrame	训练数据框	required
validation	Optional[DataFrame]	如果提供,将在训练过程中使用此数据框作为验证集. 用于早停和日志记录.如果未提供,将使用训练数据的20%作为验证集. 默认为 None.	None
train_sampler	Optional[Sampler]	自定义的 PyTorch 批次采样器,将传递给 DataLoaders. 适用于处理不平衡数据和其他自定义批次策略.	None
target_transform	Optional[Union[TransformerMixin, Tuple(Callable)]]	如果提供,在模型训练前对目标应用此变换,并在预测时应用逆变换. 参数可以是具有 inverse_transform 方法的 sklearn Transformer,或 由可调用对象组成的元组 (transform_func, inverse_transform_func).	None
seed	Optional[int]	用于可重复性的随机种子.默认为 42.	42
cache_data	str	决定如何在数据加载器中缓存数据.如果设置为 "memory",将在内存中缓存.如果设置为有效路径,将在该路径中缓存.默认为 "memory".	'memory'

Returns:

Name	Type	Description
TabularDatamodule	TabularDatamodule	准备好的数据模块

Source code in src/pytorch_tabular/tabular_model.py

    def prepare_dataloader(
        self,
        train: DataFrame,
        validation: Optional[DataFrame] = None,
        train_sampler: Optional[torch.utils.data.Sampler] = None,
        target_transform: Optional[Union[TransformerMixin, Tuple]] = None,
        seed: Optional[int] = 42,
        cache_data: str = "memory",
    ) -> TabularDatamodule:
        """准备用于训练和验证的数据加载器.

Parameters:
    train (DataFrame): 训练数据框

    validation (Optional[DataFrame], optional):
        如果提供,将在训练过程中使用此数据框作为验证集.
        用于早停和日志记录.如果未提供,将使用训练数据的20%作为验证集.
        默认为 None.

    train_sampler (Optional[torch.utils.data.Sampler], optional):
        自定义的 PyTorch 批次采样器,将传递给 DataLoaders.
        适用于处理不平衡数据和其他自定义批次策略.

    target_transform (Optional[Union[TransformerMixin, Tuple(Callable)]], optional):
        如果提供,在模型训练前对目标应用此变换,并在预测时应用逆变换.
        参数可以是具有 inverse_transform 方法的 sklearn Transformer,或
        由可调用对象组成的元组 (transform_func, inverse_transform_func).

    seed (Optional[int], optional): 用于可重复性的随机种子.默认为 42.

    cache_data (str): 决定如何在数据加载器中缓存数据.如果设置为
        "memory",将在内存中缓存.如果设置为有效路径,将在该路径中缓存.默认为 "memory".

Returns:
    TabularDatamodule: 准备好的数据模块
"""
        if self.verbose:
            logger.info("Preparing the DataLoaders")
        target_transform = self._check_and_set_target_transform(target_transform)

        datamodule = TabularDatamodule(
            train=train,
            validation=validation,
            config=self.config,
            target_transform=target_transform,
            train_sampler=train_sampler,
            seed=seed,
            cache_data=cache_data,
            verbose=self.verbose,
        )
        datamodule.prepare_data()
        datamodule.setup("fit")
        return datamodule

prepare_model(datamodule, loss=None, metrics=None, metrics_prob_inputs=None, optimizer=None, optimizer_params=None)

准备模型以进行训练.

Parameters:

Name	Type	Description	Default
datamodule	TabularDatamodule	数据模块	required
loss	(Optional[Module], 可选)	自定义损失函数,不在标准 PyTorch 库中	None
metrics	(Optional[List[Callable]], 可选)	自定义度量函数（可调用对象）,具有 metric_fn(y_hat, y) 签名并作用于 torch 张量输入	None
metrics_prob_inputs	(Optional[List[bool]], 可选)	这是分类度量的必填参数.如果度量函数需要概率作为输入,请设置为 True. 列表的长度应等于度量函数的数量.默认为 None.	None
optimizer	(Optional[Optimizer], 可选)	自定义优化器,是标准 PyTorch 优化器的直接替代品. 这应该是类,而不是初始化的对象	None
optimizer_params	(Optional[Dict], 可选)	用于初始化自定义优化器的参数.	None

Returns:

Name	Type	Description
BaseModel	BaseModel	准备好的模型

Source code in src/pytorch_tabular/tabular_model.py

    def prepare_model(
        self,
        datamodule: TabularDatamodule,
        loss: Optional[torch.nn.Module] = None,
        metrics: Optional[List[Callable]] = None,
        metrics_prob_inputs: Optional[List[bool]] = None,
        optimizer: Optional[torch.optim.Optimizer] = None,
        optimizer_params: Dict = None,
    ) -> BaseModel:
        """准备模型以进行训练.

Parameters:
    datamodule (TabularDatamodule): 数据模块

    loss (Optional[torch.nn.Module], 可选): 自定义损失函数,不在标准 PyTorch 库中

    metrics (Optional[List[Callable]], 可选): 自定义度量函数（可调用对象）,具有 metric_fn(y_hat, y) 签名并作用于 torch 张量输入

    metrics_prob_inputs (Optional[List[bool]], 可选): 这是分类度量的必填参数.如果度量函数需要概率作为输入,请设置为 True.
        列表的长度应等于度量函数的数量.默认为 None.

    optimizer (Optional[torch.optim.Optimizer], 可选):
        自定义优化器,是标准 PyTorch 优化器的直接替代品.
        这应该是类,而不是初始化的对象

    optimizer_params (Optional[Dict], 可选): 用于初始化自定义优化器的参数.

Returns:
    BaseModel: 准备好的模型
"""
        if self.verbose:
            logger.info(f"Preparing the Model: {self.config._model_name}")
        # Fetching the config as some data specific configs have been added in the datamodule
        self.inferred_config = self._read_parse_config(datamodule.update_config(self.config), InferredConfig)
        model = self.model_callable(
            self.config,
            custom_loss=loss,  # Unused in SSL tasks
            custom_metrics=metrics,  # Unused in SSL tasks
            custom_metrics_prob_inputs=metrics_prob_inputs,  # Unused in SSL tasks
            custom_optimizer=optimizer,
            custom_optimizer_params=optimizer_params or {},
            inferred_config=self.inferred_config,
        )
        # Data Aware Initialization(for the models that need it)
        model.data_aware_initialization(datamodule)
        if self.model_state_dict_path is not None:
            self._load_weights(model, self.model_state_dict_path)
        if self.track_experiment and self.config.log_target == "wandb":
            self.logger.watch(model, log=self.config.exp_watch, log_freq=self.config.exp_log_freq)
        return model

pretrain(train, validation=None, optimizer=None, optimizer_params=None, max_epochs=None, min_epochs=None, seed=42, callbacks=None, datamodule=None, cache_data='memory')

预训练方法,接收数据并触发训练.

Parameters:

Name	Type	Description	Default
train	DataFrame	训练数据框	required
validation	Optional[DataFrame]	如果提供,将在训练过程中使用此数据框作为验证集. 用于早停和日志记录.如果未提供,将使用训练数据的20%作为验证集.默认为None.	None
optimizer	Optional[Optimizer]	自定义优化器,可作为标准PyTorch优化器的替代品. 应为类,而非初始化对象.	None
optimizer_params	Optional[Dict]	用于初始化自定义优化器的参数.	None
max_epochs	Optional[int]	覆盖要运行的最大周期数.默认为None.	None
min_epochs	Optional[int]	覆盖要运行的最小周期数.默认为None.	None
seed	Optional[int]	(int): 随机种子,用于可重复性.默认为42.	42
callbacks	Optional[List[Callback]]	训练过程中使用的回调列表. 默认为None.	None
datamodule	Optional[TabularDatamodule]	数据模块.如果提供,将忽略其他参数如train、test等, 并使用数据模块.默认为None.	None
cache_data	str	决定如何在数据加载器中缓存数据.如果设置为"memory",将在内存中缓存. 如果设置为有效路径,将在该路径中缓存.默认为"memory".	'memory'

Returns: pl.Trainer: PyTorch Lightning Trainer实例

Source code in src/pytorch_tabular/tabular_model.py

    def pretrain(
        self,
        train: Optional[DataFrame],
        validation: Optional[DataFrame] = None,
        optimizer: Optional[torch.optim.Optimizer] = None,
        optimizer_params: Dict = None,
        # train_sampler: Optional[torch.utils.data.Sampler] = None,
        max_epochs: Optional[int] = None,
        min_epochs: Optional[int] = None,
        seed: Optional[int] = 42,
        callbacks: Optional[List[pl.Callback]] = None,
        datamodule: Optional[TabularDatamodule] = None,
        cache_data: str = "memory",
    ) -> pl.Trainer:
        """    预训练方法,接收数据并触发训练.

Parameters:
    train (DataFrame): 训练数据框

    validation (Optional[DataFrame], optional): 如果提供,将在训练过程中使用此数据框作为验证集.
        用于早停和日志记录.如果未提供,将使用训练数据的20%作为验证集.默认为None.

    optimizer (Optional[torch.optim.Optimizer], optional): 自定义优化器,可作为标准PyTorch优化器的替代品.
        应为类,而非初始化对象.

    optimizer_params (Optional[Dict], optional): 用于初始化自定义优化器的参数.

    max_epochs (Optional[int]): 覆盖要运行的最大周期数.默认为None.

    min_epochs (Optional[int]): 覆盖要运行的最小周期数.默认为None.

    seed: (int): 随机种子,用于可重复性.默认为42.

    callbacks (Optional[List[pl.Callback]], optional): 训练过程中使用的回调列表.
        默认为None.

    datamodule (Optional[TabularDatamodule], optional): 数据模块.如果提供,将忽略其他参数如train、test等,
        并使用数据模块.默认为None.

    cache_data (str): 决定如何在数据加载器中缓存数据.如果设置为"memory",将在内存中缓存.
        如果设置为有效路径,将在该路径中缓存.默认为"memory".
Returns:
    pl.Trainer: PyTorch Lightning Trainer实例
"""
        assert self.config.task == "ssl", (
            f"`pretrain` is not valid for {self.config.task} task. Please use `fit`" " instead."
        )
        seed = seed or self.config.seed
        if seed:
            seed_everything(seed)
        if datamodule is None:
            datamodule = self.prepare_dataloader(
                train,
                validation,
                train_sampler=None,
                target_transform=None,
                seed=seed,
                cache_data=cache_data,
            )
        else:
            if train is not None:
                warnings.warn(
                    "train data and datamodule is provided."
                    " Ignoring the train data and using the datamodule."
                    " Set either one of them to None to avoid this warning."
                )
        model = self.prepare_model(
            datamodule,
            optimizer,
            optimizer_params or {},
        )

        return self.train(model, datamodule, callbacks, max_epochs, min_epochs)

save_config(dir)

将配置保存到指定目录.

Source code in src/pytorch_tabular/tabular_model.py

def save_config(self, dir: str) -> None:
    """将配置保存到指定目录."""
    with open(os.path.join(dir, "config.yml"), "w") as fp:
        OmegaConf.save(self.config, fp, resolve=True)

save_datamodule(dir, inference_only=False)

Saves the datamodule in the specified directory.

Parameters:

Name	Type	Description	Default
dir	str	保存datamodule的目录路径	required
inference_only	bool	如果为True,将仅保存不带数据的推理datamodule. 这不能用于进一步训练,但可用于推理.默认为False.	False

Source code in src/pytorch_tabular/tabular_model.py

    def save_datamodule(self, dir: str, inference_only: bool = False) -> None:
        """    Saves the datamodule in the specified directory.

Args:
    dir (str): 保存datamodule的目录路径
    inference_only (bool): 如果为True,将仅保存不带数据的推理datamodule.
        这不能用于进一步训练,但可用于推理.默认为False.
"""
        if inference_only:
            dm = self.datamodule.inference_only_copy()
        else:
            dm = self.datamodule

        joblib.dump(dm, os.path.join(dir, "datamodule.sav"))

save_model(dir, inference_only=False)

保存模型和检查点在指定目录中.

Parameters:

Name	Type	Description	Default
dir	str	保存模型的目录路径	required
inference_only	bool	如果为True,将仅保存数据模块的推理版本	False

Source code in src/pytorch_tabular/tabular_model.py

    def save_model(self, dir: str, inference_only: bool = False) -> None:
        """    保存模型和检查点在指定目录中.

Parameters:
    dir (str): 保存模型的目录路径
    inference_only (bool): 如果为True,将仅保存数据模块的推理版本
"""
        if os.path.exists(dir) and (os.listdir(dir)):
            logger.warning("Directory is not empty. Overwriting the contents.")
            for f in os.listdir(dir):
                os.remove(os.path.join(dir, f))
        os.makedirs(dir, exist_ok=True)
        self.save_config(dir)
        self.save_datamodule(dir, inference_only=inference_only)
        if hasattr(self.config, "log_target") and self.config.log_target is not None:
            joblib.dump(self.logger, os.path.join(dir, "exp_logger.sav"))
        if hasattr(self, "callbacks"):
            joblib.dump(self.callbacks, os.path.join(dir, "callbacks.sav"))
        self.trainer.save_checkpoint(os.path.join(dir, "model.ckpt"))
        custom_params = {}
        custom_params["custom_loss"] = getattr(self.model, "custom_loss", None)
        custom_params["custom_metrics"] = getattr(self.model, "custom_metrics", None)
        custom_params["custom_metrics_prob_inputs"] = getattr(self.model, "custom_metrics_prob_inputs", None)
        custom_params["custom_optimizer"] = getattr(self.model, "custom_optimizer", None)
        custom_params["custom_optimizer_params"] = getattr(self.model, "custom_optimizer_params", None)
        joblib.dump(custom_params, os.path.join(dir, "custom_params.sav"))
        if self.custom_model:
            joblib.dump(self.model_callable, os.path.join(dir, "custom_model_callable.sav"))

save_model_for_inference(path, kind='pytorch', onnx_export_params={'opset_version': 12})

保存模型以供推理.

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	保存模型的路径	required
kind	str	"pytorch" 或 "onnx"（实验性）	'pytorch'
onnx_export_params	Dict	传递给 torch.onnx.export 的 ONNX 导出参数	{'opset_version': 12}

Returns:

Name	Type	Description
bool	bool	如果模型成功保存则为 True

Source code in src/pytorch_tabular/tabular_model.py

    def save_model_for_inference(
        self,
        path: Union[str, Path],
        kind: str = "pytorch",
        onnx_export_params: Dict = {"opset_version": 12},
    ) -> bool:
        """保存模型以供推理.

Parameters:
    path (Union[str, Path]): 保存模型的路径
    kind (str): "pytorch" 或 "onnx"（实验性）
    onnx_export_params (Dict): 传递给 torch.onnx.export 的 ONNX 导出参数

Returns:
    bool: 如果模型成功保存则为 True
"""
        if kind == "pytorch":
            torch.save(self.model, str(path))
            return True
        elif kind == "onnx":
            # Export the model
            onnx_export_params["input_names"] = ["categorical", "continuous"]
            onnx_export_params["output_names"] = onnx_export_params.get("output_names", ["output"])
            onnx_export_params["dynamic_axes"] = {
                onnx_export_params["input_names"][0]: {0: "batch_size"},
                onnx_export_params["output_names"][0]: {0: "batch_size"},
            }
            cat = torch.zeros(
                self.config.batch_size,
                len(self.config.categorical_cols),
                dtype=torch.int,
            )
            cont = torch.randn(
                self.config.batch_size,
                len(self.config.continuous_cols),
                requires_grad=True,
            )
            x = {"continuous": cont, "categorical": cat}
            torch.onnx.export(self.model, x, str(path), **onnx_export_params)
            return True
        else:
            raise ValueError("`kind` must be either pytorch or onnx")

save_weights(path)

保存模型权重到指定目录.

Parameters:

Name	Type	Description	Default
path	str	保存模型的文件路径	required

Source code in src/pytorch_tabular/tabular_model.py

    def save_weights(self, path: Union[str, Path]) -> None:
        """保存模型权重到指定目录.

Parameters:
    path (str): 保存模型的文件路径
"""
        torch.save(self.model.state_dict(), path)

summary(model=None, max_depth=-1)

打印模型的摘要.

Parameters:

Name	Type	Description	Default
max_depth	int	遍历模块并显示在摘要中的最大深度. 默认为 -1,表示将显示所有模块.	-1

Source code in src/pytorch_tabular/tabular_model.py

    def summary(self, model=None, max_depth: int = -1) -> None:
        """    打印模型的摘要.

Parameters:
    max_depth (int): 遍历模块并显示在摘要中的最大深度.
        默认为 -1,表示将显示所有模块.
"""
        if model is not None:
            print(summarize(model, max_depth=max_depth))
        elif self.has_model:
            print(summarize(self.model, max_depth=max_depth))
        else:
            rich_print(f"[bold green]{self.__class__.__name__}[/bold green]")
            rich_print("-" * 100)
            rich_print("[bold yellow]Config[/bold yellow]")
            rich_print("-" * 100)
            pprint(self.config.__dict__["_content"])
            rich_print(
                ":triangular_flag:[bold red]Full Model Summary once model has "
                "been initialized or passed in as an argument[/bold red]"
            )

train(model, datamodule, callbacks=None, max_epochs=None, min_epochs=None, handle_oom=True)

训练模型.

Parameters:

Name	Type	Description	Default
model	LightningModule	要训练的PyTorch Lightning模型.	required
datamodule	TabularDatamodule	数据模块	required
callbacks	Optional[List[Callback]]	训练期间使用的回调函数列表.默认为None.	None
max_epochs	Optional[int]	覆盖要运行的最大epoch数.默认为None.	None
min_epochs	Optional[int]	覆盖要运行的最小epoch数.默认为None.	None
handle_oom	bool	如果为True,将尝试优雅地处理OOM错误.默认为True.	True

Returns:

Type	Description
Trainer	pl.Trainer: PyTorch Lightning Trainer实例

Source code in src/pytorch_tabular/tabular_model.py

    def train(
        self,
        model: pl.LightningModule,
        datamodule: TabularDatamodule,
        callbacks: Optional[List[pl.Callback]] = None,
        max_epochs: int = None,
        min_epochs: int = None,
        handle_oom: bool = True,
    ) -> pl.Trainer:
        """    训练模型.

Parameters:
    model (pl.LightningModule): 要训练的PyTorch Lightning模型.

    datamodule (TabularDatamodule): 数据模块

    callbacks (Optional[List[pl.Callback]], optional):
        训练期间使用的回调函数列表.默认为None.

    max_epochs (Optional[int]): 覆盖要运行的最大epoch数.默认为None.

    min_epochs (Optional[int]): 覆盖要运行的最小epoch数.默认为None.

    handle_oom (bool): 如果为True,将尝试优雅地处理OOM错误.默认为True.

Returns:
    pl.Trainer: PyTorch Lightning Trainer实例
"""
        self._prepare_for_training(model, datamodule, callbacks, max_epochs, min_epochs)
        train_loader, val_loader = (
            self.datamodule.train_dataloader(),
            self.datamodule.val_dataloader(),
        )
        self.model.train()
        if self.config.auto_lr_find and (not self.config.fast_dev_run):
            if self.verbose:
                logger.info("Auto LR Find Started")
            with OutOfMemoryHandler(handle_oom=handle_oom) as oom_handler:
                result = Tuner(self.trainer).lr_find(
                    self.model,
                    train_dataloaders=train_loader,
                    val_dataloaders=val_loader,
                )
            if oom_handler.oom_triggered:
                raise OOMException(
                    "OOM detected during LR Find. Try reducing your batch_size or the"
                    " model parameters." + "/n" + "Original Error: " + oom_handler.oom_msg
                )
            if self.verbose:
                logger.info(
                    f"Suggested LR: {result.suggestion()}. For plot and detailed"
                    " analysis, use `find_learning_rate` method."
                )
            self.model.reset_weights()
            # Parameters in models needs to be initialized again after LR find
            self.model.data_aware_initialization(self.datamodule)
        self.model.train()
        if self.verbose:
            logger.info("Training Started")
        with OutOfMemoryHandler(handle_oom=handle_oom) as oom_handler:
            self.trainer.fit(self.model, train_loader, val_loader)
        if oom_handler.oom_triggered:
            raise OOMException(
                "OOM detected during Training. Try reducing your batch_size or the"
                " model parameters."
                "/n" + "Original Error: " + oom_handler.oom_msg
            )
        self._is_fitted = True
        if self.verbose:
            logger.info("Training the model completed")
        if self.config.load_best:
            self.load_best_model()
        return self.trainer

Bases: LightningDataModule

Source code in src/pytorch_tabular/tabular_datamodule.py

class TabularDatamodule(pl.LightningDataModule):
    CONTINUOUS_TRANSFORMS = {
        "quantile_uniform": {
            "callable": QuantileTransformer,
            "params": {"output_distribution": "uniform", "random_state": None},
        },
        "quantile_normal": {
            "callable": QuantileTransformer,
            "params": {"output_distribution": "normal", "random_state": None},
        },
        "box-cox": {
            "callable": PowerTransformer,
            "params": {"method": "box-cox", "standardize": False},
        },
        "yeo-johnson": {
            "callable": PowerTransformer,
            "params": {"method": "yeo-johnson", "standardize": False},
        },
    }

    class CACHE_MODES(Enum):
        MEMORY = "memory"
        DISK = "disk"
        INFERENCE = "inference"

    def __init__(
        self,
        train: DataFrame,
        config: DictConfig,
        validation: DataFrame = None,
        target_transform: Optional[Union[TransformerMixin, Tuple]] = None,
        train_sampler: Optional[torch.utils.data.Sampler] = None,
        seed: Optional[int] = 42,
        cache_data: str = "memory",
        copy_data: bool = True,
        verbose: bool = True,
    ):
        """    Pytorch Lightning 用于表格数据的 Datamodule.

Parameters:
    train (DataFrame): 训练数据框

    config (DictConfig): 从 ModelConfig、DataConfig、
        TrainerConfig、OptimizerConfig 和 ExperimentConfig 合并的配置对象

    validation (DataFrame, 可选): 验证数据框.
        如果留空,我们将使用 DataConfig 中的验证分割来随机抽取样本作为验证.
        默认为 None.

    target_transform (Optional[Union[TransformerMixin, Tuple(Callable)]], 可选):
        如果提供,将在建模前对目标应用变换,并在预测时进行逆变换.参数可以是具有 inverse_transform 方法的 sklearn Transformer,
        或由可调用对象组成的元组 (transform_func, inverse_transform_func)
        默认为 None.

    train_sampler (Optional[torch.utils.data.Sampler], 可选):
        如果提供,将使用该采样器对训练数据进行采样.默认为 None.

    seed (Optional[int], 可选): 用于可重复数据加载器的种子.默认为 42.

    cache_data (str): 决定如何在数据加载器中缓存数据.如果设置为
        "memory",将在内存中缓存.如果设置为有效路径,将在该路径中缓存.默认为 "memory".

    copy_data (bool): 如果为 True,将在预处理前复制数据框.默认为 True.

    verbose (bool): 设置 databodule 日志的详细程度
"""
        super().__init__()
        self.train = train.copy() if copy_data else train
        if validation is not None:
            self.validation = validation.copy() if copy_data else validation
        else:
            self.validation = None
        self._set_target_transform(target_transform)
        self.target = config.target or []
        self.batch_size = config.batch_size
        self.train_sampler = train_sampler
        self.config = config
        self.seed = seed
        self.verbose = verbose
        self._fitted = False
        self._setup_cache(cache_data)
        self._inferred_config = self._update_config(config)

    @property
    def categorical_encoder(self):
        """返回分类编码器."""
        return getattr(self, "_categorical_encoder", None)

    @categorical_encoder.setter
    def categorical_encoder(self, value):
        self._categorical_encoder = value

    @property
    def continuous_transform(self):
        """返回连续变换."""
        return getattr(self, "_continuous_transform", None)

    @continuous_transform.setter
    def continuous_transform(self, value):
        self._continuous_transform = value

    @property
    def scaler(self):
        """返回缩放器."""
        return getattr(self, "_scaler", None)

    @scaler.setter
    def scaler(self, value):
        self._scaler = value

    @property
    def label_encoder(self):
        """返回标签编码器."""
        return getattr(self, "_label_encoder", None)

    @label_encoder.setter
    def label_encoder(self, value):
        self._label_encoder = value

    @property
    def target_transforms(self):
        """返回目标变换."""
        if self.do_target_transform:
            return self._target_transforms
        else:
            return None

    @target_transforms.setter
    def target_transforms(self, value):
        self._target_transforms = value

    def _setup_cache(self, cache_data: Union[str, bool]) -> None:
        cache_data = cache_data.lower()
        if cache_data == self.CACHE_MODES.MEMORY.value:
            self.cache_mode = self.CACHE_MODES.MEMORY
        elif isinstance(cache_data, str):
            self.cache_mode = self.CACHE_MODES.DISK
            self.cache_dir = Path(cache_data)
            self.cache_dir.mkdir(parents=True, exist_ok=True)
        else:
            logger.warning(f"{cache_data} is not a valid path. Caching in memory")
            self.cache_mode = self.CACHE_MODES.MEMORY

    def _set_target_transform(self, target_transform: Union[TransformerMixin, Tuple]) -> None:
        if target_transform is not None:
            if isinstance(target_transform, Iterable):
                target_transform = FunctionTransformer(func=target_transform[0], inverse_func=target_transform[1])
            self.do_target_transform = True
        else:
            self.do_target_transform = False
        self.target_transform_template = target_transform

    def _update_config(self, config) -> InferredConfig:
        """计算并更新配置对象的一些关键信息.

Parameters:
    config (DictConfig): 配置对象

Returns:
    InferredConfig: 更新后的配置对象
"""
        categorical_dim = len(config.categorical_cols)
        continuous_dim = len(config.continuous_cols)
        # self._output_dim_clf = len(np.unique(self.train_dataset.y)) if config.target else None
        # self._output_dim_reg = len(config.target) if config.target else None
        if config.task == "regression":
            # self._output_dim_reg = len(config.target) if config.target else None if self.train is not None:
            output_dim = len(config.target) if config.target else None
            output_cardinality = None
        elif config.task == "classification":
            # self._output_dim_clf = len(np.unique(self.train_dataset.y)) if config.target else None
            if self.train is not None:
                output_cardinality = (
                    self.train[config.target].fillna("NA").nunique().tolist() if config.target else None
                )
                output_dim = sum(output_cardinality)
            else:
                output_cardinality = (
                    self.train_dataset.data[config.target].fillna("NA").nunique().tolist() if config.target else None
                )
                output_dim = sum(output_cardinality)
        elif config.task == "ssl":
            output_cardinality = None
            output_dim = None
        else:
            raise ValueError(f"{config.task} is an unsupported task.")
        if self.train is not None:
            categorical_cardinality = [
                int(x) + 1 for x in list(self.train[config.categorical_cols].fillna("NA").nunique().values)
            ]
        else:
            categorical_cardinality = [
                int(x) + 1 for x in list(self.train_dataset.data[config.categorical_cols].nunique().values)
            ]
        if getattr(config, "embedding_dims", None) is not None:
            embedding_dims = config.embedding_dims
        else:
            embedding_dims = [(x, min(50, (x + 1) // 2)) for x in categorical_cardinality]
        return InferredConfig(
            categorical_dim=categorical_dim,
            continuous_dim=continuous_dim,
            output_dim=output_dim,
            output_cardinality=output_cardinality,
            categorical_cardinality=categorical_cardinality,
            embedding_dims=embedding_dims,
        )

    def update_config(self, config) -> InferredConfig:
        """计算并更新配置对象的一些关键信息.逻辑在_update_config中实现.这只是为了使其可以从外部访问,并且不破坏当前的API.

Parameters:
    config (DictConfig): 配置对象

Returns:
    InferredConfig: 更新后的配置对象
"""
        if self.cache_mode is self.CACHE_MODES.INFERENCE:
            warnings.warn("Cannot update config in inference mode. Returning the cached config")
            return self._inferred_config
        else:
            return self._update_config(config)

    def _encode_date_columns(self, data: DataFrame) -> DataFrame:
        added_features = []
        for field_name, freq, format in self.config.date_columns:
            data = self.make_date(data, field_name, format)
            data, _new_feats = self.add_datepart(data, field_name, frequency=freq, prefix=None, drop=True)
            added_features += _new_feats
        return data, added_features

    def _encode_categorical_columns(self, data: DataFrame, stage: str) -> DataFrame:
        if stage != "fit":
            # Inference
            return self.categorical_encoder.transform(data)
        # Fit
        logger.debug("Encoding Categorical Columns using OrdinalEncoder")
        self.categorical_encoder = OrdinalEncoder(
            cols=self.config.categorical_cols,
            handle_unseen=("impute" if self.config.handle_unknown_categories else "error"),
            handle_missing="impute" if self.config.handle_missing_values else "error",
        )
        data = self.categorical_encoder.fit_transform(data)
        return data

    def _transform_continuous_columns(self, data: DataFrame, stage: str) -> DataFrame:
        if stage == "fit":
            transform = self.CONTINUOUS_TRANSFORMS[self.config.continuous_feature_transform]
            if "random_state" in transform["params"] and self.seed is not None:
                transform["params"]["random_state"] = self.seed
            self.continuous_transform = transform["callable"](**transform["params"])
            # can be accessed through property continuous_transform
            data.loc[:, self.config.continuous_cols] = self.continuous_transform.fit_transform(
                data.loc[:, self.config.continuous_cols]
            )
        else:
            data.loc[:, self.config.continuous_cols] = self.continuous_transform.transform(
                data.loc[:, self.config.continuous_cols]
            )
        return data

    def _normalize_continuous_columns(self, data: DataFrame, stage: str) -> DataFrame:
        if stage == "fit":
            self.scaler = StandardScaler()
            data.loc[:, self.config.continuous_cols] = self.scaler.fit_transform(
                data.loc[:, self.config.continuous_cols]
            )
        else:
            data.loc[:, self.config.continuous_cols] = self.scaler.transform(data.loc[:, self.config.continuous_cols])
        return data

    def _label_encode_target(self, data: DataFrame, stage: str) -> DataFrame:
        if self.config.task != "classification":
            return data
        if stage == "fit" or self.label_encoder is None:
            self.label_encoder = [None] * len(self.config.target)
            for i in range(len(self.config.target)):
                self.label_encoder[i] = LabelEncoder()
                data[self.config.target[i]] = self.label_encoder[i].fit_transform(data[self.config.target[i]])
        else:
            for i in range(len(self.config.target)):
                if self.config.target[i] in data.columns:
                    data[self.config.target[i]] = self.label_encoder[i].transform(data[self.config.target[i]])
        return data

    def _target_transform(self, data: DataFrame, stage: str) -> DataFrame:
        if self.config.task != "regression":
            return data
        # target transform only for regression
        if not all(col in data.columns for col in self.config.target):
            return data
        if self.do_target_transform:
            if stage == "fit" or self.target_transforms is None:
                target_transforms = []
                for col in self.config.target:
                    _target_transform = copy.deepcopy(self.target_transform_template)
                    data[col] = _target_transform.fit_transform(data[col].values.reshape(-1, 1))
                    target_transforms.append(_target_transform)
                self.target_transforms = target_transforms
            else:
                for col, _target_transform in zip(self.config.target, self.target_transforms):
                    data[col] = _target_transform.transform(data[col].values.reshape(-1, 1))
        return data

    def preprocess_data(self, data: DataFrame, stage: str = "inference") -> Tuple[DataFrame, list]:
        """    The preprocessing, like Categorical Encoding, Normalization, etc. which any dataframe should undergo before
feeding into the dataloder.

Parameters:
    data (DataFrame): 包含特征和目标的数据框
    stage (str, optional): 内部参数.用于区分训练和推理阶段.
        默认为 "inference".

Returns:
    返回处理后的数据框和添加的特征（列表）作为元组
"""
        added_features = None
        if self.config.encode_date_columns:
            data, added_features = self._encode_date_columns(data)
        # The only features that are added are the date features extracted
        # from the date which are categorical in nature
        if (added_features is not None) and (stage == "fit"):
            logger.debug(f"Added {added_features} features after encoding the date_columns")
            self.config.categorical_cols += added_features
            # Update the categorical dimension in config
            self.config.categorical_dim = (
                len(self.config.categorical_cols) if self.config.categorical_cols is not None else 0
            )
            self._inferred_config = self._update_config(self.config)
        # Encoding Categorical Columns
        if len(self.config.categorical_cols) > 0:
            data = self._encode_categorical_columns(data, stage)

        # Transforming Continuous Columns
        if (self.config.continuous_feature_transform is not None) and (len(self.config.continuous_cols) > 0):
            data = self._transform_continuous_columns(data, stage)
        # Normalizing Continuous Columns
        if (self.config.normalize_continuous_features) and (len(self.config.continuous_cols) > 0):
            data = self._normalize_continuous_columns(data, stage)
        # Converting target labels to a 0 indexed label
        data = self._label_encode_target(data, stage)
        # Target Transforms
        data = self._target_transform(data, stage)
        return data, added_features

    def _cache_dataset(self):
        train_dataset = TabularDataset(
            task=self.config.task,
            data=self.train,
            categorical_cols=self.config.categorical_cols,
            continuous_cols=self.config.continuous_cols,
            target=self.target,
        )
        self.train = None
        validation_dataset = TabularDataset(
            task=self.config.task,
            data=self.validation,
            categorical_cols=self.config.categorical_cols,
            continuous_cols=self.config.continuous_cols,
            target=self.target,
        )
        self.validation = None

        if self.cache_mode is self.CACHE_MODES.DISK:
            torch.save(train_dataset, self.cache_dir / "train_dataset")
            torch.save(validation_dataset, self.cache_dir / "validation_dataset")
        elif self.cache_mode is self.CACHE_MODES.MEMORY:
            self.train_dataset = train_dataset
            self.validation_dataset = validation_dataset
        elif self.cache_mode is self.CACHE_MODES.INFERENCE:
            self.train_dataset = None
            self.validation_dataset = None
        else:
            raise ValueError(f"{self.cache_mode} is not a valid cache mode")

    def split_train_val(self, train):
        logger.debug(
            "No validation data provided." f" Using {self.config.validation_split*100}% of train data as validation"
        )
        val_idx = train.sample(
            int(self.config.validation_split * len(train)),
            random_state=self.seed,
        ).index
        validation = train[train.index.isin(val_idx)]
        train = train[~train.index.isin(val_idx)]
        return train, validation

    def setup(self, stage: Optional[str] = None) -> None:
        """    要在所有GPU上执行的数据操作,如训练-测试拆分、转换等.在访问数据加载器之前调用此操作.

Parameters:
    stage (Optional[str], 可选):
        用于区分训练和推理的内部参数.默认为 None.
"""
        if not (stage is None or stage == "fit" or stage == "ssl_finetune"):
            return
        if self.verbose:
            logger.info(f"Setting up the datamodule for {self.config.task} task")
        is_ssl = stage == "ssl_finetune"
        if self.validation is None:
            self.train, self.validation = self.split_train_val(self.train)
        else:
            self.validation = self.validation.copy()
        # Preprocessing Train, Validation
        self.train, _ = self.preprocess_data(self.train, stage="fit" if not is_ssl else "inference")
        self.validation, _ = self.preprocess_data(self.validation, stage="inference")
        self._fitted = True
        self._cache_dataset()

    def inference_only_copy(self):
        """    创建一个数据模块的副本,移除了训练集和验证集.这对于仅推理的场景非常有用,因为我们不希望保存训练集和验证集.

Returns:
    TabularDatamodule: 移除了训练集和验证集的数据模块副本.
"""
        if not self._fitted:
            raise RuntimeError("Can create an inference only copy only after model is fitted")
        dm_inference = copy.copy(self)
        dm_inference.train_dataset = None
        dm_inference.validation_dataset = None
        dm_inference.cache_mode = self.CACHE_MODES.INFERENCE
        return dm_inference

    # adapted from gluonts
    @classmethod
    def time_features_from_frequency_str(cls, freq_str: str) -> List[str]:
        """    返回一个适合给定频率字符串的时间特征列表.

Parameters:
    freq_str (str): 频率字符串,格式为 `[倍数][粒度]`,例如 "12H", "5min", "1D" 等.

Returns:
    添加的特征列表
"""

        features_by_offsets = {
            offsets.YearBegin: [],
            offsets.YearEnd: [],
            offsets.MonthBegin: [
                "Month",
                "Quarter",
                "Is_quarter_end",
                "Is_quarter_start",
                "Is_year_end",
                "Is_year_start",
            ],
            offsets.MonthEnd: [
                "Month",
                "Quarter",
                "Is_quarter_end",
                "Is_quarter_start",
                "Is_year_end",
                "Is_year_start",
            ],
            offsets.Week: [
                "Month",
                "Quarter",
                "Is_quarter_end",
                "Is_quarter_start",
                "Is_year_end",
                "Is_year_start",
                "Is_month_start",
                "Week",
            ],
            offsets.Day: [
                "Month",
                "Quarter",
                "Is_quarter_end",
                "Is_quarter_start",
                "Is_year_end",
                "Is_year_start",
                "Is_month_start",
                "WeekDay",
                "Dayofweek",
                "Dayofyear",
            ],
            offsets.BusinessDay: [
                "Month",
                "Quarter",
                "Is_quarter_end",
                "Is_quarter_start",
                "Is_year_end",
                "Is_year_start",
                "Is_month_start",
                "WeekDay",
                "Dayofweek",
                "Dayofyear",
            ],
            offsets.Hour: [
                "Month",
                "Quarter",
                "Is_quarter_end",
                "Is_quarter_start",
                "Is_year_end",
                "Is_year_start",
                "Is_month_start",
                "WeekDay",
                "Dayofweek",
                "Dayofyear",
                "Hour",
            ],
            offsets.Minute: [
                "Month",
                "Quarter",
                "Is_quarter_end",
                "Is_quarter_start",
                "Is_year_end",
                "Is_year_start",
                "Is_month_start",
                "WeekDay",
                "Dayofweek",
                "Dayofyear",
                "Hour",
                "Minute",
            ],
        }

        offset = to_offset(freq_str)

        for offset_type, feature in features_by_offsets.items():
            if isinstance(offset, offset_type):
                return feature

        supported_freq_msg = f"""
        Unsupported frequency {freq_str}

        The following frequencies are supported:

            Y, YS   - yearly
                alias: A
            M, MS   - monthly
            W   - weekly
            D   - daily
            B   - business days
            H   - hourly
            T   - minutely
                alias: min
        """
        raise RuntimeError(supported_freq_msg)

    # adapted from fastai
    @classmethod
    def make_date(cls, df: DataFrame, date_field: str, date_format: str = "ISO8601") -> DataFrame:
        """确保 `df[date_field]` 具有正确的日期类型.

Parameters:
    df (DataFrame): 数据框

    date_field (str): 日期字段名称

Returns:
    日期字段已转换为日期时间的数据框
"""
        field_dtype = df[date_field].dtype
        if isinstance(field_dtype, DatetimeTZDtype):
            field_dtype = np.datetime64
        if not np.issubdtype(field_dtype, np.datetime64):
            df[date_field] = to_datetime(df[date_field], format=date_format)
        return df

    # adapted from fastai
    @classmethod
    def add_datepart(
        cls,
        df: DataFrame,
        field_name: str,
        frequency: str,
        prefix: str = None,
        drop: bool = True,
    ) -> Tuple[DataFrame, List[str]]:
        """    用于在`df`的`field_name`列中添加与日期相关的列的辅助函数.

Parameters:
    df (DataFrame): 数据框

    field_name (str): 日期字段名称

    frequency (str): 频率字符串,格式为`[倍数][粒度]`,例如"12H", "5min", "1D"等.

    prefix (str, 可选): 添加到新列的前缀.默认为None.

    drop (bool, 可选): 是否删除原始列.默认为True.

Returns:
    添加了新列的数据框和新增列的列表
"""
        field = df[field_name]
        prefix = (re.sub("[Dd]ate$", "", field_name) if prefix is None else prefix) + "_"
        attr = cls.time_features_from_frequency_str(frequency)
        added_features = []
        for n in attr:
            if n == "Week":
                continue
            df[prefix + n] = getattr(field.dt, n.lower())
            added_features.append(prefix + n)
        # Pandas removed `dt.week` in v1.1.10
        if "Week" in attr:
            week = field.dt.isocalendar().week if hasattr(field.dt, "isocalendar") else field.dt.week
            df.insert(3, prefix + "Week", week)
            added_features.append(prefix + "Week")
        # TODO Not adding Elapsed by default. Need to route it through config
        # mask = ~field.isna()
        # df[prefix + "Elapsed"] = np.where(
        #     mask, field.values.astype(np.int64) // 10 ** 9, None
        # )
        # added_features.append(prefix + "Elapsed")
        if drop:
            df.drop(field_name, axis=1, inplace=True)

        # Removing features woth zero variations
        # for col in added_features:
        #     if len(df[col].unique()) == 1:
        #         df.drop(columns=col, inplace=True)
        #         added_features.remove(col)
        return df, added_features

    def _load_dataset_from_cache(self, tag: str = "train"):
        if self.cache_mode is self.CACHE_MODES.MEMORY:
            try:
                dataset = getattr(self, f"_{tag}_dataset")
            except AttributeError:
                raise AttributeError(
                    f"{tag}_dataset not found in memory. Please provide the data for" f" {tag} dataloader"
                )
        elif self.cache_mode is self.CACHE_MODES.DISK:
            try:
                dataset = torch.load(self.cache_dir / f"{tag}_dataset")
            except FileNotFoundError:
                raise FileNotFoundError(
                    f"{tag}_dataset not found in {self.cache_dir}. Please provide the" f" data for {tag} dataloader"
                )
        elif self.cache_mode is self.CACHE_MODES.INFERENCE:
            raise RuntimeError("Cannot load dataset in inference mode. Use" " `prepare_inference_dataloader` instead")
        else:
            raise ValueError(f"{self.cache_mode} is not a valid cache mode")
        return dataset

    @property
    def train_dataset(self) -> TabularDataset:
        """返回训练数据集.

Returns:
    TabularDataset: 训练数据集
"""
        return self._load_dataset_from_cache("train")

    @train_dataset.setter
    def train_dataset(self, value):
        self._train_dataset = value

    @property
    def validation_dataset(self) -> TabularDataset:
        """返回验证数据集.

Returns:
    TabularDataset: 验证数据集
"""
        return self._load_dataset_from_cache("validation")

    @validation_dataset.setter
    def validation_dataset(self, value):
        self._validation_dataset = value

    def train_dataloader(self, batch_size: Optional[int] = None) -> DataLoader:
        """加载训练集的函数.

Parameters:
    batch_size (Optional[int], 可选): 批量大小.默认为 `self.batch_size`.

Returns:
    DataLoader: 训练数据加载器
"""
        return DataLoader(
            self.train_dataset,
            batch_size or self.batch_size,
            shuffle=True if self.train_sampler is None else False,
            num_workers=self.config.num_workers,
            sampler=self.train_sampler,
            pin_memory=self.config.pin_memory,
        )

    def val_dataloader(self, batch_size: Optional[int] = None) -> DataLoader:
        """加载验证集的函数.

Parameters:
    batch_size (Optional[int], 可选): 批量大小.默认为 `self.batch_size`.

Returns:
    DataLoader: 验证数据加载器
"""
        return DataLoader(
            self.validation_dataset,
            batch_size or self.batch_size,
            shuffle=False,
            num_workers=self.config.num_workers,
            pin_memory=self.config.pin_memory,
        )

    def _prepare_inference_data(self, df: DataFrame) -> DataFrame:
        """    准备数据进行推理."""
        # TODO Is the target encoding necessary?
        if len(set(self.target) - set(df.columns)) > 0:
            if self.config.task == "classification":
                for i in range(len(self.target)):
                    df.loc[:, self.target[i]] = np.array([self.label_encoder[i].classes_[0]] * len(df)).reshape(-1, 1)
            else:
                df.loc[:, self.target] = np.zeros((len(df), len(self.target)))
        df, _ = self.preprocess_data(df, stage="inference")
        return df

    def prepare_inference_dataloader(
        self, df: DataFrame, batch_size: Optional[int] = None, copy_df: bool = True
    ) -> DataLoader:
        """函数用于准备并加载新数据.

Parameters:
    df (DataFrame): 包含特征和目标的数据框
    batch_size (Optional[int], 可选): 批量大小.默认为 `self.batch_size`.
    copy_df (bool, 可选): 是否在处理前复制数据框.默认为 False.
Returns:
    DataLoader: 传入数据框的数据加载器
"""
        if copy_df:
            df = df.copy()
        df = self._prepare_inference_data(df)
        dataset = TabularDataset(
            task=self.config.task,
            data=df,
            categorical_cols=self.config.categorical_cols,
            continuous_cols=self.config.continuous_cols,
            target=(self.target if all(col in df.columns for col in self.target) else None),
        )
        return DataLoader(
            dataset,
            batch_size or self.batch_size,
            shuffle=False,
            num_workers=self.config.num_workers,
        )

    def save_dataloader(self, path: Union[str, Path]) -> None:
        """保存数据加载器到指定路径.

Parameters:
    path (Union[str, Path]): 保存数据加载器的路径
"""
        if isinstance(path, str):
            path = Path(path)
        joblib.dump(self, path)

    @classmethod
    def load_datamodule(cls, path: Union[str, Path]):
        """加载一个数据模块从指定路径.

Parameters:
    path (Union[str, Path]): 数据模块的路径

Returns:
    TabularDatamodule (TabularDatamodule): 从路径加载的数据模块
"""
        if isinstance(path, str):
            path = Path(path)
        if not path.exists():
            raise FileNotFoundError(f"{path} does not exist.")
        datamodule = joblib.load(path)
        return datamodule

    def copy(
        self,
        train: DataFrame,
        validation: DataFrame = None,
        target_transform: Optional[Union[TransformerMixin, Tuple]] = None,
        train_sampler: Optional[torch.utils.data.Sampler] = None,
        seed: Optional[int] = None,
        cache_data: str = None,
        copy_data: bool = None,
        verbose: bool = None,
        call_setup: bool = True,
        config_override: Optional[Dict] = {},
    ):
        if config_override is not None:
            for k, v in config_override.items():
                setattr(self.config, k, v)
        dm = TabularDatamodule(
            train=train,
            config=self.config,
            validation=validation,
            target_transform=target_transform or self.target_transforms,
            train_sampler=train_sampler or self.train_sampler,
            seed=seed or self.seed,
            cache_data=cache_data or self.cache_mode.value,
            copy_data=copy_data or True,
            verbose=verbose or self.verbose,
        )
        dm.categorical_encoder = self.categorical_encoder
        dm.continuous_transform = self.continuous_transform
        dm.scaler = self.scaler
        dm.label_encoder = self.label_encoder
        dm.target_transforms = self.target_transforms
        dm.setup(stage="ssl_finetune") if call_setup else None
        return dm

categorical_encoder property writable

返回分类编码器.

continuous_transform property writable

返回连续变换.

label_encoder property writable

返回标签编码器.

scaler property writable

返回缩放器.

target_transforms property writable

返回目标变换.

train_dataset: TabularDataset property writable

返回训练数据集.

Returns:

Name	Type	Description
TabularDataset	TabularDataset	训练数据集

validation_dataset: TabularDataset property writable

返回验证数据集.

Returns:

Name	Type	Description
TabularDataset	TabularDataset	验证数据集

__init__(train, config, validation=None, target_transform=None, train_sampler=None, seed=42, cache_data='memory', copy_data=True, verbose=True)

Pytorch Lightning 用于表格数据的 Datamodule.

Parameters:

Name	Type	Description	Default
train	DataFrame	训练数据框	required
config	DictConfig	从 ModelConfig、DataConfig、 TrainerConfig、OptimizerConfig 和 ExperimentConfig 合并的配置对象	required
validation	(DataFrame, 可选)	验证数据框. 如果留空,我们将使用 DataConfig 中的验证分割来随机抽取样本作为验证. 默认为 None.	None
target_transform	(Optional[Union[TransformerMixin, Tuple(Callable)]], 可选)	如果提供,将在建模前对目标应用变换,并在预测时进行逆变换.参数可以是具有 inverse_transform 方法的 sklearn Transformer, 或由可调用对象组成的元组 (transform_func, inverse_transform_func) 默认为 None.	None
train_sampler	(Optional[Sampler], 可选)	如果提供,将使用该采样器对训练数据进行采样.默认为 None.	None
seed	(Optional[int], 可选)	用于可重复数据加载器的种子.默认为 42.	42
cache_data	str	决定如何在数据加载器中缓存数据.如果设置为 "memory",将在内存中缓存.如果设置为有效路径,将在该路径中缓存.默认为 "memory".	'memory'
copy_data	bool	如果为 True,将在预处理前复制数据框.默认为 True.	True
verbose	bool	设置 databodule 日志的详细程度	True

Source code in src/pytorch_tabular/tabular_datamodule.py

    def __init__(
        self,
        train: DataFrame,
        config: DictConfig,
        validation: DataFrame = None,
        target_transform: Optional[Union[TransformerMixin, Tuple]] = None,
        train_sampler: Optional[torch.utils.data.Sampler] = None,
        seed: Optional[int] = 42,
        cache_data: str = "memory",
        copy_data: bool = True,
        verbose: bool = True,
    ):
        """    Pytorch Lightning 用于表格数据的 Datamodule.

Parameters:
    train (DataFrame): 训练数据框

    config (DictConfig): 从 ModelConfig、DataConfig、
        TrainerConfig、OptimizerConfig 和 ExperimentConfig 合并的配置对象

    validation (DataFrame, 可选): 验证数据框.
        如果留空,我们将使用 DataConfig 中的验证分割来随机抽取样本作为验证.
        默认为 None.

    target_transform (Optional[Union[TransformerMixin, Tuple(Callable)]], 可选):
        如果提供,将在建模前对目标应用变换,并在预测时进行逆变换.参数可以是具有 inverse_transform 方法的 sklearn Transformer,
        或由可调用对象组成的元组 (transform_func, inverse_transform_func)
        默认为 None.

    train_sampler (Optional[torch.utils.data.Sampler], 可选):
        如果提供,将使用该采样器对训练数据进行采样.默认为 None.

    seed (Optional[int], 可选): 用于可重复数据加载器的种子.默认为 42.

    cache_data (str): 决定如何在数据加载器中缓存数据.如果设置为
        "memory",将在内存中缓存.如果设置为有效路径,将在该路径中缓存.默认为 "memory".

    copy_data (bool): 如果为 True,将在预处理前复制数据框.默认为 True.

    verbose (bool): 设置 databodule 日志的详细程度
"""
        super().__init__()
        self.train = train.copy() if copy_data else train
        if validation is not None:
            self.validation = validation.copy() if copy_data else validation
        else:
            self.validation = None
        self._set_target_transform(target_transform)
        self.target = config.target or []
        self.batch_size = config.batch_size
        self.train_sampler = train_sampler
        self.config = config
        self.seed = seed
        self.verbose = verbose
        self._fitted = False
        self._setup_cache(cache_data)
        self._inferred_config = self._update_config(config)

add_datepart(df, field_name, frequency, prefix=None, drop=True) classmethod

用于在df的field_name列中添加与日期相关的列的辅助函数.

Parameters:

Name	Type	Description	Default
df	DataFrame	数据框	required
field_name	str	日期字段名称	required
frequency	str	频率字符串,格式为[倍数][粒度],例如"12H", "5min", "1D"等.	required
prefix	(str, 可选)	添加到新列的前缀.默认为None.	None
drop	(bool, 可选)	是否删除原始列.默认为True.	True

Returns:

Type	Description
Tuple[DataFrame, List[str]]	添加了新列的数据框和新增列的列表

Source code in src/pytorch_tabular/tabular_datamodule.py

    @classmethod
    def add_datepart(
        cls,
        df: DataFrame,
        field_name: str,
        frequency: str,
        prefix: str = None,
        drop: bool = True,
    ) -> Tuple[DataFrame, List[str]]:
        """    用于在`df`的`field_name`列中添加与日期相关的列的辅助函数.

Parameters:
    df (DataFrame): 数据框

    field_name (str): 日期字段名称

    frequency (str): 频率字符串,格式为`[倍数][粒度]`,例如"12H", "5min", "1D"等.

    prefix (str, 可选): 添加到新列的前缀.默认为None.

    drop (bool, 可选): 是否删除原始列.默认为True.

Returns:
    添加了新列的数据框和新增列的列表
"""
        field = df[field_name]
        prefix = (re.sub("[Dd]ate$", "", field_name) if prefix is None else prefix) + "_"
        attr = cls.time_features_from_frequency_str(frequency)
        added_features = []
        for n in attr:
            if n == "Week":
                continue
            df[prefix + n] = getattr(field.dt, n.lower())
            added_features.append(prefix + n)
        # Pandas removed `dt.week` in v1.1.10
        if "Week" in attr:
            week = field.dt.isocalendar().week if hasattr(field.dt, "isocalendar") else field.dt.week
            df.insert(3, prefix + "Week", week)
            added_features.append(prefix + "Week")
        # TODO Not adding Elapsed by default. Need to route it through config
        # mask = ~field.isna()
        # df[prefix + "Elapsed"] = np.where(
        #     mask, field.values.astype(np.int64) // 10 ** 9, None
        # )
        # added_features.append(prefix + "Elapsed")
        if drop:
            df.drop(field_name, axis=1, inplace=True)

        # Removing features woth zero variations
        # for col in added_features:
        #     if len(df[col].unique()) == 1:
        #         df.drop(columns=col, inplace=True)
        #         added_features.remove(col)
        return df, added_features

inference_only_copy()

创建一个数据模块的副本,移除了训练集和验证集.这对于仅推理的场景非常有用,因为我们不希望保存训练集和验证集.

Returns:

Name	Type	Description
TabularDatamodule		移除了训练集和验证集的数据模块副本.

Source code in src/pytorch_tabular/tabular_datamodule.py

    def inference_only_copy(self):
        """    创建一个数据模块的副本,移除了训练集和验证集.这对于仅推理的场景非常有用,因为我们不希望保存训练集和验证集.

Returns:
    TabularDatamodule: 移除了训练集和验证集的数据模块副本.
"""
        if not self._fitted:
            raise RuntimeError("Can create an inference only copy only after model is fitted")
        dm_inference = copy.copy(self)
        dm_inference.train_dataset = None
        dm_inference.validation_dataset = None
        dm_inference.cache_mode = self.CACHE_MODES.INFERENCE
        return dm_inference

load_datamodule(path) classmethod

加载一个数据模块从指定路径.

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	数据模块的路径	required

Returns:

Name	Type	Description
TabularDatamodule	TabularDatamodule	从路径加载的数据模块

Source code in src/pytorch_tabular/tabular_datamodule.py

    @classmethod
    def load_datamodule(cls, path: Union[str, Path]):
        """加载一个数据模块从指定路径.

Parameters:
    path (Union[str, Path]): 数据模块的路径

Returns:
    TabularDatamodule (TabularDatamodule): 从路径加载的数据模块
"""
        if isinstance(path, str):
            path = Path(path)
        if not path.exists():
            raise FileNotFoundError(f"{path} does not exist.")
        datamodule = joblib.load(path)
        return datamodule

make_date(df, date_field, date_format='ISO8601') classmethod

确保 df[date_field] 具有正确的日期类型.

Parameters:

Name	Type	Description	Default
df	DataFrame	数据框	required
date_field	str	日期字段名称	required

Returns:

Type	Description
DataFrame	日期字段已转换为日期时间的数据框

Source code in src/pytorch_tabular/tabular_datamodule.py

    @classmethod
    def make_date(cls, df: DataFrame, date_field: str, date_format: str = "ISO8601") -> DataFrame:
        """确保 `df[date_field]` 具有正确的日期类型.

Parameters:
    df (DataFrame): 数据框

    date_field (str): 日期字段名称

Returns:
    日期字段已转换为日期时间的数据框
"""
        field_dtype = df[date_field].dtype
        if isinstance(field_dtype, DatetimeTZDtype):
            field_dtype = np.datetime64
        if not np.issubdtype(field_dtype, np.datetime64):
            df[date_field] = to_datetime(df[date_field], format=date_format)
        return df

prepare_inference_dataloader(df, batch_size=None, copy_df=True)

函数用于准备并加载新数据.

Parameters:

Name	Type	Description	Default
df	DataFrame	包含特征和目标的数据框	required
batch_size	(Optional[int], 可选)	批量大小.默认为 self.batch_size.	None
copy_df	(bool, 可选)	是否在处理前复制数据框.默认为 False.	True

Returns: DataLoader: 传入数据框的数据加载器

Source code in src/pytorch_tabular/tabular_datamodule.py

    def prepare_inference_dataloader(
        self, df: DataFrame, batch_size: Optional[int] = None, copy_df: bool = True
    ) -> DataLoader:
        """函数用于准备并加载新数据.

Parameters:
    df (DataFrame): 包含特征和目标的数据框
    batch_size (Optional[int], 可选): 批量大小.默认为 `self.batch_size`.
    copy_df (bool, 可选): 是否在处理前复制数据框.默认为 False.
Returns:
    DataLoader: 传入数据框的数据加载器
"""
        if copy_df:
            df = df.copy()
        df = self._prepare_inference_data(df)
        dataset = TabularDataset(
            task=self.config.task,
            data=df,
            categorical_cols=self.config.categorical_cols,
            continuous_cols=self.config.continuous_cols,
            target=(self.target if all(col in df.columns for col in self.target) else None),
        )
        return DataLoader(
            dataset,
            batch_size or self.batch_size,
            shuffle=False,
            num_workers=self.config.num_workers,
        )

preprocess_data(data, stage='inference')

The preprocessing, like Categorical Encoding, Normalization, etc. which any dataframe should undergo before feeding into the dataloder.

Parameters:

Name	Type	Description	Default
data	DataFrame	包含特征和目标的数据框	required
stage	str	内部参数.用于区分训练和推理阶段. 默认为 "inference".	'inference'

Returns:

Type	Description
Tuple[DataFrame, list]	返回处理后的数据框和添加的特征（列表）作为元组

Source code in src/pytorch_tabular/tabular_datamodule.py

    def preprocess_data(self, data: DataFrame, stage: str = "inference") -> Tuple[DataFrame, list]:
        """    The preprocessing, like Categorical Encoding, Normalization, etc. which any dataframe should undergo before
feeding into the dataloder.

Parameters:
    data (DataFrame): 包含特征和目标的数据框
    stage (str, optional): 内部参数.用于区分训练和推理阶段.
        默认为 "inference".

Returns:
    返回处理后的数据框和添加的特征（列表）作为元组
"""
        added_features = None
        if self.config.encode_date_columns:
            data, added_features = self._encode_date_columns(data)
        # The only features that are added are the date features extracted
        # from the date which are categorical in nature
        if (added_features is not None) and (stage == "fit"):
            logger.debug(f"Added {added_features} features after encoding the date_columns")
            self.config.categorical_cols += added_features
            # Update the categorical dimension in config
            self.config.categorical_dim = (
                len(self.config.categorical_cols) if self.config.categorical_cols is not None else 0
            )
            self._inferred_config = self._update_config(self.config)
        # Encoding Categorical Columns
        if len(self.config.categorical_cols) > 0:
            data = self._encode_categorical_columns(data, stage)

        # Transforming Continuous Columns
        if (self.config.continuous_feature_transform is not None) and (len(self.config.continuous_cols) > 0):
            data = self._transform_continuous_columns(data, stage)
        # Normalizing Continuous Columns
        if (self.config.normalize_continuous_features) and (len(self.config.continuous_cols) > 0):
            data = self._normalize_continuous_columns(data, stage)
        # Converting target labels to a 0 indexed label
        data = self._label_encode_target(data, stage)
        # Target Transforms
        data = self._target_transform(data, stage)
        return data, added_features

save_dataloader(path)

保存数据加载器到指定路径.

Parameters:

Name	Type	Description	Default
path	Union[str, Path]	保存数据加载器的路径	required

Source code in src/pytorch_tabular/tabular_datamodule.py

    def save_dataloader(self, path: Union[str, Path]) -> None:
        """保存数据加载器到指定路径.

Parameters:
    path (Union[str, Path]): 保存数据加载器的路径
"""
        if isinstance(path, str):
            path = Path(path)
        joblib.dump(self, path)

setup(stage=None)

要在所有GPU上执行的数据操作,如训练-测试拆分、转换等.在访问数据加载器之前调用此操作.

Parameters:

Name	Type	Description	Default
stage	(Optional[str], 可选)	用于区分训练和推理的内部参数.默认为 None.	None

Source code in src/pytorch_tabular/tabular_datamodule.py

    def setup(self, stage: Optional[str] = None) -> None:
        """    要在所有GPU上执行的数据操作,如训练-测试拆分、转换等.在访问数据加载器之前调用此操作.

Parameters:
    stage (Optional[str], 可选):
        用于区分训练和推理的内部参数.默认为 None.
"""
        if not (stage is None or stage == "fit" or stage == "ssl_finetune"):
            return
        if self.verbose:
            logger.info(f"Setting up the datamodule for {self.config.task} task")
        is_ssl = stage == "ssl_finetune"
        if self.validation is None:
            self.train, self.validation = self.split_train_val(self.train)
        else:
            self.validation = self.validation.copy()
        # Preprocessing Train, Validation
        self.train, _ = self.preprocess_data(self.train, stage="fit" if not is_ssl else "inference")
        self.validation, _ = self.preprocess_data(self.validation, stage="inference")
        self._fitted = True
        self._cache_dataset()

time_features_from_frequency_str(freq_str) classmethod

返回一个适合给定频率字符串的时间特征列表.

Parameters:

Name	Type	Description	Default
freq_str	str	频率字符串,格式为 [倍数][粒度],例如 "12H", "5min", "1D" 等.	required

Returns:

Type	Description
List[str]	添加的特征列表

Source code in src/pytorch_tabular/tabular_datamodule.py

    @classmethod
    def time_features_from_frequency_str(cls, freq_str: str) -> List[str]:
        """    返回一个适合给定频率字符串的时间特征列表.

Parameters:
    freq_str (str): 频率字符串,格式为 `[倍数][粒度]`,例如 "12H", "5min", "1D" 等.

Returns:
    添加的特征列表
"""

        features_by_offsets = {
            offsets.YearBegin: [],
            offsets.YearEnd: [],
            offsets.MonthBegin: [
                "Month",
                "Quarter",
                "Is_quarter_end",
                "Is_quarter_start",
                "Is_year_end",
                "Is_year_start",
            ],
            offsets.MonthEnd: [
                "Month",
                "Quarter",
                "Is_quarter_end",
                "Is_quarter_start",
                "Is_year_end",
                "Is_year_start",
            ],
            offsets.Week: [
                "Month",
                "Quarter",
                "Is_quarter_end",
                "Is_quarter_start",
                "Is_year_end",
                "Is_year_start",
                "Is_month_start",
                "Week",
            ],
            offsets.Day: [
                "Month",
                "Quarter",
                "Is_quarter_end",
                "Is_quarter_start",
                "Is_year_end",
                "Is_year_start",
                "Is_month_start",
                "WeekDay",
                "Dayofweek",
                "Dayofyear",
            ],
            offsets.BusinessDay: [
                "Month",
                "Quarter",
                "Is_quarter_end",
                "Is_quarter_start",
                "Is_year_end",
                "Is_year_start",
                "Is_month_start",
                "WeekDay",
                "Dayofweek",
                "Dayofyear",
            ],
            offsets.Hour: [
                "Month",
                "Quarter",
                "Is_quarter_end",
                "Is_quarter_start",
                "Is_year_end",
                "Is_year_start",
                "Is_month_start",
                "WeekDay",
                "Dayofweek",
                "Dayofyear",
                "Hour",
            ],
            offsets.Minute: [
                "Month",
                "Quarter",
                "Is_quarter_end",
                "Is_quarter_start",
                "Is_year_end",
                "Is_year_start",
                "Is_month_start",
                "WeekDay",
                "Dayofweek",
                "Dayofyear",
                "Hour",
                "Minute",
            ],
        }

        offset = to_offset(freq_str)

        for offset_type, feature in features_by_offsets.items():
            if isinstance(offset, offset_type):
                return feature

        supported_freq_msg = f"""
        Unsupported frequency {freq_str}

        The following frequencies are supported:

            Y, YS   - yearly
                alias: A
            M, MS   - monthly
            W   - weekly
            D   - daily
            B   - business days
            H   - hourly
            T   - minutely
                alias: min
        """
        raise RuntimeError(supported_freq_msg)

train_dataloader(batch_size=None)

加载训练集的函数.

Parameters:

Name	Type	Description	Default
batch_size	(Optional[int], 可选)	批量大小.默认为 self.batch_size.	None

Returns:

Name	Type	Description
DataLoader	DataLoader	训练数据加载器

Source code in src/pytorch_tabular/tabular_datamodule.py

    def train_dataloader(self, batch_size: Optional[int] = None) -> DataLoader:
        """加载训练集的函数.

Parameters:
    batch_size (Optional[int], 可选): 批量大小.默认为 `self.batch_size`.

Returns:
    DataLoader: 训练数据加载器
"""
        return DataLoader(
            self.train_dataset,
            batch_size or self.batch_size,
            shuffle=True if self.train_sampler is None else False,
            num_workers=self.config.num_workers,
            sampler=self.train_sampler,
            pin_memory=self.config.pin_memory,
        )

update_config(config)

计算并更新配置对象的一些关键信息.逻辑在_update_config中实现.这只是为了使其可以从外部访问,并且不破坏当前的API.

Parameters:

Name	Type	Description	Default
config	DictConfig	配置对象	required

Returns:

Name	Type	Description
InferredConfig	InferredConfig	更新后的配置对象

Source code in src/pytorch_tabular/tabular_datamodule.py

    def update_config(self, config) -> InferredConfig:
        """计算并更新配置对象的一些关键信息.逻辑在_update_config中实现.这只是为了使其可以从外部访问,并且不破坏当前的API.

Parameters:
    config (DictConfig): 配置对象

Returns:
    InferredConfig: 更新后的配置对象
"""
        if self.cache_mode is self.CACHE_MODES.INFERENCE:
            warnings.warn("Cannot update config in inference mode. Returning the cached config")
            return self._inferred_config
        else:
            return self._update_config(config)

val_dataloader(batch_size=None)

加载验证集的函数.

Parameters:

Name	Type	Description	Default
batch_size	(Optional[int], 可选)	批量大小.默认为 self.batch_size.	None

Returns:

Name	Type	Description
DataLoader	DataLoader	验证数据加载器

Source code in src/pytorch_tabular/tabular_datamodule.py

    def val_dataloader(self, batch_size: Optional[int] = None) -> DataLoader:
        """加载验证集的函数.

Parameters:
    batch_size (Optional[int], 可选): 批量大小.默认为 `self.batch_size`.

Returns:
    DataLoader: 验证数据加载器
"""
        return DataLoader(
            self.validation_dataset,
            batch_size or self.batch_size,
            shuffle=False,
            num_workers=self.config.num_workers,
            pin_memory=self.config.pin_memory,
        )

表格模型调优器.

此类用于在给定搜索空间、策略和优化指标的情况下,调整表格模型的超参数.

Source code in src/pytorch_tabular/tabular_model_tuner.py

class TabularModelTuner:
    """表格模型调优器.

此类用于在给定搜索空间、策略和优化指标的情况下,调整表格模型的超参数."""

    ALLOWABLE_STRATEGIES = ["grid_search", "random_search"]
    OUTPUT = namedtuple("OUTPUT", ["trials_df", "best_params", "best_score", "best_model"])

    def __init__(
        self,
        data_config: Optional[Union[DataConfig, str]] = None,
        model_config: Optional[Union[ModelConfig, str]] = None,
        optimizer_config: Optional[Union[OptimizerConfig, str]] = None,
        trainer_config: Optional[Union[TrainerConfig, List[TrainerConfig]]] = None,
        model_callable: Optional[Callable] = None,
        model_state_dict_path: Optional[Union[str, Path]] = None,
        suppress_lightning_logger: bool = True,
        **kwargs,
    ):
        """表格模型调优器帮助您调整表格模型的超参数.

Parameters:
    data_config (可选[Union[DataConfig, str]], 可选): 表格模型的DataConfig.
        如果传递了str,将使用该路径中的yaml文件初始化DataConfig.
        默认为None.

    model_config (可选[Union[ModelConfig, List[TrainerConfig]]], 可选): 表格模型的ModelConfig.
        如果传递了str,将使用该路径中的yaml文件初始化ModelConfig.
        默认为None.

    optimizer_config (可选[Union[OptimizerConfig, str]], 可选): 表格模型的OptimizerConfig.
        如果传递了str,将使用该路径中的yaml文件初始化OptimizerConfig.
        默认为None.

    trainer_config (可选[Union[TrainerConfig, str]], 可选): 表格模型的TrainerConfig.
        如果传递了str,将使用该路径中的yaml文件初始化TrainerConfig.
        默认为None.

    model_callable (可选[Callable], 可选): 返回PyTorch表格模型的可调用对象.
        如果提供,将忽略model_config并使用此可调用对象初始化模型.
        默认为None.

    model_state_dict_path (可选[Union[str, Path]], 可选): 模型状态字典的路径.

        如果提供,将忽略model_config并使用此状态字典初始化模型.
        默认为None.

    suppress_lightning_logger (bool, 可选): 是否抑制lightning日志记录器.默认为True.

    **kwargs: 传递给TabularModel初始化的其他关键字参数.
"""
        if not isinstance(model_config, list):
            model_config = [model_config]

        if trainer_config.profiler is not None:
            warnings.warn(
                "Profiler is not supported in tuner. Set profiler=None in TrainerConfig to disable this warning."
            )
            trainer_config.profiler = None
        if trainer_config.fast_dev_run:
            warnings.warn("fast_dev_run is turned on. Tuning results won't be accurate.")
        if trainer_config.progress_bar != "none":
            # If config and tuner have progress bar enabled, it will result in a bug within the library (rich.progress)
            trainer_config.progress_bar = "none"
            warnings.warn("Turning off progress bar. Set progress_bar='none' in TrainerConfig to disable this warning.")
        trainer_config.trainer_kwargs.update({"enable_model_summary": False})
        self.data_config = data_config
        self.model_config = model_config
        self.optimizer_config = optimizer_config
        self.trainer_config = trainer_config
        self.suppress_lightning_logger = suppress_lightning_logger
        self.tabular_model_init_kwargs = {
            "model_callable": model_callable,
            "model_state_dict_path": model_state_dict_path,
            **kwargs,
        }

    def _check_assign_config(self, config, param, value):
        if isinstance(config, DictConfig):
            if param in config:
                config[param] = value
            else:
                raise ValueError(f"{param} is not a valid parameter for {str(config)}")
        elif isinstance(config, (ModelConfig, OptimizerConfig)):
            if hasattr(config, param):
                setattr(config, param, value)
            else:
                raise ValueError(f"{param} is not a valid parameter for {str(config)}")

    def _update_configs(
        self,
        optimizer_config: OptimizerConfig,
        model_config: ModelConfig,
        params: Dict,
    ):
        """更新配置以使用新参数."""
        # update configs with the new parameters
        for k, v in params.items():
            if k == "model":
                continue

            root, param = k.split("__")
            if root.startswith("trainer_config"):
                raise ValueError(
                    "The trainer_config is not supported by tuner. Please remove it from tuner parameters!"
                )
            elif root.startswith("optimizer_config"):
                self._check_assign_config(optimizer_config, param, v)
            elif root.startswith("model_config.head_config"):
                param = param.replace("model_config.head_config.", "")
                self._check_assign_config(model_config.head_config, param, v)
            elif root.startswith("model_config") and "head_config" not in root:
                self._check_assign_config(model_config, param, v)
            else:
                raise ValueError(
                    f"{k} is not in the proper format. Use __ to separate the "
                    "root and param. for eg. `optimizer_config__optimizer` should be "
                    "used to update the optimizer parameter in the optimizer_config"
                )
        return optimizer_config, model_config

    def tune(
        self,
        train: DataFrame,
        search_space: Union[Dict, List[Dict]],
        metric: Union[str, Callable],
        mode: str,
        strategy: str,
        validation: Optional[DataFrame] = None,
        n_trials: Optional[int] = None,
        cv: Optional[Union[int, Iterable, BaseCrossValidator]] = None,
        cv_agg_func: Optional[Callable] = np.mean,
        cv_kwargs: Optional[Dict] = {},
        return_best_model: bool = True,
        verbose: bool = False,
        progress_bar: bool = True,
        random_state: Optional[int] = 42,
        ignore_oom: bool = True,
        **kwargs,
    ):
        """调整TabularModel的超参数.

Parameters:
    train (DataFrame): 训练数据

    validation (DataFrame, 可选): 验证数据.默认为None.

    search_space (Dict): 一个字典,形式为{参数名: [要尝试的值]}用于网格搜索,或{参数名: 分布}用于随机搜索

    metric (Union[str, Callable]): 用于评估的指标.
        如果提供字符串,将使用定义的指标之一.
        如果提供可调用对象,将使用该函数作为指标.
        我们期望可调用对象的形式为`metric(y_true, y_pred)`.对于分类问题,`y_pred`是一个包含每个类别的概率（<class>_probability）和最终预测（prediction）的数据框.对于回归问题,它是一个包含最终预测（<target>_prediction）的数据框.
        默认为None.

    mode (str): 其中之一['max', 'min'].是否最大化或最小化指标.

    strategy (str): 其中之一['grid_search', 'random_search'].用于调整的策略.

    n_trials (int, 可选): 要运行的试验次数.仅用于随机搜索.
        默认为None.

    cv (Optional[Union[int, Iterable, BaseCrossValidator]]): 确定交叉验证的分割策略.
        可能的输入为:

        - None,不使用任何交叉验证.我们将仅使用验证数据
        - 整数,指定(Stratified)KFold中的折数,
        - 一个可迭代对象,生成(train, test)分割为索引数组.
        - 一个scikit-learn的CV分割器.
        默认为None.

    cv_agg_func (Optional[Callable], 可选): 用于聚合交叉验证分数的函数.
        默认为np.mean.

    cv_kwargs (Optional[Dict], 可选): 要传递给交叉验证方法的额外关键字参数.
        默认为{}.

    return_best_model (bool, 可选): 如果为True,将返回最佳模型.默认为True.

    verbose (bool, 可选): 是否打印每次试验的结果.默认为False.

    progress_bar (bool, 可选): 是否显示进度条.默认为True.

    random_state (Optional[int], 可选): 用于随机搜索的随机状态.默认为42.

    ignore_oom (bool, 可选): 是否忽略内存不足错误.默认为True.

    **kwargs: 要传递给TabularModel fit的额外关键字参数.

Returns:
    OUTPUT: 一个命名元组,包含以下属性:
        trials_df (DataFrame): 包含每次试验结果的数据框
        best_params (Dict): 找到的最佳参数
        best_score (float): 找到的最佳分数
        best_model (TabularModel 或 None): 如果return_best_model为True,返回best_model,否则返回None
"""
        assert strategy in self.ALLOWABLE_STRATEGIES, f"tuner must be one of {self.ALLOWABLE_STRATEGIES}"
        assert mode in ["max", "min"], "mode must be one of ['max', 'min']"
        assert metric is not None, "metric must be specified"
        assert (isinstance(search_space, dict) or (isinstance(search_space, list))) and len(
            search_space
        ) > 0, "search_space must be a non-empty dict"
        if self.suppress_lightning_logger:
            suppress_lightning_logs()
        if cv is not None and validation is not None:
            warnings.warn(
                "Both validation and cv are provided. Ignoring validation and using cv. Use "
                "`validation=None` to turn off this warning."
            )
            validation = None

        if not isinstance(search_space, list):
            search_space = [search_space]

        assert len(self.model_config) == len(search_space), "model_config and search_space must have the same length"

        verbose_tabular_model = self.tabular_model_init_kwargs.pop("verbose", False)

        with Progress() as progress:
            model_config_iterator = range(len(self.model_config))
            if progress_bar:
                model_config_iterator = progress.track(
                    model_config_iterator, description="[green]Running models config..."
                )

            datamodule = None
            trials = []
            best_model = None
            best_score = 0.0
            for idx in model_config_iterator:
                search_space_temp = {
                    **{"model": [f"{idx}-{self.model_config[idx].__class__.__name__}"]},
                    **search_space[idx],
                }

                if strategy == "grid_search":
                    assert all(
                        isinstance(v, list) for v in search_space_temp.values()
                    ), "For grid search, all values in search_space must be a list of values to try"
                    search_space_iterator = list(ParameterGrid(search_space_temp))
                    if n_trials is not None:
                        warnings.warn(
                            "n_trials is ignored for grid search to do a complete sweep of"
                            " the grid. Set n_trials=None to turn off this warning."
                        )
                    n_trials = sum(1 for _ in search_space_iterator)
                elif strategy == "random_search":
                    assert n_trials is not None, "n_trials must be specified for random search"
                    search_space_iterator = list(
                        ParameterSampler(search_space_temp, n_iter=n_trials, random_state=random_state)
                    )
                else:
                    raise NotImplementedError(f"{strategy} is not implemented yet.")

                # Sort by trainer_config to recreate the datamodule when necessary
                trainer_configs = [key for key in search_space_iterator if "trainer_config" in key]
                for key in trainer_configs:
                    search_space_iterator = sorted(
                        search_space_iterator, key=lambda search_space_iterator: search_space_iterator[key]
                    )

                if progress_bar:
                    search_space_iterator = progress.track(
                        search_space_iterator,
                        description=f"[blue]Training {idx}-{self.model_config[idx].__class__.__name__}...",
                    )

                if isinstance(metric, str):
                    is_callable_metric = False
                    metric_str = metric
                elif callable(metric):
                    is_callable_metric = True
                    metric_str = metric.__name__

                for i, params in enumerate(search_space_iterator):
                    # Copying the configs as a base
                    # Make sure all default parameters that you want to be set for all
                    # trials are in the original configs
                    trainer_config_t = deepcopy(self.trainer_config)
                    optimizer_config_t = deepcopy(self.optimizer_config)
                    model_config_t = deepcopy(self.model_config[idx])

                    optimizer_config_t, model_config_t = self._update_configs(
                        optimizer_config_t, model_config_t, params
                    )
                    # Initialize Tabular model using the new config
                    tabular_model_t = TabularModel(
                        data_config=self.data_config,
                        model_config=model_config_t,
                        optimizer_config=optimizer_config_t,
                        trainer_config=trainer_config_t,
                        verbose=verbose_tabular_model,
                        **self.tabular_model_init_kwargs,
                    )

                    # Create datamodule
                    if not datamodule:
                        prep_dl_kwargs, prep_model_kwargs, train_kwargs = tabular_model_t._split_kwargs(kwargs)
                        if "seed" not in prep_dl_kwargs:
                            prep_dl_kwargs["seed"] = random_state
                        datamodule = tabular_model_t.prepare_dataloader(
                            train=train, validation=validation, **prep_dl_kwargs
                        )
                        validation = validation if validation is not None else datamodule.validation_dataset.data

                    if cv is not None:
                        cv_verbose = cv_kwargs.pop("verbose", False)
                        cv_kwargs.pop("handle_oom", None)
                        with OutOfMemoryHandler(handle_oom=True) as handler:
                            cv_scores, _ = tabular_model_t.cross_validate(
                                cv=cv,
                                train=train,
                                metric=metric,
                                verbose=cv_verbose,
                                handle_oom=False,
                                **cv_kwargs,
                            )
                        if handler.oom_triggered:
                            if not ignore_oom:
                                raise OOMException(
                                    "Out of memory error occurred during cross validation. "
                                    "Set ignore_oom=True to ignore this error."
                                )
                            else:
                                params.update({metric_str: (np.inf if mode == "min" else -np.inf)})
                                params.update({"model": f"{params['model']} (OOM)"})
                        else:
                            params.update({metric_str: cv_agg_func(cv_scores)})
                    else:
                        model = tabular_model_t.prepare_model(
                            datamodule=datamodule,
                            **prep_model_kwargs,
                        )
                        train_kwargs.pop("handle_oom", None)
                        with OutOfMemoryHandler(handle_oom=True) as handler:
                            tabular_model_t.train(model=model, datamodule=datamodule, handle_oom=False, **train_kwargs)
                        if handler.oom_triggered:
                            if not ignore_oom:
                                raise OOMException(
                                    "Out of memory error occurred during training. "
                                    "Set ignore_oom=True to ignore this error."
                                )
                            else:
                                params.update({metric_str: (np.inf if mode == "min" else -np.inf)})
                                params.update({"model": f"{params['model']} (OOM)"})
                        else:
                            if is_callable_metric:
                                preds = tabular_model_t.predict(validation, include_input_features=False)
                                params.update({metric_str: metric(validation[tabular_model_t.config.target], preds)})
                            else:
                                result = tabular_model_t.evaluate(validation, verbose=False)
                                params.update({k.replace("test_", ""): v for k, v in result[0].items()})

                            if return_best_model:
                                # Removing the datamodule from the model to save memory
                                tabular_model_t.datamodule = None
                                if best_model is None:
                                    best_model = deepcopy(tabular_model_t)
                                    best_score = params[metric_str]
                                else:
                                    if mode == "min":
                                        if params[metric_str] < best_score:
                                            best_model = deepcopy(tabular_model_t)
                                            best_score = params[metric_str]
                                    elif mode == "max":
                                        if params[metric_str] > best_score:
                                            best_model = deepcopy(tabular_model_t)
                                            best_score = params[metric_str]

                    params.update({"trial_id": i})
                    trials.append(params)
                    if verbose:
                        logger.info(f"Trial {i+1}/{n_trials}: {params} | Score: {params[metric]}")

        trials_df = pd.DataFrame(trials)
        trials = trials_df.pop("trial_id")
        if mode == "max":
            best_idx = trials_df[metric_str].idxmax()
        elif mode == "min":
            best_idx = trials_df[metric_str].idxmin()
        else:
            raise NotImplementedError(f"{mode} is not implemented yet.")
        best_params = trials_df.iloc[best_idx].to_dict()
        best_score = best_params.pop(metric_str)
        trials_df.insert(0, "trial_id", trials)

        if verbose:
            logger.info("Model Tuner Finished")
            logger.info(f"Best Model: {best_params['model']} - Best Score ({metric_str}): {best_score}")

        if return_best_model and best_model is not None:
            best_model.datamodule = datamodule

            return self.OUTPUT(trials_df, best_params, best_score, best_model)
        else:
            return self.OUTPUT(trials_df, best_params, best_score, None)

__init__(data_config=None, model_config=None, optimizer_config=None, trainer_config=None, model_callable=None, model_state_dict_path=None, suppress_lightning_logger=True, **kwargs)

表格模型调优器帮助您调整表格模型的超参数.

Parameters:

Name	Type	Description	Default
data_config	(可选[Union[DataConfig, str]], 可选)	表格模型的DataConfig. 如果传递了str,将使用该路径中的yaml文件初始化DataConfig. 默认为None.	None
model_config	(可选[Union[ModelConfig, List[TrainerConfig]]], 可选)	表格模型的ModelConfig. 如果传递了str,将使用该路径中的yaml文件初始化ModelConfig. 默认为None.	None
optimizer_config	(可选[Union[OptimizerConfig, str]], 可选)	表格模型的OptimizerConfig. 如果传递了str,将使用该路径中的yaml文件初始化OptimizerConfig. 默认为None.	None
trainer_config	(可选[Union[TrainerConfig, str]], 可选)	表格模型的TrainerConfig. 如果传递了str,将使用该路径中的yaml文件初始化TrainerConfig. 默认为None.	None
model_callable	(可选[Callable], 可选)	返回PyTorch表格模型的可调用对象. 如果提供,将忽略model_config并使用此可调用对象初始化模型. 默认为None.	None
model_state_dict_path	(可选[Union[str, Path]], 可选)	模型状态字典的路径. 如果提供,将忽略model_config并使用此状态字典初始化模型. 默认为None.	None
suppress_lightning_logger	(bool, 可选)	是否抑制lightning日志记录器.默认为True.	True
**kwargs		传递给TabularModel初始化的其他关键字参数.	{}

Source code in src/pytorch_tabular/tabular_model_tuner.py

    def __init__(
        self,
        data_config: Optional[Union[DataConfig, str]] = None,
        model_config: Optional[Union[ModelConfig, str]] = None,
        optimizer_config: Optional[Union[OptimizerConfig, str]] = None,
        trainer_config: Optional[Union[TrainerConfig, List[TrainerConfig]]] = None,
        model_callable: Optional[Callable] = None,
        model_state_dict_path: Optional[Union[str, Path]] = None,
        suppress_lightning_logger: bool = True,
        **kwargs,
    ):
        """表格模型调优器帮助您调整表格模型的超参数.

Parameters:
    data_config (可选[Union[DataConfig, str]], 可选): 表格模型的DataConfig.
        如果传递了str,将使用该路径中的yaml文件初始化DataConfig.
        默认为None.

    model_config (可选[Union[ModelConfig, List[TrainerConfig]]], 可选): 表格模型的ModelConfig.
        如果传递了str,将使用该路径中的yaml文件初始化ModelConfig.
        默认为None.

    optimizer_config (可选[Union[OptimizerConfig, str]], 可选): 表格模型的OptimizerConfig.
        如果传递了str,将使用该路径中的yaml文件初始化OptimizerConfig.
        默认为None.

    trainer_config (可选[Union[TrainerConfig, str]], 可选): 表格模型的TrainerConfig.
        如果传递了str,将使用该路径中的yaml文件初始化TrainerConfig.
        默认为None.

    model_callable (可选[Callable], 可选): 返回PyTorch表格模型的可调用对象.
        如果提供,将忽略model_config并使用此可调用对象初始化模型.
        默认为None.

    model_state_dict_path (可选[Union[str, Path]], 可选): 模型状态字典的路径.

        如果提供,将忽略model_config并使用此状态字典初始化模型.
        默认为None.

    suppress_lightning_logger (bool, 可选): 是否抑制lightning日志记录器.默认为True.

    **kwargs: 传递给TabularModel初始化的其他关键字参数.
"""
        if not isinstance(model_config, list):
            model_config = [model_config]

        if trainer_config.profiler is not None:
            warnings.warn(
                "Profiler is not supported in tuner. Set profiler=None in TrainerConfig to disable this warning."
            )
            trainer_config.profiler = None
        if trainer_config.fast_dev_run:
            warnings.warn("fast_dev_run is turned on. Tuning results won't be accurate.")
        if trainer_config.progress_bar != "none":
            # If config and tuner have progress bar enabled, it will result in a bug within the library (rich.progress)
            trainer_config.progress_bar = "none"
            warnings.warn("Turning off progress bar. Set progress_bar='none' in TrainerConfig to disable this warning.")
        trainer_config.trainer_kwargs.update({"enable_model_summary": False})
        self.data_config = data_config
        self.model_config = model_config
        self.optimizer_config = optimizer_config
        self.trainer_config = trainer_config
        self.suppress_lightning_logger = suppress_lightning_logger
        self.tabular_model_init_kwargs = {
            "model_callable": model_callable,
            "model_state_dict_path": model_state_dict_path,
            **kwargs,
        }

tune(train, search_space, metric, mode, strategy, validation=None, n_trials=None, cv=None, cv_agg_func=np.mean, cv_kwargs={}, return_best_model=True, verbose=False, progress_bar=True, random_state=42, ignore_oom=True, **kwargs)

调整TabularModel的超参数.

Parameters:

Name	Type	Description	Default
train	DataFrame	训练数据	required
validation	(DataFrame, 可选)	验证数据.默认为None.	None
search_space	Dict	一个字典,形式为{参数名: [要尝试的值]}用于网格搜索,或{参数名: 分布}用于随机搜索	required
metric	Union[str, Callable]	用于评估的指标. 如果提供字符串,将使用定义的指标之一. 如果提供可调用对象,将使用该函数作为指标. 我们期望可调用对象的形式为metric(y_true, y_pred).对于分类问题,y_pred是一个包含每个类别的概率（_probability）和最终预测（prediction）的数据框.对于回归问题,它是一个包含最终预测（_prediction）的数据框. 默认为None.	required
mode	str	其中之一['max', 'min'].是否最大化或最小化指标.	required
strategy	str	其中之一['grid_search', 'random_search'].用于调整的策略.	required
n_trials	(int, 可选)	要运行的试验次数.仅用于随机搜索. 默认为None.	None
cv	Optional[Union[int, Iterable, BaseCrossValidator]]	确定交叉验证的分割策略. 可能的输入为: None,不使用任何交叉验证.我们将仅使用验证数据 整数,指定(Stratified)KFold中的折数, 一个可迭代对象,生成(train, test)分割为索引数组. 一个scikit-learn的CV分割器. 默认为None.	None
cv_agg_func	(Optional[Callable], 可选)	用于聚合交叉验证分数的函数. 默认为np.mean.	mean
cv_kwargs	(Optional[Dict], 可选)	要传递给交叉验证方法的额外关键字参数. 默认为{}.	{}
return_best_model	(bool, 可选)	如果为True,将返回最佳模型.默认为True.	True
verbose	(bool, 可选)	是否打印每次试验的结果.默认为False.	False
progress_bar	(bool, 可选)	是否显示进度条.默认为True.	True
random_state	(Optional[int], 可选)	用于随机搜索的随机状态.默认为42.	42
ignore_oom	(bool, 可选)	是否忽略内存不足错误.默认为True.	True
**kwargs		要传递给TabularModel fit的额外关键字参数.	{}

Returns:

Name	Type	Description
OUTPUT		一个命名元组,包含以下属性: trials_df (DataFrame): 包含每次试验结果的数据框 best_params (Dict): 找到的最佳参数 best_score (float): 找到的最佳分数 best_model (TabularModel 或 None): 如果return_best_model为True,返回best_model,否则返回None

Source code in src/pytorch_tabular/tabular_model_tuner.py

    def tune(
        self,
        train: DataFrame,
        search_space: Union[Dict, List[Dict]],
        metric: Union[str, Callable],
        mode: str,
        strategy: str,
        validation: Optional[DataFrame] = None,
        n_trials: Optional[int] = None,
        cv: Optional[Union[int, Iterable, BaseCrossValidator]] = None,
        cv_agg_func: Optional[Callable] = np.mean,
        cv_kwargs: Optional[Dict] = {},
        return_best_model: bool = True,
        verbose: bool = False,
        progress_bar: bool = True,
        random_state: Optional[int] = 42,
        ignore_oom: bool = True,
        **kwargs,
    ):
        """调整TabularModel的超参数.

Parameters:
    train (DataFrame): 训练数据

    validation (DataFrame, 可选): 验证数据.默认为None.

    search_space (Dict): 一个字典,形式为{参数名: [要尝试的值]}用于网格搜索,或{参数名: 分布}用于随机搜索

    metric (Union[str, Callable]): 用于评估的指标.
        如果提供字符串,将使用定义的指标之一.
        如果提供可调用对象,将使用该函数作为指标.
        我们期望可调用对象的形式为`metric(y_true, y_pred)`.对于分类问题,`y_pred`是一个包含每个类别的概率（<class>_probability）和最终预测（prediction）的数据框.对于回归问题,它是一个包含最终预测（<target>_prediction）的数据框.
        默认为None.

    mode (str): 其中之一['max', 'min'].是否最大化或最小化指标.

    strategy (str): 其中之一['grid_search', 'random_search'].用于调整的策略.

    n_trials (int, 可选): 要运行的试验次数.仅用于随机搜索.
        默认为None.

    cv (Optional[Union[int, Iterable, BaseCrossValidator]]): 确定交叉验证的分割策略.
        可能的输入为:

        - None,不使用任何交叉验证.我们将仅使用验证数据
        - 整数,指定(Stratified)KFold中的折数,
        - 一个可迭代对象,生成(train, test)分割为索引数组.
        - 一个scikit-learn的CV分割器.
        默认为None.

    cv_agg_func (Optional[Callable], 可选): 用于聚合交叉验证分数的函数.
        默认为np.mean.

    cv_kwargs (Optional[Dict], 可选): 要传递给交叉验证方法的额外关键字参数.
        默认为{}.

    return_best_model (bool, 可选): 如果为True,将返回最佳模型.默认为True.

    verbose (bool, 可选): 是否打印每次试验的结果.默认为False.

    progress_bar (bool, 可选): 是否显示进度条.默认为True.

    random_state (Optional[int], 可选): 用于随机搜索的随机状态.默认为42.

    ignore_oom (bool, 可选): 是否忽略内存不足错误.默认为True.

    **kwargs: 要传递给TabularModel fit的额外关键字参数.

Returns:
    OUTPUT: 一个命名元组,包含以下属性:
        trials_df (DataFrame): 包含每次试验结果的数据框
        best_params (Dict): 找到的最佳参数
        best_score (float): 找到的最佳分数
        best_model (TabularModel 或 None): 如果return_best_model为True,返回best_model,否则返回None
"""
        assert strategy in self.ALLOWABLE_STRATEGIES, f"tuner must be one of {self.ALLOWABLE_STRATEGIES}"
        assert mode in ["max", "min"], "mode must be one of ['max', 'min']"
        assert metric is not None, "metric must be specified"
        assert (isinstance(search_space, dict) or (isinstance(search_space, list))) and len(
            search_space
        ) > 0, "search_space must be a non-empty dict"
        if self.suppress_lightning_logger:
            suppress_lightning_logs()
        if cv is not None and validation is not None:
            warnings.warn(
                "Both validation and cv are provided. Ignoring validation and using cv. Use "
                "`validation=None` to turn off this warning."
            )
            validation = None

        if not isinstance(search_space, list):
            search_space = [search_space]

        assert len(self.model_config) == len(search_space), "model_config and search_space must have the same length"

        verbose_tabular_model = self.tabular_model_init_kwargs.pop("verbose", False)

        with Progress() as progress:
            model_config_iterator = range(len(self.model_config))
            if progress_bar:
                model_config_iterator = progress.track(
                    model_config_iterator, description="[green]Running models config..."
                )

            datamodule = None
            trials = []
            best_model = None
            best_score = 0.0
            for idx in model_config_iterator:
                search_space_temp = {
                    **{"model": [f"{idx}-{self.model_config[idx].__class__.__name__}"]},
                    **search_space[idx],
                }

                if strategy == "grid_search":
                    assert all(
                        isinstance(v, list) for v in search_space_temp.values()
                    ), "For grid search, all values in search_space must be a list of values to try"
                    search_space_iterator = list(ParameterGrid(search_space_temp))
                    if n_trials is not None:
                        warnings.warn(
                            "n_trials is ignored for grid search to do a complete sweep of"
                            " the grid. Set n_trials=None to turn off this warning."
                        )
                    n_trials = sum(1 for _ in search_space_iterator)
                elif strategy == "random_search":
                    assert n_trials is not None, "n_trials must be specified for random search"
                    search_space_iterator = list(
                        ParameterSampler(search_space_temp, n_iter=n_trials, random_state=random_state)
                    )
                else:
                    raise NotImplementedError(f"{strategy} is not implemented yet.")

                # Sort by trainer_config to recreate the datamodule when necessary
                trainer_configs = [key for key in search_space_iterator if "trainer_config" in key]
                for key in trainer_configs:
                    search_space_iterator = sorted(
                        search_space_iterator, key=lambda search_space_iterator: search_space_iterator[key]
                    )

                if progress_bar:
                    search_space_iterator = progress.track(
                        search_space_iterator,
                        description=f"[blue]Training {idx}-{self.model_config[idx].__class__.__name__}...",
                    )

                if isinstance(metric, str):
                    is_callable_metric = False
                    metric_str = metric
                elif callable(metric):
                    is_callable_metric = True
                    metric_str = metric.__name__

                for i, params in enumerate(search_space_iterator):
                    # Copying the configs as a base
                    # Make sure all default parameters that you want to be set for all
                    # trials are in the original configs
                    trainer_config_t = deepcopy(self.trainer_config)
                    optimizer_config_t = deepcopy(self.optimizer_config)
                    model_config_t = deepcopy(self.model_config[idx])

                    optimizer_config_t, model_config_t = self._update_configs(
                        optimizer_config_t, model_config_t, params
                    )
                    # Initialize Tabular model using the new config
                    tabular_model_t = TabularModel(
                        data_config=self.data_config,
                        model_config=model_config_t,
                        optimizer_config=optimizer_config_t,
                        trainer_config=trainer_config_t,
                        verbose=verbose_tabular_model,
                        **self.tabular_model_init_kwargs,
                    )

                    # Create datamodule
                    if not datamodule:
                        prep_dl_kwargs, prep_model_kwargs, train_kwargs = tabular_model_t._split_kwargs(kwargs)
                        if "seed" not in prep_dl_kwargs:
                            prep_dl_kwargs["seed"] = random_state
                        datamodule = tabular_model_t.prepare_dataloader(
                            train=train, validation=validation, **prep_dl_kwargs
                        )
                        validation = validation if validation is not None else datamodule.validation_dataset.data

                    if cv is not None:
                        cv_verbose = cv_kwargs.pop("verbose", False)
                        cv_kwargs.pop("handle_oom", None)
                        with OutOfMemoryHandler(handle_oom=True) as handler:
                            cv_scores, _ = tabular_model_t.cross_validate(
                                cv=cv,
                                train=train,
                                metric=metric,
                                verbose=cv_verbose,
                                handle_oom=False,
                                **cv_kwargs,
                            )
                        if handler.oom_triggered:
                            if not ignore_oom:
                                raise OOMException(
                                    "Out of memory error occurred during cross validation. "
                                    "Set ignore_oom=True to ignore this error."
                                )
                            else:
                                params.update({metric_str: (np.inf if mode == "min" else -np.inf)})
                                params.update({"model": f"{params['model']} (OOM)"})
                        else:
                            params.update({metric_str: cv_agg_func(cv_scores)})
                    else:
                        model = tabular_model_t.prepare_model(
                            datamodule=datamodule,
                            **prep_model_kwargs,
                        )
                        train_kwargs.pop("handle_oom", None)
                        with OutOfMemoryHandler(handle_oom=True) as handler:
                            tabular_model_t.train(model=model, datamodule=datamodule, handle_oom=False, **train_kwargs)
                        if handler.oom_triggered:
                            if not ignore_oom:
                                raise OOMException(
                                    "Out of memory error occurred during training. "
                                    "Set ignore_oom=True to ignore this error."
                                )
                            else:
                                params.update({metric_str: (np.inf if mode == "min" else -np.inf)})
                                params.update({"model": f"{params['model']} (OOM)"})
                        else:
                            if is_callable_metric:
                                preds = tabular_model_t.predict(validation, include_input_features=False)
                                params.update({metric_str: metric(validation[tabular_model_t.config.target], preds)})
                            else:
                                result = tabular_model_t.evaluate(validation, verbose=False)
                                params.update({k.replace("test_", ""): v for k, v in result[0].items()})

                            if return_best_model:
                                # Removing the datamodule from the model to save memory
                                tabular_model_t.datamodule = None
                                if best_model is None:
                                    best_model = deepcopy(tabular_model_t)
                                    best_score = params[metric_str]
                                else:
                                    if mode == "min":
                                        if params[metric_str] < best_score:
                                            best_model = deepcopy(tabular_model_t)
                                            best_score = params[metric_str]
                                    elif mode == "max":
                                        if params[metric_str] > best_score:
                                            best_model = deepcopy(tabular_model_t)
                                            best_score = params[metric_str]

                    params.update({"trial_id": i})
                    trials.append(params)
                    if verbose:
                        logger.info(f"Trial {i+1}/{n_trials}: {params} | Score: {params[metric]}")

        trials_df = pd.DataFrame(trials)
        trials = trials_df.pop("trial_id")
        if mode == "max":
            best_idx = trials_df[metric_str].idxmax()
        elif mode == "min":
            best_idx = trials_df[metric_str].idxmin()
        else:
            raise NotImplementedError(f"{mode} is not implemented yet.")
        best_params = trials_df.iloc[best_idx].to_dict()
        best_score = best_params.pop(metric_str)
        trials_df.insert(0, "trial_id", trials)

        if verbose:
            logger.info("Model Tuner Finished")
            logger.info(f"Best Model: {best_params['model']} - Best Score ({metric_str}): {best_score}")

        if return_best_model and best_model is not None:
            best_model.datamodule = datamodule

            return self.OUTPUT(trials_df, best_params, best_score, best_model)
        else:
            return self.OUTPUT(trials_df, best_params, best_score, None)

比较多个模型在同一数据集上的表现.

Parameters:

Name	Type	Description	Default
task	str	预测任务的类型.可以是 'classification' 或 'regression'	required
train	DataFrame	训练数据	required
test	DataFrame	用于评估性能的测试数据	required
data_config	Union[DataConfig, str]	DataConfig 对象或 yaml 文件的路径.	required
optimizer_config	Union[OptimizerConfig, str]	OptimizerConfig 对象或 yaml 文件的路径.	required
trainer_config	Union[TrainerConfig, str]	TrainerConfig 对象或 yaml 文件的路径.	required
model_list	Union[str, List[Union[ModelConfig, str]]]	要比较的模型列表. 可以是 pytorch_tabular.tabular_model_sweep.MODEL_SWEEP_PRESETS 中定义的预设之一, 或 ModelConfig 对象的列表.默认为 "lite".	'lite'
metrics	Optional[List[str]]	训练期间需要跟踪的指标列表.指标应为 torchmetrics 中实现的函数式指标之一. 默认情况下,分类任务为准确率,回归任务为均方误差.	None
metrics_prob_input	Optional[bool]	配置中定义的分类指标的强制参数.这定义了指标函数的输入是概率还是类别. 长度应与指标数量相同.默认为 None.	None
metrics_params	Optional[List]	传递给指标函数的参数.task 强制为 multiclass,因为多分类版本可以处理二分类, 并且为了简单起见,我们只使用 multiclass.	None
validation	Optional[DataFrame]	如果提供,将在训练时使用此数据框作为验证.用于早停和日志记录.如果留空,将使用 20% 的训练数据作为验证. 默认为 None.	None
experiment_config	Optional[Union[ExperimentConfig, str]]	ExperimentConfig 对象或 yaml 文件的路径.	None
common_model_args	Optional[dict]	所有模型通用的模型参数.参数列表可以在 ModelConfig 中找到. 如果未提供,将使用默认值.默认为 {}.	{}
rank_metric	Optional[Tuple[str, str]]	用于对模型进行排序的指标.元组的第一个元素是指标名称, 第二个元素是方向.默认为 ('loss', "lower_is_better").	('loss', 'lower_is_better')
return_best_model	bool	如果为 True,将返回最佳模型.默认为 True.	True
seed	int	用于可重复性的种子.默认为 42.	42
ignore_oom	bool	如果为 True,将忽略内存不足错误并继续下一个模型.	True
progress_bar	bool	如果为 True,将显示进度条.默认为 True.	True
verbose	bool	如果为 True,将打印进度.默认为 True.	True
suppress_lightning_logger	bool	如果为 True,将抑制 lightning 日志记录器.默认为 True.	True

Returns:

Name	Type	Description
results		训练结果.
best_model		如果 return_best_model 为 True,返回最佳模型,否则返回 None.

Source code in src/pytorch_tabular/tabular_model_sweep.py

def model_sweep(
    task: str,
    train: pd.DataFrame,
    test: pd.DataFrame,
    data_config: Union[DataConfig, str],
    optimizer_config: Union[OptimizerConfig, str],
    trainer_config: Union[TrainerConfig, str],
    model_list: Union[str, List[Union[ModelConfig, str]]] = "lite",
    metrics: Optional[List[Union[str, Callable]]] = None,
    metrics_params: Optional[List[dict]] = None,
    metrics_prob_input: Optional[List[bool]] = None,
    validation: Optional[pd.DataFrame] = None,
    experiment_config: Optional[Union[ExperimentConfig, str]] = None,
    common_model_args: Optional[dict] = {},
    rank_metric: Optional[Tuple[str, str]] = ("loss", "lower_is_better"),
    return_best_model: bool = True,
    seed: int = 42,
    ignore_oom: bool = True,
    progress_bar: bool = True,
    verbose: bool = True,
    suppress_lightning_logger: bool = True,
):
    """比较多个模型在同一数据集上的表现.

Parameters:
    task (str): 预测任务的类型.可以是 'classification' 或 'regression'

    train (pd.DataFrame): 训练数据

    test (pd.DataFrame): 用于评估性能的测试数据

    data_config (Union[DataConfig, str]): DataConfig 对象或 yaml 文件的路径.

    optimizer_config (Union[OptimizerConfig, str]): OptimizerConfig 对象或 yaml 文件的路径.

    trainer_config (Union[TrainerConfig, str]): TrainerConfig 对象或 yaml 文件的路径.

    model_list (Union[str, List[Union[ModelConfig, str]]], optional): 要比较的模型列表.
            可以是 ``pytorch_tabular.tabular_model_sweep.MODEL_SWEEP_PRESETS`` 中定义的预设之一,
            或 ``ModelConfig`` 对象的列表.默认为 "lite".

    metrics (Optional[List[str]]): 训练期间需要跟踪的指标列表.指标应为 ``torchmetrics`` 中实现的函数式指标之一.
            默认情况下,分类任务为准确率,回归任务为均方误差.

    metrics_prob_input (Optional[bool]): 配置中定义的分类指标的强制参数.这定义了指标函数的输入是概率还是类别.
            长度应与指标数量相同.默认为 None.

    metrics_params (Optional[List]): 传递给指标函数的参数.`task` 强制为 `multiclass`,因为多分类版本可以处理二分类,
            并且为了简单起见,我们只使用 `multiclass`.

    validation (Optional[DataFrame], optional):
            如果提供,将在训练时使用此数据框作为验证.用于早停和日志记录.如果留空,将使用 20% 的训练数据作为验证.
            默认为 None.

    experiment_config (Optional[Union[ExperimentConfig, str]], optional): ExperimentConfig 对象或 yaml 文件的路径.

    common_model_args (Optional[dict], optional): 所有模型通用的模型参数.参数列表可以在 ``ModelConfig`` 中找到.
            如果未提供,将使用默认值.默认为 {}.

    rank_metric (Optional[Tuple[str, str]], optional): 用于对模型进行排序的指标.元组的第一个元素是指标名称,
            第二个元素是方向.默认为 ('loss', "lower_is_better").

    return_best_model (bool, optional): 如果为 True,将返回最佳模型.默认为 True.

    seed (int, optional): 用于可重复性的种子.默认为 42.

    ignore_oom (bool, optional): 如果为 True,将忽略内存不足错误并继续下一个模型.

    progress_bar (bool, optional): 如果为 True,将显示进度条.默认为 True.

    verbose (bool, optional): 如果为 True,将打印进度.默认为 True.

    suppress_lightning_logger (bool, optional): 如果为 True,将抑制 lightning 日志记录器.默认为 True.

Returns:
    results: 训练结果.

    best_model: 如果 return_best_model 为 True,返回最佳模型,否则返回 None.
"""
    _validate_args(
        task=task,
        train=train,
        test=test,
        data_config=data_config,
        optimizer_config=optimizer_config,
        trainer_config=trainer_config,
        model_list=model_list,
        metrics=metrics,
        metrics_params=metrics_params,
        metrics_prob_input=metrics_prob_input,
        validation=validation,
        experiment_config=experiment_config,
        common_model_args=common_model_args,
        rank_metric=rank_metric,
    )
    if suppress_lightning_logger:
        suppress_lightning_logs()
    if progress_bar:
        if trainer_config.progress_bar != "none":
            # Turning off thie internal progress bar to avoid conflict with sweep progress bar
            warnings.warn(
                "Training Progress bar is not `none`. Set `progress_bar=none` in"
                " `trainer_config` to remove this warning"
            )
            trainer_config.progress_bar = "none"

    if model_list in ["full", "high_memory"]:
        warnings.warn(
            "The full model list is quite large and uses a lot of memory. "
            "Consider using `lite` or define configs yourselves for a faster run"
        )
    _model_args = ["metrics", "metrics_params", "metrics_prob_input"]
    # Replacing the common model args with the ones passed in the function
    for arg in _model_args:
        if locals()[arg] is not None:
            common_model_args[arg] = locals()[arg]
    if isinstance(model_list, str):
        model_list = copy.deepcopy(MODEL_SWEEP_PRESETS[model_list])
        model_list = [
            (
                getattr(models, model_config[0])(task=task, **model_config[1], **common_model_args)
                if isinstance(model_config, Tuple)
                else (
                    getattr(models, model_config)(task=task, **common_model_args)
                    if isinstance(model_config, str)
                    else model_config
                )
            )
            for model_config in model_list
        ]

    def _init_tabular_model(m):
        return TabularModel(
            data_config=data_config,
            model_config=m,
            optimizer_config=optimizer_config,
            trainer_config=trainer_config,
            experiment_config=experiment_config,
            verbose=False,
        )

    datamodule = _init_tabular_model(model_list[0]).prepare_dataloader(train=train, validation=validation, seed=seed)
    results = []
    best_model = None
    is_lower_better = rank_metric[1] == "lower_is_better"
    best_score = 1e9 if is_lower_better else -1e9
    it = track(model_list, description="Sweeping Models") if progress_bar else model_list
    ctx = Progress() if progress_bar else nullcontext()
    with ctx as progress:
        if progress_bar:
            task_p = progress.add_task("Sweeping Models", total=len(model_list))
        for model_config in model_list:
            if isinstance(model_config, str):
                model_config = getattr(models, model_config)(task=task, **common_model_args)
            else:
                for key, val in common_model_args.items():
                    if hasattr(model_config, key):
                        setattr(model_config, key, val)
                    else:
                        raise ValueError(
                            f"ModelConfig {model_config.name} does not have an" f" attribute {key} in common_model_args"
                        )
            params = model_config.__dict__
            start_time = time.time()
            tabular_model = _init_tabular_model(model_config)
            name = tabular_model.name
            if verbose:
                logger.info(f"Training {name}")
            model = tabular_model.prepare_model(datamodule)
            if progress_bar:
                progress.update(task_p, description=f"Training {name}", advance=1)
            with OutOfMemoryHandler(handle_oom=True) as handler:
                tabular_model.train(model, datamodule, handle_oom=False)
            res_dict = {
                "model": name,
                "# Params": int_to_human_readable(tabular_model.num_params),
            }
            if handler.oom_triggered:
                if not ignore_oom:
                    raise OOMException(
                        "Out of memory error occurred during cross validation. "
                        "Set ignore_oom=True to ignore this error."
                    )
                else:
                    res_dict.update(
                        {
                            f"test_{rank_metric[0]}": (np.inf if is_lower_better else -np.inf),
                            "epochs": "OOM",
                            "time_taken": "OOM",
                            "time_taken_per_epoch": "OOM",
                        }
                    )
                    res_dict["model"] = name + " (OOM)"
            else:
                if (
                    tabular_model.trainer.early_stopping_callback is not None
                    and tabular_model.trainer.early_stopping_callback.stopped_epoch != 0
                ):
                    res_dict["epochs"] = tabular_model.trainer.early_stopping_callback.stopped_epoch
                else:
                    res_dict["epochs"] = tabular_model.trainer.max_epochs
                res_dict.update(tabular_model.evaluate(test=test, verbose=False)[0])
                res_dict["time_taken"] = time.time() - start_time
                res_dict["time_taken_per_epoch"] = res_dict["time_taken"] / res_dict["epochs"]

                if return_best_model:
                    tabular_model.datamodule = None
                    if best_model is None:
                        best_model = copy.deepcopy(tabular_model)
                        best_score = res_dict[f"test_{rank_metric[0]}"]
                    else:
                        if is_lower_better:
                            if res_dict[f"test_{rank_metric[0]}"] < best_score:
                                best_model = copy.deepcopy(tabular_model)
                                best_score = res_dict[f"test_{rank_metric[0]}"]
                        else:
                            if res_dict[f"test_{rank_metric[0]}"] > best_score:
                                best_model = copy.deepcopy(tabular_model)
                                best_score = res_dict[f"test_{rank_metric[0]}"]

            if verbose:
                logger.info(f"Finished Training {name}")
                logger.info("Results:" f" {', '.join([f'{k}: {v}' for k, v in res_dict.items()])}")
            res_dict["params"] = params
            results.append(res_dict)

    if verbose:
        logger.info("Model Sweep Finished")
        logger.info(f"Best Model: {best_model.name}")
    results = pd.DataFrame(results).sort_values(by=f"test_{rank_metric[0]}", ascending=is_lower_better)
    if return_best_model and best_model is not None:
        best_model.datamodule = datamodule
        return results, best_model
    else:
        return results, None