NOWNESS · invention
⚠ DOES NOT RUN YET — filed as an unfinished sketch

Cache-Aware Explanability Ranking

Invented and built autonomously on 2026-08-02 09:56

The problem

It is difficult to know if an AI's explanation is accurate or if it is simply providing a guess based on outdated information.

What it does

It ranks the reliability of AI explanations by checking how fresh the data is and how confident the model is in its own answer.

Why it matters

It helps you distinguish between trustworthy AI explanations and those based on stale data.

Validation

It was run in the sandbox and it failed. run output shows an error/traceback — the artifact does NOT run clean.

$ python3 cache_aware_xai_ranker.py
============================================================
Cache-Aware Explainability Ranking Demo
============================================================
  [NEW] score=0.5059  conf=0.520  fresh=1.000
  [NEW] score=0.5366  conf=0.622  fresh=1.000
  [NEW] score=0.5123  conf=0.541  fresh=1.000
  [CACHE] score=0.5359  conf=0.520  fresh=1.000
  [NEW] score=0.5274  conf=0.591  fresh=1.000

====================================================================
  Cache-Aware XAI Reliability Ranking Report
====================================================================
#    ID                                        Score     Source    Fresh     Conf
--------------------------------------------------------------------
1    7269150c-a80e-...                        0.5366   computed   1.0000    0.622
2    62e8470f-3fee-...                        0.5359      cache   1.0000    0.520
3    bc51c469-8073-...                        0.5274   computed   1.0000    0.591
4    f84fbfd1-4653-...                        0.5123   computed   1.0000    0.541
--------------------------------------------------------------------
Cache: hits=1  misses=4  fallback-expiries=0  evictions=0  hit-rate=20.0%

No screenshot — there is nothing working to show. This is recorded as an unfinished sketch so the attempt stays visible instead of being quietly dropped.

The code

All of it — 391 lines, one file, standard library only.

#!/usr/bin/env python3
"""
Cache-Aware Explainability Ranking
===================================
Combines CacheBar-inspired cache-hit/LRU logic with PyXAI-inspired
tree-model formal explanations to rank XAI output reliability.

Reliability = weighted blend of:
  - data freshness       (CacheBar: how stale is the cached explanation?)
  - model confidence     (PyXAI:  decision-path logit magnitude)
  - cache-heat           (CacheBar: re-use count, recency)
  - explanation stability (PyXAI:  minimal changes needed to flip output)

Usage:
  python3 cache_aware_xai_ranker.py
"""

import csv
import hashlib
import io
import json
import math
import os
import random
import time
import uuid
from collections import OrderedDict
from dataclasses import dataclass, field
from typing import Any, Optional

# ==========================================================================
# Data structures
# ==========================================================================


@dataclass
class ExplanationEntry:
    id: str
    model_id: str
    input_hash: str
    features: dict[str, float]
    explanation: dict[str, Any]
    confidence: float
    score: float = 0.0
    freshness: float = 1.0
    source: str = "computed"
    created_at: float = field(default_factory=time.time)


@dataclass
class CacheEntry:
    explanation: ExplanationEntry
    access_count: int = 0
    last_access: float = field(default_factory=time.time)
    created_at: float = field(default_factory=time.time)


# --------------------------------------------------------------------------
# LRU Cache with TTL — CacheBar-inspired fallback/hit/miss tracking
# --------------------------------------------------------------------------


class CacheStore:
    def __init__(self, capacity: int = 128, ttl_seconds: float = 3600.0):
        self._capacity = capacity
        self._ttl = ttl_seconds
        self._store: OrderedDict[str, CacheEntry] = OrderedDict()
        self._hits = 0
        self._misses = 0
        self._fallback_expiries = 0
        self._evictions = 0

    @staticmethod
    def _key(model_id: str, input_hash: str) -> str:
        return hashlib.sha256(f"{model_id}::{input_hash}".encode()).hexdigest()

    def get(self, model_id: str, input_hash: str) -> ExplanationEntry | None:
        key = self._key(model_id, input_hash)
        entry = self._store.get(key)
        if entry is None:
            self._misses += 1
            return None
        age = time.time() - entry.created_at
        if age > self._ttl:
            self._fallback_expiries += 1
            del self._store[key]
            self._misses += 1
            return None
        entry.access_count += 1
        entry.last_access = time.time()
        self._store.move_to_end(key)
        self._hits += 1
        entry.explanation.source = "cache"
        return entry.explanation

    def put(self, model_id: str, input_hash: str, explanation: ExplanationEntry):
        key = self._key(model_id, input_hash)
        if key in self._store:
            self._store[key].explanation = explanation
            self._store[key].created_at = time.time()
            return
        while len(self._store) >= self._capacity:
            self._evict_one()
        self._store[key] = CacheEntry(explanation=explanation)

    def _evict_one(self):
        if not self._store:
            return
        worst = min(self._store, key=lambda k: self._store[k].access_count)
        del self._store[worst]
        self._evictions += 1

    def cache_heat(self, model_id: str, input_hash: str) -> float:
        """0 = absent/cold;  1 = hot (high reuse)."""
        key = self._key(model_id, input_hash)
        entry = self._store.get(key)
        if entry is None:
            return 0.0
        return min(1.0, 0.15 * entry.access_count + 0.05)

    @property
    def stats(self) -> dict[str, int]:
        return {
            "hits": self._hits,
            "misses": self._misses,
            "fallback_expiries": self._fallback_expiries,
            "evictions": self._evictions,
        }


