train_car_price_xgboost.py

print("SCRIPT STARTED - CAR PRICE PREDICTION")

import os
import json
import joblib
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

from xgboost import XGBRegressor

from sklearn.model_selection import train_test_split
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import OneHotEncoder
from sklearn.pipeline import Pipeline
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
from pathlib import Path

BASE_DIR = Path(__file__).resolve().parent


# =========================
# SETTINGS
# =========================

DATA_DIR = BASE_DIR / "archiveprices"
RESULTS_DIR = BASE_DIR / "results_xgboost_price"
RESULTS_DIR.mkdir(exist_ok=True)

MODEL_PATH = RESULTS_DIR / "car_price_xgboost_model.pkl"

CSV_FILES = {
    "audi.csv": "Audi",
    "bmw.csv": "BMW",
    "ford.csv": "Ford",
    "hyundi.csv": "Hyundai",
    "merc.csv": "Mercedes",
    "skoda.csv": "Skoda",
    "toyota.csv": "Toyota",
    "vauxhall.csv": "Vauxhall",
    "vw.csv": "Volkswagen"
}

TARGET = "price"

NUMERIC_FEATURES = [
    "year",
    "mileage",
    "tax",
    "mpg",
    "engineSize"
]

CATEGORICAL_FEATURES = [
    "brand",
    "model",
    "transmission",
    "fuelType"
]


# =========================
# LOAD AND MERGE DATASET
# =========================

all_dfs = []

for file_name, brand in CSV_FILES.items():
    file_path = os.path.join(DATA_DIR, file_name)

    if not os.path.exists(file_path):
        raise FileNotFoundError(f"Could not find file: {file_path}")

    temp_df = pd.read_csv(file_path)
    temp_df["brand"] = brand
    all_dfs.append(temp_df)

df = pd.concat(all_dfs, ignore_index=True)

print("\nMerged dataset shape:", df.shape)
print("\nColumns:")
print(df.columns)

print("\nFirst rows:")
print(df.head())

print("\nMissing values:")
print(df.isnull().sum())

print("\nBrand distribution:")
print(df["brand"].value_counts())


# =========================
# BASIC CLEANING
# =========================

# Some files use "tax", while Hyundai uses "tax(£)"
# We merge both into one common "tax" column
if "tax(£)" in df.columns:
    df["tax"] = df["tax"].fillna(df["tax(£)"])

needed_columns = CATEGORICAL_FEATURES + NUMERIC_FEATURES + [TARGET]
df = df[needed_columns].copy()

# Clean text columns
for col in CATEGORICAL_FEATURES:
    df[col] = df[col].astype(str).str.strip()

# Remove missing rows
df = df.dropna()

# Remove unrealistic / noisy values
df = df[df["price"] > 500]
df = df[df["price"] < 100000]
df = df[df["year"] >= 1990]
df = df[df["year"] <= 2025]
df = df[df["mileage"] >= 0]
df = df[df["mileage"] < 300000]
df = df[df["engineSize"] > 0]
df = df[df["engineSize"] < 8]

print("\nDataset shape after cleaning:", df.shape)
print("\nBrand distribution after cleaning:")
print(df["brand"].value_counts())

# Save cleaned dataset
df.to_csv(os.path.join(RESULTS_DIR, "clean_used_car_dataset.csv"), index=False)


# =========================
# BASIC DATASET GRAPHS
# =========================

plt.figure(figsize=(8, 5))
df["price"].hist(bins=50)
plt.title("Used Car Price Distribution")
plt.xlabel("Price")
plt.ylabel("Number of Cars")
plt.grid(True)
plt.tight_layout()
plt.savefig(os.path.join(RESULTS_DIR, "price_distribution.png"), dpi=300, bbox_inches="tight")
plt.show()

plt.figure(figsize=(8, 5))
df["brand"].value_counts().plot(kind="bar")
plt.title("Number of Cars per Brand")
plt.xlabel("Brand")
plt.ylabel("Count")
plt.xticks(rotation=45)
plt.grid(axis="y")
plt.tight_layout()
plt.savefig(os.path.join(RESULTS_DIR, "brand_distribution.png"), dpi=300, bbox_inches="tight")
plt.show()

plt.figure(figsize=(8, 5))
plt.scatter(df["mileage"], df["price"], alpha=0.25)
plt.title("Price vs Mileage")
plt.xlabel("Mileage")
plt.ylabel("Price")
plt.grid(True)
plt.tight_layout()
plt.savefig(os.path.join(RESULTS_DIR, "price_vs_mileage.png"), dpi=300, bbox_inches="tight")
plt.show()

plt.figure(figsize=(8, 5))
plt.scatter(df["year"], df["price"], alpha=0.25)
plt.title("Price vs Year")
plt.xlabel("Year")
plt.ylabel("Price")
plt.grid(True)
plt.tight_layout()
plt.savefig(os.path.join(RESULTS_DIR, "price_vs_year.png"), dpi=300, bbox_inches="tight")
plt.show()


# =========================
# TRAIN / TEST SPLIT
# =========================

X = df[CATEGORICAL_FEATURES + NUMERIC_FEATURES]
y = df[TARGET]

X_train, X_test, y_train, y_test = train_test_split(
    X,
    y,
    test_size=0.2,
    random_state=42
)

print("\nTrain shape:", X_train.shape)
print("Test shape:", X_test.shape)


# =========================
# BUILD XGBOOST PIPELINE
# =========================

try:
    onehot = OneHotEncoder(handle_unknown="ignore", sparse_output=False)
except TypeError:
    onehot = OneHotEncoder(handle_unknown="ignore", sparse=False)

