feat(backtest): ML backtest

.gitignore

@@ -3,3 +3,5 @@ data/ext-journal-2026.parquet
 __pycache__/
 .pytest_cache/
 runtime/
+tests/backtest/results/predictions.csv
+tests/backtest/results/test_channels.csv

tests/backtest/__init__.py

@@ -0,0 +1 @@
+"""Оффлайн-бэктест модели прогноза тревог (HACK-174)."""

tests/backtest/backtest.md

@@ -0,0 +1,77 @@
+# Бэктест модели прогноза (HACK-174)
+
+## Зачем
+
+Эксперты спросят: «Насколько можно верить прогнозу?» До этого у нас была
+только цифра PR-AUC 0,44, по которой на этот вопрос не ответить. Здесь
+модель проверена так, как её увидит диспетчер: пришёл на смену, посмотрел
+прогноз и через сутки узнал, сбылся ли он.
+
+## Что сделали
+
+Взяли 550 каналов, которых модель не видела при обучении, и период
+апрель–июнь 2026. На начало каждой смены (08:00 и 20:00) смотрели, что в
+этот момент показывал бы стенд по каналу и случилась ли на канале новая
+тревога или неисправность в следующие 24 ч. Ту же проверку прошло простое
+правило «вчера была тревога — будет снова».
+
+## Результат
+
+| | Из 100 алертов сбывается | Из 100 событий поймали | Алертов в смену на диспетчера | Предупреждаем за |
+|---|---|---|---|---|
+| Цель из ТЗ | 70 | 50 | — | — |
+| Модель, HIGH (≥ 0,7) | **18** | **10** | ~7 | 6 ч |
+| Модель, MEDIUM (≥ 0,4) | 14 | 15 | ~12 | 9 ч |
+| Правило «тревога вчера» | 13 | 12 | ~12 | 8 ч |
+
+## Что из этого следует
+
+1. **Заявлять экспертам высокую достоверность нельзя.** 4 из 5 алертов HIGH
+   ложные, 9 из 10 событий модель пропускает. Цель ТЗ не достигнута, это
+   идёт в «Отклонения от ТЗ».
+2. **Модель почти не лучше простого правила.** HIGH точнее правила (18
+   против 13), но ловит меньше (10 против 12). Обе разницы статистически
+   значимы.
+3. **Вероятность в карточке вводит в заблуждение.** Там, где модель пишет
+   90–100%, событие случается в 10% случаев. Показывать процент как
+   вероятность нельзя, только уровень HIGH или MEDIUM.
+4. **Предсказываем в основном поломки датчиков.** 83% сбывшихся событий —
+   переход в «Неисправен», а не авария или реальная тревога. Честнее
+   называть это «риском неисправности оборудования».
+5. **Цифры завышены, и честной проверки по времени сейчас нет.** Модель
+   обучена на журналах с 01.2024 по 06.2026 (ноутбуки 1 и 5: около 720
+   train-каналов, 9,5 млн строк), фильтра по дате нет. Периода после
+   обучения в данных нет, поэтому здесь те же месяцы, только на каналах,
+   которых модель не видела. Проверка по времени возможна только после
+   переобучения с отсечкой, например до 31.03.2026 (отдельная задача,
+   около 1 дня).
+
+## Детали
+
+- **Каналы.** Тестовые каналы воспроизведены точно по сплиту обучения
+  (seed 69, 577 шт.). Число каналов и строк в каждой части совпадает с
+  ноутбуком `2_create_train_test_datasets_V2_MAIN`. Из 577 исключено 27:
+  у 26 нет событий в периоде, 1 отсутствует в справочнике. Последняя
+  смена — 29.06 20:00, чтобы окно 24 ч целиком помещалось в данные.
+- **Событие.** Засчитывается переход канала из нормы в тревогу; повторные
+  тревоги подряд новым событием не считаются. Флаг тревоги взят из поля
+  `тревожное` журнала: из `справочник_состояний.csv` однозначно его не
+  вывести, потому что у одного типа датчика несколько наборов состояний.
+  **Отклонение от описания тикета, нужна виза автора.**
+- **Прогноз «как на стенде».** Берётся последний прогноз по событию канала
+  до начала смены с горизонтом 30 ч, если он ещё не истёк. Так устроен
+  `inference/runner.py`. Если же считать прогноз заново ровно в начале
+  смены, модель хуже правила: 9 из 100 алертов сбывается, ловит 7 из 100
+  событий.
+- **Диспетчер** — объект уровня 2 в справочнике объектов (16 районов).
+  Число алертов пересчитано на весь парк: тестовые каналы — это около 5%
+  каналов.
+- **Спокойные каналы** (в норме, без тревог 6 ч до смены): HIGH — 16 из 100
+  сбывается, поймали 7 из 100, около 5 алертов в смену.
+
+Полные таблицы: `comparison_table.csv` (оба варианта модели × 2 среза),
+`bootstrap_ci.csv` (95% CI разницы с правилом), `calibration.csv`,
+`episode_value_breakdown.csv`.
+
+Запуск (около 1 мин):
+`HACK174_DATASET_DIR=<каталог с ext-journal-*.csv> python -m tests.backtest.run`

