fix(meters): automate statistical anomaly detection

app_new/core/models/planning.py

@@ -511,6 +511,7 @@ class MeterAlertRule(db.Model):
     id = db.Column(db.Integer, primary_key=True)
     meter_id = db.Column(db.Integer, db.ForeignKey("meters.id", ondelete="CASCADE"), nullable=False, index=True)
     metric = db.Column(db.String(60), nullable=False)
+    rule_type = db.Column(db.String(30), nullable=False, default="MANUAL_THRESHOLD", server_default="MANUAL_THRESHOLD")
     period = db.Column(db.String(10), nullable=False, default="DAY")
     context_filter = db.Column(db.String(40), nullable=True)
     operator = db.Column(db.String(2), nullable=False, default=">")

app_new/core/services/meter_analytics.py

@@ -204,7 +204,7 @@ def statistical_anomaly(meter, interval, min_intervals=3, minimum_deviation_perc
     if rule is not None:
         min_intervals = rule.min_comparable_intervals
         minimum_deviation_percent = rule.min_deviation_percent
-    history = [i for i in consumption_intervals(meter) if i.end_date < interval.start_date and i.context == interval.context]
+    history = [i for i in consumption_intervals(meter) if i.end_date <= interval.start_date and i.context == interval.context]
     history = [i for i in history if not MeterAlertEvidence.query.filter_by(reading_end_id=i.reading_end.id, excluded_from_baseline=True).first()]
     if len(history) < min_intervals: return None
     reference = median(i.consumption_per_day for i in history)
@@ -346,7 +346,10 @@ def recalculate_meter_after_reading(*, meter, reading, commit=True):
     all_intervals = consumption_intervals(meter)
     for rule in rules:
         for interval in impacted:
-            evaluate_alert_rule(rule, interval, intervals=all_intervals, commit=False)
+            if rule.rule_type == "STATISTICAL_ANOMALY":
+                statistical_anomaly(meter, interval, rule=rule, commit=False)
+            else:
+                evaluate_alert_rule(rule, interval, intervals=all_intervals, commit=False)
     if commit:
         db.session.commit()
     return impacted
@@ -369,7 +372,10 @@ def recalculate_meter_after_correction(*, meter, reading, commit=True):
     all_intervals = consumption_intervals(meter)
     for rule in rules:
         for interval in impacted:
-            evaluate_alert_rule(rule, interval, intervals=all_intervals, commit=False)
+            if rule.rule_type == "STATISTICAL_ANOMALY":
+                statistical_anomaly(meter, interval, rule=rule, commit=False)
+            else:
+                evaluate_alert_rule(rule, interval, intervals=all_intervals, commit=False)
     if commit:
         db.session.commit()
     return impacted

migrations/versions/r3f4g5h6i7j8_meter_alert_rule_type.py

@@ -0,0 +1,23 @@
+"""Classify C3 alert rules without changing existing rule behavior."""
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "r3f4g5h6i7j8"
+down_revision = "q2f3g4h5i6j7"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.add_column(
+        "meter_alert_rules",
+        sa.Column(
+            "rule_type", sa.String(length=30), nullable=False,
+            server_default="MANUAL_THRESHOLD",
+        ),
+    )
+
+
+def downgrade():
+    op.drop_column("meter_alert_rules", "rule_type")

tests/integration/test_meter_checkpoint_c3.py

@@ -4,7 +4,7 @@ from time import perf_counter
 import pytest
 
 from app_new import db
-from app_new.core.models.college import Building
+from app_new.core.models.college import Building, HousingUnit
 from app_new.core.models.planning import (
     Meter, MeterReading, MeterReadingCorrection, MeterAlert, MeterAlertEvidence,
     MeterAlertRule, MeterHeatingRegime, GasConversion, MeterTariff,
@@ -227,3 +227,113 @@ def test_c3_alert_close_route_and_dashboard_summaries(app, admin_user, authentic
     dashboard = authenticated_client.get("/planning/meter-monitoring")
     assert dashboard.status_code == 200
     assert b"Eau" in dashboard.data and b"m" in dashboard.data
+
+
+def _stat_rule(meter, **kwargs):
+    values = dict(
+        meter=meter, rule_type="STATISTICAL_ANOMALY", metric="consumption_per_day",
+        period="DAY", min_comparable_intervals=3, min_deviation_percent=30,
+        operator=">", threshold=0, level="WARNING",
+    )
+    values.update(kwargs)
+    return MeterAlertRule(**values)
+
+
+def _normal_history(meter, user_id, start=date(2026, 5, 1), count=4):
+    for offset in range(count):
+        _reading(meter, offset * 10, start + __import__('datetime').timedelta(days=offset), user_id)
+
+
+def test_c3_final_statistical_anomaly_runs_from_new_reading(app, admin_user):
+    with app.app_context():
+        meter = _meter("STAT_AUTO")
+        _normal_history(meter, admin_user["id"], date(2026, 5, 4))
+        rule = _stat_rule(meter)
+        db.session.add(rule)
+        db.session.commit()
+        _reading(meter, 45, date(2026, 5, 8), admin_user["id"])
+        alert = MeterAlert.query.filter_by(meter_id=meter.id, rule_id=rule.id, alert_type="STATISTICAL_ANOMALY", status="OPEN").one()
+        assert alert.reference_value == pytest.approx(10)
+        assert alert.observed_value == pytest.approx(15)
+        assert alert.deviation_percent >= 30
+        assert alert.comparable_count >= 3
+        assert len(alert.evidence) == 1
+
+
+def test_c3_final_small_variation_and_insufficient_history_do_not_alert(app, admin_user):
+    with app.app_context():
+        small = _meter("STAT_SMALL")
+        _normal_history(small, admin_user["id"], date(2026, 6, 1))
+        small_rule = _stat_rule(small)
+        db.session.add(small_rule)
+        db.session.commit()
+        _reading(small, 41, date(2026, 6, 5), admin_user["id"])
+        assert MeterAlert.query.filter_by(meter_id=small.id, alert_type="STATISTICAL_ANOMALY").count() == 0
+
+        insufficient = _meter("STAT_INSUFFICIENT")
+        _normal_history(insufficient, admin_user["id"], date(2026, 7, 1), count=3)
+        rule = _stat_rule(insufficient)
+        db.session.add(rule)
+        db.session.commit()
+        _reading(insufficient, 35, date(2026, 7, 4), admin_user["id"])
+        assert MeterAlert.query.filter_by(meter_id=insufficient.id, alert_type="STATISTICAL_ANOMALY").count() == 0
+
+
+def test_c3_final_custom_statistical_parameters(app, admin_user):
+    with app.app_context():
+        building = Building(name="TEST_UI_C3_STAT_CUSTOM_BUILDING")
+        db.session.add(building)
+        db.session.flush()
+        housing = HousingUnit(name="TEST_UI_C3_STAT_CUSTOM_HOUSING", building_id=building.id)
+        db.session.add(housing)
+        db.session.flush()
+        meter = create_meter(name="TEST_UI_C3_STAT_CUSTOM", meter_type="eau", unit="m³", housing_unit_id=housing.id)
+        _normal_history(meter, admin_user["id"], date(2026, 8, 3), count=5)
+        rule = _stat_rule(meter, min_comparable_intervals=5, min_deviation_percent=40)
+        db.session.add(rule)
+        db.session.commit()
+        _reading(meter, 50, date(2026, 8, 8), admin_user["id"])
+        assert MeterAlert.query.filter_by(meter_id=meter.id, alert_type="STATISTICAL_ANOMALY").count() == 0
+        _reading(meter, 65, date(2026, 8, 9), admin_user["id"])
+        alert = MeterAlert.query.filter_by(meter_id=meter.id, alert_type="STATISTICAL_ANOMALY", status="OPEN").one()
+        assert alert.comparable_count >= 5 and alert.deviation_percent >= 40
+
+
+def test_c3_final_statistical_correction_normal_and_still_anomalous(app, admin_user):
+    with app.app_context():
+        meter = _meter("STAT_CORRECTION")
+        _normal_history(meter, admin_user["id"], date(2026, 9, 7))
+        rule = _stat_rule(meter)
+        db.session.add(rule)
+        db.session.commit()
+        high = _reading(meter, 45, date(2026, 9, 11), admin_user["id"])
+        original = MeterAlert.query.filter_by(meter_id=meter.id, rule_id=rule.id, status="OPEN").one()
+        correct_meter_reading(reading=high, new_value=41, user_id=admin_user["id"], reason="TEST_UI_C3_FINAL normal")
+        db.session.expire_all()
+        assert db.session.get(MeterAlert, original.id).status == "CLOSED"
+
+        high_again = db.session.get(MeterReading, high.id)
+        correct_meter_reading(reading=high_again, new_value=50, user_id=admin_user["id"], reason="TEST_UI_C3_FINAL still anomalous")
+        open_alerts = MeterAlert.query.filter_by(meter_id=meter.id, rule_id=rule.id, alert_type="STATISTICAL_ANOMALY", status="OPEN").all()
+        assert len(open_alerts) == 1 and len(open_alerts[0].evidence) == 1
+
+
+def test_c3_final_reading_recalculation_performance(app, admin_user):
+    with app.app_context():
+        meter = _meter("STAT_PERF")
+        _normal_history(meter, admin_user["id"], date(2026, 10, 1), count=30)
+        manual = MeterAlertRule(meter=meter, metric="consumption_per_day", period="DAY", operator=">", threshold=999)
+        statistical = _stat_rule(meter)
+        db.session.add_all([manual, statistical])
+        db.session.commit()
+        from sqlalchemy import event
+        queries = {"count": 0}
+        def count_query(*args):
+            queries["count"] += 1
+        event.listen(db.engine, "before_cursor_execute", count_query)
+        started = perf_counter()
+        _reading(meter, 315, date(2026, 11, 1), admin_user["id"])
+        elapsed = perf_counter() - started
+        event.remove(db.engine, "before_cursor_execute", count_query)
+        print(f"C3_FINAL_PERF reading_seconds={elapsed:.4f} reading_queries={queries['count']} intervals={len(consumption_intervals(meter))}")
+        assert elapsed >= 0 and queries["count"] > 0