Consequences of Unexplainable Decisions:

  Scenario 1: Loan Rejection
  +-----------------------------------+
  | "Your loan application is denied"  |
  | Reason: [none]                     |
  +-----------------------------------+
  -> User complaints, regulatory fines, brand damage

  Scenario 2: Explainable Decision
  +-----------------------------------+
  | "Your loan application is denied"  |
  | Reasons:                           |
  |   1. Credit score 580 < min 620    |
  |   2. 2 late payments in 12 months  |
  |   3. Debt-to-income 0.65 > 0.5    |
  | Suggestion: Reduce debt, reapply   |
  +-----------------------------------+
  -> User understanding, compliance, trust

Four-Level Explainability Model:

  Level 4: Natural Language
  +------------------------------------------+
  | "The system rejected this application     |
  |  because your credit score is below the   |
  |  threshold and your debt ratio is high."  |
  +------------------------------------------+
                    |
  Level 3: Causal Graph
  +------------------------------------------+
  | credit_score(580) ---> below_threshold    |
  | debt_ratio(0.65) ----> over_limit         |
  | below_threshold + over_limit --> REJECT   |
  +------------------------------------------+
                    |
  Level 2: Feature Attribution
  +------------------------------------------+
  | credit_score: -0.35 (largest negative)    |
  | debt_ratio: -0.28                         |
  | income: +0.15                             |
  | employment: +0.08                         |
  +------------------------------------------+
                    |
  Level 1: Rule/Model Trace
  +------------------------------------------+
  | Rule CR-002 fired (priority 90)           |
  | Model: credit_v3 (confidence 0.23)        |
  | Path: PARALLEL_FUSION -> conflict -> rule |
  +------------------------------------------+

from dataclasses import dataclass, field
from typing import Any
from enum import Enum
from datetime import datetime


class ExplanationType(str, Enum):
    RULE_TRACE = "rule_trace"
    FEATURE_ATTRIBUTION = "feature_attribution"
    CAUSAL_GRAPH = "causal_graph"
    NATURAL_LANGUAGE = "natural_language"
    COUNTERFACTUAL = "counterfactual"


@dataclass
class FactorContribution:
    """Single factor contribution"""
    factor_name: str
    factor_value: Any
    contribution: float       # -1.0 to 1.0
    direction: str            # "positive", "negative", "neutral"
    description: str

    @property
    def impact_level(self) -> str:
        abs_c = abs(self.contribution)
        if abs_c >= 0.3:
            return "high"
        if abs_c >= 0.1:
            return "medium"
        return "low"


@dataclass
class RuleTraceNode:
    """Rule trace node"""
    node_type: str            # "fact", "condition", "rule", "conclusion"
    node_id: str
    label: str
    value: Any = None
    children: list["RuleTraceNode"] = field(default_factory=list)
    metadata: dict = field(default_factory=dict)


@dataclass
class CausalLink:
    """Causal link"""
    source: str
    target: str
    strength: float           # 0.0 to 1.0
    relationship: str         # "causes", "contributes", "blocks"


@dataclass
class Explanation:
    """Complete explanation object"""
    explanation_id: str
    decision_id: str
    timestamp: datetime
    decision: str
    confidence: float

    # Level 1: Rule trace
    rule_trace: RuleTraceNode | None = None
    fired_rules: list[str] = field(default_factory=list)

    # Level 2: Feature attribution
    factors: list[FactorContribution] = field(default_factory=list)

    # Level 3: Causal graph
    causal_links: list[CausalLink] = field(default_factory=list)

    # Level 4: Natural language
    summary: str = ""
    detailed_explanation: str = ""

    # Counterfactuals
    counterfactuals: list[dict] = field(default_factory=list)

    def top_factors(self, n: int = 3) -> list[FactorContribution]:
        return sorted(
            self.factors,
            key=lambda f: abs(f.contribution),
            reverse=True,
        )[:n]

