feat(ml): incident4 pipeline

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
+}