| """Train the BioPatch clinical risk model. |
| |
| Generates 5 000 literature-calibrated synthetic snapshots |
| (:mod:`ai.synthetic`), trains a :class:`RandomForestClassifier`, evaluates |
| on a held-out 20% split, and persists the model bundle to |
| ``backend/ai/artifacts/rf_v1.joblib`` together with the training-set |
| summary statistics used at inference time for the XAI breakdown. |
| |
| Run from the ``backend/`` directory:: |
| |
| python -m scripts.seed_clinical_data |
| """ |
|
|
| from __future__ import annotations |
|
|
| import csv |
| import logging |
| from pathlib import Path |
|
|
| import joblib |
| import numpy as np |
| from sklearn.ensemble import RandomForestClassifier |
| from sklearn.metrics import classification_report |
| from sklearn.model_selection import train_test_split |
|
|
| from ai.features import FEATURE_ORDER |
| from ai.synthetic import SEVERITY_CLASSES, generate_dataset |
|
|
| logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") |
| logger = logging.getLogger("seed_clinical_data") |
|
|
|
|
| ROOT = Path(__file__).resolve().parent.parent |
| ARTIFACTS_DIR = ROOT / "ai" / "artifacts" |
| TRAINING_CSV = ARTIFACTS_DIR / "training_data.csv" |
| MODEL_PATH = ARTIFACTS_DIR / "rf_v1.joblib" |
|
|
|
|
| def _dump_csv(samples: list[dict], path: Path) -> None: |
| cols = FEATURE_ORDER + ["severity_class", "severity_label"] |
| with path.open("w", newline="") as f: |
| w = csv.DictWriter(f, fieldnames=cols) |
| w.writeheader() |
| for s in samples: |
| w.writerow({k: s[k] for k in cols}) |
|
|
|
|
| def train(n_samples: int = 5000, seed: int = 42) -> dict: |
| ARTIFACTS_DIR.mkdir(parents=True, exist_ok=True) |
|
|
| logger.info("Generating %d calibrated synthetic samples (seed=%d)", n_samples, seed) |
| samples = generate_dataset(n_samples, seed=seed) |
| _dump_csv(samples, TRAINING_CSV) |
| logger.info("Training data → %s", TRAINING_CSV) |
|
|
| X = np.array([[s[f] for f in FEATURE_ORDER] for s in samples], dtype=np.float32) |
| y = np.array([s["severity_class"] for s in samples], dtype=np.int32) |
|
|
| X_tr, X_te, y_tr, y_te = train_test_split( |
| X, y, test_size=0.20, random_state=seed, stratify=y, |
| ) |
|
|
| logger.info("Training RandomForestClassifier (n_estimators=200, max_depth=14)") |
| clf = RandomForestClassifier( |
| n_estimators=200, |
| max_depth=14, |
| min_samples_leaf=4, |
| class_weight="balanced", |
| random_state=seed, |
| n_jobs=-1, |
| ) |
| clf.fit(X_tr, y_tr) |
| acc = float(clf.score(X_te, y_te)) |
| logger.info("Held-out accuracy: %.3f", acc) |
|
|
| pred = clf.predict(X_te) |
| target_names = [SEVERITY_CLASSES[i] for i in sorted(set(y))] |
| report = classification_report( |
| y_te, pred, target_names=target_names, digits=3, zero_division=0 |
| ) |
| logger.info("Classification report:\n%s", report) |
|
|
| baseline_mean = {f: float(X_tr[:, i].mean()) for i, f in enumerate(FEATURE_ORDER)} |
| baseline_std = {f: float(max(X_tr[:, i].std(), 0.05)) for i, f in enumerate(FEATURE_ORDER)} |
|
|
| importance = dict(zip(FEATURE_ORDER, clf.feature_importances_.tolist())) |
| logger.info("Feature importance: %s", |
| ", ".join(f"{k}={v:.3f}" for k, v in importance.items())) |
|
|
| bundle = { |
| "clf": clf, |
| "version": "rf-v1", |
| "feature_order": FEATURE_ORDER, |
| "baseline_mean": baseline_mean, |
| "baseline_std": baseline_std, |
| "test_accuracy": acc, |
| "n_samples": n_samples, |
| "seed": seed, |
| "classes": SEVERITY_CLASSES, |
| } |
| joblib.dump(bundle, MODEL_PATH) |
| logger.info("Saved model bundle → %s (%.0f KB)", MODEL_PATH, MODEL_PATH.stat().st_size / 1024) |
| return bundle |
|
|
|
|
| if __name__ == "__main__": |
| train() |
|
|
