stream

def metric_results(
    self,
    level: MetricLevelEnum | Literal["macro", "micro", "window", "user"] = MetricLevelEnum.MACRO,
    only_current_timestamp: None | bool = False,
    filter_timestamp: None | int = None,
    filter_algo: None | str = None,
) -> pd.DataFrame:
    """Results of the metrics computed.

    Computes the metrics of all algorithms based on the level specified and
    return the results in a pandas DataFrame. The results can be filtered
    based on the algorithm name and the current timestamp.

    Specifics
    ---------
    - User level: User level metrics computed across all timestamps.
    - Window level: Window level metrics computed across all timestamps. This can
        be viewed as a macro level metric in the context of a single window, where
        the scores of each user is averaged within the window.
    - Macro level: Macro level metrics computed for entire timeline. This
        score is computed by averaging the scores of all windows, treating each
        window equally.
    - Micro level: Micro level metrics computed for entire timeline. This
        score is computed by averaging the scores of all users, treating each
        user and the timestamp the user is in as unique contribution to the
        overall score.

    Args:
        level: Level of the metric to compute, defaults to "macro".
        only_current_timestamp: Filter only the current timestamp, defaults to False.
        filter_timestamp: Timestamp value to filter on, defaults to None.
            If both `only_current_timestamp` and `filter_timestamp` are provided,
            `filter_timestamp` will be used.
        filter_algo: Algorithm name to filter on, defaults to None.

    Returns:
        Dataframe representation of the metric.
    """
    if isinstance(level, str) and not MetricLevelEnum.has_value(level):
        raise ValueError("Invalid level specified")
    level = MetricLevelEnum(level)

    if only_current_timestamp and filter_timestamp:
        raise ValueError("Cannot specify both only_current_timestamp and filter_timestamp.")

    timestamp = None
    if only_current_timestamp:
        timestamp = self._current_timestamp

    if filter_timestamp:
        timestamp = filter_timestamp

    return self._acc.df_metric(filter_algo=filter_algo, filter_timestamp=timestamp, level=level)

def plot_macro_level_metric(self) -> None:
    df = self.metric_results("macro")
    df = df.reset_index()
    ax = sns.barplot(
        data=df,
        x="metric",
        y="macro_score",
        hue="algorithm",
        edgecolor="black"
    )

    ax.set_xlabel("Metric")
    ax.set_ylabel("Macro score")
    ax.set_title("Macro-level scores by metric and algorithm")

    for container in ax.containers:
        ax.bar_label(container, fmt='%.4f', padding=3, fontsize=8)
    plt.legend(
        title="Algorithm",
        loc="upper center",
        bbox_to_anchor=(0.5, -0.1),
    )
    ax.grid(axis="y", alpha=0.3, linestyle="--")
    plt.show()

def plot_micro_level_metric(self) -> None:
    df = self.metric_results("micro")
    df = df.reset_index()
    ax = sns.barplot(
        data=df,
        x="metric",
        y="micro_score",
        hue="algorithm",
        edgecolor="black"
    )

    ax.set_xlabel("Metric")
    ax.set_ylabel("Micro score")
    ax.set_title("Micro-level scores by metric and algorithm")

    for container in ax.containers:
        ax.bar_label(container, fmt='%.4f', padding=3, fontsize=8)
    plt.legend(
        title="Algorithm",
        loc="upper center",
        bbox_to_anchor=(0.5, -0.1),
    )
    ax.grid(axis="y", alpha=0.3, linestyle="--")
    plt.show()

def plot_window_level_metric(self) -> None:
    df = self.metric_results("window")
    df = df.reset_index()
    metrics = df["metric"].unique()
    n_metrics = len(metrics)

    fig, axes = plt.subplots(n_metrics, 1, figsize=(10, 7), sharey=False)
    if n_metrics == 1:
        axes = [axes]

    fig.suptitle("Window-level scores over time", fontsize=14, fontweight="bold")

    for ax, metric in zip(axes, metrics):
        # Filter data for this metric
        metric_df = df[df["metric"] == metric]

        # Plot line for each algorithm
        sns.lineplot(
            data=metric_df,
            x="timestamp",
            y="window_score",
            hue="algorithm",
            marker="o",
            markersize=6,
            linewidth=2,
            ax=ax,
        )
        ax.set_xlabel("Timestamp (epoch)")
        ax.set_ylabel(f"{metric} score")
        ax.grid(axis="both", alpha=0.3, linestyle="--")

        # Remove individual legends
        if ax.get_legend() is not None:
            ax.get_legend().remove()

    # Create single shared legend at bottom
    handles, labels = axes[0].get_legend_handles_labels()

    fig.legend(
        handles,
        labels,
        title="Algorithm",
        loc="lower center",
        bbox_to_anchor=(0.5, -0.15),
        ncol=1,  # vertical stacking
        frameon=True,
        fontsize=9,
    )
    plt.show()

def restore(self) -> None:
    """Restore the generators before pickling.

    This method is used to restore the generators after loading the object
    from a pickle file.
    """
    self.setting.restore(self._run_step)
    logger.debug("Generators restored")

def current_step(self) -> int:
    """Return the current step of the evaluator.

    Returns:
        Current step of the evaluator.
    """
    return self._run_step

