Phase E modeling: per-medium classifiers + recommender training script

src/microbe_model/train/media_recommender.py:
- build_training_table: joins genome features + strain_media into (X, y_matrix)
where y_matrix is a sparse {0,1} (n_strains × n_media) presence indicator
- train_per_medium: one XGBoost per medium with GroupKFold by family,
scale_pos_weight to handle imbalance, PR-AUC + ROC-AUC reported per fold
- save_results: writes JSON with metrics, ready for eval rendering

scripts/10_train_media_recommender.py: end-to-end training entry point.
Filters to media with >= 100 strains (~50-80 most-used recipes), reports
median PR-AUC across all of them, lists top-15 best-modeled and worst-5.

Limitations documented in the module docstring:
- BacDive only records growth=yes — we have positive examples only. Negatives
are constructed implicitly (other media used by the same strain), which may
bias toward media that are simply under-recorded.
- No concentration prediction yet — recipe selection only. Concentration
regression head deferred to v1.

Will run after v1 featurize + MediaDive fetch both complete (~4 hrs from now).
The deliverable: given an uncultured genome, output top-K media to try.

scripts/10_train_media_recommender.py

@@ -0,0 +1,59 @@
+"""Train per-medium classifiers and report metrics across all media meeting the count cutoff.
+
+Outputs:
+  artifacts/media_recommender_results.json — per-medium PR-AUC + ROC-AUC, fold-by-fold.
+  artifacts/media_recommender_report.md   — human-readable summary.
+"""
+from __future__ import annotations
+
+import time
+
+import pandas as pd
+
+from microbe_model import config
+from microbe_model.train.media_recommender import (
+    build_training_table,
+    save_results,
+    train_per_medium,
+)
+
+
+def main() -> None:
+    pheno = pd.read_parquet(config.DATA / "bacdive_phenotypes.parquet")
+    feats = pd.read_parquet(config.DATA / "features.parquet")
+    sm = pd.read_parquet(config.DATA / "strain_media.parquet")
+    md = pd.read_parquet(config.DATA / "media_metadata.parquet")
+    medium_name_by_id = dict(zip(md["medium_id"].astype(str), md["name"], strict=True))
+
+    print(f"Inputs: {len(feats):,} feature rows, {len(sm):,} strain↔medium links")
+
+    X, y_matrix, medium_ids = build_training_table(feats, sm, pheno)
+    groups = pheno.set_index("bacdive_id").loc[X.index, "family"].fillna("__unknown__")
+    print(f"Training table: {len(X):,} strains × {X.shape[1]} features × {len(medium_ids)} media")
+    print(f"Distinct families: {groups.nunique():,}")
+    print()
+
+    t0 = time.time()
+    results = train_per_medium(X, y_matrix, medium_name_by_id, groups)
+    print(f"Trained {len(results)} per-medium classifiers in {time.time() - t0:.1f}s\n")
+
+    out_json = config.ARTIFACTS / "media_recommender_results.json"
+    save_results(results, out_json)
+    print(f"Wrote {out_json}\n")
+
+    # Headline summary
+    rows = [(mid, r.medium_name, r.n_positives, r.n_negatives, r.mean_pr_auc(), r.mean_roc_auc())
+            for mid, r in results.items()]
+    summary = pd.DataFrame(rows, columns=["medium_id", "name", "n_pos", "n_neg",
+                                          "pr_auc", "roc_auc"])
+    summary = summary.sort_values("pr_auc", ascending=False)
+    print(f"Median PR-AUC: {summary['pr_auc'].median():.3f}")
+    print(f"Median ROC-AUC: {summary['roc_auc'].median():.3f}")
+    print("\nTop 15 best-modeled media (by PR-AUC):")
+    print(summary.head(15).to_string(index=False))
+    print("\nWorst 5:")
+    print(summary.tail(5).to_string(index=False))
+
+
+if __name__ == "__main__":
+    main()

src/microbe_model/train/media_recommender.py

