Core API Reference¶

@abstractmethod
def run(self, adapter: BaseAdapter, data: pd.DataFrame) -> "ValidationTestResult":
    """Run the validation test.

    Args:
        adapter: The adapter to validate
        data: Input data for validation

    Returns:
        TestResult object containing test results

    """
    pass

def run_with_error_handling(self, adapter: BaseAdapter, data: pd.DataFrame) -> "ValidationTestResult":
    """Run the validation test with error handling.

    Args:
        adapter: The adapter to validate
        data: Input data for validation

    Returns:
        TestResult object containing test results

    Raises:
        MetricCalculationError: If metric calculation fails
        TestExecutionError: If test execution fails

    """
    try:
        return self.run(adapter, data)
    except ZeroDivisionError as e:
        # This is clearly a mathematical calculation issue
        raise MetricCalculationError(f"Metric calculation error in {self.test_name} test: {str(e)}") from e
    except Exception as e:
        # All other errors - let individual tests handle specific categorization if needed
        raise TestExecutionError(f"Test execution error in {self.test_name} test: {str(e)}") from e

def evaluate_framework(self, framework: str, config: BaseConfig) -> ValidationResults:
    """Evaluate an MMM framework using the unified API.

    Args:
        framework: Name of the MMM framework to evaluate
        config: Framework-specific configuration

    Returns:
        ValidationResult object containing evaluation metrics and predictions

    Raises:
        ValueError: If any test name is invalid

    """
    # Initialize the adapter
    adapter = get_adapter(framework, config)

    # Run validation tests
    validation_results = self.validation_orchestrator.validate(
        adapter=adapter,
        data=self.data,
        test_names=self.test_names,
    )

    return validation_results

def validate(
    self,
    adapter: BaseAdapter,
    data: pd.DataFrame,
    test_names: list[ValidationTestNames],
) -> ValidationResults:
    """Run validation tests on the model.

    Args:
        model: Model to validate
        data: Input data for validation
        test_names: List of test names to run
        adapter: Adapter to use for the test

    Returns:
        ValidationResults containing all test results

    Raises:
        ValueError: If any requested test is not registered

    """
    # Run tests and collect results
    results: dict[ValidationTestNames, ValidationTestResult] = {}
    for test_name in test_names:
        logger.info(f"Running test: {test_name}")
        test_instance = self.tests[test_name](adapter.date_column)
        test_result = test_instance.run_with_error_handling(adapter, data)
        results[test_name] = test_result

    return ValidationResults(results)

def get_test_result(self, test_name: ValidationTestNames) -> ValidationTestResult:
    """Get results for a specific test."""
    return self.test_results[test_name]

def to_df(self) -> pd.DataFrame:
    """Convert validation results to a flat DataFrame format."""
    return pd.concat(
        [self.get_test_result(test_name).to_df() for test_name in self.test_results.keys()],
        ignore_index=True,
    )

def to_df(self) -> pd.DataFrame:
    """Convert test results to a flat DataFrame format."""
    test_scores_df = self.test_scores.to_df()
    test_scores_df[ValidationTestAttributeNames.TEST_NAME.value] = self.test_name.value
    test_scores_df[ValidationTestAttributeNames.TIMESTAMP.value] = self.timestamp.isoformat()
    return test_scores_df

def run(self, adapter: BaseAdapter, data: pd.DataFrame) -> ValidationTestResult:
    """Run the accuracy test."""
    # Split data into train/test sets
    train, test = self._split_data_holdout(data)
    predictions = adapter.fit_and_predict(train, test)
    actual = test.groupby(self.date_column)[InputDataframeConstants.RESPONSE_COL].sum()
    assert len(actual) == len(predictions), "Actual and predicted lengths must match"

    # Calculate metrics
    test_scores = AccuracyMetricResults.populate_object_with_metrics(
        actual=pd.Series(actual),  # Ensure it's a Series
        predicted=pd.Series(predictions, index=actual.index),
    )

    logger.info(f"Saving the test results for {self.test_name} test")

    return ValidationTestResult(
        test_name=ValidationTestNames.ACCURACY,
        metric_names=AccuracyMetricNames.to_list(),
        test_scores=test_scores,
    )

def run(self, adapter: BaseAdapter, data: pd.DataFrame) -> ValidationTestResult:
    """Run the cross-validation test using time-series splits.

    Args:
        model: Model to validate
        adapter: Adapter to use for the test
        data: Input data

    Returns:
        TestResult containing cross-validation metrics

    """
    # Initialize cross-validation splitter
    cv_splits = self._split_data_time_series_cv(data)

    # Store metrics for each fold
    fold_metrics = []

    # Run cross-validation
    for i, (train_idx, test_idx) in enumerate(cv_splits):

        logger.info(f"Running cross-validation fold {i+1} of {len(cv_splits)}")

        # Get train/test data
        train = data.loc[train_idx]
        test = data.loc[test_idx]

        # Get predictions
        predictions = adapter.fit_and_predict(train, test)
        actual = test.groupby(self.date_column)[InputDataframeConstants.RESPONSE_COL].sum()
        assert len(actual) == len(predictions), "Actual and predicted lengths must match"

        # Add in fold results
        fold_metrics.append(
            AccuracyMetricResults.populate_object_with_metrics(
                actual=pd.Series(actual),  # Ensure it's a Series
                predicted=pd.Series(predictions, index=actual.index),
            )
        )

    # Calculate mean and std of metrics across folds and create metric results
    test_scores = CrossValidationMetricResults(
        mean_mape=calculate_mean_for_singular_values_across_cross_validation_folds(
            fold_metrics, AccuracyMetricNames.MAPE
        ),
        std_mape=calculate_std_for_singular_values_across_cross_validation_folds(
            fold_metrics, AccuracyMetricNames.MAPE
        ),
        mean_r_squared=calculate_mean_for_singular_values_across_cross_validation_folds(
            fold_metrics, AccuracyMetricNames.R_SQUARED
        ),
    )

    logger.info(f"Saving the test results for {self.test_name} test")

    return ValidationTestResult(
        test_name=ValidationTestNames.CROSS_VALIDATION,
        metric_names=CrossValidationMetricNames.to_list(),
        test_scores=test_scores,
    )