def start_stream(self) -> None:
    """Start the streaming process.

    Warning:
        Once `start_stream` is called, the evaluator cannot register any new algorithms.

    Raises:
        ValueError: If the stream has already started.
    """
    self._assert_state(expected=EvaluatorState.INITIALIZED, error_msg="Stream has already started")
    self.setting.restore()
    logger.debug("Preparing evaluator for streaming")
    self._acc = MetricAccumulator()
    self.load_next_window()
    logger.debug("Evaluator is ready for streaming")
    # TODO: allow programmer to register anytime
    self._transition_state(new_state=EvaluatorState.STARTED, allow_from=[EvaluatorState.INITIALIZED])

def register_model(
    self,
    algorithm: Algorithm,
    algorithm_name: None | str = None,
) -> UUID:
    """Register the algorithm with the evaluator.

    This method is called to register the algorithm with the evaluator.
    The method will assign a unique identifier to the algorithm and store
    the algorithm in the registry.

    Warning:
        Once `start_stream` is called, the evaluator cannot register any new algorithms.
    """
    self._assert_state(EvaluatorState.INITIALIZED, "Cannot register algorithms after stream started")
    algo_id = self._algo_state_mgr.register(name=algorithm_name, algorithm_ptr=algorithm)
    logger.debug(f"Algorithm {algo_id} registered")
    return algo_id

def get_algorithm_state(self, algorithm_id: UUID) -> AlgorithmStateEnum:
    """Get the state of the algorithm."""
    return self._algo_state_mgr[algorithm_id].state

def get_all_algorithm_status(self) -> dict[str, AlgorithmStateEnum]:
    """Get the status of all algorithms."""
    return self._algo_state_mgr.all_algo_states()

def load_next_window(self) -> None:
    self.user_item_base.reset_unknown_user_item_base()
    self._training_data_cache = self._get_training_data()
    self._unlabeled_data_cache, self._ground_truth_data_cache, self._current_timestamp = self._get_evaluation_data()
    self._algo_state_mgr.set_all_ready(data_segment=self._current_timestamp)

def get_training_data(self, algo_id: UUID) -> InteractionMatrix:
    """Get training data for the algorithm.

    Args:
        algo_id: Unique identifier of the algorithm.

    Raises:
        ValueError: If the stream has not started.

    Returns:
        The training data for the algorithm.
    """
    self._assert_state(expected=[EvaluatorState.STARTED, EvaluatorState.IN_PROGRESS], error_msg="Call start_stream() first")

    logger.debug(f"Getting data for algorithm {algo_id}")

    if self._strategy.should_advance_window(
        algo_state_mgr=self._algo_state_mgr,
        current_step=self._run_step,
        total_steps=self.setting.num_split,
    ):
        try:
            self.load_next_window()
        except EOWSettingError:
            self._transition_state(
                EvaluatorState.COMPLETED, allow_from=[EvaluatorState.STARTED, EvaluatorState.IN_PROGRESS]
            )
            raise RuntimeError("End of evaluation window reached")

    can_request, reason = self._algo_state_mgr.can_request_training_data(algo_id)
    if not can_request:
        raise PermissionError(f"Cannot request data: {reason}")
    # TODO handle case when algo is ready after submitting prediction, but current timestamp has not changed, meaning algo is requesting same data again
    self._algo_state_mgr.transition(
        algo_id=algo_id,
        new_state=AlgorithmStateEnum.RUNNING,
        data_segment=self._current_timestamp,
    )

    self._state = EvaluatorState.IN_PROGRESS
    # release data to the algorithm
    return self._training_data_cache

def get_unlabeled_data(self, algo_id: UUID) -> PredictionMatrix:
    """Get unlabeled data for the algorithm.

    This method is called to get the unlabeled data for the algorithm. The
    unlabeled data is the data that the algorithm will predict. It will
    contain `(user_id, -1)` pairs, where the value -1 indicates that the
    item is to be predicted.
    """
    logger.debug(f"Getting unlabeled data for algorithm {algo_id}")
    can_submit, reason = self._algo_state_mgr.can_request_unlabeled_data(algo_id)
    if not can_submit:
        raise PermissionError(f"Cannot get unlabeled data: {reason}")
    return self._unlabeled_data_cache

def submit_prediction(self, algo_id: UUID, X_pred: csr_matrix) -> None:
    """Submit the prediction of the algorithm.

    This method is called to submit the prediction of the algorithm.
    There are a few checks that are done before the prediction is
    evaluated by calling :meth:`_evaluate_algo_pred`.

    Once the prediction is evaluated, the method will update the state
    of the algorithm to PREDICTED.
    """
    logger.debug(f"Submitting prediction for algorithm {algo_id}")
    can_submit, reason = self._algo_state_mgr.can_submit_prediction(algo_id)
    if not can_submit:
        raise PermissionError(f"Cannot submit prediction: {reason}")

    self._evaluate_algo_pred(algorithm_id=algo_id, y_pred=X_pred)
    self._algo_state_mgr.transition(
        algo_id=algo_id,
        new_state=AlgorithmStateEnum.PREDICTED,
    )

@abstractmethod
def should_advance_window(self, algo_state_mgr: AlgorithmStateManager, current_step: int, total_steps: int) -> bool:
    """Determine if should advance to next window"""
    pass

def should_advance_window(self, algo_state_mgr: AlgorithmStateManager, current_step: int, total_steps: int) -> bool:
    """Advance only when all algorithms predicted"""
    return algo_state_mgr.is_all_predicted() and current_step < total_steps

def should_advance_window(self, algo_state_mgr: AlgorithmStateManager, current_step: int, total_steps: int) -> bool:
    """Advance only when all algorithms predicted"""
    return (
        algo_state_mgr.is_all_predicted()
        and algo_state_mgr.is_all_same_data_segment()
        and current_step < total_steps
    )