from fastapi import APIRouter, Request from pydantic import BaseModel, Field from typing import Optional, Dict, Any from catboost import CatBoostClassifier import pandas as pd import numpy as np from routes.predict_bag_of_heroes import predict_bag_of_heroes from routes.predict_with_players import predict_with_players router = APIRouter() # ========================= # Загрузка модели # ========================= modelPro = CatBoostClassifier() modelPro.load_model("artifacts/model_from_db_pro_v3.cbm") # ========================= # Загрузка порядка фич # ========================= def load_feature_order(path: str) -> list[str]: fo = pd.read_csv(path) first_col = fo.columns[0] return fo[first_col].tolist() FEATURE_ORDER_PRO: list[str] = load_feature_order("artifacts/feature_order_db.csv") # ========================= # Дефолты для недостающих фич # ========================= DEFAULTS: Dict[str, Any] = { "is_first_pick_radiant": 0, # Radiant heroes "r_h1": -1, "r_h2": -1, "r_h3": -1, "r_h4": -1, "r_h5": -1, # Dire heroes "d_h1": -1, "d_h2": -1, "d_h3": -1, "d_h4": -1, "d_h5": -1, # # Radiant players "r_p1": -1, "r_p2": -1, "r_p3": -1, "r_p4": -1, "r_p5": -1, # # Dire players "d_p1": -1, "d_p2": -1, "d_p3": -1, "d_p4": -1, "d_p5": -1, # Radiant positions "rp_h1": -1, "rp_h2": -1, "rp_h3": -1, "rp_h4": -1, "rp_h5": -1, # Dire positions "dp_h1": -1, "dp_h2": -1, "dp_h3": -1, "dp_h4": -1, "dp_h5": -1, } # ========================= # Входная схема # ========================= class DraftPayload(BaseModel): # флаг первого пика (0 — Dire first pick/неизвестно, 1 — Radiant first pick) is_first_pick_radiant: Optional[int] = Field(default=DEFAULTS["is_first_pick_radiant"]) # герои (IDs) r_h1: Optional[int] = Field(default=DEFAULTS["r_h1"]) r_h2: Optional[int] = Field(default=DEFAULTS["r_h2"]) r_h3: Optional[int] = Field(default=DEFAULTS["r_h3"]) r_h4: Optional[int] = Field(default=DEFAULTS["r_h4"]) r_h5: Optional[int] = Field(default=DEFAULTS["r_h5"]) d_h1: Optional[int] = Field(default=DEFAULTS["d_h1"]) d_h2: Optional[int] = Field(default=DEFAULTS["d_h2"]) d_h3: Optional[int] = Field(default=DEFAULTS["d_h3"]) d_h4: Optional[int] = Field(default=DEFAULTS["d_h4"]) d_h5: Optional[int] = Field(default=DEFAULTS["d_h5"]) # игроки (IDs) r_p1: Optional[int] = Field(default=DEFAULTS["r_p1"]) r_p2: Optional[int] = Field(default=DEFAULTS["r_p2"]) r_p3: Optional[int] = Field(default=DEFAULTS["r_p3"]) r_p4: Optional[int] = Field(default=DEFAULTS["r_p4"]) r_p5: Optional[int] = Field(default=DEFAULTS["r_p5"]) d_p1: Optional[int] = Field(default=DEFAULTS["d_p1"]) d_p2: Optional[int] = Field(default=DEFAULTS["d_p2"]) d_p3: Optional[int] = Field(default=DEFAULTS["d_p3"]) d_p4: Optional[int] = Field(default=DEFAULTS["d_p4"]) d_p5: Optional[int] = Field(default=DEFAULTS["d_p5"]) # позиции героев (1-5) rp_h1: Optional[int] = Field(default=DEFAULTS["rp_h1"]) rp_h2: Optional[int] = Field(default=DEFAULTS["rp_h2"]) rp_h3: Optional[int] = Field(default=DEFAULTS["rp_h3"]) rp_h4: Optional[int] = Field(default=DEFAULTS["rp_h4"]) rp_h5: Optional[int] = Field(default=DEFAULTS["rp_h5"]) dp_h1: Optional[int] = Field(default=DEFAULTS["dp_h1"]) dp_h2: Optional[int] = Field(default=DEFAULTS["dp_h2"]) dp_h3: Optional[int] = Field(default=DEFAULTS["dp_h3"]) dp_h4: Optional[int] = Field(default=DEFAULTS["dp_h4"]) dp_h5: Optional[int] = Field(default=DEFAULTS["dp_h5"]) # ========================= # Хелперы # ========================= def build_long_format_input(payload: dict) -> pd.DataFrame: """ Конвертирует payload в hero+position combination features для модели. Создаёт бинарные признаки вида radiant_h{hero_id}_p{position} и dire_h{hero_id}_p{position} """ features = {} # Инициализируем все признаки нулями for feat in FEATURE_ORDER_PRO: features[feat] = 0 # Radiant heroes с позициями for i in range(1, 6): hero_id = int(payload.get(f"r_h{i}", -1)) position = int(payload.get(f"rp_h{i}", -1)) if hero_id >= 0 and position >= 0: feature_name = f"radiant_h{hero_id}_p{position}" if feature_name in features: features[feature_name] = 1 # Dire heroes с позициями for i in range(1, 6): hero_id = int(payload.get(f"d_h{i}", -1)) position = int(payload.get(f"dp_h{i}", -1)) if hero_id >= 0 and position >= 0: feature_name = f"dire_h{hero_id}_p{position}" if feature_name in features: features[feature_name] = 1 # Создаём DataFrame с одной строкой в правильном порядке df = pd.DataFrame([features], columns=FEATURE_ORDER_PRO) return df def proba_percent(p: float) -> float: """Перевод вероятности в проценты (0..100) с отсечкой.""" return round(float(np.clip(p * 100.0, 0.0, 100.0))) # ========================= # Роут # ========================= @router.post("/draft/predict") async def predict(request: Request): body = await request.json() # Конвертируем все значения героев, игроков и позиций в int for key in body: if key.startswith(("r_h", "d_h", "is_first_pick_radiant")): if body[key] is not None and body[key] != "": try: body[key] = int(body[key]) except (ValueError, TypeError): body[key] = -1 else: body[key] = -1 elif key.startswith(("rp_h", "dp_h")): if body[key] == 0: body[key] = -1 else: body[key] = -1 # Hero+position combination предсказание X_pro = build_long_format_input(body) # Получаем предсказание для матча (одна строка) radiant_pro = float(modelPro.predict_proba(X_pro)[0, 1]) rp = proba_percent(radiant_pro) rd = 100.0 - rp # Предсказание bag-of-heroes модели bag_prediction = predict_bag_of_heroes(body) # Предсказание модели с игроками players_prediction = predict_with_players(body) # Предсказание стекинг модели (ленивый импорт для избежания циклической зависимости) try: from routes.predict_stacking import predict_stacking stacking_prediction = predict_stacking(body) except Exception: stacking_prediction = {"radiant_win": 50, "dire_win": 50} return { "pro-with-pos": { "radiant_win": rp, "dire_win": rd }, "pro": { "radiant_win": bag_prediction["radiant_win"], "dire_win": bag_prediction["dire_win"] }, "with-players": { "radiant_win": players_prediction["radiant_win"], "dire_win": players_prediction["dire_win"] }, "stacking": { "radiant_win": stacking_prediction["radiant_win"], "dire_win": stacking_prediction["dire_win"] } }