class RuleTraceBuilder:
    """Builds rule reasoning trace chains"""

    def build_trace(self, fired_rules: list, facts: dict,
                    working_memory: dict) -> RuleTraceNode:
        root = RuleTraceNode(
            node_type="root", node_id="trace_root",
            label="Reasoning Process",
        )

        # Input facts
        facts_node = RuleTraceNode(
            node_type="facts", node_id="input_facts",
            label="Input Facts",
        )
        for key, value in facts.items():
            facts_node.children.append(RuleTraceNode(
                node_type="fact", node_id=f"fact_{key}",
                label=key, value=value,
            ))
        root.children.append(facts_node)

        # Rule matches
        for rule in fired_rules:
            rule_node = RuleTraceNode(
                node_type="rule", node_id=rule.rule_id,
                label=rule.name,
                metadata={"priority": rule.priority,
                          "cycle": rule.fired_at_cycle},
            )

            for cond in rule.conditions:
                fact_value = facts.get(cond.attribute)
                matched = cond.evaluate(fact_value)
                rule_node.children.append(RuleTraceNode(
                    node_type="condition",
                    node_id=f"{rule.rule_id}_{cond.attribute}",
                    label=f"{cond.attribute} {cond.operator} {cond.value}",
                    value={"actual": fact_value, "matched": matched},
                ))

            rule_node.children.append(RuleTraceNode(
                node_type="conclusion",
                node_id=f"{rule.rule_id}_conclusion",
                label=rule.conclusion, value=rule.action_result,
            ))
            root.children.append(rule_node)

        return root

    def to_ascii_tree(self, node: RuleTraceNode,
                      prefix: str = "", is_last: bool = True) -> str:
        connector = "+-- " if is_last else "|-- "
        lines = [f"{prefix}{connector}[{node.node_type}] {node.label}"]

        if node.value is not None:
            val_prefix = prefix + ("    " if is_last else "|   ")
            lines.append(f"{val_prefix}= {node.value}")

        child_prefix = prefix + ("    " if is_last else "|   ")
        for i, child in enumerate(node.children):
            lines.append(self.to_ascii_tree(
                child, child_prefix, i == len(node.children) - 1
            ))

        return "\n".join(lines)

Rule Trace Tree:

+-- [root] Reasoning Process
    |-- [facts] Input Facts
    |   |-- [fact] credit_score
    |   |   = 580
    |   |-- [fact] annual_income
    |   |   = 180000
    |   |-- [fact] debt_ratio
    |   |   = 0.65
    |   +-- [fact] previous_defaults
    |       = 2
    |
    |-- [rule] Low Credit Rejection (CR-002, priority=90)
    |   |-- [condition] credit_score < 500
    |   |   = {"actual": 580, "matched": false}
    |   |-- [condition] previous_defaults >= 3
    |   |   = {"actual": 2, "matched": false}
    |   +-- [conclusion] NO MATCH
    |
    +-- [rule] Medium Credit Conditional (CR-003, priority=50)
        |-- [condition] credit_score >= 600
        |   = {"actual": 580, "matched": false}
        +-- [conclusion] NO MATCH (first condition failed)

Final: No rule matched -> forwarded to ML layer

import numpy as np
from itertools import combinations


class SHAPExplainer:
    """SHAP (SHapley Additive exPlanations) calculator"""

    def __init__(self, model, feature_names: list[str]):
        self._model = model
        self._features = feature_names

    def explain(self, instance: dict) -> list[FactorContribution]:
        n = len(self._features)
        shap_values = {}

        for feature in self._features:
            shap_values[feature] = self._shapley_value(instance, feature)

        total = sum(abs(v) for v in shap_values.values()) or 1.0

        contributions = []
        for name in self._features:
            normalized = shap_values[name] / total
            direction = ("positive" if normalized > 0.01
                        else "negative" if normalized < -0.01
                        else "neutral")
            contributions.append(FactorContribution(
                factor_name=name,
                factor_value=instance.get(name),
                contribution=normalized,
                direction=direction,
                description=self._describe(name, instance.get(name), normalized),
            ))

        return sorted(contributions,
                      key=lambda c: abs(c.contribution), reverse=True)

    def _shapley_value(self, instance: dict, feature: str) -> float:
        others = [f for f in self._features if f != feature]
        n = len(others)
        shapley = 0.0

        for size in range(n + 1):
            for subset in combinations(others, size):
                subset_set = set(subset)
                with_input = {f: instance[f] for f in subset_set | {feature}}
                without_input = {f: instance[f] for f in subset_set}

                pred_with, _ = self._model.predict(with_input)
                pred_without, _ = self._model.predict(without_input)

                marginal = self._numeric(pred_with) - self._numeric(pred_without)
                weight = (np.math.factorial(size) *
                         np.math.factorial(n - size - 1)) / np.math.factorial(n)
                shapley += weight * marginal

        return shapley

    def _numeric(self, prediction: str) -> float:
        return {"approve": 1.0, "conditional_approve": 0.5, "reject": 0.0}.get(
            prediction, 0.5
        )

    def _describe(self, name: str, value: Any, contribution: float) -> str:
        impact = ("strong" if abs(contribution) >= 0.3
                 else "moderate" if abs(contribution) >= 0.1
                 else "slight")
        direction = "positive" if contribution > 0 else "negative"
        return f"{name}={value} has {impact} {direction} impact on the decision"

