scripts/41_benchmark_media_recommender.py

"""Dry-lab benchmark for genome-to-medium recommendation.

The production media recommender is trained as one binary classifier per medium.
This script evaluates the more practical ranking question:

    if we hide a strain's known MediaDive medium links, does the model rank at
    least one true medium in its top-k recommendations?

It compares the model against simple popularity baselines under family-grouped
splits by default.
"""
from __future__ import annotations

import argparse
import json
import math
import time
from pathlib import Path
from typing import Any

import numpy as np
import pandas as pd
import xgboost as xgb
from sklearn.metrics import average_precision_score, roc_auc_score
from sklearn.model_selection import GroupKFold, KFold

from microbe_model import config
from microbe_model.train.media_recommender import build_training_table


def load_recommender_features() -> pd.DataFrame:
    """Load the same feature stack used by scripts/10_train_media_recommender.py."""
    feats = pd.read_parquet(config.DATA / "features.parquet")

    hmm_path = config.DATA / "hmm_features.parquet"
    if hmm_path.exists():
        hmm = pd.read_parquet(hmm_path)
        feats = feats.merge(hmm, on="genome_accession", how="left")

    kegg_path = config.DATA / "kegg_modules.parquet"
    if kegg_path.exists():
        kegg = pd.read_parquet(kegg_path)
        feats = feats.merge(kegg, on="genome_accession", how="left")

    iso_meta_path = config.DATA / "isolation_metadata.parquet"
    if iso_meta_path.exists():
        iso_meta = pd.read_parquet(iso_meta_path)
        iso_meta["bacdive_id"] = iso_meta["bacdive_id"].astype(int)
        feats["bacdive_id"] = feats["bacdive_id"].astype(int)
        keep = ["bacdive_id", "iso_lat", "iso_lon", "iso_collection_year"]
        keep += [
            c
            for c in iso_meta.columns
            if c.startswith(("iso_continent_", "iso_country_", "iso_host_kingdom_"))
        ]
        feats = feats.merge(iso_meta[keep], on="bacdive_id", how="left")

    return feats


def group_labels(pheno: pd.DataFrame, index: pd.Index) -> pd.Series:
    """Return stable taxonomic groups with family, then genus, then species fallback."""
    tax = pheno.set_index("bacdive_id").reindex(index)
    groups = tax["family"].copy()
    groups = groups.fillna(tax["genus"]).fillna(tax["species"]).fillna("__unknown__")
    return groups.astype(str)


def topk_metrics(y_true: np.ndarray, scores: np.ndarray, *, ks: tuple[int, ...]) -> dict[str, float]:
    """Compute strain-level top-k recovery metrics for multi-label medium rankings."""
    valid = y_true.sum(axis=1) > 0
    y = y_true[valid].astype(bool)
    s = scores[valid]
    n = int(len(y))
    if n == 0:
        return {"n_eval": 0}

    order = np.argsort(-s, axis=1)
    true_counts = y.sum(axis=1)
    out: dict[str, float] = {"n_eval": float(n), "mean_true_media": float(true_counts.mean())}

    reciprocal_ranks: list[float] = []
    for row_idx in range(n):
        ranked_truth = y[row_idx, order[row_idx]]
        hits = np.flatnonzero(ranked_truth)
        reciprocal_ranks.append(0.0 if len(hits) == 0 else 1.0 / float(hits[0] + 1))
    out["mrr"] = float(np.mean(reciprocal_ranks))

    for k in ks:
        top = order[:, :k]
        top_truth = np.take_along_axis(y, top, axis=1)
        hit = top_truth.any(axis=1)
        out[f"hit_at_{k}"] = float(hit.mean())
        out[f"recall_at_{k}"] = float((top_truth.sum(axis=1) / true_counts).mean())
        out[f"precision_at_{k}"] = float(top_truth.sum(axis=1).mean() / k)
    return out


