app.alerts

Alerts: turns red forecasts into dispatcher alerts and checks them against what actually happened.

• Raise when a forecast’s risk (the model’s probability of arriving > 2 min late) reaches risk_threshold and its target stop is 10–15 minutes ahead — so an alert is never about a stop the vehicle already reached (criterion 2: no after-the-fact alerts).

• One active alert per vehicle: later red forecasts for it are counted, not re-raised.

• Verify once the fleet detects the arrival at the target stop: verified with the actual delay, scored as a hit (actually > 2 min late) or a false alarm. No arrival detected by the end of the detection window → resolved with the reason. Precision and forecast error of alerts are therefore measured live.

Classes

Alert(alert_id, tr_id, prediction, ...[, ...])
AlertEngine(fleet, clock, *, risk_threshold, ...)
AlertStats([raised, duplicates_suppressed, ...])

class app.alerts.Alert(alert_id: 'str', tr_id: 'int', prediction: 'Prediction', segment_from: 'str | None', segment_to: 'str | None', created_at: 'float', status: 'Status' = 'active', delay_fact_s: 'float | None' = None, closed_at: 'float | None' = None, resolution: 'str | None' = None)

Базовые классы: object

__init__(alert_id: str, tr_id: int, prediction: Prediction, segment_from: str | None, segment_to: str | None, created_at: float, status: Literal['active', 'verified', 'resolved'] = 'active', delay_fact_s: float | None = None, closed_at: float | None = None, resolution: str | None = None) → None

alert_id: str

closed_at: float | None

created_at: float

delay_fact_s: float | None

property hit: bool | None

prediction: Prediction

resolution: str | None

segment_from: str | None

segment_to: str | None

status: Literal['active', 'verified', 'resolved']

tr_id: int

class app.alerts.AlertEngine(fleet: Fleet, clock: DatasetClock, *, risk_threshold: float = 0.7, verify_grace_s: float = 840, id_prefix: str = 'a', wall: Callable[[], float]=<built-in function time>)

Базовые классы: object

__init__(fleet: Fleet, clock: DatasetClock, *, risk_threshold: float = 0.7, verify_grace_s: float = 840, id_prefix: str = 'a', wall: Callable[[], float]=<built-in function time>) → None

active: dict[int, Alert]

check(T: datetime | None = None) → None

closed: deque[Alert]

on_event(fn: Callable[[Literal['alert.new', 'alert.verified', 'alert.resolved'], Alert], None]) → None

on_prediction(p: Prediction) → None

async run(tick_s: float = 5.0) → None

snapshot() → dict

class app.alerts.AlertStats(raised: 'int' = 0, duplicates_suppressed: 'int' = 0, verified: 'int' = 0, hits: 'int' = 0, resolved_unverified: 'int' = 0, abs_error_sum_s: 'float' = 0.0)

Базовые классы: object

__init__(raised: int = 0, duplicates_suppressed: int = 0, verified: int = 0, hits: int = 0, resolved_unverified: int = 0, abs_error_sum_s: float = 0.0) → None

abs_error_sum_s: float

duplicates_suppressed: int

hits: int

raised: int

resolved_unverified: int

verified: int