predict.py

#!/usr/bin/env python
"""
Run predictions using the trained model.

Usage:
    python predict.py                           # Predict on test set, print metrics
    python predict.py --input data.csv          # Predict on custom CSV
    python predict.py --single                  # Interactive single prediction
"""

from __future__ import annotations

import argparse
import json
import pickle
import sys
from pathlib import Path

import numpy as np
import pandas as pd
from loguru import logger

logger.remove()
logger.add(sys.stderr, format="<green>{time:HH:mm:ss}</green> | <level>{level:<8}</level> | {message}", level="INFO")

ROOT = Path(__file__).resolve().parent
MODEL_PATH = ROOT / "models" / "best_model.pkl"
FEATURES_PATH = ROOT / "models" / "feature_names.json"
ENCODINGS_PATH = ROOT / "data" / "processed" / "encodings.json"


def load_model():
    with open(MODEL_PATH, "rb") as f:
        model = pickle.load(f)
    with open(FEATURES_PATH) as f:
        feature_names = json.load(f)
    with open(ENCODINGS_PATH) as f:
        encodings = json.load(f)
    return model, feature_names, encodings


def predict_test_set():
    """Run predictions on the held-out test set and print metrics."""
    from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score

    model, features, encodings = load_model()
    test_df = pd.read_csv(ROOT / "data" / "processed" / "test.csv")

    X_test = test_df[features]
    y_test = test_df["rent_price_qar"]

    pred_log = model.predict(X_test)
    pred_qar = np.expm1(pred_log)

    mae = mean_absolute_error(y_test, pred_qar)
    rmse = np.sqrt(mean_squared_error(y_test, pred_qar))
    r2 = r2_score(y_test, pred_qar)
    mape = np.mean(np.abs((y_test - pred_qar) / y_test)) * 100
    median_ae = np.median(np.abs(y_test - pred_qar))

    logger.info("Test Set Predictions")
    logger.info(f"  Samples:    {len(y_test)}")
    logger.info(f"  R-squared:  {r2:.4f}")
    logger.info(f"  MAE:        {mae:,.0f} QAR")
    logger.info(f"  RMSE:       {rmse:,.0f} QAR")
    logger.info(f"  MAPE:       {mape:.1f}%")
    logger.info(f"  Median AE:  {median_ae:,.0f} QAR")

    results = pd.DataFrame({
        "actual_qar": y_test.values,
        "predicted_qar": pred_qar.round(0).astype(int),
        "error_qar": (pred_qar - y_test.values).round(0).astype(int),
    })
    output_path = ROOT / "data" / "processed" / "predictions.csv"
    results.to_csv(output_path, index=False)
    logger.info(f"  Saved predictions to {output_path}")


def predict_file(input_path: str):
    """Run predictions on a custom CSV file."""
    model, features, encodings = load_model()
    df = pd.read_csv(input_path)

    missing = [f for f in features if f not in df.columns]
    if missing:
        logger.error(f"Missing features in input: {missing}")
        sys.exit(1)

    X = df[features]
    pred_log = model.predict(X)
    pred_qar = np.expm1(pred_log)

    df["predicted_rent_qar"] = pred_qar.round(0).astype(int)

    output_path = Path(input_path).stem + "_predictions.csv"
    df.to_csv(output_path, index=False)
    logger.info(f"Saved {len(df)} predictions to {output_path}")


def predict_single():
    """Interactive single property prediction."""
    model, features, encodings = load_model()
    medians = encodings.get("medians", {})
    te_mappings = encodings.get("target_encodings", {})

    print("\n--- Qatar Rent Price Predictor ---\n")

    bedrooms = int(input("Bedrooms (0-7): ") or "2")
    bathrooms = int(input("Bathrooms (1-5): ") or "2")
    area = float(input("Area in sqm (50-500): ") or "120")
    prop_type = input("Property type (apartment/villa/studio/townhouse/penthouse): ") or "apartment"
    neighborhood = input("Neighborhood (e.g. The Pearl Island, West Bay, Al Sadd): ") or "Al Sadd"
    furnished = input("Furnished? (yes/no): ").lower().startswith("y")

    row = {f: medians.get(f, 0) for f in features}

    row["bedrooms"] = bedrooms
    row["bathrooms"] = bathrooms
    row["area_sqm"] = area
    row["log_area"] = np.log1p(area)
    row["is_furnished"] = int(furnished)
    row["amenity_score"] = 4
    row["log_beds"] = np.log1p(bedrooms)
    row["beds_x_area"] = bedrooms * area
    row["baths_x_area"] = bathrooms * area
    row["beds_x_baths"] = bedrooms * bathrooms
    row["rooms_per_sqm"] = (bedrooms + bathrooms) / max(area, 1)
    row["furnished_x_area"] = int(furnished) * area

    if "neighborhood_te" in features and "neighborhood" in te_mappings:
        row["neighborhood_te"] = te_mappings["neighborhood"].get(
            neighborhood, np.mean(list(te_mappings["neighborhood"].values()))
        )
    if "sub_neighborhood_te" in features and "sub_neighborhood" in te_mappings:
        row["sub_neighborhood_te"] = np.mean(list(te_mappings["sub_neighborhood"].values()))

    type_map = {"apartment": 0, "compound": 1, "duplex": 2, "other": 3,
                "penthouse": 4, "studio": 5, "townhouse": 6, "villa": 7, "whole_building": 8}
    row["property_type_encoded"] = type_map.get(prop_type, 0)

    X = pd.DataFrame([row])[features]
    pred_log = model.predict(X)[0]
    pred_qar = np.expm1(pred_log)

    print(f"\n  Predicted monthly rent: {pred_qar:,.0f} QAR")
    print(f"  ({pred_qar * 12:,.0f} QAR/year)\n")


def main():
    parser = argparse.ArgumentParser(description="Qatar Rent Price Predictor")
    parser.add_argument("--input", type=str, help="CSV file to predict on")
    parser.add_argument("--single", action="store_true", help="Interactive single prediction")
    args = parser.parse_args()

    if args.single:
        predict_single()
    elif args.input:
        predict_file(args.input)
    else:
        predict_test_set()


if __name__ == "__main__":
    main()