diff --git a/app/ml/evaluation.py b/app/ml/evaluation.py new file mode 100644 index 0000000..f0bcb2a --- /dev/null +++ b/app/ml/evaluation.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from typing import Sequence + +from app.ml.feature_store import FeatureVector +from app.ml.training import TrainingPipeline, TrainedArtifact + +logger = logging.getLogger(__name__) + + +@dataclass +class Metric: + name: str + value: float + threshold: float | None = None + + +@dataclass +class EvalReport: + artifact_id: str + sample_size: int + metrics: list[Metric] + + +class Evaluator: + def __init__(self, pipeline: TrainingPipeline) -> None: + self._pipeline = pipeline + + def evaluate(self, artifact_id: str, predictions: Sequence[str]) -> EvalReport: + artifacts = list(self._pipeline._artifacts) + if not artifacts: + raise ValueError("Kein trainiertes Modell für Evaluation vorhanden.") + + supported_sensors = self._pipeline.export(artifact_id).supported_sensors + unknown_hits = sum(1 for prediction in predictions if ":" not in prediction) + supported_references = sum(1 for sensor in supported_sensors for prediction in predictions if sensor in prediction) + sample_size = len(predictions) + coverage = supported_references / sample_size if sample_size else 0.0 + unknown_rate = unknown_hits / sample_size if sample_size else 0.0 + + coverage_metric = Metric(name="coverage", value=coverage, threshold=0.8) + unknown_metric = Metric(name="unknown_rate", value=unknown_rate, threshold=0.1) + + report = EvalReport( + artifact_id=artifact_id, + sample_size=sample_size, + metrics=[coverage_metric, unknown_metric], + ) + logger.info( + "Evaluation %s -> coverage=%.2f, unknown_rate=%.2f", + artifact_id, + coverage, + unknown_rate, + ) + return report \ No newline at end of file diff --git a/tests/ml/test_evaluation.py b/tests/ml/test_evaluation.py new file mode 100644 index 0000000..fac9894 --- /dev/null +++ b/tests/ml/test_evaluation.py @@ -0,0 +1,33 @@ +from __future__ import annotations + +import pytest + +from app.ml.evaluation import Evaluator +from app.ml.feature_store import FeatureStore, FeatureVector +from app.ml.training import TrainingPipeline + + +def _vector(sensor_id: str, temperature: float, label: str | None = None) -> FeatureVector: + return FeatureVector(sensor_id=sensor_id, values={"temperature": temperature}, label=label) + + +def evaluator_factory() -> Evaluator: + store = FeatureStore() + store.add_batch([_vector("sensor.kitchen", 19.0), _vector("sensor.bedroom", 18.5)]) + pipeline = TrainingPipeline(store) + pipeline.run("artifact_v1") + return Evaluator(pipeline) + + +def test_evaluate_returns_report_with_metrics() -> None: + evaluator = evaluator_factory() + report = evaluator.evaluate("artifact_v1", ["artifact_v1:sensor.kitchen:{'temperature': 21.0}", "artifact_v1:sensor.bedroom:{'temperature': 18.5}"]) + assert report.artifact_id == "artifact_v1" + assert report.sample_size == 2 + assert {metric.name for metric in report.metrics} == {"coverage", "unknown_rate"} + + +def test_evaluate_without_training_raises_value_error() -> None: + evaluator = Evaluator(TrainingPipeline(FeatureStore())) + with pytest.raises(ValueError): + evaluator.evaluate("artifact_v1", []) \ No newline at end of file