Spaces:
Running on Zero
Running on Zero
File size: 6,661 Bytes
10ec54c 8c62fe5 10ec54c 8c62fe5 10ec54c 8c62fe5 10ec54c 8c62fe5 10ec54c | 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 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 | """
fallback.py
-----------
Deterministic template-based fallback system for RazorShield explanation layer.
Activated when:
- Model is unavailable / failed to load
- Model inference times out
- Model produces invalid JSON or schema errors
- Model output fails deterministic grounding validation
Ensures 100% reliable execution with zero ungrounded claims or decision overrides.
"""
from __future__ import annotations
from src.explanation.schemas import ExplanationInput, ExplanationOutput
class DeterministicFallbackExplainer:
"""Template-based fallback explanation generator."""
@staticmethod
def generate_fallback_explanation(
input_data: ExplanationInput,
failure_reason: str = "Model fallback activated",
) -> ExplanationOutput:
"""
Generates a 100% grounded template explanation matching ExplanationOutput schema.
"""
state = input_data.incident_state
severity = input_data.severity
score = input_data.incident_score
windows = input_data.suspicious_windows
fe_ratio = input_data.fraud_excess_ratio
vel_ratio = input_data.velocity_ratio
camp_active = input_data.campaign_active
q = (input_data.user_question or "").strip().lower()
# Title
if q:
title = f"RAZOR AI Risk Analysis ({input_data.merchant_id})"
else:
title = f"RazorShield Defensive Risk Assessment: {state} ({severity} Severity)"
# Campaign context string
if camp_active:
camp_ctx = (
f"A promotional campaign is currently active for merchant {input_data.merchant_id}. "
f"Volume velocity ({vel_ratio:.1f}x baseline) is normalized, but fraud excess ({fe_ratio:.1f}x baseline) remains actionable."
)
else:
camp_ctx = (
f"No promotional campaign is active for merchant {input_data.merchant_id}. "
f"Observed volume velocity is {vel_ratio:.1f}x baseline."
)
# Base action
if state == "ALERT":
action = "Initiate immediate merchant review, enforce step-up authentication, and review high-risk transaction batches."
elif state == "INVESTIGATE":
action = "Monitor merchant temporal stream closely and apply selective verification on suspicious transactions."
else:
action = "Maintain standard automated processing."
# Dynamic Question-Specific Summary
if "driver" in q or "main risk" in q or "primary risk" in q:
summary = (
f"The primary risk drivers for merchant {input_data.merchant_id} are the Fraud Excess Ratio ({fe_ratio:.1f}x baseline) "
f"and Spike Probability ({input_data.spike_probability * 100:.1f}%). "
f"The policy engine calculated an incident score of {score:.2f} across {windows} suspicious monitoring windows."
)
elif "flag" in q or "analyst" in q or "review" in q or "should" in q:
if state in ["ALERT", "INVESTIGATE"]:
summary = (
f"Yes, an analyst should review merchant {input_data.merchant_id} because the system is in {state} state ({severity} severity) "
f"with {windows} suspicious windows detected. Recommended action: {action}"
)
else:
summary = (
f"No immediate manual flagging is required for merchant {input_data.merchant_id}. "
f"The merchant is currently in NORMAL state (policy score {score:.2f}, {windows} suspicious windows). "
f"Recommended action: {action}"
)
elif "campaign" in q or "flash sale" in q or "normalization" in q or "how" in q:
summary = (
f"Flash sale campaign normalization adjusts volume velocity thresholds during registered promotional events. "
f"This prevents legitimate traffic spikes from triggering false-positive fraud alerts. "
f"For merchant {input_data.merchant_id}, campaign status is currently {'ACTIVE' if camp_active else 'INACTIVE'}."
)
elif q:
summary = (
f"Addressing your query regarding '{input_data.user_question}': Merchant {input_data.merchant_id} is currently evaluated as {state} "
f"({severity} severity, policy score {score:.2f}). Observed fraud excess ratio is {fe_ratio:.1f}x baseline and volume velocity is {vel_ratio:.1f}x baseline."
)
elif state == "ALERT":
summary = (
f"RazorShield classified merchant {input_data.merchant_id} activity as {state} ({severity} severity, policy score {score:.2f}) "
f"because a fraud anomaly persisted across {windows} consecutive monitoring windows. "
f"The estimated fraud excess ratio is {fe_ratio:.1f}x baseline with a volume velocity of {vel_ratio:.1f}x baseline. "
f"{camp_ctx}"
)
elif state == "INVESTIGATE":
summary = (
f"RazorShield flagged merchant {input_data.merchant_id} activity for {state} ({severity} severity, policy score {score:.2f}) "
f"due to a detected anomaly in {windows} monitoring window. "
f"The fraud excess ratio is {fe_ratio:.1f}x baseline and volume velocity is {vel_ratio:.1f}x baseline. "
f"{camp_ctx}"
)
else: # NORMAL
summary = (
f"RazorShield evaluated merchant {input_data.merchant_id} activity as {state} ({severity} severity, policy score {score:.2f}). "
f"Observed fraud excess ratio is {fe_ratio:.1f}x baseline and volume velocity is {vel_ratio:.1f}x baseline. "
f"{camp_ctx}"
)
# Key signals
key_signals = [
f"Policy Incident Score: {score:.2f}",
f"Fraud Excess Ratio: {fe_ratio:.1f}x baseline",
f"Volume Velocity Ratio: {vel_ratio:.1f}x baseline",
f"Consecutive Suspicious Windows: {windows}",
]
confidence_note = (
f"Explanation generated via deterministic fallback ({failure_reason}). "
f"Decision ({state}) is authoritatively determined by RazorShield policy engine."
)
return ExplanationOutput(
title=title,
summary=summary,
key_signals=key_signals,
campaign_context=camp_ctx,
recommended_action=action,
confidence_note=confidence_note,
)
|