Cookbook 324: Offline Clinical Hematology HIT 4Ts Score, Anti-PF4 ELISA & Argatroban Engine

"""
Cookbook 324: Offline Hematology HIT 4Ts Score, Anti-PF4 ELISA & Argatroban Engine
OpenPHR Clinical AI Working Group (https://openphr.org)
"""

import math
import numpy as np
import pandas as pd
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass

@dataclass
class HITPatientTelemetry:
    patient_id: str
    age_years: float
    patient_weight_kg: float = 82.0 # kg
    # Platelet Trajectory Telemetry
    baseline_platelet_count_k_ul: float = 280.0 # x10^9/L (k/uL)
    nadir_platelet_count_k_ul: float = 68.0 # x10^9/L
    current_platelet_count_k_ul: float = 68.0 # x10^9/L
    # Heparin Chronology
    days_since_heparin_started: int = 7 # Days (Days 5-10 = classic)
    heparin_exposure_in_last_30_days: bool = False
    heparin_exposure_in_last_31_to_100_days: bool = False
    # Thrombosis & Clinical Sequelae Telemetry
    new_proven_thrombosis: bool = True # DVT/PE/Arterial
    skin_necrosis_at_injection_site: bool = False
    acute_systemic_reaction_post_bolus: bool = False
    recurrent_or_suspected_thrombosis: bool = False
    # Other Causes of Thrombocytopenia
    other_causes_apparent: str = "None" # "None" (2 pts), "Possible" (1 pt - sepsis/bypass), "Definite" (0 pts - severe DIC/ITP)
    # Laboratory Immunoassay Telemetry
    anti_pf4_elisa_od_value: Optional[float] = 1.640 # Optical density (>= 1.000 = Strongly Positive)
    sra_serotonin_release_percent: Optional[float] = 88.0 # % (>= 20% = Confirmed Positive)
    # Hepatic & ICU Hemodynamic Telemetry
    is_critically_ill_or_icu: bool = True
    total_bilirubin_mg_dl: float = 2.2 # mg/dL (> 1.5 = Hepatic impairment)
    baseline_aptt_seconds: float = 30.0 # seconds
    current_aptt_seconds: float = 32.0 # seconds
    # Active Medication Warnings
    active_warfarin_prescription: bool = True # 🚨 Deadly contraindication!

@dataclass
class HITEvaluationReport:
    patient_id: str
    four_ts_total_score: int # 0 - 8
    four_ts_probability_tier: str # "High Probability (6-8 points)", "Intermediate Probability (4-5 points)", "Low Probability (0-3 points)"
    four_ts_breakdown: List[str]
    elisa_sra_interpretation: str
    heparin_cessation_orders: List[str]
    non_heparin_anticoagulation_protocol: List[str]
    safety_sentinels: List[str]
    clinical_ash_chest_directive: str

