Phase 2: Noosphere Engine analytics modules

pyproject.toml

@@ -22,6 +22,8 @@ dependencies = [
     "structlog>=24.4.0",
     "defusedxml>=0.7.1",
     "trafilatura>=1.12.0",
+    "ruptures>=1.1.10",
+    "networkx>=3.6.1",
 ]
 
 [project.optional-dependencies]
@@ -91,6 +93,8 @@ module = [
     "googleapiclient.*",
     "youtube_transcript_api.*",
     "trafilatura.*",
+    "ruptures.*",
+    "networkx.*",
 ]
 ignore_missing_imports = true
 

src/intelstream/noosphere/attractor_dashboard/cog.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+from collections import defaultdict
+from typing import TYPE_CHECKING
+
+import discord
+import structlog
+from discord import app_commands
+from discord.ext import commands
+
+from intelstream.noosphere.attractor_dashboard.metrics import (
+    find_change_points,
+    format_dashboard,
+)
+
+if TYPE_CHECKING:
+    from intelstream.bot import IntelStreamBot
+    from intelstream.noosphere.shared.data_models import CommunityStateVector
+
+logger = structlog.get_logger(__name__)
+
+MAX_HISTORY = 168
+
+
+class AttractorDashboardCog(commands.Cog):
+    def __init__(self, bot: IntelStreamBot) -> None:
+        self.bot = bot
+        self._history: dict[str, list[CommunityStateVector]] = defaultdict(list)
+
+    @commands.Cog.listener("on_state_vector_updated")
+    async def _on_state_vector(self, csv: CommunityStateVector) -> None:
+        history = self._history[csv.guild_id]
+        history.append(csv)
+        if len(history) > MAX_HISTORY:
+            self._history[csv.guild_id] = history[-MAX_HISTORY:]
+
+    @app_commands.command(name="dashboard", description="View community attractor dashboard")
+    async def dashboard(self, interaction: discord.Interaction) -> None:
+        if interaction.guild is None:
+            await interaction.response.send_message(
+                "This command can only be used in a server.", ephemeral=True
+            )
+            return
+
+        guild_id = str(interaction.guild.id)
+        history = self._history.get(guild_id, [])
+        if not history:
+            await interaction.response.send_message(
+                "No community data available yet.", ephemeral=True
+            )
+            return
+
+        latest = history[-1]
+        change_points = find_change_points(history) if len(history) >= 10 else None
+        lines = format_dashboard(latest, change_points)
+
+        embed = discord.Embed(
+            title="Attractor Dashboard",
+            description="```\n" + "\n".join(lines) + "\n```",
+            color=0x9B59B6,
+        )
+
+        if change_points:
+            cp_summary = ", ".join(f"{cp.metric} ({cp.direction})" for cp in change_points[:5])
+            embed.add_field(name="Change Points Detected", value=cp_summary, inline=False)
+
+        embed.set_footer(text=f"History: {len(history)} snapshots")
+
+        await interaction.response.send_message(embed=embed)
+        logger.debug(
+            "attractor dashboard displayed",
+            guild_id=guild_id,
+            history_size=len(history),
+        )

