diff --git a/.gitignore b/.gitignore index 80857fe..7b53cf2 100644 --- a/.gitignore +++ b/.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 diff --git a/tests/backtest/__init__.py b/tests/backtest/__init__.py new file mode 100644 index 0000000..a3c05de --- /dev/null +++ b/tests/backtest/__init__.py @@ -0,0 +1 @@ +"""Оффлайн-бэктест модели прогноза тревог (HACK-174).""" diff --git a/tests/backtest/backtest.md b/tests/backtest/backtest.md new file mode 100644 index 0000000..fa8153d --- /dev/null +++ b/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` diff --git a/tests/backtest/data.py b/tests/backtest/data.py new file mode 100644 index 0000000..bf1c519 --- /dev/null +++ b/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()} diff --git a/tests/backtest/episodes.py b/tests/backtest/episodes.py new file mode 100644 index 0000000..9885ee1 --- /dev/null +++ b/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) + ) diff --git a/tests/backtest/metrics.py b/tests/backtest/metrics.py new file mode 100644 index 0000000..9a76d6e --- /dev/null +++ b/tests/backtest/metrics.py @@ -0,0 +1,141 @@ +"""Precision/recall, алерты на смену диспетчера, калибровка, бутстрэп CI.""" +from __future__ import annotations + +import numpy as np +import polars as pl + +HIGH_THRESHOLD = 0.7 +MEDIUM_THRESHOLD = 0.4 +N_BOOTSTRAP = 2000 + + +def _binary_metrics( + df: pl.DataFrame, predicted: pl.Series, dispatcher_scale: dict[int, float] +) -> dict[str, float]: + df = df.with_columns(predicted.alias("__pred")) + tp = df.filter(pl.col("__pred") & pl.col("outcome_24h")).height + fp = df.filter(pl.col("__pred") & ~pl.col("outcome_24h")).height + fn = df.filter(~pl.col("__pred") & pl.col("outcome_24h")).height + + by_dispatcher_shift = ( + df.filter(pl.col("dispatcher_id").is_not_null()) + .group_by(["dispatcher_id", "t0"]) + .agg(pl.col("__pred").sum().alias("alerts")) + .with_columns( + pl.col("dispatcher_id") + .replace_strict(dispatcher_scale, default=1.0, return_dtype=pl.Float64) + .alias("scale") + ) + ) + hours = df.filter(pl.col("__pred") & pl.col("outcome_24h"))["hours_to_event"].drop_nulls() + + return { + "precision": tp / (tp + fp) if (tp + fp) else float("nan"), + "recall": tp / (tp + fn) if (tp + fn) else float("nan"), + "tp": tp, + "fp": fp, + "fn": fn, + "n": df.height, + # На тестовых каналах (~5% парка) и пересчитанное на весь парк диспетчера. + "alerts_per_shift_sample": by_dispatcher_shift["alerts"].mean(), + "alerts_per_shift_fleet": (by_dispatcher_shift["alerts"] * by_dispatcher_shift["scale"]).mean(), + "median_hours_to_event": hours.median() if hours.len() else float("nan"), + } + + +def _method_predicates(df: pl.DataFrame) -> dict[str, pl.Series]: + return { + "model_fresh_t0_high_0.7": df["probability"] >= HIGH_THRESHOLD, + "model_fresh_t0_medium_0.4": df["probability"] >= MEDIUM_THRESHOLD, + "model_prodstyle_high_0.7": df["prodstyle_valid"] & (df["prodstyle_probability"] >= HIGH_THRESHOLD), + "model_prodstyle_medium_0.4": df["prodstyle_valid"] & (df["prodstyle_probability"] >= MEDIUM_THRESHOLD), + "baseline_alarm_yesterday": df["baseline_alarm_yesterday"], + "baseline_10plus_yesterday": df["baseline_10plus_yesterday"], + } + + +def build_comparison_table(df: pl.DataFrame, dispatcher_scale: dict[int, float]) -> pl.DataFrame: + slices = {"all_channels": df, "early_warning_6h": df.filter(pl.col("early_warning_ok"))} + return pl.DataFrame( + [ + {"slice": slice_name, "method": method, **_binary_metrics(sdf, predicate, dispatcher_scale)} + for slice_name, sdf in slices.items() + for method, predicate in _method_predicates(sdf).items() + ] + ) + + +def calibration_table(df: pl.DataFrame, bins: int = 10) -> pl.DataFrame: + breaks = [i / bins for i in range(1, bins)] + labels = [f"[{i / bins:.1f}, {(i + 1) / bins:.1f})" for i in range(bins)] + return ( + df.with_columns( + pl.col("probability").cut(breaks, labels=labels, left_closed=True).alias("probability_bin") + ) + .group_by("probability_bin", maintain_order=True) + .agg( + pl.len().alias("n"), + pl.col("outcome_24h").mean().alias("actual_rate"), + pl.col("probability").mean().alias("mean_probability"), + ) + .sort("probability_bin") + ) + + +def _per_channel_counts(df: pl.DataFrame, predicted: pl.Series) -> list[np.ndarray]: + counts = ( + df.with_columns(predicted.alias("__pred")) + .group_by("channel_id") + .agg( + (pl.col("__pred") & pl.col("outcome_24h")).sum().alias("tp"), + (pl.col("__pred") & ~pl.col("outcome_24h")).sum().alias("fp"), + (~pl.col("__pred") & pl.col("outcome_24h")).sum().alias("fn"), + ) + .sort("channel_id") + ) + return [counts[c].to_numpy() for c in ("tp", "fp", "fn")] + + +def bootstrap_diff_ci( + df: pl.DataFrame, predicted_a: pl.Series, predicted_b: pl.Series +) -> dict[str, float]: + """95% CI разниц precision и recall a - b. Бутстрэп по каналам, не по строкам: + смены одного канала коррелированы, построчный занизил бы дисперсию. + """ + tp_a, fp_a, fn_a = _per_channel_counts(df, predicted_a) + tp_b, fp_b, fn_b = _per_channel_counts(df, predicted_b) + + rng = np.random.default_rng(0) + n = tp_a.shape[0] + diffs = {"precision_diff": np.empty(N_BOOTSTRAP), "recall_diff": np.empty(N_BOOTSTRAP)} + + for i in range(N_BOOTSTRAP): + idx = rng.integers(0, n, n) + a_tp, a_fp, a_fn = tp_a[idx].sum(), fp_a[idx].sum(), fn_a[idx].sum() + b_tp, b_fp, b_fn = tp_b[idx].sum(), fp_b[idx].sum(), fn_b[idx].sum() + diffs["precision_diff"][i] = (a_tp / (a_tp + a_fp) if (a_tp + a_fp) else 0.0) - ( + b_tp / (b_tp + b_fp) if (b_tp + b_fp) else 0.0 + ) + diffs["recall_diff"][i] = (a_tp / (a_tp + a_fn) if (a_tp + a_fn) else 0.0) - ( + b_tp / (b_tp + b_fn) if (b_tp + b_fn) else 0.0 + ) + + result = {} + for name, values in diffs.items(): + result[name] = float(np.mean(values)) + result[f"{name}_ci_low"] = float(np.percentile(values, 2.5)) + result[f"{name}_ci_high"] = float(np.percentile(values, 97.5)) + return result + + +def build_bootstrap_table(df: pl.DataFrame) -> pl.DataFrame: + """CI каждого варианта модели против бейзлайна «тревога вчера».""" + predicates = _method_predicates(df) + baseline = predicates["baseline_alarm_yesterday"] + return pl.DataFrame( + [ + {"method": method, "vs": "baseline_alarm_yesterday", **bootstrap_diff_ci(df, predicate, baseline)} + for method, predicate in predicates.items() + if method.startswith("model_") + ] + ) diff --git a/tests/backtest/predict.py b/tests/backtest/predict.py new file mode 100644 index 0000000..e3021b0 --- /dev/null +++ b/tests/backtest/predict.py @@ -0,0 +1,97 @@ +"""Реплей событий через app.PredictionService в двух вариантах: прогноз ровно +на смене t0 и «как на стенде» — последний прогноз по событию до t0. +""" +from __future__ import annotations + +import bisect +from datetime import datetime + +import polars as pl + +from app import PredictionService +from tests.backtest.data import ID_COL + +# Стенд пишет прогноз на каждом событии с horizon_until = ts + 30ч и удаляет +# просроченные (inference/runner.py, predictions.delete_expired). +PROD_HORIZON_HOURS = 30.0 + + +def _channel_events(events_df: pl.DataFrame) -> dict[int, list[tuple[datetime, str]]]: + return { + int(channel_id): list(zip(group["ts"].to_list(), group["значение_датчика"].to_list())) + for (channel_id,), group in events_df.partition_by( + ID_COL, as_dict=True, maintain_order=True + ).items() + } + + +def predict_shifts( + service: PredictionService, + events_df: pl.DataFrame, + channel_ids: list[int], + shift_times: list[datetime], + horizon_hours: float, +) -> list[tuple[int, datetime, float, bool]]: + """(канал, t0, вероятность, history_complete) по событиям с ts < t0.""" + channel_events = _channel_events(events_df) + rows, keys = [], [] + + for channel_id in channel_ids: + events = channel_events.get(channel_id, []) + ei = 0 + for t0 in shift_times: + while ei < len(events) and events[ei][0] < t0: + service.store.update(channel_id, *events[ei]) + ei += 1 + rows.append(service._build_feature_row(channel_id, t0)) + keys.append((channel_id, t0, service.store.history_complete(channel_id, t0))) + + raw = service.model.predict(rows) + return [ + (channel_id, t0, service._probability_from_raw(float(r), horizon_hours), complete) + for (channel_id, t0, complete), r in zip(keys, raw) + ] + + +def predict_last_event( + service: PredictionService, + events_df: pl.DataFrame, + channel_ids: list[int], + shift_times: list[datetime], +) -> dict[tuple[int, datetime], tuple[float | None, float | None]]: + """{(канал, t0): (вероятность, возраст в часах)} последнего прогноза по + событию до t0; (None, None), если событий до t0 не было. + """ + channel_events = _channel_events(events_df) + result: dict[tuple[int, datetime], tuple[float | None, float | None]] = {} + rows, row_meta = [], [] + + for channel_id in channel_ids: + events = channel_events.get(channel_id, []) + ts = [event_ts for event_ts, _ in events] + + # Смены по индексу последнего события со ts < t0. + due: dict[int, list[datetime]] = {} + for t0 in shift_times: + due.setdefault(bisect.bisect_left(ts, t0) - 1, []).append(t0) + for t0 in due.get(-1, []): + result[(channel_id, t0)] = (None, None) + + for idx, (event_ts, value) in enumerate(events): + service.store.update(channel_id, event_ts, value) + t0s = due.get(idx) + if not t0s: + continue + # Следующее событие с тем же ts: прогноз стенд даст после него. + if idx + 1 < len(ts) and ts[idx + 1] == event_ts: + due.setdefault(idx + 1, []).extend(t0s) + continue + rows.append(service._build_feature_row(channel_id, event_ts)) + row_meta.append((channel_id, event_ts, t0s)) + + for (channel_id, event_ts, t0s), raw in zip(row_meta, service.model.predict(rows)): + probability = service._probability_from_raw(float(raw), PROD_HORIZON_HOURS) + for t0 in t0s: + result[(channel_id, t0)] = (probability, (t0 - event_ts).total_seconds() / 3600) + + return result diff --git a/tests/backtest/results/bootstrap_ci.csv b/tests/backtest/results/bootstrap_ci.csv new file mode 100644 index 0000000..f1867a0 --- /dev/null +++ b/tests/backtest/results/bootstrap_ci.csv @@ -0,0 +1,5 @@ +method,vs,precision_diff,precision_diff_ci_low,precision_diff_ci_high,recall_diff,recall_diff_ci_low,recall_diff_ci_high +model_fresh_t0_high_0.7,baseline_alarm_yesterday,-0.03138913755639792,-0.07050541664833714,0.012371952598976558,-0.05449899321474636,-0.08312325741661279,-0.028134374772072177 +model_fresh_t0_medium_0.4,baseline_alarm_yesterday,-0.05960757964951178,-0.08694890344125657,-0.035021413272330215,0.04908701353424739,0.0006112734910093891,0.10552878578603714 +model_prodstyle_high_0.7,baseline_alarm_yesterday,0.05103185500595758,0.02823745039543638,0.07505216804979566,-0.01956546001744728,-0.032832704205969995,-0.006099442426007712 +model_prodstyle_medium_0.4,baseline_alarm_yesterday,0.01974370976459373,0.00019372933244877623,0.04067700325721721,0.029015808611489988,0.006503228120516496,0.054738962341369164 diff --git a/tests/backtest/results/calibration.csv b/tests/backtest/results/calibration.csv new file mode 100644 index 0000000..1c5e367 --- /dev/null +++ b/tests/backtest/results/calibration.csv @@ -0,0 +1,11 @@ +probability_bin,n,actual_rate,mean_probability +"[0.0, 0.1)",86719,0.012546270136878885,0.013213212696130827 +"[0.1, 0.2)",4308,0.036211699164345405,0.13938121656866698 +"[0.2, 0.3)",1974,0.05977710233029382,0.25046106315726696 +"[0.3, 0.4)",1423,0.054813773717498245,0.34584533768545883 +"[0.4, 0.5)",1363,0.03741746148202495,0.4513625936184026 +"[0.5, 0.6)",1053,0.0731244064577398,0.5495437673289508 +"[0.6, 0.7)",897,0.059085841694537344,0.6466572509661418 +"[0.7, 0.8)",534,0.09176029962546817,0.7455367933795556 +"[0.8, 0.9)",284,0.07394366197183098,0.8464538803053107 +"[0.9, 1.0)",445,0.09887640449438202,0.9764363440869955 diff --git a/tests/backtest/results/comparison_table.csv b/tests/backtest/results/comparison_table.csv new file mode 100644 index 0000000..d8ce08f --- /dev/null +++ b/tests/backtest/results/comparison_table.csv @@ -0,0 +1,13 @@ +slice,method,precision,recall,tp,fp,fn,n,alerts_per_shift_sample,alerts_per_shift_fleet,median_hours_to_event +all_channels,model_fresh_t0_high_0.7,0.090261,0.065706,114,1149,1621,99000,0.438542,8.711691,5.316250 +all_channels,model_fresh_t0_medium_0.4,0.064467,0.170029,295,4281,1440,99000,1.588889,31.621273,7.250556 +all_channels,model_prodstyle_high_0.7,0.175879,0.100865,175,820,1560,99000,0.345486,6.923810,6.276667 +all_channels,model_prodstyle_medium_0.4,0.144290,0.149280,259,1536,1476,99000,0.623264,12.072714,8.818333 +all_channels,baseline_alarm_yesterday,0.125075,0.120461,209,1462,1526,99000,0.580208,11.778137,7.755278 +all_channels,baseline_10plus_yesterday,0.226667,0.009798,17,58,1718,99000,0.026042,0.603211,0.385000 +early_warning_6h,model_fresh_t0_high_0.7,0.074807,0.053016,87,1076,1554,97481,0.403819,8.003108,7.574444 +early_warning_6h,model_fresh_t0_medium_0.4,0.056453,0.151127,248,4145,1393,97481,1.525347,30.362544,7.668750 +early_warning_6h,model_prodstyle_high_0.7,0.158389,0.071907,118,627,1523,97481,0.258681,5.180913,5.996944 +early_warning_6h,model_prodstyle_medium_0.4,0.129161,0.118221,194,1308,1447,97481,0.521528,10.053264,7.805972 +early_warning_6h,baseline_alarm_yesterday,0.102963,0.084704,139,1211,1502,97481,0.468750,9.510626,7.099722 +early_warning_6h,baseline_10plus_yesterday,0.069767,0.001828,3,40,1638,97481,0.014931,0.350726,12.250278 diff --git a/tests/backtest/results/episode_value_breakdown.csv b/tests/backtest/results/episode_value_breakdown.csv new file mode 100644 index 0000000..477a138 --- /dev/null +++ b/tests/backtest/results/episode_value_breakdown.csv @@ -0,0 +1,13 @@ +значение_датчика,n,share +Неисправен,5543,0.8274369308852068 +Обнаружен дым,366,0.05463502015226153 +Не замкнут,262,0.03911031497238394 +Отключено устройство,229,0.0341842065979997 +Обнаружено движение,113,0.016868189281982384 +Обнаружен газ,57,0.00850873264666368 +Рычаг сдернут,56,0.008359456635318705 +Много неисправных устройств,26,0.0038811762949693984 +Питание от батарей,23,0.0034333482609344677 +Температура ниже 3ºC,13,0.0019405881474846992 +Работают все насосы в АНС,9,0.0013434841021047917 +Не определено,2,0.0002985520226899537 diff --git a/tests/backtest/run.py b/tests/backtest/run.py new file mode 100644 index 0000000..564b13e --- /dev/null +++ b/tests/backtest/run.py @@ -0,0 +1,129 @@ +"""Оффлайн-бэктест HACK-174, см. tests/backtest/backtest.md. + + HACK174_DATASET_DIR=/path/to/Sources/dataset python -m tests.backtest.run +""" +from __future__ import annotations + +import argparse +import os +from datetime import date, datetime, timedelta +from pathlib import Path + +import polars as pl + +from tests.backtest import data, episodes, metrics, predict +from tests.backtest.service import build_offline_service + +REPO_ROOT = Path(__file__).resolve().parents[2] +HORIZON = timedelta(hours=24) +SHIFT_HOURS = (8, 20) +WARMUP_PAD_DAYS = 3 + + +def _build_rows( + predictions: list[tuple[int, datetime, float, bool]], + prodstyle: dict[tuple[int, datetime], tuple[float | None, float | None]], + timelines: dict[int, episodes.ChannelTimeline], + dispatchers: dict[int, int | None], +) -> pl.DataFrame: + rows = [] + for channel_id, t0, probability, history_complete in predictions: + timeline = timelines[channel_id] + yesterday_alarms = episodes.alarms_in_window(timeline, t0 - timedelta(hours=24), t0) + # Спокойный канал: сейчас в норме и без тревог последние 6 ч. + early_warning_ok = ( + episodes.state_at(timeline, t0) is not True + and episodes.alarms_in_window(timeline, t0 - timedelta(hours=6), t0) == 0 + ) + episode_start = episodes.first_episode_after(timeline, t0, HORIZON) + prod_probability, prod_age = prodstyle[(channel_id, t0)] + + rows.append( + { + "channel_id": channel_id, + "t0": t0, + "probability": probability, + "history_complete": history_complete, + "prodstyle_probability": prod_probability, + "prodstyle_age_hours": prod_age, + "prodstyle_valid": prod_age is not None and prod_age < predict.PROD_HORIZON_HOURS, + "baseline_alarm_yesterday": yesterday_alarms > 0, + "baseline_10plus_yesterday": yesterday_alarms >= 10, + "early_warning_ok": early_warning_ok, + "outcome_24h": episode_start is not None, + "hours_to_event": ( + (episode_start - t0).total_seconds() / 3600 if episode_start is not None else None + ), + "dispatcher_id": dispatchers.get(channel_id), + } + ) + return pl.DataFrame(rows, infer_schema_length=None) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--dataset-dir", type=Path, default=os.environ.get("HACK174_DATASET_DIR")) + parser.add_argument( + "--weather-csv", + type=Path, + default=REPO_ROOT / "retraining/data/reference/open-meteo-55.75N37.63E140m.csv", + ) + parser.add_argument("--output-dir", type=Path, default=Path(__file__).resolve().parent / "results") + parser.add_argument("--start", type=date.fromisoformat, default=date(2026, 4, 1)) + parser.add_argument("--end", type=date.fromisoformat, default=date(2026, 6, 30)) + args = parser.parse_args() + if args.dataset_dir is None: + parser.error("--dataset-dir or HACK174_DATASET_DIR is required") + out = args.output_dir + out.mkdir(parents=True, exist_ok=True) + + test_ids = data.test_channel_ids(args.dataset_dir) + pl.DataFrame({data.ID_COL: test_ids}).write_csv(out / "test_channels.csv") + + load_start = args.start - timedelta(days=WARMUP_PAD_DAYS) + events_df = data.load_events( + args.dataset_dir, test_ids, load_start, args.end + timedelta(days=1, hours=1) + ) + seed_states = data.last_state_before(args.dataset_dir, test_ids, load_start) + service = build_offline_service(REPO_ROOT, args.weather_csv) + + have_events = set(events_df[data.ID_COL].unique().to_list()) + usable_ids = [cid for cid in test_ids if cid in have_events and cid in service.metadata] + + # Смены, у которых окно t0 + 24ч целиком внутри периода: последняя — 29.06 20:00. + period_start = datetime.combine(args.start, datetime.min.time()) + period_end = datetime.combine(args.end, datetime.min.time()) + timedelta(days=1) + shift_times = [ + period_start + timedelta(days=d, hours=h) + for d in range((args.end - args.start).days + 1) + for h in SHIFT_HOURS + if period_start + timedelta(days=d, hours=h) + HORIZON <= period_end + ] + + predictions = predict.predict_shifts( + service, events_df, usable_ids, shift_times, HORIZON.total_seconds() / 3600 + ) + # Store уже продвинут до последней смены: без сброса события второго + # прохода app.OnlineFeatureStore.update счёл бы out-of-order. + service.store.states.clear() + prodstyle = predict.predict_last_event(service, events_df, usable_ids, shift_times) + + timelines = episodes.build_timelines(events_df, seed_states) + dispatchers = data.channel_dispatchers( + service.config.sensors_metadata_path, service.config.objects_metadata_path + ) + df = _build_rows(predictions, prodstyle, timelines, dispatchers) + df.write_csv(out / "predictions.csv") + + episodes.episode_value_breakdown(timelines, period_start, period_end).write_csv( + out / "episode_value_breakdown.csv" + ) + comparison = metrics.build_comparison_table(df, data.dispatcher_scale(dispatchers, usable_ids)) + comparison.write_csv(out / "comparison_table.csv", float_precision=6) + metrics.calibration_table(df).write_csv(out / "calibration.csv") + metrics.build_bootstrap_table(df).write_csv(out / "bootstrap_ci.csv") + print(comparison) + + +if __name__ == "__main__": + main() diff --git a/tests/backtest/service.py b/tests/backtest/service.py new file mode 100644 index 0000000..0740170 --- /dev/null +++ b/tests/backtest/service.py @@ -0,0 +1,44 @@ +"""Боевой app.PredictionService offline: модель и справочники стенда, +CSV-погода Open-Meteo вместо weather-service, без Postgres и pickle-state. +""" +from __future__ import annotations + +import tempfile +from datetime import datetime +from pathlib import Path + +import polars as pl + +from app import PredictionService, ServiceConfig + + +class CsvWeatherClient: + """app.WeatherSource из CSV; UTC +3ч, как EXPECTED_WEATHER в ноутбуке 1.""" + + def __init__(self, csv_path: Path) -> None: + df = pl.read_csv(csv_path).with_columns( + (pl.col("time").str.to_datetime() + pl.duration(hours=3)).alias("time") + ) + self._by_hour = {row["time"]: row for row in df.iter_rows(named=True)} + self._missing = dict.fromkeys(df.columns) + + def get(self, target_hour: datetime) -> dict: + hour = target_hour.replace(minute=0, second=0, microsecond=0) + return self._by_hour.get(hour, self._missing) + + def close(self) -> None: + pass + + +def build_offline_service(repo_root: Path, weather_csv: Path) -> PredictionService: + # Свежий каталог: чужой metadata-кеш и rolling-state не подмешаются. + runtime_dir = Path(tempfile.mkdtemp(prefix="hack174-backtest-")) + config = ServiceConfig( + model_path=repo_root / "data" / "best_model.cbm", + sensors_metadata_path=repo_root / "data" / "справочник_каналов_датчиков.parquet", + objects_metadata_path=repo_root / "data" / "справочник_объектов_диспетчер.parquet", + metadata_cache_path=runtime_dir / "metadata_cache.pkl", + state_path=runtime_dir / "no_state.pkl", + log_path=runtime_dir / "service.log", + ) + return PredictionService(config=config, weather_client=CsvWeatherClient(weather_csv))