feat(data): add v0.2 generator with explicit negatives and balance; feat(train): train_from_jsonl and pipeline script; backend: prefer v02 model by default if present

Author: Nick Vaccarello

medical_diagnosis_model/backend/app.py

@@ -34,7 +34,10 @@
     allow_headers=["*"],
 )
 model = ClinicalReasoningNetwork(hidden_neurons=25, learning_rate=0.3, epochs=1000)
-MODEL_PATH = os.path.join(MODEL_ROOT, "models", "enhanced_medical_model.json")
+# Prefer v0.2 model if present; allow env override
+DEFAULT_MODEL = os.path.join(MODEL_ROOT, "models", "enhanced_medical_model.json")
+V02_MODEL = os.path.join(MODEL_ROOT, "models", "enhanced_medical_model_v02.json")
+MODEL_PATH = os.environ.get("MDM_MODEL_PATH") or (V02_MODEL if os.path.exists(V02_MODEL) else DEFAULT_MODEL)
 exporter = PDFExporter(export_dir=os.path.join(MODEL_ROOT, "exports"))
 _RATE_LIMIT_STORE: dict[str, dict[str, float | int]] = {}
 _ADAPTIVE_SESSIONS: Dict[str, Dict] = {}

medical_diagnosis_model/data/generate_v02.py

