_driver Module (Internal Workflow Definition)

This internal module defines an MMMBaseDriver class that outlines the sequential steps involved in a typical Marketing Mix Modelling workflow. It acts as a blueprint or orchestrator, calling functions from other Abacus modules to perform specific tasks.

MMMBaseDriver Class

class MMMBaseDriver:
    """Base driver class for orchestrating the MMM workflow."""

This class defines methods corresponding to the major stages of an MMM project.

Methods

load_config

    def load_config(self, config_filename: str) -> tuple[bytes, dict]:

Loads the YAML configuration file using abacus.prepro.config.load_config.

Parameters:

• config_filename (str): Path to the configuration file.

Returns:

• tuple[bytes, dict]: Raw file content and the parsed configuration dictionary.

init_output

    def init_output(self, data_dir: str = ".") -> str:

Initialises the output directory structure using os.makedirs.

Parameters:

• data_dir (str, optional): Base directory for results. Defaults to ".".

Returns:

• str: Absolute path to the created results directory.

ingest_data

    def ingest_data(self, input_path: str, config: dict, holidays_filename: str) -> InputData:

Ingests data from a source file (CSV/Excel) and transforms it into the InputData structure. Uses abacus.prepro.decode.parse_csv_generic and abacus.prepro.convert.transform_input_generic.

Parameters:

• input_path (str): Path to the raw input data file.

• config (dict): Parsed configuration dictionary.

• holidays_filename (str): Path to the holidays file.

Returns:

• InputData: An InputData object from abacus.prepro.input_data.

run_feature_engineering

    def run_feature_engineering(self, input_data: InputData, _config: dict) -> InputData:

A placeholder method for applying custom feature engineering steps. In this base implementation, it returns the input data unchanged.

Parameters:

• input_data (InputData): The input data object.

• _config (dict): The configuration dictionary (unused here).

Returns:

• InputData: The potentially modified input data object.

describe_data

    def describe_data(
        self,
        results_dir: str,
        input_data_raw: InputData,
        input_data_processed: InputData,
        raw_config: str,
        current_commit: str,
    ) -> None:

Generates descriptive summaries and plots for both raw and processed data, saving them to the results directory. Uses functions from abacus.sketch.depict (like describe_input_data, describe_config) and the dump method of InputData.

Parameters:

• results_dir (str): Directory to save outputs.

• input_data_raw (InputData): InputData before feature engineering.

• input_data_processed (InputData): InputData after feature engineering.

• raw_config (str): Raw string content of the configuration file.

• current_commit (str): Git commit hash or version identifier.

fit

    def fit(
        self,
        results_dir: str,
        input_data_processed: InputData,
        config: dict,
        model_filename: str = None,
    ) -> tuple[DataToFit, Any]: # Type hint for model depends on fit_abacus_mmm

Prepares the data for model fitting (creating DataToFit object) and then fits the MMM model using abacus.core.mmm.fit_abacus_mmm (or loads a saved model using abacus.core.mmm.load_model).

Parameters:

• results_dir (str): Directory containing results (passed to fit_abacus_mmm).

• input_data_processed (InputData): Processed input data.

• config (dict): Configuration dictionary.

• model_filename (str, optional): Path to load a pre-fitted model instead of fitting. Defaults to None.

Returns:

• tuple[DataToFit, Any]: A tuple containing the DataToFit object and the fitted model instance.

visualize

    def visualize(
        self,
        results_dir: str,
        model: Any, # Type hint depends on fit_abacus_mmm
        input_data_processed: InputData,
        data_to_fit: DataToFit,
        config: dict,
    ) -> None:

Generates standard visualisations for the fitted model using functions from abacus.sketch.depict (like describe_mmm_training, describe_mmm_prediction).

Parameters:

• results_dir (str): Directory to save plots.

• model (Any): The fitted model instance.

• input_data_processed (InputData): Processed input data.

• data_to_fit (DataToFit): Data prepared for fitting.

• config (dict): Configuration dictionary.

save_model

    def save_model(self, results_dir: str, model: Any, data_to_fit: DataToFit) -> None:

Saves the fitted model state and the DataToFit object used. Uses abacus.core.mmm.save_model and the dump method of DataToFit.

Parameters:

• results_dir (str): Directory to save outputs.

• model (Any): The fitted model instance.

• data_to_fit (DataToFit): The data prepared for fitting.

main

    def main(self, config_filename, input_filename, holidays_filename, output_folder):

Orchestrates the entire workflow by calling the other methods in sequence: load config, init output, ingest data, feature engineering, describe data, fit model, visualize, save model, and write a final output summary file.

Parameters:

• config_filename (str): Path to the configuration file.

• input_filename (str): Path to the raw input data file.

• holidays_filename (str): Path to the holidays file.

• output_folder (str): Path to the base directory for saving results.