Modelni ishlab chiqarishga chiqarish

import joblib
from pathlib import Path

Path("modellar").mkdir(exist_ok=True)

joblib.dump(quvur, "modellar/uy_narxi_v1.joblib", compress=3)

model = joblib.load("modellar/uy_narxi_v1.joblib")

import json
import sklearn
import numpy as np
from datetime import datetime

metama = {
    "nom": "uy_narxi",
    "versiya": "1.0.0",
    "yaratilgan": datetime.now().isoformat(),
    "sklearn_versiyasi": sklearn.__version__,
    "python_versiyasi": "3.12.4",
    "belgilar": list(X.columns),
    "nishon": "narx",
    "model_turi": type(quvur[-1]).__name__,
    "giperparametrlar": qidiruv.best_params_,
    "natijalar": {
        "cv_rmse": float(cv_rmse),
        "sinov_rmse": float(sinov_rmse),
        "sinov_r2": float(sinov_r2),
    },
    "oqitish_namunalari": int(len(X_o)),
}

with open("modellar/uy_narxi_v1.json", "w", encoding="utf-8") as fayl:
    json.dump(metama, fayl, ensure_ascii=False, indent=2)

class BashoratXatosi(Exception):
    pass


def tekshirish(malumot: dict, kutilgan_belgilar: list) -> None:
    """Kirish ma'lumotini modelga berishdan oldin tekshiradi."""

    yetishmayotgan = set(kutilgan_belgilar) - set(malumot)
    if yetishmayotgan:
        raise BashoratXatosi(f"Belgilar yetishmayapti: {yetishmayotgan}")

    chegaralar = {
        "maydon":  (10, 1000),
        "xonalar": (1, 20),
        "yosh":    (0, 150),
    }

    for belgi, (eng_kam, eng_kop) in chegaralar.items():
        qiymat = malumot.get(belgi)

        if qiymat is None:
            continue

        if not isinstance(qiymat, (int, float)):
            raise BashoratXatosi(f"{belgi} son bo'lishi kerak")

        if not eng_kam <= qiymat <= eng_kop:
            raise BashoratXatosi(
                f"{belgi} qiymati {eng_kam}-{eng_kop} oralig'ida bo'lishi kerak, "
                f"berilgan: {qiymat}"
            )

# api.py
from flask import Flask, request, jsonify
import joblib
import pandas as pd
import logging
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[logging.FileHandler("jurnal/bashoratlar.log"), logging.StreamHandler()],
)

ilova = Flask(__name__)
model = joblib.load("modellar/uy_narxi_v1.joblib")

with open("modellar/uy_narxi_v1.json", encoding="utf-8") as fayl:
    metama = json.load(fayl)

BELGILAR = metama["belgilar"]


@ilova.route("/salomatlik", methods=["GET"])
def salomatlik():
    return jsonify({
        "holat": "ishlayapti",
        "model": metama["nom"],
        "versiya": metama["versiya"],
    })


@ilova.route("/bashorat", methods=["POST"])
def bashorat():
    boshlandi = datetime.now()

    try:
        malumot = request.get_json()

        if malumot is None:
            return jsonify({"xato": "JSON kutilgan edi"}), 400

        tekshirish(malumot, BELGILAR)

        qator = pd.DataFrame([malumot])[BELGILAR]
        natija = float(model.predict(qator)[0])

        davomiylik = (datetime.now() - boshlandi).total_seconds() * 1000

        logging.info(
            "bashorat kirish=%s natija=%.2f vaqt=%.1fms",
            malumot, natija, davomiylik,
        )

        return jsonify({
            "bashorat": round(natija, 2),
            "birlik": "mln so'm",
            "model_versiyasi": metama["versiya"],
        })

    except BashoratXatosi as xato:
        logging.warning("noto'g'ri kirish: %s", xato)
        return jsonify({"xato": str(xato)}), 400

    except Exception as xato:
        logging.exception("kutilmagan xato")
        return jsonify({"xato": "Serverda xatolik yuz berdi"}), 500


if __name__ == "__main__":
    ilova.run(host="0.0.0.0", port=5000)

curl -X POST http://localhost:5000/bashorat \
  -H "Content-Type: application/json" \
  -d '{"maydon": 85, "xonalar": 3, "yosh": 7, "tuman": "Chilonzor", "holati": "yangi"}'

@ilova.route("/bashorat/toplam", methods=["POST"])
def toplam_bashorat():
    malumotlar = request.get_json()

    if not isinstance(malumotlar, list):
        return jsonify({"xato": "Massiv kutilgan edi"}), 400

    if len(malumotlar) > 1000:
        return jsonify({"xato": "Bir so'rovda 1000 tadan ko'p bo'lmasin"}), 400

    jadval = pd.DataFrame(malumotlar)[BELGILAR]
    natijalar = model.predict(jadval)

    return jsonify({"bashoratlar": [round(float(n), 2) for n in natijalar]})

FROM python:3.12-slim

WORKDIR /ilova

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY modellar/ modellar/
COPY api.py .

RUN mkdir -p jurnal

EXPOSE 5000

HEALTHCHECK --interval=30s --timeout=3s \
  CMD python -c "import requests; requests.get('http://localhost:5000/salomatlik')"

CMD ["gunicorn", "--bind", "0.0.0.0:5000", "--workers", "4", "api:ilova"]

docker build -t uy-narxi:1.0.0 .
docker run -p 5000:5000 uy-narxi:1.0.0

from scipy.stats import ks_2samp

