src/price_predictor.py

"""Inference helper for used-car price estimation."""

from __future__ import annotations

import json
from functools import lru_cache
from typing import Any

import joblib
import numpy as np

from src.config import MODEL_METADATA_PATH, PREPROCESSOR_PATH, PRICE_MODEL_PATH
from src.data_preprocessing import prepare_inference_input
from src.utils import to_float


def _heuristic_estimate(input_data: dict[str, Any]) -> dict[str, Any]:
    base = 26000.0
    age = to_float(input_data.get("age"), 6)
    km = to_float(input_data.get("km"), 90000)
    hp_kw = to_float(input_data.get("hp_kW"), 100)

    base -= age * 1500
    base -= (km / 1000) * 35
    base += (hp_kw - 100) * 45

    # Extract brand from make_model for adjustments
    make_model = str(input_data.get("make_model", "")).lower()
    luxury_adjustments = {
        "rolls": 70000,
        "bentley": 45000,
        "aston": 35000,
        "porsche": 30000,
        "mercedes": 12000,
        "bmw": 10000,
        "audi": 9000,
        "jaguar": 9000,
        "cadillac": 7000,
    }
    budget_adjustments = {
        "swift": -4000,
        "fiat": -2500,
        "hyundai": -1000,
        "kia": -1000,
        "mahindra": -500,
        "toyota": 2500,
        "mazda": 1500,
        "ford": 500,
        "vw": 2500,
    }

    applied = False
    for key, delta in luxury_adjustments.items():
        if key in make_model:
            base += delta
            applied = True
            break

    if not applied:
        for key, delta in budget_adjustments.items():
            if key in make_model:
                base += delta
                applied = True
                break

    if not applied and ("suv" in make_model or "benz" in make_model):
        base += 3500

    estimated = max(5000, min(180000, base))
    spread = max(2500, estimated * 0.12)

    return {
        "estimated_price": round(float(estimated), 2),
        "lower_bound": round(float(estimated - spread), 2),
        "upper_bound": round(float(estimated + spread), 2),
        "model_name": "HeuristicFallback",
        "confidence_note": "Estimated range from a simple age-and-mileage fallback because trained model artifacts are unavailable.",
    }


@lru_cache(maxsize=1)
def _load_model_bundle() -> dict[str, Any] | None:
    try:
        model = joblib.load(PRICE_MODEL_PATH)
        preprocessor = joblib.load(PREPROCESSOR_PATH)

        metadata = {}
        if MODEL_METADATA_PATH.exists():
            metadata = json.loads(MODEL_METADATA_PATH.read_text(encoding="utf-8"))

        return {
            "model": model,
            "preprocessor": preprocessor,
            "metadata": metadata,
        }
    except Exception:
        return None


def predict_price(input_data: dict[str, Any]) -> dict[str, Any]:
    """Predict used-car price and return estimate range."""
    model_bundle = _load_model_bundle()
    if model_bundle is None:
        return _heuristic_estimate(input_data)

    try:
        df_features = prepare_inference_input(input_data)
        X = model_bundle["preprocessor"].transform(df_features)
        prediction = float(model_bundle["model"].predict(X)[0])

        metadata = model_bundle.get("metadata", {})
        model_name = metadata.get("model_name", model_bundle["model"].__class__.__name__)
        rmse = (
            metadata.get("best_metrics", {}).get("rmse")
            if isinstance(metadata.get("best_metrics"), dict)
            else None
        )

        if rmse is None:
            spread = max(3000.0, prediction * 0.12)
            confidence_note = "Estimated range based on generic uncertainty band."
        else:
            spread = max(float(rmse), prediction * 0.08)
            confidence_note = "Estimated range based on validation RMSE of the trained model."

        lower = max(1000.0, prediction - spread)
        upper = prediction + spread

        return {
            "estimated_price": round(prediction, 2),
            "lower_bound": round(lower, 2),
            "upper_bound": round(upper, 2),
            "model_name": model_name,
            "confidence_note": confidence_note,
        }
    except Exception:
        return _heuristic_estimate(input_data)