def per_medium_auc(y_true: np.ndarray, scores: np.ndarray, medium_ids: list[str]) -> pd.DataFrame:
    """Compute PR-AUC and ROC-AUC per medium where both classes are present."""
    rows = []
    for j, mid in enumerate(medium_ids):
        y = y_true[:, j]
        if y.sum() == 0 or y.sum() == len(y):
            continue
        rows.append({
            "medium_id": str(mid),
            "n_pos": int(y.sum()),
            "n_neg": int((y == 0).sum()),
            "pr_auc": float(average_precision_score(y, scores[:, j])),
            "roc_auc": float(roc_auc_score(y, scores[:, j])),
        })
    return pd.DataFrame(rows)


def taxonomy_popularity_scores(
    y_train: pd.DataFrame,
    tax_train: pd.DataFrame,
    tax_test: pd.DataFrame,
    global_scores: np.ndarray,
) -> np.ndarray:
    """Score by same-genus popularity, else same-family popularity, else global popularity."""
    genus_scores = {
        str(genus): y_train.loc[idx].mean(axis=0).to_numpy(dtype=np.float32)
        for genus, idx in tax_train.groupby("genus").groups.items()
        if pd.notna(genus) and len(idx) >= 3
    }
    family_scores = {
        str(family): y_train.loc[idx].mean(axis=0).to_numpy(dtype=np.float32)
        for family, idx in tax_train.groupby("family").groups.items()
        if pd.notna(family) and len(idx) >= 3
    }
    scores = np.tile(global_scores, (len(tax_test), 1)).astype(np.float32)
    for i, (_, row) in enumerate(tax_test.iterrows()):
        genus = str(row.get("genus")) if pd.notna(row.get("genus")) else None
        family = str(row.get("family")) if pd.notna(row.get("family")) else None
        if genus and genus in genus_scores:
            scores[i] = genus_scores[genus]
        elif family and family in family_scores:
            scores[i] = family_scores[family]
    return scores


def train_fold_scores(
    X: pd.DataFrame,
    y_matrix: pd.DataFrame,
    train_idx: np.ndarray,
    test_idx: np.ndarray,
    *,
    n_estimators: int,
    max_depth: int,
    seed: int,
) -> np.ndarray:
    """Train per-medium classifiers for one fold and return test-row score matrix."""
    X_train = X.iloc[train_idx]
    X_test = X.iloc[test_idx]
    y_train = y_matrix.iloc[train_idx]
    global_scores = y_train.mean(axis=0).to_numpy(dtype=np.float32)
    scores = np.tile(global_scores, (len(test_idx), 1)).astype(np.float32)

    for j, medium_id in enumerate(y_matrix.columns):
        y = y_train[medium_id].to_numpy()
        n_pos = int(y.sum())
        n_neg = int((y == 0).sum())
        if n_pos < 5 or n_neg < 5:
            continue
        scale_pos_weight = n_neg / max(1, n_pos)
        model = xgb.XGBClassifier(
            n_estimators=n_estimators,
            max_depth=max_depth,
            learning_rate=0.05,
            tree_method="hist",
            n_jobs=-1,
            scale_pos_weight=scale_pos_weight,
            eval_metric="logloss",
            random_state=seed,
        )
        model.fit(X_train, y)
        scores[:, j] = model.predict_proba(X_test)[:, 1].astype(np.float32)
    return scores