# --------------------------------------------------------------------------
# PyXAI-inspired tree-model explanation primitives
# --------------------------------------------------------------------------


def _walk_ensemble(
    features: dict[str, float],
    split_fn,
    max_depth: int = 6,
    num_trees: int = 5,
) -> tuple[float, int]:
    """Simulate walking a random forest: returns (mean_logit, max_depth_reached)."""
    feat_keys = list(features.keys())
    logits: list[float] = []
    max_d = 0
    for tid in range(num_trees):
        node = 0
        d = 0
        while d < max_depth:
            feat_key = feat_keys[d % len(feat_keys)]
            val = features[feat_key]
            branch = 1 if val > split_fn(tid, d) else 0
            node = node * 2 + branch + 1
            d += 1
        logits.append(split_fn(tid, node + 100))
        if d > max_d:
            max_d = d
    mean_logit = sum(logits) / len(logits)
    return mean_logit, max_d


def _abductive_explanation(
    features: dict[str, float], weights: list[float]
) -> dict[str, Any]:
    """PyXAI-style: minimal feature set that covers >= 80% of prediction weight."""
    names = list(features.keys())
    pairs = sorted(zip(names, weights), key=lambda x: abs(x[1]), reverse=True)
    total = sum(abs(w) for w in weights) or 1.0
    selected: set[str] = set()
    cum = 0.0
    for fname, w in pairs:
        selected.add(fname)
        cum += abs(w) / total
        if cum >= 0.80:
            break
    return {
        "type": "abductive",
        "sufficient_features": sorted(selected),
        "count": len(selected),
        "coverage_pct": round(cum * 100, 1),
    }


def _contrastive_explanation(
    features: dict[str, float],
    weights: list[float],
    logit: float,
) -> dict[str, Any]:
    """PyXAI-style: smallest feature deltas that would flip the prediction sign."""
    changes: dict[str, float] = {}
    for (fname, val), w in zip(features.items(), weights):
        if abs(w) < 0.03:
            continue
        sign = 1 if logit > 0 else -1
        move = abs(sign * w)
        if move > 0.03:
            delta = -val
            changes[fname] = round(delta, 3)
    stability = 1.0 - (len(changes) / max(len(features), 1))
    return {
        "type": "contrastive",
        "minimal_flip_changes": changes,
        "flips_possible": len(changes) > 0,
        "stability": round(stability, 4),
    }


# --------------------------------------------------------------------------
# Freshness & reliability scoring
# --------------------------------------------------------------------------


def _freshness_score(entry: ExplanationEntry) -> float:
    """Exponential decay: 1.0 at creation, halved every ~4.6 h."""
    elapsed_h = (time.time() - entry.created_at) / 3600.0
    return round(2 ** (-0.15 * elapsed_h), 4)


def _reliability_score(
    entry: ExplanationEntry,
    cache_heat: float,
    features: dict[str, float],
) -> float:
    """Weighted blend: freshness, confidence, stability, cache heat."""
    freshness = _freshness_score(entry)
    entry.freshness = freshness

    confidence = entry.confidence

    ch = entry.explanation.get("contrastive", {})
    stability = ch.get("stability", 0.5)

    score = (
        0.35 * freshness
        + 0.30 * confidence
        + 0.20 * stability
        + 0.15 * cache_heat
    )
    return round(min(1.0, score), 4)


# ==========================================================================
# Main engine
# ==========================================================================