def run(self, adapter: BaseAdapter, data: pd.DataFrame) -> ValidationTestResult:
    """Run the perturbation test."""
    # Train model on original data
    adapter.fit(data)
    original_rois = adapter.get_channel_roi()

    # TODO: support perturbation of reach and frequency regressors
    if adapter.primary_media_regressor_type == PrimaryMediaRegressor.REACH_AND_FREQUENCY:
        logger.warning(
            "Perturbation test skipped: Reach and frequency regressor type not supported for perturbation."
        )
        # Return NaN results for each channel indicating the test was not run
        channel_names = adapter.get_channel_names()
        test_scores = PerturbationMetricResults(
            percentage_change_for_each_channel=pd.Series(np.nan, index=channel_names),
        )
        return ValidationTestResult(
            test_name=ValidationTestNames.PERTURBATION,
            metric_names=PerturbationMetricNames.to_list(),
            test_scores=test_scores,
        )

    # Add noise to primary regressor data and retrain
    noisy_data = self._add_gaussian_noise_to_primary_regressors(
        df=data,
        regressor_cols=adapter.primary_media_regressor_columns,
    )
    adapter.fit(noisy_data)
    noise_rois = adapter.get_channel_roi()

    # calculate the pct change in roi
    percentage_change = calculate_absolute_percentage_change(
        baseline_series=original_rois,
        comparison_series=noise_rois,
    )

    # Create metric results - roi % change for each channel
    test_scores = PerturbationMetricResults(
        percentage_change_for_each_channel=percentage_change,
    )

    logger.info(f"Saving the test results for {self.test_name} test")

    return ValidationTestResult(
        test_name=ValidationTestNames.PERTURBATION,
        metric_names=PerturbationMetricNames.to_list(),
        test_scores=test_scores,
    )

def run(self, adapter: BaseAdapter, data: pd.DataFrame) -> ValidationTestResult:
    """Run the stability test."""
    # Initialize cross-validation splitter
    cv_splits = self._split_data_time_series_cv(data)

    # Store metrics for each fold
    fold_metrics = []

    # Run cross-validation
    for i, (train_idx, refresh_idx) in enumerate(cv_splits):

        logger.info(f"Running refresh stability test fold {i+1} of {len(cv_splits)}")

        # Get train/test data
        # todo(): Can we somehow store these training changes in the adapter for use in time series holdout test
        current_data = data.loc[train_idx]
        # Combine current data with refresh data for retraining
        refresh_data = pd.concat([current_data, data.loc[refresh_idx]], ignore_index=True)
        # Get common dates for roi stability comparison
        common_start_date, common_end_date = self._get_common_dates(
            baseline_data=current_data,
            comparison_data=refresh_data,
            date_column=adapter.date_column,
        )

        # Train model and get coefficients
        adapter.fit(current_data)
        current_model_rois = adapter.get_channel_roi(
            start_date=common_start_date,
            end_date=common_end_date,
        )
        adapter.fit(refresh_data)
        refreshed_model_rois = adapter.get_channel_roi(
            start_date=common_start_date,
            end_date=common_end_date,
        )

        # calculate the pct change in volume
        percentage_change = calculate_absolute_percentage_change(
            baseline_series=current_model_rois,
            comparison_series=refreshed_model_rois,
        )

        fold_metrics.append(percentage_change)

    # Calculate mean and std of percentage change for each channel across cross validation folds
    test_scores = RefreshStabilityMetricResults(
        mean_percentage_change_for_each_channel=calculate_means_for_series_across_cross_validation_folds(
            fold_metrics
        ),
        std_percentage_change_for_each_channel=calculate_stds_for_series_across_cross_validation_folds(
            fold_metrics
        ),
    )

    logger.info(f"Saving the test results for {self.test_name} test")

    return ValidationTestResult(
        test_name=ValidationTestNames.REFRESH_STABILITY,
        metric_names=RefreshStabilityMetricNames.to_list(),
        test_scores=test_scores,
    )

@classmethod
def all_tests(cls) -> list["ValidationTestNames"]:
    """Return all validation test names as a list."""
    return list(cls)

@classmethod
def all_tests_as_str(cls) -> list[str]:
    """Return all validation test names as a list of strings."""
    return [test.value for test in cls]