diff --git a/apps/backend/alembic/versions/d3f1a2b7c904_rapports_de_derive.py b/apps/backend/alembic/versions/d3f1a2b7c904_rapports_de_derive.py new file mode 100644 index 0000000..b64c6ba --- /dev/null +++ b/apps/backend/alembic/versions/d3f1a2b7c904_rapports_de_derive.py @@ -0,0 +1,81 @@ +"""rapports de derive du modele de prevision + +Revision ID: d3f1a2b7c904 +Revises: c0adab96238c +Create Date: 2026-09-22 14:40:00.000000 + +`site_id` est nullable, et c'est le coeur du schema : une ligne par site, plus une ligne +globale tous sites confondus, que `NULL` designe. Un seul site qui derive est invisible dans +une moyenne d'ensemble, et une derive d'ensemble sans rupture par site signale un changement +de modele ou de saison, pas une panne. + +L'unicite passe par un index a `coalesce` et non par une `UniqueConstraint` : deux lignes +globales successives ont toutes deux `site_id` a NULL, et NULL n'est egal a aucune valeur, pas +meme a lui-meme. Meme forme que `uq_reading_source`. + +Les trois `CHECK` sont portees par la base, comme `ck_prediction_status` : un verdict sans +motif, ou un statut inconnu, ne doit pas dependre de la vigilance de l'appelant. +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "d3f1a2b7c904" +down_revision: str | Sequence[str] | None = "c0adab96238c" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.create_table( + "drift_report", + sa.Column("drift_report_id", sa.BigInteger(), autoincrement=True, nullable=False), + sa.Column( + "computed_at", + sa.DateTime(timezone=True), + server_default=sa.text("now()"), + nullable=False, + ), + sa.Column("site_id", sa.Text(), nullable=True), + sa.Column("window_start", sa.DateTime(timezone=True), nullable=False), + sa.Column("window_end", sa.DateTime(timezone=True), nullable=False), + sa.Column("reference_start", sa.DateTime(timezone=True), nullable=True), + sa.Column("reference_end", sa.DateTime(timezone=True), nullable=True), + sa.Column("n_observations", sa.Integer(), nullable=False), + sa.Column("mae", sa.Double(), nullable=True), + sa.Column("mape", sa.Double(), nullable=True), + sa.Column("bias", sa.Double(), nullable=True), + sa.Column("reference_mae", sa.Double(), nullable=True), + sa.Column("coverage_ratio", sa.Double(), nullable=True), + sa.Column("insufficient_data_ratio", sa.Double(), nullable=True), + sa.Column("model_references", postgresql.ARRAY(sa.Text()), nullable=False), + sa.Column("status", sa.Text(), nullable=False), + sa.Column("reason", sa.Text(), nullable=True), + sa.CheckConstraint( + "status IN ('stable', 'derive', 'indetermine')", name="ck_drift_report_status" + ), + sa.CheckConstraint( + "status = 'stable' OR reason IS NOT NULL", name="ck_drift_report_reason" + ), + sa.CheckConstraint("n_observations >= 0", name="ck_drift_report_observations"), + sa.ForeignKeyConstraint( + ["site_id"], ["site.site_id"], name="fk_drift_report_site", ondelete="RESTRICT" + ), + sa.PrimaryKeyConstraint("drift_report_id"), + ) + op.create_index( + "ix_drift_report_site_computed", "drift_report", ["site_id", "computed_at"], unique=False + ) + op.create_index( + "uq_drift_report_window", + "drift_report", + ["window_end", sa.literal_column("coalesce(site_id, '')")], + unique=True, + ) + + +def downgrade() -> None: + op.drop_table("drift_report") diff --git a/apps/backend/app/api/deps.py b/apps/backend/app/api/deps.py index 6662f81..dd7fe7e 100644 --- a/apps/backend/app/api/deps.py +++ b/apps/backend/app/api/deps.py @@ -24,6 +24,7 @@ from app.core.security import decode_access_token as decode_token from app.db.session import get_session from app.repositories.alert import AlertRepository from app.repositories.audit_log import AuditLogRepository +from app.repositories.drift import DriftRepository from app.repositories.login_attempt import LoginAttemptRepository from app.repositories.password_reset_attempt import PasswordResetAttemptRepository from app.repositories.password_reset_token import PasswordResetTokenRepository @@ -35,6 +36,7 @@ from app.repositories.site import SiteRepository from app.repositories.user import UserRepository from app.services.alert import AlertService from app.services.auth import AuthService, LoginPolicy, PasswordResetPolicy +from app.services.drift import DriftService from app.services.prediction import PredictionService from app.services.reading import ReadingService from app.services.recommendation import RecommendationService @@ -232,6 +234,13 @@ def get_prediction_service(session: SessionDep) -> PredictionService: PredictionServiceDep = Annotated[PredictionService, Depends(get_prediction_service)] +def get_drift_service(session: SessionDep) -> DriftService: + return DriftService(DriftRepository(session)) + + +DriftServiceDep = Annotated[DriftService, Depends(get_drift_service)] + + async def get_current_principal( credentials: CredentialsDep, session: SessionDep, diff --git a/apps/backend/app/api/openapi.py b/apps/backend/app/api/openapi.py index 8ca8c08..6860d4a 100644 --- a/apps/backend/app/api/openapi.py +++ b/apps/backend/app/api/openapi.py @@ -90,6 +90,14 @@ TAGS: Final[list[dict[str, Any]]] = [ "de scoring (`ml/`) et simplement lue ici. Accessible à partir du rôle `lecteur`." ), }, + { + "name": "monitoring", + "description": ( + "Surveillance de la dérive du modèle : écart entre les prévisions déjà écrites et " + "les lectures réellement arrivées, par site et tous sites confondus. Réservé à " + "partir du rôle `operateur`, qui agit sur un pipeline dégradé." + ), + }, ] cookie_de_rafraichissement = APIKeyCookie( @@ -153,6 +161,17 @@ REPONSES_ADMIN: Final[Reponses] = { }, } +REPONSES_OPERATEUR: Final[Reponses] = { + **REPONSES_AUTHENTIFIEES, + 403: { + "model": ErrorResponse, + "description": ( + "Droits insuffisants, ou mot de passe provisoire à changer quand `detail` vaut " + "`password_change_required`." + ), + }, +} + # `lecteur` est le rôle minimum : `require_role` n'y refuse jamais un 403 pour droits # insuffisants, seulement pour le mot de passe provisoire. REPONSES_LECTEUR: Final[Reponses] = { diff --git a/apps/backend/app/api/v1/endpoints/monitoring.py b/apps/backend/app/api/v1/endpoints/monitoring.py new file mode 100644 index 0000000..031b36a --- /dev/null +++ b/apps/backend/app/api/v1/endpoints/monitoring.py @@ -0,0 +1,20 @@ +from fastapi import APIRouter + +from app.api.deps import DriftServiceDep, OperateurDep +from app.api.openapi import REPONSE_VALIDATION +from app.schemas.drift import DriftReportResponse + +router = APIRouter() + + +@router.get( + "/drift", + response_model=list[DriftReportResponse], + summary="Dernier rapport de dérive par site, plus la ligne globale", + responses=REPONSE_VALIDATION, +) +async def get_drift( + _: OperateurDep, service: DriftServiceDep, site_id: str | None = None +) -> list[DriftReportResponse]: + rapports = await service.derniers(site_id=site_id) + return [DriftReportResponse.model_validate(rapport) for rapport in rapports] diff --git a/apps/backend/app/api/v1/router.py b/apps/backend/app/api/v1/router.py index 6079acf..c3c39d4 100644 --- a/apps/backend/app/api/v1/router.py +++ b/apps/backend/app/api/v1/router.py @@ -1,10 +1,16 @@ from fastapi import APIRouter -from app.api.openapi import REPONSE_SERVEUR, REPONSES_ADMIN, REPONSES_LECTEUR +from app.api.openapi import ( + REPONSE_SERVEUR, + REPONSES_ADMIN, + REPONSES_LECTEUR, + REPONSES_OPERATEUR, +) from app.api.v1.endpoints import ( alerts, auth, health, + monitoring, predictions, readings, recommendations, @@ -38,3 +44,6 @@ api_router.include_router( api_router.include_router( predictions.router, prefix="/predictions", tags=["predictions"], responses=REPONSES_LECTEUR ) +api_router.include_router( + monitoring.router, prefix="/monitoring", tags=["monitoring"], responses=REPONSES_OPERATEUR +) diff --git a/apps/backend/app/models/__init__.py b/apps/backend/app/models/__init__.py index 167d7ce..8032c17 100644 --- a/apps/backend/app/models/__init__.py +++ b/apps/backend/app/models/__init__.py @@ -2,7 +2,15 @@ # --autogenerate`, qui générerait alors un drop de sa table. from app.models.audit_log import AuditLog -from app.models.energy import Alert, Dataset, Prediction, Reading, Recommendation, Site +from app.models.energy import ( + Alert, + Dataset, + DriftReport, + Prediction, + Reading, + Recommendation, + Site, +) from app.models.login_attempt import LoginAttempt from app.models.password_reset_attempt import PasswordResetAttempt from app.models.password_reset_token import PasswordResetToken @@ -14,6 +22,7 @@ __all__ = [ "AppUser", "AuditLog", "Dataset", + "DriftReport", "LoginAttempt", "PasswordResetAttempt", "PasswordResetToken", diff --git a/apps/backend/app/models/energy.py b/apps/backend/app/models/energy.py index 285ad26..ce4fcac 100644 --- a/apps/backend/app/models/energy.py +++ b/apps/backend/app/models/energy.py @@ -208,3 +208,49 @@ class Recommendation(Base): explanation: Mapped[str] = mapped_column(Text) rule_reference: Mapped[str] = mapped_column(Text) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + + +class DriftReport(Base): + __tablename__ = "drift_report" + __table_args__ = ( + CheckConstraint( + "status IN ('stable', 'derive', 'indetermine')", name="ck_drift_report_status" + ), + CheckConstraint("status = 'stable' OR reason IS NOT NULL", name="ck_drift_report_reason"), + CheckConstraint("n_observations >= 0", name="ck_drift_report_observations"), + Index("ix_drift_report_site_computed", "site_id", "computed_at"), + ) + + drift_report_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + computed_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + # `NULL` porte la ligne globale, tous sites confondus : une derive d'ensemble et la derive + # d'un seul site ne se lisent pas dans le meme chiffre. + site_id: Mapped[str | None] = mapped_column( + Text, ForeignKey("site.site_id", name="fk_drift_report_site", ondelete="RESTRICT") + ) + window_start: Mapped[datetime] = mapped_column(DateTime(timezone=True)) + window_end: Mapped[datetime] = mapped_column(DateTime(timezone=True)) + reference_start: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + reference_end: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + n_observations: Mapped[int] = mapped_column(Integer) + mae: Mapped[float | None] = mapped_column(Double) + mape: Mapped[float | None] = mapped_column(Double) + bias: Mapped[float | None] = mapped_column(Double) + reference_mae: Mapped[float | None] = mapped_column(Double) + coverage_ratio: Mapped[float | None] = mapped_column(Double) + insufficient_data_ratio: Mapped[float | None] = mapped_column(Double) + model_references: Mapped[list[str]] = mapped_column(ARRAY(Text)) + status: Mapped[str] = mapped_column(Text) + reason: Mapped[str | None] = mapped_column(Text) + + +# Piège : une `UniqueConstraint` ne dédoublonnerait pas les lignes globales, dont `site_id` est +# NULL et qu'aucune n'est égale à une autre. Même forme que `uq_reading_source`. +Index( + "uq_drift_report_window", + DriftReport.window_end, + func.coalesce(DriftReport.site_id, text("''")), + unique=True, +) diff --git a/apps/backend/app/monitoring/__init__.py b/apps/backend/app/monitoring/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/apps/backend/app/monitoring/drift.py b/apps/backend/app/monitoring/drift.py new file mode 100644 index 0000000..70ee970 --- /dev/null +++ b/apps/backend/app/monitoring/drift.py @@ -0,0 +1,105 @@ +# Surveillance de dérive du modèle de prévision (EC06, issue #45) : même gabarit que +# `app.detection.internal_alerts`, ordonnancé par le DAG `derive`. + +from __future__ import annotations + +import argparse +import asyncio +import sys +from datetime import UTC, datetime, timedelta + +from app.core.config import get_settings +from app.db.session import get_session_factory +from app.repositories.drift import DriftRepository, NouveauRapportDerive +from app.services.drift import STATUT_DERIVE, DriftService, Seuils + + +async def run_drift( + *, now: datetime | None = None, site_id: str | None = None, seuils: Seuils | None = None +) -> list[NouveauRapportDerive]: + """Calcule les rapports de la fenêtre et les enregistre. Rend ce qui a été calculé, que la + ligne ait été écrite ou ignorée par l'index d'idempotence.""" + async with get_session_factory()() as session: + depot = DriftRepository(session) + rapports = await DriftService(depot, seuils=seuils).evaluate(now=now, site_id=site_id) + await depot.enregistre(rapports) + await session.commit() + return rapports + + +def _parse_instant(valeur: str) -> datetime: + instant = datetime.fromisoformat(valeur) + return instant if instant.tzinfo is not None else instant.replace(tzinfo=UTC) + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + defauts = Seuils() + parser = argparse.ArgumentParser( + prog="python -m app.monitoring.drift", + description="Surveillance de dérive du modèle de prévision EnerVision", + ) + parser.add_argument("--site-id", default=None, help="Limite le calcul à un seul site.") + parser.add_argument( + "--now", + type=_parse_instant, + default=None, + help=( + "Instant de référence (ISO 8601, UTC si le fuseau est omis). Défaut : l'heure courante." + ), + ) + parser.add_argument( + "--window-hours", + type=int, + default=int(defauts.fenetre.total_seconds() // 3600), + help="Durée de la fenêtre récente, et de la fenêtre de référence qui la précède.", + ) + parser.add_argument( + "--grace-hours", + type=int, + default=int(defauts.grace.total_seconds() // 3600), + help="Délai laissé à l'ingestion avant qu'une prévision soit jugée vérifiable.", + ) + parser.add_argument( + "--min-observations", + type=int, + default=defauts.min_observations, + help="En deçà, le verdict est `indetermine` plutôt qu'un chiffre trompeur.", + ) + parser.add_argument( + "--fail-on-drift", + action="store_true", + help="Sort en code non nul si une dérive est constatée, pour que la tâche rougisse.", + ) + return parser.parse_args(argv) + + +def seuils_depuis(args: argparse.Namespace) -> Seuils: + return Seuils( + fenetre=timedelta(hours=args.window_hours), + grace=timedelta(hours=args.grace_hours), + min_observations=args.min_observations, + ) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + # Échoue tôt si `APP_SECRET_KEY`/`DATABASE_URL` manquent, avant toute requête à la base. + get_settings() + rapports = asyncio.run( + run_drift(now=args.now, site_id=args.site_id, seuils=seuils_depuis(args)) + ) + + for rapport in rapports: + cible = rapport.site_id or "TOUS SITES" + mae = f"{rapport.mae:.2f}" if rapport.mae is not None else "-" + print( + f"{cible} : {rapport.status}, MAE {mae} kWh sur {rapport.n_observations} prévision(s)" + f"{' : ' + rapport.reason if rapport.reason else ''}" + ) + + derive = any(rapport.status == STATUT_DERIVE for rapport in rapports) + return 1 if derive and args.fail_on_drift else 0 + + +if __name__ == "__main__": # pragma: no cover + sys.exit(main()) diff --git a/apps/backend/app/repositories/drift.py b/apps/backend/app/repositories/drift.py new file mode 100644 index 0000000..712228c --- /dev/null +++ b/apps/backend/app/repositories/drift.py @@ -0,0 +1,184 @@ +"""Piège : deux dédoublonnages, pas un - DriftRepository.paires() + +`prediction` n'a pas d'unicité sur `(site_id, target_at)` : chaque run de scoring empile une +ligne de plus. `uq_reading_source` autorise de son côté deux lectures au même instant quand la +`source` diffère. Joindre les deux tables sans `DISTINCT ON` des deux côtés compterait donc la +même heure plusieurs fois, et la moyenne d'erreur pèserait ces sites en double. + +On retient la prédiction du run le plus récent, celle que sert `GET /api/v1/predictions`, avec +`prediction_id` en départage : `created_at` vaut l'heure de début de transaction et ne +distingue pas deux lignes du même run. +""" + +from collections.abc import Sequence +from dataclasses import asdict, dataclass +from datetime import datetime + +from sqlalchemy import Subquery, func, select +from sqlalchemy.dialects.postgresql import insert +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.energy import DriftReport, Prediction, Reading + +TARGET_METRIC = "consumption_kwh" +STATUT_DISPONIBLE = "available" + + +@dataclass(frozen=True, slots=True) +class PaireDerive: + site_id: str + target_at: datetime + predicted_value: float + actual_value: float + model_reference: str + + +@dataclass(frozen=True, slots=True) +class NouveauRapportDerive: + site_id: str | None + window_start: datetime + window_end: datetime + reference_start: datetime | None + reference_end: datetime | None + n_observations: int + mae: float | None + mape: float | None + bias: float | None + reference_mae: float | None + coverage_ratio: float | None + insufficient_data_ratio: float | None + model_references: list[str] + status: str + reason: str | None + + +@dataclass(frozen=True, slots=True) +class ComptageStatut: + site_id: str + status: str + nombre: int + + +def _predictions_retenues(*, debut: datetime, fin: datetime, site_id: str | None) -> Subquery: + requete = ( + select( + Prediction.site_id, + Prediction.target_at, + Prediction.predicted_value, + Prediction.model_reference, + Prediction.status, + ) + .distinct(Prediction.site_id, Prediction.target_at) + .where( + Prediction.target_metric == TARGET_METRIC, + Prediction.target_at >= debut, + Prediction.target_at < fin, + ) + .order_by(Prediction.site_id, Prediction.target_at, Prediction.prediction_id.desc()) + ) + if site_id is not None: + requete = requete.where(Prediction.site_id == site_id) + return requete.subquery() + + +def _lectures_retenues(*, debut: datetime, fin: datetime, site_id: str | None) -> Subquery: + requete = ( + select(Reading.site_id, Reading.timestamp, Reading.consumption_kwh) + .distinct(Reading.site_id, Reading.timestamp) + .where( + Reading.timestamp >= debut, + Reading.timestamp < fin, + Reading.consumption_kwh.is_not(None), + ) + .order_by(Reading.site_id, Reading.timestamp, Reading.reading_id.desc()) + ) + if site_id is not None: + requete = requete.where(Reading.site_id == site_id) + return requete.subquery() + + +class DriftRepository: + def __init__(self, session: AsyncSession) -> None: + self._session = session + + async def paires( + self, *, debut: datetime, fin: datetime, site_id: str | None = None + ) -> Sequence[PaireDerive]: + predictions = _predictions_retenues(debut=debut, fin=fin, site_id=site_id) + lectures = _lectures_retenues(debut=debut, fin=fin, site_id=site_id) + requete = ( + select( + predictions.c.site_id, + predictions.c.target_at, + predictions.c.predicted_value, + lectures.c.consumption_kwh, + predictions.c.model_reference, + ) + .select_from(predictions) + .join( + lectures, + (lectures.c.site_id == predictions.c.site_id) + & (lectures.c.timestamp == predictions.c.target_at), + ) + .where(predictions.c.status == STATUT_DISPONIBLE) + .order_by(predictions.c.site_id, predictions.c.target_at) + ) + + lignes = await self._session.execute(requete) + return [ + PaireDerive( + site_id=ligne[0], + target_at=ligne[1], + predicted_value=ligne[2], + actual_value=ligne[3], + model_reference=ligne[4], + ) + for ligne in lignes + ] + + async def comptages( + self, *, debut: datetime, fin: datetime, site_id: str | None = None + ) -> Sequence[ComptageStatut]: + predictions = _predictions_retenues(debut=debut, fin=fin, site_id=site_id) + requete = ( + select(predictions.c.site_id, predictions.c.status, func.count()) + .select_from(predictions) + .group_by(predictions.c.site_id, predictions.c.status) + ) + + lignes = await self._session.execute(requete) + return [ + ComptageStatut(site_id=ligne[0], status=ligne[1], nombre=ligne[2]) for ligne in lignes + ] + + # Pourquoi : l'idempotence est déléguée à `uq_drift_report_window` plutôt qu'à une lecture + # préalable, comme pour les recommandations. Rejouer la commande sur la même fenêtre ne + # duplique donc rien. + async def enregistre(self, rapports: Sequence[NouveauRapportDerive]) -> int: + if not rapports: + return 0 + + valeurs = [asdict(rapport) for rapport in rapports] + requete = ( + insert(DriftReport) + .values(valeurs) + .on_conflict_do_nothing( + index_elements=[DriftReport.window_end, func.coalesce(DriftReport.site_id, "")] + ) + .returning(DriftReport.drift_report_id) + ) + return len((await self._session.scalars(requete)).all()) + + async def derniers(self, *, site_id: str | None = None) -> Sequence[DriftReport]: + requete = ( + select(DriftReport) + .distinct(DriftReport.site_id) + .order_by( + DriftReport.site_id, + DriftReport.computed_at.desc(), + DriftReport.drift_report_id.desc(), + ) + ) + if site_id is not None: + requete = requete.where(DriftReport.site_id == site_id) + return (await self._session.scalars(requete)).all() diff --git a/apps/backend/app/schemas/drift.py b/apps/backend/app/schemas/drift.py new file mode 100644 index 0000000..6c6780c --- /dev/null +++ b/apps/backend/app/schemas/drift.py @@ -0,0 +1,31 @@ +from datetime import datetime +from enum import StrEnum + +from pydantic import BaseModel, ConfigDict + + +class DriftStatus(StrEnum): + STABLE = "stable" + DERIVE = "derive" + INDETERMINE = "indetermine" + + +class DriftReportResponse(BaseModel): + model_config = ConfigDict(from_attributes=True) + + site_id: str | None + computed_at: datetime + window_start: datetime + window_end: datetime + reference_start: datetime | None + reference_end: datetime | None + n_observations: int + mae: float | None + mape: float | None + bias: float | None + reference_mae: float | None + coverage_ratio: float | None + insufficient_data_ratio: float | None + model_references: list[str] + status: DriftStatus + reason: str | None diff --git a/apps/backend/app/services/drift.py b/apps/backend/app/services/drift.py new file mode 100644 index 0000000..1ede7f5 --- /dev/null +++ b/apps/backend/app/services/drift.py @@ -0,0 +1,231 @@ +"""Contrainte : la dérive se mesure sur ce qui a déjà eu lieu - DriftService.evaluate() + +Une prévision ne devient vérifiable que quand la lecture de son instant cible est ingérée. La +fenêtre est donc fermée à droite par un délai de grâce : sans lui, la dernière heure ferait +chuter le taux de couverture à chaque exécution, et le verdict dirait « dérive » alors que +seule l'ingestion n'avait pas fini son tour. + +La comparaison se fait entre deux fenêtres vives de même durée, pas contre la métrique de +référence du modèle journalisée à l'entraînement. Ce ne sont pas les mêmes grandeurs : +l'entraînement mesure un backtest où la météo de l'heure cible est connue, le scoring prévoit +une heure future dont la météo ne l'est pas. Les comparer classerait le modèle « en dérive » +dès le premier jour, ce qui ne prouverait rien. +""" + +from collections.abc import Sequence +from dataclasses import dataclass, replace +from datetime import UTC, datetime, timedelta + +from app.models.energy import DriftReport +from app.repositories.drift import ( + ComptageStatut, + DriftRepository, + NouveauRapportDerive, + PaireDerive, +) + +STATUT_STABLE = "stable" +STATUT_DERIVE = "derive" +STATUT_INDETERMINE = "indetermine" + +STATUT_INSUFFISANT = "insufficient_data" +STATUT_DISPONIBLE = "available" + + +@dataclass(frozen=True, slots=True) +class Seuils: + # 168 h, la saisonnalité hebdomadaire que le modèle apprend par son lag principal : une + # fenêtre plus courte comparerait un week-end à une semaine ouvrée. + fenetre: timedelta = timedelta(hours=168) + grace: timedelta = timedelta(hours=2) + min_observations: int = 24 + ratio_derive: float = 1.25 + mae_plancher: float = 0.0 + seuil_biais: float = 0.0 + seuil_couverture: float = 0.8 + + +@dataclass(frozen=True, slots=True) +class Metriques: + n_observations: int + mae: float | None + mape: float | None + bias: float | None + model_references: list[str] + + +def mesure(paires: Sequence[PaireDerive]) -> Metriques: + if not paires: + return Metriques(n_observations=0, mae=None, mape=None, bias=None, model_references=[]) + + ecarts = [paire.predicted_value - paire.actual_value for paire in paires] + # Le MAPE diverge sur une consommation nulle : les sites à l'arrêt sortent de ce seul + # rapport, jamais des autres métriques. + ratios = [ + abs(ecart / paire.actual_value) + for ecart, paire in zip(ecarts, paires, strict=True) + if paire.actual_value != 0 + ] + + return Metriques( + n_observations=len(paires), + mae=sum(abs(ecart) for ecart in ecarts) / len(ecarts), + mape=(sum(ratios) / len(ratios) * 100) if ratios else None, + bias=sum(ecarts) / len(ecarts), + model_references=sorted({paire.model_reference for paire in paires}), + ) + + +@dataclass(frozen=True, slots=True) +class Verdict: + status: str + reason: str | None + + +class DriftService: + def __init__(self, depot: DriftRepository, *, seuils: Seuils | None = None) -> None: + self._depot = depot + self._seuils = seuils or Seuils() + + async def derniers(self, *, site_id: str | None = None) -> Sequence[DriftReport]: + """Ce que sert l'API : le dernier rapport de chaque site, plus la ligne globale.""" + return await self._depot.derniers(site_id=site_id) + + async def evaluate( + self, *, now: datetime | None = None, site_id: str | None = None + ) -> list[NouveauRapportDerive]: + """Une ligne par site, plus une ligne globale dont le `site_id` est nul.""" + fin = (now or datetime.now(UTC)) - self._seuils.grace + debut = fin - self._seuils.fenetre + reference_fin = debut + reference_debut = reference_fin - self._seuils.fenetre + + recentes = await self._depot.paires(debut=debut, fin=fin, site_id=site_id) + anciennes = await self._depot.paires( + debut=reference_debut, fin=reference_fin, site_id=site_id + ) + comptages = await self._depot.comptages(debut=debut, fin=fin, site_id=site_id) + + gabarit = NouveauRapportDerive( + site_id=None, + window_start=debut, + window_end=fin, + reference_start=reference_debut, + reference_end=reference_fin, + n_observations=0, + mae=None, + mape=None, + bias=None, + reference_mae=None, + coverage_ratio=None, + insufficient_data_ratio=None, + model_references=[], + status=STATUT_INDETERMINE, + reason=None, + ) + + rapports = [ + self._rapport( + gabarit, + site=site, + recentes=[p for p in recentes if p.site_id == site], + anciennes=[p for p in anciennes if p.site_id == site], + comptages=[c for c in comptages if c.site_id == site], + ) + for site in sorted( + {paire.site_id for paire in recentes} | {c.site_id for c in comptages} + ) + ] + rapports.append( + self._rapport( + gabarit, site=None, recentes=recentes, anciennes=anciennes, comptages=comptages + ) + ) + return rapports + + def _rapport( + self, + gabarit: NouveauRapportDerive, + *, + site: str | None, + recentes: Sequence[PaireDerive], + anciennes: Sequence[PaireDerive], + comptages: Sequence[ComptageStatut], + ) -> NouveauRapportDerive: + metriques = mesure(recentes) + reference = mesure(anciennes) + couverture = _couverture(len(recentes), comptages) + verdict = self._verdict(metriques, reference_mae=reference.mae, couverture=couverture) + + return replace( + gabarit, + site_id=site, + n_observations=metriques.n_observations, + mae=metriques.mae, + mape=metriques.mape, + bias=metriques.bias, + reference_mae=reference.mae, + coverage_ratio=couverture, + insufficient_data_ratio=_part_insuffisante(comptages), + model_references=metriques.model_references, + status=verdict.status, + reason=verdict.reason, + ) + + def _verdict( + self, metriques: Metriques, *, reference_mae: float | None, couverture: float | None + ) -> Verdict: + seuils = self._seuils + if metriques.n_observations < seuils.min_observations: + return Verdict( + STATUT_INDETERMINE, + f"{metriques.n_observations} prévision(s) vérifiée(s) sur la fenêtre, " + f"minimum {seuils.min_observations}.", + ) + + if couverture is not None and couverture < seuils.seuil_couverture: + return Verdict( + STATUT_DERIVE, + f"Couverture de {couverture:.0%}, sous le seuil de {seuils.seuil_couverture:.0%} : " + "le pipeline, pas le modèle.", + ) + + plafond = _plafond(reference_mae, ratio=seuils.ratio_derive, plancher=seuils.mae_plancher) + if metriques.mae is not None and plafond is not None and metriques.mae > plafond: + return Verdict( + STATUT_DERIVE, + f"MAE de {metriques.mae:.2f} kWh au-delà de {plafond:.2f} kWh, " + "seuil dérivé de la fenêtre de référence.", + ) + + if ( + seuils.seuil_biais > 0 + and metriques.bias is not None + and abs(metriques.bias) > seuils.seuil_biais + ): + return Verdict( + STATUT_DERIVE, + f"Biais de {metriques.bias:+.2f} kWh : le modèle se trompe toujours du même côté.", + ) + + return Verdict(STATUT_STABLE, None) + + +def _plafond(reference_mae: float | None, *, ratio: float, plancher: float) -> float | None: + if reference_mae is None: + return plancher or None + return max(plancher, reference_mae * ratio) + + +def _couverture(apparie: int, comptages: Sequence[ComptageStatut]) -> float | None: + """Part des prévisions disponibles qui ont trouvé leur réalisé. Mesure l'ingestion et + l'ordonnancement, pas la qualité du modèle.""" + disponibles = sum(c.nombre for c in comptages if c.status == STATUT_DISPONIBLE) + return apparie / disponibles if disponibles else None + + +def _part_insuffisante(comptages: Sequence[ComptageStatut]) -> float | None: + total = sum(c.nombre for c in comptages) + if not total: + return None + return sum(c.nombre for c in comptages if c.status == STATUT_INSUFFISANT) / total diff --git a/apps/backend/openapi.json b/apps/backend/openapi.json index 67b3877..df12efe 100644 --- a/apps/backend/openapi.json +++ b/apps/backend/openapi.json @@ -1859,6 +1859,94 @@ } ] } + }, + "/api/v1/monitoring/drift": { + "get": { + "tags": [ + "monitoring" + ], + "summary": "Dernier rapport de dérive par site, plus la ligne globale", + "operationId": "get_drift_api_v1_monitoring_drift_get", + "security": [ + { + "Jeton d'accès": [] + } + ], + "parameters": [ + { + "name": "site_id", + "in": "query", + "required": false, + "schema": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Site Id" + } + } + ], + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": { + "type": "array", + "items": { + "$ref": "#/components/schemas/DriftReportResponse" + }, + "title": "Response Get Drift Api V1 Monitoring Drift Get" + } + } + } + }, + "500": { + "description": "Erreur interne. `correlation` identifie la trace côté serveur, qui n'est pas renvoyée au client.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/InternalErrorResponse" + } + } + } + }, + "401": { + "description": "Jeton absent, illisible, périmé, ou rendu caduc par un changement de rôle ou une désactivation. L'en-tête `WWW-Authenticate` porte la cause dans `error=`.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "403": { + "description": "Droits insuffisants, ou mot de passe provisoire à changer quand `detail` vaut `password_change_required`.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ErrorResponse" + } + } + } + }, + "422": { + "description": "Corps invalide. Le détail nomme le champ fautif et le type d'erreur, jamais la valeur envoyée.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/ValidationErrorResponse" + } + } + } + } + } + } } }, "components": { @@ -1977,6 +2065,180 @@ ], "title": "AlertType" }, + "DriftReportResponse": { + "properties": { + "site_id": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Site Id" + }, + "computed_at": { + "type": "string", + "format": "date-time", + "title": "Computed At" + }, + "window_start": { + "type": "string", + "format": "date-time", + "title": "Window Start" + }, + "window_end": { + "type": "string", + "format": "date-time", + "title": "Window End" + }, + "reference_start": { + "anyOf": [ + { + "type": "string", + "format": "date-time" + }, + { + "type": "null" + } + ], + "title": "Reference Start" + }, + "reference_end": { + "anyOf": [ + { + "type": "string", + "format": "date-time" + }, + { + "type": "null" + } + ], + "title": "Reference End" + }, + "n_observations": { + "type": "integer", + "title": "N Observations" + }, + "mae": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "title": "Mae" + }, + "mape": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "title": "Mape" + }, + "bias": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "title": "Bias" + }, + "reference_mae": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "title": "Reference Mae" + }, + "coverage_ratio": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "title": "Coverage Ratio" + }, + "insufficient_data_ratio": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "title": "Insufficient Data Ratio" + }, + "model_references": { + "items": { + "type": "string" + }, + "type": "array", + "title": "Model References" + }, + "status": { + "$ref": "#/components/schemas/DriftStatus" + }, + "reason": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "title": "Reason" + } + }, + "type": "object", + "required": [ + "site_id", + "computed_at", + "window_start", + "window_end", + "reference_start", + "reference_end", + "n_observations", + "mae", + "mape", + "bias", + "reference_mae", + "coverage_ratio", + "insufficient_data_ratio", + "model_references", + "status", + "reason" + ], + "title": "DriftReportResponse" + }, + "DriftStatus": { + "type": "string", + "enum": [ + "stable", + "derive", + "indetermine" + ], + "title": "DriftStatus" + }, "ErrorResponse": { "properties": { "detail": { @@ -3277,6 +3539,10 @@ { "name": "predictions", "description": "Dernière prévision de consommation par site, calculée hors ligne par le pipeline de scoring (`ml/`) et simplement lue ici. Accessible à partir du rôle `lecteur`." + }, + { + "name": "monitoring", + "description": "Surveillance de la dérive du modèle : écart entre les prévisions déjà écrites et les lectures réellement arrivées, par site et tous sites confondus. Réservé à partir du rôle `operateur`, qui agit sur un pipeline dégradé." } ] } diff --git a/apps/backend/tests/api/acces.py b/apps/backend/tests/api/acces.py index 2b38374..2b1f640 100644 --- a/apps/backend/tests/api/acces.py +++ b/apps/backend/tests/api/acces.py @@ -56,6 +56,7 @@ ROLE_MINIMUM: Final[dict[Route, Role]] = { ("GET", "/api/v1/readings"): Role.LECTEUR, ("GET", "/api/v1/predictions"): Role.LECTEUR, ("GET", "/api/v1/sensors/status"): Role.ADMIN, + ("GET", "/api/v1/monitoring/drift"): Role.OPERATEUR, ("GET", "/api/v1/users"): Role.ADMIN, ("POST", "/api/v1/users"): Role.ADMIN, ("PATCH", "/api/v1/users/{user_id}"): Role.ADMIN, diff --git a/apps/backend/tests/api/test_monitoring.py b/apps/backend/tests/api/test_monitoring.py new file mode 100644 index 0000000..68b2da1 --- /dev/null +++ b/apps/backend/tests/api/test_monitoring.py @@ -0,0 +1,99 @@ +from collections.abc import Iterator, Sequence +from datetime import UTC, datetime, timedelta +from uuid import uuid4 + +import pytest +from fastapi import FastAPI +from httpx import AsyncClient + +from app.api.deps import get_current_principal, get_drift_service +from app.core.principal import Principal +from app.core.roles import AccountKind, Role +from app.models.energy import DriftReport + +INSTANT = datetime(2026, 9, 22, 12, tzinfo=UTC) + + +def operateur() -> Principal: + return Principal( + id=uuid4(), + email="operateur@enervision.fr", + role=Role.OPERATEUR, + kind=AccountKind.HUMAIN, + must_change_password=False, + ) + + +def rapport(*, site_id: str | None) -> DriftReport: + return DriftReport( + drift_report_id=1, + computed_at=INSTANT, + site_id=site_id, + window_start=INSTANT - timedelta(hours=168), + window_end=INSTANT, + reference_start=None, + reference_end=None, + n_observations=48, + mae=1.5, + mape=12.0, + bias=0.3, + reference_mae=1.2, + coverage_ratio=0.95, + insufficient_data_ratio=0.0, + model_references=["lightgbm-aaa"], + status="stable", + reason=None, + ) + + +class FauxService: + def __init__(self, rapports: Sequence[DriftReport]) -> None: + self.rapports = list(rapports) + self.site_demande: str | None = None + + async def derniers(self, *, site_id: str | None = None) -> Sequence[DriftReport]: + self.site_demande = site_id + return self.rapports + + +@pytest.fixture +def servi(app: FastAPI) -> Iterator[list[DriftReport]]: + rapports = [rapport(site_id="SITE001"), rapport(site_id=None)] + service = FauxService(rapports) + app.dependency_overrides[get_current_principal] = operateur + app.dependency_overrides[get_drift_service] = lambda: service + yield rapports + app.dependency_overrides.clear() + + +async def test_drift_returns_the_latest_report_of_every_site( + servi: list[DriftReport], client: AsyncClient +) -> None: + reponse = await client.get("/api/v1/monitoring/drift") + + assert reponse.status_code == 200 + assert [ligne["site_id"] for ligne in reponse.json()] == ["SITE001", None] + + +async def test_drift_exposes_the_metrics_of_the_stored_report( + servi: list[DriftReport], client: AsyncClient +) -> None: + reponse = await client.get("/api/v1/monitoring/drift") + + premier = reponse.json()[0] + assert premier["status"] == "stable" + assert premier["mae"] == 1.5 + assert premier["model_references"] == ["lightgbm-aaa"] + + +async def test_drift_returns_an_empty_list_when_no_report_exists( + app: FastAPI, client: AsyncClient +) -> None: + app.dependency_overrides[get_current_principal] = operateur + app.dependency_overrides[get_drift_service] = lambda: FauxService([]) + + reponse = await client.get("/api/v1/monitoring/drift") + + assert reponse.status_code == 200 + assert reponse.json() == [] + app.dependency_overrides.clear() diff --git a/apps/backend/tests/db/test_data_schema.py b/apps/backend/tests/db/test_data_schema.py index 52295aa..e816d9c 100644 --- a/apps/backend/tests/db/test_data_schema.py +++ b/apps/backend/tests/db/test_data_schema.py @@ -1,5 +1,5 @@ from collections.abc import AsyncIterator -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from uuid import uuid4 import pytest @@ -9,7 +9,15 @@ from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncConnection, create_async_engine from app.core.config import get_settings -from app.models.energy import Alert, Dataset, Prediction, Reading, Recommendation, Site +from app.models.energy import ( + Alert, + Dataset, + DriftReport, + Prediction, + Reading, + Recommendation, + Site, +) pytestmark = pytest.mark.integration MOMENT = datetime(2024, 1, 1, tzinfo=UTC) @@ -269,3 +277,62 @@ async def test_recommendation_is_unique_when_alert_and_rule_match( with pytest.raises(IntegrityError): async with savepoint: await data_connection.execute(statement) + + +def _rapport(**remplacements: object) -> dict[str, object]: + defauts: dict[str, object] = { + "site_id": None, + "window_start": MOMENT, + "window_end": MOMENT, + "n_observations": 12, + "model_references": ["lightgbm-aaa"], + "status": "stable", + "reason": None, + } + return {**defauts, **remplacements} + + +async def test_drift_report_rejects_an_unknown_status(data_connection: AsyncConnection) -> None: + statement = insert(DriftReport).values(**_rapport(status="douteux", reason="x")) + savepoint = data_connection.begin_nested() + + with pytest.raises(IntegrityError): + async with savepoint: + await data_connection.execute(statement) + + +async def test_drift_report_rejects_a_drift_without_a_reason( + data_connection: AsyncConnection, +) -> None: + statement = insert(DriftReport).values(**_rapport(status="derive")) + savepoint = data_connection.begin_nested() + + with pytest.raises(IntegrityError): + async with savepoint: + await data_connection.execute(statement) + + +async def test_drift_report_accepts_one_global_row_without_a_site( + data_connection: AsyncConnection, +) -> None: + identifiant = ( + await data_connection.execute( + insert(DriftReport).values(**_rapport()).returning(DriftReport.drift_report_id) + ) + ).scalar_one() + + assert identifiant is not None + + +async def test_drift_report_is_unique_when_window_and_site_match( + data_connection: AsyncConnection, +) -> None: + fenetre = MOMENT + timedelta(days=1) + statement = insert(DriftReport).values(**_rapport(window_end=fenetre)) + await data_connection.execute(statement) + + savepoint = data_connection.begin_nested() + + with pytest.raises(IntegrityError): + async with savepoint: + await data_connection.execute(statement) diff --git a/apps/backend/tests/repositories/test_drift.py b/apps/backend/tests/repositories/test_drift.py new file mode 100644 index 0000000..255abbe --- /dev/null +++ b/apps/backend/tests/repositories/test_drift.py @@ -0,0 +1,187 @@ +from datetime import UTC, datetime, timedelta + +import pytest +from sqlalchemy.dialects import postgresql +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.sql import ClauseElement + +from app.repositories.drift import ( + DriftRepository, + NouveauRapportDerive, + _lectures_retenues, + _predictions_retenues, +) +from tests.repositories.test_prediction import creer_prediction +from tests.repositories.test_reading import creer_lecture +from tests.repositories.test_site import creer as creer_site + +DEBUT = datetime(2026, 9, 15, tzinfo=UTC) +FIN = datetime(2026, 9, 22, tzinfo=UTC) +CIBLE = datetime(2026, 9, 16, 12, tzinfo=UTC) + + +def sql(requete: ClauseElement) -> str: + return str(requete.compile(dialect=postgresql.dialect())) # type: ignore[no-untyped-call] + + +def rapport(**remplacements: object) -> NouveauRapportDerive: + defauts: dict[str, object] = { + "site_id": None, + "window_start": DEBUT, + "window_end": FIN, + "reference_start": None, + "reference_end": None, + "n_observations": 10, + "mae": 1.0, + "mape": 5.0, + "bias": 0.1, + "reference_mae": None, + "coverage_ratio": 1.0, + "insufficient_data_ratio": 0.0, + "model_references": ["lightgbm-aaa"], + "status": "stable", + "reason": None, + } + return NouveauRapportDerive(**{**defauts, **remplacements}) # type: ignore[arg-type] + + +def test_predictions_keep_one_row_per_site_and_target_in_sql() -> None: + requete = sql(_predictions_retenues(debut=DEBUT, fin=FIN, site_id=None).element) + + assert "DISTINCT ON (prediction.site_id, prediction.target_at)" in requete + assert "prediction.prediction_id DESC" in requete + + +def test_readings_keep_one_row_per_site_and_instant_in_sql() -> None: + requete = sql(_lectures_retenues(debut=DEBUT, fin=FIN, site_id=None).element) + + assert "DISTINCT ON (reading.site_id, reading.timestamp)" in requete + assert "reading.reading_id DESC" in requete + + +def test_predictions_restrict_themselves_to_the_requested_site_in_sql() -> None: + requete = sql(_predictions_retenues(debut=DEBUT, fin=FIN, site_id="SITE001").element) + + assert requete.count("prediction.site_id = ") == 1 + + +def test_readings_ignore_a_missing_consumption_in_sql() -> None: + requete = sql(_lectures_retenues(debut=DEBUT, fin=FIN, site_id=None).element) + + assert "reading.consumption_kwh IS NOT NULL" in requete + + +@pytest.mark.integration +async def test_repository_pairs_a_prediction_with_the_reading_of_the_same_instant( + session: AsyncSession, +) -> None: + site = await creer_site(session) + await creer_prediction(session, site_id=site.site_id, target_at=CIBLE, predicted_value=12.0) + await creer_lecture(session, site_id=site.site_id, timestamp=CIBLE, consumption_kwh=10.0) + + paires = await DriftRepository(session).paires(debut=DEBUT, fin=FIN, site_id=site.site_id) + await session.rollback() + + assert [(p.predicted_value, p.actual_value) for p in paires] == [(12.0, 10.0)] + + +@pytest.mark.integration +async def test_repository_keeps_the_latest_run_when_several_predictions_share_a_target( + session: AsyncSession, +) -> None: + site = await creer_site(session) + await creer_prediction(session, site_id=site.site_id, target_at=CIBLE, predicted_value=12.0) + await creer_prediction(session, site_id=site.site_id, target_at=CIBLE, predicted_value=99.0) + await creer_lecture(session, site_id=site.site_id, timestamp=CIBLE, consumption_kwh=10.0) + + paires = await DriftRepository(session).paires(debut=DEBUT, fin=FIN, site_id=site.site_id) + await session.rollback() + + assert [p.predicted_value for p in paires] == [99.0] + + +@pytest.mark.integration +async def test_repository_keeps_one_reading_per_instant_when_two_sources_wrote_the_same_hour( + session: AsyncSession, +) -> None: + site = await creer_site(session) + await creer_prediction(session, site_id=site.site_id, target_at=CIBLE, predicted_value=12.0) + await creer_lecture( + session, site_id=site.site_id, timestamp=CIBLE, source="api_current", consumption_kwh=10.0 + ) + await creer_lecture( + session, site_id=site.site_id, timestamp=CIBLE, source="api_history", consumption_kwh=20.0 + ) + + paires = await DriftRepository(session).paires(debut=DEBUT, fin=FIN, site_id=site.site_id) + await session.rollback() + + assert [p.actual_value for p in paires] == [20.0] + + +@pytest.mark.integration +async def test_repository_excludes_an_insufficient_data_prediction_from_the_pairs( + session: AsyncSession, +) -> None: + site = await creer_site(session) + await creer_prediction( + session, + site_id=site.site_id, + target_at=CIBLE, + predicted_value=None, + status="insufficient_data", + failure_reason="historique trop court", + ) + await creer_lecture(session, site_id=site.site_id, timestamp=CIBLE, consumption_kwh=10.0) + + depot = DriftRepository(session) + paires = await depot.paires(debut=DEBUT, fin=FIN, site_id=site.site_id) + comptages = await depot.comptages(debut=DEBUT, fin=FIN, site_id=site.site_id) + await session.rollback() + + assert paires == [] + assert [(c.status, c.nombre) for c in comptages] == [("insufficient_data", 1)] + + +@pytest.mark.integration +async def test_repository_excludes_a_target_outside_the_window(session: AsyncSession) -> None: + site = await creer_site(session) + hors_fenetre = FIN + timedelta(hours=1) + await creer_prediction( + session, site_id=site.site_id, target_at=hors_fenetre, predicted_value=12.0 + ) + await creer_lecture(session, site_id=site.site_id, timestamp=hors_fenetre, consumption_kwh=10.0) + + paires = await DriftRepository(session).paires(debut=DEBUT, fin=FIN, site_id=site.site_id) + await session.rollback() + + assert paires == [] + + +@pytest.mark.integration +async def test_repository_reads_back_the_global_report_it_wrote(session: AsyncSession) -> None: + depot = DriftRepository(session) + fenetre = datetime(2035, 3, 1, tzinfo=UTC) + + ecrites = await depot.enregistre([rapport(window_end=fenetre)]) + derniers = await depot.derniers() + globaux = [r for r in derniers if r.site_id is None and r.window_end == fenetre] + await session.rollback() + + assert ecrites == 1 + assert len(globaux) == 1 + + +@pytest.mark.integration +async def test_repository_ignores_a_second_report_for_the_same_window_and_site( + session: AsyncSession, +) -> None: + depot = DriftRepository(session) + fenetre = datetime(2035, 4, 1, tzinfo=UTC) + + premiere = await depot.enregistre([rapport(window_end=fenetre)]) + seconde = await depot.enregistre([rapport(window_end=fenetre, status="derive", reason="x")]) + await session.rollback() + + assert premiere == 1 + assert seconde == 0 diff --git a/apps/backend/tests/services/test_drift.py b/apps/backend/tests/services/test_drift.py new file mode 100644 index 0000000..ad6a0f2 --- /dev/null +++ b/apps/backend/tests/services/test_drift.py @@ -0,0 +1,220 @@ +from collections.abc import Sequence +from datetime import UTC, datetime, timedelta + +import pytest + +from app.repositories.drift import ComptageStatut, PaireDerive +from app.services.drift import ( + STATUT_DERIVE, + STATUT_INDETERMINE, + STATUT_STABLE, + DriftService, + Seuils, + mesure, +) + +INSTANT = datetime(2026, 9, 22, 12, 0, tzinfo=UTC) + + +def paire( + *, site_id: str = "SITE001", prevu: float, reel: float, reference: str = "lightgbm-aaa" +) -> PaireDerive: + return PaireDerive( + site_id=site_id, + target_at=INSTANT, + predicted_value=prevu, + actual_value=reel, + model_reference=reference, + ) + + +def paires( + *, site_id: str = "SITE001", nombre: int, prevu: float, reel: float +) -> list[PaireDerive]: + return [paire(site_id=site_id, prevu=prevu, reel=reel) for _ in range(nombre)] + + +class FauxDepot: + def __init__( + self, + *, + recentes: Sequence[PaireDerive] = (), + anciennes: Sequence[PaireDerive] = (), + comptages: Sequence[ComptageStatut] = (), + ) -> None: + self.recentes = list(recentes) + self.anciennes = list(anciennes) + self._comptages = list(comptages) + self.fenetres: list[tuple[datetime, datetime]] = [] + + async def paires( + self, *, debut: datetime, fin: datetime, site_id: str | None = None + ) -> Sequence[PaireDerive]: + self.fenetres.append((debut, fin)) + return self.recentes if len(self.fenetres) == 1 else self.anciennes + + async def comptages( + self, *, debut: datetime, fin: datetime, site_id: str | None = None + ) -> Sequence[ComptageStatut]: + return self._comptages + + +def service(depot: FauxDepot, **surcharges: object) -> DriftService: + return DriftService(depot, seuils=Seuils(**surcharges)) # type: ignore[arg-type] + + +def test_drift_averages_the_absolute_gap_between_forecast_and_actual() -> None: + metriques = mesure([paire(prevu=12.0, reel=10.0), paire(prevu=8.0, reel=10.0)]) + + assert metriques.mae == 2.0 + assert metriques.n_observations == 2 + + +def test_drift_computes_a_signed_bias_when_the_model_overforecasts() -> None: + metriques = mesure([paire(prevu=12.0, reel=10.0), paire(prevu=14.0, reel=10.0)]) + + assert metriques.bias == 3.0 + + +def test_drift_computes_a_negative_bias_when_the_model_underforecasts() -> None: + metriques = mesure([paire(prevu=8.0, reel=10.0), paire(prevu=6.0, reel=10.0)]) + + assert metriques.bias == -3.0 + + +def test_drift_excludes_a_zero_actual_from_the_mape_only() -> None: + metriques = mesure([paire(prevu=11.0, reel=10.0), paire(prevu=5.0, reel=0.0)]) + + assert metriques.mape == 10.0 + assert metriques.n_observations == 2 + assert metriques.mae == 3.0 + + +def test_drift_reports_no_mape_when_every_actual_is_zero() -> None: + metriques = mesure([paire(prevu=1.0, reel=0.0)]) + + assert metriques.mape is None + + +def test_drift_lists_every_model_reference_seen_in_the_window() -> None: + metriques = mesure( + [paire(prevu=10.0, reel=10.0, reference="lightgbm-bbb"), paire(prevu=10.0, reel=10.0)] + ) + + assert metriques.model_references == ["lightgbm-aaa", "lightgbm-bbb"] + + +async def test_drift_reports_indetermine_when_the_window_holds_too_few_observations() -> None: + depot = FauxDepot(recentes=paires(nombre=3, prevu=10.0, reel=10.0)) + + rapports = await service(depot, min_observations=24).evaluate(now=INSTANT) + + assert {rapport.status for rapport in rapports} == {STATUT_INDETERMINE} + assert all(rapport.reason for rapport in rapports) + + +async def test_drift_reports_derive_when_the_recent_mae_exceeds_the_reference_ratio() -> None: + depot = FauxDepot( + recentes=paires(nombre=30, prevu=14.0, reel=10.0), + anciennes=paires(nombre=30, prevu=11.0, reel=10.0), + comptages=[ComptageStatut(site_id="SITE001", status="available", nombre=30)], + ) + + rapports = await service(depot, min_observations=10).evaluate(now=INSTANT) + + global_ = next(rapport for rapport in rapports if rapport.site_id is None) + assert global_.status == STATUT_DERIVE + assert global_.mae == 4.0 + assert global_.reference_mae == 1.0 + + +async def test_drift_reports_stable_when_the_recent_mae_stays_close_to_the_reference() -> None: + depot = FauxDepot( + recentes=paires(nombre=30, prevu=11.0, reel=10.0), + anciennes=paires(nombre=30, prevu=11.0, reel=10.0), + comptages=[ComptageStatut(site_id="SITE001", status="available", nombre=30)], + ) + + rapports = await service(depot, min_observations=10).evaluate(now=INSTANT) + + global_ = next(rapport for rapport in rapports if rapport.site_id is None) + assert global_.status == STATUT_STABLE + assert global_.reason is None + + +async def test_drift_reports_derive_when_the_coverage_ratio_falls_under_the_threshold() -> None: + depot = FauxDepot( + recentes=paires(nombre=30, prevu=10.0, reel=10.0), + anciennes=paires(nombre=30, prevu=10.0, reel=10.0), + comptages=[ComptageStatut(site_id="SITE001", status="available", nombre=100)], + ) + + rapports = await service(depot, min_observations=10).evaluate(now=INSTANT) + + global_ = next(rapport for rapport in rapports if rapport.site_id is None) + assert global_.status == STATUT_DERIVE + assert global_.coverage_ratio == 0.3 + + +async def test_drift_reports_one_line_per_site_and_one_global_line() -> None: + depot = FauxDepot( + recentes=[ + *paires(site_id="SITE001", nombre=12, prevu=10.0, reel=10.0), + *paires(site_id="SITE002", nombre=12, prevu=10.0, reel=10.0), + ], + comptages=[ + ComptageStatut(site_id="SITE001", status="available", nombre=12), + ComptageStatut(site_id="SITE002", status="available", nombre=12), + ], + ) + + rapports = await service(depot, min_observations=10).evaluate(now=INSTANT) + + assert [rapport.site_id for rapport in rapports] == ["SITE001", "SITE002", None] + assert next(r for r in rapports if r.site_id is None).n_observations == 24 + + +async def test_drift_measures_the_share_of_sites_left_without_enough_history() -> None: + depot = FauxDepot( + recentes=paires(nombre=30, prevu=10.0, reel=10.0), + comptages=[ + ComptageStatut(site_id="SITE001", status="available", nombre=30), + ComptageStatut(site_id="SITE001", status="insufficient_data", nombre=10), + ], + ) + + rapports = await service(depot, min_observations=10).evaluate(now=INSTANT) + + assert next(r for r in rapports if r.site_id is None).insufficient_data_ratio == 0.25 + + +async def test_drift_closes_the_window_before_the_grace_delay() -> None: + depot = FauxDepot() + + await service(depot, grace=timedelta(hours=2), fenetre=timedelta(hours=168)).evaluate( + now=INSTANT + ) + + recente, reference = depot.fenetres + assert recente[1] == INSTANT - timedelta(hours=2) + assert recente[0] == INSTANT - timedelta(hours=170) + assert reference[1] == recente[0] + + +@pytest.mark.parametrize( + ("prevu", "attendu"), + [(10.0, STATUT_STABLE), (30.0, STATUT_DERIVE)], + ids=["mae_stable", "mae_triplee"], +) +async def test_drift_compares_the_recent_window_to_the_reference_one( + prevu: float, attendu: str +) -> None: + depot = FauxDepot( + recentes=paires(nombre=30, prevu=prevu, reel=10.0), + anciennes=paires(nombre=30, prevu=10.0, reel=10.0), + comptages=[ComptageStatut(site_id="SITE001", status="available", nombre=30)], + ) + + rapports = await service(depot, min_observations=10, mae_plancher=1.0).evaluate(now=INSTANT) + + assert next(r for r in rapports if r.site_id is None).status == attendu diff --git a/apps/backend/tests/test_drift_cli.py b/apps/backend/tests/test_drift_cli.py new file mode 100644 index 0000000..2400472 --- /dev/null +++ b/apps/backend/tests/test_drift_cli.py @@ -0,0 +1,108 @@ +from datetime import UTC, datetime, timedelta + +import pytest + +from app.monitoring import drift as cli +from app.repositories.drift import NouveauRapportDerive +from app.services.drift import STATUT_DERIVE, STATUT_STABLE, Seuils + +INSTANT = datetime(2026, 9, 22, 12, tzinfo=UTC) + + +def rapport(*, site_id: str | None, status: str, reason: str | None = None) -> NouveauRapportDerive: + return NouveauRapportDerive( + site_id=site_id, + window_start=INSTANT - timedelta(hours=168), + window_end=INSTANT, + reference_start=None, + reference_end=None, + n_observations=48, + mae=1.5, + mape=12.0, + bias=0.3, + reference_mae=1.2, + coverage_ratio=1.0, + insufficient_data_ratio=0.0, + model_references=["lightgbm-aaa"], + status=status, + reason=reason, + ) + + +def installe(monkeypatch: pytest.MonkeyPatch, rapports: list[NouveauRapportDerive]) -> None: + async def fausse_execution( + *, now: datetime | None, site_id: str | None, seuils: Seuils | None + ) -> list[NouveauRapportDerive]: + return rapports + + monkeypatch.setattr(cli, "run_drift", fausse_execution) + + +def test_parse_args_defaults_to_the_standard_window() -> None: + arguments = cli.parse_args([]) + + assert arguments.window_hours == 168 + assert arguments.grace_hours == 2 + assert arguments.fail_on_drift is False + + +def test_parse_args_reads_the_site_id() -> None: + assert cli.parse_args(["--site-id", "SITE001"]).site_id == "SITE001" + + +def test_parse_args_parses_the_instant_option() -> None: + arguments = cli.parse_args(["--now", "2026-09-22T12:00:00+00:00"]) + + assert arguments.now == INSTANT + + +def test_parse_instant_treats_a_naive_datetime_as_utc() -> None: + assert cli._parse_instant("2026-09-22T12:00:00") == INSTANT + + +def test_seuils_depuis_translates_the_hour_options_into_durations() -> None: + seuils = cli.seuils_depuis(cli.parse_args(["--window-hours", "24", "--grace-hours", "1"])) + + assert seuils.fenetre == timedelta(hours=24) + assert seuils.grace == timedelta(hours=1) + + +def test_main_prints_the_verdict_of_every_line( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + installe( + monkeypatch, + [ + rapport(site_id="SITE001", status=STATUT_STABLE), + rapport(site_id=None, status=STATUT_STABLE), + ], + ) + + code = cli.main([]) + + sortie = capsys.readouterr().out + assert code == 0 + assert "SITE001" in sortie + assert "TOUS SITES" in sortie + + +def test_main_exits_non_zero_when_drift_is_detected_and_the_flag_is_set( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + installe(monkeypatch, [rapport(site_id=None, status=STATUT_DERIVE, reason="MAE doublée")]) + + code = cli.main(["--fail-on-drift"]) + + assert code == 1 + assert "MAE doublée" in capsys.readouterr().out + + +def test_main_exits_zero_when_drift_is_detected_without_the_flag( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + installe(monkeypatch, [rapport(site_id=None, status=STATUT_DERIVE, reason="MAE doublée")]) + + code = cli.main([]) + + assert code == 0 + assert capsys.readouterr().out != ""