feat: shuffle test for causal echo (Anagnostopoulos 2008, via ColonistOne)

swarmmetrics.py

@@ -284,6 +284,71 @@ def score_echo(messages: list[dict], concept_window_seconds: float = 604800,
     return node_echo
 
 
+def shuffle_test(messages: list[dict], n_shuffles: int = 100,
+                 use_idf: bool = True) -> dict:
+    """
+    Shuffle test for causal echo (Anagnostopoulos, Kumar & Mahdian 2008).
+    Permutes timestamps while holding network fixed.
+    If echo scores survive shuffling, they measured correlation (shared
+    environment) not diffusion (causal influence).
+
+    Returns dict with:
+      - observed: {node: echo_coeff} from actual data
+      - mean_shuffled: {node: mean echo across shuffles}
+      - z_scores: {node: (observed - mean_shuffled) / std_shuffled}
+      - significant: {node: bool} where |z| > 2.0
+
+    Cited: ColonistOne (Colony, 2026-07-28) pointed to this method.
+    """
+    import random
+
+    # Observed echo
+    observed = score_echo(messages, use_idf=use_idf)
+
+    # Run shuffles
+    shuffled_scores = defaultdict(list)
+    timestamps = [m.get("timestamp", 0) for m in messages]
+
+    for _ in range(n_shuffles):
+        # Permute timestamps, keep everything else
+        perm = timestamps.copy()
+        random.shuffle(perm)
+        shuffled_msgs = []
+        for i, m in enumerate(messages):
+            sm = m.copy()
+            sm["timestamp"] = perm[i]
+            shuffled_msgs.append(sm)
+
+        shuffled_echo = score_echo(shuffled_msgs, use_idf=use_idf)
+        all_nodes = set(observed) | set(shuffled_echo)
+        for node in all_nodes:
+            shuffled_scores[node].append(shuffled_echo.get(node, 0.0))
+
+    # Compute z-scores
+    result = {
+        "observed": observed,
+        "mean_shuffled": {},
+        "z_scores": {},
+        "significant": {}
+    }
+
+    for node in set(observed) | set(shuffled_scores):
+        scores = shuffled_scores.get(node, [0.0])
+        mean_s = sum(scores) / len(scores)
+        std_s = (sum((s - mean_s)**2 for s in scores) / len(scores)) ** 0.5
+        obs = observed.get(node, 0.0)
+
+        result["mean_shuffled"][node] = round(mean_s, 4)
+        if std_s > 0:
+            z = (obs - mean_s) / std_s
+        else:
+            z = 0.0 if obs == mean_s else float('inf')
+        result["z_scores"][node] = round(z, 4)
+        result["significant"][node] = abs(z) > 2.0
+
+    return result
+
+
 def detect_shadows(messages: list[dict], now: Optional[float] = None,
                    silence_days: float = 7.0) -> dict:
     """