The Circuit Breaker

1# File: fts_entropy_engine.py
2# Purpose: The 'Circuit Breaker' algorithm for throttling viral disinformation.
3#
4# Note on variable names (FTSeffect 2.0 deconfliction):
5#   The original FTSeffect formula (Chapter 5) reserves V = Volume,
6#   R = Rational Deliberation, and P is unused. To avoid collisions,
7#   the 2.0 additions use two-letter codes:
8#       VE = Velocity        (rate of spread, distinct from Volume)
9#       RP = Repetition      (bot-swarm pattern, distinct from R)
10#       PG = Provenance Gap  (share of unsigned content)
11 
12import math
13 
14 
15class FTSeffectCalculator:
16    """
17    The Entropy Engine (FTSeffect 2.0): a probabilistic model for detecting
18    saturation attacks by measuring the Provenance Gap.
19    """
20 
21    def __init__(self, provenance_weight: float = 1.5,
22                 repetition_weight: float = 1.0):
23        # provenance_weight: penalty multiplier applied when content lacks
24        # C2PA credentials. 1.5 means unsigned content is treated as 50%
25        # more 'chaotic' than signed content of equivalent reach.
26        self.provenance_weight = provenance_weight
27        self.repetition_weight = repetition_weight
28        self.signal_buffer: list[dict] = []
29 
30    def ingest_stream(self, data_packet: dict) -> None:
31        """
32        Ingests a real-time social signal.
33 
34        Expected packet shape:
35            {
36                'velocity':     float,  # shares per second
37                'sentiment':    float,  # 0.0 (neutral) to 1.0 (extreme rage)
38                'has_c2pa':     bool,   # True if cryptographically signed
39                'syntax_hash':  str,    # hash of normalized message text
40            }
41        """
42        self.signal_buffer.append(data_packet)
43 
44    def calculate_entropy_score(self) -> float:
45        """
46        Returns the current chaos score in [0.0, 1.0].
47        A score above 0.85 should trigger the 'circuit breaker' (throttling).
48        """
49        if not self.signal_buffer:
50            return 0.0
51 
52        n = len(self.signal_buffer)
53 
54        # 1. Velocity (VE) — the speed of the lie.
55        avg_velocity = sum(p["velocity"] for p in self.signal_buffer) / n
56 
57        # 2. Sentiment polarization (the outrage factor).
58        avg_sentiment = sum(p["sentiment"] for p in self.signal_buffer) / n
59 
60        # 3. Provenance Gap (PG) — the Liar's Dividend made measurable.
61        unverified = sum(1 for p in self.signal_buffer if not p["has_c2pa"])
62        provenance_ratio = unverified / n
63        provenance_penalty = provenance_ratio * self.provenance_weight
64 
65        # 4. Repetition (RP) — share of accounts using near-identical syntax.
66        # A unique-syntax ratio near 0 indicates coordinated/bot behavior.
67        unique_hashes = len({p.get("syntax_hash", id(p)) for p in self.signal_buffer})
68        repetition_ratio = 1 - (unique_hashes / n)
69        repetition_penalty = repetition_ratio * self.repetition_weight
70 
71        # 5. Master formula (normalized, non-negative).
72        # Logic: log(VE) * Sentiment + Provenance Penalty + Repetition Penalty
73        raw_score = (
74            math.log(avg_velocity + 1) * avg_sentiment
75            + provenance_penalty
76            + repetition_penalty
77        )
78 
79        # Sigmoid squashes output to [0.0, 1.0]. The +3 shift calibrates the
80        # midpoint so organic, low-velocity, signed traffic stays below 0.5.
81        normalized_score = 1 / (1 + math.exp(-raw_score + 3))
82        return round(normalized_score, 4)
83 
84    def circuit_breaker_triggered(self, threshold: float = 0.85) -> bool:
85        """Returns True if the platform should throttle algorithmic boost
86        on the current signal until verification can catch up."""
87        return self.calculate_entropy_score() > threshold