API
pytorch_tabular.tabular_model.TabularModel
Source code in 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,
) -> None:
"""The core model which orchestrates everything from initializing the datamodule, the model, trainer, etc.
Args:
config (Optional[Union[DictConfig, str]], optional): Single OmegaConf DictConfig object or
the path to the yaml file holding all the config parameters. Defaults to None.
data_config (Optional[Union[DataConfig, str]], optional): DataConfig object or path to the yaml file. Defaults to None.
model_config (Optional[Union[ModelConfig, str]], optional): A subclass of ModelConfig or path to the yaml file.
Determines which model to run from the type of config. Defaults to None.
optimizer_config (Optional[Union[OptimizerConfig, str]], optional): OptimizerConfig object or path to the yaml file.
Defaults to None.
trainer_config (Optional[Union[TrainerConfig, str]], optional): TrainerConfig object or path to the yaml file.
Defaults to None.
experiment_config (Optional[Union[ExperimentConfig, str]], optional): ExperimentConfig object or path to the yaml file.
If Provided configures the experiment tracking. Defaults to None.
model_callable (Optional[Callable], optional): If provided, will override the model callable that will be loaded from the config.
Typically used when providing Custom Models
"""
super().__init__()
self.exp_manager = ExperimentRunManager()
if config is None:
assert (
(data_config is not None)
or (model_config is not None)
or (optimizer_config is not None)
or (trainer_config is not None)
), "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 experiment_config is None:
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:
logger.info("Experiment Tracking is turned off")
self.track_experiment = False
self.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(
getattr(models, self.config._module_src), self.config._model_name
)
self.custom_model = False
else:
self.model_callable = model_callable
self.custom_model = True
self._run_validation()
def _run_validation(self):
"""Validates the Config params and throws errors if something is wrong
Raises:
NotImplementedError: If you provide a multi-target config to a classification task
ValueError: If there is a problem with Target Range
"""
if self.config.task == "classification":
if len(self.config.target) > 1:
raise NotImplementedError(
"Multi-Target Classification is not implemented."
)
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(
getattr(models, _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]:
"""Gets the name of the experiment and increments version by 1
Returns:
tuple[str, int]: Returns the name and version number
"""
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):
"""Sets up the Experiment Tracking Framework according to the choices made in the Experimentconfig
Raises:
NotImplementedError: Raises an Error for invalid choices of log_target
"""
if self.config.log_target == "tensorboard":
self.logger = pl.loggers.TensorBoardLogger(
name=self.name, save_dir=self.config.project_name, version=self.uid
)
elif self.config.log_target == "wandb":
self.logger = pl.loggers.WandbLogger(
name=f"{self.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:
"""Prepares the necesary callbacks to the Trainer based on the configuration
Returns:
List: A list of callbacks
"""
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,
verbose=False,
mode=self.config.early_stopping_mode,
)
callbacks.append(early_stop_callback)
if self.config.checkpoints:
ckpt_name = f"{self.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,
)
callbacks.append(model_checkpoint)
self.config.checkpoint_callback = True
else:
self.config.checkpoint_callback = False
if self.config.progress_bar == "rich":
callbacks.append(RichProgressBar())
logger.debug(f"Callbacks used: {callbacks}")
return callbacks
def _prepare_dataloader(
self, train, validation, test, target_transform=None, train_sampler=None
):
logger.info("Preparing the DataLoaders...")
if (
hasattr(self, "datamodule")
and self.datamodule is not None
and self.datamodule._fitted
):
logger.debug("Data Module is already fitted. Using it to get loaders")
else:
self.datamodule = TabularDatamodule(
train=train,
validation=validation,
config=self.config,
test=test,
target_transform=target_transform,
train_sampler=train_sampler,
)
self.datamodule.prepare_data()
self.datamodule.setup("fit")
train_loader = self.datamodule.train_dataloader()
val_loader = self.datamodule.val_dataloader()
return train_loader, val_loader
def _prepare_model(
self, loss, metrics, optimizer, optimizer_params, reset, trained_backbone
):
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.config = self.datamodule.config
if hasattr(self, "model") and self.model is not None and not reset:
logger.debug("Using the trained model...")
else:
logger.debug("Re-initializing the model. Trained weights are ignored.")
self.model = self.model_callable(
self.config,
custom_loss=loss,
custom_metrics=metrics,
custom_optimizer=optimizer,
custom_optimizer_params=optimizer_params,
)
# Data Aware Initialization(for the models that need it)
self.model.data_aware_initialization(self.datamodule)
if trained_backbone:
self.model.backbone = trained_backbone
def _prepare_trainer(self, max_epochs=None, min_epochs=None):
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["checkpoint_callback"] = self.config.checkpoint_callback
# 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)
self.trainer = pl.Trainer(
logger=self.logger,
callbacks=self.callbacks,
**trainer_args_config,
)
def load_best_model(self):
"""Loads the best model after training is done"""
if self.trainer.checkpoint_callback is not None:
logger.info("Loading the best model...")
ckpt_path = self.trainer.checkpoint_callback.best_model_path
if ckpt_path != "":
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.info(
"No best model available to load. Did you run it more than 1 epoch?..."
)
else:
logger.info(
"No best model available to load. Did you run it more than 1 epoch?..."
)
def _pre_fit(
self,
train: pd.DataFrame,
validation: Optional[pd.DataFrame],
test: Optional[pd.DataFrame],
loss: Optional[torch.nn.Module],
metrics: Optional[List[Callable]],
optimizer: Optional[torch.optim.Optimizer],
optimizer_params: Dict,
train_sampler: Optional[torch.utils.data.Sampler],
target_transform: Optional[Union[TransformerMixin, Tuple]],
max_epochs: int,
min_epochs: int,
reset: bool,
trained_backbone: Optional[pl.LightningModule],
callbacks: Optional[List[pl.Callback]],
):
"""Prepares the dataloaders, trainer, and model for the fit process"""
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
train_loader, val_loader = self._prepare_dataloader(
train, validation, test, target_transform, train_sampler
)
self._prepare_model(
loss, metrics, optimizer, optimizer_params, reset, trained_backbone
)
if self.track_experiment and self.config.log_target == "wandb":
self.logger.watch(
self.model, log=self.config.exp_watch, log_freq=self.config.exp_log_freq
)
self.callbacks = self._prepare_callbacks(callbacks)
self._prepare_trainer(max_epochs, min_epochs)
return train_loader, val_loader
def fit(
self,
train: pd.DataFrame,
validation: Optional[pd.DataFrame] = None,
test: Optional[pd.DataFrame] = None,
loss: Optional[torch.nn.Module] = None,
metrics: Optional[List[Callable]] = None,
optimizer: Optional[torch.optim.Optimizer] = None,
optimizer_params: Dict = {},
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,
reset: bool = False,
seed: Optional[int] = None,
trained_backbone: Optional[pl.LightningModule] = None,
callbacks: Optional[List[pl.Callback]] = None,
) -> None:
"""The fit method which takes in the data and triggers the training
Args:
train (pd.DataFrame): Training Dataframe
validation (Optional[pd.DataFrame], optional): If provided, will use this dataframe as the validation while training.
Used in Early Stopping and Logging. If left empty, will use 20% of Train data as validation. Defaults to None.
test (Optional[pd.DataFrame], optional): If provided, will use as the hold-out data,
which you'll be able to check performance after the model is trained. Defaults to None.
loss (Optional[torch.nn.Module], optional): Custom Loss functions which are not in standard pytorch library
metrics (Optional[List[Callable]], optional): Custom metric functions(Callable) which has the
signature metric_fn(y_hat, y) and works on torch tensor inputs
optimizer (Optional[torch.optim.Optimizer], optional): Custom optimizers which are a drop in replacements for standard PyToch optimizers.
This should be the Class and not the initialized object
optimizer_params (Optional[Dict], optional): The parmeters to initialize the custom optimizer.
train_sampler (Optional[torch.utils.data.Sampler], optional): Custom PyTorch batch samplers which will be passed to the DataLoaders. Useful for dealing with imbalanced data and other custom batching strategies
target_transform (Optional[Union[TransformerMixin, Tuple(Callable)]], optional): If provided, applies the transform to the target before modelling
and inverse the transform during prediction. The parameter can either be a sklearn Transformer which has an inverse_transform method, or
a tuple of callables (transform_func, inverse_transform_func)
max_epochs (Optional[int]): Overwrite maximum number of epochs to be run
min_epochs (Optional[int]): Overwrite minimum number of epochs to be run
reset: (bool): Flag to reset the model and train again from scratch
seed: (int): If you have to override the default seed set as part of of ModelConfig
trained_backbone (pl.LightningModule): this module contains the weights for a pretrained backbone
callbacks (Optional[List[pl.Callback]], optional): Custom callbacks to be used during training.
"""
seed_everything(seed if seed is not None else self.config.seed)
train_loader, val_loader = self._pre_fit(
train,
validation,
test,
loss,
metrics,
optimizer,
optimizer_params,
train_sampler,
target_transform,
max_epochs,
min_epochs,
reset,
trained_backbone,
callbacks,
)
self.model.train()
if self.config.auto_lr_find and (not self.config.fast_dev_run):
self.trainer.tune(self.model, train_loader, val_loader)
# Parameters in models needs to be initialized again after LR find
self.model.data_aware_initialization(self.datamodule)
self.model.train()
self.trainer.fit(self.model, train_loader, val_loader)
logger.info("Training the model completed...")
if self.config.load_best:
self.load_best_model()
def find_learning_rate(
self,
train: pd.DataFrame,
validation: Optional[pd.DataFrame] = None,
test: Optional[pd.DataFrame] = None,
loss: Optional[torch.nn.Module] = None,
metrics: Optional[List[Callable]] = None,
optimizer: Optional[torch.optim.Optimizer] = None,
optimizer_params: Dict = {},
min_lr: float = 1e-8,
max_lr: float = 1,
num_training: int = 100,
mode: str = "exponential",
early_stop_threshold: float = 4.0,
plot=True,
train_sampler: Optional[torch.utils.data.Sampler] = None,
trained_backbone=None,
) -> None:
"""Enables the user to do a range test of good initial learning rates, to reduce the amount of guesswork in picking a good starting learning rate.
Args:
train (pd.DataFrame): Training Dataframe
validation (Optional[pd.DataFrame], optional): If provided, will use this dataframe as the validation while training.
Used in Early Stopping and Logging. If left empty, will use 20% of Train data as validation. Defaults to None.
test (Optional[pd.DataFrame], optional): If provided, will use as the hold-out data,
which you'll be able to check performance after the model is trained. Defaults to None.
loss (Optional[torch.nn.Module], optional): Custom Loss functions which are not in standard pytorch library
metrics (Optional[List[Callable]], optional): Custom metric functions(Callable) which has the signature metric_fn(y_hat, y)
optimizer (Optional[torch.optim.Optimizer], optional): Custom optimizers which are a drop in replacements for standard PyToch optimizers.
This should be the Class and not the initialized object
optimizer_params (Optional[Dict], optional): The parmeters to initialize the custom optimizer.
min_lr (Optional[float], optional): minimum learning rate to investigate
max_lr (Optional[float], optional): maximum learning rate to investigate
num_training (Optional[int], optional): number of learning rates to test
mode (Optional[str], optional): search strategy, either 'linear' or 'exponential'. If set to
'linear' the learning rate will be searched by linearly increasing
after each batch. If set to 'exponential', will increase learning
rate exponentially.
early_stop_threshold(Optional[float], optional): threshold for stopping the search. If the
loss at any point is larger than early_stop_threshold*best_loss
then the search is stopped. To disable, set to None.
plot(bool, optional): If true, will plot using matplotlib
trained_backbone (pl.LightningModule): this module contains the weights for a pretrained backbone
train_sampler (Optional[torch.utils.data.Sampler], optional): Custom PyTorch batch samplers which will be passed to the DataLoaders. Useful for dealing with imbalanced data and other custom batching strategies
"""
train_loader, val_loader = self._pre_fit(
train,
validation,
test,
loss,
metrics,
optimizer,
optimizer_params,
target_transform=None,
max_epochs=None,
min_epochs=None,
reset=True,
trained_backbone=trained_backbone,
train_sampler=train_sampler,
)
lr_finder = self.trainer.tuner.lr_find(
self.model,
train_loader,
val_loader,
min_lr,
max_lr,
num_training,
mode,
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
return new_lr, pd.DataFrame(lr_finder.results)
def evaluate(self, test: Optional[pd.DataFrame]) -> Union[dict, list]:
"""Evaluates the dataframe using the loss and metrics already set in config
Args:
test (Optional[pd.DataFrame]): The dataframe to be evaluated. If not provided, will try to use the
test provided during fit. If that was also not provided will return an empty dictionary
Returns:
Union[dict, list]: The final test result dictionary.
"""
if test is not None:
test_loader = self.datamodule.prepare_inference_dataloader(test)
elif self.test is not None:
test_loader = self.datamodule.test_dataloader()
else:
return {}
result = self.trainer.test(
model=self.model,
test_dataloaders=test_loader,
ckpt_path=None,
)
return result
def predict(
self,
test: pd.DataFrame,
quantiles: Optional[List] = [0.25, 0.5, 0.75],
n_samples: Optional[int] = 100,
ret_logits=False,
) -> pd.DataFrame:
"""Uses the trained model to predict on new data and return as a dataframe
Args:
test (pd.DataFrame): The new dataframe with the features defined during training
quantiles (Optional[List]): For probabilistic models like Mixture Density Networks, this specifies
the different quantiles to be extracted apart from the `central_tendency` and added to the dataframe.
For other models it is ignored. Defaults to [0.25, 0.5, 0.75]
n_samples (Optional[int]): Number of samples to draw from the posterior to estimate the quantiles.
Ignored for non-probabilistic models. Defaults to 100
ret_logits (bool): Flag to return raw model outputs/logits except the backbone features along
with the dataframe. Defaults to False
Returns:
pd.DataFrame: Returns a dataframe with predictions and features.
If classification, it returns probabilities and final prediction
"""
assert all(
[q <= 1 and q >= 0 for q in quantiles]
), "Quantiles should be a decimal between 0 and 1"
self.model.eval()
inference_dataloader = self.datamodule.prepare_inference_dataloader(test)
point_predictions = []
quantile_predictions = []
logits_predictions = defaultdict(list)
is_probabilistic = (
hasattr(self.model.hparams, "_probabilistic")
and self.model.hparams._probabilistic
)
for batch in tqdm(inference_dataloader, desc="Generating Predictions..."):
for k, v in batch.items():
if isinstance(v, list) and (len(v) == 0):
# Skipping empty list
continue
batch[k] = v.to(self.model.device)
if is_probabilistic:
samples, ret_value = self.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 = self.model.predict(batch, ret_model_output=True)
if ret_logits:
for k, v in ret_value.items():
# if k == "backbone_features":
# continue
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)
pred_df = test.copy()
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):
pred_df[
f"{target_col}_q{int(q*100)}"
] = 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":
point_predictions = nn.Softmax(dim=-1)(point_predictions).numpy()
for i, class_ in enumerate(self.datamodule.label_encoder.classes_):
pred_df[f"{class_}_probability"] = point_predictions[:, i]
pred_df["prediction"] = self.datamodule.label_encoder.inverse_transform(
np.argmax(point_predictions, axis=1)
)
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 save_model(self, dir: str):
"""Saves the model and checkpoints in the specified directory
Args:
dir (str): The path to the directory to save the model
"""
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)
with open(os.path.join(dir, "config.yml"), "w") as fp:
OmegaConf.save(self.config, fp, resolve=True)
joblib.dump(self.datamodule, os.path.join(dir, "datamodule.sav"))
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"] = self.model.custom_loss
custom_params["custom_metrics"] = self.model.custom_metrics
custom_params["custom_optimizer"] = self.model.custom_optimizer
custom_params["custom_optimizer_params"] = self.model.custom_optimizer_params
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]):
"""Saves the model weights in the specified directory
Args:
path (str): The path to the directory to save the model
"""
torch.save(self.model.state_dict(), path)
# TODO Need to test ONNX export
def save_model_for_inference(
self,
path: Union[str, Path],
kind: str = "pytorch",
onnx_export_params: Dict = {},
):
"""Saves the model for inference
path (Union[str, Path]): path to save the model
kind (str): "pytorch" or "onnx" (Experimental)
onnx_export_params (Dict): parameters for onnx export to be
passed to torch.onnx.export
"""
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 = dict(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")
@classmethod
def load_from_checkpoint(cls, dir: str, map_location=None, strict=True):
"""Loads a saved model from the directory
Args:
dir (str): The directory where the model wa saved, along with the checkpoints
map_location (Union[Dict[str, str], str, device, int, Callable, None]) – If your checkpoint
saved a GPU model and you now load on CPUs or a different number of GPUs, use this to map
to the new setup. The behaviour is the same as in torch.load()
strict (bool) – Whether to strictly enforce that the keys in checkpoint_path match the keys
returned by this module’s state dict. Default: True.
Returns:
TabularModel: The saved 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(
getattr(models, config._module_src), config._model_name
)
custom_model = False
custom_params = joblib.load(os.path.join(dir, "custom_params.sav"))
model_args = {}
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["metric_params"] = [{}] # 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
model = model_callable.load_from_checkpoint(
checkpoint_path=os.path.join(dir, "model.ckpt"),
map_location=map_location,
strict=strict,
**model_args,
)
# Updating config with custom parameters for experiment tracking
if custom_params.get("custom_loss") is not None:
model.custom_loss = custom_params["custom_loss"]
if custom_params.get("custom_metrics") is not None:
model.custom_metrics = custom_params["custom_metrics"]
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"]
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._prepare_trainer()
tabular_model.trainer.model = model
tabular_model.logger = logger
return tabular_model
def summary(self, max_depth=-1):
print(summarize(self.model, max_depth=max_depth))
def __str__(self) -> str:
return self.summary()
__init__(self, config=None, data_config=None, model_config=None, optimizer_config=None, trainer_config=None, experiment_config=None, model_callable=None)
special
The core model which orchestrates everything from initializing the datamodule, the model, trainer, etc.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
config |
Optional[Union[DictConfig, str]] |
Single OmegaConf DictConfig object or the path to the yaml file holding all the config parameters. Defaults to None. |
None |
data_config |
Optional[Union[DataConfig, str]] |
DataConfig object or path to the yaml file. Defaults to None. |
None |
model_config |
Optional[Union[ModelConfig, str]] |
A subclass of ModelConfig or path to the yaml file. Determines which model to run from the type of config. Defaults to None. |
None |
optimizer_config |
Optional[Union[OptimizerConfig, str]] |
OptimizerConfig object or path to the yaml file. Defaults to None. |
None |
trainer_config |
Optional[Union[TrainerConfig, str]] |
TrainerConfig object or path to the yaml file. Defaults to None. |
None |
experiment_config |
Optional[Union[ExperimentConfig, str]] |
ExperimentConfig object or path to the yaml file. If Provided configures the experiment tracking. Defaults to None. |
None |
model_callable |
Optional[Callable] |
If provided, will override the model callable that will be loaded from the config. Typically used when providing Custom Models |
None |
Source code in 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,
) -> None:
"""The core model which orchestrates everything from initializing the datamodule, the model, trainer, etc.
Args:
config (Optional[Union[DictConfig, str]], optional): Single OmegaConf DictConfig object or
the path to the yaml file holding all the config parameters. Defaults to None.
data_config (Optional[Union[DataConfig, str]], optional): DataConfig object or path to the yaml file. Defaults to None.
model_config (Optional[Union[ModelConfig, str]], optional): A subclass of ModelConfig or path to the yaml file.
Determines which model to run from the type of config. Defaults to None.
optimizer_config (Optional[Union[OptimizerConfig, str]], optional): OptimizerConfig object or path to the yaml file.
Defaults to None.
trainer_config (Optional[Union[TrainerConfig, str]], optional): TrainerConfig object or path to the yaml file.
Defaults to None.
experiment_config (Optional[Union[ExperimentConfig, str]], optional): ExperimentConfig object or path to the yaml file.
If Provided configures the experiment tracking. Defaults to None.
model_callable (Optional[Callable], optional): If provided, will override the model callable that will be loaded from the config.
Typically used when providing Custom Models
"""
super().__init__()
self.exp_manager = ExperimentRunManager()
if config is None:
assert (
(data_config is not None)
or (model_config is not None)
or (optimizer_config is not None)
or (trainer_config is not None)
), "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 experiment_config is None:
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:
logger.info("Experiment Tracking is turned off")
self.track_experiment = False
self.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(
getattr(models, self.config._module_src), self.config._model_name
)
self.custom_model = False
else:
self.model_callable = model_callable
self.custom_model = True
self._run_validation()
evaluate(self, test)
Evaluates the dataframe using the loss and metrics already set in config
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
test |
Optional[pd.DataFrame] |
The dataframe to be evaluated. If not provided, will try to use the test provided during fit. If that was also not provided will return an empty dictionary |
required |
Returns:
| Type | Description |
|---|---|
Union[dict, list] |
The final test result dictionary. |
Source code in pytorch_tabular/tabular_model.py
def evaluate(self, test: Optional[pd.DataFrame]) -> Union[dict, list]:
"""Evaluates the dataframe using the loss and metrics already set in config
Args:
test (Optional[pd.DataFrame]): The dataframe to be evaluated. If not provided, will try to use the
test provided during fit. If that was also not provided will return an empty dictionary
Returns:
Union[dict, list]: The final test result dictionary.
"""
if test is not None:
test_loader = self.datamodule.prepare_inference_dataloader(test)
elif self.test is not None:
test_loader = self.datamodule.test_dataloader()
else:
return {}
result = self.trainer.test(
model=self.model,
test_dataloaders=test_loader,
ckpt_path=None,
)
return result
find_learning_rate(self, train, validation=None, test=None, loss=None, metrics=None, optimizer=None, optimizer_params={}, min_lr=1e-08, max_lr=1, num_training=100, mode='exponential', early_stop_threshold=4.0, plot=True, train_sampler=None, trained_backbone=None)
Enables the user to do a range test of good initial learning rates, to reduce the amount of guesswork in picking a good starting learning rate.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
train |
pd.DataFrame |
Training Dataframe |
required |
validation |
Optional[pd.DataFrame] |
If provided, will use this dataframe as the validation while training. Used in Early Stopping and Logging. If left empty, will use 20% of Train data as validation. Defaults to None. |
None |
test |
Optional[pd.DataFrame] |
If provided, will use as the hold-out data, which you'll be able to check performance after the model is trained. Defaults to None. |
None |
loss |
Optional[torch.nn.Module] |
Custom Loss functions which are not in standard pytorch library |
None |
metrics |
Optional[List[Callable]] |
Custom metric functions(Callable) which has the signature metric_fn(y_hat, y) |
None |
optimizer |
Optional[torch.optim.Optimizer] |
Custom optimizers which are a drop in replacements for standard PyToch optimizers. This should be the Class and not the initialized object |
None |
optimizer_params |
Optional[Dict] |
The parmeters to initialize the custom optimizer. |
{} |
min_lr |
Optional[float] |
minimum learning rate to investigate |
1e-08 |
max_lr |
Optional[float] |
maximum learning rate to investigate |
1 |
num_training |
Optional[int] |
number of learning rates to test |
100 |
mode |
Optional[str] |
search strategy, either 'linear' or 'exponential'. If set to 'linear' the learning rate will be searched by linearly increasing after each batch. If set to 'exponential', will increase learning rate exponentially. |
'exponential' |
early_stop_threshold(Optional[float], |
optional |
threshold for stopping the search. If the loss at any point is larger than early_stop_threshold*best_loss then the search is stopped. To disable, set to None. |
required |
plot(bool, |
optional |
If true, will plot using matplotlib |
required |
trained_backbone |
pl.LightningModule |
this module contains the weights for a pretrained backbone |
None |
train_sampler |
Optional[torch.utils.data.Sampler] |
Custom PyTorch batch samplers which will be passed to the DataLoaders. Useful for dealing with imbalanced data and other custom batching strategies |
None |
Source code in pytorch_tabular/tabular_model.py
def find_learning_rate(
self,
train: pd.DataFrame,
validation: Optional[pd.DataFrame] = None,
test: Optional[pd.DataFrame] = None,
loss: Optional[torch.nn.Module] = None,
metrics: Optional[List[Callable]] = None,
optimizer: Optional[torch.optim.Optimizer] = None,
optimizer_params: Dict = {},
min_lr: float = 1e-8,
max_lr: float = 1,
num_training: int = 100,
mode: str = "exponential",
early_stop_threshold: float = 4.0,
plot=True,
train_sampler: Optional[torch.utils.data.Sampler] = None,
trained_backbone=None,
) -> None:
"""Enables the user to do a range test of good initial learning rates, to reduce the amount of guesswork in picking a good starting learning rate.
Args:
train (pd.DataFrame): Training Dataframe
validation (Optional[pd.DataFrame], optional): If provided, will use this dataframe as the validation while training.
Used in Early Stopping and Logging. If left empty, will use 20% of Train data as validation. Defaults to None.
test (Optional[pd.DataFrame], optional): If provided, will use as the hold-out data,
which you'll be able to check performance after the model is trained. Defaults to None.
loss (Optional[torch.nn.Module], optional): Custom Loss functions which are not in standard pytorch library
metrics (Optional[List[Callable]], optional): Custom metric functions(Callable) which has the signature metric_fn(y_hat, y)
optimizer (Optional[torch.optim.Optimizer], optional): Custom optimizers which are a drop in replacements for standard PyToch optimizers.
This should be the Class and not the initialized object
optimizer_params (Optional[Dict], optional): The parmeters to initialize the custom optimizer.
min_lr (Optional[float], optional): minimum learning rate to investigate
max_lr (Optional[float], optional): maximum learning rate to investigate
num_training (Optional[int], optional): number of learning rates to test
mode (Optional[str], optional): search strategy, either 'linear' or 'exponential'. If set to
'linear' the learning rate will be searched by linearly increasing
after each batch. If set to 'exponential', will increase learning
rate exponentially.
early_stop_threshold(Optional[float], optional): threshold for stopping the search. If the
loss at any point is larger than early_stop_threshold*best_loss
then the search is stopped. To disable, set to None.
plot(bool, optional): If true, will plot using matplotlib
trained_backbone (pl.LightningModule): this module contains the weights for a pretrained backbone
train_sampler (Optional[torch.utils.data.Sampler], optional): Custom PyTorch batch samplers which will be passed to the DataLoaders. Useful for dealing with imbalanced data and other custom batching strategies
"""
train_loader, val_loader = self._pre_fit(
train,
validation,
test,
loss,
metrics,
optimizer,
optimizer_params,
target_transform=None,
max_epochs=None,
min_epochs=None,
reset=True,
trained_backbone=trained_backbone,
train_sampler=train_sampler,
)
lr_finder = self.trainer.tuner.lr_find(
self.model,
train_loader,
val_loader,
min_lr,
max_lr,
num_training,
mode,
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
return new_lr, pd.DataFrame(lr_finder.results)
fit(self, train, validation=None, test=None, loss=None, metrics=None, optimizer=None, optimizer_params={}, train_sampler=None, target_transform=None, max_epochs=None, min_epochs=None, reset=False, seed=None, trained_backbone=None, callbacks=None)
The fit method which takes in the data and triggers the training
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
train |
pd.DataFrame |
Training Dataframe |
required |
validation |
Optional[pd.DataFrame] |
If provided, will use this dataframe as the validation while training. Used in Early Stopping and Logging. If left empty, will use 20% of Train data as validation. Defaults to None. |
None |
test |
Optional[pd.DataFrame] |
If provided, will use as the hold-out data, which you'll be able to check performance after the model is trained. Defaults to None. |
None |
loss |
Optional[torch.nn.Module] |
Custom Loss functions which are not in standard pytorch library |
None |
metrics |
Optional[List[Callable]] |
Custom metric functions(Callable) which has the signature metric_fn(y_hat, y) and works on torch tensor inputs |
None |
optimizer |
Optional[torch.optim.Optimizer] |
Custom optimizers which are a drop in replacements for standard PyToch optimizers. This should be the Class and not the initialized object |
None |
optimizer_params |
Optional[Dict] |
The parmeters to initialize the custom optimizer. |
{} |
train_sampler |
Optional[torch.utils.data.Sampler] |
Custom PyTorch batch samplers which will be passed to the DataLoaders. Useful for dealing with imbalanced data and other custom batching strategies |
None |
target_transform |
Optional[Union[TransformerMixin, Tuple(Callable)]] |
If provided, applies the transform to the target before modelling and inverse the transform during prediction. The parameter can either be a sklearn Transformer which has an inverse_transform method, or a tuple of callables (transform_func, inverse_transform_func) |
None |
max_epochs |
Optional[int] |
Overwrite maximum number of epochs to be run |
None |
min_epochs |
Optional[int] |
Overwrite minimum number of epochs to be run |
None |
reset |
bool |
(bool): Flag to reset the model and train again from scratch |
False |
seed |
Optional[int] |
(int): If you have to override the default seed set as part of of ModelConfig |
None |
trained_backbone |
pl.LightningModule |
this module contains the weights for a pretrained backbone |
None |
callbacks |
Optional[List[pl.Callback]] |
Custom callbacks to be used during training. |
None |
Source code in pytorch_tabular/tabular_model.py
def fit(
self,
train: pd.DataFrame,
validation: Optional[pd.DataFrame] = None,
test: Optional[pd.DataFrame] = None,
loss: Optional[torch.nn.Module] = None,
metrics: Optional[List[Callable]] = None,
optimizer: Optional[torch.optim.Optimizer] = None,
optimizer_params: Dict = {},
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,
reset: bool = False,
seed: Optional[int] = None,
trained_backbone: Optional[pl.LightningModule] = None,
callbacks: Optional[List[pl.Callback]] = None,
) -> None:
"""The fit method which takes in the data and triggers the training
Args:
train (pd.DataFrame): Training Dataframe
validation (Optional[pd.DataFrame], optional): If provided, will use this dataframe as the validation while training.
Used in Early Stopping and Logging. If left empty, will use 20% of Train data as validation. Defaults to None.
test (Optional[pd.DataFrame], optional): If provided, will use as the hold-out data,
which you'll be able to check performance after the model is trained. Defaults to None.
loss (Optional[torch.nn.Module], optional): Custom Loss functions which are not in standard pytorch library
metrics (Optional[List[Callable]], optional): Custom metric functions(Callable) which has the
signature metric_fn(y_hat, y) and works on torch tensor inputs
optimizer (Optional[torch.optim.Optimizer], optional): Custom optimizers which are a drop in replacements for standard PyToch optimizers.
This should be the Class and not the initialized object
optimizer_params (Optional[Dict], optional): The parmeters to initialize the custom optimizer.
train_sampler (Optional[torch.utils.data.Sampler], optional): Custom PyTorch batch samplers which will be passed to the DataLoaders. Useful for dealing with imbalanced data and other custom batching strategies
target_transform (Optional[Union[TransformerMixin, Tuple(Callable)]], optional): If provided, applies the transform to the target before modelling
and inverse the transform during prediction. The parameter can either be a sklearn Transformer which has an inverse_transform method, or
a tuple of callables (transform_func, inverse_transform_func)
max_epochs (Optional[int]): Overwrite maximum number of epochs to be run
min_epochs (Optional[int]): Overwrite minimum number of epochs to be run
reset: (bool): Flag to reset the model and train again from scratch
seed: (int): If you have to override the default seed set as part of of ModelConfig
trained_backbone (pl.LightningModule): this module contains the weights for a pretrained backbone
callbacks (Optional[List[pl.Callback]], optional): Custom callbacks to be used during training.
"""
seed_everything(seed if seed is not None else self.config.seed)
train_loader, val_loader = self._pre_fit(
train,
validation,
test,
loss,
metrics,
optimizer,
optimizer_params,
train_sampler,
target_transform,
max_epochs,
min_epochs,
reset,
trained_backbone,
callbacks,
)
self.model.train()
if self.config.auto_lr_find and (not self.config.fast_dev_run):
self.trainer.tune(self.model, train_loader, val_loader)
# Parameters in models needs to be initialized again after LR find
self.model.data_aware_initialization(self.datamodule)
self.model.train()
self.trainer.fit(self.model, train_loader, val_loader)
logger.info("Training the model completed...")
if self.config.load_best:
self.load_best_model()
load_best_model(self)
Loads the best model after training is done
Source code in pytorch_tabular/tabular_model.py
def load_best_model(self):
"""Loads the best model after training is done"""
if self.trainer.checkpoint_callback is not None:
logger.info("Loading the best model...")
ckpt_path = self.trainer.checkpoint_callback.best_model_path
if ckpt_path != "":
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.info(
"No best model available to load. Did you run it more than 1 epoch?..."
)
else:
logger.info(
"No best model available to load. Did you run it more than 1 epoch?..."
)
load_from_checkpoint(dir, map_location=None, strict=True)
classmethod
Loads a saved model from the directory
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
dir |
str |
The directory where the model wa saved, along with the checkpoints |
required |
strict |
bool) – Whether to strictly enforce that the keys in checkpoint_path match the keys
returned by this module’s state dict. Default |
True. |
True |
Returns:
| Type | Description |
|---|---|
TabularModel |
The saved TabularModel |
Source code in pytorch_tabular/tabular_model.py
@classmethod
def load_from_checkpoint(cls, dir: str, map_location=None, strict=True):
"""Loads a saved model from the directory
Args:
dir (str): The directory where the model wa saved, along with the checkpoints
map_location (Union[Dict[str, str], str, device, int, Callable, None]) – If your checkpoint
saved a GPU model and you now load on CPUs or a different number of GPUs, use this to map
to the new setup. The behaviour is the same as in torch.load()
strict (bool) – Whether to strictly enforce that the keys in checkpoint_path match the keys
returned by this module’s state dict. Default: True.
Returns:
TabularModel: The saved 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(
getattr(models, config._module_src), config._model_name
)
custom_model = False
custom_params = joblib.load(os.path.join(dir, "custom_params.sav"))
model_args = {}
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["metric_params"] = [{}] # 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
model = model_callable.load_from_checkpoint(
checkpoint_path=os.path.join(dir, "model.ckpt"),
map_location=map_location,
strict=strict,
**model_args,
)
# Updating config with custom parameters for experiment tracking
if custom_params.get("custom_loss") is not None:
model.custom_loss = custom_params["custom_loss"]
if custom_params.get("custom_metrics") is not None:
model.custom_metrics = custom_params["custom_metrics"]
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"]
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._prepare_trainer()
tabular_model.trainer.model = model
tabular_model.logger = logger
return tabular_model
predict(self, test, quantiles=[0.25, 0.5, 0.75], n_samples=100, ret_logits=False)
Uses the trained model to predict on new data and return as a dataframe
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
test |
pd.DataFrame |
The new dataframe with the features defined during training |
required |
quantiles |
Optional[List] |
For probabilistic models like Mixture Density Networks, this specifies
the different quantiles to be extracted apart from the |
[0.25, 0.5, 0.75] |
n_samples |
Optional[int] |
Number of samples to draw from the posterior to estimate the quantiles. Ignored for non-probabilistic models. Defaults to 100 |
100 |
ret_logits |
bool |
Flag to return raw model outputs/logits except the backbone features along with the dataframe. Defaults to False |
False |
Returns:
| Type | Description |
|---|---|
pd.DataFrame |
Returns a dataframe with predictions and features. If classification, it returns probabilities and final prediction |
Source code in pytorch_tabular/tabular_model.py
def predict(
self,
test: pd.DataFrame,
quantiles: Optional[List] = [0.25, 0.5, 0.75],
n_samples: Optional[int] = 100,
ret_logits=False,
) -> pd.DataFrame:
"""Uses the trained model to predict on new data and return as a dataframe
Args:
test (pd.DataFrame): The new dataframe with the features defined during training
quantiles (Optional[List]): For probabilistic models like Mixture Density Networks, this specifies
the different quantiles to be extracted apart from the `central_tendency` and added to the dataframe.
For other models it is ignored. Defaults to [0.25, 0.5, 0.75]
n_samples (Optional[int]): Number of samples to draw from the posterior to estimate the quantiles.
Ignored for non-probabilistic models. Defaults to 100
ret_logits (bool): Flag to return raw model outputs/logits except the backbone features along
with the dataframe. Defaults to False
Returns:
pd.DataFrame: Returns a dataframe with predictions and features.
If classification, it returns probabilities and final prediction
"""
assert all(
[q <= 1 and q >= 0 for q in quantiles]
), "Quantiles should be a decimal between 0 and 1"
self.model.eval()
inference_dataloader = self.datamodule.prepare_inference_dataloader(test)
point_predictions = []
quantile_predictions = []
logits_predictions = defaultdict(list)
is_probabilistic = (
hasattr(self.model.hparams, "_probabilistic")
and self.model.hparams._probabilistic
)
for batch in tqdm(inference_dataloader, desc="Generating Predictions..."):
for k, v in batch.items():
if isinstance(v, list) and (len(v) == 0):
# Skipping empty list
continue
batch[k] = v.to(self.model.device)
if is_probabilistic:
samples, ret_value = self.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 = self.model.predict(batch, ret_model_output=True)
if ret_logits:
for k, v in ret_value.items():
# if k == "backbone_features":
# continue
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)
pred_df = test.copy()
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):
pred_df[
f"{target_col}_q{int(q*100)}"
] = 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":
point_predictions = nn.Softmax(dim=-1)(point_predictions).numpy()
for i, class_ in enumerate(self.datamodule.label_encoder.classes_):
pred_df[f"{class_}_probability"] = point_predictions[:, i]
pred_df["prediction"] = self.datamodule.label_encoder.inverse_transform(
np.argmax(point_predictions, axis=1)
)
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
save_model(self, dir)
Saves the model and checkpoints in the specified directory
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
dir |
str |
The path to the directory to save the model |
required |
Source code in pytorch_tabular/tabular_model.py
def save_model(self, dir: str):
"""Saves the model and checkpoints in the specified directory
Args:
dir (str): The path to the directory to save the model
"""
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)
with open(os.path.join(dir, "config.yml"), "w") as fp:
OmegaConf.save(self.config, fp, resolve=True)
joblib.dump(self.datamodule, os.path.join(dir, "datamodule.sav"))
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"] = self.model.custom_loss
custom_params["custom_metrics"] = self.model.custom_metrics
custom_params["custom_optimizer"] = self.model.custom_optimizer
custom_params["custom_optimizer_params"] = self.model.custom_optimizer_params
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(self, path, kind='pytorch', onnx_export_params={})
Saves the model for inference path (Union[str, Path]): path to save the model kind (str): "pytorch" or "onnx" (Experimental) onnx_export_params (Dict): parameters for onnx export to be passed to torch.onnx.export
Source code in pytorch_tabular/tabular_model.py
def save_model_for_inference(
self,
path: Union[str, Path],
kind: str = "pytorch",
onnx_export_params: Dict = {},
):
"""Saves the model for inference
path (Union[str, Path]): path to save the model
kind (str): "pytorch" or "onnx" (Experimental)
onnx_export_params (Dict): parameters for onnx export to be
passed to torch.onnx.export
"""
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 = dict(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(self, path)
Saves the model weights in the specified directory
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path |
str |
The path to the directory to save the model |
required |
Source code in pytorch_tabular/tabular_model.py
def save_weights(self, path: Union[str, Path]):
"""Saves the model weights in the specified directory
Args:
path (str): The path to the directory to save the model
"""
torch.save(self.model.state_dict(), path)
pytorch_tabular.tabular_datamodule.TabularDatamodule (LightningDataModule)
Source code in pytorch_tabular/tabular_datamodule.py
class TabularDatamodule(pl.LightningDataModule):
CONTINUOUS_TRANSFORMS = {
"quantile_uniform": {
"callable": QuantileTransformer,
"params": dict(output_distribution="uniform", random_state=42),
},
"quantile_normal": {
"callable": QuantileTransformer,
"params": dict(output_distribution="normal", random_state=42),
},
"box-cox": {
"callable": PowerTransformer,
"params": dict(method="box-cox", standardize=False),
},
"yeo-johnson": {
"callable": PowerTransformer,
"params": dict(method="yeo-johnson", standardize=False),
},
}
def __init__(
self,
train: pd.DataFrame,
config: DictConfig,
validation: pd.DataFrame = None,
test: pd.DataFrame = None,
target_transform: Optional[Union[TransformerMixin, Tuple]] = None,
train_sampler: Optional[torch.utils.data.Sampler] = None,
):
"""The Pytorch Lightning Datamodule for Tabular Data
Args:
train (pd.DataFrame): The Training Dataframe
config (DictConfig): Merged configuration object from ModelConfig, DataConfig,
TrainerConfig, OptimizerConfig & ExperimentConfig
validation (pd.DataFrame, optional): Validation Dataframe.
If left empty, we use the validation split from DataConfig to split a random sample as validation.
Defaults to None.
test (pd.DataFrame, optional): Holdout DataFrame to check final performance on.
Defaults to None.
target_transform (Optional[Union[TransformerMixin, Tuple(Callable)]], optional): If provided, applies the transform to the target before modelling
and inverse the transform during prediction. The parameter can either be a sklearn Transformer which has an inverse_transform method, or
a tuple of callables (transform_func, inverse_transform_func)
"""
super().__init__()
self.train = train.copy()
self.validation = validation
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
self.test = test if test is None else test.copy()
self.target = config.target
self.batch_size = config.batch_size
self.train_sampler = train_sampler
self.config = config
self._fitted = False
def update_config(self) -> None:
"""Calculates and updates a few key information to the config object
Raises:
NotImplementedError: [description]
"""
if self.config.task == "regression":
self.config.output_dim = len(self.config.target)
elif self.config.task == "classification":
self.config.output_dim = len(self.train[self.config.target[0]].unique())
elif self.config.task == "ssl":
self.config.output_dim = len(self.config.categorical_cols) + len(
self.config.continuous_cols
)
if not self.do_leave_one_out_encoder():
self.config.categorical_cardinality = [
int(self.train[col].fillna("NA").nunique()) + 1
for col in self.config.categorical_cols
]
if (
hasattr(self.config, "embedding_dims")
and self.config.embedding_dims is None
):
self.config.embedding_dims = [
(x, min(50, (x + 1) // 2))
for x in self.config.categorical_cardinality
]
def do_leave_one_out_encoder(self) -> bool:
"""Checks the special condition for NODE where we use a LeaveOneOutEncoder to encode categorical columns
Returns:
bool
"""
return (self.config._model_name == "NODEModel") and (
not self.config.embed_categorical
)
def preprocess_data(
self, data: pd.DataFrame, stage: str = "inference"
) -> Tuple[pd.DataFrame, list]:
"""The preprocessing, like Categorical Encoding, Normalization, etc. which any dataframe should undergo before feeding into the dataloder
Args:
data (pd.DataFrame): A dataframe with the features and target
stage (str, optional): Internal parameter. Used to distinguisj between fit and inference. Defaults to "inference".
Returns:
tuple[pd.DataFrame, list]: Returns the processed dataframe and the added features(list) as a tuple
"""
logger.info(f"Preprocessing data: Stage: {stage}...")
added_features = None
if self.config.encode_date_columns:
for field_name, freq in self.config.date_columns:
data = self.make_date(data, field_name)
data, added_features = self.add_datepart(
data, field_name, frequency=freq, prefix=None, drop=True
)
# 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
self.config.categorical_dim = (
len(self.config.categorical_cols)
if self.config.categorical_cols is not None
else 0
)
# Encoding Categorical Columns
if len(self.config.categorical_cols) > 0:
if stage == "fit":
if self.do_leave_one_out_encoder():
logger.debug("Encoding Categorical Columns using LeavOneOutEncoder")
self.categorical_encoder = ce.LeaveOneOutEncoder(
cols=self.config.categorical_cols, random_state=42
)
# Multi-Target Regression uses the first target to encode the categorical columns
if len(self.config.target) > 1:
logger.warning(
f"Multi-Target Regression: using the first target({self.config.target[0]}) to encode the categorical columns"
)
data = self.categorical_encoder.fit_transform(
data, data[self.config.target[0]]
)
else:
logger.debug("Encoding Categorical Columns using OrdinalEncoder")
self.categorical_encoder = OrdinalEncoder(
cols=self.config.categorical_cols
)
data = self.categorical_encoder.fit_transform(data)
else:
data = self.categorical_encoder.transform(data)
# Transforming Continuous Columns
if (self.config.continuous_feature_transform is not None) and (
len(self.config.continuous_cols) > 0
):
if stage == "fit":
transform = self.CONTINUOUS_TRANSFORMS[
self.config.continuous_feature_transform
]
self.continuous_transform = transform["callable"](**transform["params"])
# TODO implement quantile noise
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]
)
# Normalizing Continuous Columns
if (self.config.normalize_continuous_features) and (
len(self.config.continuous_cols) > 0
):
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]
)
# Converting target labels to a 0 indexed label
if self.config.task == "classification":
if stage == "fit":
self.label_encoder = LabelEncoder()
data[self.config.target[0]] = self.label_encoder.fit_transform(
data[self.config.target[0]]
)
else:
if self.config.target[0] in data.columns:
data[self.config.target[0]] = self.label_encoder.transform(
data[self.config.target[0]]
)
# Target Transforms
if all([col in data.columns for col in self.config.target]):
if self.do_target_transform:
if stage == "fit":
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, added_features
def setup(self, stage: Optional[str] = None) -> None:
"""Data Operations you want to perform on all GPUs, like train-test split, transformations, etc.
This is called before accessing the dataloaders
Args:
stage (Optional[str], optional): Internal parameter to distinguish between fit and inference. Defaults to None.
"""
if stage == "fit" or stage is None:
if self.validation is None:
logger.debug(
f"No validation data provided. Using {self.config.validation_split*100}% of train data as validation"
)
val_idx = self.train.sample(
int(self.config.validation_split * len(self.train)), random_state=42
).index
self.validation = self.train[self.train.index.isin(val_idx)]
self.train = self.train[~self.train.index.isin(val_idx)]
else:
self.validation = self.validation.copy()
# Preprocessing Train, Validation
self.train, _ = self.preprocess_data(self.train, stage="fit")
self.validation, _ = self.preprocess_data(
self.validation, stage="inference"
)
if self.test is not None:
self.test, _ = self.preprocess_data(self.test, stage="inference")
# Calculating the categorical dims and embedding dims etc and updating the config
self.update_config()
self._fitted = True
# adapted from gluonts
@classmethod
def time_features_from_frequency_str(cls, freq_str: str) -> List[str]:
"""
Returns a list of time features that will be appropriate for the given frequency string.
Parameters
----------
freq_str
Frequency string of the form [multiple][granularity] such as "12H", "5min", "1D" etc.
"""
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",
"Week" "Day",
"Dayofweek",
"Dayofyear",
],
offsets.BusinessDay: [
"Month",
"Quarter",
"Is_quarter_end",
"Is_quarter_start",
"Is_year_end",
"Is_year_start",
"Is_month_start",
"Week" "Day",
"Dayofweek",
"Dayofyear",
],
offsets.Hour: [
"Month",
"Quarter",
"Is_quarter_end",
"Is_quarter_start",
"Is_year_end",
"Is_year_start",
"Is_month_start",
"Week" "Day",
"Dayofweek",
"Dayofyear",
"Hour",
],
offsets.Minute: [
"Month",
"Quarter",
"Is_quarter_end",
"Is_quarter_start",
"Is_year_end",
"Is_year_start",
"Is_month_start",
"Week" "Day",
"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: pd.DataFrame, date_field: str):
"Make sure `df[date_field]` is of the right date type."
field_dtype = df[date_field].dtype
if isinstance(field_dtype, pd.core.dtypes.dtypes.DatetimeTZDtype):
field_dtype = np.datetime64
if not np.issubdtype(field_dtype, np.datetime64):
df[date_field] = pd.to_datetime(df[date_field], infer_datetime_format=True)
return df
# adapted from fastai
@classmethod
def add_datepart(
cls,
df: pd.DataFrame,
field_name: str,
frequency: str,
prefix: str = None,
drop: bool = True,
):
"Helper function that adds columns relevant to a date in the column `field_name` of `df`."
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 train_dataloader(self, batch_size: Optional[int] = None) -> DataLoader:
"""Function that loads the train set."""
dataset = TabularDataset(
task=self.config.task,
data=self.train,
categorical_cols=self.config.categorical_cols,
continuous_cols=self.config.continuous_cols,
embed_categorical=(not self.do_leave_one_out_encoder()),
target=self.target,
)
return DataLoader(
dataset,
batch_size if batch_size is not None else 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) -> DataLoader:
"""Function that loads the validation set."""
dataset = TabularDataset(
task=self.config.task,
data=self.validation,
categorical_cols=self.config.categorical_cols,
continuous_cols=self.config.continuous_cols,
embed_categorical=(not self.do_leave_one_out_encoder()),
target=self.target,
)
return DataLoader(
dataset,
self.batch_size,
shuffle=False,
num_workers=self.config.num_workers,
pin_memory=self.config.pin_memory,
)
def test_dataloader(self) -> DataLoader:
"""Function that loads the validation set."""
if self.test is not None:
dataset = TabularDataset(
task=self.config.task,
data=self.test,
categorical_cols=self.config.categorical_cols,
continuous_cols=self.config.continuous_cols,
embed_categorical=(not self.do_leave_one_out_encoder()),
target=self.target,
)
return DataLoader(
dataset,
self.batch_size,
shuffle=False,
num_workers=self.config.num_workers,
pin_memory=self.config.pin_memory,
)
def prepare_inference_dataloader(self, df: pd.DataFrame) -> DataLoader:
"""Function that prepares and loads the new data.
Args:
df (pd.DataFrame): Dataframe with the features and target
Returns:
DataLoader: The dataloader for the passed in dataframe
"""
df = df.copy()
if len(set(self.target) - set(df.columns)) > 0:
if self.config.task == "classification":
df.loc[:, self.target] = np.array(
[self.label_encoder.classes_[0]] * len(df)
)
else:
df.loc[:, self.target] = np.zeros((len(df), len(self.target)))
df, _ = self.preprocess_data(df, stage="inference")
dataset = TabularDataset(
task=self.config.task,
data=df,
categorical_cols=self.config.categorical_cols,
continuous_cols=self.config.continuous_cols,
embed_categorical=(not self.do_leave_one_out_encoder()),
target=self.target
if all([col in df.columns for col in self.target])
else None,
)
return DataLoader(
dataset,
self.batch_size,
shuffle=False,
num_workers=self.config.num_workers,
)
__init__(self, train, config, validation=None, test=None, target_transform=None, train_sampler=None)
special
The Pytorch Lightning Datamodule for Tabular Data
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
train |
pd.DataFrame |
The Training Dataframe |
required |
config |
DictConfig |
Merged configuration object from ModelConfig, DataConfig, |
required |
validation |
pd.DataFrame |
Validation Dataframe. |
None |
test |
pd.DataFrame |
Holdout DataFrame to check final performance on. |
None |
target_transform |
Optional[Union[TransformerMixin, Tuple(Callable)]] |
If provided, applies the transform to the target before modelling |
None |
Source code in pytorch_tabular/tabular_datamodule.py
def __init__(
self,
train: pd.DataFrame,
config: DictConfig,
validation: pd.DataFrame = None,
test: pd.DataFrame = None,
target_transform: Optional[Union[TransformerMixin, Tuple]] = None,
train_sampler: Optional[torch.utils.data.Sampler] = None,
):
"""The Pytorch Lightning Datamodule for Tabular Data
Args:
train (pd.DataFrame): The Training Dataframe
config (DictConfig): Merged configuration object from ModelConfig, DataConfig,
TrainerConfig, OptimizerConfig & ExperimentConfig
validation (pd.DataFrame, optional): Validation Dataframe.
If left empty, we use the validation split from DataConfig to split a random sample as validation.
Defaults to None.
test (pd.DataFrame, optional): Holdout DataFrame to check final performance on.
Defaults to None.
target_transform (Optional[Union[TransformerMixin, Tuple(Callable)]], optional): If provided, applies the transform to the target before modelling
and inverse the transform during prediction. The parameter can either be a sklearn Transformer which has an inverse_transform method, or
a tuple of callables (transform_func, inverse_transform_func)
"""
super().__init__()
self.train = train.copy()
self.validation = validation
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
self.test = test if test is None else test.copy()
self.target = config.target
self.batch_size = config.batch_size
self.train_sampler = train_sampler
self.config = config
self._fitted = False
add_datepart(df, field_name, frequency, prefix=None, drop=True)
classmethod
Helper function that adds columns relevant to a date in the column field_name of df.
Source code in pytorch_tabular/tabular_datamodule.py
@classmethod
def add_datepart(
cls,
df: pd.DataFrame,
field_name: str,
frequency: str,
prefix: str = None,
drop: bool = True,
):
"Helper function that adds columns relevant to a date in the column `field_name` of `df`."
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
do_leave_one_out_encoder(self)
Checks the special condition for NODE where we use a LeaveOneOutEncoder to encode categorical columns
Returns:
| Type | Description |
|---|---|
bool |
bool |
Source code in pytorch_tabular/tabular_datamodule.py
def do_leave_one_out_encoder(self) -> bool:
"""Checks the special condition for NODE where we use a LeaveOneOutEncoder to encode categorical columns
Returns:
bool
"""
return (self.config._model_name == "NODEModel") and (
not self.config.embed_categorical
)
make_date(df, date_field)
classmethod
Make sure df[date_field] is of the right date type.
Source code in pytorch_tabular/tabular_datamodule.py
@classmethod
def make_date(cls, df: pd.DataFrame, date_field: str):
"Make sure `df[date_field]` is of the right date type."
field_dtype = df[date_field].dtype
if isinstance(field_dtype, pd.core.dtypes.dtypes.DatetimeTZDtype):
field_dtype = np.datetime64
if not np.issubdtype(field_dtype, np.datetime64):
df[date_field] = pd.to_datetime(df[date_field], infer_datetime_format=True)
return df
prepare_inference_dataloader(self, df)
Function that prepares and loads the new data.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
df |
pd.DataFrame |
Dataframe with the features and target |
required |
Returns:
| Type | Description |
|---|---|
DataLoader |
The dataloader for the passed in dataframe |
Source code in pytorch_tabular/tabular_datamodule.py
def prepare_inference_dataloader(self, df: pd.DataFrame) -> DataLoader:
"""Function that prepares and loads the new data.
Args:
df (pd.DataFrame): Dataframe with the features and target
Returns:
DataLoader: The dataloader for the passed in dataframe
"""
df = df.copy()
if len(set(self.target) - set(df.columns)) > 0:
if self.config.task == "classification":
df.loc[:, self.target] = np.array(
[self.label_encoder.classes_[0]] * len(df)
)
else:
df.loc[:, self.target] = np.zeros((len(df), len(self.target)))
df, _ = self.preprocess_data(df, stage="inference")
dataset = TabularDataset(
task=self.config.task,
data=df,
categorical_cols=self.config.categorical_cols,
continuous_cols=self.config.continuous_cols,
embed_categorical=(not self.do_leave_one_out_encoder()),
target=self.target
if all([col in df.columns for col in self.target])
else None,
)
return DataLoader(
dataset,
self.batch_size,
shuffle=False,
num_workers=self.config.num_workers,
)
preprocess_data(self, 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 |
pd.DataFrame |
A dataframe with the features and target |
required |
stage |
str |
Internal parameter. Used to distinguisj between fit and inference. Defaults to "inference". |
'inference' |
Returns:
| Type | Description |
|---|---|
tuple[pd.DataFrame, list] |
Returns the processed dataframe and the added features(list) as a tuple |
Source code in pytorch_tabular/tabular_datamodule.py
def preprocess_data(
self, data: pd.DataFrame, stage: str = "inference"
) -> Tuple[pd.DataFrame, list]:
"""The preprocessing, like Categorical Encoding, Normalization, etc. which any dataframe should undergo before feeding into the dataloder
Args:
data (pd.DataFrame): A dataframe with the features and target
stage (str, optional): Internal parameter. Used to distinguisj between fit and inference. Defaults to "inference".
Returns:
tuple[pd.DataFrame, list]: Returns the processed dataframe and the added features(list) as a tuple
"""
logger.info(f"Preprocessing data: Stage: {stage}...")
added_features = None
if self.config.encode_date_columns:
for field_name, freq in self.config.date_columns:
data = self.make_date(data, field_name)
data, added_features = self.add_datepart(
data, field_name, frequency=freq, prefix=None, drop=True
)
# 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
self.config.categorical_dim = (
len(self.config.categorical_cols)
if self.config.categorical_cols is not None
else 0
)
# Encoding Categorical Columns
if len(self.config.categorical_cols) > 0:
if stage == "fit":
if self.do_leave_one_out_encoder():
logger.debug("Encoding Categorical Columns using LeavOneOutEncoder")
self.categorical_encoder = ce.LeaveOneOutEncoder(
cols=self.config.categorical_cols, random_state=42
)
# Multi-Target Regression uses the first target to encode the categorical columns
if len(self.config.target) > 1:
logger.warning(
f"Multi-Target Regression: using the first target({self.config.target[0]}) to encode the categorical columns"
)
data = self.categorical_encoder.fit_transform(
data, data[self.config.target[0]]
)
else:
logger.debug("Encoding Categorical Columns using OrdinalEncoder")
self.categorical_encoder = OrdinalEncoder(
cols=self.config.categorical_cols
)
data = self.categorical_encoder.fit_transform(data)
else:
data = self.categorical_encoder.transform(data)
# Transforming Continuous Columns
if (self.config.continuous_feature_transform is not None) and (
len(self.config.continuous_cols) > 0
):
if stage == "fit":
transform = self.CONTINUOUS_TRANSFORMS[
self.config.continuous_feature_transform
]
self.continuous_transform = transform["callable"](**transform["params"])
# TODO implement quantile noise
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]
)
# Normalizing Continuous Columns
if (self.config.normalize_continuous_features) and (
len(self.config.continuous_cols) > 0
):
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]
)
# Converting target labels to a 0 indexed label
if self.config.task == "classification":
if stage == "fit":
self.label_encoder = LabelEncoder()
data[self.config.target[0]] = self.label_encoder.fit_transform(
data[self.config.target[0]]
)
else:
if self.config.target[0] in data.columns:
data[self.config.target[0]] = self.label_encoder.transform(
data[self.config.target[0]]
)
# Target Transforms
if all([col in data.columns for col in self.config.target]):
if self.do_target_transform:
if stage == "fit":
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, added_features
setup(self, stage=None)
Data Operations you want to perform on all GPUs, like train-test split, transformations, etc. This is called before accessing the dataloaders
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
stage |
Optional[str] |
Internal parameter to distinguish between fit and inference. Defaults to None. |
None |
Source code in pytorch_tabular/tabular_datamodule.py
def setup(self, stage: Optional[str] = None) -> None:
"""Data Operations you want to perform on all GPUs, like train-test split, transformations, etc.
This is called before accessing the dataloaders
Args:
stage (Optional[str], optional): Internal parameter to distinguish between fit and inference. Defaults to None.
"""
if stage == "fit" or stage is None:
if self.validation is None:
logger.debug(
f"No validation data provided. Using {self.config.validation_split*100}% of train data as validation"
)
val_idx = self.train.sample(
int(self.config.validation_split * len(self.train)), random_state=42
).index
self.validation = self.train[self.train.index.isin(val_idx)]
self.train = self.train[~self.train.index.isin(val_idx)]
else:
self.validation = self.validation.copy()
# Preprocessing Train, Validation
self.train, _ = self.preprocess_data(self.train, stage="fit")
self.validation, _ = self.preprocess_data(
self.validation, stage="inference"
)
if self.test is not None:
self.test, _ = self.preprocess_data(self.test, stage="inference")
# Calculating the categorical dims and embedding dims etc and updating the config
self.update_config()
self._fitted = True
test_dataloader(self)
Function that loads the validation set.
Source code in pytorch_tabular/tabular_datamodule.py
def test_dataloader(self) -> DataLoader:
"""Function that loads the validation set."""
if self.test is not None:
dataset = TabularDataset(
task=self.config.task,
data=self.test,
categorical_cols=self.config.categorical_cols,
continuous_cols=self.config.continuous_cols,
embed_categorical=(not self.do_leave_one_out_encoder()),
target=self.target,
)
return DataLoader(
dataset,
self.batch_size,
shuffle=False,
num_workers=self.config.num_workers,
pin_memory=self.config.pin_memory,
)
time_features_from_frequency_str(freq_str)
classmethod
Returns a list of time features that will be appropriate for the given frequency string.
Parameters
freq_str Frequency string of the form [multiple][granularity] such as "12H", "5min", "1D" etc.
Source code in pytorch_tabular/tabular_datamodule.py
@classmethod
def time_features_from_frequency_str(cls, freq_str: str) -> List[str]:
"""
Returns a list of time features that will be appropriate for the given frequency string.
Parameters
----------
freq_str
Frequency string of the form [multiple][granularity] such as "12H", "5min", "1D" etc.
"""
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",
"Week" "Day",
"Dayofweek",
"Dayofyear",
],
offsets.BusinessDay: [
"Month",
"Quarter",
"Is_quarter_end",
"Is_quarter_start",
"Is_year_end",
"Is_year_start",
"Is_month_start",
"Week" "Day",
"Dayofweek",
"Dayofyear",
],
offsets.Hour: [
"Month",
"Quarter",
"Is_quarter_end",
"Is_quarter_start",
"Is_year_end",
"Is_year_start",
"Is_month_start",
"Week" "Day",
"Dayofweek",
"Dayofyear",
"Hour",
],
offsets.Minute: [
"Month",
"Quarter",
"Is_quarter_end",
"Is_quarter_start",
"Is_year_end",
"Is_year_start",
"Is_month_start",
"Week" "Day",
"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(self, batch_size=None)
Function that loads the train set.
Source code in pytorch_tabular/tabular_datamodule.py
def train_dataloader(self, batch_size: Optional[int] = None) -> DataLoader:
"""Function that loads the train set."""
dataset = TabularDataset(
task=self.config.task,
data=self.train,
categorical_cols=self.config.categorical_cols,
continuous_cols=self.config.continuous_cols,
embed_categorical=(not self.do_leave_one_out_encoder()),
target=self.target,
)
return DataLoader(
dataset,
batch_size if batch_size is not None else 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(self)
Calculates and updates a few key information to the config object
Exceptions:
| Type | Description |
|---|---|
NotImplementedError |
[description] |
Source code in pytorch_tabular/tabular_datamodule.py
def update_config(self) -> None:
"""Calculates and updates a few key information to the config object
Raises:
NotImplementedError: [description]
"""
if self.config.task == "regression":
self.config.output_dim = len(self.config.target)
elif self.config.task == "classification":
self.config.output_dim = len(self.train[self.config.target[0]].unique())
elif self.config.task == "ssl":
self.config.output_dim = len(self.config.categorical_cols) + len(
self.config.continuous_cols
)
if not self.do_leave_one_out_encoder():
self.config.categorical_cardinality = [
int(self.train[col].fillna("NA").nunique()) + 1
for col in self.config.categorical_cols
]
if (
hasattr(self.config, "embedding_dims")
and self.config.embedding_dims is None
):
self.config.embedding_dims = [
(x, min(50, (x + 1) // 2))
for x in self.config.categorical_cardinality
]
val_dataloader(self)
Function that loads the validation set.
Source code in pytorch_tabular/tabular_datamodule.py
def val_dataloader(self) -> DataLoader:
"""Function that loads the validation set."""
dataset = TabularDataset(
task=self.config.task,
data=self.validation,
categorical_cols=self.config.categorical_cols,
continuous_cols=self.config.continuous_cols,
embed_categorical=(not self.do_leave_one_out_encoder()),
target=self.target,
)
return DataLoader(
dataset,
self.batch_size,
shuffle=False,
num_workers=self.config.num_workers,
pin_memory=self.config.pin_memory,
)
pytorch_tabular.models.category_embedding.category_embedding_model.CategoryEmbeddingModel (BaseModel)
Source code in pytorch_tabular/models/category_embedding/category_embedding_model.py
class CategoryEmbeddingModel(BaseModel):
def __init__(self, config: DictConfig, **kwargs):
# The concatenated output dim of the embedding layer
self.embedding_cat_dim = sum([y for x, y in config.embedding_dims])
super().__init__(config, **kwargs)
def _build_network(self):
# Backbone
self.backbone = CategoryEmbeddingBackbone(self.hparams)
# Adding the last layer
self.head = nn.Linear(
self.backbone.output_dim, self.hparams.output_dim
) # output_dim auto-calculated from other config
_initialize_layers(
self.hparams.activation, self.hparams.initialization, self.head
)
def extract_embedding(self):
return self.backbone.embedding_layers
pytorch_tabular.models.category_embedding.config.CategoryEmbeddingModelConfig (ModelConfig)
dataclass
CategoryEmbeddingModel configuration
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
task |
str |
Specify whether the problem is regression of classification.Choices are: regression classification |
required |
learning_rate |
float |
The learning rate of the model |
0.001 |
loss |
Optional[str] |
The loss function to be applied. By Default it is MSELoss for regression and CrossEntropyLoss for classification. Unless you are sure what you are doing, leave it at MSELoss or L1Loss for regression and CrossEntropyLoss for classification |
None |
metrics |
Optional[List[str]] |
the list of metrics you need to track during training. The metrics should be one of the metrics implemented in PyTorch Lightning. By default, it is Accuracy if classification and MeanSquaredLogError for regression |
None |
metrics_params |
Optional[List] |
The parameters to be passed to the Metrics initialized |
None |
target_range |
Optional[List] |
The range in which we should limit the output variable. Currently ignored for multi-target regression Typically used for Regression problems. If left empty, will not apply any restrictions |
None |
layers |
str |
Hyphen-separated number of layers and units in the classification head. eg. 32-64-32. |
'128-64-32' |
batch_norm_continuous_input |
bool |
If True, we will normalize the contiinuous layer by passing it through a BatchNorm layer |
True |
activation |
str |
The activation type in the classification head. The default activation in PyTorch like ReLU, TanH, LeakyReLU, etc. https://pytorch.org/docs/stable/nn.html#non-linear-activations-weighted-sum-nonlinearity |
'ReLU' |
embedding_dims |
Optional[List[int]] |
The dimensions of the embedding for each categorical column as a list of tuples (cardinality, embedding_dim). If left empty, will infer using the cardinality of the categorical column using the rule min(50, (x + 1) // 2) |
None |
embedding_dropout |
float |
probability of an embedding element to be zeroed. |
0.5 |
dropout |
float |
probability of an classification element to be zeroed. |
0.5 |
use_batch_norm |
bool |
Flag to include a BatchNorm layer after each Linear Layer+DropOut |
False |
initialization |
str |
Initialization scheme for the linear layers. Choices are: |
'kaiming' |
Exceptions:
| Type | Description |
|---|---|
NotImplementedError |
Raises an error if task is not in ['regression','classification'] |
Source code in pytorch_tabular/models/category_embedding/config.py
class CategoryEmbeddingModelConfig(ModelConfig):
"""CategoryEmbeddingModel configuration
Args:
task (str): Specify whether the problem is regression of classification.Choices are: regression classification
learning_rate (float): The learning rate of the model
loss (Union[str, NoneType]): The loss function to be applied.
By Default it is MSELoss for regression and CrossEntropyLoss for classification.
Unless you are sure what you are doing, leave it at MSELoss or L1Loss for regression and CrossEntropyLoss for classification
metrics (Union[List[str], NoneType]): the list of metrics you need to track during training.
The metrics should be one of the metrics implemented in PyTorch Lightning.
By default, it is Accuracy if classification and MeanSquaredLogError for regression
metrics_params (Union[List, NoneType]): The parameters to be passed to the Metrics initialized
target_range (Union[List, NoneType]): The range in which we should limit the output variable. Currently ignored for multi-target regression
Typically used for Regression problems. If left empty, will not apply any restrictions
layers (str): Hyphen-separated number of layers and units in the classification head. eg. 32-64-32.
batch_norm_continuous_input (bool): If True, we will normalize the contiinuous layer by passing it through a BatchNorm layer
activation (str): The activation type in the classification head.
The default activation in PyTorch like ReLU, TanH, LeakyReLU, etc.
https://pytorch.org/docs/stable/nn.html#non-linear-activations-weighted-sum-nonlinearity
embedding_dims (Union[List[int], NoneType]): The dimensions of the embedding for each categorical column
as a list of tuples (cardinality, embedding_dim). If left empty, will infer using the cardinality of the categorical column
using the rule min(50, (x + 1) // 2)
embedding_dropout (float): probability of an embedding element to be zeroed.
dropout (float): probability of an classification element to be zeroed.
use_batch_norm (bool): Flag to include a BatchNorm layer after each Linear Layer+DropOut
initialization (str): Initialization scheme for the linear layers. Choices are: `kaiming` `xavier` `random`
Raises:
NotImplementedError: Raises an error if task is not in ['regression','classification']
"""
layers: str = field(
default="128-64-32",
metadata={
"help": "Hyphen-separated number of layers and units in the classification head. eg. 32-64-32."
},
)
batch_norm_continuous_input: bool = field(
default=True,
metadata={
"help": "If True, we will normalize the contiinuous layer by passing it through a BatchNorm layer"
},
)
activation: str = field(
default="ReLU",
metadata={
"help": "The activation type in the classification head. The default activaion in PyTorch like ReLU, TanH, LeakyReLU, etc. https://pytorch.org/docs/stable/nn.html#non-linear-activations-weighted-sum-nonlinearity"
},
)
embedding_dropout: float = field(
default=0.5,
metadata={"help": "probability of an embedding element to be zeroed."},
)
dropout: float = field(
default=0.5,
metadata={"help": "probability of an classification element to be zeroed."},
)
use_batch_norm: bool = field(
default=False,
metadata={
"help": "Flag to include a BatchNorm layer after each Linear Layer+DropOut"
},
)
initialization: str = field(
default="kaiming",
metadata={
"help": "Initialization scheme for the linear layers",
"choices": ["kaiming", "xavier", "random"],
},
)
_module_src: str = field(default="category_embedding")
_model_name: str = field(default="CategoryEmbeddingModel")
_config_name: str = field(default="CategoryEmbeddingModelConfig")
pytorch_tabular.models.node.config.NodeConfig (ModelConfig)
dataclass
Model configuration
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
task |
str |
Specify whether the problem is regression of classification.Choices are: regression classification |
required |
learning_rate |
float |
The learning rate of the model |
0.001 |
loss |
Optional[str] |
The loss function to be applied. By Default it is MSELoss for regression and CrossEntropyLoss for classification. Unless you are sure what you are doing, leave it at MSELoss or L1Loss for regression and CrossEntropyLoss for classification |
None |
metrics |
Optional[List[str]] |
the list of metrics you need to track during training. The metrics should be one of the metrics implemented in PyTorch Lightning. By default, it is Accuracy if classification and MeanSquaredLogError for regression |
None |
metrics_params |
Optional[List] |
The parameters to be passed to the Metrics initialized |
None |
target_range |
Optional[List] |
The range in which we should limit the output variable. Currently ignored for multi-target regression Typically used for Regression problems. If left empty, will not apply any restrictions |
None |
num_layers |
int |
Number of Oblivious Decision Tree Layers in the Dense Architecture |
1 |
num_trees |
int |
Number of Oblivious Decision Trees in each layer |
2048 |
additional_tree_output_dim |
int |
The additional output dimensions which is only used to
pass through different layers of the architectures. Only the first |
3 |
depth |
int |
The depth of the individual Oblivious Decision Trees |
6 |
choice_function |
str |
Generates a sparse probability distribution to be used as feature weights(aka, soft feature selection) Choices are: ['entmax15', 'sparsemax'] |
'entmax15' |
bin_function |
str |
Generates a sparse probability distribution to be used as tree leaf weights Choices are: ['entmoid15', 'sparsemoid'] |
'entmoid15' |
max_features |
Optional[int] |
If not None, sets a max limit on the number of features to be carried forward from layer to layer in the Dense Architecture |
None |
input_dropout |
float |
Dropout to be applied to the inputs between layers of the Dense Architecture |
0.0 |
initialize_response |
str |
Initializing the response variable in the Oblivious Decision Trees. By default, it is a standard normal distribution. Choices are: ['normal', 'uniform'] |
'normal' |
initialize_selection_logits |
str |
Initializing the feature selector. By default is a uniform distribution across the features. Choices are: ['uniform', 'normal'] |
'uniform' |
threshold_init_beta |
float |
Used in the Data-aware initialization of thresholds where the threshold is initialized randomly (with a beta distribution) to feature values in the first batch. It initializes threshold to a q-th quantile of data points. where q ~ Beta(:threshold_init_beta:, :threshold_init_beta:) If this param is set to 1, initial thresholds will have the same distribution as data points If greater than 1 (e.g. 10), thresholds will be closer to median data value If less than 1 (e.g. 0.1), thresholds will approach min/max data values. |
1.0 |
threshold_init_cutoff |
float |
Used in the Data-aware initialization of scales(used in the scaling ODTs). It is initialized in such a way that all the samples in the first batch belong to the linear region of the entmoid/sparsemoid(bin-selectors) and thereby have non-zero gradients Threshold log-temperatures initializer, in (0, inf) By default(1.0), log-temperatures are initialized in such a way that all bin selectors end up in the linear region of sparse-sigmoid. The temperatures are then scaled by this parameter. Setting this value > 1.0 will result in some margin between data points and sparse-sigmoid cutoff value Setting this value < 1.0 will cause (1 - value) part of data points to end up in flat sparse-sigmoid region For instance, threshold_init_cutoff = 0.9 will set 10% points equal to 0.0 or 1.0 Setting this value > 1.0 will result in a margin between data points and sparse-sigmoid cutoff value All points will be between (0.5 - 0.5 / threshold_init_cutoff) and (0.5 + 0.5 / threshold_init_cutoff) |
1.0 |
embed_categorical |
bool |
Flag to embed categorical columns using an Embedding Layer. If turned off, the categorical columns are encoded using LeaveOneOutEncoder |
False |
embedding_dims |
Optional[List[int]] |
The dimensions of the embedding for each categorical column as a list of tuples (cardinality, embedding_dim). If left empty, will infer using the cardinality of the categorical column using the rule min(50, (x + 1) // 2) |
None |
embedding_dropout |
float |
probability of an embedding element to be zeroed. |
0.0 |
Exceptions:
| Type | Description |
|---|---|
NotImplementedError |
Raises an error if task is not in ['regression','classification'] |
Source code in pytorch_tabular/models/node/config.py
class NodeConfig(ModelConfig):
"""Model configuration
Args:
task (str): Specify whether the problem is regression of classification.Choices are: regression classification
learning_rate (float): The learning rate of the model
loss (Union[str, NoneType]): The loss function to be applied.
By Default it is MSELoss for regression and CrossEntropyLoss for classification.
Unless you are sure what you are doing, leave it at MSELoss or L1Loss for regression and CrossEntropyLoss for classification
metrics (Union[List[str], NoneType]): the list of metrics you need to track during training.
The metrics should be one of the metrics implemented in PyTorch Lightning.
By default, it is Accuracy if classification and MeanSquaredLogError for regression
metrics_params (Union[List, NoneType]): The parameters to be passed to the Metrics initialized
target_range (Union[List, NoneType]): The range in which we should limit the output variable. Currently ignored for multi-target regression
Typically used for Regression problems. If left empty, will not apply any restrictions
num_layers (int): Number of Oblivious Decision Tree Layers in the Dense Architecture
num_trees (int): Number of Oblivious Decision Trees in each layer
additional_tree_output_dim (int): The additional output dimensions which is only used to
pass through different layers of the architectures. Only the first `output_dim` outputs will be used for prediction
depth (int): The depth of the individual Oblivious Decision Trees
choice_function (str): Generates a sparse probability distribution to be used as feature weights(aka, soft feature selection)
Choices are: ['entmax15', 'sparsemax']
bin_function (str): Generates a sparse probability distribution to be used as tree leaf weights
Choices are: ['entmoid15', 'sparsemoid']
max_features (Union[int, NoneType]): If not None, sets a max limit on the number of features to be carried forward from layer to layer in the Dense Architecture
input_dropout (float): Dropout to be applied to the inputs between layers of the Dense Architecture
initialize_response (str): Initializing the response variable in the Oblivious Decision Trees.
By default, it is a standard normal distribution. Choices are: ['normal', 'uniform']
initialize_selection_logits (str): Initializing the feature selector.
By default is a uniform distribution across the features. Choices are: ['uniform', 'normal']
threshold_init_beta (float):
Used in the Data-aware initialization of thresholds where the threshold is initialized randomly
(with a beta distribution) to feature values in the first batch.
It initializes threshold to a q-th quantile of data points.
where q ~ Beta(:threshold_init_beta:, :threshold_init_beta:)
If this param is set to 1, initial thresholds will have the same distribution as data points
If greater than 1 (e.g. 10), thresholds will be closer to median data value
If less than 1 (e.g. 0.1), thresholds will approach min/max data values.
threshold_init_cutoff (float):
Used in the Data-aware initialization of scales(used in the scaling ODTs).
It is initialized in such a way that all the samples in the first batch belong to the linear
region of the entmoid/sparsemoid(bin-selectors) and thereby have non-zero gradients
Threshold log-temperatures initializer, in (0, inf)
By default(1.0), log-temperatures are initialized in such a way that all bin selectors
end up in the linear region of sparse-sigmoid. The temperatures are then scaled by this parameter.
Setting this value > 1.0 will result in some margin between data points and sparse-sigmoid cutoff value
Setting this value < 1.0 will cause (1 - value) part of data points to end up in flat sparse-sigmoid region
For instance, threshold_init_cutoff = 0.9 will set 10% points equal to 0.0 or 1.0
Setting this value > 1.0 will result in a margin between data points and sparse-sigmoid cutoff value
All points will be between (0.5 - 0.5 / threshold_init_cutoff) and (0.5 + 0.5 / threshold_init_cutoff)
embed_categorical (bool): Flag to embed categorical columns using an Embedding Layer.
If turned off, the categorical columns are encoded using LeaveOneOutEncoder
embedding_dims (Union[List[int], NoneType]): The dimensions of the embedding for each categorical column as a
list of tuples (cardinality, embedding_dim). If left empty, will infer using the cardinality of the categorical column
using the rule min(50, (x + 1) // 2)
embedding_dropout (float): probability of an embedding element to be zeroed.
Raises:
NotImplementedError: Raises an error if task is not in ['regression','classification']
"""
num_layers: int = field(
default=1,
metadata={
"help": "Number of Oblivious Decision Tree Layers in the Dense Architecture"
},
)
num_trees: int = field(
default=2048,
metadata={"help": "Number of Oblivious Decision Trees in each layer"},
)
additional_tree_output_dim: int = field(
default=3,
metadata={
"help": "The additional output dimensions which is only used to pass through different layers of the architectures. Only the first output_dim outputs will be used for prediction"
},
)
depth: int = field(
default=6,
metadata={"help": "The depth of the individual Oblivious Decision Trees"},
)
choice_function: str = field(
default="entmax15",
metadata={
"help": "Generates a sparse probability distribution to be used as feature weights(aka, soft feature selection)",
"choices": ["entmax15", "sparsemax"],
},
)
bin_function: str = field(
default="entmoid15",
metadata={
"help": "Generates a sparse probability distribution to be used as tree leaf weights",
"choices": ["entmoid15", "sparsemoid"],
},
)
max_features: Optional[int] = field(
default=None,
metadata={
"help": "If not None, sets a max limit on the number of features to be carried forward from layer to layer in the Dense Architecture"
},
)
input_dropout: float = field(
default=0.0,
metadata={
"help": "Dropout to be applied to the inputs between layers of the Dense Architecture"
},
)
initialize_response: str = field(
default="normal",
metadata={
"help": "Initializing the response variable in the Oblivious Decision Trees. By default, it is a standard normal distribution",
"choices": ["normal", "uniform"],
},
)
initialize_selection_logits: str = field(
default="uniform",
metadata={
"help": "Initializing the feature selector. By default is a uniform distribution across the features",
"choices": ["uniform", "normal"],
},
)
threshold_init_beta: float = field(
default=1.0,
metadata={
"help": """
Used in the Data-aware initialization of thresholds where the threshold is initialized randomly
(with a beta distribution) to feature values in the first batch.
It initializes threshold to a q-th quantile of data points.
where q ~ Beta(:threshold_init_beta:, :threshold_init_beta:)
If this param is set to 1, initial thresholds will have the same distribution as data points
If greater than 1 (e.g. 10), thresholds will be closer to median data value
If less than 1 (e.g. 0.1), thresholds will approach min/max data values.
"""
},
)
threshold_init_cutoff: float = field(
default=1.0,
metadata={
"help": """
Used in the Data-aware initialization of scales(used in the scaling ODTs).
It is initialized in such a way that all the samples in the first batch belong to the linear
region of the entmoid/sparsemoid(bin-selectors) and thereby have non-zero gradients
Threshold log-temperatures initializer, in (0, inf)
By default(1.0), log-temperatures are initialized in such a way that all bin selectors
end up in the linear region of sparse-sigmoid. The temperatures are then scaled by this parameter.
Setting this value > 1.0 will result in some margin between data points and sparse-sigmoid cutoff value
Setting this value < 1.0 will cause (1 - value) part of data points to end up in flat sparse-sigmoid region
For instance, threshold_init_cutoff = 0.9 will set 10% points equal to 0.0 or 1.0
Setting this value > 1.0 will result in a margin between data points and sparse-sigmoid cutoff value
All points will be between (0.5 - 0.5 / threshold_init_cutoff) and (0.5 + 0.5 / threshold_init_cutoff)
"""
},
)
embed_categorical: bool = field(
default=False,
metadata={
"help": "Flag to embed categorical columns using an Embedding Layer. If turned off, the categorical columns are encoded using LeaveOneOutEncoder"
},
)
embedding_dropout: float = field(
default=0.0,
metadata={"help": "probability of an embedding element to be zeroed."},
)
_module_src: str = field(default="node")
_model_name: str = field(default="NODEModel")
_config_name: str = field(default="NodeConfig")
pytorch_tabular.models.node.node_model.NODEModel (BaseModel)
Source code in pytorch_tabular/models/node/node_model.py
class NODEModel(BaseModel):
def __init__(self, config: DictConfig, **kwargs):
super().__init__(config, **kwargs)
def subset(self, x):
return x[..., : self.hparams.output_dim].mean(dim=-2)
def data_aware_initialization(self, datamodule):
"""Performs data-aware initialization for NODE"""
logger.info("Data Aware Initialization....")
# Need a big batch to initialize properly
alt_loader = datamodule.train_dataloader(batch_size=2000)
batch = next(iter(alt_loader))
for k, v in batch.items():
if isinstance(v, list) and (len(v) == 0):
# Skipping empty list
continue
# batch[k] = v.to("cpu" if self.config.gpu == 0 else "cuda")
batch[k] = v.to(self.device)
# single forward pass to initialize the ODST
with torch.no_grad():
self(batch)
def _build_network(self):
self.backbone = NODEBackbone(self.hparams)
# average first n channels of every tree, where n is the number of output targets for regression
# and number of classes for classification
self.head = utils.Lambda(self.subset)
def extract_embedding(self):
if self.hparams.embed_categorical:
if self.backbone.embedding_cat_dim != 0:
return self.backbone.embedding_layers
else:
raise ValueError(
"Model has been trained with no categorical feature and therefore can't be used as a Categorical Encoder"
)
data_aware_initialization(self, datamodule)
Performs data-aware initialization for NODE
Source code in pytorch_tabular/models/node/node_model.py
def data_aware_initialization(self, datamodule):
"""Performs data-aware initialization for NODE"""
logger.info("Data Aware Initialization....")
# Need a big batch to initialize properly
alt_loader = datamodule.train_dataloader(batch_size=2000)
batch = next(iter(alt_loader))
for k, v in batch.items():
if isinstance(v, list) and (len(v) == 0):
# Skipping empty list
continue
# batch[k] = v.to("cpu" if self.config.gpu == 0 else "cuda")
batch[k] = v.to(self.device)
# single forward pass to initialize the ODST
with torch.no_grad():
self(batch)
pytorch_tabular.models.tabnet.tabnet_model.TabNetModel (BaseModel)
Source code in pytorch_tabular/models/tabnet/tabnet_model.py
class TabNetModel(BaseModel):
def __init__(self, config: DictConfig, **kwargs):
assert config.task in ["regression", "classification"], "TabNet is only implemented for Regression and Classification"
super().__init__(config, **kwargs)
def _build_network(self):
self.backbone = TabNetBackbone(self.hparams)
setattr(self.backbone, "output_dim", self.hparams.output_dim)
self.head = nn.Identity()
pytorch_tabular.models.tabnet.config.TabNetModelConfig (ModelConfig)
dataclass
Model configuration
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
task |
str |
Specify whether the problem is regression of classification.Choices are: regression classification |
required |
learning_rate |
float |
The learning rate of the model |
0.001 |
loss |
Optional[str] |
The loss function to be applied. |
None |
metrics |
Optional[List[str]] |
the list of metrics you need to track during training. |
None |
metrics_params |
Optional[List] |
The parameters to be passed to the Metrics initialized |
None |
target_range |
Optional[List] |
The range in which we should limit the output variable. Currently ignored for multi-target regression |
None |
n_d |
int |
Dimension of the prediction layer (usually between 4 and 64) |
8 |
n_a |
int |
Dimension of the attention layer (usually between 4 and 64) |
8 |
n_steps |
int |
Number of sucessive steps in the newtork (usually betwenn 3 and 10) |
3 |
gamma |
float |
Float above 1, scaling factor for attention updates (usually betwenn 1.0 to 2.0) |
1.3 |
embedding_dims |
Optional[List[int]] |
The dimensions of the embedding for each categorical column as |
None |
n_independent |
int |
Number of independent GLU layer in each GLU block (default 2) |
2 |
n_shared |
int |
Number of independent GLU layer in each GLU block (default 2) |
2 |
virtual_batch_size |
int |
Batch size for Ghost Batch Normalization |
128 |
mask_type |
str |
Either 'sparsemax' or 'entmax' : this is the masking function to useChoices are: sparsemax entmax |
'sparsemax' |
Exceptions:
| Type | Description |
|---|---|
NotImplementedError |
Raises an error if task is not in ['regression','classification'] |
Source code in pytorch_tabular/models/tabnet/config.py
class TabNetModelConfig(ModelConfig):
"""Model configuration
Args:
task (str): Specify whether the problem is regression of classification.Choices are: regression classification
learning_rate (float): The learning rate of the model
loss (Union[str, NoneType]): The loss function to be applied.
By Default it is MSELoss for regression and CrossEntropyLoss for classification.
Unless you are sure what you are doing, leave it at MSELoss or L1Loss for regression and CrossEntropyLoss for classification
metrics (Union[List[str], NoneType]): the list of metrics you need to track during training.
The metrics should be one of the metrics implemented in PyTorch Lightning.
By default, it is Accuracy if classification and MeanSquaredLogError for regression
metrics_params (Union[List, NoneType]): The parameters to be passed to the Metrics initialized
target_range (Union[List, NoneType]): The range in which we should limit the output variable. Currently ignored for multi-target regression
Typically used for Regression problems. If left empty, will not apply any restrictions
n_d (int): Dimension of the prediction layer (usually between 4 and 64)
n_a (int): Dimension of the attention layer (usually between 4 and 64)
n_steps (int): Number of sucessive steps in the newtork (usually betwenn 3 and 10)
gamma (float): Float above 1, scaling factor for attention updates (usually betwenn 1.0 to 2.0)
embedding_dims (Union[List[int], NoneType]): The dimensions of the embedding for each categorical column as
a list of tuples (cardinality, embedding_dim). If left empty, will infer using the cardinality of the categorical column
using the rule min(50, (x + 1) // 2)
n_independent (int): Number of independent GLU layer in each GLU block (default 2)
n_shared (int): Number of independent GLU layer in each GLU block (default 2)
virtual_batch_size (int): Batch size for Ghost Batch Normalization
mask_type (str): Either 'sparsemax' or 'entmax' : this is the masking function to useChoices are: sparsemax entmax
Raises:
NotImplementedError: Raises an error if task is not in ['regression','classification']
"""
n_d: int = field(
default=8,
metadata={
"help": "Dimension of the prediction layer (usually between 4 and 64)"
},
)
n_a: int = field(
default=8,
metadata={
"help": "Dimension of the attention layer (usually between 4 and 64)"
},
)
n_steps: int = field(
default=3,
metadata={
"help": "Number of sucessive steps in the newtork (usually betwenn 3 and 10)"
},
)
gamma: float = field(
default=1.3,
metadata={
"help": "Float above 1, scaling factor for attention updates (usually betwenn 1.0 to 2.0)"
},
)
n_independent: int = field(
default=2,
metadata={
"help": "Number of independent GLU layer in each GLU block (default 2)"
},
)
n_shared: int = field(
default=2,
metadata={
"help": "Number of independent GLU layer in each GLU block (default 2)"
},
)
virtual_batch_size: int = field(
default=128,
metadata={"help": "Batch size for Ghost Batch Normalization"},
)
mask_type: str = field(
default="sparsemax",
metadata={
"help": "Either 'sparsemax' or 'entmax' : this is the masking function to use",
"choices": ["sparsemax", "entmax"],
},
)
_module_src: str = field(default="tabnet")
_model_name: str = field(default="TabNetModel")
_config_name: str = field(default="TabNetModelConfig")
pytorch_tabular.config.config.TrainerConfig
dataclass
Trainer configuration
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
batch_size |
int |
Number of samples in each batch of training |
64 |
fast_dev_run |
bool |
Quick Debug Run of Val |
False |
max_epochs |
int |
Maximum number of epochs to be run |
10 |
min_epochs |
int |
Force training for at least these many epochs. 1 by default |
1 |
max_time |
Optional[int] |
Stop training after this amount of time has passed. Disabled by default (None) |
None |
gpus |
int |
Number of gpus to train on (int) or which GPUs to train on (list or str). -1 uses all available GPUs. By default uses CPU (None) |
None |
accumulate_grad_batches |
int |
Accumulates grads every k batches or as set up in the dict. Trainer also calls optimizer.step() for the last indivisible step number. |
1 |
auto_lr_find |
bool |
Runs a learning rate finder algorithm (see this paper) when calling trainer.tune(), to find optimal initial learning rate. |
False |
auto_select_gpus |
bool |
If enabled and |
True |
check_val_every_n_epoch |
int |
Check val every n train epochs. |
1 |
deterministic |
bool |
If true enables cudnn.deterministic. Might make your system slower, but ensures reproducibility. |
False |
gradient_clip_val |
float |
Gradient clipping value |
0.0 |
overfit_batches |
float |
Uses this much data of the training set. If nonzero, will use the same training set for validation and testing. If the training dataloaders have shuffle=True, Lightning will automatically disable it. Useful for quickly debugging or trying to overfit on purpose. |
0.0 |
profiler |
Optional[str, NoneType] |
To profile individual steps during training and assist in identifying bottlenecks. Choices are: 'None' 'simple' 'advanced' |
None |
early_stopping |
str |
The loss/metric that needed to be monitored for early stopping. If None, there will be no early stopping |
'valid_loss' |
early_stopping_min_delta |
float |
The minimum delta in the loss/metric which qualifies as an improvement in early stopping |
0.001 |
early_stopping_mode |
str |
The direction in which the loss/metric should be optimized. Choices are |
'min' |
early_stopping_patience |
int |
The number of epochs to wait until there is no further improvements in loss/metric |
3 |
checkpoints |
str |
The loss/metric that needed to be monitored for checkpoints. If None, there will be no checkpoints |
'valid_loss' |
checkpoints_path |
str |
The path where the saved models will be |
'saved_models' |
checkpoints_name(Optional[str]) |
The name under which the models will be saved.
If left blank, first it will look for |
required | |
checkpoints_mode |
str |
The direction in which the loss/metric should be optimized |
'min' |
checkpoints_save_top_k |
int |
The number of best models to save |
1 |
load_best |
bool |
Flag to load the best model saved during training |
True |
track_grad_norm |
int |
Track and Log Gradient Norms in the logger. -1 by default means no tracking. 1 for the L1 norm, 2 for L2 norm, etc. |
-1 |
progress_bar |
str |
Progress bar type. Can be one of: |
'rich' |
precision |
int |
Lightning supports either double precision (64), full precision (32), or half precision (16) training. Half precision, or mixed precision, is the combined use of 32 and 16 bit floating points to reduce memory footprint during model training. This can result in improved performance, achieving +3X speedups on modern GPUs. |
32 |
trainer_kwargs |
dict[str, Any] |
Additional kwargs to be passed to PyTorch Lightning Trainer. See https://pytorch-lightning.readthedocs.io/en/latest/api/pytorch_lightning.trainer.html#pytorch_lightning.trainer.Trainer |
<factory> |
Source code in pytorch_tabular/config/config.py
class TrainerConfig:
"""Trainer configuration
Args:
batch_size (int): Number of samples in each batch of training
fast_dev_run (bool): Quick Debug Run of Val
max_epochs (int): Maximum number of epochs to be run
min_epochs (int): Force training for at least these many epochs. 1 by default
max_time (Optional[int]): Stop training after this amount of time has passed. Disabled by default (None)
gpus (int): Number of gpus to train on (int) or which GPUs to train on (list or str). -1 uses all available GPUs. By default uses CPU (None)
accumulate_grad_batches (int): Accumulates grads every k batches or as set up in the dict.
Trainer also calls optimizer.step() for the last indivisible step number.
auto_lr_find (bool): Runs a learning rate finder algorithm (see this paper) when calling trainer.tune(),
to find optimal initial learning rate.
auto_select_gpus (bool): If enabled and `gpus` is an integer, pick available gpus automatically.
This is especially useful when GPUs are configured to be in 'exclusive mode', such that only one
process at a time can access them.
check_val_every_n_epoch (int): Check val every n train epochs.
deterministic (bool): If true enables cudnn.deterministic. Might make your system slower, but ensures reproducibility.
gradient_clip_val (float): Gradient clipping value
overfit_batches (float): Uses this much data of the training set. If nonzero, will use the same training set
for validation and testing. If the training dataloaders have shuffle=True, Lightning will automatically disable it.
Useful for quickly debugging or trying to overfit on purpose.
profiler (Optional[str, NoneType]): To profile individual steps during training and assist in identifying bottlenecks.
Choices are: 'None' 'simple' 'advanced'
early_stopping (str): The loss/metric that needed to be monitored for early stopping. If None, there will be no early stopping
early_stopping_min_delta (float): The minimum delta in the loss/metric which qualifies as an improvement in early stopping
early_stopping_mode (str): The direction in which the loss/metric should be optimized. Choices are `max` and `min`
early_stopping_patience (int): The number of epochs to wait until there is no further improvements in loss/metric
checkpoints (str): The loss/metric that needed to be monitored for checkpoints. If None, there will be no checkpoints
checkpoints_path (str): The path where the saved models will be
checkpoints_name(Optional[str]): The name under which the models will be saved.
If left blank, first it will look for `run_name` in experiment_config and if that is also None
then it will use a generic name like task_version.
checkpoints_mode (str): The direction in which the loss/metric should be optimized
checkpoints_save_top_k (int): The number of best models to save
load_best (bool): Flag to load the best model saved during training
track_grad_norm (int): Track and Log Gradient Norms in the logger.
-1 by default means no tracking. 1 for the L1 norm, 2 for L2 norm, etc.
progress_bar (str): Progress bar type. Can be one of: `none`, `simple`,
`rich`. Defaults to `rich`.
precision (int): Lightning supports either double precision (64), full
precision (32), or half precision (16) training. Half precision, or
mixed precision, is the combined use of 32 and 16 bit floating points
to reduce memory footprint during model training. This can result in
improved performance, achieving +3X speedups on modern GPUs.
trainer_kwargs (dict[str, Any]): Additional kwargs to be passed to PyTorch Lightning Trainer.
See https://pytorch-lightning.readthedocs.io/en/latest/api/pytorch_lightning.trainer.html#pytorch_lightning.trainer.Trainer
"""
batch_size: int = field(
default=64, metadata={"help": "Number of samples in each batch of training"}
)
fast_dev_run: bool = field(
default=False,
metadata={
"help": "runs n if set to ``n`` (int) else 1 if set to ``True`` batch(es) of train, val and test to find any bugs (ie: a sort of unit test)."
},
)
max_epochs: int = field(
default=10, metadata={"help": "Maximum number of epochs to be run"}
)
min_epochs: Optional[int] = field(
default=1,
metadata={
"help": "Force training for at least these many epochs. 1 by default"
},
)
max_time: Optional[int] = field(
default=None,
metadata={
"help": "Stop training after this amount of time has passed. Disabled by default (None)"
},
)
gpus: Optional[int] = field(
default=None,
metadata={
"help": "Number of gpus to train on (int). -1 uses all available GPUs. By default uses CPU (None)"
},
)
accumulate_grad_batches: int = field(
default=1,
metadata={
"help": "Accumulates grads every k batches or as set up in the dict. Trainer also calls optimizer.step() for the last indivisible step number."
},
)
auto_lr_find: bool = field(
default=False,
metadata={
"help": "Runs a learning rate finder algorithm (see this paper) when calling trainer.tune(), to find optimal initial learning rate."
},
)
auto_select_gpus: bool = field(
default=True,
metadata={
"help": "If enabled and `gpus` is an integer, pick available gpus automatically. This is especially useful when GPUs are configured to be in 'exclusive mode', such that only one process at a time can access them."
},
)
check_val_every_n_epoch: int = field(
default=1, metadata={"help": "Check val every n train epochs."}
)
gradient_clip_val: float = field(
default=0.0, metadata={"help": "Gradient clipping value"}
)
overfit_batches: float = field(
default=0.0,
metadata={
"help": "Uses this much data of the training set. If nonzero, will use the same training set for validation and testing. If the training dataloaders have shuffle=True, Lightning will automatically disable it. Useful for quickly debugging or trying to overfit on purpose."
},
)
deterministic: bool = field(
default=False,
metadata={
"help": "If true enables cudnn.deterministic. Might make your system slower, but ensures reproducibility."
},
)
profiler: Optional[str] = field(
default=None,
metadata={
"help": "To profile individual steps during training and assist in identifying bottlenecks. None, simple or advanced",
"choices": [None, "simple", "advanced"],
},
)
early_stopping: Optional[str] = field(
default="valid_loss",
metadata={
"help": "The loss/metric that needed to be monitored for early stopping. If None, there will be no early stopping"
},
)
early_stopping_min_delta: float = field(
default=0.001,
metadata={
"help": "The minimum delta in the loss/metric which qualifies as an improvement in early stopping"
},
)
early_stopping_mode: str = field(
default="min",
metadata={
"help": "The direction in which the loss/metric should be optimized",
"choices": ["max", "min"],
},
)
early_stopping_patience: int = field(
default=3,
metadata={
"help": "The number of epochs to wait until there is no further improvements in loss/metric"
},
)
checkpoints: Optional[str] = field(
default="valid_loss",
metadata={
"help": "The loss/metric that needed to be monitored for checkpoints. If None, there will be no checkpoints"
},
)
checkpoints_path: str = field(
default="saved_models",
metadata={"help": "The path where the saved models will be"},
)
checkpoints_name: Optional[str] = field(
default=None,
metadata={
"help": "The name under which the models will be saved. If left blank, first it will look for `run_name` in experiment_config and if that is also None then it will use a generic name like task_version."
},
)
checkpoints_mode: str = field(
default="min",
metadata={"help": "The direction in which the loss/metric should be optimized"},
)
checkpoints_save_top_k: int = field(
default=1,
metadata={"help": "The number of best models to save"},
)
load_best: bool = field(
default=True,
metadata={"help": "Flag to load the best model saved during training"},
)
track_grad_norm: int = field(
default=-1,
metadata={
"help": "Track and Log Gradient Norms in the logger. -1 by default means no tracking. 1 for the L1 norm, 2 for L2 norm, etc."
},
)
progress_bar: str = field(
default="rich",
metadata={
"help": "Progress bar type. Can be one of: `none`, `simple`, `rich`. Defaults to `rich`."
},
)
precision: int = field(
default=32,
metadata={
"help": "Precision of the model. Can be one of: `32`, `16`, `64`. Defaults to `32`.",
"choices": [32, 16, 64],
},
)
seed: int = field(
default=42,
metadata={"help": "Seed for random number generators. Defaults to 42"},
)
trainer_kwargs: Dict[str, Any] = field(
default_factory=dict,
metadata={
"help": "Additional kwargs to be passed to PyTorch Lightning Trainer. See https://pytorch-lightning.readthedocs.io/en/latest/api/pytorch_lightning.trainer.html#pytorch_lightning.trainer.Trainer"
},
)
def __post_init__(self):
_validate_choices(self)
pytorch_tabular.config.config.DataConfig
dataclass
Data configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
target |
List[str] |
A list of strings with the names of the target column(s) |
required |
continuous_cols |
List[str] |
Column names of the numeric fields. Defaults to [] |
<factory> |
categorical_cols |
List |
Column names of the categorical fields to treat differently. Defaults to [] |
<factory> |
date_columns |
List |
(Column names, Freq) tuples of the date fields. For eg. a field named introduction_date and with a monthly frequency should have an entry ('intro_date','M'} |
<factory> |
encode_date_columns |
bool |
Whether or not to encode the derived variables from date |
True |
validation_split |
Optional[float, NoneType] |
Percentage of Training rows to keep aside as validation. Used only if Validation Data is not given separately |
0.2 |
continuous_feature_transform |
Optional[str, NoneType] |
Whether or not to transform the features before modelling. By default it is turned off.Choices are: None "yeo-johnson" "box-cox" "quantile_normal" "quantile_uniform" |
None |
normalize_continuous_features |
bool |
Flag to normalize the input features(continuous) |
True |
quantile_noise |
int |
NOT IMPLEMENTED. If specified fits QuantileTransformer on data with added gaussian noise with std = :quantile_noise: * data.std ; this will cause discrete values to be more separable. Please note that this transformation does NOT apply gaussian noise to the resulting data, the noise is only applied for QuantileTransformer |
0 |
num_workers |
Optional[int, NoneType] |
The number of workers used for data loading. For Windows always set to 0 |
0 |
pin_memory |
Optional[bool, NoneType] |
Whether or not to use pin_memory for data loading. |
False |
Source code in pytorch_tabular/config/config.py
class DataConfig:
"""Data configuration.
Args:
target (List[str]): A list of strings with the names of the target column(s)
continuous_cols (List[str]): Column names of the numeric fields. Defaults to []
categorical_cols (List): Column names of the categorical fields to treat differently. Defaults to []
date_columns (List): (Column names, Freq) tuples of the date fields. For eg. a field named introduction_date
and with a monthly frequency should have an entry ('intro_date','M'}
encode_date_columns (bool): Whether or not to encode the derived variables from date
validation_split (Optional[float, NoneType]): Percentage of Training rows to keep aside as validation.
Used only if Validation Data is not given separately
continuous_feature_transform (Optional[str, NoneType]): Whether or not to transform the features before modelling.
By default it is turned off.Choices are: None "yeo-johnson" "box-cox" "quantile_normal" "quantile_uniform"
normalize_continuous_features (bool): Flag to normalize the input features(continuous)
quantile_noise (int): NOT IMPLEMENTED. If specified fits QuantileTransformer on data with added
gaussian noise with std = :quantile_noise: * data.std ; this will cause discrete values to be more separable.
Please note that this transformation does NOT apply gaussian noise to the resulting data,
the noise is only applied for QuantileTransformer
num_workers (Optional[int, NoneType]): The number of workers used for data loading. For Windows always set to 0
pin_memory (Optional[bool, NoneType]): Whether or not to use pin_memory for data loading.
"""
target: List[str] = field(
default=MISSING,
metadata={"help": "A list of strings with the names of the target column(s)"},
)
continuous_cols: List = field(
default_factory=list,
metadata={"help": "Column names of the numeric fields. Defaults to []"},
)
categorical_cols: List = field(
default_factory=list,
metadata={
"help": "Column names of the categorical fields to treat differently. Defaults to []"
},
)
date_columns: List = field(
default_factory=list,
metadata={
"help": "(Column names, Freq) tuples of the date fields. For eg. a field named introduction_date and with a monthly frequency should have an entry ('intro_date','M'}"
},
)
encode_date_columns: bool = field(
default=True,
metadata={"help": "Whether or not to encode the derived variables from date"},
)
validation_split: Optional[float] = field(
default=0.2,
metadata={
"help": "Percentage of Training rows to keep aside as validation. Used only if Validation Data is not given separately"
},
)
continuous_feature_transform: Optional[str] = field(
default=None,
metadata={
"help": "Whether or not to transform the features before modelling. By default it is turned off.",
"choices": [
None,
"yeo-johnson",
"box-cox",
"quantile_normal",
"quantile_uniform",
],
},
)
normalize_continuous_features: bool = field(
default=True,
metadata={"help": "Flag to normalize the input features(continuous)"},
)
quantile_noise: int = field(
default=0,
metadata={
"help": "NOT IMPLEMENTED. If specified fits QuantileTransformer on data with added gaussian noise with std = :quantile_noise: * data.std ; this will cause discrete values to be more separable. Please not that this transformation does NOT apply gaussian noise to the resulting data, the noise is only applied for QuantileTransformer"
},
)
num_workers: Optional[int] = field(
default=0,
metadata={
"help": "The number of workers used for data loading. For windows always set to 0"
},
)
pin_memory: bool = field(
default=False,
metadata={"help": "Whether or not to pin memory for data loading."},
)
categorical_dim: int = field(init=False)
continuous_dim: int = field(init=False)
# output_dim: int = field(init=False)
def __post_init__(self):
assert (
len(self.categorical_cols)
+ len(self.continuous_cols)
+ len(self.date_columns)
> 0
), "There should be at-least one feature defined in categorical, continuous, or date columns"
self.categorical_dim = (
len(self.categorical_cols) if self.categorical_cols is not None else 0
)
self.continuous_dim = (
len(self.continuous_cols) if self.continuous_cols is not None else 0
)
_validate_choices(self)
if os.name == "nt" and self.num_workers != 0:
print("Windows does not support num_workers > 0. Setting num_workers to 0")
self.num_workers = 0
pytorch_tabular.config.config.ModelConfig
dataclass
Base Model configuration
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
task |
str |
Specify whether the problem is regression, classification, or self supervised learning. Choices are: regression classification ssl |
required |
ssl_task |
str |
Specify the kind of self supervised algorithm to use. Choices are: Denoising, Contrastive, None |
None |
aug_task |
str |
Specify the kind of augmentations algorithm to use for ssl. Choices are: cutmix, mixup, None |
None |
embedding_dims |
Optional[List[int], NoneType] |
The dimensions of the embedding for each categorical column as a list of tuples (cardinality, embedding_dim). If left empty, will infer using the cardinality of the categorical column using the rule min(50, (x + 1) // 2). Will only be used if the model uses categorical embedding. |
None |
learning_rate |
float |
The learning rate of the model |
0.001 |
loss |
Optional[str, NoneType] |
The loss function to be applied. By Default it is MSELoss for regression and CrossEntropyLoss for classification. Unless you are sure what you are doing, leave it at MSELoss or L1Loss for regression and CrossEntropyLoss for classification |
None |
metrics |
Optional[List[str], NoneType] |
the list of metrics you need to track during training. The metrics should be one of the metrics implemented in PyTorch Lightning. By default, it is accuracy if classification and mean_squared_error for regression |
None |
metrics_params |
Optional[List, NoneType] |
The parameters to be passed to the metrics function |
None |
target_range |
Optional[List, NoneType] |
The range in which we should limit the output variable. Currently ignored for multi-target regression Typically used for Regression problems. If left empty, will not apply any restrictions |
None |
Exceptions:
| Type | Description |
|---|---|
NotImplementedError |
Raises an error if task is not regression or classification |
Source code in pytorch_tabular/config/config.py
class ModelConfig:
"""Base Model configuration
Args:
task (str): Specify whether the problem is regression, classification, or self supervised learning. Choices are: regression classification ssl
ssl_task (str): Specify the kind of self supervised algorithm to use. Choices are: Denoising, Contrastive, None
aug_task (str): Specify the kind of augmentations algorithm to use for ssl. Choices are: cutmix, mixup, None
embedding_dims (Optional[List[int], NoneType]): The dimensions of the embedding for each categorical column
as a list of tuples (cardinality, embedding_dim). If left empty, will infer using the cardinality of the categorical column
using the rule min(50, (x + 1) // 2). Will only be used if the model uses categorical embedding.
learning_rate (float): The learning rate of the model
loss (Optional[str, NoneType]): The loss function to be applied.
By Default it is MSELoss for regression and CrossEntropyLoss for classification.
Unless you are sure what you are doing, leave it at MSELoss or L1Loss for regression and CrossEntropyLoss for classification
metrics (Optional[List[str], NoneType]): the list of metrics you need to track during training.
The metrics should be one of the metrics implemented in PyTorch Lightning.
By default, it is accuracy if classification and mean_squared_error for regression
metrics_params (Optional[List, NoneType]): The parameters to be passed to the metrics function
target_range (Optional[List, NoneType]): The range in which we should limit the output variable. Currently ignored for multi-target regression
Typically used for Regression problems. If left empty, will not apply any restrictions
Raises:
NotImplementedError: Raises an error if task is not regression or classification
"""
task: str = field(
# default="regression",
metadata={
"help": "Specify whether the problem is regression of classification.",
"choices": ["regression", "classification", "ssl"],
}
)
ssl_task: Optional[str] = field(
default=None,
metadata={
"help": "Specify the kind of self supervised algorithm to use.",
"choices": ["Denoising", "Contrastive", None],
},
)
aug_task: Optional[str] = field(
default=None,
metadata={
"help": "Specify the kind of augmentations algorithm to use for ssl.",
"choices": ["cutmix", "mixup", None],
},
)
embedding_dims: Optional[List[int]] = field(
default=None,
metadata={
"help": "The dimensions of the embedding for each categorical column as a list of tuples "
"(cardinality, embedding_dim). If left empty, will infer using the cardinality of the "
"categorical column using the rule min(50, (x + 1) // 2)"
},
)
learning_rate: float = field(
default=1e-3, metadata={"help": "The learning rate of the model"}
)
loss: Optional[str] = field(
default=None,
metadata={
"help": "The loss function to be applied. By Default it is MSELoss for regression "
"and CrossEntropyLoss for classification. Unless you are sure what you are doing, "
"leave it at MSELoss or L1Loss for regression and CrossEntropyLoss for classification"
},
)
metrics: Optional[List[str]] = field(
default=None,
metadata={
"help": "the list of metrics you need to track during training. The metrics should be one "
"of the functional metrics implemented in ``torchmetrics``. By default, "
"it is accuracy if classification and mean_squared_error for regression"
},
)
metrics_params: Optional[List] = field(
default=None,
metadata={"help": "The parameters to be passed to the metrics function"},
)
target_range: Optional[List] = field(
default=None,
metadata={
"help": "The range in which we should limit the output variable. "
"Currently ignored for multi-target regression. Typically used for Regression problems. "
"If left empty, will not apply any restrictions"
},
)
seed: int = field(
default=42,
metadata={"help": "The seed for reproducibility. Defaults to 42"},
)
def __post_init__(self):
if self.task == "regression":
self.loss = "MSELoss" if self.loss is None else self.loss
self.metrics = (
["mean_squared_error"] if self.metrics is None else self.metrics
)
self.metrics_params = (
[{} for _ in self.metrics]
if self.metrics_params is None
else self.metrics_params
)
elif self.task == "classification":
self.loss = "CrossEntropyLoss" if self.loss is None else self.loss
self.metrics = ["accuracy"] if self.metrics is None else self.metrics
self.metrics_params = (
[{} for _ in self.metrics]
if self.metrics_params is None
else self.metrics_params
)
elif self.task == "ssl":
assert self.ssl_task, "if task is ssl, ssl_task cannot be None"
assert self.aug_task, "if task is ssl, aug_task cannot be None"
if self.ssl_task == "Contrastive":
if self.loss:
logger.warning(
"In case of Contrastive the loss cannot be specified and will be ignored"
)
self.loss = "ContrastiveLoss" if self.loss is None else self.loss
else:
self.loss = "MSELoss" if self.loss is None else self.loss
self.metrics = (
["mean_squared_error"] if self.metrics is None else self.metrics
)
self.metrics_params = [{}]
else:
raise NotImplementedError(
f"{self.task} is not a valid task. Should be one of "
f"{self.__dataclass_fields__['task'].metadata['choices']}"
)
assert len(self.metrics) == len(
self.metrics_params
), "metrics and metric_params should have same length"
_validate_choices(self)
pytorch_tabular.config.config.ExperimentConfig
dataclass
Experiment configuration. Experiment Tracking with WandB and Tensorboard
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
project_name |
str |
The name of the project under which all runs will be logged. For Tensorboard this defines the folder under which the logs will be saved and for W&B it defines the project name. |
required |
run_name |
Optional[str, NoneType] |
The name of the run; a specific identifier to recognize the run. If left blank, will be assigned a auto-generated name |
None |
exp_watch |
Optional[str, NoneType] |
The level of logging required.
Can be |
None |
log_target |
str |
Determines where logging happens - Tensorboard or W&BChoices are: wandb tensorboard |
'tensorboard' |
log_logits |
bool |
Turn this on to log the logits as a histogram in W&B |
False |
exp_log_freq |
int |
step count between logging of gradients and parameters. |
100 |
_exp_version_manager |
str |
The location of the yaml file which manages versions of experiments |
required |
Source code in pytorch_tabular/config/config.py
class ExperimentConfig:
"""Experiment configuration. Experiment Tracking with WandB and Tensorboard
Args:
project_name (str): The name of the project under which all runs will be logged.
For Tensorboard this defines the folder under which the logs will be saved and
for W&B it defines the project name.
run_name (Optional[str, NoneType]): The name of the run; a specific identifier to
recognize the run. If left blank, will be assigned a auto-generated name
exp_watch (Optional[str, NoneType]): The level of logging required.
Can be `gradients`, `parameters`, `all` or `None`. Defaults to None
log_target (str): Determines where logging happens - Tensorboard or W&BChoices are: wandb tensorboard
log_logits (bool): Turn this on to log the logits as a histogram in W&B
exp_log_freq (int): step count between logging of gradients and parameters.
_exp_version_manager (str): The location of the yaml file which manages versions of experiments
"""
project_name: str = field(
default=MISSING,
metadata={
"help": "The name of the project under which all runs will be logged. For Tensorboard this defines the folder under which the logs will be saved and for W&B it defines the project name"
},
)
run_name: Optional[str] = field(
default=None,
metadata={
"help": "The name of the run; a specific identifier to recognize the run. If left blank, will be assigned a auto-generated name"
},
)
exp_watch: Optional[str] = field(
default=None,
metadata={
"help": "The level of logging required. Can be `gradients`, `parameters`, `all` or `None`. Defaults to None",
"choices": ["gradients", "parameters", "all", None],
},
)
log_target: str = field(
default="tensorboard",
metadata={
"help": "Determines where logging happens - Tensorboard or W&B",
"choices": ["wandb", "tensorboard"],
},
)
log_logits: bool = field(
default=False,
metadata={"help": "Turn this on to log the logits as a histogram in W&B"},
)
exp_log_freq: int = field(
default=100,
metadata={"help": "step count between logging of gradients and parameters."},
)
def __post_init__(self):
_validate_choices(self)
if self.log_target == "wandb":
try:
import wandb # noqa: F401
except ImportError:
raise ImportError(
"No W&B installation detected. `pip install wandb` to install W&B if you set log_target as `wandb`"
)
pytorch_tabular.config.config.OptimizerConfig
dataclass
Optimizer and Learning Rate Scheduler configuration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
optimizer |
str |
Any of the standard optimizers from
torch.optim. Defaults to |
'Adam' |
optimizer_params |
dict |
The parameters for the optimizer. If left blank, will use default parameters. |
<factory> |
lr_scheduler |
Optional[str, NoneType] |
The name of the LearningRateScheduler to use, if any, from torch.optim.lr_scheduler.
If None, will not use any scheduler. Defaults to |
None |
lr_scheduler_params |
Optional[dict, NoneType] |
The parameters for the LearningRateScheduler. If left blank, will use default parameters. |
<factory> |
lr_scheduler_monitor_metric |
Optional[str, NoneType] |
Used with |
'valid_loss' |
Source code in pytorch_tabular/config/config.py
class OptimizerConfig:
"""Optimizer and Learning Rate Scheduler configuration.
Args:
optimizer (str): Any of the standard optimizers from
[torch.optim](https://pytorch.org/docs/stable/optim.html#algorithms). Defaults to `Adam`"
optimizer_params (dict): The parameters for the optimizer. If left blank, will use default parameters.
lr_scheduler (Optional[str, NoneType]): The name of the LearningRateScheduler to use, if any, from [torch.optim.lr_scheduler](https://pytorch.org/docs/stable/optim.html#how-to-adjust-learning-rate).
If None, will not use any scheduler. Defaults to `None`
lr_scheduler_params (Optional[dict, NoneType]): The parameters for the LearningRateScheduler. If left blank, will use default parameters.
lr_scheduler_monitor_metric (Optional[str, NoneType]): Used with `ReduceLROnPlateau`, where the plateau is decided based on this metric
"""
optimizer: str = field(
default="Adam",
metadata={
"help": "Any of the standard optimizers from [torch.optim](https://pytorch.org/docs/stable/optim.html#algorithms)."
},
)
optimizer_params: dict = field(
default_factory=lambda: {"weight_decay": 0, "amsgrad": False},
metadata={
"help": "The parameters for the optimizer. If left blank, will use default parameters."
},
)
lr_scheduler: Optional[str] = field(
default=None,
metadata={
"help": "The name of the LearningRateScheduler to use, if any, from [torch.optim.lr_scheduler](https://pytorch.org/docs/stable/optim.html#how-to-adjust-learning-rate). If None, will not use any scheduler. Defaults to `None`",
},
)
lr_scheduler_params: Optional[dict] = field(
default_factory=lambda: {},
metadata={
"help": "The parameters for the LearningRateScheduler. If left blank, will use default parameters."
},
)
lr_scheduler_monitor_metric: Optional[str] = field(
default="valid_loss",
metadata={
"help": "Used with ReduceLROnPlateau, where the plateau is decided based on this metric"
},
)
@staticmethod
def read_from_yaml(filename: str = "config/optimizer_config.yml"):
config = _read_yaml(filename)
if config["lr_scheduler_params"] is None:
config["lr_scheduler_params"] = {}
return OptimizerConfig(**config)
pytorch_tabular.config.config.ExperimentRunManager
Source code in pytorch_tabular/config/config.py
class ExperimentRunManager:
def __init__(
self,
exp_version_manager: str = ".pt_tmp/exp_version_manager.yml",
) -> None:
"""The manages the versions of the experiments based on the name. It is a simple dictionary(yaml) based lookup.
Primary purpose is to avoid overwriting of saved models while runing the training without changing the experiment name.
Args:
exp_version_manager (str, optional): The path of the yml file which acts as version control.
Defaults to ".pt_tmp/exp_version_manager.yml".
"""
super().__init__()
self._exp_version_manager = exp_version_manager
if os.path.exists(exp_version_manager):
self.exp_version_manager = OmegaConf.load(exp_version_manager)
else:
self.exp_version_manager = OmegaConf.create({})
os.makedirs(os.path.split(exp_version_manager)[0], exist_ok=True)
with open(self._exp_version_manager, "w") as file:
OmegaConf.save(config=self.exp_version_manager, f=file)
def update_versions(self, name):
if name in self.exp_version_manager.keys():
uid = self.exp_version_manager[name] + 1
else:
uid = 1
self.exp_version_manager[name] = uid
with open(self._exp_version_manager, "w") as file:
OmegaConf.save(config=self.exp_version_manager, f=file)
return uid
__init__(self, exp_version_manager='.pt_tmp/exp_version_manager.yml')
special
The manages the versions of the experiments based on the name. It is a simple dictionary(yaml) based lookup. Primary purpose is to avoid overwriting of saved models while runing the training without changing the experiment name.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
exp_version_manager |
str |
The path of the yml file which acts as version control. |
'.pt_tmp/exp_version_manager.yml' |
Source code in pytorch_tabular/config/config.py
def __init__(
self,
exp_version_manager: str = ".pt_tmp/exp_version_manager.yml",
) -> None:
"""The manages the versions of the experiments based on the name. It is a simple dictionary(yaml) based lookup.
Primary purpose is to avoid overwriting of saved models while runing the training without changing the experiment name.
Args:
exp_version_manager (str, optional): The path of the yml file which acts as version control.
Defaults to ".pt_tmp/exp_version_manager.yml".
"""
super().__init__()
self._exp_version_manager = exp_version_manager
if os.path.exists(exp_version_manager):
self.exp_version_manager = OmegaConf.load(exp_version_manager)
else:
self.exp_version_manager = OmegaConf.create({})
os.makedirs(os.path.split(exp_version_manager)[0], exist_ok=True)
with open(self._exp_version_manager, "w") as file:
OmegaConf.save(config=self.exp_version_manager, f=file)