def siljishni_tekshir(oqitish_malumoti, yangi_malumot, belgilar, chegara=0.05):
    """Kolmogorov-Smirnov testi bilan taqsimot o'zgarishini tekshiradi."""

    natijalar = {}

    for belgi in belgilar:
        statistika, p_qiymat = ks_2samp(
            oqitish_malumoti[belgi].dropna(),
            yangi_malumot[belgi].dropna(),
        )

        natijalar[belgi] = {
            "statistika": round(float(statistika), 4),
            "p_qiymat": round(float(p_qiymat), 4),
            "siljish_bor": bool(p_qiymat < chegara),
        }

    return natijalar


natija = siljishni_tekshir(X_oqitish, X_yangi, ["maydon", "yosh", "narx_m2"])

for belgi, holat in natija.items():
    belgi_matni = "SILJISH" if holat["siljish_bor"] else "normal"
    print(f"{belgi:12s} p={holat['p_qiymat']:.4f}  {belgi_matni}")

import time
from collections import deque

class Kuzatuvchi:
    def __init__(self, oyna=1000):
        self.bashoratlar = deque(maxlen=oyna)
        self.vaqtlar = deque(maxlen=oyna)
        self.xatolar = 0
        self.jami = 0

    def yozing(self, bashorat, davomiylik_ms, xato=False):
        self.jami += 1

        if xato:
            self.xatolar += 1
            return

        self.bashoratlar.append(bashorat)
        self.vaqtlar.append(davomiylik_ms)

    def hisobot(self):
        if not self.vaqtlar:
            return {"holat": "ma'lumot yo'q"}

        vaqtlar = sorted(self.vaqtlar)

        return {
            "jami_sorov": self.jami,
            "xato_foizi": round(self.xatolar / max(self.jami, 1) * 100, 2),
            "ortacha_vaqt_ms": round(sum(vaqtlar) / len(vaqtlar), 1),
            "p95_vaqt_ms": round(vaqtlar[int(len(vaqtlar) * 0.95)], 1),
            "ortacha_bashorat": round(sum(self.bashoratlar) / len(self.bashoratlar), 2),
        }


kuzatuvchi = Kuzatuvchi()

import random

class ModelTanlovchi:
    def __init__(self, eski_model, yangi_model, yangi_ulush=0.1):
        self.eski = eski_model
        self.yangi = yangi_model
        self.yangi_ulush = yangi_ulush

    def bashorat(self, malumot, foydalanuvchi_id):
        # Bir foydalanuvchi har doim bir xil modelga tushsin
        tasodif = random.Random(foydalanuvchi_id).random()

        if tasodif < self.yangi_ulush:
            return self.yangi.predict(malumot), "yangi"

        return self.eski.predict(malumot), "eski"

# test_model.py
import pytest
import joblib
import pandas as pd

@pytest.fixture
def model():
    return joblib.load("modellar/uy_narxi_v1.joblib")


def test_model_yuklandi(model):
    assert model is not None
    assert hasattr(model, "predict")


def test_bashorat_musbat(model):
    uy = pd.DataFrame([{
        "maydon": 85, "xonalar": 3, "yosh": 7,
        "tuman": "Chilonzor", "holati": "yangi",
    }])

    natija = model.predict(uy)[0]
    assert natija > 0
    assert 100 < natija < 10000


def test_katta_uy_qimmatroq(model):
    kichik = pd.DataFrame([{
        "maydon": 50, "xonalar": 2, "yosh": 10,
        "tuman": "Chilonzor", "holati": "o'rta",
    }])
    katta = kichik.copy()
    katta["maydon"] = 120
    katta["xonalar"] = 4

    assert model.predict(katta)[0] > model.predict(kichik)[0]


def test_yangi_uy_qimmatroq(model):
    eski = pd.DataFrame([{
        "maydon": 80, "xonalar": 3, "yosh": 40,
        "tuman": "Chilonzor", "holati": "eski",
    }])
    yangi = eski.copy()
    yangi["yosh"] = 1
    yangi["holati"] = "yangi"

    assert model.predict(yangi)[0] > model.predict(eski)[0]


def test_sifat_chegarasi(model):
    X_sinov = pd.read_csv("malumotlar/sinov.csv")
    y_sinov = X_sinov.pop("narx")

    from sklearn.metrics import r2_score
    r2 = r2_score(y_sinov, model.predict(X_sinov))

    assert r2 > 0.85, f"Model sifati tushdi: R2={r2:.3f}"

ml-loyiha/
├── malumotlar/
│   ├── xom/
│   └── tayyor/
├── daftarlar/
│   ├── 01-tahlil.ipynb
│   └── 02-tajribalar.ipynb
├── manba/
│   ├── tayyorlash.py
│   ├── oqitish.py
│   ├── baholash.py
│   └── bashorat.py
├── modellar/
│   ├── uy_narxi_v1.joblib
│   └── uy_narxi_v1.json
├── testlar/
│   └── test_model.py
├── jurnal/
├── api.py
├── Dockerfile
├── requirements.txt
├── .gitignore
└── README.md

import mlflow
import mlflow.sklearn

mlflow.set_experiment("uy-narxi")

with mlflow.start_run(run_name="rf-v3"):
    mlflow.log_params(qidiruv.best_params_)

    mlflow.log_metrics({
        "cv_rmse": cv_rmse,
        "sinov_rmse": sinov_rmse,
        "sinov_r2": sinov_r2,
    })

    mlflow.sklearn.log_model(quvur, "model")
    mlflow.log_artifact("rasmlar/qoldiqlar.png")

mlflow ui