Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 12 additions & 8 deletions retraining/README.md
Original file line number Diff line number Diff line change
@@ -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/<run_id>/notebooks/`. В них подставляются годы входных событий и числа доступных чанков. Результат зависит от фактически переданных файлов, настроек и версии кода; пример из репозитория не заменяет полный набор событий.
Эта папка — самостоятельная точка повторного обучения модели на исходных и новых данных. Добавьте новые журналы событий в `data/incoming_events/` и запустите один скрипт: он объединит все записи, сформирует признаки, заново разобьёт устройства на train/val/test, выполнит feature selection, Optuna-тюнинг, обучит итоговую модель и посчитает метрики.

Шаблоны в `notebooks/` — копии исходных тетрадок. Они **не перезаписываются**. Для каждого прогона скрипт создаёт рабочие копии в `artifacts/runs/<run_id>/notebooks/` и только там подставляет найденные годы и реально доступные чанки. По умолчанию feature selection и подбор параметров используют ограниченные выборки, а итоговая модель обучается на всех полученных train-чанках. Validation и test остаются отдельными выборками для оценки качества; финальная оценка охватывает все test-файлы.

Результат зависит от фактически переданных файлов, настроек и версии кода; пример из репозитория не заменяет полный набор событий. Скрипт не заменяет модель сервиса инференса.

## Состав

Expand Down Expand Up @@ -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`.
Expand All @@ -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, если они не предназначены для публикации.
78 changes: 78 additions & 0 deletions retraining/baseline_previous_day.py
Original file line number Diff line number Diff line change
@@ -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()
45 changes: 45 additions & 0 deletions retraining/cloud/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
# Обучение в Yandex Cloud

`job.py` выполняет три отдельные операции: кладёт снимок кода и Parquet в закрытый Object Storage, создаёт временную VM через Terraform/OpenTofu и читает статус. На VM `bootstrap.py` монтирует отдельный диск, а `runner.py` выполняет пять ноутбуков, сравнивает результат с правилом «тревога вчера» и старой моделью, сохраняет артефакты в `results/<run-id>/` и выключает 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/<run-id>/`, затем выполняет `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/<run-id>/status.json`. `phase: failed` означает, что нужно смотреть `run.log`, `bootstrap-status.json` и выполненные тетрадки в том же префиксе. Перед удалением VM проверьте, что результаты в бакете появились. Модель не заменяется автоматически: сначала сравнить PR-AUC и операционные метрики с правилами «≥1 тревога вчера» и «≥10 тревог вчера» на одном test-срезе.

## API для кнопки

`serve.py` предоставляет `POST /jobs` с `{"runId":"..."}` и `GET /jobs/<run-id>` для 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='<secret>' python retraining/cloud/serve.py \
--config /private/path/orchestrator.json
```

Все файлы данных, ключи и Terraform state должны оставаться вне Git. API использует уже подготовленный каталог Parquet на управляющем хосте; обновлять его перед повторным обучением нужно отдельно. Конфигурация не разворачивает управляющий хост и не меняет прод без явной настройки.
Loading
Loading