tests/backtest/data.py

@@ -0,0 +1,123 @@
+"""Журналы ext-journal-*.csv и справочники: сплит каналов, события, диспетчеры."""
+from __future__ import annotations
+
+from collections import Counter
+from datetime import date, datetime
+from pathlib import Path
+
+import polars as pl
+
+ID_COL = "ид_канала_данных"
+SEED = 69
+TRAIN_SIZE = 0.9
+VAL_SIZE = 0.05
+
+
+def _scan(dataset_dir: Path, year: int) -> pl.LazyFrame:
+    # В журналах есть вклеенные строки-заголовки (см. ноутбук 1), отбрасываем.
+    return pl.scan_csv(
+        dataset_dir / f"ext-journal-{year}.csv",
+        schema_overrides={ID_COL: pl.Utf8, "дата": pl.Utf8, "значение_датчика": pl.Utf8},
+        ignore_errors=True,
+    ).filter(pl.col(ID_COL) != ID_COL)
+
+
+def _with_ts(df: pl.DataFrame) -> pl.DataFrame:
+    return df.with_columns(
+        pl.col("дата").str.to_date(),
+        pl.col("время").str.to_time("%H:%M:%S", strict=False),
+        (pl.col("тревожное") == "t").alias("тревожное"),
+    ).with_columns(pl.col("дата").dt.combine(pl.col("время")).alias("ts"))
+
+
+def test_channel_ids(dataset_dir: Path) -> list[int]:
+    """Тестовая часть get_device_split из ноутбука 2_create_train_test_datasets_V2_MAIN:
+    ид каналов 2024-2026, sort + shuffle(seed=69), train 90% / val 5% / test остаток.
+    """
+    ids = (
+        pl.concat(
+            _scan(dataset_dir, year).select(ID_COL).unique()
+            .collect(engine="streaming").with_columns(pl.col(ID_COL).cast(pl.Int64))
+            for year in (2024, 2025, 2026)
+        )
+        .unique()
+        .sort(ID_COL)
+        .sample(fraction=1.0, shuffle=True, seed=SEED)
+    )
+    n_head = int(ids.height * TRAIN_SIZE) + int(ids.height * VAL_SIZE)
+    return ids.slice(n_head)[ID_COL].sort().to_list()
+
+
+def load_events(
+    dataset_dir: Path, channel_ids: list[int], start: date, end: date
+) -> pl.DataFrame:
+    """События channel_ids за [start, end] из журнала 2026, по каналу и времени."""
+    df = (
+        _scan(dataset_dir, 2026)
+        .select("ид_события", ID_COL, "дата", "время", "тревожное", "значение_датчика")
+        .with_columns(pl.col(ID_COL).cast(pl.Int64))
+        .filter(pl.col(ID_COL).is_in(channel_ids))
+        .filter(pl.col("дата").is_between(pl.lit(start.isoformat()), pl.lit(end.isoformat())))
+        .collect(engine="streaming")
+    )
+    return _with_ts(df).sort([ID_COL, "ts", "ид_события"])
+
+
+def last_state_before(
+    dataset_dir: Path, channel_ids: list[int], before: date
+) -> dict[int, bool]:
+    """Флаг тревожное последнего события канала до before по всем годам: без него
+    первая тревога в окне читалась бы как новый эпизод (канал 334183 в тревоге с 12.03).
+    """
+    df = pl.concat(
+        _scan(dataset_dir, year)
+        .select(ID_COL, "дата", "время", "тревожное")
+        .with_columns(pl.col(ID_COL).cast(pl.Int64))
+        .filter(pl.col(ID_COL).is_in(channel_ids))
+        .collect(engine="streaming")
+        for year in range(2019, 2027)
+    )
+    df = (
+        _with_ts(df)
+        .filter(pl.col("ts") < before)
+        .sort([ID_COL, "ts"])
+        .group_by(ID_COL, maintain_order=True)
+        .last()
+    )
+    return dict(zip(df[ID_COL].to_list(), df["тревожное"].to_list()))
+
+
+def channel_dispatchers(sensors_path: Path, objects_path: Path) -> dict[int, int | None]:
+    """Диспетчер канала — ближайший предок уровня 2 в справочнике объектов
+    (Q&A_2: уровень 2 — «Диспетчер района», 16 объектов).
+    """
+    objects = pl.read_parquet(objects_path)
+    parent = dict(zip(objects["ид_объект"], objects["родитель"]))
+    level = dict(zip(objects["ид_объект"], objects["иерархия_уровень"]))
+
+    def dispatcher_of(current: int | None) -> int | None:
+        seen = set()
+        while current is not None and current not in seen:
+            if level.get(current) == 2:
+                return int(current)
+            seen.add(current)
+            current = parent.get(current)
+        return None
+
+    sensors = pl.read_parquet(sensors_path, columns=[ID_COL, "ид_объект"])
+    return {
+        int(channel_id): dispatcher_of(int(object_id))
+        for channel_id, object_id in zip(sensors[ID_COL], sensors["ид_объект"])
+    }
+
+
+def dispatcher_scale(
+    dispatchers: dict[int, int | None], sample_ids: list[int]
+) -> dict[int, float]:
+    """Множитель «все каналы диспетчера / тестовые каналы диспетчера»."""
+    full = Counter(d for d in dispatchers.values() if d is not None)
+    sample_set = set(sample_ids)
+    sample = Counter(
+        d for cid, d in dispatchers.items() if d is not None and cid in sample_set
+    )
+    return {d: full[d] / count for d, count in sample.items()}