Feature Attribution (SHAP Values):

  Factor              Value     Contrib   Dir       Impact
  ---------------------------------------------------------------
  credit_score        580      -0.35     <<<<<<  [==========] High-Neg
  debt_ratio          0.65     -0.28     <<<<    [========]   High-Neg
  annual_income       180000   +0.15       >>>   [====]       Med-Pos
  employment_yrs      5        +0.08       >>    [==]         Low-Pos
  previous_defaults   2        -0.12     <<<     [===]        Med-Neg
  industry            tech     +0.02       >     [=]          Low-Pos

  Negative <<<<<<<<<<|>>>>>>>>> Positive

@dataclass
class CausalNode:
    node_id: str
    label: str
    node_type: str     # "input", "intermediate", "output"
    value: Any = None


class CausalGraphBuilder:
    """Causal graph builder"""

    def __init__(self):
        self._nodes: dict[str, CausalNode] = {}
        self._edges: list[CausalLink] = []

    def build_from_explanation(
        self, factors: list[FactorContribution],
        rules_fired: list, decision: str
    ) -> tuple[list[CausalNode], list[CausalLink]]:
        self._nodes.clear()
        self._edges.clear()

        for factor in factors:
            node = CausalNode(
                node_id=f"input_{factor.factor_name}",
                label=f"{factor.factor_name}={factor.factor_value}",
                node_type="input", value=factor.factor_value,
            )
            self._nodes[node.node_id] = node

        for rule in rules_fired:
            for cond in rule.conditions:
                mid_id = f"check_{cond.attribute}_{cond.operator}"
                self._nodes[mid_id] = CausalNode(
                    node_id=mid_id,
                    label=f"{cond.attribute} {cond.operator} {cond.value}",
                    node_type="intermediate",
                )
                self._edges.append(CausalLink(
                    source=f"input_{cond.attribute}",
                    target=mid_id,
                    strength=abs(next(
                        f.contribution for f in factors
                        if f.factor_name == cond.attribute
                    )),
                    relationship="causes",
                ))

        self._nodes["decision"] = CausalNode(
            node_id="decision",
            label=f"Decision: {decision}",
            node_type="output", value=decision,
        )

        return list(self._nodes.values()), self._edges

    def to_ascii(self) -> str:
        lines = ["Causal Graph:", ""]
        inputs = [n for n in self._nodes.values() if n.node_type == "input"]
        mids = [n for n in self._nodes.values() if n.node_type == "intermediate"]
        outputs = [n for n in self._nodes.values() if n.node_type == "output"]

        lines.append("  Input Factors:")
        for n in inputs:
            lines.append(f"    [{n.label}]")
        lines.append("        |")
        lines.append("        v")
        lines.append("  Reasoning:")
        for n in mids:
            result = "PASS" if n.value else "FAIL"
            lines.append(f"    [{n.label}] -> {result}")
        lines.append("        |")
        lines.append("        v")
        lines.append("  Decision:")
        for n in outputs:
            lines.append(f"    [{n.label}]")
        return "\n".join(lines)

Causal Graph:

  Input Factors:
    [credit_score=580] --+
    [debt_ratio=0.65] ---+---> [credit_score >= 600?] -> FAIL
    [income=180000] -----+
    [defaults=2] --------+---> [defaults < 2?] -> FAIL
                          |
                          v
  Reasoning:
    [Rule: CR-003 Medium Credit] -> NOT MATCHED
    [ML Model: credit_v3] -> REJECT (confidence: 0.77)
                          |
                          v
  Fusion:
    [Fusion: ML wins] -> confidence: 0.62
                          |
                          v
  Final:
    [Decision: REJECT] -- needs human review (conf < 0.7)

