offline

learn_offline(environment, learner, inputs, outputs, *, input_transform=None, output_transform=None)

Learn from an offline environment.

Learn from an offline environment by learning from the input-output pairs.

Parameters:

Name	Type	Description	Default
environment	OfflineEnvironment	The offline environment.	required
learner	SupervisedLearner	The supervised learner.	required
inputs	list[str]	The input feature names.	required
outputs	list[str]	The output feature names.	required
input_transform	Transform | None	The transform to apply to the input features. Will be part of the final model.	None
output_transform	InvertibleTransform | None	The transform to apply to the output features. Its inverse will be part of the final model.	None

Returns:

Type	Description
Model	The model learned from the environment.

Source code in src/flowcean/core/strategies/offline.py

def learn_offline(
    environment: OfflineEnvironment,
    learner: SupervisedLearner,
    inputs: list[str],
    outputs: list[str],
    *,
    input_transform: Transform | None = None,
    output_transform: InvertibleTransform | None = None,
) -> Model:
    """Learn from an offline environment.

    Learn from an offline environment by learning from the input-output pairs.

    Args:
        environment: The offline environment.
        learner: The supervised learner.
        inputs: The input feature names.
        outputs: The output feature names.
        input_transform: The transform to apply to the input features.
            Will be part of the final model.
        output_transform: The transform to apply to the output features.
            Its inverse will be part of the final model.

    Returns:
        The model learned from the environment.
    """
    if input_transform is None:
        input_transform = Identity()
    if output_transform is None:
        output_transform = Identity()

    logger.info("Learning with offline strategy")
    data = environment.observe()

    logger.info("Selecting input and output features")
    input_features = data.select(inputs)
    output_features = data.select(outputs)

    logger.info("Fitting transforms and applying them to features")
    input_transform.fit(input_features)
    input_features = input_transform.apply(input_features)

    logger.info("Fitting output transform and applying it to output features")
    output_transform.fit(output_features)
    output_features = output_transform.apply(output_features)

    logger.info("Learning model")
    model = learner.learn(inputs=input_features, outputs=output_features)

    model.pre_transform |= input_transform
    model.post_transform |= output_transform.inverse()

    return model

evaluate_offline(models, environment, inputs, outputs, metrics)

Evaluate a model on an offline environment.

Evaluate a model on an offline environment by predicting the outputs from the inputs and comparing them to the true outputs.

Parameters:

Name	Type	Description	Default
models	Model | Iterable[Model]	The models to evaluate.	required
environment	OfflineEnvironment	The offline environment.	required
inputs	Sequence[str]	The input feature names.	required
outputs	Sequence[str]	The output feature names.	required
metrics	Sequence[Metric]	The metrics to evaluate the model with.	required

Returns:

Type	Description
Report	The evaluation report.

Source code in src/flowcean/core/strategies/offline.py

def evaluate_offline(
    models: Model | Iterable[Model],
    environment: OfflineEnvironment,
    inputs: Sequence[str],
    outputs: Sequence[str],
    metrics: Sequence[Metric],
) -> Report:
    """Evaluate a model on an offline environment.

    Evaluate a model on an offline environment by predicting the outputs from
    the inputs and comparing them to the true outputs.

    Args:
        models: The models to evaluate.
        environment: The offline environment.
        inputs: The input feature names.
        outputs: The output feature names.
        metrics: The metrics to evaluate the model with.

    Returns:
        The evaluation report.
    """
    if not isinstance(models, Iterable):
        models = [models]
    data = environment.observe()
    input_features = data.select(inputs)
    output_features = data.select(outputs)
    entries: dict[str, ReportEntry] = {}

    for model in models:
        predictions = model.predict(input_features)

        entries[model.name] = ReportEntry(
            {
                metric.name: metric(output_features, predictions.lazy())
                for metric in metrics
            },
        )
    return Report(entries)

tune_threshold(model, environment, inputs, outputs, metric, *, thresholds=None, num_thresholds=19)

Find optimal decision threshold for a classifier.

Evaluates the model at multiple threshold values and returns the threshold that maximizes the given metric.

Parameters:

