src/train.py

"""Train and compare rating-prediction models (the ML Numeric block).

- Target: average_rating (a book's community reception).
- Features: book metadata + popularity + the NLP-derived sentiment feature + genre one-hots.
- Compares Ridge / RandomForest / GradientBoosting (+ XGBoost if installed).
- Saves: best pipeline (rating_model.joblib), a comparison report, an error-analysis file,
  and writes predicted_rating back into the catalog -> this is the ML output the ranking
  and LLM-explanation blocks consume.

Run:  python -m src.train
"""
import warnings

import joblib
import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.ensemble import GradientBoostingRegressor, RandomForestRegressor
from sklearn.impute import SimpleImputer
from sklearn.linear_model import Ridge
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
from sklearn.model_selection import cross_val_score, train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

from src import config as cfg

try:
    from xgboost import XGBRegressor
    HAS_XGB = True
except Exception:
    HAS_XGB = False

NUMERIC = ["num_pages", "ratings_count", "text_reviews_count", "publication_year", "sentiment_compound"]
CATEGORICAL = ["language_code"]


def load_xy():
    df = pd.read_parquet(cfg.TRAINING_PARQUET)
    # log-transform highly skewed count features
    for c in ["ratings_count", "text_reviews_count"]:
        if c in df.columns:
            df[c] = np.log1p(df[c])
    df = df.dropna(subset=[cfg.TARGET]).reset_index(drop=True)

    genre_cols = [c for c in df.columns if c.startswith("genre_") and c != "genre_str"]
    features = (
        [c for c in NUMERIC if c in df.columns]
        + [c for c in CATEGORICAL if c in df.columns]
        + genre_cols
    )
    return df, df[features], df[cfg.TARGET], features, genre_cols


def build_preprocessor(features, genre_cols):
    num = [c for c in NUMERIC if c in features]
    cat = [c for c in CATEGORICAL if c in features]
    transformers = []
    if num:
        transformers.append(("num", Pipeline([("imp", SimpleImputer(strategy="median")),
                                               ("sc", StandardScaler())]), num))
    if cat:
        transformers.append(("cat", Pipeline([("imp", SimpleImputer(strategy="most_frequent")),
                                               ("oh", OneHotEncoder(handle_unknown="ignore",
                                                                    sparse_output=False))]), cat))
    if genre_cols:
        transformers.append(("genre", "passthrough", genre_cols))
    return ColumnTransformer(transformers)


def main():
    warnings.filterwarnings("ignore")
    df, X, y, features, genre_cols = load_xy()
    print(f"Training on {len(df)} books, {len(features)} features.")

    Xtr, Xte, ytr, yte = train_test_split(X, y, test_size=cfg.TEST_SIZE, random_state=cfg.RANDOM_STATE)
    pre = build_preprocessor(features, genre_cols)

    models = {
        "Ridge": Ridge(alpha=1.0),
        "RandomForest": RandomForestRegressor(n_estimators=300, random_state=cfg.RANDOM_STATE, n_jobs=-1),
        "GradientBoosting": GradientBoostingRegressor(random_state=cfg.RANDOM_STATE),
    }
    if HAS_XGB:
        models["XGBoost"] = XGBRegressor(
            n_estimators=400, learning_rate=0.05, max_depth=4, subsample=0.8,
            random_state=cfg.RANDOM_STATE, n_jobs=-1,
        )

    rows, fitted = [], {}
    for name, est in models.items():
        pipe = Pipeline([("pre", pre), ("model", est)])
        pipe.fit(Xtr, ytr)
        pred = pipe.predict(Xte)
        rmse = float(np.sqrt(mean_squared_error(yte, pred)))
        mae = float(mean_absolute_error(yte, pred))
        r2 = float(r2_score(yte, pred))
        cv = cross_val_score(pipe, X, y, cv=5, scoring="neg_mean_squared_error")
        cv_rmse = float(np.sqrt(-cv.mean()))
        rows.append({"model": name, "RMSE": rmse, "MAE": mae, "R2": r2, "CV_RMSE": cv_rmse})
        fitted[name] = pipe
        print(f"{name:16s} RMSE={rmse:.3f}  MAE={mae:.3f}  R2={r2:.3f}  CV_RMSE={cv_rmse:.3f}")

    report = pd.DataFrame(rows).sort_values("RMSE").reset_index(drop=True)
    report.to_csv(cfg.REPORTS_DIR / "model_comparison.csv", index=False)
    best_name = report.iloc[0]["model"]
    best = fitted[best_name]
    joblib.dump(best, cfg.RATING_MODEL)
    print(f"\nBest model: {best_name}  ->  {cfg.RATING_MODEL}")

    # --- Error analysis: worst predictions + simple residual stats -----------
    pred_te = best.predict(Xte)
    err = pd.DataFrame({"y_true": yte.values, "y_pred": pred_te})
    err["residual"] = err["y_true"] - err["y_pred"]
    err["abs_err"] = err["residual"].abs()
    err.sort_values("abs_err", ascending=False).head(25).to_csv(
        cfg.REPORTS_DIR / "worst_predictions.csv", index=False
    )
    print(f"Residual mean={err['residual'].mean():.3f}  std={err['residual'].std():.3f}")

    # --- Feature importance (if available) -----------------------------------
    try:
        model_obj = best.named_steps["model"]
        if hasattr(model_obj, "feature_importances_"):
            names = best.named_steps["pre"].get_feature_names_out()
            fi = pd.DataFrame({"feature": names, "importance": model_obj.feature_importances_})
            fi.sort_values("importance", ascending=False).to_csv(
                cfg.REPORTS_DIR / "feature_importance.csv", index=False
            )
    except Exception as e:
        print(f"(feature importance skipped: {e})")

    # --- ML output -> other blocks: predicted_rating for every catalog book --
    full = pd.read_parquet(cfg.TRAINING_PARQUET)
    for c in ["ratings_count", "text_reviews_count"]:
        if c in full.columns:
            full[c] = np.log1p(full[c])
    preds = pd.DataFrame({"book_id": full["book_id"], "predicted_rating": best.predict(full[features])})

    catalog = pd.read_parquet(cfg.CATALOG_PARQUET)
    catalog = catalog.drop(columns=["predicted_rating"], errors="ignore").merge(preds, on="book_id", how="left")
    catalog.to_parquet(cfg.CATALOG_PARQUET, index=False)
    print("Wrote predicted_rating into catalog.parquet")


if __name__ == "__main__":
    main()