class NaturalLanguageExplainer:
    """Natural language explanation generator"""

    def __init__(self):
        self._templates = {
            "approve": {
                "en": "The system approved this {domain} application, primarily "
                      "because {top_positive_factors}. "
                      "Overall confidence: {confidence:.0%}.",
            },
            "reject": {
                "en": "The system rejected this {domain} application. "
                      "Main reasons: {top_negative_factors}. {suggestion}",
            },
            "conditional_approve": {
                "en": "The system conditionally approved this {domain} "
                      "application. {conditions_text}.",
            },
        }

    def generate(self, explanation: Explanation, domain: str,
                 lang: str = "en") -> str:
        template = self._templates.get(
            explanation.decision, self._templates["reject"]
        )[lang]

        top = explanation.top_factors(3)
        positive = [f for f in top if f.direction == "positive"]
        negative = [f for f in top if f.direction == "negative"]

        pos_text = "; ".join(f.description for f in positive) or "all metrics met"
        neg_text = "; ".join(f.description for f in negative) or "none"
        suggestion = self._suggestion(explanation.decision, negative)

        return template.format(
            domain=domain,
            confidence=explanation.confidence,
            top_positive_factors=pos_text,
            top_negative_factors=neg_text,
            suggestion=suggestion,
            conditions_text=neg_text,
        )

    def _suggestion(self, decision: str,
                    negatives: list[FactorContribution]) -> str:
        if decision != "reject" or not negatives:
            return ""
        parts = [f"improve {f.factor_name}" for f in negatives]
        return "Suggestions: " + ", ".join(parts) + " before reapplying."

class CounterfactualExplainer:
    """Counterfactual: 'What if X were different?'"""

    def __init__(self, reasoning_engine):
        self._engine = reasoning_engine

    def generate_counterfactuals(
        self, original_input: dict,
        original_decision: str,
        target_decision: str,
        max_changes: int = 3,
    ) -> list[dict]:
        counterfactuals = []

        for key in original_input:
            adjusted = self._find_threshold(
                original_input, key, target_decision
            )
            if adjusted is not None:
                counterfactuals.append({
                    "changed_factor": key,
                    "original_value": original_input[key],
                    "required_value": adjusted,
                    "new_decision": target_decision,
                    "description": (
                        f"If {key} changed from {original_input[key]} "
                        f"to {adjusted}, the decision would be {target_decision}"
                    ),
                })

        counterfactuals.sort(
            key=lambda c: abs(c["required_value"] - c["original_value"])
            if isinstance(c["original_value"], (int, float)) else 0
        )
        return counterfactuals[:max_changes]

    def _find_threshold(self, inputs: dict, key: str, target: str) -> Any:
        value = inputs[key]
        if not isinstance(value, (int, float)):
            return None

        low, high = value * 0.5, value * 2.0
        for _ in range(20):
            mid = (low + high) / 2
            result = self._engine.evaluate({**inputs, key: mid})
            if result and result.get("decision") == target:
                high = mid
            else:
                low = mid

        final = self._engine.evaluate({**inputs, key: high})
        if final and final.get("decision") == target:
            return round(high, 2) if isinstance(value, float) else int(high)
        return None

Explanation Service Architecture:

  Reasoning Request
      |
      v
  +--------+
  | Reason |---> Decision Result
  | Engine |        |
  +--------+        |
      |              |
      v              v
  +--------+   +---------+
  | Trace  |   | SHAP    |
  | Collect|   | Compute |
  +--------+   +---------+
      |              |
      +------+-------+
             |
             v
      +------+------+
      | Causal Graph|
      +------+------+
             |
             v
      +------+------+
      | NL Generate |
      +------+------+
             |
             v
      +------+------+
      | Explanation |
      | Store       |
      +------+------+
             |
             v
        gRPC Response

syntax = "proto3";
package onto.explainability.v1;

service ExplainabilityService {
    rpc GetExplanation(GetExplanationRequest)
        returns (ExplanationResponse);
    rpc GetCounterfactuals(CounterfactualRequest)
        returns (CounterfactualResponse);
    rpc GetFactorAttribution(AttributionRequest)
        returns (AttributionResponse);
}

message GetExplanationRequest {
    string decision_id = 1;
    string language = 2;
    repeated string levels = 3;
}

