app.predict.predictor

The predictor: every tick, finds each vehicle’s target stop 10–15 minutes ahead and gets a delay forecast.

Target = the first planned visit in (T + 10 min, T + 15 min] — the dataset’s own definition, so the horizon criterion holds by construction. One prediction per (vehicle, target stop): as T moves on, the target rolls forward and a fresh prediction follows, roughly every 1–3 minutes per vehicle.

If the ML service is unreachable, the backend issues its own baseline forecast (delay = current deviation, risk from the same sigmoid the ML contract uses) marked source="fallback" and retries ML after retry_s — the dashboard keeps working in degraded mode (criterion 5).

Functions

fallback_response(req)	What the backend answers itself when ML is down: the baseline (delay = current deviation), same shape as ML.
level_for(risk)
risk_from_delay(d)	The ML contract's fallback risk: sigmoid((delay − 120) / 60), i.e. 0.5 at +2 min late.

Classes

Prediction(tr_id, sample_id, T, target_pos, ...)
Predictor(fleet, clock, client, *, ...)
PredictorStats([batches, predictions_ml, ...])

class app.predict.predictor.Prediction(tr_id: 'int', sample_id: 'str', T: 'datetime', target_pos: 'int', target_stop_id: 'int', target_plan: 'datetime', cur_dev_s: 'float | None', delay_pred_s: 'float', risk_score: 'float', risk_level: 'str', confidence: 'float', source: "Literal['ml', 'fallback']", made_at: 'float', response: 'dict')

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

T: datetime

__init__(tr_id: int, sample_id: str, T: datetime, target_pos: int, target_stop_id: int, target_plan: datetime, cur_dev_s: float | None, delay_pred_s: float, risk_score: float, risk_level: str, confidence: float, source: Literal['ml', 'fallback'], made_at: float, response: dict) → None

confidence: float

cur_dev_s: float | None

delay_pred_s: float

property horizon_ok: bool

property lead_s: float

made_at: float

response: dict

risk_level: str

risk_score: float

sample_id: str

source: Literal['ml', 'fallback']

target_plan: datetime

target_pos: int

target_stop_id: int

tr_id: int

class app.predict.predictor.Predictor(fleet: Fleet, clock: DatasetClock, client: MlClient, *, max_ping_age_s: float = 900.0, telemetry_span_s: float = 4500.0, batch_max: int = 64, retry_s: float = 30.0, wall: Callable[[], float]=<built-in function time>)

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

__init__(fleet: Fleet, clock: DatasetClock, client: MlClient, *, max_ping_age_s: float = 900.0, telemetry_span_s: float = 4500.0, batch_max: int = 64, retry_s: float = 30.0, wall: Callable[[], float]=<built-in function time>) → None

due(T: datetime) → Iterator[tuple[VehicleState, StopVisit]]

last_error: str | None

last_requests: dict[int, dict]

latest: dict[int, Prediction]

ml_available: bool | None

ml_batch_ms: deque[float]

on_prediction(fn: Callable[[Prediction], None]) → None

recent: deque[Prediction]

async run(tick_s: float = 5.0) → None

snapshot() → dict

async tick(T: datetime | None = None) → int

class app.predict.predictor.PredictorStats(batches: 'int' = 0, predictions_ml: 'int' = 0, predictions_fallback: 'int' = 0, ml_errors: 'int' = 0, horizon_ok: 'int' = 0)

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

__init__(batches: int = 0, predictions_ml: int = 0, predictions_fallback: int = 0, ml_errors: int = 0, horizon_ok: int = 0) → None

batches: int

horizon_ok: int

ml_errors: int

predictions_fallback: int

predictions_ml: int

app.predict.predictor.fallback_response(req: dict) → dict

What the backend answers itself when ML is down: the baseline (delay = current deviation), same shape as ML.

app.predict.predictor.level_for(risk: float) → str

app.predict.predictor.risk_from_delay(d: float) → float

The ML contract’s fallback risk: sigmoid((delay − 120) / 60), i.e. 0.5 at +2 min late.