feat(backend): moteur de règles de recommandations et route de génération

`recommendation` n'avait aucun écrivain : les quatre couches de lecture étaient
livrées, mais rien ne produisait de ligne. Le moteur comble ce trou.

Le catalogue `REGLES` vit dans `app/services/`, pas dans `ml/` : il lit `alert.type`,
`alert.severity`, `alert.value` et `alert.threshold`, sans modèle ni feature, et
s'appuie sur deux repositories existants. L'arbitrage avec l'ADR 0005, qui annonçait
#38 du côté ML, est tranché par l'ADR 0006.

Sept règles, cinq par type d'alerte et deux transverses (sévérité critique,
dépassement d'au moins 20 % du seuil), donc une à trois recommandations par alerte.
L'idempotence est portée par la base : `create_missing()` insère en
`ON CONFLICT DO NOTHING` sur `uq_recommendation_alert_rule`, ce qui supprime la
fenêtre entre un contrôle préalable et l'insertion. `rule_reference` devient de ce
fait une clé fonctionnelle, d'où le suffixe de version sur chaque référence.

Deux déclencheurs : `POST /api/v1/recommendations/generate` réservé `admin`, et
`python -m app.cli generate-recommendations` (cible `make recommendations`).

Limite connue : aucune source n'alimente `alert` aujourd'hui, ni détection interne
(#104) ni ingestion de l'API Mock. La route répond, le rapport reste à zéro, et la
chaîne s'allume sans retoucher le moteur le jour où les alertes existent.

Tests : 80 unitaires et API verts, plus 6 d'intégration dont l'idempotence jouée
contre PostgreSQL.

Closes #38
This commit is contained in:
Johan LEROY
2026-09-18 15:49:54 +02:00
parent 619024f547
commit aeb07e14db
21 changed files with 854 additions and 23 deletions
@@ -0,0 +1,142 @@
from datetime import UTC, datetime
import pytest
from app.models.energy import Alert
from app.services.recommendation_rules import FACTEUR_DEPASSEMENT_MAJEUR, applique_les_regles
MOMENT = datetime(2024, 1, 1, tzinfo=UTC)
def alerte(
*,
alert_id: int = 1,
type_alerte: str = "spike",
severity: str = "high",
value: float | None = None,
threshold: float | None = None,
metric: str | None = None,
site_id: str = "SITE001",
) -> Alert:
return Alert(
alert_id=alert_id,
source_alert_id=f"ALR-{alert_id}",
site_id=site_id,
source="api_mock",
timestamp=MOMENT,
type=type_alerte,
severity=severity,
message="Alerte de test",
value=value,
threshold=threshold,
metric=metric,
prediction_id=None,
raw_data={},
)
@pytest.mark.parametrize(
("type_alerte", "attendue"),
[
("spike", "spike-delestage-v1"),
("threshold", "threshold-reduction-v1"),
("outage", "outage-secours-v1"),
("sensor", "sensor-maintenance-v1"),
("anomaly", "anomaly-verification-v1"),
],
ids=["pic", "seuil", "coupure", "capteur", "anomalie"],
)
def test_each_alert_type_yields_its_own_rule(type_alerte: str, attendue: str) -> None:
proposees = applique_les_regles(alerte(type_alerte=type_alerte))
assert [p.rule_reference for p in proposees] == [attendue]
def test_a_critical_alert_adds_the_escalation_rule() -> None:
proposees = applique_les_regles(alerte(severity="critical"))
assert "escalade-astreinte-v1" in {p.rule_reference for p in proposees}
@pytest.mark.parametrize("severity", ["low", "medium", "high"], ids=["faible", "moyenne", "haute"])
def test_a_non_critical_alert_does_not_escalate(severity: str) -> None:
proposees = applique_les_regles(alerte(severity=severity))
assert "escalade-astreinte-v1" not in {p.rule_reference for p in proposees}
def test_a_large_overshoot_adds_the_contract_rule() -> None:
proposees = applique_les_regles(
alerte(value=720.0 * FACTEUR_DEPASSEMENT_MAJEUR, threshold=720.0)
)
assert "contrat-puissance-v1" in {p.rule_reference for p in proposees}
def test_an_overshoot_below_the_factor_does_not_add_the_contract_rule() -> None:
proposees = applique_les_regles(alerte(value=800.0, threshold=720.0))
assert "contrat-puissance-v1" not in {p.rule_reference for p in proposees}
@pytest.mark.parametrize(
("value", "threshold"),
[(None, 720.0), (900.0, None), (900.0, 0.0), (900.0, -10.0)],
ids=["sans mesure", "sans seuil", "seuil nul", "seuil negatif"],
)
def test_the_contract_rule_stays_silent_without_an_exploitable_threshold(
value: float | None, threshold: float | None
) -> None:
proposees = applique_les_regles(alerte(value=value, threshold=threshold))
assert "contrat-puissance-v1" not in {p.rule_reference for p in proposees}
def test_the_explanation_quotes_the_measure_and_the_threshold() -> None:
proposees = applique_les_regles(alerte(value=812.5, threshold=720.0, metric="consumption_kw"))
assert "(consumption_kw mesurée à 812.5, seuil 720.0)" in proposees[0].explanation
def test_the_explanation_quotes_the_measure_alone_when_no_threshold_is_known() -> None:
proposees = applique_les_regles(alerte(value=812.5, metric="consumption_kw"))
assert "(consumption_kw mesurée à 812.5)" in proposees[0].explanation
def test_the_explanation_omits_the_measure_when_the_alert_carries_none() -> None:
proposees = applique_les_regles(alerte())
assert "(" not in proposees[0].explanation
def test_the_explanation_names_the_site() -> None:
proposees = applique_les_regles(alerte(site_id="SITE042"))
assert "SITE042" in proposees[0].explanation
def test_every_proposal_carries_the_alert_identifier() -> None:
proposees = applique_les_regles(alerte(alert_id=77, severity="critical"))
assert {p.alert_id for p in proposees} == {77}
def test_an_alert_never_yields_the_same_rule_twice() -> None:
proposees = applique_les_regles(
alerte(severity="critical", value=900.0, threshold=720.0, metric="consumption_kw")
)
assert len(proposees) == len({p.rule_reference for p in proposees})
def test_a_critical_alert_over_the_threshold_yields_the_three_rules() -> None:
proposees = applique_les_regles(
alerte(severity="critical", value=900.0, threshold=720.0, metric="consumption_kw")
)
assert {p.rule_reference for p in proposees} == {
"spike-delestage-v1",
"escalade-astreinte-v1",
"contrat-puissance-v1",
}