Сравнить коммиты

..

Ничего общего в коммитах. 'master' и 'lab_2/master' имеют совершенно разные истории.

3
.gitignore поставляемый

@ -1,5 +1,4 @@
### Python
__pycache__/
*.pyc
### Jupyter
@ -10,5 +9,3 @@ __pycache__/
# virtual environments
.venv/
.venv*/
# .env files
*.env

@ -2,27 +2,15 @@
**Выполняет**: **Сыропятов В.В.** (А-01м-24)
Реализация цикла разработки ML-сервиса:
* подготовка и исследование данных ([Jupyter](https://jupyter.org/), scientific Python),
* создание модели ML ([MLFlow](https://mlflow.org/), [scikit-learn](https://scikit-learn.org/stable/)),
* разработка REST-сервиса для выполнения модели ([FastAPI](https://fastapi.tiangolo.com/) (недоступна на 2025‑12‑08)),
* конфигурация инфрастуктуры сервиса с мониторингом ([Docker](https://www.docker.com/) (частично недоступна на 2025‑12‑08)), [Docker Compose](https://docs.docker.com/compose/) (частично недоступна на 2025‑12‑08)), [Prometheus](https://prometheus.io/) (недоступна на 2025‑12‑08)), [Grafana OSS](https://grafana.com/)).
## Данные
Используемый датасет: [Car price prediction(used cars)
](https://www.kaggle.com/datasets/vijayaadithyanvg/car-price-predictionused-cars/data) —
продажа подержанных автомобилей на рынке в Индии.
## Сервис предсказания цен
См. `services/README.md`.
## Исследовательская часть проекта
### Установка
## Установка
#### Общий порядок
### Общий порядок
**Внимание**: Здесь описан только общий порядок установки. Определённые части проекта могут требовать установки по отдельным инструкциям.
@ -59,13 +47,13 @@
5. **При необходимости** скачайте данные. Каноническое расположение для данных проекта: `data/`.
#### Зависимости
### Зависимости
##### Общие зависимости
#### Общие зависимости
Зависимости — пакеты Python — записаны в файле `requirements/requirements.txt` (см. **Пакеты Python**).
##### Пакеты Python
#### Пакеты Python
Установка/обновление пакетов Python в активное окружение из файла `requirements/requirements.txt`:
@ -73,10 +61,10 @@
pip install -U -r requirements/requirements.txt
```
### Разведочный анализ данных (EDA)
## Разведочный анализ данных (EDA)
См. `eda/README.md`.
### Исследование и настройка предсказательной модели
## Исследование и настройка предсказательной модели
См. `research/README.md`.

@ -2,7 +2,7 @@
set -eu
. ./_mlflow_config_common.sh
. _mlflow_config_common.sh
DEFAULT_ARTIFACT_ROOT="./mlflow/mlartifacts/"

@ -1,169 +0,0 @@
# Сервис предсказания цен
Веб-сервис предсказания цен на подержанные автомобили. Мониторинг в комплекте.
Обзор сервисов (по `compose.yaml`, см. о развёртывании ниже):
| Профили Compose | Имя | Объекты | Описание |
|-----------------|------------------|------------------|------------------|
| — | `prices-predictor` | код: `ml_service/` | Веб-сервис предсказания цен, только stateless API. Об используемой предсказательной модели см. `research/README.md`. |
| — | `prometheus` | конфигурация: `prometheus/` | Мониторинг сервиса ([Prometheus](https://prometheus.io/)). |
| — | `grafana` | сохранённая конфигурация: `grafana/` | Аналитика и визуализация данных мониторига сервиса ([Grafana](https://grafana.com/)). |
| `with-testers` | `load-tester` | код: `load-tester/` | Генератор потока случайных запросов к `prices-predictor` для тестирования. |
Дополнительно:
* `models/` — расположение файла модели `model.pkl` для использования сервисом `prices-predictor`.
* `fetch_model_as_pickle_from_mlflow.py` — скрипт для экспорта предиктивной модели scikit-learn из MLFlow в файл.
## API сервиса предсказания цен
**Базовый URL**: `/api`. Все указанные далее URL записаны **относительно базового URL**, если не указано иное.
* Полная интерактивная документация (Swagger UI): `/docs`.
* Предсказать цену подержанного автомобиля: `/predict` (POST).
Пример запроса:
* requst query: `item_id=16` (параметр `item_id` необходим!);
* request body:
```json
{
"selling_price": 5.59,
"driven_kms": 27000.0,
"age": 5.0,
"fuel_type": "petrol",
"selling_type": "dealer",
"transmission_type": "manual"
}
```
* response body:
```json
{
"item_id": 16,
"price": 3.743508852258851
}
```
* Тестовый эндпоинт: `/` (GET).
Возвращает простой демонстрационный объект JSON.
Может использоваться для проверки состояния сервиса.
## Мониторинг
### Prometheus UI
#### Примеры запросов
Гистограмма предсказанных цен `model_prediction_value_bucket` (запрос: `rate(model_prediction_value_bucket[5m]`):
![Гистограмма предсказанных цен как временные ряды](docs/screenshot-prometheus-query-model-1.png)
Гистограмма продолжительности предсказания цен моделью ML `model_prediction_seconds_bucket` (запрос: `rate(model_prediction_seconds_bucket[5m]`):
![Гистограмма продолжительности предсказания цен моделью ML как временные ряды](docs/screenshot-prometheus-query-model-2.png)
Интенсивность потока запросов к сервису предсказания цен с разными результатами (успех — коды HTTP `2xx`, ошибки со стороны клиента — коды HTTP `4xx`) `http_requests_total{handler="/predict"}` (запрос: `rate(http_requests_total{handler="/predict"}[5m]`):
![Интенсивность потока запросов к сервису предсказания цен с разными результатами](docs/screenshot-prometheus-query-http-1.png)
Интенсивность потока запросов к **веб-серверу** сервиса предсказания цен **с ошибками** `http_requests_total{handler="/predict"}` (запрос: `sum without(handler, method) (rate(http_requests_total{status=~"4..|5.."}[5m]))`):
![Интенсивность потока запросов к веб-серверу сервиса предсказания цен, заканчивающихся ошибками](docs/screenshot-prometheus-query-http-1.png)
### Дашборд в Grafana
Дашборд экспортирован в файл: `grafana/objects/dashboard-1765200932880.json`.
![Дашборд в Grafana](docs/screenshot-grafana-dashboard.png)
Элементы:
* мониторинг модели:
* гистограмма распределения предсказанных цен за период времени (10 мин);
* прикладной уровень:
* интенсивность потока запросов (всех запросов; запросов, заканчивающихся ошибкой);
* инфраструктурный уровень:
* состояние сервиса (up/down);
* выделенный процессу объём VRAM.
## Развёртывание
### Файл модели
Файл используемой предсказательной модели можно извлечь из MLFlow скриптом `models/fetch_model_as_pickle_from_mlflow.py`. Файл модели можно размещается в `models/model.pkl`.
Например, извлечь модель по имени (`<model-name>`) и версии (`<model-version>`) (например, `UsedCardPricePredictionFinal/1`) (команда запускается из корневой директории проекта &mdash; от этого зависит путь к создаваемому файлу):
python services/models/fetch_model_as_pickle_from_mlflow.py --model "models:/<model-name>/<model-version>" services/models/model.pkl
Можно указать адрес tracking сервера MLFlow, например: `--tracking-uri "http://localhost:5000"`.
Информация о других опциях доступна:
python services/models/fetch_model_as_pickle_from_mlflow.py --help
### Образы Docker
#### `ml_model` (для сервиса `prices-predictor`)
**Сборка образа** (замените `<version>` на номер версии) (команда запускается из корневой директории проекта &mdash; от этого зависит путь к директории):
docker build -t ml_service:<version> services/ml_service/
**Независимый запуск** (замените `<version>` на номер версии образа, `<models-dir>` на **абсолютный** путь к директории, где размещён файл предсказательной модели `model.pkl`, `<port>` на порт для запуска веб-сервиса (например, `8000`)):
docker run -v "<models-dir>:/models" -p <port>:8000 ml_service:<version>
Модель может быть размещена в `models/`; тогда, например, при запуске команды из корна проекта: `$(pwd)/services/models` (здесь `$(pwd)` используется потому, что необходим абсолютный путь).
#### `load-tester` (для сервиса `load-tester`)
**Сборка образа** (замените `<version>` на номер версии) (команда запускается из корневой директории проекта &mdash; от этого зависит путь к директории):
docker build -t load_tester:<version> services/load_tester/
**Независимый запуск** (замените `<version>` на номер версии образа, `<api-base-url>` на базовый URL сервиса `prices-predictor` (например, `http://prices-predictor:8000/api`)):
docker run -e "API_BASE_URL=<api-base-url>" ml_service:<version>
### Развёртывание сервиса посредством Compose
Конфигурация описана в файле `compose.yaml`. Имя системы: `mpei-iis-system`.
Рекомендуется (не обязательно) использовать env-файл `compose.env`. Используйте файл `compose.env.template` как шаблон.
**Директория `models/` используется сервисом `prices-predictor` как том** с файлом модели `model.pkl`. Поместите туда файл модели, см. [Файл модели](#файл-модели).
**Управление сервисом с мониторингом** (замените `<command>` и `[options...]`):
docker compose -f services/compose.yaml --env-file services/compose.env <command> [options...]
**Для запуска вместе с генераторами тестовых запросов** используйте опцию compose `--profile=with-tester`.
Основные команды `docker compose`:
* `up`: создать и запустить контейнеры (также тома, сети и прочее); оставляет вывод логов прикреплённым к терминалу, `SIGINT` останавливает контейнеры, **но не удаляет созданные объекты**;
* опция `-d`: то же, но открепляет процесс от терминала.
* `down`: остановить и удалить контейнеры (также сети и прочее; для удаления томов используйте опцию `-v`).
* `start`: запустить существующие контейнеры.
* `stop`: остановить контейнеры.
* `restart`: перезапустить контейнеры.
**Открытые на хосте интерфейсы**:
* `localhost:8010`: Сервис `prices-predictor`. Базовый URL: `/api`.
* `localhost:9090`: UI Prometheus.
* `localhost:3000`: Grafana.
**Доступные на хосте тома**:
* `mpei-iis-system_prometheus-storage`: БД Prometheus.
* `mpei-iis-system_grafana-storage`: БД Grafana.

@ -1,2 +0,0 @@
GF_SECURITY_ADMIN_USER=admin
GF_SECURITY_ADMIN_PASSWORD=admin

@ -1,64 +0,0 @@
name: mpei-iis-system
services:
prices-predictor:
image: ml_service:2
ports:
- "8010:8000"
volumes:
- './models:/models:ro'
load-tester:
image: load_tester:1
environment:
API_BASE_URL: "http://prices-predictor:8000/api"
# XXX: Предотвращает аварийный выход тестера при отсутствии ответа от prices-predictor
# во время его (потенциально долгого) запуска.
depends_on:
- prices-predictor
deploy:
replicas: 2
profiles:
- "with-testers"
prometheus:
image: prom/prometheus:v3.7.3
ports:
- "9090:9090"
user: nobody
command:
- "--config.file=/etc/prometheus/prometheus.yaml"
volumes:
- "./prometheus/prometheus.yaml:/etc/prometheus/prometheus.yaml:ro"
- "prometheus-storage:/prometheus"
grafana:
image: grafana/grafana:12.4.0-20012734117
ports:
- "3000:3000"
#environment:
# GF_SECURITY_ADMIN_USER: "$__file{/run/secrets/grafana-admin-user}"
# GF_SECURITY_ADMIN_PASSWORD: "$__file{/run/secrets/grafana-admin-password}"
#secrets:
# - grafana-admin-user
# - grafana-admin-password
environment:
GF_SECURITY_ADMIN_USER: "${GF_SECURITY_ADMIN_USER:-admin}"
GF_SECURITY_ADMIN_PASSWORD: "${GF_SECURITY_ADMIN_PASSWORD:-admin}"
volumes:
- "grafana-storage:/var/lib/grafana"
volumes:
prometheus-storage: {}
grafana-storage: {}
#secrets:
#
# grafana-admin-user:
# environment: GF_SECURITY_ADMIN_USER
#
# grafana-admin-password:
# environment: GF_SECURITY_ADMIN_PASSWORD

Двоичный файл не отображается.

До

Ширина:  |  Высота:  |  Размер: 164 KiB

Двоичный файл не отображается.

До

Ширина:  |  Высота:  |  Размер: 87 KiB

Двоичный файл не отображается.

До

Ширина:  |  Высота:  |  Размер: 78 KiB

Двоичный файл не отображается.

До

Ширина:  |  Высота:  |  Размер: 115 KiB

Двоичный файл не отображается.

До

Ширина:  |  Высота:  |  Размер: 99 KiB

1
services/grafana/.gitattributes поставляемый

@ -1 +0,0 @@
objects/*.json -text

@ -1,543 +0,0 @@
{
"apiVersion": "dashboard.grafana.app/v2beta1",
"kind": "Dashboard",
"metadata": {
"name": "adcdfv7",
"generation": 9,
"creationTimestamp": "2025-12-08T12:51:53Z",
"labels": {},
"annotations": {}
},
"spec": {
"annotations": [
{
"kind": "AnnotationQuery",
"spec": {
"builtIn": true,
"enable": true,
"hide": true,
"iconColor": "rgba(0, 211, 255, 1)",
"name": "Annotations & Alerts",
"query": {
"group": "grafana",
"kind": "DataQuery",
"spec": {},
"version": "v0"
}
}
}
],
"cursorSync": "Off",
"editable": true,
"elements": {
"panel-1": {
"kind": "Panel",
"spec": {
"data": {
"kind": "QueryGroup",
"spec": {
"queries": [
{
"kind": "PanelQuery",
"spec": {
"hidden": false,
"query": {
"group": "prometheus",
"kind": "DataQuery",
"spec": {
"editorMode": "builder",
"exemplar": false,
"expr": "process_virtual_memory_bytes",
"instant": false,
"interval": "10s",
"legendFormat": "__auto",
"range": true
},
"version": "v0"
},
"refId": "A"
}
}
],
"queryOptions": {},
"transformations": []
}
},
"description": "",
"id": 1,
"links": [],
"title": "Выделенный объём VRAM",
"vizConfig": {
"group": "timeseries",
"kind": "VizConfig",
"spec": {
"fieldConfig": {
"defaults": {
"color": {
"mode": "palette-classic"
},
"custom": {
"axisBorderShow": false,
"axisCenteredZero": false,
"axisColorMode": "text",
"axisLabel": "",
"axisPlacement": "auto",
"barAlignment": 0,
"barWidthFactor": 0.6,
"drawStyle": "bars",
"fillOpacity": 0,
"gradientMode": "none",
"hideFrom": {
"legend": false,
"tooltip": false,
"viz": false
},
"insertNulls": false,
"lineInterpolation": "stepAfter",
"lineWidth": 1,
"pointSize": 5,
"scaleDistribution": {
"type": "linear"
},
"showPoints": "auto",
"showValues": false,
"spanNulls": false,
"stacking": {
"group": "A",
"mode": "none"
},
"thresholdsStyle": {
"mode": "off"
}
},
"thresholds": {
"mode": "absolute",
"steps": [
{
"color": "green",
"value": 0
},
{
"color": "red",
"value": 80
}
]
},
"unit": "bytes"
},
"overrides": []
},
"options": {
"legend": {
"calcs": [],
"displayMode": "list",
"placement": "bottom",
"showLegend": true
},
"tooltip": {
"hideZeros": false,
"mode": "single",
"sort": "none"
}
}
},
"version": "12.4.0-20012734117"
}
}
},
"panel-2": {
"kind": "Panel",
"spec": {
"data": {
"kind": "QueryGroup",
"spec": {
"queries": [
{
"kind": "PanelQuery",
"spec": {
"hidden": false,
"query": {
"group": "prometheus",
"kind": "DataQuery",
"spec": {
"editorMode": "builder",
"expr": "up",
"legendFormat": "__auto",
"range": true
},
"version": "v0"
},
"refId": "A"
}
}
],
"queryOptions": {},
"transformations": []
}
},
"description": "",
"id": 2,
"links": [],
"title": "Состояние",
"vizConfig": {
"group": "state-timeline",
"kind": "VizConfig",
"spec": {
"fieldConfig": {
"defaults": {
"color": {
"mode": "continuous-GrYlRd"
},
"custom": {
"axisPlacement": "auto",
"fillOpacity": 70,
"hideFrom": {
"legend": false,
"tooltip": false,
"viz": false
},
"insertNulls": false,
"lineWidth": 0,
"spanNulls": false
},
"thresholds": {
"mode": "absolute",
"steps": [
{
"color": "green",
"value": 0
}
]
},
"unit": "bool_on_off"
},
"overrides": []
},
"options": {
"alignValue": "left",
"legend": {
"displayMode": "list",
"placement": "bottom",
"showLegend": true
},
"mergeValues": true,
"rowHeight": 0.9,
"showValue": "auto",
"tooltip": {
"hideZeros": false,
"mode": "single",
"sort": "none"
}
}
},
"version": "12.4.0-20012734117"
}
}
},
"panel-3": {
"kind": "Panel",
"spec": {
"data": {
"kind": "QueryGroup",
"spec": {
"queries": [
{
"kind": "PanelQuery",
"spec": {
"hidden": false,
"query": {
"group": "prometheus",
"kind": "DataQuery",
"spec": {
"editorMode": "builder",
"expr": "sum without(instance, method, status) (rate(http_requests_total{handler=\"/predict\"}[$__rate_interval]))",
"interval": "1m",
"legendFormat": "__auto",
"range": true
},
"version": "v0"
},
"refId": "A"
}
},
{
"kind": "PanelQuery",
"spec": {
"hidden": false,
"query": {
"group": "prometheus",
"kind": "DataQuery",
"spec": {
"editorMode": "builder",
"expr": "rate(http_requests_total{handler=\"/predict\", status=~\"4..|5..\"}[$__rate_interval])",
"instant": false,
"interval": "1m",
"legendFormat": "__auto",
"range": true
},
"version": "v0"
},
"refId": "B"
}
}
],
"queryOptions": {},
"transformations": []
}
},
"description": "",
"id": 3,
"links": [],
"title": "HTTP-запросы",
"vizConfig": {
"group": "timeseries",
"kind": "VizConfig",
"spec": {
"fieldConfig": {
"defaults": {
"color": {
"mode": "palette-classic"
},
"custom": {
"axisBorderShow": false,
"axisCenteredZero": false,
"axisColorMode": "text",
"axisLabel": "",
"axisPlacement": "auto",
"barAlignment": 0,
"barWidthFactor": 0.6,
"drawStyle": "line",
"fillOpacity": 0,
"gradientMode": "none",
"hideFrom": {
"legend": false,
"tooltip": false,
"viz": false
},
"insertNulls": false,
"lineInterpolation": "linear",
"lineWidth": 1,
"pointSize": 5,
"scaleDistribution": {
"type": "linear"
},
"showPoints": "auto",
"showValues": false,
"spanNulls": false,
"stacking": {
"group": "A",
"mode": "none"
},
"thresholdsStyle": {
"mode": "off"
}
},
"thresholds": {
"mode": "absolute",
"steps": [
{
"color": "green",
"value": 0
}
]
},
"unit": "reqps"
},
"overrides": []
},
"options": {
"legend": {
"calcs": [],
"displayMode": "list",
"placement": "bottom",
"showLegend": true
},
"tooltip": {
"hideZeros": false,
"mode": "single",
"sort": "none"
}
}
},
"version": "12.4.0-20012734117"
}
}
},
"panel-4": {
"kind": "Panel",
"spec": {
"data": {
"kind": "QueryGroup",
"spec": {
"queries": [
{
"kind": "PanelQuery",
"spec": {
"hidden": false,
"query": {
"group": "prometheus",
"kind": "DataQuery",
"spec": {
"editorMode": "builder",
"exemplar": false,
"expr": "sum without(instance) (increase(model_prediction_value_bucket[10m])) / on() group_left sum(increase(model_prediction_value_count[10m]))",
"format": "heatmap",
"instant": false,
"interval": "10m",
"legendFormat": "<{{le}}",
"range": true
},
"version": "v0"
},
"refId": "A"
}
}
],
"queryOptions": {},
"transformations": []
}
},
"description": "Подпись под каждым столбцом обозначает его максимальное соответствующее значение",
"id": 4,
"links": [],
"title": "Предсказанные цены за 10 минут",
"vizConfig": {
"group": "bargauge",
"kind": "VizConfig",
"spec": {
"fieldConfig": {
"defaults": {
"color": {
"mode": "thresholds"
},
"thresholds": {
"mode": "absolute",
"steps": [
{
"color": "green",
"value": 0
}
]
},
"unit": "percentunit"
},
"overrides": []
},
"options": {
"displayMode": "gradient",
"legend": {
"calcs": [],
"displayMode": "list",
"placement": "bottom",
"showLegend": false
},
"maxVizHeight": 300,
"minVizHeight": 16,
"minVizWidth": 8,
"namePlacement": "auto",
"orientation": "auto",
"reduceOptions": {
"calcs": [
"lastNotNull"
],
"fields": "",
"values": false
},
"showUnfilled": false,
"sizing": "auto",
"valueMode": "color"
}
},
"version": "12.4.0-20012734117"
}
}
}
},
"layout": {
"kind": "GridLayout",
"spec": {
"items": [
{
"kind": "GridLayoutItem",
"spec": {
"element": {
"kind": "ElementReference",
"name": "panel-4"
},
"height": 8,
"width": 12,
"x": 0,
"y": 0
}
},
{
"kind": "GridLayoutItem",
"spec": {
"element": {
"kind": "ElementReference",
"name": "panel-2"
},
"height": 8,
"width": 12,
"x": 12,
"y": 0
}
},
{
"kind": "GridLayoutItem",
"spec": {
"element": {
"kind": "ElementReference",
"name": "panel-3"
},
"height": 8,
"width": 12,
"x": 0,
"y": 8
}
},
{
"kind": "GridLayoutItem",
"spec": {
"element": {
"kind": "ElementReference",
"name": "panel-1"
},
"height": 8,
"width": 12,
"x": 12,
"y": 8
}
}
]
}
},
"links": [],
"liveNow": false,
"preload": false,
"tags": [],
"timeSettings": {
"autoRefresh": "30s",
"autoRefreshIntervals": [
"5s",
"10s",
"30s",
"1m",
"5m",
"15m",
"30m",
"1h",
"2h",
"1d"
],
"fiscalYearStartMonth": 0,
"from": "now-1h",
"hideTimepicker": false,
"timezone": "browser",
"to": "now"
},
"title": "Сервис предсказания цен",
"variables": []
},
"status": {}
}

@ -1,6 +0,0 @@
### Python
__pycache__/
*.pyc
### Project
*.unused

@ -1,13 +0,0 @@
FROM python:3.11-slim
WORKDIR /load_tester
COPY ./requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["python", "-m", "tester"]
# docker build -t load_tester:1 services/load_tester/
# docker run -e "API_BASE_URL=http://prices-predictor:8000/api" load_tester:1

@ -1,280 +0,0 @@
from argparse import ArgumentParser
from collections.abc import Callable, MutableMapping
from dataclasses import dataclass, asdict
from enum import Enum
import logging
from os import getenv
from random import randint, uniform, expovariate, choice
from signal import SIGINT, SIGTERM, signal
import sys
from time import sleep
from types import FrameType
from typing import Any, cast
from requests import RequestException, Response, Session
def fixup_payload_enum_value(mapping: MutableMapping[str, Any], key: str) -> None:
mapping[key] = mapping[key].value
ENDPOINT_URL: str = '/predict'
class FuelType(Enum):
PETROL = 'petrol'
DIESEL = 'diesel'
CNG = 'cng'
class SellingType(Enum):
DEALER = 'dealer'
INDIVIDUAL = 'individual'
class TransmissionType(Enum):
MANUAL = 'manual'
AUTOMATIC = 'automatic'
@dataclass
class PricePredictionFeatures:
selling_price: float
driven_kms: float
age: float
fuel_type: FuelType
selling_type: SellingType
transmission_type: TransmissionType
MAX_RETRIES_DEFAULT = 3
def exp_delay_from_attempt_number(attempt_i: int) -> float:
return 0.2 * (2 ** attempt_i)
def post_item(
session: Session, url: str, item_id: int, features: PricePredictionFeatures,
*, max_retries: int = MAX_RETRIES_DEFAULT,
) -> Response:
if max_retries < 0:
raise ValueError('max_retries must be >= 0')
payload = asdict(features)
for k in ('fuel_type', 'selling_type', 'transmission_type'):
fixup_payload_enum_value(payload, k)
excs = []
for attempt_i in range(max_retries + 1):
try:
response = session.post(url, params={'item_id': item_id}, json=payload, timeout=10)
except RequestException as err:
excs.append(err)
sleep(exp_delay_from_attempt_number(attempt_i))
else:
return response
assert len(excs) > 0
# XXX: ...
raise IOError(
f'Failed to post an item in {max_retries + 1} attempts;'
' see the latest exception in __cause__'
) from excs[-1]
def generate_request_data() -> tuple[int, PricePredictionFeatures]:
item_id = randint(1, 100)
features = PricePredictionFeatures(
selling_price=round(uniform(2.0, 16.0), 2),
driven_kms=round(uniform(0.0, 100000.0), 0),
age=round(uniform(0.0, 10.0), 1),
fuel_type=choice(list(FuelType)),
selling_type=choice(list(SellingType)),
transmission_type=choice(list(TransmissionType)),
)
return (item_id, features)
INTERVAL_MEAN_DEFAULT = 4.0
INTERVAL_BOUNDS_DEFAULT: tuple[float | None, float | None] = (0.5, 10.0)
class Requester:
def __init__(
self,
base_url: str,
interval_mean: float = INTERVAL_MEAN_DEFAULT,
interval_bounds: tuple[float | None, float | None] = INTERVAL_BOUNDS_DEFAULT,
*, max_retries: int = MAX_RETRIES_DEFAULT,
):
self.base_url = base_url
self.interval_mean = interval_mean
self.interval_bounds = interval_bounds
self.max_retries = max_retries
self._session = Session()
self._stop_requested: bool = False
@property
def endpoint(self) -> str:
endpoint_url = ENDPOINT_URL
if (len(endpoint_url) > 0) and (not endpoint_url.startswith('/')):
endpoint_url = '/' + endpoint_url
return (self.base_url + endpoint_url)
@property
def session(self) -> Session:
return self._session
@property
def stop_requested(self) -> bool:
return self._stop_requested
def stop(self) -> None:
self._stop_requested = True
def _decide_delay(self) -> float:
interval_bounds = self.interval_bounds
val = expovariate(1. / self.interval_mean)
if interval_bounds[0] is not None:
val = max(val, interval_bounds[0])
if interval_bounds[1] is not None:
val = min(val, interval_bounds[1])
return val
def run(self) -> None:
while not self._stop_requested:
item_id, features = generate_request_data()
try:
response = post_item(
self._session, self.endpoint, item_id, features, max_retries=self.max_retries,
)
except IOError as err:
logging.warning('%s: %s', str(err), str(err.__cause__))
raise err
else:
logging.debug('Success: %s %s', response.status_code, response.reason)
sleep(self._decide_delay())
def _build_termination_handler(requester: Requester) -> Callable[[int, FrameType | None], None]:
def termination_handler(sig: int, frame: FrameType | None) -> None:
_ = sig
_ = frame
requester.stop()
return termination_handler
def _configure_logging(level: int, quiet: bool) -> None:
if quiet:
level = logging.CRITICAL + 1
logging.basicConfig(
level=level, format='%(asctime)s %(levelname)s %(message)s', stream=sys.stderr,
)
def _setup_signal_handlers(requester: Requester) -> None:
termination_handler = _build_termination_handler(requester)
for sig in (SIGINT, SIGTERM):
signal(sig, termination_handler)
def _validate_cli_interval_bound(string: str) -> float | None:
string = string.lower()
if string in ('', 'null', 'none'):
return None
return float(string)
def _validate_cli_interval_bounds(string: str) -> tuple[float | None, float | None]:
string = string.lower()
if string in ('', 'null', 'none'):
return (None, None)
min_string, max_string = string.split(',', 1)
return cast(
tuple[float | None, float | None],
tuple(map(_validate_cli_interval_bound, (min_string, max_string)))
)
def _validate_cli_max_retries(string: str) -> int:
val = int(string)
if val < 0:
raise ValueError(f'Max retries should be >=0, given {val}')
return val
def _validate_cli_logging_level(string: str) -> int:
return {
'debug': logging.DEBUG,
'info': logging.INFO,
'warning': logging.WARNING,
'error': logging.ERROR,
'critical': logging.CRITICAL,
}[string]
def parse_args(argv):
parser = ArgumentParser(
description=(
'Регулярная отправка POST-запросов на эндпоинт предсказания цены.'
' Остановка по SIGINT / SIGTERM.'
),
allow_abbrev=False,
exit_on_error=True,
)
parser.add_argument('base_url', type=str, nargs='?')
parser.add_argument('--interval-mean', type=float, dest='interval_mean')
parser.add_argument(
'--interval-bounds', type=_validate_cli_interval_bounds, dest='interval_bounds',
)
parser.add_argument(
'--max-retries',
type=_validate_cli_max_retries,
default=MAX_RETRIES_DEFAULT,
dest='max_retries',
)
parser.add_argument('-q', '--quiet', action='store_true', dest='quiet')
parser.add_argument(
'--log-level',
default=logging.WARNING,
type=_validate_cli_logging_level,
dest='logging_level',
)
args = parser.parse_args(argv[1:])
if args.base_url is None:
args.base_url = getenv('API_BASE_URL')
if args.base_url is None:
raise RuntimeError('No API base URL specified')
if (args.interval_mean is not None) and (args.interval_mean <= 0):
raise ValueError(f'Interval mean should be > 0, given {args.interval_mean}')
if (
(args.interval_bounds is not None)
and all((b is not None) for b in args.interval_bounds)
and (args.interval_bounds[0] > args.interval_bounds[1])
):
raise ValueError(f'Interval bounds should be b_1 <= b_2, given {args.interval_bounds!r}')
if args.interval_mean is not None:
if args.interval_bounds is None:
args.interval_bounds = ((args.interval_mean / 5), (args.interval_mean * 5))
else:
args.interval_mean = INTERVAL_MEAN_DEFAULT
args.interval_bounds = INTERVAL_BOUNDS_DEFAULT
return args
def main(argv):
args = parse_args(argv)
_configure_logging(args.logging_level, args.quiet)
logging.debug('Creating a Requester with base URL: %s', args.base_url)
requester = Requester(
args.base_url,
interval_mean=args.interval_mean,
interval_bounds=args.interval_bounds,
max_retries=args.max_retries,
)
_setup_signal_handlers(requester)
requester.run()
return 0
if __name__ == '__main__':
sys.exit(int(main(sys.argv) or 0))

@ -1,37 +0,0 @@
from os import (
getenv,
#kill,
)
#from signal import SIGINT, SIGTERM
from subprocess import (
run,
#Popen,
)
import sys
#_child_proc: Popen | None = None
#def _forward_signal(sig, frame):
# _ = frame
# if _child_proc is None:
# return
# if _child_proc.pid in (SIGINT, SIGTERM):
# kill(_child_proc.pid, sig)
# else:
# raise RuntimeError(f'Attempted to forward an unexpected signal: {sig}')
def main(argv):
argv = list(argv) # copy
base_url = getenv('API_BASE_URL')
if base_url is None:
raise RuntimeError('API_BASE_URL is not specified')
argv.append(base_url) # HACK: ...
result = run(['python', '-m', 'tester', *argv[1:]], check=False)
return result.returncode
if __name__ == '__main__':
sys.exit(int(main(sys.argv) or 0))

@ -1,19 +0,0 @@
FROM python:3.11-slim
WORKDIR /service
COPY ./requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
VOLUME /models
EXPOSE 8000/tcp
ENV MODELS_PATH=/models
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]
# docker build -t ml_service:2 services/ml_service/
# docker run -v "$(pwd)/services/models:/models" -p 8000:8000 ml_service:2

@ -1,2 +0,0 @@
from ._meta import PACKAGE_PATH
from .main import app

@ -1,4 +0,0 @@
from pathlib import Path
PACKAGE_PATH = Path(__file__).parent

@ -1,72 +0,0 @@
from os import getenv
from pathlib import Path
from fastapi import FastAPI
from prometheus_fastapi_instrumentator import Instrumentator
from pydantic import BaseModel, Field
from ._meta import PACKAGE_PATH
from .predictor import (
FuelType, SellingType, TransmissionType, PricePredictionFeatures, PricePredictor,
)
MODELS_PATH = getenv('MODELS_PATH', None)
if MODELS_PATH is not None:
MODELS_PATH = Path(MODELS_PATH)
else:
SERVICES_PATH = PACKAGE_PATH.parents[1]
assert SERVICES_PATH.name == 'services'
MODELS_PATH = SERVICES_PATH / 'models'
MODEL_PATH = MODELS_PATH / 'model.pkl'
predictor = PricePredictor(MODEL_PATH)
API_BASE_PATH = '/api'
app = FastAPI(
title='Сервис ML',
version='0.1.0',
root_path=API_BASE_PATH,
#redoc_url=None,
)
_ = (
Instrumentator(excluded_handlers=['/metrics'])
.instrument(app)
.expose(app, endpoint='/metrics')
)
@app.get('/', summary='Тестовый эндпоинт')
async def root():
return {'Hello': 'World'}
class PricePredictionRequest(BaseModel):
selling_price: float = Field(..., gt=0)
driven_kms: float = Field(..., ge=0)
age: float = Field(..., ge=0)
fuel_type: FuelType
selling_type: SellingType
transmission_type: TransmissionType
@app.post('/predict', summary='Предсказать цену подержанного автомобиля')
def predict_price(item_id: int, req: PricePredictionRequest):
features = PricePredictionFeatures(
selling_price=req.selling_price,
driven_kms=req.driven_kms,
age=req.age,
fuel_type=req.fuel_type,
selling_type=req.selling_type,
transmission_type=req.transmission_type,
)
pred = predictor.predict(features)
return {'item_id': item_id, 'price': pred}

@ -1,111 +0,0 @@
from dataclasses import dataclass
from enum import Enum
from itertools import chain
from pandas import DataFrame
from pickle import load
from prometheus_client import Counter, Histogram
def open_model_file(file, *, buffering=-1, opener=None, **kwargs_extra):
open_kwargs_extra = {}
if 'closefd' in kwargs_extra:
open_kwargs_extra['closefd'] = kwargs_extra.pop('closefd')
if len(kwargs_extra) > 0:
raise TypeError(
'Unexpected keyword arguments given: {}'
.format(', '.join(map(repr, kwargs_extra.keys())))
)
return open(file, 'rb', buffering=buffering, opener=opener)
def load_model_from_file(file):
return load(file)
def load_model_from_path(path, *, buffering=-1, opener=None, **kwargs_extra):
open_kwargs_extra = {}
for k in ('closefd',):
if k in kwargs_extra:
open_kwargs_extra[k] = kwargs_extra.pop(k)
if len(kwargs_extra) > 0:
raise TypeError(
'Unexpected keyword arguments given: {}'.format(', '.join(kwargs_extra.keys()))
)
with open_model_file(
path, buffering=buffering, opener=opener, **open_kwargs_extra,
) as model_file:
return load_model_from_file(model_file)
class FuelType(Enum):
PETROL = 'petrol'
DIESEL = 'diesel'
CNG = 'cng'
class SellingType(Enum):
DEALER = 'dealer'
INDIVIDUAL = 'individual'
class TransmissionType(Enum):
MANUAL = 'manual'
AUTOMATIC = 'automatic'
@dataclass
class PricePredictionFeatures:
selling_price: float
driven_kms: float
age: float
fuel_type: FuelType
selling_type: SellingType
transmission_type: TransmissionType
metric_prediction_latency = Histogram(
'model_prediction_seconds', 'Время вычислений в модели',
buckets=(
list(chain.from_iterable((v * (10 ** p) for v in (1, 2, 5)) for p in range(-4, (1 + 1))))
+ [float('+inf')]
),
)
metric_prediction_errors = Counter(
'model_prediction_errors_total', 'Ошибки вычислений в модели по типу', ('error_type',),
)
metric_prediction_value = Histogram(
'model_prediction_value', 'Предсказанное значение цены',
buckets=(
list(chain.from_iterable((v * (10 ** p) for v in (1, 2, 5)) for p in range(-1, (2 + 1))))
+ [float('+inf')]
),
)
class PricePredictor:
def __init__(self, model_path):
self._model = load_model_from_path(model_path)
def predict(self, features):
# WARN: порядок столбцов вроде имеет значение
features_df = DataFrame([{
'selling_price': features.selling_price,
'driven_kms': features.driven_kms,
'fuel_type': features.fuel_type.value,
'selling_type': features.selling_type.value,
'transmission': features.transmission_type.value,
'age': features.age,
}])
try:
with metric_prediction_latency.time():
predictions = self._model.predict(features_df)
except Exception as err:
metric_prediction_errors.labels(error_type=type(err).__name__).inc()
raise
assert len(predictions) == 1
value = float(predictions[0])
metric_prediction_value.observe(value)
return value

@ -1,7 +0,0 @@
fastapi ~=0.120.4
mlxtend ~=0.23.4
pandas >=2.3.1,<3
prometheus_client ~=0.23.1
prometheus_fastapi_instrumentator >=7.0.2,<8
scikit-learn >=1.7.2,<2
uvicorn ~=0.38.0

1
services/models/.gitignore поставляемый

@ -1 +0,0 @@
*.pkl

@ -1,83 +0,0 @@
from argparse import ArgumentParser
from pathlib import Path
from pickle import dump
from sys import exit as sys_exit, argv as sys_argv
from mlflow import set_tracking_uri, set_registry_uri
from mlflow.sklearn import load_model
MLFLOW_TRACKING_URI_DEFAULT = 'http://localhost:5000'
def open_file_for_model(file, *, buffering=-1, opener=None, **kwargs_extra):
open_kwargs_extra = {}
if 'closefd' in kwargs_extra:
open_kwargs_extra['closefd'] = kwargs_extra.pop('closefd')
if len(kwargs_extra) > 0:
raise TypeError(
'Unexpected keyword arguments given: {}'
.format(', '.join(map(repr, kwargs_extra.keys())))
)
return open(file, 'wb', buffering=buffering, opener=opener)
def dump_model_to_file(model, file):
return dump(model, file)
def dump_model_to_path(model, path, *, buffering=-1, opener=None, **kwargs_extra):
open_kwargs_extra = {}
for k in ('closefd',):
if k in kwargs_extra:
open_kwargs_extra[k] = kwargs_extra.pop(k)
if len(kwargs_extra) > 0:
raise TypeError(
'Unexpected keyword arguments given: {}'
.format(', '.join(map(repr, kwargs_extra.keys())))
)
with open_file_for_model(
path, buffering=buffering, opener=opener, **open_kwargs_extra,
) as model_file:
return dump_model_to_file(model, model_file)
def parse_args(argv):
parser = ArgumentParser(
description=(
'Скачать модель с tracking server MLFlow и сохранить в локальный файл pickle'
),
allow_abbrev=False,
exit_on_error=True,
)
model_ref_parser = parser.add_mutually_exclusive_group(required=True)
model_ref_parser.add_argument('-m', '--model', type=str, dest='model_uri')
model_ref_parser.add_argument('--run', type=str, dest='run_id')
parser.add_argument(
'--tracking-uri', default=MLFLOW_TRACKING_URI_DEFAULT, type=str, dest='tracking_uri',
)
parser.add_argument('--registry-uri', type=str, dest='registry_uri')
parser.add_argument('out_path', default=Path('.'), type=Path)
args = parser.parse_args(argv)
return args
def main(argv):
args = parse_args(argv)
set_tracking_uri(args.tracking_uri)
if args.registry_uri is not None:
set_registry_uri(args.registry_uri)
if args.model_uri is not None:
model_uri = args.model_uri
elif args.run_id is not None:
model_uri = f'runs:/{args.run_id}/model'
else:
assert False
return 1
model = load_model(model_uri)
dump_model_to_path(model, args.out_path)
return 0
if __name__ == '__main__':
sys_exit(int(main(sys_argv) or 0))

@ -1,15 +0,0 @@
global:
scrape_interval: 15s
scrape_timeout: 5s
scrape_configs:
- job_name: "prices_predictor"
static_configs:
- targets:
- "prices-predictor:8000"
scheme: http
metrics_path: "/metrics"
#relabel_configs:
# - source_labels: ["__address__"]
# target_labels: "instance"
Загрузка…
Отмена
Сохранить