preprocessor = ColumnTransformer(
    transformers=[
        ("categorical", onehot, CATEGORICAL_FEATURES),
        ("numeric", "passthrough", NUMERIC_FEATURES)
    ]
)

xgb_model = XGBRegressor(
    n_estimators=800,
    learning_rate=0.03,
    max_depth=6,
    subsample=0.85,
    colsample_bytree=0.85,
    objective="reg:squarederror",
    random_state=42,
    n_jobs=-1
)

model = Pipeline(
    steps=[
        ("preprocessor", preprocessor),
        ("regressor", xgb_model)
    ]
)


# =========================
# TRAIN MODEL
# =========================

print("\nTraining XGBoost model...")
model.fit(X_train, y_train)

print("\nTraining finished.")


# =========================
# EVALUATE MODEL
# =========================

y_pred = model.predict(X_test)

mae = mean_absolute_error(y_test, y_pred)
rmse = np.sqrt(mean_squared_error(y_test, y_pred))
r2 = r2_score(y_test, y_pred)

print("\nTest Results:")
print("MAE:", mae)
print("RMSE:", rmse)
print("R2 score:", r2)

metrics = {
    "MAE": float(mae),
    "RMSE": float(rmse),
    "R2_score": float(r2)
}

with open(os.path.join(RESULTS_DIR, "price_model_metrics.json"), "w") as f:
    json.dump(metrics, f, indent=4)


# =========================
# RESULT GRAPHS
# =========================

plt.figure(figsize=(8, 6))
plt.scatter(y_test, y_pred, alpha=0.35)
plt.plot([y_test.min(), y_test.max()], [y_test.min(), y_test.max()])
plt.title("Actual Price vs Predicted Price")
plt.xlabel("Actual Price")
plt.ylabel("Predicted Price")
plt.grid(True)
plt.tight_layout()
plt.savefig(os.path.join(RESULTS_DIR, "actual_vs_predicted_price.png"), dpi=300, bbox_inches="tight")
plt.show()

residuals = y_test - y_pred

plt.figure(figsize=(8, 5))
plt.scatter(y_pred, residuals, alpha=0.35)
plt.axhline(0)
plt.title("Residual Plot")
plt.xlabel("Predicted Price")
plt.ylabel("Residual: Actual - Predicted")
plt.grid(True)
plt.tight_layout()
plt.savefig(os.path.join(RESULTS_DIR, "residual_plot.png"), dpi=300, bbox_inches="tight")
plt.show()

plt.figure(figsize=(8, 5))
plt.hist(residuals, bins=50)
plt.title("Residual Distribution")
plt.xlabel("Prediction Error")
plt.ylabel("Count")
plt.grid(True)
plt.tight_layout()
plt.savefig(os.path.join(RESULTS_DIR, "residual_distribution.png"), dpi=300, bbox_inches="tight")
plt.show()


# =========================
# FEATURE IMPORTANCE
# =========================

trained_preprocessor = model.named_steps["preprocessor"]
trained_xgb = model.named_steps["regressor"]

cat_feature_names = trained_preprocessor.named_transformers_["categorical"].get_feature_names_out(CATEGORICAL_FEATURES)
feature_names = list(cat_feature_names) + NUMERIC_FEATURES

importances = trained_xgb.feature_importances_

importance_df = pd.DataFrame({
    "feature": feature_names,
    "importance": importances
}).sort_values(by="importance", ascending=False)

importance_df.to_csv(os.path.join(RESULTS_DIR, "feature_importance.csv"), index=False)

top_importance = importance_df.head(20)

plt.figure(figsize=(10, 7))
plt.barh(top_importance["feature"][::-1], top_importance["importance"][::-1])
plt.title("Top 20 Feature Importances - XGBoost")
plt.xlabel("Importance")
plt.ylabel("Feature")
plt.tight_layout()
plt.savefig(os.path.join(RESULTS_DIR, "feature_importance_top20.png"), dpi=300, bbox_inches="tight")
plt.show()


# =========================
# SAVE MODEL
# =========================

joblib.dump(model, MODEL_PATH)
print("\nSaved XGBoost price model:", MODEL_PATH)


# =========================
# EXAMPLE CLEAN PRICE PREDICTION
# =========================

example_car = pd.DataFrame([{
    "brand": "Ford",
    "model": "Focus",
    "transmission": "Manual",
    "fuelType": "Petrol",
    "year": 2016,
    "mileage": 60000,
    "tax": 145,
    "mpg": 55.0,
    "engineSize": 1.6
}])

clean_price = model.predict(example_car)[0]

print("\nExample clean used-car price prediction:")
print(example_car)
print("Predicted clean price:", round(clean_price, 2))


# =========================
# DAMAGE PRICE ADJUSTMENT EXAMPLE
# =========================
# Later these values will come from your ResNet50 damage model.

damage_probs = {
    "dent": 0.90,
    "scratch": 0.80,
    "crack": 0.10,
    "glass_shatter": 0.05,
    "lamp_broken": 0.20,
    "tire_flat": 0.05
}

damage_weights = {
    "scratch": 0.08,
    "dent": 0.12,
    "crack": 0.15,
    "glass_shatter": 0.25,
    "lamp_broken": 0.15,
    "tire_flat": 0.10
}

damage_reduction = 0

for label, prob in damage_probs.items():
    damage_reduction += prob * damage_weights[label]

# Cap the reduction so the formula does not become too extreme
damage_reduction = min(damage_reduction, 0.50)

damaged_price = clean_price * (1 - damage_reduction)

print("\nDamage adjustment example:")
print("Damage probabilities:", damage_probs)
print("Damage reduction:", round(damage_reduction * 100, 2), "%")
print("Final damaged price:", round(damaged_price, 2))

print("\nSCRIPT FINISHED")