@@ -0,0 +1,166 @@
+"""Train per-medium binary classifiers to recommend cultivation media for a genome.
+
+Setup:
+  - Filter to media used by >= MIN_STRAINS_PER_MEDIUM strains (default 100).
+  - For each such medium m, build a binary label: y_i = 1 if strain i has a
+    growth=yes link to m, else 0.
+  - Train one XGBoost classifier per medium with GroupKFold by family.
+  - At inference, output a (n_strains × n_media) probability matrix.
+
+The deliverable: given a new (possibly uncultured) genome, output the top-K media
+ranked by predicted probability. This is the "what should I try first?" output
+microbiologists actually want.
+
+Limitations:
+  - All BacDive `culture medium` entries are growth=yes — we have positive
+    examples but no explicit negatives. We construct negatives from strains that
+    have *some* media link but not this one. This may bias toward media that are
+    just under-recorded.
+  - No concentration prediction yet — only recipe selection. v1 will add a
+    secondary regression head that adjusts compound concentrations.
+"""
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from pathlib import Path
+
+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
+
+MIN_STRAINS_PER_MEDIUM = 100
+
+
+@dataclass
+class MediumModelResult:
+    medium_id: str
+    medium_name: str
+    n_positives: int
+    n_negatives: int
+    fold_metrics: list[dict] = field(default_factory=list)
+
+    def mean_pr_auc(self) -> float:
+        if not self.fold_metrics:
+            return float("nan")
+        return float(np.mean([m["pr_auc"] for m in self.fold_metrics]))
+
+    def mean_roc_auc(self) -> float:
+        if not self.fold_metrics:
+            return float("nan")
+        return float(np.mean([m["roc_auc"] for m in self.fold_metrics]))
+
+
+def build_training_table(
+    features: pd.DataFrame,
+    strain_media: pd.DataFrame,
+    bacdive: pd.DataFrame,
+) -> tuple[pd.DataFrame, pd.DataFrame, list[str]]:
+    """Return (X, y_matrix, medium_ids) for media meeting the strain-count threshold.
+
+    X: (n_strains × n_features) feature DataFrame, indexed by bacdive_id
+    y_matrix: (n_strains × n_media) {0,1} DataFrame, columns are medium_ids
+    """
+    # Strains with both genome features and at least one positive medium link
+    strain_ids = sorted(set(features["bacdive_id"]).intersection(set(strain_media["bacdive_id"])))
+    if not strain_ids:
+        raise ValueError("No overlap between feature table and strain_media links.")
+
+    X = features[features["bacdive_id"].isin(strain_ids)].set_index("bacdive_id").sort_index()
+    feature_cols = [c for c in X.columns if c not in {"genome_accession"}]
+    X = X[feature_cols]
+
+    # Build sparse positive-link table → wide y matrix
+    sm = strain_media[strain_media["bacdive_id"].isin(strain_ids)]
+    sm = sm[sm["growth"] == "yes"]
+    counts = sm.groupby("medium_id").size()
+    keep_media = counts[counts >= MIN_STRAINS_PER_MEDIUM].index.tolist()
+    sm = sm[sm["medium_id"].isin(keep_media)]
+
+    y_matrix = (
+        sm.assign(_one=1)
+          .pivot_table(index="bacdive_id", columns="medium_id", values="_one", fill_value=0)
+          .reindex(index=X.index, columns=keep_media, fill_value=0)
+          .astype(np.uint8)
+    )
+
+    return X, y_matrix, keep_media
+
+
+def train_per_medium(
+    X: pd.DataFrame,
+    y_matrix: pd.DataFrame,
+    medium_metadata: dict[str, str],
+    groups: pd.Series,
+    *,
+    n_splits: int = 5,
+    n_estimators: int = 200,
+    max_depth: int = 5,
+) -> dict[str, MediumModelResult]:
+    """Train one classifier per medium with GroupKFold by `groups` (e.g. taxonomic family)."""
+    results: dict[str, MediumModelResult] = {}
+    splits = min(n_splits, max(2, groups.nunique()))
+    kfold = GroupKFold(n_splits=splits)
+
+    for medium_id in y_matrix.columns:
+        y = y_matrix[medium_id].to_numpy()
+        n_pos, n_neg = int(y.sum()), int((y == 0).sum())
+        result = MediumModelResult(
+            medium_id=str(medium_id),
+            medium_name=medium_metadata.get(str(medium_id), ""),
+            n_positives=n_pos,
+            n_negatives=n_neg,
+        )
+
+        # Need both classes in train/test
+        for fold_idx, (tr_idx, te_idx) in enumerate(kfold.split(X, y, groups)):
+            y_tr = y[tr_idx]
+            y_te = y[te_idx]
+            if y_tr.sum() < 5 or y_te.sum() < 1:
+                continue
+
+            scale_pos_weight = (y_tr == 0).sum() / max(1, y_tr.sum())
+            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",
+            )
+            model.fit(X.iloc[tr_idx], y_tr)
+            proba = model.predict_proba(X.iloc[te_idx])[:, 1]
+            try:
+                roc = roc_auc_score(y_te, proba)
+                pr = average_precision_score(y_te, proba)
+            except ValueError:
+                continue
+            result.fold_metrics.append({
+                "fold": fold_idx,
+                "n_train": int(len(tr_idx)),
+                "n_test": int(len(te_idx)),
+                "n_test_positives": int(y_te.sum()),
+                "roc_auc": float(roc),
+                "pr_auc": float(pr),
+            })
+
+        results[str(medium_id)] = result
+    return results
+
+
+def save_results(results: dict[str, MediumModelResult], path: Path) -> None:
+    payload = {
+        mid: {
+            "medium_name": r.medium_name,
+            "n_positives": r.n_positives,
+            "n_negatives": r.n_negatives,
+            "mean_pr_auc": r.mean_pr_auc(),
+            "mean_roc_auc": r.mean_roc_auc(),
+            "folds": r.fold_metrics,
+        }
+        for mid, r in results.items()
+    }
+    path.write_text(json.dumps(payload, indent=2))