ENI-projet-piscine/apps/backend/app/monitoring/drift.py

# 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(
        "--bias-threshold",
        type=float,
        default=defauts.seuil_biais,
        help=(
            "Biais absolu en kWh au-delà duquel le verdict bascule en dérive. "
            "Zéro, le défaut, laisse le biais informatif : voir l'ADR 0011."
        ),
    )
    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,
        seuil_biais=args.bias_threshold,
    )


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())