src/intelstream/noosphere/attractor_dashboard/metrics.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+import numpy as np
+import ruptures
+
+if TYPE_CHECKING:
+    from intelstream.noosphere.shared.data_models import CommunityStateVector
+
+METRIC_FIELDS = [
+    "semantic_coherence",
+    "semantic_momentum",
+    "topic_entropy",
+    "topic_churn",
+    "activity_entropy",
+    "reply_depth",
+    "interaction_modularity",  # populated by Phase 1 metrics_computer
+    "anthrophony_ratio",
+    "biophony_ratio",
+]
+
+METRIC_LABELS = {
+    "semantic_coherence": "Coherence",
+    "semantic_momentum": "Momentum",
+    "topic_entropy": "Topic Entropy",
+    "topic_churn": "Topic Churn",
+    "activity_entropy": "Activity Entropy",
+    "reply_depth": "Reply Depth",
+    "interaction_modularity": "Modularity",
+    "anthrophony_ratio": "Anthrophony",
+    "biophony_ratio": "Biophony",
+}
+
+
+@dataclass
+class ChangePoint:
+    metric: str
+    index: int
+    direction: str
+
+
+def extract_metric_series(
+    history: list[CommunityStateVector], field: str
+) -> np.ndarray[tuple[int], np.dtype[np.float64]]:
+    return np.array([getattr(csv, field) for csv in history], dtype=np.float64)
+
+
+def detect_change_points(
+    series: np.ndarray[tuple[int], np.dtype[np.float64]],
+    min_size: int = 5,
+    penalty: float = 3.0,
+) -> list[int]:
+    if len(series) < min_size * 2:
+        return []
+    algo = ruptures.Pelt(model="rbf", min_size=min_size).fit(series)
+    breakpoints: list[int] = algo.predict(pen=penalty)
+    return [bp for bp in breakpoints if bp < len(series)]
+
+
+def find_change_points(
+    history: list[CommunityStateVector],
+    min_size: int = 5,
+    penalty: float = 3.0,
+) -> list[ChangePoint]:
+    results: list[ChangePoint] = []
+    for field_name in METRIC_FIELDS:
+        series = extract_metric_series(history, field_name)
+        cps = detect_change_points(series, min_size=min_size, penalty=penalty)
+        for cp in cps:
+            before = float(np.mean(series[max(0, cp - min_size) : cp]))
+            after_end = min(cp + min_size, len(series))
+            after = float(np.mean(series[cp:after_end])) if cp < len(series) else before
+            direction = "up" if after > before else "down"
+            results.append(ChangePoint(metric=field_name, index=cp, direction=direction))
+    return results
+
+
+def format_dashboard(
+    csv: CommunityStateVector,
+    change_points: list[ChangePoint] | None = None,
+) -> list[str]:
+    lines: list[str] = []
+    cp_map: dict[str, str] = {}
+    if change_points:
+        for cp in change_points:
+            cp_map[cp.metric] = " [!]"
+
+    for field_name in METRIC_FIELDS:
+        label = METRIC_LABELS[field_name]
+        value = getattr(csv, field_name)
+        flag = cp_map.get(field_name, "")
+        lines.append(f"{label:<18} {value:>6.3f}{flag}")
+
+    return lines

src/intelstream/noosphere/constants.py

@@ -0,0 +1,13 @@
+import enum
+import math
+
+PHI: float = (1.0 + math.sqrt(5.0)) / 2.0
+GOLDEN_ANGLE: float = 2.0 * math.pi / (PHI**2)
+
+FIBONACCI_SEQ: list[int] = [1, 1, 2, 3, 5, 8, 13, 21, 34, 55]
+
+
+class MessageClassification(enum.StrEnum):
+    ANTHROPHONY = "anthrophony"
+    BIOPHONY = "biophony"
+    GEOPHONY = "geophony"

src/intelstream/noosphere/cordyceps_audit/audit.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class CordycepsReport:
+    herfindahl_index: float
+    vocabulary_jaccard: float
+    parasitism_score: float
+    flagged: bool
+
+
+def herfindahl_index(message_counts: dict[str, int]) -> float:
+    total = sum(message_counts.values())
+    if total == 0:
+        return 0.0
+    return sum((count / total) ** 2 for count in message_counts.values())
+
+
+def vocabulary_jaccard(bot_terms: set[str], community_terms: set[str]) -> float:
+    if not bot_terms and not community_terms:
+        return 0.0
+    intersection = bot_terms & community_terms
+    union = bot_terms | community_terms
+    if not union:
+        return 0.0
+    return len(intersection) / len(union)
+
+
+def compute_parasitism_score(
+    hhi: float,
+    vocab_jaccard: float,
+    hhi_weight: float = 0.5,
+    vocab_weight: float = 0.5,
+) -> float:
+    return hhi_weight * hhi + vocab_weight * vocab_jaccard
+
+
+def run_audit(
+    message_counts: dict[str, int],
+    bot_terms: set[str],
+    community_terms: set[str],
+    parasitism_threshold: float = 0.6,
+) -> CordycepsReport:
+    hhi = herfindahl_index(message_counts)
+    vjac = vocabulary_jaccard(bot_terms, community_terms)
+    score = compute_parasitism_score(hhi, vjac)
+    return CordycepsReport(
+        herfindahl_index=hhi,
+        vocabulary_jaccard=vjac,
+        parasitism_score=score,
+        flagged=score > parasitism_threshold,
+    )

