deepreads / scripts /train_classical.py
yash-b18
feat: all three models — naive popularity, LightGBM reranker, Two-Tower neural
fa98c11
Raw History Blame Contribute Delete
4.99 kB
"""
Script: train_classical.py
Trains the LightGBM reranker on the full feature set and saves the model.
Outputs:
models/lgbm_model.pkl
data/outputs/figures/feature_importance.png
Usage:
python scripts/train_classical.py
python scripts/train_classical.py --debug
python scripts/train_classical.py --feature-set id_metadata # ablation tier
"""
from __future__ import annotations
import argparse
import pickle
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import matplotlib.pyplot as plt
import pandas as pd
from src.config import FIGURES_DIR, MODELS_DIR, PROCESSED_DIR, ensure_dirs, get_config
from src.features.builder import FEATURE_SETS, build_feature_matrix
from src.models.classical import LightGBMReranker
from src.utils.logging import get_logger
logger = get_logger(__name__)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Train LightGBM reranker.")
parser.add_argument("--debug", action="store_true")
parser.add_argument(
"--feature-set",
type=str,
default="id_metadata_text_history",
choices=FEATURE_SETS,
help="Feature tier to use for training.",
)
parser.add_argument(
"--output-name",
type=str,
default="lgbm_model",
help="Stem of the output .pkl filename (in models/).",
)
return parser.parse_args()
def plot_feature_importance(model: LightGBMReranker, output_path: Path) -> None:
"""Save a horizontal bar chart of feature importances."""
imp = model.feature_importance()
top = imp.head(20)
fig, ax = plt.subplots(figsize=(9, 6))
ax.barh(top["feature"][::-1], top["importance"][::-1], color="#3b82f6")
ax.set_xlabel("Importance (Gain)")
ax.set_title("LightGBM Feature Importance (Top 20)")
plt.tight_layout()
output_path.parent.mkdir(parents=True, exist_ok=True)
fig.savefig(output_path, dpi=150)
plt.close(fig)
logger.info("Feature importance plot saved → %s", output_path)
def main() -> None:
args = parse_args()
cfg = get_config(debug=args.debug)
ensure_dirs(cfg)
logger.info("Loading pre-built feature matrices…")
# Try to load pre-built features; fall back to rebuilding
if cfg.data.train_features_path.exists():
train_feat = pd.read_parquet(cfg.data.train_features_path)
val_feat = pd.read_parquet(cfg.data.val_features_path)
else:
logger.warning("Pre-built features not found. Loading raw splits and rebuilding…")
train_df = pd.read_parquet(cfg.data.train_path)
val_df = pd.read_parquet(cfg.data.val_path)
user_features = pd.read_parquet(cfg.data.user_features_path)
item_features = pd.read_parquet(cfg.data.item_features_path)
artifacts_path = PROCESSED_DIR / "feature_artifacts.pkl"
with open(artifacts_path, "rb") as f:
artifacts = pickle.load(f)
user_profiles_path = PROCESSED_DIR / "user_profiles.parquet"
if user_profiles_path.exists():
user_profiles = pd.read_parquet(user_profiles_path)
else:
from src.features.text_features import build_user_text_profiles
user_profiles = build_user_text_profiles(train_df, item_features)
train_feat, _ = build_feature_matrix(
train_df, user_features, item_features, user_profiles,
artifacts["vectorizer"], feature_set=args.feature_set,
)
val_feat, _ = build_feature_matrix(
val_df, user_features, item_features, user_profiles,
artifacts["vectorizer"], feature_set=args.feature_set,
)
# Determine feature columns (everything except 'label' and index cols)
exclude = {"label", "user_idx", "item_idx", "user_id", "item_id", "rating", "timestamp"}
feature_cols = [c for c in train_feat.columns if c not in exclude]
logger.info("Training LightGBM on %d features, %d rows…", len(feature_cols), len(train_feat))
model = LightGBMReranker(
n_estimators=cfg.classical.n_estimators,
learning_rate=cfg.classical.learning_rate,
num_leaves=cfg.classical.num_leaves,
max_depth=cfg.classical.max_depth,
min_child_samples=cfg.classical.min_child_samples,
subsample=cfg.classical.subsample,
colsample_bytree=cfg.classical.colsample_bytree,
reg_alpha=cfg.classical.reg_alpha,
reg_lambda=cfg.classical.reg_lambda,
n_jobs=cfg.classical.n_jobs,
random_state=cfg.data.random_seed,
early_stopping_rounds=cfg.classical.early_stopping_rounds,
)
model.fit(train_feat, val_feat, feature_cols)
# Save model
model_path = MODELS_DIR / f"{args.output_name}.pkl"
model.save(model_path)
# Feature importance plot
plot_feature_importance(model, FIGURES_DIR / "feature_importance.png")
logger.info("LightGBM training complete.")
if __name__ == "__main__":
main()