diff --git a/retraining/README.md b/retraining/README.md index 6bea2a3..db4b492 100644 --- a/retraining/README.md +++ b/retraining/README.md @@ -1,8 +1,12 @@ # Общий пайплайн обучения -Этот каталог содержит пять ноутбуков и `run_pipeline.py` для последовательной подготовки данных, разделения на train/val/test, отбора признаков, подбора параметров и обучения CatBoost. Скрипт не запускает облачную VM и не заменяет модель сервиса инференса. Описание относится к коду ветки `main`; экспериментальная автоматизация облачного запуска в неё не входит. +Для запуска на временной VM в Yandex Cloud см. [cloud/README.md](cloud/README.md). -Шаблоны в `notebooks/` не перезаписываются. Для каждого запуска создаются рабочие копии в `artifacts/runs//notebooks/`. В них подставляются годы входных событий и числа доступных чанков. Результат зависит от фактически переданных файлов, настроек и версии кода; пример из репозитория не заменяет полный набор событий. +Эта папка — самостоятельная точка повторного обучения модели на исходных и новых данных. Добавьте новые журналы событий в `data/incoming_events/` и запустите один скрипт: он объединит все записи, сформирует признаки, заново разобьёт устройства на train/val/test, выполнит feature selection, Optuna-тюнинг, обучит итоговую модель и посчитает метрики. + +Шаблоны в `notebooks/` — копии исходных тетрадок. Они **не перезаписываются**. Для каждого прогона скрипт создаёт рабочие копии в `artifacts/runs//notebooks/` и только там подставляет найденные годы и реально доступные чанки. По умолчанию feature selection и подбор параметров используют ограниченные выборки, а итоговая модель обучается на всех полученных train-чанках. Validation и test остаются отдельными выборками для оценки качества; финальная оценка охватывает все test-файлы. + +Результат зависит от фактически переданных файлов, настроек и версии кода; пример из репозитория не заменяет полный набор событий. Скрипт не заменяет модель сервиса инференса. ## Состав @@ -37,7 +41,7 @@ python run_pipeline.py ## Как добавить новые данные -1. Скопируйте один или несколько parquet-файлов в `data/incoming_events/`. Скрипт читает только файлы по `input.events_glob` из `config.yaml`; по умолчанию это `*.parquet`. Записи одного года объединяются. Если файл не должен участвовать в запуске, уберите его из входного каталога или настройте glob до запуска. +1. Скопируйте один или несколько parquet-файлов в `data/incoming_events/`. Скрипт читает только файлы по `input.events_glob` из `config.yaml` (по умолчанию `*.parquet`); записи одного года объединяются. Для полного прогона положите журналы в отдельный каталог и укажите его в `input.events_dir`, либо настройте `input.events_glob`, чтобы исключить `журнал_событий_пример.parquet`. Не удаляйте старые реальные файлы, если они тоже должны остаться в обучающей выборке. 2. Убедитесь, что у каждого файла есть колонки: `ид_события`, `ид_канала_данных`, `дата`, `время`, `значение_датчика`, `тревожное`. 3. При изменении справочников или погоды замените соответствующий файл в `data/reference/` либо поменяйте путь в `config.yaml`. 4. Запустите `python run_pipeline.py`. @@ -46,22 +50,22 @@ python run_pipeline.py ## Конфигурация -В `config.yaml` можно изменить каталоги данных, Jupyter kernel, таймаут и максимальное число чанков для этапов feature selection и tuning. Скрипт уменьшает лимит до числа существующих чанков; при нуле чанков этап завершится ошибкой. +В `config.yaml` можно изменить каталоги данных, Jupyter kernel, таймаут и число чанков для этапов feature selection, tuning и final training. Значение `all` у `final_training.train_chunks` и `final_training.val_chunks` означает все доступные чанки соответствующего сплита. Целочисленные лимиты автоматически ограничиваются числом существующих чанков. На test лимит не действует: финальный ноутбук оценивает все test-файлы. При нуле чанков этап завершится ошибкой. Новые данные должны быть в том же формате, что и пример. Дата должна приводиться к календарной дате; погодный CSV должен содержать колонку `time` и погодные поля из исходной тетрадки. -## Фактическое использование чанков в `main` +## Фактическое использование чанков | Этап | Train | Validation | Test | | --- | --- | --- | --- | | Отбор признаков | До 4 чанков | До 20 чанков | До 20 чанков | | Подбор параметров | До 8 чанков | До 80 чанков | До 20 чанков в шаблоне | -| Итоговое обучение | До 8 чанков | До 80 чанков | Все `chunk_*.parquet` для итоговой оценки | +| Итоговое обучение | Все (`final_training.train_chunks: all`) | Все (`final_training.val_chunks: all`) | Все `chunk_*.parquet` для итоговой оценки | -Числа заданы в `notebook_limits` и ограничиваются фактическим количеством файлов. Итоговый ноутбук получает лимиты train/val из секции `tuning`; поэтому текущий код `main` **не гарантирует обучение на всех train-чанках**. Все test-файлы читаются в итоговой оценке пакетами. Test не используется для обучения итоговой модели, но ноутбук отбора признаков считает `test_pr_auc` по кандидатам. Поэтому итоговую test-метрику нельзя считать оценкой на полностью закрытой выборке. +Числа заданы в `notebook_limits` и ограничиваются фактическим количеством файлов. Итоговый ноутбук берёт лимиты train/val из секции `final_training`; если её нет в конфиге, используются лимиты `tuning`. Все test-файлы читаются в итоговой оценке пакетами. Test не используется для обучения итоговой модели, но ноутбук отбора признаков считает `test_pr_auc` по кандидатам. Поэтому итоговую test-метрику нельзя считать оценкой на полностью закрытой выборке. ## Проверка результата -Прогон считается завершённым после успешного выхода `run_pipeline.py` и проверки `best_model.cbm`, `scores_final.csv`, `best_params.json` и выполненных ноутбуков в одном каталоге запуска. В выводе последнего ноутбука сверяйте число найденных test-файлов с входным набором. Для сравнения с прежней моделью и правилами «тревога вчера» и «не менее десяти тревог вчера» нужны отдельные расчёты на тех же строках и горизонте прогноза; текущий скрипт `main` их не выполняет. Автоматической выкладки модели нет. +Прогон считается завершённым после успешного выхода `run_pipeline.py` и проверки `best_model.cbm`, `scores_final.csv`, `best_params.json` и выполненных ноутбуков в одном каталоге запуска. В выводе последнего ноутбука сверяйте число найденных test-файлов с входным набором. Для сравнения с прежней моделью и правилами «тревога вчера» и «не менее десяти тревог вчера» используйте `compare_old_model.py` и `baseline_previous_day.py` — они считают метрики на тех же test-строках и горизонте прогноза. Автоматической выкладки модели нет. Для воспроизводимости фиксируйте commit, `config.yaml`, перечень Parquet и их SHA-256, число train/val/test-чанков и итоговые артефакты. Данные и секреты размещайте вне Git, если они не предназначены для публикации. diff --git a/retraining/baseline_previous_day.py b/retraining/baseline_previous_day.py new file mode 100644 index 0000000..4466ef4 --- /dev/null +++ b/retraining/baseline_previous_day.py @@ -0,0 +1,78 @@ +#!/usr/bin/env python3 +"""Evaluate previous-day alarm rules on the exact final test event rows.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +import numpy as np +import polars as pl +from sklearn.metrics import average_precision_score + + +def score(y_true: np.ndarray, predictions: np.ndarray) -> dict: + predicted = predictions.astype(bool) + positive = y_true.astype(bool) + selected = int(predicted.sum()) + hits = int(np.count_nonzero(predicted & positive)) + return { + "pr_auc": float(average_precision_score(y_true, predictions)), + "precision": hits / selected if selected else None, + "recall": hits / int(positive.sum()) if positive.any() else None, + "alert_rate": selected / len(y_true), + "alerts": selected, + "true_positives": hits, + } + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--run-dir", type=Path, required=True) + args = parser.parse_args() + data = args.run_dir / "data" + test = pl.scan_parquet(str(data / "final_test" / "chunk_*.parquet")).select( + "ид_события", "ид_канала_данных", "дата", "target_lower", "target_upper" + ) + test_ids = test.select("ид_канала_данных").unique() + alarms = ( + pl.scan_parquet(str(data / "ext-journal-*.parquet")) + .select("ид_канала_данных", "дата", "тревожное") + .join(test_ids, on="ид_канала_данных", how="semi") + .with_columns(pl.col("дата").cast(pl.Date)) + .group_by("ид_канала_данных", "дата") + .agg(pl.col("тревожное").eq("t").sum().alias("alarm_count")) + .with_columns((pl.col("дата") + pl.duration(days=1)).alias("дата")) + ) + evaluated = ( + test.join(alarms, on=["ид_канала_данных", "дата"], how="left") + .select( + "ид_события", + ((pl.col("target_upper") != -1) & (pl.col("target_lower") <= 24)) + .cast(pl.Int8) + .alias("target"), + pl.col("alarm_count").fill_null(0).alias("alarm_count_yesterday"), + ) + .collect(engine="streaming") + ) + y_true = evaluated["target"].to_numpy() + yesterday = evaluated["alarm_count_yesterday"].to_numpy() + result = { + "test_rows": len(y_true), + "test_positive_rate": float(y_true.mean()), + "unit": "event row", + "horizon_hours": 24, + "rules": { + "any_alarm_yesterday": score(y_true, (yesterday >= 1).astype(np.int8)), + "ten_alarms_yesterday": score(y_true, (yesterday >= 10).astype(np.int8)), + }, + } + destination = args.run_dir / "survival_optuna" / "baseline_previous_day.json" + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8") + print(json.dumps(result, ensure_ascii=False, indent=2), flush=True) + + +if __name__ == "__main__": + main() diff --git a/retraining/cloud/README.md b/retraining/cloud/README.md new file mode 100644 index 0000000..ae2bd5c --- /dev/null +++ b/retraining/cloud/README.md @@ -0,0 +1,45 @@ +# Обучение в Yandex Cloud + +`job.py` выполняет три отдельные операции: кладёт снимок кода и Parquet в закрытый Object Storage, создаёт временную VM через Terraform/OpenTofu и читает статус. На VM `bootstrap.py` монтирует отдельный диск, а `runner.py` выполняет пять ноутбуков, сравнивает результат с правилом «тревога вчера» и старой моделью, сохраняет артефакты в `results//` и выключает VM. Terraform state остаётся на машине запуска вне Git. После завершения VM и диск нужно удалить командой `destroy`: выключение VM не удаляет диск. + +## Подготовка + +1. Установить Python с зависимостью `polars` из `retraining/requirements.txt`, `yc` с доступом к Yandex Cloud и `tofu` (или Terraform). Авторизовать `yc` для загрузки в Object Storage и создания IAM-токена. Сервисный аккаунт VM должен читать `jobs/` и писать `results/` в приватном бакете. +2. Подготовить Parquet с колонками `ид_события`, `ид_канала_данных`, `дата`, `время`, `значение_датчика`, `тревожное`. Для стенда есть `../export_stand_events.py`; пароль SSH передаётся путём к файлу вне Git. Справочники и погода берутся из снимка репозитория. +3. Скопировать `cloud-settings.example.json` в файл вне репозитория и заполнить ID существующих сети, группы безопасности, сервисного аккаунта и бакета. Группа безопасности должна разрешать исходящий HTTPS; входящие порты обучению не нужны. Квота должна вмещать заданные `cores` и `memory_gib`. + +## Один запуск + +Из корня репозитория: + +```bash +python retraining/cloud/job.py launch \ + --run-id full-20260927 \ + --settings /private/path/cloud-settings.json \ + --state-dir /private/path/terraform-state \ + --events-dir /private/path/events-parquet +``` + +Команда проверяет схему и считает SHA-256 всех файлов, загружает их в `jobs//`, затем выполняет `tofu apply`. При повторном использовании **того же** уже загруженного набора данных можно добавить `--reuse-events-prefix events/r3`; файлы всё равно проверяются по SHA-256 на VM. Для нового набора этот флаг не использовать. Версия кода фиксируется в отдельном архиве каждого запуска. + +Статус и уборка: + +```bash +python retraining/cloud/job.py status --run-id full-20260927 \ + --settings /private/path/cloud-settings.json --state-dir /private/path/terraform-state +python retraining/cloud/job.py destroy --run-id full-20260927 \ + --settings /private/path/cloud-settings.json --state-dir /private/path/terraform-state +``` + +Окончание — `phase: completed` в `results//status.json`. `phase: failed` означает, что нужно смотреть `run.log`, `bootstrap-status.json` и выполненные тетрадки в том же префиксе. Перед удалением VM проверьте, что результаты в бакете появились. Модель не заменяется автоматически: сначала сравнить PR-AUC и операционные метрики с правилами «≥1 тревога вчера» и «≥10 тревог вчера» на одном test-срезе. + +## API для кнопки + +`serve.py` предоставляет `POST /jobs` с `{"runId":"..."}` и `GET /jobs/` для Java-бэкенда. Он запускается на отдельном управляющем хосте с теми же `yc`, `tofu`, доступом к каталогу Parquet и приватному state. По умолчанию слушает только `127.0.0.1:8099`; при удалённом доступе нужен закрытый канал до хоста. Установите длинный случайный `ML_TRAINING_API_TOKEN` на этом хосте и тот же токен как `ML_TRAINING_ORCHESTRATOR_TOKEN` в бэкенде. Адрес задаётся `ML_TRAINING_ORCHESTRATOR_URL`. Без этих переменных Java-ручка отвечает 503. Пример файла настроек — `orchestrator.example.json`. + +```bash +ML_TRAINING_API_TOKEN='' python retraining/cloud/serve.py \ + --config /private/path/orchestrator.json +``` + +Все файлы данных, ключи и Terraform state должны оставаться вне Git. API использует уже подготовленный каталог Parquet на управляющем хосте; обновлять его перед повторным обучением нужно отдельно. Конфигурация не разворачивает управляющий хост и не меняет прод без явной настройки. diff --git a/retraining/cloud/bootstrap.py b/retraining/cloud/bootstrap.py new file mode 100644 index 0000000..b55122b --- /dev/null +++ b/retraining/cloud/bootstrap.py @@ -0,0 +1,75 @@ +#!/usr/bin/env python3 +"""Set up the training disk and launch the isolated Yandex Cloud runner.""" + +from __future__ import annotations + +import json +import subprocess +import sys +import time +import traceback +import urllib.request +from pathlib import Path + + +CONFIG = json.loads(Path("/etc/ml-job.json").read_text()) +BUCKET = CONFIG["bucket"] +RUN_ID = CONFIG["run_id"] +DISK = Path("/dev/disk/by-id/virtio-ml-data") +WORK = Path("/work") +METADATA_URL = "http://169.254.169.254/computeMetadata/v1/instance/service-accounts/default/token" + + +def token() -> str: + request = urllib.request.Request(METADATA_URL, headers={"Metadata-Flavor": "Google"}) + with urllib.request.urlopen(request, timeout=20) as response: + return json.load(response)["access_token"] + + +def s3(key: str, method: str, data: bytes | None = None) -> bytes: + headers = {"Authorization": f"Bearer {token()}"} + request = urllib.request.Request( + f"https://storage.yandexcloud.net/{BUCKET}/{key}", + headers=headers, method=method, data=data, + ) + with urllib.request.urlopen(request, timeout=120) as response: + return response.read() + + +def main() -> None: + # Guard against an OOM kill of both the notebook and this bootstrap process. + subprocess.run(["shutdown", "-h", "+4320"], check=True) + print("Waiting for training disk", flush=True) + for _ in range(120): + if DISK.exists(): + break + time.sleep(1) + else: + raise RuntimeError(f"Training disk not found: {DISK}") + + if subprocess.run(["blkid", str(DISK)], stdout=subprocess.DEVNULL).returncode: + subprocess.run(["mkfs.ext4", "-F", str(DISK)], check=True) + WORK.mkdir(exist_ok=True) + subprocess.run(["mount", str(DISK), str(WORK)], check=True) + print("Training disk mounted", flush=True) + + subprocess.run(["apt-get", "update", "-qq"], check=True, timeout=1800) + subprocess.run(["apt-get", "install", "-y", "-qq", "python3-venv"], check=True, timeout=1800) + runner = WORK / "ml-cloud-runner.py" + runner.write_bytes(s3(f"jobs/{RUN_ID}/runner.py", "GET")) + print("Launching ML runner", flush=True) + subprocess.run([sys.executable, "-u", str(runner)], check=True) + + +if __name__ == "__main__": + result = {"ok": True} + try: + main() + except Exception as error: + result = {"ok": False, "error": str(error), "traceback": traceback.format_exc()[-3000:]} + print(result["traceback"], flush=True) + try: + s3(f"results/{RUN_ID}/bootstrap-status.json", "PUT", json.dumps(result).encode()) + except Exception as error: + print(f"Could not upload bootstrap status: {error}", flush=True) + subprocess.run(["shutdown", "-h", "now"], check=False) diff --git a/retraining/cloud/cloud-settings.example.json b/retraining/cloud/cloud-settings.example.json new file mode 100644 index 0000000..6cdcebc --- /dev/null +++ b/retraining/cloud/cloud-settings.example.json @@ -0,0 +1,11 @@ +{ + "folder_id": "", + "zone": "ru-central1-a", + "subnet_id": "", + "security_group_id": "", + "service_account_id": "", + "bucket_name": "", + "cores": 16, + "memory_gib": 128, + "work_disk_gib": 450 +} diff --git a/retraining/cloud/job.py b/retraining/cloud/job.py new file mode 100644 index 0000000..5341161 --- /dev/null +++ b/retraining/cloud/job.py @@ -0,0 +1,101 @@ +#!/usr/bin/env python3 +"""Stage, provision, inspect and remove an isolated cloud training job.""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import subprocess +import sys +import tempfile +from pathlib import Path + + +HERE = Path(__file__).resolve().parent +TERRAFORM = HERE / "terraform" +RUN_ID = re.compile(r"[a-z0-9][a-z0-9-]{2,50}\Z") + + +def command(args: list[str], *, env: dict[str, str] | None = None) -> str: + result = subprocess.run(args, check=True, text=True, stdout=subprocess.PIPE, env=env) + return result.stdout.strip() + + +def settings(path: Path) -> dict: + config = json.loads(path.read_text(encoding="utf-8")) + required = {"folder_id", "subnet_id", "security_group_id", "service_account_id", "bucket_name"} + missing = required - config.keys() + if missing: + raise ValueError(f"Missing cloud settings: {', '.join(sorted(missing))}") + return config + + +def tofu(args: argparse.Namespace, action: str, config: dict) -> None: + state_dir = args.state_dir.resolve() + state_dir.mkdir(parents=True, exist_ok=True) + state_file = state_dir / f"{args.run_id}.tfstate" + env = dict(os.environ) + env["YC_TOKEN"] = command([args.yc, "iam", "create-token"]) + env["TF_DATA_DIR"] = str(state_dir / f"{args.run_id}.terraform") + env.setdefault("TF_CLI_CONFIG_FILE", str(HERE / "tofurc")) + base = [args.tofu, f"-chdir={TERRAFORM}"] + command(base + ["init", "-backend=false", "-input=false"], env=env) + variables = dict(config, run_id=args.run_id, + instance_name=f"ml-training-{args.run_id}") + allowed = {"folder_id", "zone", "subnet_id", "security_group_id", "service_account_id", + "bucket_name", "run_id", "instance_name", "cores", "memory_gib", "work_disk_gib"} + flags = [f"-var={name}={value}" for name, value in variables.items() if name in allowed] + subprocess.run(base + [action, "-input=false", "-auto-approve", f"-state={state_file}", *flags], + check=True, env=env) + + +def stage(args: argparse.Namespace, config: dict) -> None: + cmd = [sys.executable, str(HERE / "prepare_job.py"), "--events-dir", str(args.events_dir), + "--bucket", config["bucket_name"], "--run-id", args.run_id, "--yc", args.yc] + if args.reuse_events_prefix: + cmd += ["--reuse-events-prefix", args.reuse_events_prefix] + subprocess.run(cmd, check=True) + + +def status(args: argparse.Namespace, config: dict) -> None: + with tempfile.TemporaryDirectory(prefix="ml-job-status-") as temporary: + local = Path(temporary) / "status.json" + command([args.yc, "storage", "s3", "cp", + f"s3://{config['bucket_name']}/results/{args.run_id}/status.json", str(local)]) + print(local.read_text(encoding="utf-8")) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("action", choices=["launch", "status", "destroy"]) + parser.add_argument("--run-id", required=True) + parser.add_argument("--settings", type=Path, required=True, help="Cloud IDs JSON kept outside Git") + parser.add_argument("--state-dir", type=Path, required=True, help="Private directory for Terraform state") + parser.add_argument("--events-dir", type=Path, help="Parquet export; required for launch") + parser.add_argument("--reuse-events-prefix", help="Existing S3 data prefix for an unchanged export") + parser.add_argument("--yc", default="yc") + parser.add_argument("--tofu", default="tofu") + args = parser.parse_args() + if not RUN_ID.fullmatch(args.run_id): + parser.error("Invalid run-id") + config = settings(args.settings) + if args.action == "launch": + if args.events_dir is None: + parser.error("--events-dir is required for launch") + if (args.state_dir / f"{args.run_id}.tfstate").exists(): + parser.error("Terraform state already exists for this run-id") + stage(args, config) + tofu(args, "apply", config) + elif args.action == "status": + status(args, config) + else: + if not (args.state_dir / f"{args.run_id}.tfstate").exists(): + parser.error("Terraform state not found; refusing destroy") + tofu(args, "destroy", config) + (args.state_dir / f"{args.run_id}.destroyed").write_text("destroyed\n", encoding="utf-8") + + +if __name__ == "__main__": + main() diff --git a/retraining/cloud/orchestrator.example.json b/retraining/cloud/orchestrator.example.json new file mode 100644 index 0000000..e8cfbc8 --- /dev/null +++ b/retraining/cloud/orchestrator.example.json @@ -0,0 +1,7 @@ +{ + "settings": "/private/path/cloud-settings.json", + "state_dir": "/private/path/terraform-state", + "events_dir": "/private/path/events-parquet", + "yc": "yc", + "tofu": "tofu" +} diff --git a/retraining/cloud/prepare_job.py b/retraining/cloud/prepare_job.py new file mode 100644 index 0000000..9fa814e --- /dev/null +++ b/retraining/cloud/prepare_job.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 +"""Stage a versioned training job in a private Yandex Object Storage bucket.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import re +import subprocess +import tarfile +import tempfile +from pathlib import Path + +import polars as pl + + +REPO = Path(__file__).resolve().parents[2] +REQUIRED = {"ид_события", "ид_канала_данных", "дата", "время", "значение_датчика", "тревожное"} + + +def digest(path: Path) -> str: + sha = hashlib.sha256() + with path.open("rb") as source: + for block in iter(lambda: source.read(1024 * 1024), b""): + sha.update(block) + return sha.hexdigest() + + +def include(entry: tarfile.TarInfo) -> tarfile.TarInfo | None: + parts = Path(entry.name).parts + if parts and parts[0] == ".git": + return None + if "__pycache__" in parts or ".DS_Store" in parts or ".terraform" in parts: + return None + if any(part in {".env", "terraform.tfstate", "terraform.tfstate.backup"} + or part.endswith((".tfvars", ".key", ".pem")) for part in parts): + return None + if parts[:2] == ("retraining", "artifacts"): + return None + if parts[:3] == ("retraining", "data", "incoming_events"): + return None + return entry + + +def upload(yc: str, path: Path, bucket: str, key: str) -> None: + subprocess.run([yc, "storage", "s3", "cp", str(path), f"s3://{bucket}/{key}"], check=True) + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--events-dir", type=Path, required=True) + parser.add_argument("--bucket", required=True) + parser.add_argument("--run-id", required=True) + parser.add_argument("--yc", default="yc", help="Path to authenticated Yandex Cloud CLI") + parser.add_argument("--reuse-events-prefix", help="Existing S3 prefix with matching files") + parser.add_argument("--dry-run", action="store_true") + args = parser.parse_args() + + if not re.fullmatch(r"[a-z0-9][a-z0-9-]{2,50}", args.run_id): + parser.error("run-id must contain 3-51 lowercase letters, digits or hyphens") + files = sorted(args.events_dir.glob("*.parquet")) + if not files: + parser.error("No Parquet event files found") + if any(path.name == "журнал_событий_пример.parquet" for path in files): + parser.error("Do not include the example journal in a full run") + + prefix = f"jobs/{args.run_id}" + events_prefix = args.reuse_events_prefix or f"{prefix}/events" + with tempfile.TemporaryDirectory(prefix="ml-training-job-") as temporary: + scratch = Path(temporary) + archive = scratch / "ml-code.tar.gz" + with tarfile.open(archive, "w:gz") as tar: + for path in sorted(REPO.iterdir()): + tar.add(path, arcname=path.name, filter=include) + entries = [] + for path in files: + scan = pl.scan_parquet(path) + missing = REQUIRED - set(scan.collect_schema().names()) + if missing: + raise ValueError(f"{path}: missing columns {sorted(missing)}") + entries.append({ + "name": path.name, + "rows": scan.select(pl.len()).collect().item(), + "bytes": path.stat().st_size, + "sha256": digest(path), + }) + manifest = { + "run_id": args.run_id, + "events_prefix": events_prefix, + "code_sha256": digest(archive), + "total_rows": sum(entry["rows"] for entry in entries), + "files": entries, + } + manifest_path = scratch / "manifest.json" + manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8") + if not args.dry_run: + if not args.reuse_events_prefix: + for path in files: + upload(args.yc, path, args.bucket, f"{events_prefix}/{path.name}") + upload(args.yc, archive, args.bucket, f"{prefix}/ml-code.tar.gz") + upload(args.yc, Path(__file__).with_name("runner.py"), args.bucket, f"{prefix}/runner.py") + upload(args.yc, manifest_path, args.bucket, f"{prefix}/manifest.json") + print(json.dumps({ + "run_id": args.run_id, + "bucket": args.bucket, + "manifest_key": f"{prefix}/manifest.json", + "events_prefix": events_prefix, + "total_rows": manifest["total_rows"], + "files": len(files), + "uploaded": not args.dry_run, + }, ensure_ascii=False)) + + +if __name__ == "__main__": + main() diff --git a/retraining/cloud/runner.py b/retraining/cloud/runner.py new file mode 100644 index 0000000..5670cc1 --- /dev/null +++ b/retraining/cloud/runner.py @@ -0,0 +1,227 @@ +#!/usr/bin/env python3 +"""Run one isolated retraining job on a Yandex Compute Cloud VM.""" + +from __future__ import annotations + +import hashlib +import json +import os +import shutil +import signal +import subprocess +import sys +import tarfile +import threading +import time +import traceback +import urllib.request +from datetime import datetime, timezone +from pathlib import Path + + +CONFIG = json.loads(Path("/etc/ml-job.json").read_text()) +BUCKET = CONFIG["bucket"] +RUN_ID = CONFIG["run_id"] +JOB_PREFIX = f"jobs/{RUN_ID}" +ROOT = Path("/work") +REPO = ROOT / "ml" +TRAINING = REPO / "retraining" +RESULTS = f"results/{RUN_ID}" +METADATA_URL = "http://169.254.169.254/computeMetadata/v1/instance/service-accounts/default/token" +BASE_URL = f"https://storage.yandexcloud.net/{BUCKET}/" +MAX_TRAINING_SECONDS = 68 * 3600 + +state = {"phase": "starting", "started_at": datetime.now(timezone.utc).isoformat()} +state_lock = threading.Lock() +stop_reporter = threading.Event() + + +def token() -> str: + request = urllib.request.Request(METADATA_URL, headers={"Metadata-Flavor": "Google"}) + with urllib.request.urlopen(request, timeout=20) as response: + return json.load(response)["access_token"] + + +def request_object(key: str, method: str, data: bytes | None = None): + headers = {"Authorization": f"Bearer {token()}"} + if data is not None: + headers["Content-Type"] = "application/octet-stream" + request = urllib.request.Request(BASE_URL + key, data=data, headers=headers, method=method) + return urllib.request.urlopen(request, timeout=120) + + +def download(key: str, destination: Path, expected_sha256: str | None = None) -> None: + destination.parent.mkdir(parents=True, exist_ok=True) + temporary = destination.with_suffix(destination.suffix + ".part") + digest = hashlib.sha256() + with request_object(key, "GET") as response, temporary.open("wb") as output: + while block := response.read(1024 * 1024): + output.write(block) + digest.update(block) + if expected_sha256 and digest.hexdigest() != expected_sha256: + temporary.unlink(missing_ok=True) + raise ValueError(f"Checksum mismatch: {key}") + os.replace(temporary, destination) + + +def upload(key: str, data: bytes) -> None: + with request_object(key, "PUT", data) as response: + response.read() + + +def upload_file(key: str, path: Path) -> None: + # Final model and log files fit within the single PUT object limit. + if path.stat().st_size > 1024 * 1024 * 1024: + raise ValueError(f"Artifact is larger than 1 GiB: {path}") + upload(key, path.read_bytes()) + + +def metrics() -> dict: + result = {} + try: + lines = Path("/proc/meminfo").read_text().splitlines() + memory = {line.split(":")[0]: int(line.split()[1]) for line in lines} + result["ram_used_gib"] = round((memory["MemTotal"] - memory["MemAvailable"]) / 1048576, 2) + result["ram_total_gib"] = round(memory["MemTotal"] / 1048576, 2) + disk = shutil.disk_usage(ROOT) + result["disk_used_gib"] = round(disk.used / 1073741824, 2) + result["disk_total_gib"] = round(disk.total / 1073741824, 2) + log_file = ROOT / "run.log" + if log_file.exists(): + with log_file.open("rb") as handle: + handle.seek(max(0, log_file.stat().st_size - 4000)) + result["log_tail"] = handle.read().decode(errors="replace")[-3000:] + except Exception as error: + result["monitor_error"] = str(error) + return result + + +def report() -> None: + observed = metrics() + with state_lock: + for metric in ("ram_used_gib", "disk_used_gib"): + if metric in observed: + peak = f"peak_{metric}" + state[peak] = max(state.get(peak, 0), observed[metric]) + body = dict(state) + body.update(observed) + body["updated_at"] = datetime.now(timezone.utc).isoformat() + upload(f"{RESULTS}/status.json", json.dumps(body, ensure_ascii=False).encode()) + + +def reporter() -> None: + while not stop_reporter.is_set(): + try: + report() + except Exception as error: + print(f"Status upload failed: {error}", flush=True) + stop_reporter.wait(120) + + +def set_phase(phase: str, **details) -> None: + with state_lock: + state.update(phase=phase, **details) + print(f"{datetime.now(timezone.utc).isoformat()} {phase} {details}", flush=True) + report() + + +def run_logged(command: list[str], cwd: Path, log_name: str, timeout: int | None = None) -> None: + log_path = ROOT / log_name + with log_path.open("ab") as log: + process = subprocess.Popen(command, cwd=cwd, stdout=log, stderr=subprocess.STDOUT, start_new_session=True) + try: + code = process.wait(timeout=timeout) + except subprocess.TimeoutExpired: + os.killpg(process.pid, signal.SIGTERM) + try: + process.wait(timeout=300) + except subprocess.TimeoutExpired: + os.killpg(process.pid, signal.SIGKILL) + process.wait() + raise TimeoutError(f"Timed out after {timeout} seconds: {' '.join(command)}") + if code: + raise RuntimeError(f"Process exited with code {code}: {' '.join(command)}") + + +def upload_artifacts() -> None: + run_dir = TRAINING / "artifacts" / "runs" / RUN_ID + if run_dir.exists(): + for path in run_dir.rglob("*"): + if not path.is_file() or "data" in path.relative_to(run_dir).parts: + continue + key = f"{RESULTS}/artifacts/{path.relative_to(run_dir).as_posix()}" + try: + upload_file(key, path) + except Exception as error: + print(f"Artifact upload failed for {path}: {error}", flush=True) + for name in ("install.log", "run.log", "baseline.log", "comparison.log"): + path = ROOT / name + if path.exists(): + try: + upload_file(f"{RESULTS}/{name}", path) + except Exception as error: + print(f"Log upload failed for {name}: {error}", flush=True) + + +def main() -> None: + ROOT.mkdir(exist_ok=True) + thread = threading.Thread(target=reporter, daemon=True) + thread.start() + try: + set_phase("downloading_manifest") + manifest_path = ROOT / "manifest.json" + download(f"{JOB_PREFIX}/manifest.json", manifest_path) + manifest = json.loads(manifest_path.read_text()) + + set_phase("downloading_code") + archive = ROOT / "ml-code.tar.gz" + download(f"{JOB_PREFIX}/ml-code.tar.gz", archive, manifest["code_sha256"]) + REPO.mkdir(exist_ok=True) + with tarfile.open(archive, "r:gz") as tar: + tar.extractall(REPO, filter="data") + + set_phase("downloading_data") + incoming = TRAINING / "data" / "incoming_events" + incoming.mkdir(parents=True, exist_ok=True) + (incoming / "журнал_событий_пример.parquet").unlink(missing_ok=True) + for index, file in enumerate(manifest["files"], 1): + download(f"{manifest['events_prefix']}/{file['name']}", incoming / file["name"], file["sha256"]) + if index % 5 == 0 or index == len(manifest["files"]): + set_phase("downloading_data", files_downloaded=index, files_total=len(manifest["files"])) + + set_phase("installing_dependencies") + venv = ROOT / "venv" + run_logged([sys.executable, "-m", "venv", str(venv)], ROOT, "install.log") + run_logged([str(venv / "bin" / "python"), "-m", "pip", "install", "-r", "requirements.txt"], + TRAINING, "install.log", timeout=3600) + + set_phase("training", run_id=RUN_ID) + run_logged([str(venv / "bin" / "python"), "-u", "run_pipeline.py", "--run-id", RUN_ID], + TRAINING, "run.log", timeout=MAX_TRAINING_SECONDS) + set_phase("evaluating_baselines") + run_logged([ + str(venv / "bin" / "python"), "-u", "baseline_previous_day.py", + "--run-dir", str(TRAINING / "artifacts" / "runs" / RUN_ID), + ], TRAINING, "baseline.log", timeout=7200) + set_phase("comparing_models") + run_logged([ + str(venv / "bin" / "python"), "-u", "compare_old_model.py", + "--run-dir", str(TRAINING / "artifacts" / "runs" / RUN_ID), + "--old-model", str(REPO / "data" / "best_model.cbm"), + ], TRAINING, "comparison.log", timeout=7200) + set_phase("completed") + except Exception as error: + set_phase("failed", error=str(error), traceback=traceback.format_exc()[-3000:]) + raise + finally: + upload_artifacts() + stop_reporter.set() + thread.join(timeout=5) + try: + report() + except Exception as error: + print(f"Final status upload failed: {error}", flush=True) + + +if __name__ == "__main__": + main() diff --git a/retraining/cloud/serve.py b/retraining/cloud/serve.py new file mode 100644 index 0000000..8312d9d --- /dev/null +++ b/retraining/cloud/serve.py @@ -0,0 +1,133 @@ +#!/usr/bin/env python3 +"""Small private HTTP adapter for starting and inspecting cloud training jobs.""" + +from __future__ import annotations + +import argparse +import hmac +import json +import os +import re +import subprocess +import sys +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + + +HERE = Path(__file__).resolve().parent +RUN_ID = re.compile(r"[a-z0-9][a-z0-9-]{2,50}\Z") + + +def handler(config: dict, secret: str): + state_dir = Path(config["state_dir"]).resolve() + state_dir.mkdir(parents=True, exist_ok=True) + active: dict[str, subprocess.Popen] = {} + lock = threading.Lock() + + class Handler(BaseHTTPRequestHandler): + def respond(self, code: int, value: dict) -> None: + body = json.dumps(value, ensure_ascii=False).encode() + self.send_response(code) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def authorized(self) -> bool: + supplied = self.headers.get("Authorization", "") + if hmac.compare_digest(supplied, f"Bearer {secret}"): + return True + self.respond(401, {"error": "unauthorized"}) + return False + + def do_POST(self) -> None: + if not self.authorized(): + return + if self.path != "/jobs": + self.respond(404, {"error": "not found"}) + return + try: + size = int(self.headers.get("Content-Length", "0")) + if not 1 <= size <= 1024: + raise ValueError("Invalid body size") + run_id = json.loads(self.rfile.read(size))["runId"] + if not isinstance(run_id, str) or not RUN_ID.fullmatch(run_id): + raise ValueError("Invalid runId") + except (ValueError, KeyError, json.JSONDecodeError): + self.respond(400, {"error": "Invalid job request"}) + return + with lock: + if any(process.poll() is None for process in active.values()): + self.respond(409, {"error": "Another job is being provisioned"}) + return + if any(not path.with_suffix(".destroyed").exists() + for path in state_dir.glob("*.tfstate")): + self.respond(409, {"error": "A cloud VM still needs to be destroyed"}) + return + if (state_dir / f"{run_id}.log").exists(): + self.respond(409, {"error": "runId already exists"}) + return + command = [sys.executable, str(HERE / "job.py"), "launch", "--run-id", run_id, + "--settings", config["settings"], "--state-dir", str(state_dir), + "--events-dir", config["events_dir"]] + if config.get("reuse_events_prefix"): + command += ["--reuse-events-prefix", config["reuse_events_prefix"]] + if config.get("yc"): + command += ["--yc", config["yc"]] + if config.get("tofu"): + command += ["--tofu", config["tofu"]] + with (state_dir / f"{run_id}.log").open("x") as log: + active[run_id] = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT) + self.respond(202, {"runId": run_id, "phase": "provisioning"}) + + def do_GET(self) -> None: + if not self.authorized(): + return + match = re.fullmatch(r"/jobs/([a-z0-9][a-z0-9-]{2,50})", self.path) + if not match: + self.respond(404, {"error": "not found"}) + return + run_id = match.group(1) + if not (state_dir / f"{run_id}.log").exists(): + self.respond(404, {"error": "unknown runId"}) + return + command = [sys.executable, str(HERE / "job.py"), "status", "--run-id", run_id, + "--settings", config["settings"], "--state-dir", str(state_dir)] + if config.get("yc"): + command += ["--yc", config["yc"]] + result = subprocess.run(command, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + if result.returncode == 0: + self.respond(200, json.loads(result.stdout)) + return + process = active.get(run_id) + if process is None: + self.respond(200, {"runId": run_id, "phase": "status_unavailable", + "error": "No cloud status after orchestrator restart; inspect provisioning log"}) + elif process.poll() is None: + self.respond(200, {"runId": run_id, "phase": "provisioning"}) + else: + self.respond(200, {"runId": run_id, "phase": "failed", + "error": "Cloud job did not publish status; inspect provisioning log"}) + + return Handler + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--config", type=Path, required=True) + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=8099) + args = parser.parse_args() + secret = os.environ.get("ML_TRAINING_API_TOKEN", "") + if len(secret) < 32: + parser.error("ML_TRAINING_API_TOKEN must contain at least 32 characters") + config = json.loads(args.config.read_text(encoding="utf-8")) + for required in ("settings", "state_dir", "events_dir"): + if required not in config: + parser.error(f"Missing config key: {required}") + ThreadingHTTPServer((args.host, args.port), handler(config, secret)).serve_forever() + + +if __name__ == "__main__": + main() diff --git a/retraining/cloud/terraform/.gitignore b/retraining/cloud/terraform/.gitignore new file mode 100644 index 0000000..45a7581 --- /dev/null +++ b/retraining/cloud/terraform/.gitignore @@ -0,0 +1,6 @@ +.terraform/ +*.tfstate +*.tfstate.* +*.tfplan +*.tfvars +crash.log diff --git a/retraining/cloud/terraform/.terraform.lock.hcl b/retraining/cloud/terraform/.terraform.lock.hcl new file mode 100644 index 0000000..49fce0c --- /dev/null +++ b/retraining/cloud/terraform/.terraform.lock.hcl @@ -0,0 +1,11 @@ +# This file is maintained automatically by "tofu init". +# Manual edits may be lost in future updates. + +provider "registry.terraform.io/yandex-cloud/yandex" { + version = "0.215.0" + constraints = "0.215.0" + hashes = [ + "h1:D5dxRfp+hEos4MOTLb1riN455KNtdpNeP5MEtA9wX+o=", + "h1:XcpKtZSqo9Z2NyMvwXfDcFCzbZkGTF8q4q2otcLgsEs=", + ] +} diff --git a/retraining/cloud/terraform/cloud-init.yaml.tftpl b/retraining/cloud/terraform/cloud-init.yaml.tftpl new file mode 100644 index 0000000..69300ec --- /dev/null +++ b/retraining/cloud/terraform/cloud-init.yaml.tftpl @@ -0,0 +1,12 @@ +#cloud-config +write_files: + - path: /etc/ml-job.json + permissions: '0600' + encoding: b64 + content: ${config_b64} + - path: /usr/local/bin/ml-bootstrap.py + permissions: '0755' + encoding: b64 + content: ${bootstrap_b64} +runcmd: + - ['python3', '-u', '/usr/local/bin/ml-bootstrap.py'] diff --git a/retraining/cloud/terraform/main.tf b/retraining/cloud/terraform/main.tf new file mode 100644 index 0000000..eba36b5 --- /dev/null +++ b/retraining/cloud/terraform/main.tf @@ -0,0 +1,89 @@ +terraform { + required_version = ">= 1.6.0" + + required_providers { + yandex = { + source = "registry.terraform.io/yandex-cloud/yandex" + version = "0.215.0" + } + } +} + +provider "yandex" { + folder_id = var.folder_id + zone = var.zone +} + +data "yandex_compute_image" "ubuntu" { + family = "ubuntu-2404-lts" +} + +resource "yandex_compute_disk" "work" { + name = "${var.instance_name}-data" + zone = var.zone + type = "network-hdd" + size = var.work_disk_gib + + labels = { + task = "ml-training" + } +} + +resource "yandex_compute_instance" "training" { + name = var.instance_name + folder_id = var.folder_id + zone = var.zone + platform_id = "standard-v3" + service_account_id = var.service_account_id + + resources { + cores = var.cores + memory = var.memory_gib + core_fraction = 100 + } + + boot_disk { + auto_delete = true + + initialize_params { + image_id = data.yandex_compute_image.ubuntu.id + size = 30 + type = "network-ssd" + } + } + + secondary_disk { + disk_id = yandex_compute_disk.work.id + device_name = "ml-data" + auto_delete = false + } + + network_interface { + subnet_id = var.subnet_id + nat = true + security_group_ids = [var.security_group_id] + } + + metadata = { + user-data = templatefile("${path.module}/cloud-init.yaml.tftpl", { + bootstrap_b64 = filebase64("${path.module}/../bootstrap.py") + config_b64 = base64encode(jsonencode({ + bucket = var.bucket_name + run_id = var.run_id + })) + }) + } + + labels = { + task = "ml-training" + run = var.run_id + } +} + +output "instance_id" { + value = yandex_compute_instance.training.id +} + +output "instance_name" { + value = yandex_compute_instance.training.name +} diff --git a/retraining/cloud/terraform/variables.tf b/retraining/cloud/terraform/variables.tf new file mode 100644 index 0000000..6e26ebd --- /dev/null +++ b/retraining/cloud/terraform/variables.tf @@ -0,0 +1,57 @@ +variable "folder_id" { + type = string +} + +variable "zone" { + type = string + default = "ru-central1-a" +} + +variable "subnet_id" { + type = string +} + +variable "security_group_id" { + type = string +} + +variable "service_account_id" { + type = string +} + +variable "bucket_name" { + type = string +} + +variable "run_id" { + type = string + + validation { + condition = can(regex("^[a-z0-9][a-z0-9-]{2,50}$", var.run_id)) + error_message = "run_id must be 3-51 lowercase letters, digits or hyphens." + } +} + +variable "instance_name" { + type = string + + validation { + condition = can(regex("^[a-z][a-z0-9-]{2,62}$", var.instance_name)) + error_message = "instance_name must be a valid Yandex Compute Cloud VM name." + } +} + +variable "cores" { + type = number + default = 16 +} + +variable "memory_gib" { + type = number + default = 128 +} + +variable "work_disk_gib" { + type = number + default = 450 +} diff --git a/retraining/cloud/tofurc b/retraining/cloud/tofurc new file mode 100644 index 0000000..4c9c5e8 --- /dev/null +++ b/retraining/cloud/tofurc @@ -0,0 +1,9 @@ +provider_installation { + network_mirror { + url = "https://terraform-mirror.yandexcloud.net/" + include = ["registry.terraform.io/yandex-cloud/yandex"] + } + direct { + exclude = ["registry.terraform.io/yandex-cloud/yandex"] + } +} diff --git a/retraining/compare_old_model.py b/retraining/compare_old_model.py new file mode 100644 index 0000000..620d52c --- /dev/null +++ b/retraining/compare_old_model.py @@ -0,0 +1,80 @@ +#!/usr/bin/env python3 +"""Evaluate the deployed model on the exact test split of a new training run.""" + +from __future__ import annotations + +import argparse +import csv +import json +import re +from pathlib import Path + +import numpy as np +import polars as pl +from catboost import CatBoostRegressor, Pool +from scipy.special import ndtr +from sklearn.metrics import average_precision_score + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--run-dir", type=Path, required=True) + parser.add_argument("--old-model", type=Path, required=True) + args = parser.parse_args() + + model = CatBoostRegressor() + model.load_model(str(args.old_model)) + features = model.feature_names_ + if not features or any(not name for name in features): + raise ValueError("Old model does not contain usable feature names") + categorical = [features[i] for i in model.get_cat_feature_indices()] + loss = model.get_all_params()["loss_function"] + match = re.search(r"(?:^|;)scale=([0-9.eE+-]+)", loss) + if not match: + raise ValueError(f"SurvivalAft scale is missing from old model: {loss}") + scale = float(match.group(1)) + + labels = [] + probabilities = [] + test_files = sorted((args.run_dir / "data" / "final_test").glob("chunk_*.parquet")) + if not test_files: + raise FileNotFoundError("New run has no test chunks") + for index, path in enumerate(test_files, 1): + frame = pl.scan_parquet(path).select(features + ["target_lower", "target_upper"]) + if categorical: + frame = frame.with_columns(pl.col(categorical).fill_null("null")) + frame = frame.collect(engine="streaming") + labels.append( + ((frame["target_upper"] != -1) & (frame["target_lower"] <= 24.0)) + .cast(pl.Int8).to_numpy() + ) + pool = Pool(frame.select(features), cat_features=categorical) + prediction = model.predict(pool) + probabilities.append(ndtr((np.log(24.0) - prediction) / scale)) + if index % 10 == 0 or index == len(test_files): + print(f"Old model evaluated on {index}/{len(test_files)} test chunks", flush=True) + + y_test = np.concatenate(labels) + p_test = np.concatenate(probabilities) + with (args.run_dir / "survival_optuna" / "scores_final.csv").open(newline="") as file: + new_scores = next(csv.DictReader(file)) + if len(y_test) != int(new_scores["test_rows"]): + raise ValueError("Old and new model row counts differ") + old_pr_auc = float(average_precision_score(y_test, p_test)) + new_pr_auc = float(new_scores["test_pr_auc"]) + result = { + "test_rows": len(y_test), + "test_positive_rate": float(y_test.mean()), + "old_model_test_pr_auc": old_pr_auc, + "new_model_test_pr_auc": new_pr_auc, + "delta_pr_auc": new_pr_auc - old_pr_auc, + "old_model_scale": scale, + "test_chunks": len(test_files), + } + destination = args.run_dir / "survival_optuna" / "comparison_old_vs_new.json" + destination.write_text(json.dumps(result, ensure_ascii=False, indent=2)) + print(json.dumps(result, ensure_ascii=False, indent=2), flush=True) + + +if __name__ == "__main__": + main() diff --git a/retraining/config.yaml b/retraining/config.yaml index 4653d6c..ed47c42 100644 --- a/retraining/config.yaml +++ b/retraining/config.yaml @@ -24,3 +24,6 @@ notebook_limits: train_chunks: 8 val_chunks: 80 test_chunks: 20 + final_training: + train_chunks: all + val_chunks: all diff --git a/retraining/disk_pool.py b/retraining/disk_pool.py new file mode 100644 index 0000000..2068657 --- /dev/null +++ b/retraining/disk_pool.py @@ -0,0 +1,57 @@ +"""Build numeric CatBoost pools from Parquet without collecting every row in Python.""" + +from __future__ import annotations + +from pathlib import Path + +import numpy as np +import polars as pl +from catboost.utils import quantize + + +def quantized_pool( + frame: pl.LazyFrame, + features: list[str], + target_columns: list[str], + output_dir: Path, + split: str, + input_borders: Path | None = None, +): + output_dir.mkdir(parents=True, exist_ok=True) + table = output_dir / f"{split}.tsv" + description = output_dir / "survival.cd" + description.write_text("0\tLabel\n1\tLabel\n", encoding="utf-8") + + schema = frame.select(features).collect_schema() + non_numeric = [name for name in features if not schema[name].is_numeric()] + if non_numeric: + raise ValueError(f"Disk quantization requires numeric features: {non_numeric}") + + frame.select(target_columns + features).sink_csv( + table, + separator="\t", + null_value="nan", + ) + options = { + "column_description": str(description), + "has_header": True, + } + if input_borders is not None: + options["input_borders"] = str(input_borders) + pool = quantize(str(table), **options) + table.unlink() + return pool + + +def binary_target(frame: pl.LazyFrame, horizon: float) -> np.ndarray: + return ( + frame.select( + ( + (pl.col("target_upper") != -1) + & (pl.col("target_lower") <= horizon) + ).cast(pl.Int8) + ) + .collect(engine="streaming") + .to_series() + .to_numpy() + ) diff --git a/retraining/export_stand_events.py b/retraining/export_stand_events.py new file mode 100644 index 0000000..369c5e7 --- /dev/null +++ b/retraining/export_stand_events.py @@ -0,0 +1,127 @@ +#!/usr/bin/env python3 +"""Export historical stand events as monthly Parquet files for retraining.""" + +from __future__ import annotations + +import argparse +import json +import os +import subprocess +from datetime import date +from pathlib import Path + +import polars as pl + + +REMOTE_PSQL = ( + "docker exec -i backend-postgres-1 sh -c " + "'exec psql -q -X -v ON_ERROR_STOP=1 -U \"$POSTGRES_USER\" -d \"$POSTGRES_DB\"'" +) +SCHEMA = { + "ид_события": pl.Int64, + "ид_канала_данных": pl.Int64, + "дата": pl.Date, + "время": pl.String, + "значение_датчика": pl.String, + "тревожное": pl.String, +} + + +def months(first: date, last: date): + current = first + while current <= last: + year, month = current.year, current.month + next_month = date(year + month // 12, month % 12 + 1, 1) + yield current, next_month + current = next_month + + +def query(start: date, end: date, limit: int | None) -> str: + # PostgreSQL stores the original local date/time as a timestamptz in MSK. + rows = f"LIMIT {limit}" if limit is not None else "" + return f"""COPY ( + SELECT event_id AS "ид_события", + channel_id AS "ид_канала_данных", + (ts AT TIME ZONE 'Europe/Moscow')::date AS "дата", + to_char(ts AT TIME ZONE 'Europe/Moscow', 'HH24:MI:SS') AS "время", + raw_value AS "значение_датчика", + CASE WHEN is_alarm THEN 't' ELSE 'f' END AS "тревожное" + FROM events + WHERE ts >= '{start.isoformat()} 00:00:00+03'::timestamptz + AND ts < '{end.isoformat()} 00:00:00+03'::timestamptz + {rows} + ) TO STDOUT WITH (FORMAT CSV, HEADER TRUE); +""" + + +def export_month( + start: date, + end: date, + output_dir: Path, + ssh_host: str, + password_file: Path, + limit: int | None, +) -> dict: + name = f"events-{start:%Y-%m}" + destination = output_dir / f"{name}.parquet" + if destination.exists(): + frame = pl.scan_parquet(destination) + if frame.collect_schema().names() != list(SCHEMA): + raise ValueError(f"Unexpected schema in existing file: {destination}") + return {"file": destination.name, "rows": frame.select(pl.len()).collect().item(), "skipped": True} + + csv_file = output_dir / f".{name}.csv.part" + parquet_file = output_dir / f".{name}.parquet.part" + csv_file.unlink(missing_ok=True) + parquet_file.unlink(missing_ok=True) + + command = [ + "sshpass", "-f", str(password_file), "ssh", "-C", + "-o", "StrictHostKeyChecking=yes", "-o", "ConnectTimeout=15", + ssh_host, REMOTE_PSQL, + ] + with csv_file.open("wb") as output: + result = subprocess.run( + command, input=query(start, end, limit).encode(), stdout=output, + stderr=subprocess.PIPE, check=False, + ) + if result.returncode: + csv_file.unlink(missing_ok=True) + raise RuntimeError(f"Export {name} failed: {result.stderr.decode(errors='replace')}") + + frame = pl.scan_csv(csv_file, schema_overrides=SCHEMA, try_parse_dates=True) + row_count = frame.select(pl.len()).collect().item() + if row_count: + frame.sink_parquet(parquet_file, compression="zstd") + os.replace(parquet_file, destination) + csv_file.unlink() + return {"file": destination.name, "rows": row_count, "skipped": False} + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--from-month", default="2019-01") + parser.add_argument("--through-month", default="2026-06") + parser.add_argument("--output-dir", type=Path, required=True) + parser.add_argument("--ssh-host", required=True) + parser.add_argument("--password-file", type=Path, required=True) + parser.add_argument("--limit-per-month", type=int, help="For a small format smoke test only") + args = parser.parse_args() + + first = date.fromisoformat(args.from_month + "-01") + last = date.fromisoformat(args.through_month + "-01") + if first > last: + parser.error("--from-month must be no later than --through-month") + if args.limit_per_month is not None and args.limit_per_month < 1: + parser.error("--limit-per-month must be positive") + if not args.password_file.is_file(): + parser.error("--password-file does not exist") + args.output_dir.mkdir(parents=True, exist_ok=True) + + for start, end in months(first, last): + result = export_month(start, end, args.output_dir, args.ssh_host, args.password_file, args.limit_per_month) + print(json.dumps(result, ensure_ascii=False), flush=True) + + +if __name__ == "__main__": + main() diff --git a/retraining/run_pipeline.py b/retraining/run_pipeline.py index 2eea506..632a328 100644 --- a/retraining/run_pipeline.py +++ b/retraining/run_pipeline.py @@ -6,13 +6,13 @@ import argparse import json import shutil -import subprocess -import sys from datetime import datetime from pathlib import Path +import nbformat import polars as pl import yaml +from nbconvert.preprocessors import ExecutePreprocessor ROOT = Path(__file__).resolve().parent @@ -97,27 +97,35 @@ def patch_notebook(source: Path, destination: Path, replacements: dict[str, str] # В исходной тетрадке есть случайный символ, делающий ячейку невалидной Python. code = code.replace("import polars as plА", "import polars as pl") cell["source"] = code.splitlines(keepends=True) + cell["execution_count"] = None + cell["outputs"] = [] destination.write_text(json.dumps(notebook, ensure_ascii=False, indent=1), encoding="utf-8") def execute_notebook(notebook: Path, run_dir: Path, kernel: str, timeout: int) -> None: - timeout_value = "None" if timeout == 0 else str(timeout) - command = [ - sys.executable, "-m", "jupyter", "nbconvert", "--to", "notebook", "--execute", "--inplace", - f"--ExecutePreprocessor.kernel_name={kernel}", - f"--ExecutePreprocessor.timeout={timeout_value}", str(notebook), - ] print(f"\n>>> Выполняется {notebook.name}") - subprocess.run(command, cwd=run_dir, check=True) + node = nbformat.read(notebook, as_version=4) + preprocessor = ExecutePreprocessor( + kernel_name=kernel, + timeout=None if timeout == 0 else timeout, + ) + try: + preprocessor.preprocess(node, {"metadata": {"path": str(run_dir)}}) + finally: + nbformat.write(node, notebook) def chunk_count(run_dir: Path, split: str) -> int: return len(list((run_dir / "data" / f"final_{split}").glob("chunk_*.parquet"))) -def chunks(limit: int, available: int, split: str) -> list[int]: +def chunks(limit: int | str, available: int, split: str) -> list[int]: if available == 0: raise RuntimeError(f"После разбиения не найдено чанков {split}.") + if limit == "all": + return list(range(available)) + if not isinstance(limit, int) or limit < 1: + raise ValueError(f"Неверный лимит чанков {split}: {limit!r}") return list(range(min(limit, available))) @@ -137,6 +145,8 @@ def main() -> None: event_files = sorted(incoming_dir.glob(config["input"]["events_glob"])) if not event_files: raise FileNotFoundError(f"В {incoming_dir} нет файлов {config['input']['events_glob']}") + if len(event_files) > 1 and any(file.name == "журнал_событий_пример.parquet" for file in event_files): + raise ValueError("Исключите журнал_событий_пример.parquet из полного прогона через events_dir или events_glob") run_dir.mkdir(parents=True) data_dir = run_dir / "data" @@ -176,9 +186,40 @@ def main() -> None: patch_notebook(TEMPLATES_DIR / NOTEBOOKS[3], notebooks_dir / NOTEBOOKS[3], tuning_replacements) execute_notebook(notebooks_dir / NOTEBOOKS[3], run_dir, kernel, timeout) + final = limits.get("final_training", tuning) + shutil.copy2(ROOT / "disk_pool.py", run_dir / "disk_pool.py") + full_pool = '''train_pool, y_train_binary, n_train = make_pool( + "train", + TRAIN_CHUNKS, +) + +val_pool, y_val_binary, n_val = make_pool( + "val", + VAL_CHUNKS, +)''' + disk_pool = '''from disk_pool import binary_target, quantized_pool + +quantized_dir = Path("data/quantized") +train_frame = scan_split("train", TRAIN_CHUNKS) +val_frame = scan_split("val", VAL_CHUNKS) +train_pool = quantized_pool( + train_frame, selected_features, TARGET_COLS, quantized_dir, "train" +) +n_train = train_pool.num_row() +borders_path = quantized_dir / "borders.txt" +train_pool.save_quantization_borders(str(borders_path)) +val_pool = quantized_pool( + val_frame, selected_features, TARGET_COLS, quantized_dir, "val", borders_path +) +n_val = val_pool.num_row() +y_val_binary = binary_target(val_frame, HORIZON)''' final_replacements = { - "TRAIN_CHUNKS = [*range(8)]": f"TRAIN_CHUNKS = {chunks(tuning['train_chunks'], available['train'], 'train')}", - "VAL_CHUNKS = [*range(80)]": f"VAL_CHUNKS = {chunks(tuning['val_chunks'], available['val'], 'val')}", + "TRAIN_CHUNKS = [*range(8)]": f"TRAIN_CHUNKS = {chunks(final['train_chunks'], available['train'], 'train')}", + "VAL_CHUNKS = [*range(80)]": f"VAL_CHUNKS = {chunks(final['val_chunks'], available['val'], 'val')}", + "# Те же чанки, что были в 4_tuning.ipynb": "# Финальное обучение использует лимиты из config.yaml", + full_pool: disk_pool, + 'print(f"Model saved to: {BEST_MODEL_PATH}")': 'print(f"Model saved to: {BEST_MODEL_PATH}")\n\ndel train_pool\ngc.collect()', + 'del p_val\n\ngc.collect()': 'del p_val\ndel val_pool\ndel y_val_binary\n\ngc.collect()', } patch_notebook(TEMPLATES_DIR / NOTEBOOKS[4], notebooks_dir / NOTEBOOKS[4], final_replacements) execute_notebook(notebooks_dir / NOTEBOOKS[4], run_dir, kernel, timeout)