def write_report(
    *,
    path: Path,
    payload: dict[str, Any],
    auc_table: pd.DataFrame,
    medium_names: dict[str, str],
) -> None:
    lines = [
        "# Media Recommender Dry-Lab Benchmark",
        "",
        "This benchmark hides known BacDive/MediaDive strain-medium links and asks",
        "whether the genome-only recommender recovers at least one known medium in",
        "the top-k ranked recommendations.",
        "",
        "## Setup",
        "",
        f"- Split mode: `{payload['split_mode']}`",
        f"- Folds: {payload['n_splits']}",
        f"- Evaluation strains: {int(payload['model']['n_eval'])}",
        f"- Media labels: {payload['n_media']}",
        f"- Feature columns: {payload['n_features']}",
        f"- XGBoost trees per medium per fold: {payload['n_estimators']}",
        "",
        "## Ranking Metrics",
        "",
        "| Method | MRR | Hit@1 | Hit@3 | Hit@5 | Recall@5 | Precision@5 |",
        "|---|---:|---:|---:|---:|---:|---:|",
    ]
    for key, label in [
        ("model", "XGBoost medium recommender"),
        ("global_popularity", "Global popularity baseline"),
        ("taxonomy_popularity", "Taxonomic popularity baseline"),
    ]:
        m = payload[key]
        lines.append(
            f"| {label} | {m['mrr']:.3f} | {m['hit_at_1']:.3f} | {m['hit_at_3']:.3f} | "
            f"{m['hit_at_5']:.3f} | {m['recall_at_5']:.3f} | {m['precision_at_5']:.3f} |"
        )

    valid_auc = auc_table.dropna(subset=["pr_auc", "roc_auc"]) if not auc_table.empty else auc_table
    lines += [
        "",
        "## Per-Medium AUC",
        "",
        f"- Valid media with both classes: {len(valid_auc)}",
        f"- Median ROC-AUC: {valid_auc['roc_auc'].median():.3f}" if len(valid_auc) else "- Median ROC-AUC: n/a",
        f"- Median PR-AUC: {valid_auc['pr_auc'].median():.3f}" if len(valid_auc) else "- Median PR-AUC: n/a",
        "",
        "Top media by PR-AUC:",
        "",
        "| Medium | Positives | PR-AUC | ROC-AUC |",
        "|---|---:|---:|---:|",
    ]
    if len(valid_auc):
        for row in valid_auc.sort_values("pr_auc", ascending=False).head(10).itertuples():
            name = medium_names.get(str(row.medium_id), "")
            lines.append(f"| {row.medium_id} {name} | {row.n_pos} | {row.pr_auc:.3f} | {row.roc_auc:.3f} |")

    lines += [
        "",
        "## Interpretation",
        "",
        "Hit@k is the easiest practical readout: it measures whether at least one",
        "known growth medium appears in the top-k suggestions. PR-AUC is expected",
        "to be much lower than ROC-AUC because medium labels are sparse and heavily",
        "imbalanced; a high ROC-AUC with modest PR-AUC means the model is useful for",
        "ranking candidates, not for guaranteeing growth.",
        "",
    ]
    path.write_text("\n".join(lines))


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--split-mode", choices=("family", "random"), default="family")
    parser.add_argument("--n-splits", type=int, default=5)
    parser.add_argument("--n-estimators", type=int, default=100)
    parser.add_argument("--max-depth", type=int, default=5)
    parser.add_argument("--seed", type=int, default=7)
    parser.add_argument("--max-media", type=int, default=None, help="Debug option: only score first N media.")
    parser.add_argument("--out-json", type=Path, default=config.ARTIFACTS / "media_recommender_drylab_benchmark.json")
    parser.add_argument("--out-md", type=Path, default=config.ARTIFACTS / "media_recommender_drylab_benchmark.md")
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    t0 = time.time()
    pheno = pd.read_parquet(config.DATA / "bacdive_phenotypes.parquet")
    feats = load_recommender_features()
    strain_media = pd.read_parquet(config.DATA / "strain_media.parquet")
    media_meta = pd.read_parquet(config.DATA / "media_metadata.parquet")
    medium_names = dict(zip(media_meta["medium_id"].astype(str), media_meta["name"], strict=True))

    X, y_matrix, medium_ids = build_training_table(feats, strain_media, pheno)
    if args.max_media is not None:
        medium_ids = medium_ids[: args.max_media]
        y_matrix = y_matrix[medium_ids]
    X = X.fillna(0)
    tax = pheno.set_index("bacdive_id").reindex(X.index)[["family", "genus", "species"]]
    groups = group_labels(pheno, X.index)

    if args.split_mode == "family":
        splitter = GroupKFold(n_splits=min(args.n_splits, groups.nunique()))
        splits = list(splitter.split(X, y_matrix, groups))
    else:
        splitter = KFold(n_splits=args.n_splits, shuffle=True, random_state=args.seed)
        splits = list(splitter.split(X))

    model_scores = np.zeros(y_matrix.shape, dtype=np.float32)
    global_scores = np.zeros(y_matrix.shape, dtype=np.float32)
    taxonomy_scores = np.zeros(y_matrix.shape, dtype=np.float32)
    y_all = y_matrix.to_numpy(dtype=np.uint8)

    print(
        f"Benchmark: {len(X):,} strains x {X.shape[1]:,} features x "
        f"{len(medium_ids):,} media, split={args.split_mode}"
    )
    for fold_idx, (train_idx, test_idx) in enumerate(splits):
        fold_t0 = time.time()
        y_train = y_matrix.iloc[train_idx]
        global_prevalence = y_train.mean(axis=0).to_numpy(dtype=np.float32)
        global_scores[test_idx] = np.tile(global_prevalence, (len(test_idx), 1))
        taxonomy_scores[test_idx] = taxonomy_popularity_scores(
            y_train,
            tax.iloc[train_idx],
            tax.iloc[test_idx],
            global_prevalence,
        )
        model_scores[test_idx] = train_fold_scores(
            X,
            y_matrix,
            train_idx,
            test_idx,
            n_estimators=args.n_estimators,
            max_depth=args.max_depth,
            seed=args.seed + fold_idx,
        )
        print(
            f"  fold {fold_idx}: {len(train_idx):,} train / {len(test_idx):,} test "
            f"in {time.time() - fold_t0:.1f}s",
            flush=True,
        )

    ks = (1, 3, 5, 10)
    payload: dict[str, Any] = {
        "split_mode": args.split_mode,
        "n_splits": len(splits),
        "n_features": int(X.shape[1]),
        "n_media": int(len(medium_ids)),
        "n_estimators": args.n_estimators,
        "elapsed_s": time.time() - t0,
        "model": topk_metrics(y_all, model_scores, ks=ks),
        "global_popularity": topk_metrics(y_all, global_scores, ks=ks),
        "taxonomy_popularity": topk_metrics(y_all, taxonomy_scores, ks=ks),
    }
    auc = per_medium_auc(y_all, model_scores, [str(m) for m in medium_ids])
    if not auc.empty:
        payload["model"]["median_roc_auc"] = float(auc["roc_auc"].median())
        payload["model"]["median_pr_auc"] = float(auc["pr_auc"].median())
        payload["per_medium_auc"] = auc.to_dict(orient="records")
    else:
        payload["per_medium_auc"] = []

    args.out_json.parent.mkdir(parents=True, exist_ok=True)
    args.out_json.write_text(json.dumps(payload, indent=2))
    write_report(path=args.out_md, payload=payload, auc_table=auc, medium_names=medium_names)

    print(f"\nWrote {args.out_json}")
    print(f"Wrote {args.out_md}")
    print(json.dumps({
        "model_hit_at_1": payload["model"]["hit_at_1"],
        "model_hit_at_3": payload["model"]["hit_at_3"],
        "model_hit_at_5": payload["model"]["hit_at_5"],
        "global_hit_at_5": payload["global_popularity"]["hit_at_5"],
        "taxonomy_hit_at_5": payload["taxonomy_popularity"]["hit_at_5"],
        "median_roc_auc": payload["model"].get("median_roc_auc"),
        "median_pr_auc": payload["model"].get("median_pr_auc"),
    }, indent=2))


if __name__ == "__main__":
    main()