class HeparinInducedThrombocytopeniaEngine:
    """
    Offline clinical engine for 4Ts probability scoring, Anti-PF4 ELISA / SRA interpretation,
    Argatroban / DTI pharmacokinetic titration, and Warfarin-induced venous limb gangrene prevention.
    """

    def calculate_four_ts_score(self, d: HITPatientTelemetry) -> Tuple[int, str, List[str]]:
        score = 0
        breakdown = []

        # 1. Thrombocytopenia
        platelet_drop_pct = ((d.baseline_platelet_count_k_ul - d.nadir_platelet_count_k_ul) / max(d.baseline_platelet_count_k_ul, 1.0)) * 100.0

        if platelet_drop_pct > 50.0 and d.nadir_platelet_count_k_ul >= 20.0:
            t1 = 2
            breakdown.append(f"1. Thrombocytopenia (2 pts): Platelet drop of {platelet_drop_pct:.1f}% (>50%) with nadir {d.nadir_platelet_count_k_ul:.0f} k/uL (>= 20 k/uL).")
        elif (30.0 <= platelet_drop_pct <= 50.0) or (10.0 <= d.nadir_platelet_count_k_ul <= 19.0):
            t1 = 1
            breakdown.append(f"1. Thrombocytopenia (1 pt): Platelet drop of {platelet_drop_pct:.1f}% (30-50%) or nadir {d.nadir_platelet_count_k_ul:.0f} k/uL.")
        else:
            t1 = 0
            breakdown.append(f"1. Thrombocytopenia (0 pts): Platelet drop <30% or nadir <10 k/uL.")
        score += t1

        # 2. Timing of Platelet Count Fall
        if (5 <= d.days_since_heparin_started <= 10) or (d.days_since_heparin_started <= 1 and d.heparin_exposure_in_last_30_days):
            t2 = 2
            desc = f"Clear onset on Day {d.days_since_heparin_started} (Days 5-10)" if (5 <= d.days_since_heparin_started <= 10) else "Rapid onset <=1 day with heparin exposure within past 30 days"
            breakdown.append(f"2. Timing (2 pts): {desc}.")
        elif (d.days_since_heparin_started > 10) or (d.days_since_heparin_started <= 1 and d.heparin_exposure_in_last_31_to_100_days):
            t2 = 1
            desc = "Onset after Day 10" if d.days_since_heparin_started > 10 else "Onset <=1 day with heparin exposure in past 31-100 days"
            breakdown.append(f"2. Timing (1 pt): {desc}.")
        else:
            t2 = 0
            breakdown.append(f"2. Timing (0 pts): Platelet fall < Day 4 without recent heparin exposure.")
        score += t2

        # 3. Thrombosis or Other Sequelae
        if d.new_proven_thrombosis or d.skin_necrosis_at_injection_site or d.acute_systemic_reaction_post_bolus:
            t3 = 2
            breakdown.append("3. Thrombosis (2 pts): Proven new thrombosis, skin necrosis, or acute post-bolus systemic reaction.")
        elif d.recurrent_or_suspected_thrombosis:
            t3 = 1
            breakdown.append("3. Thrombosis (1 pt): Recurrent or suspected thrombosis.")
        else:
            t3 = 0
            breakdown.append("3. Thrombosis (0 pts): No thrombosis or cutaneous lesions.")
        score += t3

        # 4. oTher Causes of Thrombocytopenia
        if d.other_causes_apparent == "None":
            t4 = 2
            breakdown.append("4. Other Causes (2 pts): No other apparent cause for thrombocytopenia.")
        elif d.other_causes_apparent == "Possible":
            t4 = 1
            breakdown.append("4. Other Causes (1 pt): Possible other cause present (e.g. sepsis, bypass, medication).")
        else:
            t4 = 0
            breakdown.append("4. Other Causes (0 pts): Definite other explanation identified.")
        score += t4

        if score >= 6:
            tier = "High Probability of HIT (6-8 points, ~64% True HIT Incidence)"
        elif score >= 4:
            tier = "Intermediate Probability of HIT (4-5 points, ~14% True HIT Incidence)"
        else:
            tier = "Low Probability of HIT (0-3 points, >99% Negative Predictive Value)"

        return score, tier, breakdown

    def interpret_immunoassays(self, od: Optional[float], sra: Optional[float]) -> str:
        if od is None:
            return "Anti-PF4 ELISA pending / not yet resulted."

        if od < 0.400:
            return f"Anti-PF4 ELISA Negative (OD {od:.3f} < 0.400) -> Rules out HIT; look for alternative causes."
        elif od >= 2.000:
            sra_str = f" Confirmatory SRA: {sra:.0f}% release (Positive >= 20%)." if sra is not None else " Confirmatory SRA pending."
            return f"Anti-PF4 ELISA STRONGLY POSITIVE (OD {od:.3f} >= 2.000, >90% specificity).{sra_str} Definite immune HIT confirmed."
        elif od >= 1.000:
            sra_str = f" Confirmatory SRA: {sra:.0f}% release." if sra is not None else " Awaiting SRA confirmation."
            return f"Anti-PF4 ELISA Moderately Positive (OD {od:.3f} >= 1.000).{sra_str}"
        else:
            return f"Anti-PF4 ELISA Weakly Positive / Borderline (OD {od:.3f} between 0.400-0.999). High false-positive rate; strictly rely on functional SRA."

    def determine_management_protocol(self, score: int, tier: str, d: HITPatientTelemetry) -> Tuple[List[str], List[str], List[str]]:
        cessation = []
        dti_protocol = []
        sentinels = []

        if score <= 3:
            cessation.append("1. LOW RISK (NPV >99%): Do NOT discontinue heparin solely for HIT suspicion.")
            cessation.append("2. DO NOT ORDER Anti-PF4 ELISA (prevents false-positive diagnostic entrapment).")
            cessation.append("3. Investigate non-HIT causes of thrombocytopenia (medications, sepsis, hemodilution).")
            return cessation, dti_protocol, sentinels

        # Intermediate or High Probability Management
        cessation.append("1. STAT HEPARIN CESSATION: Discontinue ALL unfractionated heparin (UFH), low-molecular-weight heparin (LMWH), heparin flushes, and remove heparin-bonded central venous catheters.")
        cessation.append("2. ALLERGY RECORD: Enter 'Heparin Allergy - HIT' into electronic health record.")
        cessation.append("3. LABORATORY ORDERS: STAT Anti-PF4/Heparin IgG ELISA and functional Serotonin Release Assay (SRA).")
        cessation.append("4. DUPLEX ULTRASOUND: Order bilateral lower extremity venous duplex ultrasound to screen for silent DVT (present in up to 50% of HIT).")

        # Argatroban Dosing & Titration
        # Adjust for hepatic impairment / ICU
        has_hepatic_or_icu = d.is_critically_ill_or_icu or d.total_bilirubin_mg_dl > 1.5

        if has_hepatic_or_icu:
            init_dose = 0.5 # mcg/kg/min
            dti_protocol.append(f"1. REDUCED DOSE ARGATROBAN (Hepatic Impairment / ICU Status): Start continuous IV infusion at 0.5 mcg/kg/min ({d.patient_weight_kg * init_dose:.1f} mcg/min).")
        else:
            init_dose = 2.0 # mcg/kg/min
            dti_protocol.append(f"1. STANDARD ARGATROBAN INFUSION: Start continuous IV infusion at 2.0 mcg/kg/min ({d.patient_weight_kg * init_dose:.1f} mcg/min).")

        target_aptt_low = round(d.baseline_aptt_seconds * 1.5, 1)
        target_aptt_high = round(d.baseline_aptt_seconds * 3.0, 1)
        dti_protocol.append(f"2. aPTT MONITORING: Check aPTT at 2 hours post-initiation and 2 hours after every rate adjustment. Target aPTT: {target_aptt_low} - {target_aptt_high} seconds (1.5 - 3.0x baseline {d.baseline_aptt_seconds:.0f}s; max limit 100s).")
        dti_protocol.append("3. ALTERNATIVE DTI (BIVALIRUDIN): If hepatic/renal combined failure, consider Bivalirudin 0.15 mg/kg/h IV infusion.")

        # Warfarin Gangrene Sentinel
        if d.active_warfarin_prescription:
            sentinels.append("🚨 DEADLY WARFARIN CONTRAINDICATION (VENOUS LIMB GANGRENE RISK): Warfarin (Coumadin) must NEVER be given during acute HIT. It rapidly depletes Protein C before prothrombin declines, causing massive microvascular thrombosis and limb gangrene. IMMEDIATELY DISCONTINUE WARFARIN and administer Oral/IV Vitamin K 5 - 10 mg STAT to re-establish Protein C synthesis!")

        sentinels.append("PLATELET TRANSFUSION WARNING: Do NOT administer prophylactic platelet transfusions in HIT (platelet infusions fuel antibody-mediated thrombosis; restrict strictly to life-threatening hemorrhage).")

        return cessation, dti_protocol, sentinels

    def evaluate_case(self, data: HITPatientTelemetry) -> HITEvaluationReport:
        score, tier, breakdown = self.calculate_four_ts_score(data)
        lab_interp = self.interpret_immunoassays(data.anti_pf4_elisa_od_value, data.sra_serotonin_release_percent)
        cessation, dti_plan, sentinels = self.determine_management_protocol(score, tier, data)

        directives = []
        directives.append(f"4Ts SCORE: {score}/8 ({tier}).")
        directives.append(f"LAB STATUS: {lab_interp}.")
        directives.append("THERAPY: Immediate heparin cessation and therapeutic DTI initiation.")

        return HITEvaluationReport(
            patient_id=data.patient_id,
            four_ts_total_score=score,
            four_ts_probability_tier=tier,
            four_ts_breakdown=breakdown,
            elisa_sra_interpretation=lab_interp,
            heparin_cessation_orders=cessation,
            non_heparin_anticoagulation_protocol=dti_plan,
            safety_sentinels=sentinels,
            clinical_ash_chest_directive=" ".join(directives)
        )

