fix(meters): automate statistical anomaly detection

This commit is contained in:
root 2026-08-24 12:51:01 +00:00
parent eca0b88195
commit cac9dbc632
4 changed files with 144 additions and 4 deletions

View file

@ -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=">")

View file

@ -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

View file

@ -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")

View file

@ -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="", 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