from __future__ import annotations from dataclasses import dataclass, field import math from typing import Any from .risk_rules import derive_case_risk_signals from .schema import canonical_reason_codes, normalize_text DEFAULT_HYPOTHESES = [ "safe", "bank_fraud", "duplicate_billing", "vendor_takeover", "ceo_bec", "phantom_vendor", "supply_chain_compromise", "insider_collusion", "multi_entity_layering", "campaign_fraud", "split_payment", "threshold_evasion", ] HYPOTHESIS_TO_DECISION = { "safe": "PAY", "bank_fraud": "ESCALATE_FRAUD", "duplicate_billing": "HOLD", "vendor_takeover": "ESCALATE_FRAUD", "ceo_bec": "ESCALATE_FRAUD", "phantom_vendor": "ESCALATE_FRAUD", "supply_chain_compromise": "ESCALATE_FRAUD", "insider_collusion": "ESCALATE_FRAUD", "multi_entity_layering": "ESCALATE_FRAUD", "campaign_fraud": "ESCALATE_FRAUD", "split_payment": "HOLD", "threshold_evasion": "NEEDS_REVIEW", } ATTACK_NAME_TO_HYPOTHESIS = { "bank_override_attack": "bank_fraud", "vendor_takeover_attack": "vendor_takeover", "ceo_fraud_attack": "ceo_bec", "domain_typosquat_attack": "vendor_takeover", "near_duplicate_invoice_attack": "duplicate_billing", "fake_receipt_attack": "duplicate_billing", "phantom_vendor_attack": "phantom_vendor", "inflated_line_items_attack": "duplicate_billing", "urgency_spoof_attack": "ceo_bec", "approval_threshold_evasion_attack": "threshold_evasion", "workflow_override_attack": "insider_collusion", "split_payment_attack": "split_payment", "coordinated_campaign_attack": "campaign_fraud", "supply_chain_compromise_attack": "supply_chain_compromise", "insider_collusion_attack": "insider_collusion", "multi_entity_layering_attack": "multi_entity_layering", } LIKELIHOOD_TABLES: dict[str, dict[str, dict[str, float]]] = { "compare_bank_account": { "mismatch": { "safe": 0.02, "bank_fraud": 0.95, "duplicate_billing": 0.05, "vendor_takeover": 0.88, "ceo_bec": 0.72, "phantom_vendor": 0.60, "supply_chain_compromise": 0.84, "insider_collusion": 0.46, "multi_entity_layering": 0.68, "campaign_fraud": 0.64, "split_payment": 0.18, "threshold_evasion": 0.16, }, "match": { "safe": 0.98, "bank_fraud": 0.05, "duplicate_billing": 0.95, "vendor_takeover": 0.12, "ceo_bec": 0.28, "phantom_vendor": 0.40, "supply_chain_compromise": 0.16, "insider_collusion": 0.54, "multi_entity_layering": 0.32, "campaign_fraud": 0.36, "split_payment": 0.82, "threshold_evasion": 0.84, }, }, "search_ledger": { "duplicate_found": { "safe": 0.03, "bank_fraud": 0.10, "duplicate_billing": 0.92, "vendor_takeover": 0.14, "ceo_bec": 0.08, "phantom_vendor": 0.06, "supply_chain_compromise": 0.16, "insider_collusion": 0.20, "multi_entity_layering": 0.72, "campaign_fraud": 0.80, "split_payment": 0.88, "threshold_evasion": 0.74, }, "no_duplicate": { "safe": 0.97, "bank_fraud": 0.90, "duplicate_billing": 0.08, "vendor_takeover": 0.86, "ceo_bec": 0.92, "phantom_vendor": 0.94, "supply_chain_compromise": 0.84, "insider_collusion": 0.80, "multi_entity_layering": 0.28, "campaign_fraud": 0.20, "split_payment": 0.12, "threshold_evasion": 0.26, }, }, "inspect_email_thread": { "domain_spoof_detected": { "safe": 0.01, "bank_fraud": 0.78, "duplicate_billing": 0.05, "vendor_takeover": 0.92, "ceo_bec": 0.94, "phantom_vendor": 0.62, "supply_chain_compromise": 0.87, "insider_collusion": 0.55, "multi_entity_layering": 0.50, "campaign_fraud": 0.67, "split_payment": 0.18, "threshold_evasion": 0.24, }, "domain_clean": { "safe": 0.99, "bank_fraud": 0.22, "duplicate_billing": 0.95, "vendor_takeover": 0.08, "ceo_bec": 0.06, "phantom_vendor": 0.38, "supply_chain_compromise": 0.13, "insider_collusion": 0.45, "multi_entity_layering": 0.50, "campaign_fraud": 0.33, "split_payment": 0.82, "threshold_evasion": 0.76, }, }, "lookup_vendor_history": { "suspicious_history": { "safe": 0.10, "bank_fraud": 0.74, "duplicate_billing": 0.18, "vendor_takeover": 0.70, "ceo_bec": 0.30, "phantom_vendor": 0.82, "supply_chain_compromise": 0.78, "insider_collusion": 0.60, "multi_entity_layering": 0.52, "campaign_fraud": 0.48, "split_payment": 0.35, "threshold_evasion": 0.32, }, "clean_history": { "safe": 0.90, "bank_fraud": 0.26, "duplicate_billing": 0.82, "vendor_takeover": 0.30, "ceo_bec": 0.70, "phantom_vendor": 0.18, "supply_chain_compromise": 0.22, "insider_collusion": 0.40, "multi_entity_layering": 0.48, "campaign_fraud": 0.52, "split_payment": 0.65, "threshold_evasion": 0.68, }, }, "callback_verification_result": { "callback_dispute": { "safe": 0.01, "bank_fraud": 0.88, "duplicate_billing": 0.04, "vendor_takeover": 0.85, "ceo_bec": 0.36, "phantom_vendor": 0.66, "supply_chain_compromise": 0.82, "insider_collusion": 0.40, "multi_entity_layering": 0.46, "campaign_fraud": 0.40, "split_payment": 0.10, "threshold_evasion": 0.08, }, "callback_clean": { "safe": 0.94, "bank_fraud": 0.05, "duplicate_billing": 0.90, "vendor_takeover": 0.08, "ceo_bec": 0.32, "phantom_vendor": 0.24, "supply_chain_compromise": 0.15, "insider_collusion": 0.55, "multi_entity_layering": 0.40, "campaign_fraud": 0.44, "split_payment": 0.78, "threshold_evasion": 0.82, }, "callback_suspicious_confirm": { "safe": 0.04, "bank_fraud": 0.08, "duplicate_billing": 0.06, "vendor_takeover": 0.20, "ceo_bec": 0.34, "phantom_vendor": 0.12, "supply_chain_compromise": 0.28, "insider_collusion": 0.62, "multi_entity_layering": 0.36, "campaign_fraud": 0.28, "split_payment": 0.10, "threshold_evasion": 0.12, }, }, "duplicate_cluster_report": { "cluster_detected": { "safe": 0.02, "bank_fraud": 0.06, "duplicate_billing": 0.95, "vendor_takeover": 0.10, "ceo_bec": 0.04, "phantom_vendor": 0.06, "supply_chain_compromise": 0.12, "insider_collusion": 0.18, "multi_entity_layering": 0.76, "campaign_fraud": 0.82, "split_payment": 0.90, "threshold_evasion": 0.78, }, "no_cluster": { "safe": 0.98, "bank_fraud": 0.94, "duplicate_billing": 0.05, "vendor_takeover": 0.90, "ceo_bec": 0.96, "phantom_vendor": 0.94, "supply_chain_compromise": 0.88, "insider_collusion": 0.82, "multi_entity_layering": 0.24, "campaign_fraud": 0.18, "split_payment": 0.10, "threshold_evasion": 0.22, }, }, "bank_change_approval_chain": { "mismatch_found": { "safe": 0.01, "bank_fraud": 0.92, "duplicate_billing": 0.05, "vendor_takeover": 0.78, "ceo_bec": 0.40, "phantom_vendor": 0.55, "supply_chain_compromise": 0.84, "insider_collusion": 0.62, "multi_entity_layering": 0.48, "campaign_fraud": 0.44, "split_payment": 0.08, "threshold_evasion": 0.06, }, "chain_clean": { "safe": 0.99, "bank_fraud": 0.08, "duplicate_billing": 0.95, "vendor_takeover": 0.22, "ceo_bec": 0.60, "phantom_vendor": 0.45, "supply_chain_compromise": 0.16, "insider_collusion": 0.38, "multi_entity_layering": 0.52, "campaign_fraud": 0.56, "split_payment": 0.92, "threshold_evasion": 0.94, }, }, "po_reconciliation_report": { "reconciled_with_flags": { "safe": 0.04, "bank_fraud": 0.08, "duplicate_billing": 0.70, "vendor_takeover": 0.12, "ceo_bec": 0.10, "phantom_vendor": 0.74, "supply_chain_compromise": 0.34, "insider_collusion": 0.58, "multi_entity_layering": 0.22, "campaign_fraud": 0.24, "split_payment": 0.42, "threshold_evasion": 0.46, }, "po_clean": { "safe": 0.96, "bank_fraud": 0.92, "duplicate_billing": 0.30, "vendor_takeover": 0.88, "ceo_bec": 0.90, "phantom_vendor": 0.26, "supply_chain_compromise": 0.66, "insider_collusion": 0.42, "multi_entity_layering": 0.78, "campaign_fraud": 0.76, "split_payment": 0.58, "threshold_evasion": 0.54, }, }, "receipt_reconciliation_report": { "reconciled_with_flags": { "safe": 0.06, "bank_fraud": 0.10, "duplicate_billing": 0.62, "vendor_takeover": 0.14, "ceo_bec": 0.10, "phantom_vendor": 0.70, "supply_chain_compromise": 0.30, "insider_collusion": 0.50, "multi_entity_layering": 0.20, "campaign_fraud": 0.20, "split_payment": 0.38, "threshold_evasion": 0.40, }, "receipt_clean": { "safe": 0.94, "bank_fraud": 0.90, "duplicate_billing": 0.38, "vendor_takeover": 0.86, "ceo_bec": 0.90, "phantom_vendor": 0.30, "supply_chain_compromise": 0.70, "insider_collusion": 0.50, "multi_entity_layering": 0.80, "campaign_fraud": 0.80, "split_payment": 0.62, "threshold_evasion": 0.60, }, }, } @dataclass class SPRTState: hypotheses: list[str] log_likelihood_ratios: dict[str, float] upper_boundaries: dict[str, float] lower_boundaries: dict[str, float] observations_used: int = 0 decision_ready: bool = False optimal_stopping_reached: bool = False expected_sample_number: float = 0.0 distance_to_boundary: dict[str, float] = field(default_factory=dict) prior: dict[str, float] = field(default_factory=dict) posterior_probabilities: dict[str, float] = field(default_factory=dict) accepted_hypothesis: str | None = None recommended_decision: str | None = None belief_entropy: float = 0.0 last_observation: dict[str, Any] = field(default_factory=dict) def _normalize_prior(hypotheses: list[str], prior: dict[str, float] | None) -> dict[str, float]: if not hypotheses: raise ValueError("SPRT requires at least one hypothesis.") if prior is None: base = 1.0 / len(hypotheses) return {hypothesis: base for hypothesis in hypotheses} cleaned = {hypothesis: max(0.0, float(prior.get(hypothesis, 0.0) or 0.0)) for hypothesis in hypotheses} total = sum(cleaned.values()) if total <= 0: base = 1.0 / len(hypotheses) return {hypothesis: base for hypothesis in hypotheses} return {hypothesis: value / total for hypothesis, value in cleaned.items()} def _posterior_from_llrs( hypotheses: list[str], prior: dict[str, float], llrs: dict[str, float], ) -> dict[str, float]: safe_mass = max(prior.get("safe", 1e-6), 1e-9) numerators: dict[str, float] = {"safe": safe_mass} for hypothesis in hypotheses: if hypothesis == "safe": continue numerators[hypothesis] = max(prior.get(hypothesis, 1e-9), 1e-9) * math.exp(llrs.get(hypothesis, 0.0)) total = sum(numerators.values()) return {hypothesis: numerators.get(hypothesis, 0.0) / total for hypothesis in hypotheses} def _entropy(probabilities: dict[str, float]) -> float: entropy = 0.0 for probability in probabilities.values(): if probability > 0.0: entropy -= probability * math.log(probability) return entropy def _recompute_summary(state: SPRTState) -> SPRTState: state.posterior_probabilities = _posterior_from_llrs( state.hypotheses, state.prior, state.log_likelihood_ratios, ) distances: dict[str, float] = {} accepted: str | None = None accepted_score = float("-inf") for hypothesis in state.hypotheses: if hypothesis == "safe": distances[hypothesis] = round(1.0 - state.posterior_probabilities.get("safe", 0.0), 4) continue upper = max(state.upper_boundaries.get(hypothesis, 1.0), 1e-6) llr = state.log_likelihood_ratios.get(hypothesis, 0.0) distances[hypothesis] = round(max(0.0, (upper - llr) / upper), 4) if llr >= upper and llr > accepted_score: accepted = hypothesis accepted_score = llr rejected_all = all( state.log_likelihood_ratios.get(hypothesis, 0.0) <= state.lower_boundaries.get(hypothesis, -math.inf) for hypothesis in state.hypotheses if hypothesis != "safe" ) if accepted is None and rejected_all: accepted = "safe" state.distance_to_boundary = distances state.accepted_hypothesis = accepted state.recommended_decision = HYPOTHESIS_TO_DECISION.get( accepted or max(state.posterior_probabilities, key=state.posterior_probabilities.get), "NEEDS_REVIEW", ) state.decision_ready = accepted is not None average_gap = 0.0 active = 0 for hypothesis in state.hypotheses: if hypothesis == "safe": continue active += 1 average_gap += max(0.0, state.upper_boundaries[hypothesis] - state.log_likelihood_ratios[hypothesis]) average_gap = average_gap / max(active, 1) state.expected_sample_number = round(max(0.0, average_gap / 0.45), 4) state.belief_entropy = round(_entropy(state.posterior_probabilities), 6) return state def initialize_sprt( hypotheses: list[str] | None = None, alpha: float = 0.05, beta: float = 0.10, prior: dict[str, float] | None = None, ) -> SPRTState: hypotheses = list(hypotheses or DEFAULT_HYPOTHESES) if "safe" not in hypotheses: hypotheses.insert(0, "safe") prior_distribution = _normalize_prior(hypotheses, prior) upper = math.log((1.0 - beta) / max(alpha, 1e-9)) lower = math.log(max(beta, 1e-9) / (1.0 - alpha)) state = SPRTState( hypotheses=hypotheses, log_likelihood_ratios={hypothesis: 0.0 for hypothesis in hypotheses if hypothesis != "safe"}, upper_boundaries={hypothesis: upper for hypothesis in hypotheses if hypothesis != "safe"}, lower_boundaries={hypothesis: lower for hypothesis in hypotheses if hypothesis != "safe"}, prior=prior_distribution, ) return _recompute_summary(state) def latent_hypothesis_from_case(case: dict[str, Any]) -> str: metadata = case.get("generator_metadata", {}) or {} attacks = metadata.get("applied_attacks", []) or [] for attack in attacks: mapped = ATTACK_NAME_TO_HYPOTHESIS.get(str(attack)) if mapped: return mapped signals = set(derive_case_risk_signals(case.get("gold", {}) or {})) signals.discard("unsafe_if_pay") if not signals: return "safe" if {"shared_bank_account", "coordinated_timing", "vendor_account_takeover_suspected"} <= signals: return "multi_entity_layering" if {"shared_bank_account", "coordinated_timing"} & signals: return "campaign_fraud" if {"vendor_account_takeover_suspected", "bank_override_attempt"} <= signals: return "supply_chain_compromise" if {"policy_bypass_attempt", "approval_threshold_evasion"} <= signals: return "insider_collusion" if {"urgent_payment_pressure", "sender_domain_spoof"} & signals: return "ceo_bec" if {"sender_domain_spoof", "vendor_account_takeover_suspected"} & signals: return "vendor_takeover" if {"duplicate_near_match", "approval_threshold_evasion"} <= signals: return "split_payment" if "approval_threshold_evasion" in signals: return "threshold_evasion" if "duplicate_near_match" in signals: return "duplicate_billing" if "bank_override_attempt" in signals: return "bank_fraud" return "safe" def infer_tool_observation(tool_name: str, observation: dict[str, Any]) -> str | None: tool = normalize_text(tool_name) if tool == "compare_bank_account": return "match" if bool(observation.get("matched")) else "mismatch" if tool == "search_ledger": return "duplicate_found" if int(observation.get("count", 0) or 0) > 0 else "no_duplicate" if tool == "inspect_email_thread": thread = observation.get("thread", {}) or {} sender_profile = thread.get("sender_profile", {}) or {} request_signals = thread.get("request_signals", {}) or {} suspicious = ( normalize_text(sender_profile.get("domain_alignment")) == "mismatch" or bool(request_signals.get("callback_discouraged")) or bool(request_signals.get("policy_override_language")) or bool(request_signals.get("urgency_language")) ) return "domain_spoof_detected" if suspicious else "domain_clean" if tool == "lookup_vendor_history": suspicious_flags = observation.get("derived_flags", []) or [] return "suspicious_history" if suspicious_flags or observation.get("history") else "clean_history" if tool == "callback_verification_result": details = observation.get("details", {}) or {} risk_signal = normalize_text(details.get("risk_signal") or details.get("outcome")) if risk_signal in {"callback_dispute_confirmed", "dispute", "callback_dispute"}: return "callback_dispute" if risk_signal in {"callback_suspicious_confirm", "adversarial_confirm"}: return "callback_suspicious_confirm" return "callback_clean" if tool == "duplicate_cluster_report": details = observation.get("details", {}) or {} return "cluster_detected" if normalize_text(details.get("status")) == "cluster_detected" else "no_cluster" if tool == "bank_change_approval_chain": details = observation.get("details", {}) or {} return "mismatch_found" if normalize_text(details.get("status")) == "mismatch_found" else "chain_clean" if tool == "po_reconciliation_report": details = observation.get("details", {}) or {} return "reconciled_with_flags" if normalize_text(details.get("status")) == "reconciled_with_flags" else "po_clean" if tool == "receipt_reconciliation_report": details = observation.get("details", {}) or {} return "reconciled_with_flags" if normalize_text(details.get("status")) == "reconciled_with_flags" else "receipt_clean" return None def possible_observations(tool_name: str) -> list[str]: return list(LIKELIHOOD_TABLES.get(normalize_text(tool_name), {}).keys()) def observation_probability(tool_name: str, observation_key: str, hypothesis: str) -> float: table = LIKELIHOOD_TABLES.get(normalize_text(tool_name), {}) observation_row = table.get(observation_key, {}) probability = float(observation_row.get(hypothesis, 0.5) or 0.5) return min(0.999, max(0.001, probability)) def update_sprt( state: SPRTState, tool_name: str, observation: dict[str, Any], likelihood_model: dict[str, dict[str, dict[str, float]]] | None = None, ) -> SPRTState: model = likelihood_model or LIKELIHOOD_TABLES observation_key = observation.get("observation_key") if isinstance(observation, dict) else None if not observation_key: observation_key = infer_tool_observation(tool_name, observation) table = model.get(normalize_text(tool_name), {}) state.observations_used += 1 state.last_observation = { "tool_name": tool_name, "observation_key": observation_key, } if not table or observation_key is None or observation_key not in table: return _recompute_summary(state) for hypothesis in state.hypotheses: if hypothesis == "safe": continue numerator = observation_probability(tool_name, observation_key, hypothesis) denominator = observation_probability(tool_name, observation_key, "safe") state.log_likelihood_ratios[hypothesis] += math.log(numerator / denominator) state.log_likelihood_ratios[hypothesis] = round(state.log_likelihood_ratios[hypothesis], 6) return _recompute_summary(state) def sprt_potential(state: SPRTState) -> float: ratios = [] for hypothesis, llr in state.log_likelihood_ratios.items(): upper = max(state.upper_boundaries.get(hypothesis, 1.0), 1e-6) ratios.append(max(0.0, min(1.0, llr / upper))) if not ratios: return 0.0 return round(max(ratios), 4) def optimal_stopping_check( state: SPRTState, budget_remaining: float, *, max_remaining_voi: float | None = None, min_tool_cost: float = 0.15, ) -> dict[str, Any]: posterior = state.posterior_probabilities best_hypothesis = max(posterior, key=posterior.get) best_confidence = posterior[best_hypothesis] should_stop = False if state.decision_ready: should_stop = True elif budget_remaining <= min_tool_cost: should_stop = True elif max_remaining_voi is not None and max_remaining_voi < min_tool_cost: should_stop = True elif best_confidence >= 0.86 and state.expected_sample_number > budget_remaining / max(min_tool_cost, 1e-6): should_stop = True recommendation = state.accepted_hypothesis or best_hypothesis decision = HYPOTHESIS_TO_DECISION.get(recommendation, "NEEDS_REVIEW") evidence_sufficiency = 1.0 - min(1.0, min(state.distance_to_boundary.values() or [1.0])) state.optimal_stopping_reached = should_stop return { "should_stop": should_stop, "recommended_hypothesis": recommendation, "recommended_decision": decision, "confidence": round(best_confidence, 4), "evidence_sufficiency": round(evidence_sufficiency, 4), } def sprt_state_payload(state: SPRTState) -> dict[str, Any]: return { "hypotheses": list(state.hypotheses), "log_likelihood_ratios": { hypothesis: round(value, 4) for hypothesis, value in state.log_likelihood_ratios.items() }, "posterior_probabilities": { hypothesis: round(value, 4) for hypothesis, value in state.posterior_probabilities.items() }, "upper_boundaries": { hypothesis: round(value, 4) for hypothesis, value in state.upper_boundaries.items() }, "lower_boundaries": { hypothesis: round(value, 4) for hypothesis, value in state.lower_boundaries.items() }, "observations_used": state.observations_used, "decision_ready": state.decision_ready, "optimal_stopping_reached": state.optimal_stopping_reached, "expected_sample_number": round(state.expected_sample_number, 4), "distance_to_boundary": dict(state.distance_to_boundary), "accepted_hypothesis": state.accepted_hypothesis, "recommended_decision": state.recommended_decision, "belief_entropy": round(state.belief_entropy, 6), "potential": sprt_potential(state), "last_observation": dict(state.last_observation), } def canonical_risky_hypotheses(values: list[Any]) -> list[str]: reasons = set(canonical_reason_codes(values)) hypotheses: list[str] = [] if {"shared_bank_account", "coordinated_timing", "vendor_account_takeover_suspected"} <= reasons: hypotheses.append("multi_entity_layering") if {"shared_bank_account", "coordinated_timing"} & reasons: hypotheses.append("campaign_fraud") if {"vendor_account_takeover_suspected", "bank_override_attempt"} <= reasons: hypotheses.append("supply_chain_compromise") if {"policy_bypass_attempt", "approval_threshold_evasion"} <= reasons: hypotheses.append("insider_collusion") if {"urgent_payment_pressure", "sender_domain_spoof"} & reasons: hypotheses.append("ceo_bec") if {"sender_domain_spoof", "vendor_account_takeover_suspected"} & reasons: hypotheses.append("vendor_takeover") if {"duplicate_near_match", "approval_threshold_evasion"} <= reasons: hypotheses.append("split_payment") if "approval_threshold_evasion" in reasons: hypotheses.append("threshold_evasion") if "duplicate_near_match" in reasons: hypotheses.append("duplicate_billing") if "bank_override_attempt" in reasons: hypotheses.append("bank_fraud") return list(dict.fromkeys(hypotheses))