Name	Type	Description	Default
model	ClassifierModel	The classifier model to tune.	required
environment	OfflineEnvironment	The offline environment with validation/test data.	required
inputs	Sequence[str]	The input feature names.	required
outputs	Sequence[str]	The output feature names.	required
metric	Metric	The metric to optimize (e.g., FBetaScore, Accuracy).	required
thresholds	Sequence[float] | None	Specific thresholds to evaluate. If None, generates num_thresholds evenly spaced values between 0.05 and 0.95.	None
num_thresholds	int	Number of thresholds to evaluate if thresholds is None (default: 19).	19

Returns:

Type	Description
float	Tuple of (best_threshold, results_dict) where results_dict maps
dict[float, float]	each threshold to its metric score.

Example

from flowcean.sklearn.metrics.classification import FBetaScore metric = FBetaScore(beta=1.0) best_threshold, results = tune_threshold( ... model, eval_env, inputs, outputs, metric ... ) print(f"Best threshold: {best_threshold:.3f}") model.threshold = best_threshold # Apply the best threshold

Source code in src/flowcean/core/strategies/offline.py

def tune_threshold(
    model: ClassifierModel,
    environment: OfflineEnvironment,
    inputs: Sequence[str],
    outputs: Sequence[str],
    metric: Metric,
    *,
    thresholds: Sequence[float] | None = None,
    num_thresholds: int = 19,
) -> tuple[float, dict[float, float]]:
    """Find optimal decision threshold for a classifier.

    Evaluates the model at multiple threshold values and returns the threshold
    that maximizes the given metric.

    Args:
        model: The classifier model to tune.
        environment: The offline environment with validation/test data.
        inputs: The input feature names.
        outputs: The output feature names.
        metric: The metric to optimize (e.g., FBetaScore, Accuracy).
        thresholds: Specific thresholds to evaluate. If None, generates
            num_thresholds evenly spaced values between 0.05 and 0.95.
        num_thresholds: Number of thresholds to evaluate if thresholds is
            None (default: 19).

    Returns:
        Tuple of (best_threshold, results_dict) where results_dict maps
        each threshold to its metric score.

    Example:
        >>> from flowcean.sklearn.metrics.classification import FBetaScore
        >>> metric = FBetaScore(beta=1.0)
        >>> best_threshold, results = tune_threshold(
        ...     model, eval_env, inputs, outputs, metric
        ... )
        >>> print(f"Best threshold: {best_threshold:.3f}")
        >>> model.threshold = best_threshold  # Apply the best threshold
    """
    import numpy as np

    # Generate thresholds if not provided
    thresholds_to_test: Sequence[float]
    if thresholds is None:
        thresholds_to_test = list(np.linspace(0.05, 0.95, num_thresholds))
    else:
        thresholds_to_test = thresholds

    # Store original threshold
    original_threshold = model.threshold

    # Evaluate at each threshold
    results: dict[float, float] = {}
    best_threshold = 0.5
    best_score = -float("inf")

    logger.info(
        "Tuning threshold for %s using %s",
        model.name,
        metric.name,
    )

    try:
        for threshold in thresholds_to_test:
            # Set threshold and evaluate
            model.threshold = float(threshold)
            report = evaluate_offline(
                model,
                environment,
                inputs,
                outputs,
                [metric],
            )

            # Extract metric score from report
            # Report is dict[str, ReportEntry]
            # ReportEntry is dict[str, Reportable]
            model_entry = report.get(model.name)
            if model_entry is None:
                continue

            score_value = model_entry.get(metric.name)
            if score_value is None:
                continue

            # Convert Reportable to float (it should be a numeric value)
            if isinstance(score_value, (int, float)):
                score = float(score_value)
            else:
                # Try to convert via string representation
                try:
                    score = float(str(score_value))
                except (ValueError, TypeError):
                    logger.warning(
                        "Could not convert metric value to float: %s",
                        score_value,
                    )
                    continue

            results[float(threshold)] = score

            # Track best threshold
            if score > best_score:
                best_score = score
                best_threshold = float(threshold)

            logger.debug(
                "Threshold %.3f: %s = %.4f",
                threshold,
                metric.name,
                score,
            )

    finally:
        # Restore original threshold
        model.threshold = original_threshold

    return best_threshold, results