""" decision_engine.py ================== Synthesises all per-model and cross-model signals into: - Unified probability - Dominant mechanism (Splice-driven / Protein-driven / Consensus / Ambiguous) - Confidence level (High / Moderate / Low) - Structured JSON report - Human-readable final explanation Prediction never appears without explanation. """ from __future__ import annotations import json import datetime from dataclasses import dataclass import numpy as np from explainability_engine import ( SpliceModelSignals, V4ModelSignals, ClassicModelSignals, CrossModelAnalysis, classify_risk_tier, ) # ═══════════════════════════════════════════════════════════════════════════════ # Decision result # ═══════════════════════════════════════════════════════════════════════════════ @dataclass class DecisionResult: # Variant chrom: str pos: int ref: str alt: str ref_seq: str mut_seq: str mutation_pos: int # Model outputs splice: SpliceModelSignals v4: V4ModelSignals classic: ClassicModelSignals xai: CrossModelAnalysis # Unified decision unified_probability: float dominant_mechanism: str confidence: str risk_tier: str # Interpretation text splice_analysis: str protein_analysis: str agreement_analysis: str final_explanation: str # Timestamp timestamp: str = "" def __post_init__(self): if not self.timestamp: self.timestamp = datetime.datetime.utcnow().isoformat() + "Z" def to_dict(self) -> dict: return { "variant": { "chromosome": self.chrom, "position": self.pos, "reference": self.ref, "alternate": self.alt, "hgvs_like": f"chr{self.chrom}:g.{self.pos}{self.ref}>{self.alt}", "mutation_pos_in_window": self.mutation_pos, }, "prediction": { "unified_probability": self.unified_probability, "dominant_mechanism": self.dominant_mechanism, "confidence": self.confidence, "risk_tier": self.risk_tier, }, "model_outputs": { "splice": { "probability": self.splice.probability, "importance_score": self.splice.imp_score, "risk_tier": self.splice.risk_tier, "risk_tier_desc": self.splice.risk_tier_desc, "region_importance": self.splice.region_imp.tolist(), "splice_importance": self.splice.splice_imp.tolist(), "counterfactual_delta": self.splice.counterfactual_delta, "counterfactual_table": self.splice.counterfactual_table, "feature_ablation": self.splice.ablation, "splice_aura_donor_bp": self.splice.splice_aura_donor, "splice_aura_acceptor_bp": self.splice.splice_aura_acceptor, "splice_risk_donor": self.splice.splice_risk_donor, "splice_risk_acceptor": self.splice.splice_risk_acceptor, }, "v4": { "probability": self.v4.probability, "importance_score": self.v4.imp_score, "mutation_peak": self.v4.mutation_peak, "mutation_peak_ratio": self.v4.mutation_peak_ratio, "region_importance": self.v4.region_imp.tolist(), "splice_importance": self.v4.splice_imp.tolist(), }, "classic": { "probability": self.classic.probability, "importance_score": self.classic.imp_score, "mutation_peak": self.classic.mutation_peak, }, }, "explainability_analysis": { "mutation_peak_ratio": self.xai.mutation_peak_ratio, "counterfactual_magnitude": self.xai.counterfactual_magnitude, "cross_model_locality_score": self.xai.cross_model_locality_score, "signal_concentration_index": self.xai.signal_concentration_index, "explainability_strength_score": self.xai.explainability_strength, "activation_pattern_type": self.xai.activation_pattern_type, "model_agreement": self.xai.model_agreement, }, "interpretation": { "splice_analysis": self.splice_analysis, "protein_analysis": self.protein_analysis, "agreement_analysis": self.agreement_analysis, "final_explanation": self.final_explanation, }, "analysis_timestamp": self.timestamp, "disclaimers": [ "For research use only.", "Not a clinical diagnostic tool.", "Results must be reviewed by a qualified clinical geneticist.", ], } def to_json(self) -> str: return json.dumps(self.to_dict(), indent=2) # ═══════════════════════════════════════════════════════════════════════════════ # Core decision logic # ═══════════════════════════════════════════════════════════════════════════════ def build_decision( chrom: str, pos: int, ref: str, alt: str, ref_seq: str, mut_seq: str, mutation_pos: int, splice: SpliceModelSignals, v4: V4ModelSignals, classic: ClassicModelSignals, xai: CrossModelAnalysis, ) -> DecisionResult: # ── Unified probability: weighted ensemble ───────────────── # Weights reflect model complexity: splice > v4 > classic unified_prob = round( 0.40 * splice.probability + 0.35 * v4.probability + 0.25 * classic.probability, 4 ) # ── Dominant mechanism ───────────────────────────────────── dominant_mechanism = _determine_mechanism(splice, v4, classic, xai) # ── Confidence level ────────────────────────────────────── confidence = _determine_confidence(xai, splice, v4, classic) # ── Risk tier from unified prob ─────────────────────────── risk_tier, _ = classify_risk_tier(unified_prob) # ── Interpretation paragraphs ───────────────────────────── splice_analysis = _write_splice_analysis(splice, mutation_pos) protein_analysis = _write_protein_analysis(v4, classic) agreement_analysis = _write_agreement_analysis(splice, v4, classic, xai) final_explanation = _write_final_explanation( chrom, pos, ref, alt, unified_prob, risk_tier, dominant_mechanism, confidence, splice, v4, classic, xai, mutation_pos ) return DecisionResult( chrom=chrom, pos=pos, ref=ref, alt=alt, ref_seq=ref_seq, mut_seq=mut_seq, mutation_pos=mutation_pos, splice=splice, v4=v4, classic=classic, xai=xai, unified_probability=unified_prob, dominant_mechanism=dominant_mechanism, confidence=confidence, risk_tier=risk_tier, splice_analysis=splice_analysis, protein_analysis=protein_analysis, agreement_analysis=agreement_analysis, final_explanation=final_explanation, ) # ═══════════════════════════════════════════════════════════════════════════════ # Mechanism + confidence logic # ═══════════════════════════════════════════════════════════════════════════════ def _determine_mechanism( splice: SpliceModelSignals, v4: V4ModelSignals, classic: ClassicModelSignals, xai: CrossModelAnalysis, ) -> str: sp = splice.probability v4p = v4.probability cl = classic.probability splice_dominant = sp > 0.55 and sp > v4p + 0.15 and sp > cl + 0.15 protein_dominant = (v4p + cl) / 2 > sp + 0.15 # Splice-specific signals splice_near_site = ( (splice.splice_aura_donor is not None and splice.splice_aura_donor <= 8) or (splice.splice_aura_acceptor is not None and splice.splice_aura_acceptor <= 8) ) splice_signal_strong = ( float(splice.splice_imp[0]) > 0.55 or float(splice.splice_imp[1]) > 0.55 ) if splice_dominant or (splice_near_site and splice_signal_strong): return "Splice-driven" if protein_dominant: return "Protein-driven" all_agree = max(sp, v4p, cl) - min(sp, v4p, cl) < 0.15 if all_agree and (sp + v4p + cl) / 3 > 0.45: return "Consensus" return "Ambiguous" def _determine_confidence( xai: CrossModelAnalysis, splice: SpliceModelSignals, v4: V4ModelSignals, classic: ClassicModelSignals, ) -> str: score = ( 0.30 * xai.explainability_strength + 0.30 * xai.model_agreement + 0.20 * min(xai.counterfactual_magnitude / 0.4, 1.0) + 0.20 * xai.cross_model_locality_score ) if score >= 0.65: return "High" if score >= 0.40: return "Moderate" return "Low" # ═══════════════════════════════════════════════════════════════════════════════ # Interpretation text generators # ═══════════════════════════════════════════════════════════════════════════════ def _write_splice_analysis(s: SpliceModelSignals, mutation_pos: int) -> str: lines = [] lines.append( f"The splice disruption model assigns probability {s.probability:.4f} " f"({s.risk_tier}, {s.risk_tier_desc})." ) if s.splice_aura_donor is not None and s.splice_aura_donor <= 8: lines.append( f"Mutation at position {mutation_pos} is {s.splice_aura_donor} bp from " f"a GT splice donor (nearest at position {s.nearest_donor_pos}) — " f"risk: {s.splice_risk_donor}." ) else: lines.append("No GT splice donor within critical 8 bp window.") if s.splice_aura_acceptor is not None and s.splice_aura_acceptor <= 8: lines.append( f"Mutation is {s.splice_aura_acceptor} bp from " f"an AG splice acceptor (nearest at position {s.nearest_acceptor_pos}) — " f"risk: {s.splice_risk_acceptor}." ) else: lines.append("No AG splice acceptor within critical 8 bp window.") don_imp = float(s.splice_imp[0]) acc_imp = float(s.splice_imp[1]) reg_imp = float(s.splice_imp[2]) if don_imp > 0.55 or acc_imp > 0.55: lines.append( f"Elevated splice importance: donor={don_imp:.3f}, " f"acceptor={acc_imp:.3f}, region={reg_imp:.3f}. " "Potential exon skipping or aberrant splice site activation." ) lines.append( f"Counterfactual delta: {s.counterfactual_delta:.4f} — " + ("strong causal signal at this position." if s.counterfactual_delta > 0.20 else "moderate positional causality.") ) abl = s.ablation dom = max([("splice", abl["splice_pct"]), ("region", abl["region_pct"]), ("mutation type", abl["mutation_pct"])], key=lambda x: x[1]) lines.append( f"Feature ablation: {dom[0]} features account for {dom[1]:.0f}% of the signal " f"(splice Δ{abl['splice_causal_effect']:.3f}, " f"region Δ{abl['region_causal_effect']:.3f}, " f"mut-type Δ{abl['mutation_causal_effect']:.3f})." ) lines.append( f"Activation pattern at mutation site: {_peak_desc(s.conv3_profile, mutation_pos)}." ) return " ".join(lines) def _write_protein_analysis(v4: V4ModelSignals, classic: ClassicModelSignals) -> str: lines = [] lines.append( f"The v4 protein-context model assigns probability {v4.probability:.4f}. " f"Mutation-centered importance score: {v4.imp_score:.4f}. " f"Peak ratio at mutation site: {v4.mutation_peak_ratio:.3f}x above window mean." ) exon_imp = float(v4.region_imp[0]) intron_imp = float(v4.region_imp[1]) if exon_imp > intron_imp: lines.append( f"v4 assigns higher exonic importance ({exon_imp:.3f} vs intron {intron_imp:.3f}), " "consistent with coding-sequence disruption." ) else: lines.append( f"v4 assigns higher intronic importance ({intron_imp:.3f} vs exon {exon_imp:.3f}), " "suggesting intronic regulatory or splicing mechanism." ) lines.append( f"Classic model independently assigns probability {classic.probability:.4f} " f"with importance score {classic.imp_score:.4f}." ) return " ".join(lines) def _write_agreement_analysis( splice: SpliceModelSignals, v4: V4ModelSignals, classic: ClassicModelSignals, xai: CrossModelAnalysis, ) -> str: probs = [splice.probability, v4.probability, classic.probability] spread = max(probs) - min(probs) mean_p = sum(probs) / 3 if spread < 0.10: agree_str = f"All three models show strong agreement (spread={spread:.3f})." elif spread < 0.25: agree_str = f"Models show moderate agreement (spread={spread:.3f})." else: agree_str = ( f"Models show divergence (spread={spread:.3f}) — " "dominant mechanism may be model-specific." ) return ( f"{agree_str} " f"Splice={splice.probability:.3f}, V4={v4.probability:.3f}, " f"Classic={classic.probability:.3f}. " f"Cross-model locality score: {xai.cross_model_locality_score:.3f} — " + ("activation profiles are spatially aligned across models." if xai.cross_model_locality_score > 0.7 else "some divergence in spatial activation profiles.") + f" Explainability strength: {xai.explainability_strength:.3f}." ) def _write_final_explanation( chrom, pos, ref, alt, unified_prob, risk_tier, mechanism, confidence, splice: SpliceModelSignals, v4: V4ModelSignals, classic: ClassicModelSignals, xai: CrossModelAnalysis, mutation_pos: int, ) -> str: var_str = f"chr{chrom}:g.{pos}{ref}>{alt}" # Opening sentence — prediction grounded in signals opening = ( f"Variant {var_str} is predicted {risk_tier} (unified probability {unified_prob:.4f}) " f"with {confidence} confidence. The primary mechanism is {mechanism}." ) # Why — key signal chain if mechanism == "Splice-driven": why = ( f"The splice model's conv3 activation profile shows a " f"{xai.activation_pattern_type.lower()} peak at the mutation site " f"(peak ratio {xai.mutation_peak_ratio:.2f}x window mean). " ) if splice.splice_aura_donor is not None and splice.splice_aura_donor <= 8: why += ( f"The mutation falls {splice.splice_aura_donor} bp from a GT donor site " f"({splice.splice_risk_donor}), directly within the splice recognition window. " ) if float(splice.splice_imp[0]) > 0.5 or float(splice.splice_imp[1]) > 0.5: why += "Elevated splice donor/acceptor importance scores confirm splice disruption." elif mechanism == "Protein-driven": why = ( f"The v4 and classic models drive the prediction " f"(v4={v4.probability:.3f}, classic={classic.probability:.3f}). " f"Mutation importance: {v4.imp_score:.4f}. " f"Activation concentrated at mutation site (concentration index " f"{xai.signal_concentration_index:.3f})." ) elif mechanism == "Consensus": why = ( f"All three models converge on a pathogenic signal " f"(splice={splice.probability:.3f}, v4={v4.probability:.3f}, " f"classic={classic.probability:.3f}). " f"Model agreement: {xai.model_agreement:.3f}. " "Convergent evidence strengthens confidence." ) else: why = ( f"Model signals are divergent. Splice probability={splice.probability:.3f}, " f"v4={v4.probability:.3f}, classic={classic.probability:.3f}. " "The variant may have context-dependent effects not captured by a single mechanism." ) # Counterfactual sentence cf = ( f"Counterfactual analysis (all alternative mutations at position {mutation_pos}) " f"yielded a probability range of {splice.counterfactual_delta:.4f}, " + (f"confirming strong positional causality." if splice.counterfactual_delta > 0.20 else f"indicating limited positional causality.") ) # Ablation sentence abl = splice.ablation dom = max([("splice context", abl["splice_pct"]), ("genomic region", abl["region_pct"]), ("mutation type", abl["mutation_pct"])], key=lambda x: x[1]) abl_s = ( f"Feature ablation identifies {dom[0]} as the dominant driver " f"({dom[1]:.0f}% of causal signal)." ) # Close close = ( f"Explainability strength score: {xai.explainability_strength:.3f}/1.0. " "⚠ For research use only — not a clinical diagnostic tool." ) return f"{opening}\n\n{why}\n\n{cf}\n\n{abl_s}\n\n{close}" # ── Helpers ─────────────────────────────────────────────────────────────────── def _peak_desc(profile: np.ndarray, pos: int) -> str: if pos < 0 or pos >= len(profile): return "not assessable" v = profile[pos] pct = int(np.searchsorted(np.sort(profile), v) / len(profile) * 100) return f"value={v:.3f}, top {100-pct}% percentile"