From 72a726f99f5022dacbf07a936c01eaa0a7d8a888 Mon Sep 17 00:00:00 2001 From: UsatovPavel Date: Mon, 28 Sep 2026 20:19:24 +0300 Subject: [PATCH] feat(ml): add incident4 v4 training pipeline 4-class target (FIRE/FLOOD/GAS/INTRUSION) notebooks 1-3 with outputs stripped, segmented training, one-shot blind evaluation and README. Co-Authored-By: Claude Opus 5.5 --- .../incident4/1_prepare_dataset.ipynb | 675 ++++++++++++++++ ...2_create_train_test_datasets_V2_MAIN.ipynb | 130 ++++ .../incident4/3_feature_selection.ipynb | 736 ++++++++++++++++++ retraining/notebooks/incident4/README.md | 97 +++ .../notebooks/incident4/best_params.json | 7 + .../notebooks/incident4/check_lookahead.py | 21 + retraining/notebooks/incident4/eval_v4.py | 88 +++ retraining/notebooks/incident4/pipeline_v4.sh | 22 + retraining/notebooks/incident4/train_v4.py | 105 +++ 9 files changed, 1881 insertions(+) create mode 100644 retraining/notebooks/incident4/1_prepare_dataset.ipynb create mode 100644 retraining/notebooks/incident4/2_create_train_test_datasets_V2_MAIN.ipynb create mode 100644 retraining/notebooks/incident4/3_feature_selection.ipynb create mode 100644 retraining/notebooks/incident4/README.md create mode 100644 retraining/notebooks/incident4/best_params.json create mode 100644 retraining/notebooks/incident4/check_lookahead.py create mode 100644 retraining/notebooks/incident4/eval_v4.py create mode 100755 retraining/notebooks/incident4/pipeline_v4.sh create mode 100644 retraining/notebooks/incident4/train_v4.py diff --git a/retraining/notebooks/incident4/1_prepare_dataset.ipynb b/retraining/notebooks/incident4/1_prepare_dataset.ipynb new file mode 100644 index 0000000..17451eb --- /dev/null +++ b/retraining/notebooks/incident4/1_prepare_dataset.ipynb @@ -0,0 +1,675 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": null, + "id": "421e300e", + "metadata": {}, + "outputs": [], + "source": [ + "# https://open-meteo.com/en/docs/historical-forecast-api?utm_source=chatgpt.com&latitude=55.7558&longitude=37.6173&start_date=2021-01-01&end_date=2026-09-18&hourly=temperature_2m,relative_humidity_2m,precipitation_probability,precipitation,rain,snowfall,snow_depth,weather_code,cloud_cover,pressure_msl#hourly_weather_variables\n", + "# Сайт с погодой" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "1a3531d4", + "metadata": {}, + "outputs": [], + "source": [ + "import polars as pl\n", + "\n", + "# Делаем бустинг. \n", + "# Если поле - float делаем отдельную value колонку\n", + "# Если поле - категория из топ 100 (одна из 31), то добавляем как отдельный категориальный признак под неё\n", + "# Делаем статистики временные по float и по категориальным полям\n", + "# Делаем много статистик, потом режем дерево через shap и формируем итоговую модель\n", + "# Статистики для категорий - counts в разрезе 4-3-2-1-0.5 дней (так я это вижу)\n", + "# Текущий час нужно добавить \n", + "# Номер месяца\n", + "# День недели\n", + "# все категории из df_sensors + тег_инженерной_системы тэг первый (split('-')[0]))\n", + "\n", + "# Таргет - часов до события критического (обучаем модель \"выживаемости\")\n", + "\n", + "# То как на инференсе обновлять прогнозы все эти - большой вопрос\n", + "\n", + "pl.Config.set_tbl_rows(10)\n", + "\n", + "\n", + "# НУЖНО СДЕЛАТЬ ТАК, ЧТОБЫ \n", + "# 2025 лучше поменять на всякий случай \n", + "lf = (\n", + " pl.concat([\n", + " pl.scan_parquet(f\"data/ext-journal-{year}.parquet\")\n", + " for year in [2019, 2020, 2022, 2023, 2024, 2025, 2026] # 2021 исключён\n", + " ])\n", + " # .filter(pl.col(\"дата\") != \"дата\")\n", + " .cast({\"дата\": pl.Date})\n", + ")\n", + "df_sensors = pl.scan_parquet('data/справочник_каналов_датчиков.parquet')\n", + "df_territory = pl.scan_parquet('data/справочник_объектов_диспетчер.parquet')" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "bb737a3e", + "metadata": {}, + "outputs": [], + "source": [ + "DF_SENSORS: pl.LazyFrame = (\n", + " pl.scan_parquet('data/справочник_каналов_датчиков.parquet')\n", + " .select(\n", + " pl.col('ид_канала_данных'),\n", + " pl.col('тип_инж_системы'),\n", + " pl.col('тип_датчика'),\n", + " pl.col('ид_объект').cast(pl.Int64),\n", + " pl.col('тег_инженерной_системы').str.split('-').list.get(0)\n", + " )\n", + ")\n", + "DF_TERRITORY: pl.LazyFrame = (\n", + " pl.scan_parquet('data/справочник_объектов_диспетчер.parquet')\n", + " .drop('диспетчерское_название_объекта')\n", + ")\n", + "EXPECTED_WEATHER: pl.LazyFrame = (\n", + " pl.scan_csv('data/open-meteo-55.75N37.63E140m.csv')\n", + " .with_columns(\n", + " pl.col(\"time\")\n", + " .str.to_datetime()\n", + " + pl.duration(hours=3)\n", + " )\n", + ")\n", + "MIN_CAT_VALUE_COUNT: int = 100\n", + "\n", + "# В город уходят только 4 класса аварий, класс канала задаёт тип датчика.\n", + "# Неисправности («Неисправен», «Обесточен»...) целью не считаются.\n", + "INCIDENT_BY_SENSOR: dict[str, str] = {\n", + " \"Датчик дыма\": \"FIRE\",\n", + " \"Ручной извещатель\": \"FIRE\",\n", + " \"Тепловой датчик\": \"FIRE\",\n", + " \"Состояние УИР-Р\": \"FIRE\",\n", + " \"Состояние насоса\": \"FLOOD\",\n", + " \"Датчик затопления\": \"FLOOD\",\n", + " \"Газовый датчик\": \"GAS\",\n", + " \"Датчик движения\": \"INTRUSION\",\n", + " \"КД Дверь\": \"INTRUSION\",\n", + " \"КД Люк\": \"INTRUSION\",\n", + " \"КД АВ\": \"INTRUSION\",\n", + " \"Стекло\": \"INTRUSION\",\n", + " \"9-секционный люк\": \"INTRUSION\",\n", + "}\n", + "INCIDENT_VALUES: dict[str, list[str]] = {\n", + " \"FIRE\": [\n", + " \"Обнаружен дым\",\n", + " \"Рычаг сдернут\",\n", + " \"Рычаг сдернут влево\",\n", + " \"Рычаг сдернут вправо\",\n", + " \"Оба рычага сдернуты\",\n", + " ],\n", + " \"FLOOD\": [\n", + " \"Затоплен\",\n", + " \"Работают все насосы в АНС\",\n", + " \"Работают все насосы АНС\",\n", + " \"Включены все насосы АНС\",\n", + " ],\n", + " \"GAS\": [\"Обнаружен газ\"],\n", + " \"INTRUSION\": [\n", + " \"Обнаружено движение\",\n", + " \"Движение вверх\",\n", + " \"Движение вниз\",\n", + " \"Движение влево\",\n", + " \"Движение вправо\",\n", + " ],\n", + "}\n", + "# «Не замкнут» у этих датчиков — тревога класса самого датчика\n", + "OPEN_CONTACT_SENSORS: list[str] = [\n", + " \"Ручной извещатель\",\n", + " \"Тепловой датчик\",\n", + " \"Датчик дыма\", # по справочнику «Не замкнут» тревожный\n", + " \"Состояние УИР-Р\",\n", + " \"Датчик затопления\",\n", + " \"КД Дверь\",\n", + " \"КД Люк\",\n", + " \"КД АВ\",\n", + " \"Стекло\",\n", + " \"9-секционный люк\",\n", + "]\n", + "INCIDENT_MAPPING_PATH: str = \"data/incident_by_sensor.json\"\n", + "# Колонка тип_датчика в справочнике перепутана, наборы цельные: берём названия с однозначным флагом\n", + "_states = pl.read_csv(\"data/справочник_состояний.csv\").group_by(\"название_состояния\").agg(pl.col(\"тревожное\").unique())\n", + "STATE_ALARM: dict[str, bool] = {r[0]: bool(r[1][0]) for r in _states.iter_rows() if len(r[1]) == 1}\n", + "ROWS_PER_CHUNK: int = 1_500_000\n", + "\n", + "\n", + "def incident_alarm() -> pl.Expr:\n", + " \"\"\"Тревога одного из 4 классов, совпадающего с классом датчика.\"\"\"\n", + " sensor_incident = pl.col(\"тип_датчика\").replace_strict(\n", + " INCIDENT_BY_SENSOR, default=None\n", + " )\n", + " value = pl.col(\"значение_датчика\")\n", + "\n", + " matches = pl.lit(False)\n", + " for incident, values in INCIDENT_VALUES.items():\n", + " matches = matches | ((sensor_incident == incident) & value.is_in(values))\n", + "\n", + " open_contact = (\n", + " pl.col(\"тип_датчика\").is_in(OPEN_CONTACT_SENSORS)\n", + " & (value == \"Не замкнут\")\n", + " )\n", + " # Вариант 1: FIRE/FLOOD/GAS — флаг справочника состояний (join по названию),\n", + " # INTRUSION — флаг журнала (в справочнике любое движение тревожное)\n", + " flag_ref = value.replace_strict(STATE_ALARM, default=False, return_dtype=pl.Boolean)\n", + " flag = pl.when(sensor_incident == \"INTRUSION\").then(pl.col(\"тревожное\") == 1).otherwise(flag_ref)\n", + " return flag.fill_null(False) & (matches | open_contact).fill_null(False)\n", + "\n", + "\n", + "def preproc(\n", + " lf: pl.LazyFrame\n", + " ) -> pl.LazyFrame:\n", + " '''Добавляет новые фичи в данные'''\n", + "\n", + " # Приводим к нужному формату типы\n", + " lf = (\n", + " lf\n", + " .with_columns(\n", + " pl.col(\"ид_события\").cast(pl.Int64),\n", + " pl.col(\"ид_канала_данных\").cast(pl.Int64),\n", + " pl.col(\"дата\").cast(pl.Date),\n", + " pl.col(\"время\").str.to_time(\"%H:%M:%S\"),\n", + " pl.col(\"тревожное\").eq(\"t\").cast(pl.Int8),\n", + " pl.col(\"значение_датчика\").cast(pl.Float64, strict=False).alias(\"value_float\"),\n", + " \n", + " pl.when(\n", + " pl.col(\"значение_датчика\").cast(pl.Float64, strict=False).is_null()\n", + " )\n", + " .then(pl.col(\"значение_датчика\"))\n", + " .otherwise(None)\n", + " .alias(\"value_cat\"),\n", + " )\n", + " )\n", + "\n", + " # Делаем из значение_датчика категориальных - отдельные категории \n", + " cat_values = (\n", + " lf\n", + " .drop_nulls('value_cat')\n", + " .group_by(\"value_cat\")\n", + " .agg(pl.len().alias(\"count\"))\n", + " .filter(pl.col(\"count\") >= MIN_CAT_VALUE_COUNT)\n", + " .select('value_cat')\n", + " .collect()\n", + " ['value_cat']\n", + " .to_list()\n", + " )\n", + " cat_values = [c for c in cat_values if f\"value_cat__{c}\" in TIME_DYNAMIC_CATS]\n", + " assert len(cat_values) == len(TIME_DYNAMIC_CATS), sorted(set(TIME_DYNAMIC_CATS) - {f\"value_cat__{c}\" for c in cat_values})\n", + "\n", + " lf = (\n", + " lf\n", + " .with_columns(\n", + " pl.when(pl.col(\"value_cat\") == cat)\n", + " .then(1)\n", + " .otherwise(0)\n", + " .cast(pl.Int8)\n", + " .alias(f\"value_cat__{cat}\")\n", + " for cat in cat_values\n", + " )\n", + " .drop('value_cat')\n", + " )\n", + "\n", + " # категории из DF_SENSORS и DF_TERRITORY\n", + " lf = (\n", + " lf\n", + " .join(\n", + " DF_SENSORS\n", + " .join(\n", + " DF_TERRITORY,\n", + " on='ид_объект',\n", + " how='left'\n", + " )\n", + " .drop('ид_объект')\n", + " .cast({'ид_канала_данных': pl.Int64}), \n", + " on='ид_канала_данных', \n", + " how='left'\n", + " )\n", + " )\n", + "\n", + " # Цель — только тревоги 4 классов; исходное значение больше не нужно\n", + " lf = (\n", + " lf\n", + " .with_columns(\n", + " incident_alarm().cast(pl.Int8).alias(\"тревожное\")\n", + " )\n", + " .drop(\"значение_датчика\")\n", + " )\n", + "\n", + " # Дата и время\n", + " lf = lf.with_columns(\n", + " pl.col(\"дата\").dt.weekday().alias(\"day_of_week\"),\n", + " pl.col(\"дата\").dt.month().alias(\"month\"),\n", + " pl.col(\"время\").dt.hour().alias(\"hour\"),\n", + " )\n", + "\n", + " # Погода\n", + " lf = (\n", + " lf\n", + " .with_columns(\n", + " pl.datetime(\n", + " pl.col(\"дата\").dt.year(),\n", + " pl.col(\"дата\").dt.month(),\n", + " pl.col(\"дата\").dt.day(),\n", + " pl.col(\"время\").dt.hour(),\n", + " 0,\n", + " 0,\n", + " ).alias(\"datetime\")\n", + " )\n", + " .join(\n", + " EXPECTED_WEATHER,\n", + " left_on=\"datetime\",\n", + " right_on=\"time\",\n", + " how=\"left\",\n", + " )\n", + " )\n", + "\n", + " return lf" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "a078cf8f", + "metadata": {}, + "outputs": [], + "source": [ + "from datetime import timedelta\n", + "\n", + "TARGET: str = \"тревожное\"\n", + "ID_COLS: list[str] = [\"ид_канала_данных\"]\n", + "TIME_COLS: list[str] = [\"дата\", \"время\"]\n", + "TIME_FIXED_CATS: list[str] = [\n", + " \"тип_инж_системы\",\n", + " \"тип_датчика\",\n", + " \"тег_инженерной_системы\",\n", + " \"day_of_week\",\n", + " # \"month\", # Не на нескольких годах обучаюсь, поэтому убрать\n", + " \"hour\",\n", + " 'иерархия_уровень',\n", + " 'родитель',\n", + " 'вид_объекта'\n", + "]\n", + "TIME_DYNAMIC_CATS: list[str] = [\n", + " \"value_cat__Есть питание\",\n", + " \"value_cat__На охране\",\n", + " \"value_cat__Дыма нет\",\n", + " \"value_cat__Неисправен\",\n", + " \"value_cat__Движения нет\",\n", + " \"value_cat__Рычаг сдернут\",\n", + " \"value_cat__Не определено\",\n", + " \"value_cat__Температура ниже 3ºC\",\n", + " \"value_cat__Вызов\",\n", + " \"value_cat__Много неисправных устройств\",\n", + " \"value_cat__Норма\",\n", + " \"value_cat__Устройства на объекте исправны\",\n", + " \"value_cat__Обнаружен газ\",\n", + " \"value_cat__Выключен\",\n", + " \"value_cat__Рычаг норма\",\n", + " \"value_cat__01.01.1970 03:00:00\",\n", + " \"value_cat__Обнаружено движение\",\n", + " \"value_cat__Питание от батарей\",\n", + " \"value_cat__Обнаружен дым\",\n", + " \"value_cat__Работают все насосы в АНС\",\n", + " \"value_cat__Включен\",\n", + " \"value_cat__Разговор\",\n", + " \"value_cat__В норме от +3 до +40\",\n", + " \"value_cat__01.01.1970 03:00:01\",\n", + " \"value_cat__Снято с охраны\",\n", + " \"value_cat__Неопределен\",\n", + " \"value_cat__Питание от сети\",\n", + " \"value_cat__Отключено устройство\",\n", + " \"value_cat__Обесточен\",\n", + " \"value_cat__Затоплен\",\n", + " \"value_cat__Не замкнут\",\n", + "]\n", + "TIME_CONTINUOUS: list[str] = [\"value_float\"]\n", + "VALUE_STATISTICS_OVER_DAYS: list[float] = [\n", + " 3,\n", + " 2,\n", + " 1,\n", + " 12/24,\n", + " 6/24,\n", + " 4/24,\n", + " 2/24,\n", + " 1/24,\n", + " 0,\n", + "]\n", + "FUTURE_VALUES: list[str] = [\n", + " 'temperature_2m (°C)',\n", + " 'relative_humidity_2m (%)',\n", + " 'precipitation_probability (%)',\n", + " 'precipitation (mm)',\n", + " 'rain (mm)',\n", + " 'snowfall (cm)',\n", + " 'snow_depth (m)',\n", + " 'weather_code (wmo code)',\n", + " 'cloud_cover (%)',\n", + " 'pressure_msl (hPa)'\n", + "]\n", + "FUTUTE_LAGS_DAYS = [0, 1/24, 4/24, 8/24, 12/24, 16/24, 20/24, 24/24]\n", + "\n", + "\n", + "def _window_name(days: float) -> str:\n", + " if days == 0:\n", + " return \"current\"\n", + "\n", + " hours = days * 24\n", + "\n", + " if float(days).is_integer():\n", + " return f\"{int(days)}d\"\n", + " if float(hours).is_integer():\n", + " return f\"{int(hours)}h\"\n", + "\n", + " minutes = hours * 60\n", + " if float(minutes).is_integer():\n", + " return f\"{int(minutes)}m\"\n", + "\n", + " return f\"{days:g}d\"\n", + "\n", + "def create_train_dataset(lf: pl.LazyFrame) -> pl.LazyFrame:\n", + " \"\"\"Создаёт временные фичи, погодные фичи будущих часов и таргет.\"\"\"\n", + "\n", + " date_col, time_col = TIME_COLS\n", + "\n", + " # Делаем единый datetime и сортируем данные по времени внутри каждого канала\n", + " lf = (\n", + " lf\n", + " .with_columns(\n", + " pl.datetime(\n", + " pl.col(date_col).dt.year(),\n", + " pl.col(date_col).dt.month(),\n", + " pl.col(date_col).dt.day(),\n", + " pl.col(time_col).dt.hour(),\n", + " pl.col(time_col).dt.minute(),\n", + " pl.col(time_col).dt.second(),\n", + " ).alias(\"__datetime\")\n", + " )\n", + " .sort([*ID_COLS, \"__datetime\", \"ид_события\"])\n", + " )\n", + "\n", + " # Находим время предыдущего активного TARGET, не включая текущую строку\n", + " prev_target_datetime = (\n", + " pl.when(pl.col(TARGET) == 1)\n", + " .then(pl.col(\"__datetime\"))\n", + " .otherwise(None)\n", + " .shift(1)\n", + " .fill_null(strategy=\"forward\")\n", + " .over(ID_COLS)\n", + " )\n", + "\n", + " # Находим время следующего активного TARGET, не включая текущую строку\n", + " next_target_datetime = (\n", + " pl.when(pl.col(TARGET) == 1)\n", + " .then(pl.col(\"__datetime\"))\n", + " .otherwise(None)\n", + " .shift(-1)\n", + " .fill_null(strategy=\"backward\")\n", + " .over(ID_COLS)\n", + " )\n", + "\n", + " # Добавляем расстояние до предыдущего и следующего активного TARGET в часах\n", + " lf = (\n", + " lf\n", + " .with_columns(\n", + " prev_target_datetime.alias(\"__prev_target_datetime\"),\n", + " next_target_datetime.alias(\"__next_target_datetime\"),\n", + " )\n", + " .with_columns(\n", + " (\n", + " (\n", + " pl.col(\"__datetime\")\n", + " - pl.col(\"__prev_target_datetime\")\n", + " )\n", + " .dt.total_seconds()\n", + " / 3600.0\n", + " ).alias(\"hours_since_prev_target\"),\n", + "\n", + " (\n", + " (\n", + " pl.col(\"__next_target_datetime\")\n", + " - pl.col(\"__datetime\")\n", + " )\n", + " .dt.total_seconds()\n", + " / 3600.0\n", + " ).alias(TARGET),\n", + " )\n", + " )\n", + "\n", + " feature_exprs: list[pl.Expr] = []\n", + "\n", + " # closed=\"right\" захватывает и более поздние строки той же секунды — в проде их ещё нет.\n", + " # Вычитаем их: сумма по (канал, секунда) минус накопленная сумма до текущей строки.\n", + " same_ts = [*ID_COLS, \"__datetime\"]\n", + " lf = lf.with_columns(\n", + " *[(pl.col(c).cast(pl.Int32).sum().over(same_ts) - pl.col(c).cast(pl.Int32).cum_sum().over(same_ts)).alias(f\"__later_{c}\")\n", + " for c in TIME_DYNAMIC_CATS],\n", + " *[(pl.col(c).fill_null(0).sum().over(same_ts) - pl.col(c).fill_null(0).cum_sum().over(same_ts)).alias(f\"__later_{c}\")\n", + " for c in TIME_CONTINUOUS],\n", + " *[(pl.col(c).is_not_null().cast(pl.Int32).sum().over(same_ts) - pl.col(c).is_not_null().cast(pl.Int32).cum_sum().over(same_ts)).alias(f\"__later_n_{c}\")\n", + " for c in TIME_CONTINUOUS],\n", + " )\n", + "\n", + " # Строим временные статистики по каждому заданному окну\n", + " for days in VALUE_STATISTICS_OVER_DAYS:\n", + " suffix = _window_name(days)\n", + "\n", + " # Для текущей точки просто сохраняем текущее значение\n", + " if days == 0:\n", + " feature_exprs.extend(\n", + " pl.col(col)\n", + " .alias(f\"{col}__sum_{suffix}\")\n", + " for col in TIME_DYNAMIC_CATS\n", + " )\n", + "\n", + " feature_exprs.extend(\n", + " pl.col(col)\n", + " .alias(f\"{col}__mean_{suffix}\")\n", + " for col in TIME_CONTINUOUS\n", + " )\n", + "\n", + " continue\n", + "\n", + " window = timedelta(days=days)\n", + "\n", + " # Считаем количество категориальных событий за окно\n", + " feature_exprs.extend(\n", + " (\n", + " pl.col(col).cast(pl.Int32)\n", + " .rolling_sum_by(\n", + " by=\"__datetime\",\n", + " window_size=window,\n", + " closed=\"right\",\n", + " )\n", + " .over(ID_COLS)\n", + " - pl.col(f\"__later_{col}\")\n", + " )\n", + " .alias(f\"{col}__sum_{suffix}\")\n", + " for col in TIME_DYNAMIC_CATS\n", + " )\n", + "\n", + " # Считаем среднее непрерывных значений за окно\n", + " # Среднее — через сумму и число непустых без поздних строк той же секунды\n", + " for col in TIME_CONTINUOUS:\n", + " s = pl.col(col).fill_null(0).rolling_sum_by(by=\"__datetime\", window_size=window, closed=\"right\").over(ID_COLS) - pl.col(f\"__later_{col}\")\n", + " n = pl.col(col).is_not_null().cast(pl.Int32).rolling_sum_by(by=\"__datetime\", window_size=window, closed=\"right\").over(ID_COLS) - pl.col(f\"__later_n_{col}\")\n", + " feature_exprs.append(pl.when(n > 0).then(s / n).otherwise(None).alias(f\"{col}__mean_{suffix}\"))\n", + "\n", + " # Добавляем rolling-фичи\n", + " lf = lf.with_columns(feature_exprs)\n", + "\n", + " # ============================================================\n", + " # Погодные прогнозы на текущий час и будущие часы\n", + " # ============================================================\n", + "\n", + " for days in FUTUTE_LAGS_DAYS:\n", + " hours = int(round(days * 24))\n", + " suffix = _window_name(days)\n", + "\n", + " join_col = f\"__future_datetime_{suffix}\"\n", + "\n", + " # Для события 14:37:\n", + " # 0h -> 14:00\n", + " # 1h -> 15:00\n", + " # 4h -> 18:00\n", + " # ...\n", + " lf = lf.with_columns(\n", + " (\n", + " pl.col(\"__datetime\").dt.truncate(\"1h\")\n", + " + pl.duration(hours=hours)\n", + " ).alias(join_col)\n", + " )\n", + "\n", + " weather_lf = (\n", + " EXPECTED_WEATHER\n", + " .select(\n", + " pl.col(\"time\"),\n", + " *[\n", + " pl.col(col).alias(f\"{col}__future_{suffix}\")\n", + " for col in FUTURE_VALUES\n", + " ],\n", + " )\n", + " )\n", + "\n", + " lf = (\n", + " lf\n", + " .join(\n", + " weather_lf,\n", + " left_on=join_col,\n", + " right_on=\"time\",\n", + " how=\"left\",\n", + " )\n", + " .drop(join_col)\n", + " )\n", + "\n", + " # Удаляем технические и исходные колонки\n", + " return (\n", + " lf\n", + " .drop(\n", + " \"__prev_target_datetime\",\n", + " \"__next_target_datetime\",\n", + " \"__datetime\",\n", + " *[f\"__later_{c}\" for c in TIME_DYNAMIC_CATS + TIME_CONTINUOUS],\n", + " *[f\"__later_n_{c}\" for c in TIME_CONTINUOUS],\n", + " *TIME_DYNAMIC_CATS,\n", + " *TIME_CONTINUOUS,\n", + " *FUTURE_VALUES,\n", + " # *TIME_COLS,\n", + " # *ID_COLS,\n", + " 'month'\n", + " )\n", + " )" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "9bf8d723", + "metadata": {}, + "outputs": [], + "source": [ + "from pathlib import Path\n", + "from tqdm.auto import tqdm\n", + "import polars as pl\n", + "\n", + "\n", + "def process_dataset_by_chunks(\n", + " lf: pl.LazyFrame,\n", + " output_dir: str = \"data/preprocessed_dataset\",\n", + " chunk_size: int = 500,\n", + ") -> None:\n", + "\n", + " output_dir = Path(output_dir)\n", + " output_dir.mkdir(parents=True, exist_ok=True)\n", + "\n", + " # Собираем в память только уникальные ID\n", + " # Детерминированная раскладка: каналы по порядку id, чанк ~ROWS_PER_CHUNK строк\n", + " ids = (\n", + " lf.group_by(ID_COLS).len().sort(ID_COLS).collect()\n", + " .with_columns((pl.col(\"len\").cum_sum() // ROWS_PER_CHUNK).rank(\"dense\").cast(pl.Int64).sub(1).alias(\"chunk\"))\n", + " )\n", + " ids.write_parquet(output_dir.parent / \"chunk_map.parquet\")\n", + " # Файл читаем один раз и режем по чанкам в памяти: скан 3.7 GB на каждый чанк давал ~70 с/чанк\n", + " parts = (\n", + " lf.join(ids.lazy().select(*ID_COLS, \"chunk\"), on=ID_COLS, how=\"inner\")\n", + " .collect()\n", + " .partition_by(\"chunk\", as_dict=True, include_key=False)\n", + " )\n", + "\n", + " # Один чанк грузит ~5 ядер из 32: считаем 5 чанков параллельно (polars отпускает GIL)\n", + " from concurrent.futures import ThreadPoolExecutor\n", + "\n", + " def one(chunk_idx: int) -> None:\n", + " parts.pop((chunk_idx,)).lazy().pipe(create_train_dataset).sink_parquet(\n", + " output_dir / f\"chunk_{chunk_idx:05d}.parquet\"\n", + " )\n", + "\n", + " n_chunks = ids[\"chunk\"].max() + 1\n", + " with ThreadPoolExecutor(max_workers=5) as ex:\n", + " list(tqdm(ex.map(one, range(n_chunks)), total=n_chunks, desc=\"Processing chunks\", unit=\"chunk\"))\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "6617e32a", + "metadata": {}, + "outputs": [], + "source": [ + "import json\n", + "import polars as pl\n", + "\n", + "with open(INCIDENT_MAPPING_PATH, \"w\", encoding=\"utf-8\") as f:\n", + " json.dump(INCIDENT_BY_SENSOR, f, ensure_ascii=False, indent=2)\n", + "\n", + "# Полные дубли строк журнала убираем (2023: 1.12M); одинаковые ид на разных каналах остаются.\n", + "# Только каналы 4 классов; V4_SMOKE=1 — ~2% каналов для проверки\n", + "import os\n", + "preproc(lf.unique()).filter(\n", + " pl.col(\"тип_датчика\").is_in(list(INCIDENT_BY_SENSOR))\n", + " & ((pl.col(\"ид_канала_данных\").hash(1) % 50 == 0) | pl.lit(os.environ.get(\"V4_SMOKE\") != \"1\"))\n", + ").sink_parquet('промежуточный df.parquet')\n", + "df = pl.scan_parquet('промежуточный df.parquet')\n", + "\n", + "process_dataset_by_chunks(df, chunk_size=100)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "c5cf056b", + "metadata": {}, + "outputs": [], + "source": [] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.13.15" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/retraining/notebooks/incident4/2_create_train_test_datasets_V2_MAIN.ipynb b/retraining/notebooks/incident4/2_create_train_test_datasets_V2_MAIN.ipynb new file mode 100644 index 0000000..2ad1d6b --- /dev/null +++ b/retraining/notebooks/incident4/2_create_train_test_datasets_V2_MAIN.ipynb @@ -0,0 +1,130 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": null, + "id": "a28dc8fe", + "metadata": {}, + "outputs": [], + "source": [ + "# %%\n", + "import gc\n", + "import json\n", + "from datetime import datetime\n", + "from pathlib import Path\n", + "\n", + "import polars as pl\n", + "from tqdm.auto import tqdm\n", + "\n", + "INPUT_DIR = Path(\"data/preprocessed_dataset\")\n", + "# blind = 10% каналов (слепой a), june = июнь 2026 всех каналов (слепой b)\n", + "OUTPUT_DIRS = {s: Path(f\"data/final_{s}\") for s in [\"train\", \"val\", \"blind\", \"june\"]}\n", + "\n", + "ID_COL = \"ид_канала_данных\"\n", + "SENSOR_COL = \"тип_датчика\"\n", + "with open(\"data/incident_by_sensor.json\", encoding=\"utf-8\") as f:\n", + " INCIDENT_BY_SENSOR: dict[str, str] = json.load(f)\n", + "\n", + "TARGET = \"тревожное\"\n", + "TARGET_UPPER_BOUND = 48\n", + "RANDOM_STATE = 69\n", + "BLIND_SIZE, VAL_SIZE = 0.10, 0.10 # train 80%\n", + "CUTOFF = datetime(2026, 6, 1)\n", + "# GAS и INTRUSION шлют поток строк: прореживаем только train, по хэшу ид_события\n", + "TRAIN_ROW_FRACTION = {\"GAS\": 0.15, \"INTRUSION\": 0.5}\n", + "\n", + "ROW_DT = pl.col(\"дата\").dt.combine(pl.col(\"время\"))\n", + "\n", + "\n", + "def make_target(lf: pl.LazyFrame, censor: bool) -> pl.LazyFrame:\n", + " t = pl.col(TARGET)\n", + " if censor:\n", + " # train/val: тревоги из июня (слепой период) не видим — цензура на CUTOFF\n", + " to_cut = (pl.lit(CUTOFF) - ROW_DT).dt.total_seconds() / 3600.0\n", + " t = pl.when(t.is_not_null() & (t <= to_cut)).then(t).otherwise(None)\n", + " bound = pl.min_horizontal(pl.lit(float(TARGET_UPPER_BOUND)), to_cut)\n", + " else:\n", + " bound = pl.lit(float(TARGET_UPPER_BOUND))\n", + " return lf.with_columns(\n", + " pl.when(t.is_not_null()).then(pl.min_horizontal(t, pl.lit(float(TARGET_UPPER_BOUND)))).otherwise(bound).alias(\"target_lower\"),\n", + " pl.when(t <= TARGET_UPPER_BOUND).then(t).otherwise(-1.0).alias(\"target_upper\"),\n", + " ).drop(TARGET)\n", + "\n", + "\n", + "def device_split(files: list[Path]) -> pl.DataFrame:\n", + " \"\"\"Внутри класса: каналы по id, shuffle seed 69 -> blind 10%, val 10%, train 80%.\"\"\"\n", + " ch = (\n", + " pl.scan_parquet(files).select(ID_COL, SENSOR_COL).unique(ID_COL)\n", + " .with_columns(pl.col(SENSOR_COL).replace_strict(INCIDENT_BY_SENSOR, default=None).alias(\"__incident\"))\n", + " .drop_nulls(\"__incident\").sort(ID_COL).collect()\n", + " )\n", + " parts = []\n", + " for _, g in ch.group_by(\"__incident\", maintain_order=True):\n", + " ids = g.select(ID_COL).sample(fraction=1.0, shuffle=True, seed=RANDOM_STATE)\n", + " n = ids.height\n", + " nb, nv = int(n * BLIND_SIZE), int(n * VAL_SIZE)\n", + " parts.append(ids.with_columns(\n", + " pl.when(pl.int_range(n) < nb).then(pl.lit(\"blind\"))\n", + " .when(pl.int_range(n) < nb + nv).then(pl.lit(\"val\"))\n", + " .otherwise(pl.lit(\"train\")).alias(\"__split\")))\n", + " return pl.concat(parts)\n", + "\n", + "\n", + "for d in OUTPUT_DIRS.values():\n", + " d.mkdir(parents=True, exist_ok=True)\n", + "files = sorted(INPUT_DIR.glob(\"*.parquet\"))\n", + "split = device_split(files)\n", + "split.write_parquet(\"data/device_split.parquet\")\n", + "print(split.group_by(\"__split\").len().sort(\"__split\"))\n", + "\n", + "keep = pl.col(\"ид_события\").hash(RANDOM_STATE) % 10_000 < (\n", + " pl.col(SENSOR_COL).replace_strict(INCIDENT_BY_SENSOR, default=None)\n", + " .replace_strict(TRAIN_ROW_FRACTION, default=1.0) * 10_000\n", + ")\n", + "before = ROW_DT < pl.lit(CUTOFF)\n", + "for fp in tqdm(files, desc=\"Processing chunks\"):\n", + " lf = pl.scan_parquet(fp).join(split.lazy(), on=ID_COL, how=\"inner\")\n", + " s = pl.col(\"__split\")\n", + " make_target(lf.filter(before & (s == \"train\") & keep), True).drop(\"__split\").sink_parquet(OUTPUT_DIRS[\"train\"] / fp.name)\n", + " make_target(lf.filter(before & (s == \"val\")), True).drop(\"__split\").sink_parquet(OUTPUT_DIRS[\"val\"] / fp.name)\n", + " make_target(lf.filter(before & (s == \"blind\")), False).drop(\"__split\").sink_parquet(OUTPUT_DIRS[\"blind\"] / fp.name)\n", + " make_target(lf.filter(~before), False).with_columns((s == \"blind\").alias(\"__blind_channel\")).drop(\"__split\").sink_parquet(OUTPUT_DIRS[\"june\"] / fp.name)\n", + " gc.collect()\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "b6ba13a7", + "metadata": {}, + "outputs": [], + "source": [ + "for s in [\"train\", \"val\", \"blind\", \"june\"]:\n", + " print(s, pl.scan_parquet(f\"data/final_{s}/*.parquet\").select(\n", + " pl.len(), pl.col(\"ид_канала_данных\").n_unique(),\n", + " ((pl.col(\"target_upper\") != -1) & (pl.col(\"target_lower\") <= 24)).sum().alias(\"pos24\")).collect())\n" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.13.15" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/retraining/notebooks/incident4/3_feature_selection.ipynb b/retraining/notebooks/incident4/3_feature_selection.ipynb new file mode 100644 index 0000000..859ce6b --- /dev/null +++ b/retraining/notebooks/incident4/3_feature_selection.ipynb @@ -0,0 +1,736 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": null, + "id": "d55ed3bb", + "metadata": {}, + "outputs": [], + "source": [ + "# %%\n", + "# Импорты и настройки\n", + "\n", + "import gc\n", + "import json\n", + "import math\n", + "from pathlib import Path\n", + "\n", + "import numpy as np\n", + "import polars as pl\n", + "\n", + "from catboost import CatBoostClassifier, Pool\n", + "from sklearn.metrics import average_precision_score\n", + "\n", + "\n", + "HORIZON = 24.0\n", + "\n", + "TO_DROP = [\n", + " \"ид_события\",\n", + " \"ид_канала_данных\",\n", + " \"дата\",\n", + " \"время\",\n", + " \"datetime\",\n", + "]\n", + "\n", + "TARGET_COLS = [\n", + " \"target_lower\",\n", + " \"target_upper\",\n", + "]\n", + "\n", + "\n", + "# Чанки можно задавать отдельно\n", + "# train прорежен в ноутбуке 2, поэтому берём каждый 3-й чанк, а не первые 4\n", + "N_CHUNKS = len(list(Path(\"data/final_train\").glob(\"chunk_*.parquet\")))\n", + "TRAIN_CHUNKS = [*range(0, N_CHUNKS, 4)] # четверть прореженного train\n", + "VAL_CHUNKS = [*range(N_CHUNKS)]\n", + "TEST_CHUNKS = [] # слепые наборы в отборе не участвуют\n", + "\n", + "\n", + "DROP_PER_STEP = 20\n", + "MIN_FEATURES = 10\n", + "\n", + "# Сколько строк validation использовать для SHAP\n", + "SHAP_ROWS = 50_000\n", + "\n", + "\n", + "OUTPUT_DIR = Path(\"feature_selection_shap_rfe\")\n", + "OUTPUT_DIR.mkdir(parents=True, exist_ok=True)\n", + "\n", + "HISTORY_PATH = OUTPUT_DIR / \"history.csv\"\n", + "SELECTED_STEP_PATH = OUTPUT_DIR / \"selected_step.json\"\n", + "SENSOR_COL = \"тип_датчика\"\n", + "\n", + "\n", + "with open(\"data/incident_by_sensor.json\", encoding=\"utf-8\") as f:\n", + " INCIDENT_BY_SENSOR: dict[str, str] = json.load(f)\n", + "\n", + "\n", + "def macro_pr_auc(\n", + " y: np.ndarray,\n", + " p: np.ndarray,\n", + " sensor: np.ndarray,\n", + ") -> float:\n", + " \"\"\"Среднее PR-AUC по 4 классам: GAS (88% строк) не перевешивает остальных.\"\"\"\n", + " incident = pl.Series(sensor).replace_strict(\n", + " INCIDENT_BY_SENSOR, default=None\n", + " ).to_numpy()\n", + " scores = [\n", + " average_precision_score(y[incident == c], p[incident == c])\n", + " for c in sorted(set(INCIDENT_BY_SENSOR.values()))\n", + " if y[incident == c].any()\n", + " ]\n", + " return float(np.mean(scores))\n", + "\n", + "\n", + "MODEL_PARAMS = {\n", + " \"loss_function\": \"Logloss\",\n", + " \"eval_metric\": \"PRAUC:type=Classic\",\n", + " \"iterations\": 100,\n", + " \"depth\": 6,\n", + " \"learning_rate\": 0.05,\n", + " \"random_seed\": 42,\n", + " \"verbose\": 1,\n", + "}" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ebd1457d", + "metadata": {}, + "outputs": [], + "source": [ + "# %%\n", + "# LazyFrame для нужного split\n", + "\n", + "def scan_split(\n", + " split: str,\n", + " chunks: list[int],\n", + ") -> pl.LazyFrame:\n", + "\n", + " files = [\n", + " f\"data/final_{split}/chunk_{i:05d}.parquet\"\n", + " for i in chunks\n", + " ]\n", + "\n", + " return (\n", + " pl.scan_parquet(files)\n", + " .drop(TO_DROP)\n", + " .with_columns(\n", + " (\n", + " (pl.col(\"target_upper\") != -1)\n", + " & (pl.col(\"target_lower\") <= HORIZON)\n", + " )\n", + " .cast(pl.Int8)\n", + " .alias(\"target\")\n", + " )\n", + " )\n", + "\n", + "\n", + "# Схему берем без загрузки данных\n", + "train_schema = scan_split(\n", + " \"train\",\n", + " [TRAIN_CHUNKS[0]],\n", + ").collect_schema()\n", + "\n", + "\n", + "feature_cols = [\n", + " c\n", + " for c in train_schema.names()\n", + " if c not in TARGET_COLS + [\"target\"]\n", + "]\n", + "\n", + "\n", + "cat_features = [\n", + " c\n", + " for c in feature_cols\n", + " if train_schema[c] in (\n", + " pl.String,\n", + " pl.Categorical,\n", + " pl.Enum,\n", + " )\n", + "]\n", + "\n", + "\n", + "print(f\"Features: {len(feature_cols)}\")\n", + "print(f\"Cat feats: {len(cat_features)}\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "ba78dd73", + "metadata": {}, + "outputs": [], + "source": [ + "# %%\n", + "# Служебные функции\n", + "\n", + "def collect_df(\n", + " split: str,\n", + " chunks: list[int],\n", + " features: list[str],\n", + " cat_features_current: list[str],\n", + " with_target: bool = True,\n", + " head: int | None = None,\n", + ") -> pl.DataFrame:\n", + "\n", + " cols = [*dict.fromkeys(\n", + " features + ([\"target\", SENSOR_COL] if with_target else [])\n", + " )]\n", + "\n", + " lf = (\n", + " scan_split(split, chunks)\n", + " .select(cols)\n", + " )\n", + "\n", + " # Null в категориях заполняем до collect\n", + " if cat_features_current:\n", + " lf = lf.with_columns(\n", + " pl.col(cat_features_current).fill_null(\"null\")\n", + " )\n", + "\n", + " if head is not None:\n", + " lf = lf.head(head)\n", + "\n", + " return lf.collect(\n", + " engine=\"streaming\"\n", + " )\n", + "\n", + "\n", + "def make_pool(\n", + " split: str,\n", + " chunks: list[int],\n", + " features: list[str],\n", + " cat_features_current: list[str],\n", + " return_y: bool = False,\n", + ") -> Pool | tuple[Pool, np.ndarray]:\n", + "\n", + " df = collect_df(\n", + " split=split,\n", + " chunks=chunks,\n", + " features=features,\n", + " cat_features_current=cat_features_current,\n", + " with_target=True,\n", + " )\n", + "\n", + " y = df[\"target\"].to_numpy()\n", + " sensor = df[SENSOR_COL].to_numpy()\n", + "\n", + " pool = Pool(\n", + " data=df.select(features),\n", + " label=y,\n", + " cat_features=cat_features_current,\n", + " )\n", + "\n", + " del df\n", + " gc.collect()\n", + "\n", + " if return_y:\n", + " return pool, y, sensor\n", + "\n", + " del y\n", + " gc.collect()\n", + "\n", + " return pool" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "173067ab", + "metadata": {}, + "outputs": [], + "source": [ + "# %%\n", + "# Test score считаем по чанкам\n", + "# В RAM одновременно лежит только один test chunk\n", + "\n", + "def get_test_pr_auc(\n", + " model: CatBoostClassifier,\n", + " features: list[str],\n", + " cat_features_current: list[str],\n", + ") -> float:\n", + "\n", + " if not TEST_CHUNKS:\n", + " return float(\"nan\")\n", + "\n", + " y_parts = []\n", + " p_parts = []\n", + "\n", + " for chunk in TEST_CHUNKS:\n", + "\n", + " test_pool, y_chunk, _ = make_pool(\n", + " split=\"test\",\n", + " chunks=[chunk],\n", + " features=features,\n", + " cat_features_current=cat_features_current,\n", + " return_y=True,\n", + " )\n", + "\n", + " p_chunk = model.predict_proba(\n", + " test_pool\n", + " )[:, 1]\n", + "\n", + " y_parts.append(y_chunk)\n", + " p_parts.append(p_chunk)\n", + "\n", + " del test_pool\n", + " del y_chunk\n", + " del p_chunk\n", + "\n", + " gc.collect()\n", + "\n", + "\n", + " # Для точного PR-AUC нужны prediction по всему test\n", + " # Это только два одномерных массива, поэтому памяти мало\n", + " y_test = np.concatenate(y_parts)\n", + " p_test = np.concatenate(p_parts)\n", + "\n", + " score = average_precision_score(\n", + " y_test,\n", + " p_test,\n", + " )\n", + "\n", + " del y_parts\n", + " del p_parts\n", + " del y_test\n", + " del p_test\n", + "\n", + " gc.collect()\n", + "\n", + " return score" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "5fb0c15a", + "metadata": {}, + "outputs": [], + "source": [ + "# %%\n", + "# SHAP считаем частями по validation chunks\n", + "# Полная SHAP-матрица для всего validation в RAM не создается\n", + "\n", + "def get_mean_abs_shap(\n", + " model: CatBoostClassifier,\n", + " features: list[str],\n", + " cat_features_current: list[str],\n", + ") -> np.ndarray:\n", + "\n", + " shap_sum = np.zeros(\n", + " len(features),\n", + " dtype=np.float64,\n", + " )\n", + "\n", + " total_rows = 0\n", + "\n", + " rows_per_chunk = math.ceil(\n", + " SHAP_ROWS / len(VAL_CHUNKS)\n", + " )\n", + "\n", + "\n", + " for chunk in VAL_CHUNKS:\n", + "\n", + " shap_df = collect_df(\n", + " split=\"val\",\n", + " chunks=[chunk],\n", + " features=features,\n", + " cat_features_current=cat_features_current,\n", + " with_target=False,\n", + " head=rows_per_chunk,\n", + " )\n", + "\n", + " if shap_df.height == 0:\n", + " del shap_df\n", + " gc.collect()\n", + " continue\n", + "\n", + "\n", + " shap_pool = Pool(\n", + " data=shap_df,\n", + " cat_features=cat_features_current,\n", + " )\n", + "\n", + " del shap_df\n", + " gc.collect()\n", + "\n", + "\n", + " shap_values = model.get_feature_importance(\n", + " shap_pool,\n", + " type=\"ShapValues\",\n", + " )\n", + "\n", + "\n", + " shap_sum += np.abs(\n", + " shap_values[:, :-1]\n", + " ).sum(axis=0)\n", + "\n", + " total_rows += shap_values.shape[0]\n", + "\n", + "\n", + " del shap_pool\n", + " del shap_values\n", + "\n", + " gc.collect()\n", + "\n", + "\n", + " if total_rows == 0:\n", + " raise ValueError(\"Не удалось получить строки для SHAP\")\n", + "\n", + "\n", + " return shap_sum / total_rows" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "3c05a522", + "metadata": {}, + "outputs": [], + "source": [ + "# %%\n", + "# Feature selection\n", + "# На каждом шаге:\n", + "# 1. Загружаем только текущие фичи\n", + "# 2. Обучаем модель\n", + "# 3. Сразу удаляем train/val Pool\n", + "# 4. Считаем test по чанкам\n", + "# 5. Считаем SHAP на ограниченной части validation\n", + "# 6. Убираем 10 худших фичей\n", + "\n", + "history = []\n", + "\n", + "current_features = feature_cols.copy()\n", + "\n", + "removed_this_step = []\n", + "\n", + "step = 0\n", + "\n", + "\n", + "while True:\n", + "\n", + " print()\n", + " print(\"=\" * 80)\n", + " print(\n", + " f\"step={step} | \"\n", + " f\"features={len(current_features)}\"\n", + " )\n", + "\n", + "\n", + " current_cat_features = [\n", + " c\n", + " for c in cat_features\n", + " if c in current_features\n", + " ]\n", + "\n", + "\n", + " # Train Pool\n", + " train_pool = make_pool(\n", + " split=\"train\",\n", + " chunks=TRAIN_CHUNKS,\n", + " features=current_features,\n", + " cat_features_current=current_cat_features,\n", + " return_y=False,\n", + " )\n", + "\n", + "\n", + " # Validation Pool\n", + " val_pool, y_val, sensor_val = make_pool(\n", + " split=\"val\",\n", + " chunks=VAL_CHUNKS,\n", + " features=current_features,\n", + " cat_features_current=current_cat_features,\n", + " return_y=True,\n", + " )\n", + "\n", + "\n", + " # Обучение\n", + " model = CatBoostClassifier(\n", + " **MODEL_PARAMS\n", + " )\n", + "\n", + " model.fit(\n", + " train_pool,\n", + " eval_set=val_pool,\n", + " early_stopping_rounds=50,\n", + " )\n", + "\n", + "\n", + " # Validation score\n", + " p_val = model.predict_proba(\n", + " val_pool\n", + " )[:, 1]\n", + "\n", + " val_pr_auc = average_precision_score(\n", + " y_val,\n", + " p_val,\n", + " )\n", + "\n", + " val_macro_pr_auc = macro_pr_auc(\n", + " y_val,\n", + " p_val,\n", + " sensor_val,\n", + " )\n", + "\n", + "\n", + " # Train и validation больше не нужны\n", + " del train_pool\n", + " del val_pool\n", + " del y_val\n", + " del p_val\n", + " del sensor_val\n", + "\n", + " gc.collect()\n", + "\n", + "\n", + " # Test score по одному чанку\n", + " test_pr_auc = get_test_pr_auc(\n", + " model=model,\n", + " features=current_features,\n", + " cat_features_current=current_cat_features,\n", + " )\n", + "\n", + "\n", + " print(\n", + " f\"val={val_pr_auc:.6f} | \"\n", + " f\"val_macro={val_macro_pr_auc:.6f} | \"\n", + " f\"test={test_pr_auc:.6f}\"\n", + " )\n", + "\n", + "\n", + " # Сохраняем результат шага\n", + " history.append({\n", + " \"step\": step,\n", + " \"n_features\": len(current_features),\n", + " \"val_pr_auc\": val_pr_auc,\n", + " \"val_macro_pr_auc\": val_macro_pr_auc,\n", + " \"test_pr_auc\": test_pr_auc,\n", + " \"best_iteration\": model.get_best_iteration(),\n", + " \"removed_features\": \" | \".join(\n", + " removed_this_step\n", + " ),\n", + " })\n", + "\n", + "\n", + " pl.DataFrame(\n", + " history\n", + " ).write_csv(\n", + " HISTORY_PATH\n", + " )\n", + "\n", + "\n", + " # Если дошли до минимального числа фичей\n", + " if len(current_features) <= MIN_FEATURES:\n", + "\n", + " del model\n", + "\n", + " gc.collect()\n", + "\n", + " break\n", + "\n", + "\n", + " # SHAP на ограниченной части validation\n", + " mean_abs_shap = get_mean_abs_shap(\n", + " model=model,\n", + " features=current_features,\n", + " cat_features_current=current_cat_features,\n", + " )\n", + "\n", + "\n", + " shap_importance = (\n", + " pl.DataFrame({\n", + " \"feature\": current_features,\n", + " \"mean_abs_shap\": mean_abs_shap,\n", + " })\n", + " .sort(\"mean_abs_shap\")\n", + " )\n", + "\n", + "\n", + " n_drop = min(\n", + " DROP_PER_STEP,\n", + " len(current_features) - MIN_FEATURES,\n", + " )\n", + "\n", + "\n", + " removed_this_step = (\n", + " shap_importance\n", + " .head(n_drop)\n", + " [\"feature\"]\n", + " .to_list()\n", + " )\n", + "\n", + "\n", + " print(\"remove:\")\n", + "\n", + " for feature in removed_this_step:\n", + " print(f\"# {feature}\")\n", + "\n", + "\n", + " removed_set = set(\n", + " removed_this_step\n", + " )\n", + "\n", + "\n", + " current_features = [\n", + " c\n", + " for c in current_features\n", + " if c not in removed_set\n", + " ]\n", + "\n", + "\n", + " step += 1\n", + "\n", + "\n", + " del mean_abs_shap\n", + " del shap_importance\n", + " del model\n", + "\n", + " gc.collect()" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "d248695f", + "metadata": {}, + "outputs": [], + "source": [ + "# %%\n", + "# Итоговая таблица\n", + "\n", + "history_df = pl.read_csv(\n", + " HISTORY_PATH\n", + ")\n", + "\n", + "history_df" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "76df07d2", + "metadata": {}, + "outputs": [], + "source": [ + "history_df.sort(\"val_macro_pr_auc\", descending=True)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "9e536161", + "metadata": {}, + "outputs": [], + "source": [ + "# %%\n", + "# Берем набор фичей на выбранном шаге\n", + "\n", + "history_df = pl.read_csv(HISTORY_PATH)\n", + "\n", + "# Минимум фичей при macro PR-AUC на val не хуже лучшего шага на 0.005\n", + "best_macro = history_df[\"val_macro_pr_auc\"].max()\n", + "SELECTED_STEP = (\n", + " history_df\n", + " .filter(pl.col(\"val_macro_pr_auc\") >= best_macro - 0.005)\n", + " .sort(\"n_features\")\n", + " [\"step\"][0]\n", + ")\n", + "SELECTED_STEP_PATH.write_text(json.dumps({\"step\": int(SELECTED_STEP)}))\n", + "\n", + "removed_features = (\n", + " history_df\n", + " .filter(pl.col(\"step\") <= SELECTED_STEP)\n", + " .select(\"removed_features\")\n", + " .drop_nulls()\n", + " [\"removed_features\"]\n", + " .to_list()\n", + ")\n", + "\n", + "removed_features = {\n", + " feature\n", + " for row in removed_features\n", + " for feature in row.split(\" | \")\n", + " if feature\n", + "}\n", + "\n", + "selected_features = [\n", + " feature\n", + " for feature in feature_cols\n", + " if feature not in removed_features\n", + "]\n", + "\n", + "print(f\"Step: {SELECTED_STEP}\")\n", + "print(f\"Features: {len(selected_features)}\")\n", + "\n", + "selected_features" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "20219cf3", + "metadata": {}, + "outputs": [], + "source": [ + "# %%\n", + "def get_n_rows(\n", + " split: str,\n", + " chunks: list[int],\n", + ") -> int:\n", + "\n", + " files = [\n", + " f\"data/final_{split}/chunk_{i:05d}.parquet\"\n", + " for i in chunks\n", + " ]\n", + "\n", + " return (\n", + " pl.scan_parquet(files)\n", + " .select(pl.len())\n", + " .collect()\n", + " .item()\n", + " )\n", + "\n", + "\n", + "n_train = get_n_rows(\n", + " \"train\",\n", + " TRAIN_CHUNKS,\n", + ")\n", + "\n", + "n_val = get_n_rows(\n", + " \"val\",\n", + " VAL_CHUNKS,\n", + ")\n", + "\n", + "n_test = 0 # test не используется\n", + "\n", + "\n", + "print(f\"Train: {n_train:,}\")\n", + "print(f\"Val: {n_val:,}\")\n", + "print(f\"Test: {n_test:,}\")" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.13.15" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/retraining/notebooks/incident4/README.md b/retraining/notebooks/incident4/README.md new file mode 100644 index 0000000..243cf30 --- /dev/null +++ b/retraining/notebooks/incident4/README.md @@ -0,0 +1,97 @@ +# incident4 — модель по 4 классам инцидентов (прогон v4) + +Отдельный пайплайн рядом с общими ноутбуками `retraining/notebooks/`. Ноутбуки и скрипты в `retraining/notebooks/` он не меняет. + +## Назначение + +Модель в проде предсказывает «любую тревогу на канале», и 58% таких тревог — «Неисправен». Эта модель предсказывает **событие одного из 4 классов: FIRE, FLOOD, GAS, INTRUSION** на канале датчика в ближайшие 24 ч. Класс канала задаётся типом датчика (`data/incident_by_sensor.json`, пишет nb1). Каналы других типов не прогнозируются. + +Модель: CatBoostRegressor, `SurvivalAft`. Вероятность считается как `P(T ≤ h) = ndtr((ln h − pred) / scale)`, метрики — PR-AUC при h = 24 ч. + +## Отличия от общих ноутбуков + +| | Общие ноутбуки 1–5 | incident4 v4 | +|---|---|---| +| Цель | любая тревога | событие класса канала. FIRE/FLOOD/GAS — по флагу `справочник_состояний.csv` («вариант 1», сопоставление по названию состояния), INTRUSION — по флагу `тревожное` журнала | +| «Не замкнут» | — | у «Датчик дыма» и «Состояние УИР-Р» считается событием FIRE | +| Скользящие окна | `closed="right"` видит более поздние строки той же секунды | эти строки вычитаются (`__later_*`), сортировка по (канал, время, `ид_события`). Проверка: `check_lookahead.py` | +| Дубли журнала | остаются | полные дубли строк удаляются (`lf.unique()`, в 2023 году 1.12 млн) | +| Годы | 2024–2026 | 2019, 2020, 2022–2026 (2021 исключён) | +| Сплит | train/val/test | внутри класса по каналам: 80% train / 10% val / 10% слепой (а), seed 69. Слепой (б) — июнь 2026 всех каналов | +| Цензура | — | цель train/val цензурирована на 01.06.2026 | +| Прореживание | — | только train: GAS 0.15, INTRUSION 0.5 (по хэшу `ид_события`) | +| Отбор признаков (nb3) | test участвует | по macro PR-AUC на val (выбирается минимум признаков в пределах 0.005 от лучшего шага); слепые наборы не участвуют | +| Обучение | ноутбуки 4–5 | `train_v4.py`: параметры v1 (`best_params.json`), весь train, сегменты по 100 деревьев, ранний стоп по macro PR-AUC на val (терпение 3 сегмента, потолок 1500) | +| Оценка | test в ноутбуке 5 | `eval_v4.py` один раз на слепых наборах: новая модель, v3, прод, правила | + +## Запуск + +Входные файлы кладутся в `/data/`: + +- `ext-journal-{2019,2020,2022,2023,2024,2025,2026}.parquet` — журналы событий; +- `справочник_каналов_датчиков.parquet`, `справочник_объектов_диспетчер.parquet` (есть в `retraining/data/reference/`); +- `справочник_состояний.csv` из датасета Заказчика; +- `open-meteo-55.75N37.63E140m.csv` — погода Open-Meteo за 2019–2026. В репозитории файл начинается с 2021 года. Период 2019–2020 докачан с того же endpoint (`historical-forecast-api.open-meteo.com/v1/forecast`, 55.7558 / 37.6173, те же 10 переменных). + +Для `eval_v4.py` нужны модели и прогоны для сравнения. По умолчанию они ищутся в `/compare/`. Пути задаются переменными `COMPARE_DIR`, `V3_MODEL`, `PROD_MODEL`, `V3_CHUNK_MAP` и `V2_CHUNK_MAP`. + +```bash +# полный прогон: nb1 -> nb2 -> nb3 -> smoke обучения -> обучение -> слепая оценка +env -u V4_SMOKE PY=python retraining/notebooks/incident4/pipeline_v4.sh [RUN_DIR] +``` + +- `RUN_DIR` передаётся аргументом или переменной окружения. По умолчанию это `retraining/artifacts/runs/incident4_v4/` (каталог в `.gitignore`). Скрипт копирует туда ноутбуки и `best_params.json` и выполняет ноутбуки через `jupyter nbconvert --execute`. +- `V4_SMOKE=1` оставляет в nb1 ~2% каналов. Эта настройка только для проверки, при полном прогоне переменная не задаётся. +- `python check_lookahead.py ` проверяет после nb1 фикс look-ahead перебором по первому чанку. Ожидается `mismatch 0`. +- Результаты: `survival_optuna/best_model.cbm`, `best_trees.json`, `val_by_trees.csv`, `model_.cbm` на каждый сегмент, `selected_features.json`, `eval_blind.csv`. + +Ресурсы: пик RAM ~120 GB (nb1), полный прогон ~7 ч на 32 vCPU. Из них nb3 занимает ~1.6 ч, обучение ~17.5 мин на 100 деревьев. Прогону хватило диска на 150 GB. + +## Результаты v4 (1500 деревьев, 177 признаков) + +Обучение шло на 49.4 млн строк train и не остановилось: 1500 — потолок, val macro к концу ещё росло (0.636). Слепой тест посчитан один раз, в конце. + +(а) 10% каналов, которые v4 не видела, все годы до июня 2026: + +| Класс | Каналов | **v4** | v3 | прод | правило | +|---|---|---|---|---|---| +| FIRE | 570 | **0.697** | 0.533 | 0.141 | 0.312 | +| FLOOD | 20 | 0.816 | **0.856** | 0.415 | 0.726 | +| GAS | 52 | **0.450** | 0.168 | 0.200 | 0.039 | +| INTRUSION | 221 | **0.737** | 0.711 | 0.570 | 0.688 | +| Среднее | 863 | **0.675** | 0.567 | 0.332 | 0.441 | + +(б) июнь 2026, все каналы: + +| Класс | Каналов | v4 | **v3** | прод | правило | +|---|---|---|---|---|---| +| FIRE | 3412 | 0.411 | **0.634** | 0.043 | 0.229 | +| FLOOD | 181 | **0.754** | 0.736 | 0.528 | 0.636 | +| GAS | 396 | 0.194 | **0.381** | 0.019 | 0.038 | +| INTRUSION | 1818 | 0.476 | **0.510** | 0.335 | 0.400 | +| Среднее | 5807 | 0.459 | **0.565** | 0.231 | 0.326 | + +Только каналы, которых не видели v3 и прод: + +| Срез | Каналов | v4 | v3 | прод | правило | +|---|---|---|---|---|---| +| (а) | 45 | **0.750** | 0.684 | 0.436 | 0.538 | +| (б) июнь | 260 | 0.661 | **0.698** | 0.331 | 0.549 | + +Правило — `1/(1 + часы с прошлого события класса на канале)`. Полная таблица с подмножеством `b_june_blind_channels` и правилом `rule24` находится в `eval_blind.csv` рядом с моделью. + +Оговорки: + +- v3 и прод учились на каналах 2024–2026 и, вероятно, видели июнь, поэтому сравнение с ними на (б) им подыгрывает. +- На июне v4 слабее v3 по FIRE (0.41 против 0.63). Гипотеза — дрейф поведения пожарных датчиков в 2026, она не проверена. +- Кривая nb3 плоская (~0.50 от 377 до 57 признаков), так что выбор шага по сути шум. +- Мало каналов редких классов: FLOOD 20 и GAS 52 в (а). Оценки по ним шумные. +- Сплит идёт по каналам, не по времени. Проверка по времени — только (б). +- На границах данных (конец июня, перед 01.06.2026, разрыв 2020→2022) строки размечены как «события не было», хотя исход неизвестен. ~1000 каналов в год за 2019–2020 выпадают: их нет в справочнике каналов. +- В 2019–2020 у погоды нет `precipitation_probability` и `snow_depth` (NaN). +- Соответствие «тип датчика → класс» выведено по данным и Заказчиком не подтверждено. +- Слепой тест уже посмотрен. Если выбирать по нему модель или число деревьев, он перестаёт быть независимым. + +## Модели + +Бинарники в репозиторий не коммитятся. Модель `incident4_v4_model1500` лежит в командной папке `Models/`. Там же `selected_features.json`, `incident_by_sensor.json`, `val_by_trees.csv` и `eval_blind.csv`. Данные прогона и слепые предсказания сохранены на отключённом облачном диске команды. diff --git a/retraining/notebooks/incident4/best_params.json b/retraining/notebooks/incident4/best_params.json new file mode 100644 index 0000000..9a98064 --- /dev/null +++ b/retraining/notebooks/incident4/best_params.json @@ -0,0 +1,7 @@ +{ + "scale": 0.6519151388952278, + "iterations": 100, + "depth": 9, + "learning_rate": 0.14762522256363309, + "l2_leaf_reg": 22.647264220860485 +} \ No newline at end of file diff --git a/retraining/notebooks/incident4/check_lookahead.py b/retraining/notebooks/incident4/check_lookahead.py new file mode 100644 index 0000000..7094eda --- /dev/null +++ b/retraining/notebooks/incident4/check_lookahead.py @@ -0,0 +1,21 @@ +"""Проверка фикса look-ahead: sum_1h = число строк категории в (t-1h, t] до текущей строки включительно (brute force).""" +import sys +import numpy as np, polars as pl +run = sys.argv[1] +d = pl.read_parquet(f"{run}/data/preprocessed_dataset/chunk_00000.parquet") +col = "value_cat__Норма" +f1h = f"{col}__sum_1h" +d = d.with_columns((pl.col("дата").dt.combine(pl.col("время"))).alias("dt")).sort("ид_канала_данных", "dt", "ид_события") +bad = checked = ties = 0 +for ch, g in d.group_by("ид_канала_данных", maintain_order=True): + g = g.head(3000) + # исходного индикатора в выходе нет: восстанавливаем из sum_current + ind = g[f"{col}__sum_current"].to_numpy().astype(int) + t = g["dt"].to_numpy() + for i in range(len(g)): + lo = t[i] - np.timedelta64(3600, "s") + exp = int(ind[: i + 1][(t[: i + 1] > lo)].sum()) + checked += 1 + ties += int(i + 1 < len(g) and t[i + 1] == t[i]) + bad += int(exp != g[f1h][i]) +print("checked", checked, "rows with same-second successor", ties, "mismatch", bad) diff --git a/retraining/notebooks/incident4/eval_v4.py b/retraining/notebooks/incident4/eval_v4.py new file mode 100644 index 0000000..e46c68b --- /dev/null +++ b/retraining/notebooks/incident4/eval_v4.py @@ -0,0 +1,88 @@ +"""Слепая оценка incident4_v4 (один раз): new vs v3 vs prod(old) vs правила; (a) слепые каналы, (b) июнь 2026.""" +import gc, json, os, sys, time +from pathlib import Path +import numpy as np, polars as pl +from catboost import CatBoostRegressor, Pool +from scipy.special import ndtr +from sklearn.metrics import average_precision_score + +RUN = Path(sys.argv[1]).resolve() +H = 24.0 +# Модели и прогоны для сравнения — вне репозитория, пути через env (по умолчанию RUN/compare/) +CMP = Path(os.environ.get("COMPARE_DIR", RUN / "compare")) +MODELS = {"new": RUN / "survival_optuna/best_model.cbm", + "v3": Path(os.environ.get("V3_MODEL", CMP / "incident4_v3_500.cbm")), + "prod": Path(os.environ.get("PROD_MODEL", CMP / "old_main.cbm"))} +V3_CHUNK_MAP = Path(os.environ.get("V3_CHUNK_MAP", CMP / "run_v3/data/chunk_map.parquet")) +V2_CHUNK_MAP = Path(os.environ.get("V2_CHUNK_MAP", CMP / "run_v2/data/chunk_map.parquet")) +t0 = time.time() +inc = json.load(open(RUN / "data/incident_by_sensor.json")) +models = {} +for k, p in MODELS.items(): + m = CatBoostRegressor(); m.load_model(str(p)) + scale = float(m.get_all_params()["loss_function"].split("scale=")[1]) + models[k] = (m, scale, m.feature_names_, [m.feature_names_[i] for i in m.get_cat_feature_indices()]) + print(k, m.tree_count_, len(m.feature_names_), round(scale, 4), flush=True) + + +def predict(split): + files = sorted((RUN / f"data/final_{split}").glob("chunk_*.parquet")) + need = sorted({f for _, _, fs, _ in models.values() for f in fs} | {"target_lower", "target_upper", "тип_датчика", "ид_канала_данных", "hours_since_prev_target"}) + extra = ["__blind_channel"] if split == "june" else [] + parts = [] + for b in np.array_split(np.array(files, dtype=object), 16): + if len(b) == 0: + continue + df = pl.scan_parquet(list(b)).select(need + extra).collect(engine="streaming") + if df.height == 0: + continue + out = {"ch": df["ид_канала_данных"].cast(pl.Int64), "sensor": df["тип_датчика"], + "y": ((df["target_upper"] != -1) & (df["target_lower"] <= H)).cast(pl.Int8), + "hprev": df["hours_since_prev_target"].cast(pl.Float64)} + if extra: + out["blind_ch"] = df["__blind_channel"] + for k, (m, scale, fs, cats) in models.items(): + X = df.select(fs).with_columns(pl.col(cats).cast(pl.String).fill_null("null")) + out[f"p_{k}"] = ndtr((np.log(H) - m.predict(Pool(X, cat_features=cats))) / scale) + parts.append(pl.DataFrame(out)); del df; gc.collect() + r = pl.concat(parts).with_columns( + pl.col("sensor").replace_strict(inc, default=None).alias("cls"), + pl.when(pl.col("hprev").is_not_null()).then(1.0 / (1.0 + pl.col("hprev"))).otherwise(0.0).alias("p_rule"), + (pl.col("hprev") <= H).fill_null(False).cast(pl.Float64).alias("p_rule24")) + r.write_parquet(RUN / f"eval_preds_{split}.parquet") + print(split, "rows", r.height, "channels", r["ch"].n_unique(), "pos", int(r["y"].sum()), f"t={time.time()-t0:.0f}s", flush=True) + return r + + +def ap(d, c): + return round(float(average_precision_score(d["y"].to_numpy(), d[c].to_numpy())), 3) if d.height and d["y"].sum() else None + + +# Каналы, которые v3/prod видели при обучении или выборе (для честного сравнения) +v3_seen = pl.concat([pl.read_parquet(V3_CHUNK_MAP).select("ид_канала_данных"), + pl.read_parquet(V2_CHUNK_MAP).filter(pl.col("split") != "test").select("ид_канала_данных")]) +ids = pl.concat( + pl.scan_parquet(RUN / f"data/ext-journal-{y}.parquet").select(pl.col("ид_канала_данных").cast(pl.Int64, strict=False)).unique().collect() + for y in (2024, 2025, 2026) +).drop_nulls().unique().sort("ид_канала_данных").sample(fraction=1.0, shuffle=True, seed=69) +prod_seen = ids["ид_канала_данных"][: int(ids.height * 0.9)] # сплит prod: seed 69, 90% train (как tests/backtest) +SEEN = set(v3_seen["ид_канала_данных"].cast(pl.Int64).to_list()) | set(prod_seen.to_list()) + +K = ["new", "v3", "prod", "rule", "rule24"] +CLS = ["FIRE", "FLOOD", "GAS", "INTRUSION"] +blind, june = predict("blind"), predict("june") +rows = [] +unseen = pl.col("ch").is_in(list(SEEN)).not_() +for subset, d0 in [("a_blind_channels", blind), ("a_blind_unseen_by_v3_prod", blind.filter(unseen)), + ("b_june_all", june), ("b_june_unseen_by_v3_prod", june.filter(unseen)), + ("b_june_blind_channels", june.filter(pl.col("blind_ch")))]: + for n, d in [(c, d0.filter(pl.col("cls") == c)) for c in CLS] + [("ALL_rows", d0)]: + rows.append(dict(subset=subset, cls=n, rows=d.height, channels=d["ch"].n_unique(), pos=int(d["y"].sum()), **{k: ap(d, f"p_{k}") for k in K})) + per = [x for x in rows if x["subset"] == subset and x["cls"] in CLS] + rows.append(dict(subset=subset, cls="MACRO", rows=None, channels=None, pos=None, + **{k: round(float(np.mean([x[k] for x in per if x[k] is not None])), 3) for k in K})) +res = pl.DataFrame(rows) +res.write_csv(RUN / "eval_blind.csv") +pl.Config.set_tbl_rows(60); pl.Config.set_tbl_cols(20); pl.Config.set_tbl_width_chars(220) +print(res) +print("EVAL_DONE", round(time.time() - t0), flush=True) diff --git a/retraining/notebooks/incident4/pipeline_v4.sh b/retraining/notebooks/incident4/pipeline_v4.sh new file mode 100755 index 0000000..e57c697 --- /dev/null +++ b/retraining/notebooks/incident4/pipeline_v4.sh @@ -0,0 +1,22 @@ +#!/bin/bash +# incident4_v4: nb1 -> nb2 -> nb3 -> обучение сегментами -> слепая оценка (один раз). +# Каталог прогона — аргумент или RUN_DIR, по умолчанию retraining/artifacts/runs/incident4_v4. +# В нём заранее лежит data/ (журналы, справочники, погода); см. README.md. +HERE=$(cd "$(dirname "$0")" && pwd) +RUN=${1:-${RUN_DIR:-$HERE/../../artifacts/runs/incident4_v4}} +mkdir -p "$RUN/survival_optuna" && RUN=$(cd "$RUN" && pwd) && cd "$RUN" || exit 1 +PY=${PY:-python} +cp "$HERE"/[123]_*.ipynb . +[ -f survival_optuna/best_params.json ] || cp "$HERE/best_params.json" survival_optuna/ +run_nbs() { + for nb in "$@"; do + echo ">>> $nb" + $PY -m jupyter nbconvert --to notebook --execute --inplace --ExecutePreprocessor.timeout=None "$nb" || return 1 + done +} +run_nbs 1_prepare_dataset.ipynb 2_create_train_test_datasets_V2_MAIN.ipynb 3_feature_selection.ipynb \ + && $PY "$HERE/train_v4.py" "$RUN" --smoke \ + && rm -f survival_optuna/model_*.cbm survival_optuna/val_by_trees.csv survival_optuna/best_model.cbm \ + && $PY "$HERE/train_v4.py" "$RUN" \ + && $PY "$HERE/eval_v4.py" "$RUN" +echo "PIPELINE_EXIT $?" diff --git a/retraining/notebooks/incident4/train_v4.py b/retraining/notebooks/incident4/train_v4.py new file mode 100644 index 0000000..831e9f1 --- /dev/null +++ b/retraining/notebooks/incident4/train_v4.py @@ -0,0 +1,105 @@ +"""incident4_v4: весь прореженный train (80% каналов, до июня 2026), признаки выбранного шага nb3, параметры v1. +Обучаем сегментами по SEG деревьев (init_model), после каждого — macro PR-AUC на val; стоп без улучшения PATIENCE сегментов.""" +import gc, json, sys, time +from pathlib import Path +import numpy as np, polars as pl +from catboost import CatBoostRegressor, Pool, FeaturesData +from scipy.special import ndtr +from sklearn.metrics import average_precision_score + +RUN = Path(sys.argv[1]).resolve() +SMOKE = "--smoke" in sys.argv +H, SEG, MAX_TREES, PATIENCE = 24.0, 100, 1500, 3 +if SMOKE: + SEG, MAX_TREES = 10, 30 +TO_DROP = ["ид_события", "ид_канала_данных", "дата", "время", "datetime"] + + +def selected(): + # как в nb5: все признаки минус удалённые до выбранного шага nb3 включительно + step = json.loads((RUN / "feature_selection_shap_rfe/selected_step.json").read_text())["step"] + hist = pl.read_csv(RUN / "feature_selection_shap_rfe/history.csv") + removed = {f for r in hist.filter(pl.col("step") <= step)["removed_features"].drop_nulls() for f in r.split(" | ") if f} + sch = pl.scan_parquet(RUN / "data/final_train/chunk_00000.parquet").drop(TO_DROP).collect_schema() + fs = [c for c in sch.names() if c not in ("target_lower", "target_upper") and c not in removed] + assert len(fs) == hist.filter(pl.col("step") == step)["n_features"].item() + out = {"step": step, "features": fs, "cat_features": [c for c in fs if sch[c] in (pl.String, pl.Categorical)]} + (RUN / "selected_features.json").write_text(json.dumps(out, ensure_ascii=False, indent=1)) + return out + + +feat = selected() +FEATS, CATS = feat["features"], feat["cat_features"] +NUMS = [f for f in FEATS if f not in CATS] +inc = json.load(open(RUN / "data/incident_by_sensor.json")) +CLS = sorted(set(inc.values())) +params = json.load(open(RUN / "survival_optuna/best_params.json")) +scale = float(params.pop("scale")); params.pop("iterations", None) +OUT = RUN / "survival_optuna" +t0 = time.time() +log = lambda *a: print(f"[{time.strftime('%H:%M:%S')} t={time.time()-t0:.0f}s]", *a, flush=True) + + +def load(files): + # числа — сразу в заранее выделенный float32, категории — строками (null -> "null", как в v2) + n = sum(pl.scan_parquet(f).select(pl.len()).collect().item() for f in files) + X = np.empty((n, len(NUMS)), dtype=np.float32) + C = np.empty((n, len(CATS)), dtype=object) + y = np.empty((n, 2), dtype=np.float64); sensor = np.empty(n, dtype=object); i = 0 + for f in files: + d = pl.read_parquet(f, columns=[*dict.fromkeys(FEATS + ["target_lower", "target_upper", "тип_датчика"])]) + k = d.height + if k == 0: + continue + X[i:i + k] = d.select(pl.col(NUMS).cast(pl.Float32)).to_numpy() + for j, c in enumerate(CATS): + s = d[c].cast(pl.String).fill_null("null").cast(pl.Categorical) + C[i:i + k, j] = np.array(s.cat.get_categories().to_list(), dtype=object)[s.to_physical().to_numpy()] + y[i:i + k] = d.select("target_lower", "target_upper").to_numpy() + sensor[i:i + k] = d["тип_датчика"].to_numpy(); i += k + del d + assert i == n + pool = Pool(FeaturesData(num_feature_data=X, cat_feature_data=C, num_feature_names=NUMS, cat_feature_names=CATS), label=y) + ybin = ((y[:, 1] != -1) & (y[:, 0] <= H)).astype(np.int8) + del X, C, y; gc.collect() + return pool, ybin, sensor + + +def macro(y, p, sensor): + cls = pl.Series(sensor).replace_strict(inc, default=None).to_numpy() + per = {c: float(average_precision_score(y[cls == c], p[cls == c])) for c in CLS if y[cls == c].any()} + return float(np.mean(list(per.values()))), per + + +train_files = sorted((RUN / "data/final_train").glob("chunk_*.parquet")) +val_files = sorted((RUN / "data/final_val").glob("chunk_*.parquet")) +if SMOKE: + train_files, val_files = train_files[:3], val_files[:40] +train_pool, ytr, _ = load(train_files) +log("train rows", train_pool.num_row(), "pos", int(ytr.sum()), "files", len(train_files)) +train_pool.quantize(); gc.collect() +log("quantized") +val_pool, yval, sval = load(val_files) +log("val rows", val_pool.num_row(), "pos", int(yval.sum())) + +model, hist, best = None, [], (-1.0, 0) +while True: + m = CatBoostRegressor(loss_function=f"SurvivalAft:dist=Normal;scale={scale}", iterations=SEG, **params, + random_seed=42, verbose=max(SEG // 4, 1), allow_writing_files=False) + m.fit(train_pool, init_model=model) + model = m; n = model.tree_count_ + p = ndtr((np.log(H) - model.predict(val_pool)) / scale) + mac, per = macro(yval, p, sval) + hist.append(dict(trees=n, val_macro=mac, val_all=float(average_precision_score(yval, p)), **per)) + pl.DataFrame(hist).write_csv(OUT / "val_by_trees.csv") + model.save_model(str(OUT / f"model_{n}.cbm")) + log("trees", n, "val_macro", round(mac, 4), {k: round(v, 3) for k, v in per.items()}) + if mac > best[0]: + best = (mac, n) + if n - best[1] >= PATIENCE * SEG or n >= MAX_TREES: + break +log("best trees", best[1], "val_macro", round(best[0], 4)) +bm = CatBoostRegressor(); bm.load_model(str(OUT / f"model_{best[1]}.cbm")) +bm.save_model(str(OUT / "best_model.cbm")) +json.dump({"best_trees": best[1], "val_macro": best[0], "scale": scale}, open(OUT / "best_trees.json", "w")) +print("TRAIN_DONE", flush=True)