File size: 3,626 Bytes
10795db | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | """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()
|