From cac9dbc6325e092da1b7ee708991c50518d68ec3 Mon Sep 17 00:00:00 2001 From: root Date: Mon, 24 Aug 2026 12:51:01 +0000 Subject: [PATCH] fix(meters): automate statistical anomaly detection --- app_new/core/models/planning.py | 1 + app_new/core/services/meter_analytics.py | 12 +- .../r3f4g5h6i7j8_meter_alert_rule_type.py | 23 ++++ tests/integration/test_meter_checkpoint_c3.py | 112 +++++++++++++++++- 4 files changed, 144 insertions(+), 4 deletions(-) create mode 100644 migrations/versions/r3f4g5h6i7j8_meter_alert_rule_type.py diff --git a/app_new/core/models/planning.py b/app_new/core/models/planning.py index 573198c..94d2b8d 100644 --- a/app_new/core/models/planning.py +++ b/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=">") diff --git a/app_new/core/services/meter_analytics.py b/app_new/core/services/meter_analytics.py index ef75ba5..e50716e 100644 --- a/app_new/core/services/meter_analytics.py +++ b/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 diff --git a/migrations/versions/r3f4g5h6i7j8_meter_alert_rule_type.py b/migrations/versions/r3f4g5h6i7j8_meter_alert_rule_type.py new file mode 100644 index 0000000..afc5bf6 --- /dev/null +++ b/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") diff --git a/tests/integration/test_meter_checkpoint_c3.py b/tests/integration/test_meter_checkpoint_c3.py index a3f6392..95b1e39 100644 --- a/tests/integration/test_meter_checkpoint_c3.py +++ b/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