from __future__ import annotations from dataclasses import dataclass from typing import Any @dataclass(frozen=True) class RetrainDecision: baseline_precision: float current_precision: float | None deterioration: float | None threshold: float should_flag: bool reason: str def as_dict(self) -> dict[str, Any]: return self.__dict__ def calculate_deterioration(baseline_precision: float, current_precision: float) -> float: if baseline_precision <= 0: raise ValueError("baseline_precision must be > 0") return (baseline_precision - current_precision) / baseline_precision def evaluate_retrain_trigger( baseline_precision: float, current_precision: float | None, *, deterioration_threshold: float = 0.15, ) -> RetrainDecision: if not 0 < deterioration_threshold < 1: raise ValueError("deterioration_threshold must be between 0 and 1") if not 0 < baseline_precision <= 1: raise ValueError("baseline_precision must be in (0, 1]") if current_precision is None: return RetrainDecision( baseline_precision, None, None, deterioration_threshold, False, "No current 20-day precision available." ) if not 0 <= current_precision <= 1: raise ValueError("current_precision must be between 0 and 1") deterioration = calculate_deterioration(baseline_precision, current_precision) should_flag = deterioration >= deterioration_threshold - 1e-12 if should_flag: reason = ( f"20-day precision deteriorated by {deterioration:.2%}, " f"meeting/exceeding the {deterioration_threshold:.2%} threshold." ) elif deterioration > 0: reason = ( f"20-day precision deteriorated by {deterioration:.2%}, " f"below the {deterioration_threshold:.2%} threshold." ) else: reason = ( f"20-day precision is not below baseline " f"(change={-deterioration:.2%})." ) return RetrainDecision( baseline_precision, current_precision, deterioration, deterioration_threshold, should_flag, reason )