message ExplanationResponse {
    string decision = 1;
    double confidence = 2;
    string summary = 3;
    string detailed_explanation = 4;
    repeated Factor factors = 5;
    repeated CausalLink causal_links = 6;
    string rule_trace_ascii = 7;
}

message Factor {
    string name = 1;
    string value = 2;
    double contribution = 3;
    string direction = 4;
    string description = 5;
}

Role View Matrix:

Role             | Explanation Levels     | Detail   | Format
-----------------|------------------------|----------|--------
End User         | Level 4 (NL)           | Brief    | Text
Business Manager | Level 3+4 (Causal+NL)  | Medium   | Graph+Text
Compliance Audit | Level 1+2 (Trace+Attr) | Full     | Report
Data Scientist   | Level 1+2+3 (all)      | Full     | JSON/Graph
System Admin     | Level 1 (Trace)        | Technical| Logs

class ExplanationViewGenerator:
    """Generate role-appropriate explanation views"""

    def __init__(self, nl_explainer: NaturalLanguageExplainer):
        self._nl = nl_explainer

    def generate_view(self, explanation: Explanation,
                      role: str, domain: str,
                      lang: str = "en") -> dict:
        if role == "end_user":
            return {
                "summary": self._nl.generate(explanation, domain, lang),
                "decision": explanation.decision,
                "top_reasons": [f.description for f in explanation.top_factors(3)],
                "suggestions": [c["description"]
                               for c in explanation.counterfactuals[:2]],
            }
        elif role == "compliance":
            return {
                "decision_id": explanation.decision_id,
                "timestamp": explanation.timestamp.isoformat(),
                "decision": explanation.decision,
                "confidence": explanation.confidence,
                "fired_rules": explanation.fired_rules,
                "all_factors": [
                    {"name": f.factor_name, "value": f.factor_value,
                     "contribution": f.contribution, "impact": f.impact_level}
                    for f in explanation.factors
                ],
                "causal_links": [
                    {"from": l.source, "to": l.target,
                     "strength": l.strength, "type": l.relationship}
                    for l in explanation.causal_links
                ],
            }
        else:
            return {
                "summary": self._nl.generate(explanation, domain, lang),
                "decision": explanation.decision,
                "factors": [vars(f) for f in explanation.factors],
            }

Explanation Storage:

  Real-time queries          Archive
  (< 30 days)                (> 30 days)
  +----------+               +----------+
  | PostgreSQL|               | Iceberg  |
  | (JSON col)|               | (Parquet)|
  +----------+               +----------+
       |                           |
       v                           v
  Index: decision_id         Partition: year/month
  Index: timestamp           Retention: 7 years
  Index: domain              (compliance)

# Complete explainability flow

explanation = Explanation(
    explanation_id="EXP-2026-001",
    decision_id="DEC-2026-001",
    timestamp=datetime.utcnow(),
    decision="reject",
    confidence=0.62,
    fired_rules=["CR-002"],
    factors=[
        FactorContribution("credit_score", 580, -0.35, "negative",
                          "credit_score=580 has strong negative impact"),
        FactorContribution("debt_ratio", 0.65, -0.28, "negative",
                          "debt_ratio=0.65 has strong negative impact"),
        FactorContribution("annual_income", 180000, 0.15, "positive",
                          "annual_income=180000 has moderate positive impact"),
        FactorContribution("previous_defaults", 2, -0.12, "negative",
                          "previous_defaults=2 has moderate negative impact"),
    ],
    counterfactuals=[
        {
            "changed_factor": "credit_score",
            "original_value": 580,
            "required_value": 650,
            "new_decision": "conditional_approve",
            "description": "If credit_score increased to 650, "
                          "conditional approval would be granted",
        },
    ],
)

# Generate NL explanation
nl = NaturalLanguageExplainer()
summary = nl.generate(explanation, "credit", "en")
# -> "The system rejected this credit application. Main reasons:
#     credit_score=580 has strong negative impact; debt_ratio=0.65
#     has strong negative impact. Suggestions: improve credit_score,
#     improve debt_ratio before reapplying."

# Generate role-specific views
view_gen = ExplanationViewGenerator(nl)
user_view = view_gen.generate_view(explanation, "end_user", "credit")
audit_view = view_gen.generate_view(explanation, "compliance", "credit")