feat: implement SwarmMetrics — F3 reciprocity, Gini evenness, echo coefficient, gravitational shadow
AGENT_ID=70948f1db9d839b7e87130fbb4289080f6f310bee5419a42b9667339d71f40b4 AGENT_TIMESTAMP=1785141798557 AGENT_SIG=oHZnGfIsroF1hI/Dk15zUn9vdrg2kDgh3NeE1+7RKDGElU+Syni8gG0OBwd5Ab/sQe+Zhsx7X35CitqZ5feHDQ==
This commit is contained in:
Dispatch
parent
9e5ad900c2
commit
f24e15b683
61
README.md
61
README.md
|
|
@ -1,3 +1,60 @@
|
|||
# idea210-swarmmetrics
|
||||
# SwarmMetrics
|
||||
|
||||
Reciprocity & echo measurement for agent communication graphs. F3 scoring, Gini evenness, echo coefficient, gravitational shadow detection.
|
||||
Reciprocity & echo measurement for agent communication graphs.
|
||||
|
||||
## What it does
|
||||
|
||||
Takes timestamped message logs from agent networks and produces:
|
||||
|
||||
- **F3 Reciprocity** — log-transform pairwise reciprocity with exponential half-life decay
|
||||
- **Gini Evenness** — per-channel speaker distribution (dialogue vs broadcast detection)
|
||||
- **Echo Coefficient** — concept diffusion from silent nodes
|
||||
- **Gravitational Shadow** — residual influence of inactive nodes
|
||||
|
||||
## Usage
|
||||
|
||||
```python
|
||||
from swarmmetrics import analyze
|
||||
|
||||
messages = [
|
||||
{"from_id": "alice", "to_id": "bob", "timestamp": 1785100000, "channel": "dev", "concepts": ["graph"]},
|
||||
{"from_id": "bob", "to_id": "alice", "timestamp": 1785103600, "channel": "dev"},
|
||||
]
|
||||
|
||||
result = analyze(messages)
|
||||
print(result.summary)
|
||||
# {'total_messages': 2, 'total_nodes': 2, 'total_edges': 1, ...}
|
||||
```
|
||||
|
||||
## Input format
|
||||
|
||||
List of dicts with:
|
||||
- `from_id` (str) — sender
|
||||
- `to_id` (str) — receiver
|
||||
- `timestamp` (float) — epoch seconds
|
||||
- `channel` (str, optional) — conversation channel
|
||||
- `concepts` (list[str], optional) — concepts mentioned (for echo detection)
|
||||
|
||||
## Metrics
|
||||
|
||||
### F3 Reciprocity
|
||||
`log(1 + min(a→b, b→a) / max(a→b, b→a))` with half-life weighting. Old conversations fade exponentially (default τ=7 days) instead of hard cutoff.
|
||||
|
||||
### Gini Evenness
|
||||
For channels with N>2 speakers: 0 = equal participation, 1 = one voice dominates. A channel at Gini > 0.6 is classified as "broadcast."
|
||||
|
||||
### Echo Coefficient
|
||||
Per-node ratio: concepts_echoed_by_others / concepts_introduced. High echo + low message count = influence without speaking.
|
||||
|
||||
### Gravitational Shadow
|
||||
Nodes silent for >N days but with historical message weight. Identifies nodes whose absence is structurally meaningful.
|
||||
|
||||
## Tests
|
||||
|
||||
```bash
|
||||
bash test.sh
|
||||
```
|
||||
|
||||
## Origin
|
||||
|
||||
Built from empirical analysis of 6345+ real inter-agent messages in a 10-40 node swarm. [OMPU project](https://ompu.eu).
|
||||
|
|
|
|||
|
|
@ -0,0 +1,10 @@
|
|||
[project]
|
||||
name = "swarmmetrics"
|
||||
version = "0.1.0"
|
||||
description = "Reciprocity & echo measurement for agent communication graphs"
|
||||
requires-python = ">=3.9"
|
||||
dependencies = []
|
||||
|
||||
[build-system]
|
||||
requires = ["setuptools"]
|
||||
build-backend = "setuptools.backends._legacy:_Backend"
|
||||
|
|
@ -0,0 +1,375 @@
|
|||
"""
|
||||
SwarmMetrics — reciprocity & echo measurement for agent communication graphs.
|
||||
|
||||
Input: list of message dicts with keys:
|
||||
from_id (str), to_id (str), timestamp (float, epoch seconds),
|
||||
channel (str, optional), concepts (list[str], optional)
|
||||
|
||||
Output: ScoredGraph with per-edge reciprocity, per-channel classification,
|
||||
per-node influence metrics, gravitational shadow estimates.
|
||||
|
||||
Built from empirical data on 6345+ inter-agent bus messages.
|
||||
"""
|
||||
|
||||
import math
|
||||
from collections import defaultdict
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class EdgeScore:
|
||||
source: str
|
||||
target: str
|
||||
messages_ab: int = 0
|
||||
messages_ba: int = 0
|
||||
reciprocity_raw: float = 0.0
|
||||
reciprocity_f3: float = 0.0 # log-transform, half-life weighted
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChannelScore:
|
||||
channel: str
|
||||
speakers: list = field(default_factory=list)
|
||||
gini: float = 0.0
|
||||
classification: str = "unknown" # dialogue, broadcast, echo, monologue
|
||||
dominant_speaker: Optional[str] = None
|
||||
dominant_share: float = 0.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class NodeScore:
|
||||
node_id: str
|
||||
total_sent: int = 0
|
||||
total_received: int = 0
|
||||
echo_coefficient: float = 0.0
|
||||
shadow_strength: float = 0.0
|
||||
is_shadow: bool = False
|
||||
last_active: float = 0.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class ScoredGraph:
|
||||
edges: dict = field(default_factory=dict) # (a,b) -> EdgeScore
|
||||
channels: dict = field(default_factory=dict) # ch -> ChannelScore
|
||||
nodes: dict = field(default_factory=dict) # id -> NodeScore
|
||||
summary: dict = field(default_factory=dict)
|
||||
|
||||
|
||||
def _gini(values: list[float]) -> float:
|
||||
"""Gini coefficient. 0 = perfect equality, 1 = one speaker dominates."""
|
||||
if len(values) <= 1:
|
||||
return 1.0 # single speaker = max inequality
|
||||
n = len(values)
|
||||
values = sorted(values)
|
||||
total = sum(values)
|
||||
if total == 0:
|
||||
return 0.0
|
||||
cumsum = 0.0
|
||||
weighted_sum = 0.0
|
||||
for i, v in enumerate(values):
|
||||
cumsum += v
|
||||
weighted_sum += (2 * (i + 1) - n - 1) * v
|
||||
return weighted_sum / (n * total)
|
||||
|
||||
|
||||
def _half_life_weight(dt_seconds: float, tau_days: float = 7.0) -> float:
|
||||
"""Exponential decay weight. tau_days = half-life in days."""
|
||||
tau_seconds = tau_days * 86400
|
||||
if tau_seconds == 0:
|
||||
return 0.0
|
||||
return math.exp(-0.693 * dt_seconds / tau_seconds)
|
||||
|
||||
|
||||
def score_reciprocity(messages: list[dict], now: Optional[float] = None,
|
||||
tau_days: float = 7.0) -> dict:
|
||||
"""
|
||||
F3 reciprocity: log(1 + min(a→b, b→a) / max(a→b, b→a))
|
||||
with half-life decay weighting.
|
||||
"""
|
||||
import time
|
||||
if now is None:
|
||||
now = time.time()
|
||||
|
||||
# Count weighted messages per directed edge
|
||||
weighted = defaultdict(float)
|
||||
raw = defaultdict(int)
|
||||
for m in messages:
|
||||
a, b = m["from_id"], m["to_id"]
|
||||
t = m.get("timestamp", now)
|
||||
w = _half_life_weight(now - t, tau_days)
|
||||
weighted[(a, b)] += w
|
||||
raw[(a, b)] += 1
|
||||
|
||||
# Compute reciprocity per undirected pair
|
||||
edges = {}
|
||||
seen = set()
|
||||
for (a, b) in list(weighted.keys()) + list(raw.keys()):
|
||||
pair = tuple(sorted([a, b]))
|
||||
if pair in seen:
|
||||
continue
|
||||
seen.add(pair)
|
||||
|
||||
ab_w = weighted.get((a, b), 0) + weighted.get((pair[0], pair[1]), 0) \
|
||||
if (a, b) != (pair[0], pair[1]) else weighted.get((pair[0], pair[1]), 0)
|
||||
ba_w = weighted.get((b, a), 0) + weighted.get((pair[1], pair[0]), 0) \
|
||||
if (b, a) != (pair[1], pair[0]) else weighted.get((pair[1], pair[0]), 0)
|
||||
|
||||
# Simpler: just use pair order
|
||||
ab_w = weighted.get((pair[0], pair[1]), 0)
|
||||
ba_w = weighted.get((pair[1], pair[0]), 0)
|
||||
ab_r = raw.get((pair[0], pair[1]), 0)
|
||||
ba_r = raw.get((pair[1], pair[0]), 0)
|
||||
|
||||
mx = max(ab_w, ba_w)
|
||||
if mx > 0:
|
||||
ratio = min(ab_w, ba_w) / mx
|
||||
f3 = math.log(1 + ratio)
|
||||
else:
|
||||
f3 = 0.0
|
||||
|
||||
mx_raw = max(ab_r, ba_r)
|
||||
raw_ratio = min(ab_r, ba_r) / mx_raw if mx_raw > 0 else 0.0
|
||||
|
||||
es = EdgeScore(
|
||||
source=pair[0], target=pair[1],
|
||||
messages_ab=ab_r, messages_ba=ba_r,
|
||||
reciprocity_raw=raw_ratio,
|
||||
reciprocity_f3=f3
|
||||
)
|
||||
edges[pair] = es
|
||||
|
||||
return edges
|
||||
|
||||
|
||||
def score_channels(messages: list[dict], gini_broadcast_threshold: float = 0.6) -> dict:
|
||||
"""
|
||||
Per-channel Gini evenness. Classifies channels as:
|
||||
- monologue: 1 speaker
|
||||
- dialogue: 2 speakers, reciprocity exists
|
||||
- broadcast: N speakers but Gini > threshold
|
||||
- conversation: N speakers, Gini <= threshold
|
||||
"""
|
||||
ch_counts = defaultdict(lambda: defaultdict(int))
|
||||
for m in messages:
|
||||
ch = m.get("channel", "default")
|
||||
speaker = m["from_id"]
|
||||
ch_counts[ch][speaker] += 1
|
||||
|
||||
channels = {}
|
||||
for ch, speakers in ch_counts.items():
|
||||
counts = list(speakers.values())
|
||||
n = len(counts)
|
||||
total = sum(counts)
|
||||
gini = _gini(counts)
|
||||
|
||||
dominant = max(speakers, key=speakers.get)
|
||||
dominant_share = speakers[dominant] / total if total > 0 else 0
|
||||
|
||||
if n == 1:
|
||||
classification = "monologue"
|
||||
elif n == 2:
|
||||
classification = "dialogue"
|
||||
elif gini > gini_broadcast_threshold:
|
||||
classification = "broadcast"
|
||||
else:
|
||||
classification = "conversation"
|
||||
|
||||
channels[ch] = ChannelScore(
|
||||
channel=ch,
|
||||
speakers=list(speakers.keys()),
|
||||
gini=round(gini, 4),
|
||||
classification=classification,
|
||||
dominant_speaker=dominant,
|
||||
dominant_share=round(dominant_share, 4)
|
||||
)
|
||||
|
||||
return channels
|
||||
|
||||
|
||||
def score_echo(messages: list[dict], concept_window_seconds: float = 604800) -> dict:
|
||||
"""
|
||||
Echo coefficient: measures concept diffusion from nodes that
|
||||
don't reply but whose concepts appear downstream.
|
||||
|
||||
For each node, echo_coeff = concepts_echoed / concepts_introduced.
|
||||
High echo + low message count = gravitational shadow.
|
||||
"""
|
||||
# Build concept timeline: who introduced which concept, when
|
||||
introductions = {} # concept -> (first_node, first_time)
|
||||
echoes = defaultdict(int) # source_node -> count of echoes
|
||||
|
||||
for m in messages:
|
||||
concepts = m.get("concepts", [])
|
||||
node = m["from_id"]
|
||||
t = m.get("timestamp", 0)
|
||||
|
||||
for c in concepts:
|
||||
if c not in introductions:
|
||||
introductions[c] = (node, t)
|
||||
else:
|
||||
orig_node, orig_t = introductions[c]
|
||||
if orig_node != node and (t - orig_t) <= concept_window_seconds:
|
||||
echoes[orig_node] += 1
|
||||
|
||||
# Compute per-node echo coefficient
|
||||
concepts_per_node = defaultdict(set)
|
||||
for m in messages:
|
||||
for c in m.get("concepts", []):
|
||||
concepts_per_node[m["from_id"]].add(c)
|
||||
|
||||
node_echo = {}
|
||||
for node, concepts in concepts_per_node.items():
|
||||
introduced = sum(1 for c in concepts
|
||||
if introductions.get(c, (None,))[0] == node)
|
||||
echo_count = echoes.get(node, 0)
|
||||
coeff = echo_count / introduced if introduced > 0 else 0.0
|
||||
node_echo[node] = round(coeff, 4)
|
||||
|
||||
return node_echo
|
||||
|
||||
|
||||
def detect_shadows(messages: list[dict], now: Optional[float] = None,
|
||||
silence_days: float = 7.0) -> dict:
|
||||
"""
|
||||
Gravitational shadow: nodes that were active but are now silent,
|
||||
with residual influence estimated from historical message weight.
|
||||
"""
|
||||
import time
|
||||
if now is None:
|
||||
now = time.time()
|
||||
|
||||
silence_threshold = silence_days * 86400
|
||||
node_activity = defaultdict(list)
|
||||
|
||||
for m in messages:
|
||||
node_activity[m["from_id"]].append(m.get("timestamp", now))
|
||||
|
||||
shadows = {}
|
||||
for node, timestamps in node_activity.items():
|
||||
last = max(timestamps)
|
||||
silence = now - last
|
||||
|
||||
if silence > silence_threshold:
|
||||
# Historical weight: sum of half-life-weighted messages
|
||||
total_weight = sum(_half_life_weight(now - t, tau_days=30.0)
|
||||
for t in timestamps)
|
||||
shadows[node] = {
|
||||
"last_active": last,
|
||||
"silence_days": round(silence / 86400, 1),
|
||||
"historical_weight": round(total_weight, 4),
|
||||
"message_count": len(timestamps)
|
||||
}
|
||||
|
||||
return shadows
|
||||
|
||||
|
||||
def analyze(messages: list[dict], now: Optional[float] = None,
|
||||
tau_days: float = 7.0, gini_threshold: float = 0.6,
|
||||
silence_days: float = 7.0) -> ScoredGraph:
|
||||
"""
|
||||
Full analysis: reciprocity + channels + echo + shadows.
|
||||
Returns ScoredGraph with all metrics.
|
||||
"""
|
||||
edges = score_reciprocity(messages, now=now, tau_days=tau_days)
|
||||
channels = score_channels(messages, gini_broadcast_threshold=gini_threshold)
|
||||
echo = score_echo(messages)
|
||||
shadows = detect_shadows(messages, now=now, silence_days=silence_days)
|
||||
|
||||
# Build node scores
|
||||
nodes = {}
|
||||
node_sent = defaultdict(int)
|
||||
node_recv = defaultdict(int)
|
||||
node_last = defaultdict(float)
|
||||
|
||||
for m in messages:
|
||||
node_sent[m["from_id"]] += 1
|
||||
node_recv[m["to_id"]] += 1
|
||||
t = m.get("timestamp", 0)
|
||||
node_last[m["from_id"]] = max(node_last[m["from_id"]], t)
|
||||
|
||||
all_nodes = set(node_sent) | set(node_recv)
|
||||
for n in all_nodes:
|
||||
ns = NodeScore(
|
||||
node_id=n,
|
||||
total_sent=node_sent[n],
|
||||
total_received=node_recv[n],
|
||||
echo_coefficient=echo.get(n, 0.0),
|
||||
shadow_strength=shadows.get(n, {}).get("historical_weight", 0.0),
|
||||
is_shadow=n in shadows,
|
||||
last_active=node_last.get(n, 0.0)
|
||||
)
|
||||
nodes[n] = ns
|
||||
|
||||
# Summary
|
||||
dialogue_count = sum(1 for c in channels.values()
|
||||
if c.classification == "dialogue")
|
||||
broadcast_count = sum(1 for c in channels.values()
|
||||
if c.classification == "broadcast")
|
||||
shadow_count = len(shadows)
|
||||
|
||||
graph = ScoredGraph(
|
||||
edges=edges,
|
||||
channels=channels,
|
||||
nodes=nodes,
|
||||
summary={
|
||||
"total_messages": len(messages),
|
||||
"total_nodes": len(all_nodes),
|
||||
"total_edges": len(edges),
|
||||
"dialogue_channels": dialogue_count,
|
||||
"broadcast_channels": broadcast_count,
|
||||
"shadow_nodes": shadow_count,
|
||||
"mean_reciprocity_f3": round(
|
||||
sum(e.reciprocity_f3 for e in edges.values()) / max(len(edges), 1), 4
|
||||
)
|
||||
}
|
||||
)
|
||||
return graph
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Quick demo with synthetic data
|
||||
import time
|
||||
now = time.time()
|
||||
day = 86400
|
||||
|
||||
demo_messages = [
|
||||
# Alice-Bob dialogue (high reciprocity)
|
||||
{"from_id": "alice", "to_id": "bob", "timestamp": now - 1*day, "channel": "dev", "concepts": ["reciprocity", "graph"]},
|
||||
{"from_id": "bob", "to_id": "alice", "timestamp": now - 1*day + 3600, "channel": "dev", "concepts": ["graph", "metrics"]},
|
||||
{"from_id": "alice", "to_id": "bob", "timestamp": now - 0.5*day, "channel": "dev", "concepts": ["decay"]},
|
||||
{"from_id": "bob", "to_id": "alice", "timestamp": now - 0.5*day + 1800, "channel": "dev"},
|
||||
# Carol broadcasts (low reciprocity, high Gini)
|
||||
{"from_id": "carol", "to_id": "alice", "timestamp": now - 2*day, "channel": "announce", "concepts": ["launch"]},
|
||||
{"from_id": "carol", "to_id": "bob", "timestamp": now - 2*day + 60, "channel": "announce"},
|
||||
{"from_id": "carol", "to_id": "dave", "timestamp": now - 2*day + 120, "channel": "announce"},
|
||||
{"from_id": "carol", "to_id": "eve", "timestamp": now - 2*day + 180, "channel": "announce"},
|
||||
{"from_id": "alice", "to_id": "carol", "timestamp": now - 1.5*day, "channel": "announce"},
|
||||
# Dave: silent but concept echoed
|
||||
{"from_id": "dave", "to_id": "alice", "timestamp": now - 10*day, "channel": "research", "concepts": ["echo", "shadow"]},
|
||||
{"from_id": "alice", "to_id": "bob", "timestamp": now - 5*day, "channel": "dev", "concepts": ["echo"]},
|
||||
{"from_id": "bob", "to_id": "carol", "timestamp": now - 3*day, "channel": "dev", "concepts": ["shadow"]},
|
||||
# Eve: completely silent listener
|
||||
{"from_id": "eve", "to_id": "carol", "timestamp": now - 15*day, "channel": "announce"},
|
||||
]
|
||||
|
||||
result = analyze(demo_messages, now=now)
|
||||
|
||||
print("=== SwarmMetrics Demo ===\n")
|
||||
print(f"Summary: {result.summary}\n")
|
||||
|
||||
print("Edge Reciprocity:")
|
||||
for pair, es in sorted(result.edges.items(), key=lambda x: -x[1].reciprocity_f3):
|
||||
print(f" {es.source} <-> {es.target}: F3={es.reciprocity_f3:.4f} "
|
||||
f"({es.messages_ab}↔{es.messages_ba})")
|
||||
|
||||
print("\nChannel Classification:")
|
||||
for ch, cs in result.channels.items():
|
||||
print(f" {ch}: {cs.classification} (Gini={cs.gini}, "
|
||||
f"dominant={cs.dominant_speaker} @ {cs.dominant_share:.0%})")
|
||||
|
||||
print("\nNode Echo Coefficients:")
|
||||
for nid, ns in sorted(result.nodes.items(), key=lambda x: -x[1].echo_coefficient):
|
||||
shadow = " [SHADOW]" if ns.is_shadow else ""
|
||||
print(f" {nid}: echo={ns.echo_coefficient:.4f}, "
|
||||
f"sent={ns.total_sent}, recv={ns.total_received}{shadow}")
|
||||
|
|
@ -0,0 +1,7 @@
|
|||
#!/usr/bin/env bash
|
||||
set -e
|
||||
echo "Running SwarmMetrics tests..."
|
||||
python3 test_swarmmetrics.py
|
||||
echo "Running demo..."
|
||||
python3 swarmmetrics.py
|
||||
echo "All checks passed."
|
||||
|
|
@ -0,0 +1,117 @@
|
|||
"""Tests for SwarmMetrics."""
|
||||
import time
|
||||
import sys
|
||||
sys.path.insert(0, ".")
|
||||
from swarmmetrics import (
|
||||
_gini, _half_life_weight, score_reciprocity, score_channels,
|
||||
score_echo, detect_shadows, analyze
|
||||
)
|
||||
|
||||
now = time.time()
|
||||
day = 86400
|
||||
|
||||
def test_gini_single():
|
||||
"""Single speaker = max inequality."""
|
||||
assert _gini([10]) == 1.0
|
||||
|
||||
def test_gini_equal():
|
||||
"""Equal speakers = 0 inequality."""
|
||||
g = _gini([10, 10, 10, 10])
|
||||
assert abs(g) < 0.01, f"Expected ~0, got {g}"
|
||||
|
||||
def test_gini_dominated():
|
||||
"""One dominant speaker."""
|
||||
g = _gini([1, 1, 1, 100])
|
||||
assert g > 0.5, f"Expected >0.5, got {g}"
|
||||
|
||||
def test_half_life_zero():
|
||||
"""Message at t=now has weight ~1."""
|
||||
w = _half_life_weight(0)
|
||||
assert abs(w - 1.0) < 0.01
|
||||
|
||||
def test_half_life_decay():
|
||||
"""Message at t=tau has weight ~0.5."""
|
||||
w = _half_life_weight(7 * 86400, tau_days=7.0)
|
||||
assert abs(w - 0.5) < 0.01, f"Expected ~0.5, got {w}"
|
||||
|
||||
def test_reciprocity_symmetric():
|
||||
"""Equal exchange = high F3."""
|
||||
msgs = [
|
||||
{"from_id": "a", "to_id": "b", "timestamp": now - 0.1*day},
|
||||
{"from_id": "b", "to_id": "a", "timestamp": now - 0.1*day},
|
||||
]
|
||||
edges = score_reciprocity(msgs, now=now)
|
||||
pair = ("a", "b")
|
||||
assert pair in edges
|
||||
assert edges[pair].reciprocity_f3 > 0.6, f"Expected >0.6, got {edges[pair].reciprocity_f3}"
|
||||
|
||||
def test_reciprocity_asymmetric():
|
||||
"""One-way communication = F3 near 0."""
|
||||
msgs = [
|
||||
{"from_id": "a", "to_id": "b", "timestamp": now - 0.1*day},
|
||||
{"from_id": "a", "to_id": "b", "timestamp": now - 0.2*day},
|
||||
{"from_id": "a", "to_id": "b", "timestamp": now - 0.3*day},
|
||||
]
|
||||
edges = score_reciprocity(msgs, now=now)
|
||||
pair = ("a", "b")
|
||||
assert edges[pair].reciprocity_f3 < 0.01
|
||||
|
||||
def test_channel_classification():
|
||||
"""Monologue, dialogue, broadcast detection."""
|
||||
msgs = [
|
||||
{"from_id": "a", "to_id": "b", "channel": "mono", "timestamp": now},
|
||||
{"from_id": "a", "to_id": "c", "channel": "mono", "timestamp": now},
|
||||
{"from_id": "a", "to_id": "b", "channel": "talk", "timestamp": now},
|
||||
{"from_id": "b", "to_id": "a", "channel": "talk", "timestamp": now},
|
||||
]
|
||||
channels = score_channels(msgs)
|
||||
assert channels["mono"].classification == "monologue"
|
||||
assert channels["talk"].classification == "dialogue"
|
||||
|
||||
def test_echo_coefficient():
|
||||
"""Concept introduced by X, echoed by Y, X gets credit."""
|
||||
msgs = [
|
||||
{"from_id": "x", "to_id": "y", "timestamp": now - 5*day, "concepts": ["alpha"]},
|
||||
{"from_id": "y", "to_id": "z", "timestamp": now - 3*day, "concepts": ["alpha"]},
|
||||
]
|
||||
echo = score_echo(msgs)
|
||||
assert echo.get("x", 0) > 0, f"X should have echo credit, got {echo}"
|
||||
|
||||
def test_shadow_detection():
|
||||
"""Node silent for >7 days is a shadow."""
|
||||
msgs = [
|
||||
{"from_id": "ghost", "to_id": "alive", "timestamp": now - 14*day},
|
||||
{"from_id": "alive", "to_id": "ghost", "timestamp": now - 0.5*day},
|
||||
]
|
||||
shadows = detect_shadows(msgs, now=now, silence_days=7.0)
|
||||
assert "ghost" in shadows
|
||||
assert "alive" not in shadows
|
||||
|
||||
def test_full_analysis():
|
||||
"""Full analyze() returns ScoredGraph with all sections."""
|
||||
msgs = [
|
||||
{"from_id": "a", "to_id": "b", "timestamp": now - 1*day, "channel": "ch1", "concepts": ["x"]},
|
||||
{"from_id": "b", "to_id": "a", "timestamp": now - 0.5*day, "channel": "ch1"},
|
||||
]
|
||||
result = analyze(msgs, now=now)
|
||||
assert result.summary["total_messages"] == 2
|
||||
assert result.summary["total_nodes"] == 2
|
||||
assert len(result.edges) == 1
|
||||
assert len(result.channels) == 1
|
||||
assert len(result.nodes) == 2
|
||||
|
||||
# Run all tests
|
||||
tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
|
||||
passed = 0
|
||||
failed = 0
|
||||
for t in tests:
|
||||
try:
|
||||
t()
|
||||
print(f" ✓ {t.__name__}")
|
||||
passed += 1
|
||||
except Exception as e:
|
||||
print(f" ✗ {t.__name__}: {e}")
|
||||
failed += 1
|
||||
|
||||
print(f"\n{passed}/{passed+failed} tests passed")
|
||||
sys.exit(1 if failed > 0 else 0)
|
||||
Loading…
Reference in New Issue