# Example Execution & Verification
if __name__ == "__main__":
    engine = HeparinInducedThrombocytopeniaEngine()

    print("=" * 80)
    print("OpenPHR Clinical Hematology HIT 4Ts Score, PF4 ELISA & Argatroban Engine")
    print("=" * 80)

    # Test Case 1: 66-year-old male post-cardiac surgery on Day 7 of unfractionated heparin.
    # Platelets dropped from 280 to 68 k/uL (75.7% drop, nadir >= 20).
    # Developed acute right femoral DVT. No other cause apparent.
    # 4Ts Score: 8/8 (High Probability)! Anti-PF4 ELISA OD = 1.640.
    # Critically ill with Bilirubin 2.2 mg/dL -> Reduced Argatroban 0.5 mcg/kg/min.
    # Active Warfarin flagged -> STAT Vitamin K 10 mg and Discontinuation sentinel!
    hit1 = HITPatientTelemetry(
        patient_id="HEM-HIT-7703",
        age_years=66.0,
        patient_weight_kg=82.0,
        baseline_platelet_count_k_ul=280.0,
        nadir_platelet_count_k_ul=68.0,
        days_since_heparin_started=7,
        new_proven_thrombosis=True,
        other_causes_apparent="None",
        anti_pf4_elisa_od_value=1.640,
        sra_serotonin_release_percent=88.0,
        is_critically_ill_or_icu=True,
        total_bilirubin_mg_dl=2.2,
        baseline_aptt_seconds=30.0,
        active_warfarin_prescription=True
    )

    rep1 = engine.evaluate_case(hit1)

    print(f"\n[Patient {rep1.patient_id} - Hemostasis Assessment]")
    print(f"4Ts Clinical Score: {rep1.four_ts_total_score}/8 ({rep1.four_ts_probability_tier})")
    print("Score Breakdown:")
    for b in rep1.four_ts_breakdown:
        print(f"  • {b}")
    print(f"\nImmunoassay Interpretation:\n  {rep1.elisa_sra_interpretation}")
    print("\nHeparin Cessation Protocol:")
    for c in rep1.heparin_cessation_orders:
        print(f"  {c}")
    print("\nNon-Heparin Anticoagulation (Argatroban DTI):")
    for dti in rep1.non_heparin_anticoagulation_protocol:
        print(f"  {dti}")
    if rep1.safety_sentinels:
        print("\nSafety Sentinels:")
        for s in rep1.safety_sentinels:
            print(f"  🚨 {s}")
    print(f"\nASH / CHEST Consensus Directive:\n{rep1.clinical_ash_chest_directive}")