@@ -0,0 +1,104 @@
+#!/usr/bin/env python3
+"""
+Generate v0.2 balanced JSONL training data with explicit negative evidence.
+
+Format per line:
+  {
+    "symptoms": {"Cough": 6, "Runny Nose": 5, ...},
+    "label_name": "Viral Upper Respiratory Infection"
+  }
+"""
+from __future__ import annotations
+
+import json
+import os
+import random
+from pathlib import Path
+from typing import Dict, Tuple, List
+
+
+def _load_schema():
+    # Resolve imports locally
+    import sys
+    here = Path(__file__).resolve().parent
+    model_root = here.parent
+    repo_root = model_root.parent
+    for p in (str(repo_root), str(model_root)):
+        if p not in sys.path:
+            sys.path.append(p)
+    from versions.v2.medical_disease_schema_v2 import DISEASES_V2
+    from medical_symptom_schema import SYMPTOMS
+    return DISEASES_V2, SYMPTOMS
+
+
+def _sample_case(disease_id: int, diseases: dict, symptoms: dict, explicit_neg: bool) -> Tuple[Dict[str, float], str]:
+    dis = diseases[disease_id]
+    name = dis["name"]
+    pats = dis.get("symptom_patterns", {})
+    out: Dict[str, float] = {}
+    # Positive sampling from patterns
+    for sid, pat in pats.items():
+        if sid not in symptoms:
+            continue
+        freq = pat.get("frequency", 0.0)
+        sev_lo, sev_hi = pat.get("severity_range", (0.2, 0.6))
+        if random.random() < freq:
+            sev = random.uniform(sev_lo, sev_hi)
+            out[symptoms[sid]["name"]] = round(min(max(sev * 10.0, 0.0), 10.0), 1)
+
+    # Mild/early tweak: 30% chance reduce severities
+    if random.random() < 0.3:
+        for k in list(out.keys()):
+            out[k] = round(out[k] * random.uniform(0.5, 0.8), 1)
+
+    # Explicit negatives across syndromes
+    if explicit_neg:
+        # For respiratory: ensure GU keys absent; for GU: reduce respiratory signals
+        if name in ("Viral Upper Respiratory Infection", "Influenza-like Illness", "COVID-19-like Illness", "Viral Syndrome", "Pneumonia Syndrome"):
+            for sid in (26, 27):  # Frequency, Dysuria
+                out.setdefault(symptoms[sid]["name"], 0.0)
+        if name == "Urinary Tract Infection":
+            for sid in (3, 7, 8):  # Cough, Rhinorrhea, Congestion
+                out.setdefault(symptoms[sid]["name"], 0.0)
+
+    return out, name
+
+
+def generate_balanced(per_disease: int = 200, seed: int = 42) -> List[Dict]:
+    random.seed(seed)
+    DISEASES_V2, SYMPTOMS = _load_schema()
+    # Focus set: common respiratory + GU UTI
+    target_names = {
+        "Viral Upper Respiratory Infection",
+        "Influenza-like Illness",
+        "COVID-19-like Illness",
+        "Viral Syndrome",
+        "Urinary Tract Infection",
+    }
+    target_ids = [did for did, d in DISEASES_V2.items() if d["name"] in target_names]
+    data: List[Dict] = []
+    for did in target_ids:
+        for _ in range(per_disease):
+            s, label = _sample_case(did, DISEASES_V2, SYMPTOMS, explicit_neg=True)
+            data.append({"symptoms": s, "label_name": label})
+    random.shuffle(data)
+    return data
+
+
+def main() -> int:
+    root = Path(__file__).resolve().parents[1]
+    out_dir = root / "data" / "v02"
+    out_dir.mkdir(parents=True, exist_ok=True)
+    path = out_dir / "cases_v02.jsonl"
+    data = generate_balanced(per_disease=150)
+    with path.open("w", encoding="utf-8") as f:
+        for row in data:
+            f.write(json.dumps(row) + "\n")
+    print(f"Wrote {len(data)} cases to {path}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
+
+

medical_diagnosis_model/tools/train_pipeline.py

@@ -0,0 +1,66 @@
+#!/usr/bin/env python3
+"""
+One-shot training pipeline:
+ - Generate v0.2 balanced dataset (explicit negatives)
+ - Train v2 from JSONL
+ - Calibrate and save model to models/enhanced_medical_model_v02.json
+ - Optionally run a quick confusion summary on held-out set (counts only)
+"""
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+
+def _setup_paths() -> None:
+    import os, sys
+    here = Path(__file__).resolve().parent
+    model_root = here.parent
+    repo_root = model_root.parent
+    for p in (str(repo_root), str(model_root)):
+        if p not in sys.path:
+            sys.path.append(p)
+
+
+def generate_dataset(per_disease: int) -> Path:
+    from data.generate_v02 import generate_balanced
+    root = Path(__file__).resolve().parents[1]
+    out = root / "data" / "v02"
+    out.mkdir(parents=True, exist_ok=True)
+    path = out / "cases_v02.jsonl"
+    data = generate_balanced(per_disease=per_disease)
+    with path.open("w", encoding="utf-8") as f:
+        for row in data:
+            f.write(json.dumps(row) + "\n")
+    return path
+
+
+def train_model(jsonl_path: Path, epochs: int) -> Path:
+    from versions.v2.medical_neural_network_v2 import ClinicalReasoningNetwork
+    m = ClinicalReasoningNetwork(hidden_neurons=25, learning_rate=0.3, epochs=epochs)
+    m.train_from_jsonl(str(jsonl_path), verbose=False)
+    out = Path(__file__).resolve().parents[1] / "models" / "enhanced_medical_model_v02.json"
+    m.save_model(str(out))
+    return out
+
+
+def main() -> int:
+    _setup_paths()
+    ap = argparse.ArgumentParser(description="Train v0.2 model from generated data")
+    ap.add_argument("--per-disease", type=int, default=200)
+    ap.add_argument("--epochs", type=int, default=5000)
+    args = ap.parse_args()
+
+    jsonl = generate_dataset(args.per_disease)
+    print(f"Generated dataset: {jsonl}")
+    model_path = train_model(jsonl, args.epochs)
+    print(f"Saved model: {model_path}")
+    print("Set MDM_MODEL_PATH to use this model in the API if not picked by default.")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
+
+

medical_diagnosis_model/versions/v2/medical_neural_network_v2.py

@@ -162,9 +162,15 @@ def diagnose_with_reasoning(self, symptoms_dict, has_test_results=None):
         for symptom_name, severity in symptoms_dict.items():
             sid, symptom = get_symptom_by_name(symptom_name)
             if sid is not None and sid < self.num_symptoms:
-                symptom_vector[sid] = 1
-                severity_vector[sid] = severity / 10.0
-                symptom_ids.append(sid)
+                try:
+                    sev_norm = float(severity) / 10.0
+                except Exception:
+                    sev_norm = 0.0
+                # Treat zero (or negative) severity as absent
+                if sev_norm > 0.0:
+                    symptom_vector[sid] = 1
+                    severity_vector[sid] = sev_norm
+                    symptom_ids.append(sid)
 
         # Determine syndrome
         syndrome = get_syndrome_from_symptoms(symptom_ids)
@@ -462,6 +468,61 @@ def load_model(self, filename="models/enhanced_medical_model.json"):
             self.network.append(rebuilt)
         print(f"Model loaded from {filename}")
 
+    # ===== Training from JSONL (v0.2) =====
+    def train_from_jsonl(self, jsonl_path: str, seed: int = 42, verbose: bool = True):
+        import random
+        random.seed(seed)
+        # Load data
+        rows = []
+        import json
+        with open(jsonl_path, "r", encoding="utf-8") as f:
+            for line in f:
+                line = line.strip()
+                if not line:
+                    continue
+                obj = json.loads(line)
+                rows.append(obj)
+        # Build vectors
+        dataset = []
+        for obj in rows:
+            sym = obj.get("symptoms", {})
+            label_name = obj.get("label_name")
+            # Map label to id
+            label_id = None
+            for did, d in DISEASES_V2.items():
+                if d.get("name") == label_name:
+                    label_id = did
+                    break
+            if label_id is None:
+                continue
+            symptom_vector = [0] * self.num_symptoms
+            severity_vector = [0.0] * self.num_symptoms
+            for name, sev in sym.items():
+                sid, _ = get_symptom_by_name(name)
+                if sid is None or sid >= self.num_symptoms:
+                    continue
+                try:
+                    sevn = float(sev) / 10.0
+                except Exception:
+                    sevn = 0.0
+                if sevn > 0.0:
+                    symptom_vector[sid] = 1
+                    severity_vector[sid] = sevn
+            features = symptom_vector + severity_vector + [label_id]
+            dataset.append(features)
+        # Shuffle and split
+        random.shuffle(dataset)
+        split = int(0.8 * len(dataset))
+        train_set = dataset[:split]
+        val_set = dataset[split:]
+        # Init and train
+        self.network = initialize_network(self.num_features, self.hidden_neurons, self.num_diseases)
+        history = self._train_softmax_cross_entropy(self.network, train_set, val_set, verbose=verbose)
+        self.temperature = self._calibrate_temperature(val_set)
+        if verbose:
+            print(f"Calibration: selected T={self.temperature:.2f}")
+        return history
+    
     def _apply_clinical_rules(self, nn_outputs, symptom_ids, severity_vector, has_test_results):
         """Apply clinical decision rules to adjust probabilities"""
         adjusted = nn_outputs.copy()
@@ -572,6 +633,9 @@ def _idx(name: str):
                     logits[uri_idx] += 1.5
                 if fever < 0.6 and myalgia < 0.6:
                     logits[uri_idx] += 0.5
+                # Negative evidence: GU keys absent with strong URI pattern present
+                if uti_idx is not None and (rhinorrhea > 0.3 or congestion > 0.3) and (26 not in symptom_ids) and (27 not in symptom_ids):
+                    logits[uti_idx] -= 6.0
 
             # ILI: high fever + myalgia (+/- severe fatigue)
             if ili_idx is not None: