src/scoring.py

import numpy as np
import pandas as pd
from .chemistry import descriptors
from .literature import alignment, extract_drug_direction


def _benefit_scale(series: pd.Series, lower_is_better: bool = False) -> pd.Series:
    x = pd.to_numeric(series, errors="coerce")
    valid = x.dropna()
    if valid.empty:
        return pd.Series(0.0, index=series.index)
    lo, hi = valid.quantile(0.05), valid.quantile(0.95)
    if np.isclose(lo, hi):
        scaled = pd.Series(0.5, index=series.index)
    else:
        scaled = ((x - lo) / (hi - lo)).clip(0, 1).fillna(0)
    return 1 - scaled if lower_is_better else scaled


def run_pipeline(frame: pd.DataFrame, ai_weight: float, docking_weight: float, md_weight: float, qc_weight: float, uncertainty_penalty: float, apply_filter: bool, use_placeholders: bool) -> pd.DataFrame:
    df = frame.copy()
    required = {"drug", "target"}
    missing = required - set(df.columns)
    if missing:
        raise ValueError(f"Missing required columns: {', '.join(sorted(missing))}")
    defaults = {"smiles": "", "disease_direction": "", "evidence": "", "docking_score": np.nan, "md_stability": np.nan, "qc_energy": np.nan, "source": "user input"}
    for col, default in defaults.items():
        if col not in df:
            df[col] = default
    extracted = df["evidence"].map(extract_drug_direction)
    df[["drug_direction", "literature_confidence", "extraction_note"]] = pd.DataFrame(extracted.tolist(), index=df.index)
    df["direction_alignment"] = [alignment(d, r) for d, r in zip(df["disease_direction"], df["drug_direction"])]
    df["ai_score"] = ((df["direction_alignment"] + 1) / 2 * df["literature_confidence"]).round(4)
    chem = pd.DataFrame([descriptors(s) for s in df["smiles"]], index=df.index)
    df = pd.concat([df, chem], axis=1)

    # Deterministic placeholders are for UI demonstration only, never claimed as simulations.
    keys = df["drug"].astype(str) + "|" + df["target"].astype(str)
    seed = keys.map(lambda x: sum((i + 1) * ord(c) for i, c in enumerate(x)))
    if use_placeholders:
        df["docking_score"] = pd.to_numeric(df["docking_score"], errors="coerce").fillna(-4.0 - (seed % 45) / 10)
        df["md_stability"] = pd.to_numeric(df["md_stability"], errors="coerce").fillna(0.35 + (seed % 55) / 100)
        df["qc_energy"] = pd.to_numeric(df["qc_energy"], errors="coerce").fillna(-20.0 - (seed % 600) / 10)
    df["docking_norm"] = _benefit_scale(df["docking_score"], lower_is_better=True)
    df["md_norm"] = pd.to_numeric(df["md_stability"], errors="coerce").clip(0, 1).fillna(0)
    df["qc_norm"] = _benefit_scale(df["qc_energy"], lower_is_better=True)
    physical_available = df[["docking_score", "md_stability", "qc_energy"]].notna().mean(axis=1)
    total_w = max(ai_weight + docking_weight + md_weight + qc_weight, 1e-9)
    raw = (ai_weight * df["ai_score"] + docking_weight * df["docking_norm"] + md_weight * df["md_norm"] + qc_weight * df["qc_norm"]) / total_w
    df["evidence_completeness"] = ((df["literature_confidence"] > 0).astype(float) + 3 * physical_available) / 4
    df["composite_score"] = (100 * raw * (1 - uncertainty_penalty * (1 - df["evidence_completeness"]))).clip(0, 100).round(2)
    if apply_filter:
        df = df[df["chem_pass"]].copy()
    df = df.sort_values("composite_score", ascending=False).reset_index(drop=True)
    df.insert(0, "rank", range(1, len(df) + 1))
    return df