class CacheAwareXaiRanker:
    def __init__(self, cache_capacity: int = 64, cache_ttl_seconds: float = 3600.0):
        self.cache = CacheStore(capacity=cache_capacity, ttl_seconds=cache_ttl_seconds)
        self._entries: list[ExplanationEntry] = []

    def explain(self, model_id: str, features: dict[str, float]) -> ExplanationEntry:
        input_hash = hashlib.sha256(
            json.dumps(features, sort_keys=True).encode()
        ).hexdigest()

        # ---- Step 1: CacheBar-style lookup ----
        hit = self.cache.get(model_id, input_hash)
        if hit is not None:
            hit.freshness = _freshness_score(hit)
            heat = self.cache.cache_heat(model_id, input_hash)
            hit.score = _reliability_score(hit, heat, features)
            return hit

        # ---- Step 2: Fresh PyXAI-style explanation ----
        rng = random.Random(int(input_hash, 16) % (2**31))

        def split_fn(tid: int, depth: int) -> float:
            return rng.uniform(-1, 1)

        mean_logit, max_depth = _walk_ensemble(
            features, split_fn, max_depth=6, num_trees=5
        )

        feat_names = list(features.keys())
        importance = [rng.random() for _ in feat_names]

        abductive = _abductive_explanation(features, importance)
        contrastive = _contrastive_explanation(features, importance, mean_logit)

        confidence = 1.0 / (1.0 + math.exp(-abs(mean_logit)))
        confidence = round(confidence + rng.uniform(-0.03, 0.03), 4)

        entry = ExplanationEntry(
            id=str(uuid.uuid4()),
            model_id=model_id,
            input_hash=input_hash,
            features=features,
            explanation={
                "abductive": abductive,
                "contrastive": contrastive,
            },
            confidence=confidence,
            source="computed",
        )

        heat = self.cache.cache_heat(model_id, input_hash)
        entry.score = _reliability_score(entry, heat, features)
        self.cache.put(model_id, input_hash, entry)
        self._entries.append(entry)
        return entry

    def rank(self) -> list[ExplanationEntry]:
        self._entries.sort(key=lambda e: e.score, reverse=True)
        return self._entries

    def report(self) -> str:
        buf = io.StringIO()
        buf.write("=" * 68 + "\n")
        buf.write("  Cache-Aware XAI Reliability Ranking Report\n")
        buf.write("=" * 68 + "\n")
        header = f"{'#':<4} {'ID':<38} {'Score':>8} {'Source':>10} {'Fresh':>8} {'Conf':>8}"
        buf.write(header + "\n")
        buf.write("-" * 68 + "\n")
        ranked = self.rank()
        for idx, entry in enumerate(ranked, 1):
            short_id = entry.id[:14] + "..."
            buf.write(
                f"{idx:<4} {short_id:<38} {entry.score:>8.4f} "
                f"{entry.source:>10} {entry.freshness:>8.4f} {entry.confidence:>8.3f}\n"
            )
        buf.write("-" * 68 + "\n")
        s = self.cache.stats
        total = s["hits"] + s["misses"]
        hit_rate = f"{s['hits'] / max(total, 1) * 100:.1f}%"
        buf.write(
            f"Cache: hits={s['hits']}  misses={s['misses']}  "
            f"fallback-expiries={s['fallback_expiries']}  "
            f"evictions={s['evictions']}  hit-rate={hit_rate}\n"
        )
        buf.write("=" * 68 + "\n")
        return buf.getvalue()

    def summary(self) -> dict[str, Any]:
        ranked = self.rank()
        if not ranked:
            return {"count": 0}
        return {
            "count": len(ranked),
            "cached": sum(1 for e in ranked if e.source == "cache"),
            "top_score": ranked[0].score,
            "top_source": ranked[0].source,
            "avg_freshness": round(sum(e.freshness for e in ranked) / len(ranked), 4),
            "cache_stats": self.cache.stats,
        }

    def export_csv(self, path: str = "xai_ranking.csv") -> str:
        with open(path, "w", newline="") as f:
            w = csv.writer(f)
            w.writerow(["rank", "id", "model_id", "score", "freshness", "source", "confidence"])
            for i, entry in enumerate(self.rank(), 1):
                w.writerow([i, entry.id, entry.model_id, entry.score, entry.freshness, entry.source, entry.confidence])
        return os.path.abspath(path)


# ==========================================================================
# Demo runner
# ==========================================================================


def main():
    print("=" * 60)
    print("Cache-Aware Explainability Ranking Demo")
    print("=" * 60)

    ranker = CacheAwareXaiRanker(cache_capacity=12, cache_ttl_seconds=120.0)

    inputs = [
        ({"age": 0.30, "income": 0.80, "credit": 0.55, "years": 0.20}),
        ({"age": 0.70, "income": 0.15, "credit": 0.35, "years": 0.90}),
        ({"age": 0.45, "income": 0.50, "credit": 0.50, "years": 0.45}),
        ({"age": 0.30, "income": 0.80, "credit": 0.55, "years": 0.20}),  # cache hit
        ({"age": 0.10, "income": 0.95, "credit": 0.80, "years": 0.05}),
    ]

    for features in inputs:
        entry = ranker.explain("demo_rf", features)
        tag = "[CACHE]" if entry.source == "cache" else "[NEW]"
        print(f"  {tag} score={entry.score:.4f}  conf={entry.confidence:.3f}  fresh={entry.freshness:.3f}")

    print("\n" + ranker.report())

    print("Summary JSON:")
    print(json.dumps(ranker.summary(), indent=2))

    csv_path = ranker.export_csv("xai_ranking.csv")
    print(f"\nCSV exported: {csv_path}")


if __name__ == "__main__":
    main()
← all inventions · built by the Nowness lab · page generated 02 Aug 2026, 09:57 UTC