src/intelstream/noosphere/cordyceps_audit/cog.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+from collections import defaultdict
+from typing import TYPE_CHECKING
+
+import discord
+import structlog
+from discord import app_commands
+from discord.ext import commands
+
+from intelstream.noosphere.cordyceps_audit.audit import run_audit
+from intelstream.noosphere.cordyceps_audit.vocabulary_tracker import VocabularyTracker
+
+if TYPE_CHECKING:
+    from intelstream.bot import IntelStreamBot
+    from intelstream.noosphere.shared.data_models import ProcessedMessage
+
+logger = structlog.get_logger(__name__)
+
+
+class CordycepsAuditCog(commands.Cog):
+    def __init__(self, bot: IntelStreamBot) -> None:
+        self.bot = bot
+        self._trackers: dict[str, VocabularyTracker] = defaultdict(VocabularyTracker)
+        self._message_counts: dict[str, dict[str, int]] = defaultdict(lambda: defaultdict(int))
+
+    @commands.Cog.listener("on_message_processed")
+    async def _on_message(self, msg: ProcessedMessage) -> None:
+        tracker = self._trackers[msg.guild_id]
+        counts = self._message_counts[msg.guild_id]
+        counts[msg.user_id] += 1
+
+        if msg.is_bot:
+            tracker.record_bot_message(msg.content)
+        else:
+            tracker.record_community_message(msg.content)
+
+    @app_commands.command(name="cordyceps", description="Run a Cordyceps influence audit")
+    async def cordyceps(self, interaction: discord.Interaction) -> None:
+        if interaction.guild is None:
+            await interaction.response.send_message(
+                "This command can only be used in a server.", ephemeral=True
+            )
+            return
+
+        guild_id = str(interaction.guild.id)
+        tracker = self._trackers.get(guild_id)
+        counts = self._message_counts.get(guild_id)
+
+        if tracker is None or counts is None or not counts:
+            await interaction.response.send_message(
+                "Insufficient data for a Cordyceps audit.", ephemeral=True
+            )
+            return
+
+        report = run_audit(
+            message_counts=dict(counts),
+            bot_terms=tracker.bot_terms,
+            community_terms=tracker.community_terms,
+        )
+
+        color = 0xE74C3C if report.flagged else 0x2ECC71
+        status = "FLAGGED" if report.flagged else "Healthy"
+
+        embed = discord.Embed(
+            title="Cordyceps Audit",
+            color=color,
+        )
+        embed.add_field(name="Status", value=status, inline=True)
+        embed.add_field(
+            name="Parasitism Score",
+            value=f"{report.parasitism_score:.3f}",
+            inline=True,
+        )
+        embed.add_field(
+            name="Herfindahl Index",
+            value=f"{report.herfindahl_index:.3f}",
+            inline=True,
+        )
+        embed.add_field(
+            name="Vocabulary Overlap",
+            value=f"{report.vocabulary_jaccard:.3f}",
+            inline=True,
+        )
+
+        await interaction.response.send_message(embed=embed)
+        logger.info(
+            "cordyceps audit completed",
+            guild_id=guild_id,
+            parasitism_score=report.parasitism_score,
+            flagged=report.flagged,
+        )

src/intelstream/noosphere/cordyceps_audit/vocabulary_tracker.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import re
+from collections import Counter
+
+_WORD_RE = re.compile(r"\b[a-zA-Z]{3,}\b")
+
+
+def extract_terms(text: str) -> list[str]:
+    return [m.lower() for m in _WORD_RE.findall(text)]
+
+
+class VocabularyTracker:
+    def __init__(self, top_n: int = 200) -> None:
+        self._top_n = top_n
+        self._bot_counter: Counter[str] = Counter()
+        self._community_counter: Counter[str] = Counter()
+
+    def record_bot_message(self, text: str) -> None:
+        self._bot_counter.update(extract_terms(text))
+
+    def record_community_message(self, text: str) -> None:
+        self._community_counter.update(extract_terms(text))
+
+    @property
+    def bot_terms(self) -> set[str]:
+        return {term for term, _ in self._bot_counter.most_common(self._top_n)}
+
+    @property
+    def community_terms(self) -> set[str]:
+        return {term for term, _ in self._community_counter.most_common(self._top_n)}
+
+    def reset(self) -> None:
+        self._bot_counter.clear()
+        self._community_counter.clear()