Spaces:
Running
Running
File size: 4,012 Bytes
d29b763 | 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 | #!/usr/bin/env python3
"""Validation for the advanced pharmacology-aware DDI model."""
from __future__ import annotations
import argparse
import json
import logging
import sys
from pathlib import Path
import numpy as np
import pandas as pd
from preprocessing.artifact_manager import manager
import torch
from sklearn.model_selection import train_test_split
ROOT = Path(__file__).resolve().parents[2]
if str(ROOT / 'src') not in sys.path:
sys.path.insert(0, str(ROOT / 'src'))
from training.advanced_feature_engineering import AdvancedBiomedicalFeatureEngineer, AdvancedFeatureConfig, load_metadata_map
from training.advanced_ddi_model import AdvancedDDINet, AdvancedModelConfig
from training.healthcare_safe_pipeline import compute_publication_metrics, optimize_severe_threshold, save_publication_outputs, set_deterministic_seed
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)
BASE_DIR = Path(__file__).resolve().parents[2]
MODELS_DIR = BASE_DIR / "models"
REPORTS_DIR = MODELS_DIR / "reports"
CHECKPOINT_PATH = MODELS_DIR / "advanced_ddi_safe.pt"
REPORTS_DIR.mkdir(parents=True, exist_ok=True)
def main() -> None:
parser = argparse.ArgumentParser(description="Validate the advanced DDI model")
parser.add_argument("--seed", type=int, default=2026)
parser.add_argument("--metadata-path", type=str, default="")
parser.add_argument("--sample-limit", type=int, default=0)
args = parser.parse_args()
set_deterministic_seed(args.seed)
if not CHECKPOINT_PATH.exists():
raise FileNotFoundError(f"Missing checkpoint: {CHECKPOINT_PATH}")
metadata = load_metadata_map(args.metadata_path) if args.metadata_path else {}
engineer = AdvancedBiomedicalFeatureEngineer(AdvancedFeatureConfig(), metadata=metadata)
df = manager.load_artifact('ddinter_combined')
if args.sample_limit and args.sample_limit > 0 and args.sample_limit < len(df):
df = df.sample(n=args.sample_limit, random_state=args.seed).reset_index(drop=True)
X = []
y = []
for _, row in df.iterrows():
feats = engineer.pair_features(row["Drug_A"], row["Drug_B"])
X.append(feats)
y.append({"unknown": 0, "minor": 1, "moderate": 2, "major": 3}.get(str(row["Level"]).strip().lower(), 0))
idx = np.arange(len(X))
_, test_idx = train_test_split(idx, test_size=0.2, random_state=args.seed, stratify=np.array(y))
test_y = np.array([y[i] for i in test_idx], dtype=np.int64)
test_feats = {key: np.vstack([X[i][key] for i in test_idx]).astype(np.float32) for key in X[0].keys()}
payload = torch.load(CHECKPOINT_PATH, map_location="cpu")
config = AdvancedModelConfig(**payload["config"])
model = AdvancedDDINet(config)
model.load_state_dict(payload["model_state_dict"])
model.eval()
with torch.no_grad():
logits, _ = model(
torch.from_numpy(test_feats["fingerprint"]),
torch.from_numpy(test_feats["semantic"]),
torch.from_numpy(test_feats["pharmacology"]),
torch.from_numpy(test_feats["pairwise"]),
torch.from_numpy(test_feats["molecular_pair"]),
)
probs = torch.softmax(logits, dim=1).numpy()
threshold = float(payload.get("threshold", 0.5))
threshold_info = optimize_severe_threshold(probs, test_y, precision_floor=0.25)
threshold = threshold_info["threshold"]
metrics = compute_publication_metrics(test_y, probs, threshold)
save_publication_outputs(metrics, REPORTS_DIR, prefix="advanced_ddi_validation")
summary = {
"checkpoint": str(CHECKPOINT_PATH),
"threshold": threshold,
"threshold_info": threshold_info,
"metrics": metrics,
}
(REPORTS_DIR / "advanced_ddi_validation_summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8")
logger.info("Validation complete. Severe recall: %.4f", metrics["per_class"]["major"]["recall"])
if __name__ == "__main__":
main()
|