tests/backtest/episodes.py

@@ -0,0 +1,94 @@
+"""Эпизод — переход канала из нормы в тревогу; повторные тревоги подряд
+эпизод не продлевают и не открывают заново.
+"""
+from __future__ import annotations
+
+import bisect
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+
+import polars as pl
+
+from tests.backtest.data import ID_COL
+
+
+@dataclass(slots=True)
+class ChannelTimeline:
+    ts: list[datetime]
+    is_alarm: list[bool]
+    alarm_ts: list[datetime]
+    episode_starts: list[datetime]
+    episode_values: list[str]
+    seed_is_alarm: bool | None
+
+
+def build_timelines(
+    events_df: pl.DataFrame, seed_states: dict[int, bool]
+) -> dict[int, ChannelTimeline]:
+    timelines: dict[int, ChannelTimeline] = {}
+
+    for (channel_id,), group in events_df.partition_by(
+        ID_COL, as_dict=True, maintain_order=True
+    ).items():
+        timestamps = group["ts"].to_list()
+        is_alarm = group["тревожное"].to_list()
+        seed = seed_states.get(int(channel_id))
+
+        starts, start_values, in_alarm = [], [], bool(seed)
+        for ts, alarm, value in zip(timestamps, is_alarm, group["значение_датчика"]):
+            if alarm and not in_alarm:
+                starts.append(ts)
+                start_values.append(value)
+            in_alarm = alarm
+
+        timelines[int(channel_id)] = ChannelTimeline(
+            ts=timestamps,
+            is_alarm=is_alarm,
+            alarm_ts=[ts for ts, alarm in zip(timestamps, is_alarm) if alarm],
+            episode_starts=starts,
+            episode_values=start_values,
+            seed_is_alarm=seed,
+        )
+
+    return timelines
+
+
+def alarms_in_window(timeline: ChannelTimeline, start: datetime, end: datetime) -> int:
+    """Число тревожных событий с ts в [start, end)."""
+    return bisect.bisect_left(timeline.alarm_ts, end) - bisect.bisect_left(timeline.alarm_ts, start)
+
+
+def state_at(timeline: ChannelTimeline, t0: datetime) -> bool | None:
+    """Тревожное/норма перед t0 (последнее событие с ts < t0, иначе seed)."""
+    idx = bisect.bisect_left(timeline.ts, t0) - 1
+    return timeline.is_alarm[idx] if idx >= 0 else timeline.seed_is_alarm
+
+
+def first_episode_after(
+    timeline: ChannelTimeline, t0: datetime, horizon: timedelta
+) -> datetime | None:
+    """Первое начало эпизода в (t0, t0 + horizon], иначе None."""
+    lo = bisect.bisect_right(timeline.episode_starts, t0)
+    if lo < len(timeline.episode_starts) and timeline.episode_starts[lo] <= t0 + horizon:
+        return timeline.episode_starts[lo]
+    return None
+
+
+def episode_value_breakdown(
+    timelines: dict[int, ChannelTimeline], start: datetime, end: datetime
+) -> pl.DataFrame:
+    """Начала эпизодов в [start, end) по значению датчика: часть — «Неисправен»."""
+    df = pl.DataFrame(
+        [
+            {"значение_датчика": value}
+            for timeline in timelines.values()
+            for ts, value in zip(timeline.episode_starts, timeline.episode_values)
+            if start <= ts < end
+        ]
+    )
+    return (
+        df.group_by("значение_датчика")
+        .agg(pl.len().alias("n"))
+        .with_columns((pl.col("n") / df.height).alias("share"))
+        .sort("n", descending=True)
+    )