Explainer¶

def explain(self, x0: np.ndarray) -> ExplanationResult:  # noqa: PLR0915
    """Generate a directional counterfactual explanation for a single point.

    Args:
        x0 (np.ndarray): The baseline point to explain, as a 1D array.

    Returns:
        ExplanationResult: The optimal direction and metrics.

    Raises:
        NotFittedError: If the explainer requires fitting but `fit` has not been called.
        InvalidParameterError: If `x0` is not a 1D array.
    """
    if not self._is_fitted:
        msg = "Dicex.fit must be called before explain when scaler='auto'."
        raise NotFittedError(msg)

    if x0.ndim != VECTOR_NDIM:
        msg = f"x0 must be a 1D array. Got shape {x0.shape}"
        raise InvalidParameterError(msg)

    if self._d is None:
        self._d = len(x0)

    _logger.info(
        "Explaining point: d=%d, alpha=%.3f, task=%s",
        len(x0),
        self.alpha,
        self.task,
    )

    model_for_optimizer = self.model
    x0_optimizer = x0.astype(float, copy=False)
    metadata: dict[str, Any] = {}
    perturbation_for_optimizer = self.perturbation

    if self.scaler is not None:
        x0_optimizer = self.scaler.transform(x0.reshape(1, -1)).reshape(-1)
        model_for_optimizer = self.scaler.wrap_model(self.model)

    optimizer_params = self._optimizer_params()

    optimizer = VmfVns(
        model=cast("Any", model_for_optimizer),
        x0=x0_optimizer,
        perturbation=cast("BasePerturbation", perturbation_for_optimizer),
        alpha=self.alpha,
        target_class=self.target_class,
        **cast("Any", optimizer_params),
    )
    result_scaled = optimizer.run()

    direction = result_scaled.direction
    if self.scaler is not None:
        direction = self.scaler.transform_direction(result_scaled.direction)
        metadata["direction_scaled"] = result_scaled.direction

    metadata.update(result_scaled.metadata)

    _logger.info(
        "Explanation complete: CVaR=%.6f, %d model evals",
        result_scaled.robust_value,
        result_scaled.metadata.get("n_model_evals", 0),
    )

    # Compute baseline CVaR (c=0) with the same reevaluation budget.
    n_baseline = int(optimizer_params["n_evaluation"])
    baseline_t0 = time.perf_counter()
    env_samples = cast("BasePerturbation", perturbation_for_optimizer).sample_env(n_baseline, len(x0_optimizer))
    pts_baseline = x0_optimizer + env_samples

    if self.target_class is not None:
        cls_model = cast("Any", model_for_optimizer)
        y_baseline = cls_model.predict_proba(pts_baseline)[:, self.target_class]
        y0 = cls_model.predict_proba(x0_optimizer.reshape(1, -1))[0, self.target_class]
    else:
        reg_model = cast("Any", model_for_optimizer)
        y_baseline = reg_model.predict(pts_baseline).flatten()
        y0 = reg_model.predict(x0_optimizer.reshape(1, -1)).flatten()[0]

    z_baseline = y_baseline - float(y0)
    baseline_cvar = float(lower_tail_cvar_empirical(z_baseline, self.alpha))
    metadata["baseline_cvar"] = baseline_cvar
    phase_eval_counts = dict(metadata.get("phase_eval_counts", {}))
    phase_wall_times = dict(metadata.get("phase_wall_times", {}))
    phase_eval_counts["baseline_evaluation"] = phase_eval_counts.get("baseline_evaluation", 0) + n_baseline + 1
    phase_wall_times["baseline_evaluation"] = round(
        phase_wall_times.get("baseline_evaluation", 0.0) + (time.perf_counter() - baseline_t0),
        6,
    )
    metadata["phase_eval_counts"] = phase_eval_counts
    metadata["phase_wall_times"] = phase_wall_times
    metadata.setdefault("n_test", optimizer_params["n_test"])
    metadata.setdefault("n_evaluation", n_baseline)
    metadata.setdefault("n_bootstrap", optimizer_params.get("n_bootstrap", 1000))
    metadata.setdefault("trace_level", optimizer_params.get("trace_level", "none"))

    if result_scaled.robust_value > baseline_cvar:
        _logger.info(
            "Action recommended: Found direction strictly better than inaction (%.4f > %.4f)",
            result_scaled.robust_value,
            baseline_cvar,
        )
    else:
        _logger.info(
            "Inaction recommended: Best direction is not better than absorbing ambient noise (%.4f <= %.4f)",
            result_scaled.robust_value,
            baseline_cvar,
        )

    return ExplanationResult(
        direction=direction,
        robust_value=result_scaled.robust_value,
        alpha=self.alpha,
        metadata=metadata,
    )

def explain_batch(self, x: np.ndarray) -> list[ExplanationResult]:
    """Explain each row of a 2D input array independently.

    Args:
        x (np.ndarray): Array of points to explain, of shape (n_samples, n_features).

    Returns:
        list[ExplanationResult]: A list of explanation results.

    Raises:
        InvalidParameterError: If `x` is not a 2D array.
    """
    if x.ndim != MATRIX_NDIM:
        msg = f"x must be a 2D array. Got shape {x.shape}"
        raise InvalidParameterError(msg)

    _logger.info("Batch explanation: %d points", len(x))
    return [self.explain(row) for row in x]

def fit(self, x_train: np.ndarray) -> Self:
    """Fit the internal scaler when scaling is enabled.

    Args:
        x_train (np.ndarray): Training data array of shape (n_samples, n_features).

    Returns:
        Self: The fitted explainer instance.

    Raises:
        InvalidParameterError: If `x_train` is not a 2D array.
    """
    if x_train.ndim != MATRIX_NDIM:
        msg = f"x_train must be a 2D array. Got shape {x_train.shape}"
        raise InvalidParameterError(msg)

    if self.scaler is not None:
        self.scaler.fit(x_train)
        _logger.info("Scaler fitted on %d samples, d=%d", len(x_train), x_train.shape[1])

    self._d = x_train.shape[1]
    self._is_fitted = True
    return self