Spaces:
Sleeping
Sleeping
File size: 4,201 Bytes
2c527f4 bac6ed4 2c527f4 bac6ed4 2c527f4 bac6ed4 2c527f4 bac6ed4 2c527f4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 | """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)
|