python-math #7: feat(math): group-k-smoother in clojure-legacy engine mode (PR-D smoother)

delphi/polismath/conversation/conversation.py

@@ -26,6 +26,7 @@
 )
 from polismath.pca_kmeans_rep.repness import conv_repness
 from polismath.pca_kmeans_rep.corr import compute_correlation
+from polismath.pca_kmeans_rep.group_k_smoother import group_k_smoother_update
 from polismath.utils.engine_mode import resolve_engine_mode, ENGINE_MODE_LEGACY
 
 
@@ -847,8 +848,31 @@ def _compute_clusters(self,
 
         logger.info(f"Selected k={best_k} with silhouette={best_score:.4f}")
 
-        # Use the best clustering
-        group_labels, group_centers, group_member_lists, _ = group_clusterings[best_k]
+        # Engine-mode K selection (PR-D). 'improved' (default) keeps best_k
+        # exactly as computed above — bit-for-bit unchanged, including its
+        # strict-'>' tie-break (LOWER k wins ties). 'clojure-legacy' instead
+        # runs the group-k-smoother (conversation.clj:454-478): it damps K
+        # flicker (K only switches after :group-k-buffer=4 consecutive ticks
+        # agree) and uses Clojure's max-key HIGHER-k-wins tie-break, threading
+        # {last_k, last_k_count, smoothed_k} plus the per-k clusterings across
+        # ticks on the conv. These threaded fields are NOT persisted (matching
+        # conv_man.clj:52-74) — they live in-memory across update_votes only.
+        if resolve_engine_mode() == ENGINE_MODE_LEGACY:
+            silhouettes_by_k = {k: group_clusterings[k][3] for k in group_clusterings}
+            new_smoother_state, selected_k = group_k_smoother_update(
+                prev_group_k_smoother or {}, silhouettes_by_k)
+            self.group_clusterings = group_clusterings
+            self.group_k_smoother = new_smoother_state
+            logger.info(f"Legacy group-k-smoother: best_k={best_k} "
+                        f"smoothed_k={selected_k} state={new_smoother_state}")
+        else:
+            selected_k = best_k
+
+        # Use the selected clustering (best_k in improved mode, smoothed_k in
+        # legacy mode). The smoother's clamp guarantees selected_k is a key of
+        # group_clusterings, so this never KeyErrors and group_clusters is never
+        # None.
+        group_labels, group_centers, group_member_lists, _ = group_clusterings[selected_k]
 
         # Convert to dictionary format with base cluster IDs as members
         group_clusters = []

delphi/polismath/pca_kmeans_rep/group_k_smoother.py

@@ -0,0 +1,129 @@
+"""
+Group-K smoother: port of Clojure :group-k-smoother (conversation.clj:454-478).
+
+Damps flicker in K (the number of opinion groups). Left un-smoothed, K would
+jump every tick to whatever k currently maximizes the silhouette; the smoother
+only lets K switch to a new best value after `:group-k-buffer` (= 4,
+conversation.clj:154) consecutive ticks agree on it.
+
+The state {last_k, last_k_count, smoothed_k} is threaded ON THE CONV across
+conv-update ticks (conversation.clj:457) and is NOT persisted (Clojure's
+math_main whitelist omits it, conv_man.clj:52-74). This module is a pure
+function so it can be unit-tested in isolation and reused by
+Conversation._compute_clusters.
+
+NOTE: Python has no subgroups (conversation.py hardcodes subgroup_clusters =
+{}), so ONLY the top-level group-k-smoother is ported here — the parallel
+subgroup smoother (conversation.clj:520-560) is intentionally not.
+"""
+
+from typing import Any, Dict, Mapping, Optional, Tuple
+
+# Clojure :group-k-buffer default (conversation.clj:154): switch K only after
+# this many consecutive ticks agree on a new best-K.
+GROUP_K_BUFFER = 4
+
+
+def _argmax_silhouette_higher_k_wins(silhouettes_by_k: Mapping[int, float]) -> Optional[int]:
+    """
+    argmax_k silhouette[k], breaking ties toward the HIGHER k.
+
+    Clojure computes this as
+        (apply max-key group-clusterings-silhouettes (keys group-clusterings))
+    (conversation.clj:461). Clojure's `max-key` returns the LAST argument among
+    equal-valued maxima, and the group-clusterings map (built by
+    `plmb/map-from-keys` over `(range 2 (inc max-k))`) iterates its keys in
+    ASCENDING order for the small array-maps used here. So on a silhouette tie
+    the HIGHER k is kept. We reproduce that by scanning k ascending and
+    replacing the incumbent on `>=` (not strict `>`).
+
+    Returns None only if `silhouettes_by_k` is empty (the caller guarantees at
+    least k=2 is present, conversation.py group loop over range(2, max_k+1)
+    with max_k >= 2).
+    """
+    best_k: Optional[int] = None
+    best_score: Optional[float] = None
+    for k in sorted(silhouettes_by_k):
+        score = silhouettes_by_k[k]
+        if best_score is None or score >= best_score:
+            best_k = k
+            best_score = score
+    return best_k
+
+
+def group_k_smoother_update(
+    prev_state: Optional[Mapping[str, Any]],
+    silhouettes_by_k: Mapping[int, float],
+    buffer: int = GROUP_K_BUFFER,
+) -> Tuple[Dict[str, Any], int]:
+    """
+    Advance the group-K smoother by one tick. Pure function.
+
+    Port of Clojure :group-k-smoother (conversation.clj:454-478).
+
+    State carried on the conv across ticks (conversation.clj:457):
+      last_k        - the best-K from the previous tick (None on the first tick)
+      last_k_count  - consecutive ticks this_k has equalled last_k
+                      (Clojure `:or {last-k-count 0}`, conversation.clj:457)
+      smoothed_k    - the K actually used last tick (None on the first tick)
+
+    Update rule (conversation.clj:461-478):
+      this_k       = argmax_k silhouette   (ties -> higher k; see helper)
+      same         = last_k is not None and this_k == last_k
+      this_k_count = last_k_count + 1 if same else 1
+      smoothed_k   = this_k                          if this_k_count >= buffer
+                     else (prev smoothed_k if not None else this_k)
+      clamp (#2536, conversation.clj:469-478): if smoothed_k is not among the
+             current clusterings' k-values, fall back to this_k.
+
+    First tick (smoothed_k is None): accepts this_k immediately — the
+    cold-start invariant that makes 'improved' and 'clojure-legacy' coincide on
+    tick 1.
+
+    Args:
+        prev_state: previous {last_k, last_k_count, smoothed_k}; None/{} on the
+            first tick.
+        silhouettes_by_k: {k: silhouette} for THIS tick's clusterings. Its keys
+            are the valid k-values used by the clamp.
+        buffer: consecutive-agreement threshold before switching K
+            (Clojure :group-k-buffer, default 4).
+
+    Returns:
+        (new_state, smoothed_k). `smoothed_k` is guaranteed to be a key of
+        `silhouettes_by_k`, so `clusterings[smoothed_k]` never KeyErrors.
+
+    Raises:
+        ValueError: if `silhouettes_by_k` is empty — the membership guarantee
+            above would be impossible to honor.
+    """
+    if not silhouettes_by_k:
+        raise ValueError(
+            "group_k_smoother_update requires a non-empty silhouettes_by_k: "
+            "smoothed_k must be one of its keys")
+    state = prev_state or {}
+    last_k = state.get('last_k')                  # None if absent
+    last_k_count = state.get('last_k_count', 0)   # Clojure :or {last-k-count 0}
+    prev_smoothed_k = state.get('smoothed_k')     # None if absent
+
+    this_k = _argmax_silhouette_higher_k_wins(silhouettes_by_k)
+
+    same = last_k is not None and this_k == last_k
+    this_k_count = last_k_count + 1 if same else 1
+
+    if this_k_count >= buffer:
+        smoothed_k = this_k
+    else:
+        smoothed_k = prev_smoothed_k if prev_smoothed_k is not None else this_k
+
+    # Clamp (#2536, conversation.clj:469-478): a carried smoothed_k that no
+    # longer exists this tick (e.g. the base-cluster count shrank so max-k
+    # dropped) falls back to the current best available k.
+    if smoothed_k not in silhouettes_by_k:
+        smoothed_k = this_k
+
+    new_state = {
+        'last_k': this_k,
+        'last_k_count': this_k_count,
+        'smoothed_k': smoothed_k,
+    }
+    return new_state, smoothed_k

delphi/tests/test_group_k_smoother.py

@@ -0,0 +1,206 @@
+#!/usr/bin/env python3
+"""
+Tests for the group-K smoother (PR-D smoother part).
+
+Port of Clojure :group-k-smoother (conversation.clj:454-478), which damps the
+number-of-opinion-groups (K) so it only switches to a new best value after
+`:group-k-buffer` (4, conversation.clj:154) consecutive ticks agree on it.
+
+Two layers:
+  1. Pure-function unit tests (buffer counting, reset-on-change, clamp,
+     first-tick, higher-k tie-break) — fast, deterministic.
+  2. A chained-update_votes integration test proving the smoother is threaded
+     across ticks in 'clojure-legacy' mode (no flicker on brief alternation,
+     switch after 4 consecutive) and is inert in 'improved' mode.
+"""
+
+import os
+import sys
+
+import pytest
+
+sys.path.append(os.path.abspath(os.path.dirname(__file__)))
+
+from polismath.pca_kmeans_rep.group_k_smoother import (
+    group_k_smoother_update,
+    GROUP_K_BUFFER,
+)
+from polismath.utils.engine_mode import ENGINE_MODE_ENV_VAR
+from polismath.pca_kmeans_rep.pca import PCA_IMPL_ENV_VAR
+import polismath.conversation.conversation as conv_mod
+from polismath.conversation.conversation import Conversation
+
+
+# ---------------------------------------------------------------------------
+# Pure-function unit tests
+# ---------------------------------------------------------------------------
+
+def _sils(chosen, ks=(2, 3)):
+    """Silhouettes that make `chosen` the unique argmax over ks."""
+    return {k: (1.0 if k == chosen else 0.0) for k in ks}
+
+
+def _drive(this_k_seq, ks=(2, 3), buffer=GROUP_K_BUFFER):
+    """Chain the smoother over a sequence of desired this_k values, returning
+    the list of smoothed_k emitted at each tick."""
+    state = {}
+    out = []
+    for chosen in this_k_seq:
+        state, sm = group_k_smoother_update(state, _sils(chosen, ks), buffer=buffer)
+        out.append(sm)
+    return out
+
+
+class TestGroupKSmootherPure:
+
+    def test_default_buffer_is_four(self):
+        assert GROUP_K_BUFFER == 4  # Clojure :group-k-buffer, conversation.clj:154
+
+    def test_first_tick_accepts_best_k(self):
+        state, sm = group_k_smoother_update({}, {2: 0.1, 3: 0.9})
+        assert sm == 3
+        assert state == {'last_k': 3, 'last_k_count': 1, 'smoothed_k': 3}
+
+    def test_first_tick_accepts_best_k_none_state(self):
+        _, sm = group_k_smoother_update(None, {2: 0.9, 3: 0.1})
+        assert sm == 2
+
+    def test_switch_only_after_four_consecutive(self):
+        # smoothed_k established at 2, then this_k flips to 3 and must wait 4
+        # consecutive ticks before smoothed_k follows.
+        out = _drive([2, 2, 2, 2, 3, 3, 3, 3])
+        assert out == [2, 2, 2, 2, 2, 2, 2, 3]
+
+    def test_reset_on_change(self):
+        # A single interrupting this_k=2 (tick index 6) resets the 3-streak, so
+        # the switch is delayed until 4 fresh consecutive 3's accumulate.
+        out = _drive([2, 2, 2, 2, 3, 3, 2, 3, 3, 3, 3])
+        assert out == [2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 3]
+        # Interrupt really did reset the counter (would have switched at index 7
+        # without the reset).
+        assert out[7] == 2
+
+    def test_brief_alternation_does_not_flicker(self):
+        # this_k alternates but never reaches 4-in-a-row -> smoothed_k pinned.
+        out = _drive([2, 3, 2, 3, 2, 3])
+        assert out == [2, 2, 2, 2, 2, 2]
+
+    def test_clamp_missing_carried_smoothed_falls_back_to_this_k(self):
+        # Carried smoothed_k=5 no longer exists among {2,3} -> fall back to
+        # this_k, never KeyError (Clojure clamp #2536, conversation.clj:469-478).
+        prev = {'last_k': 5, 'last_k_count': 10, 'smoothed_k': 5}
+        state, sm = group_k_smoother_update(prev, {2: 0.9, 3: 0.1})
+        assert sm == 2  # this_k
+        assert state['smoothed_k'] == 2
+
+    def test_clamp_present_carried_smoothed_is_kept(self):
+        prev = {'last_k': 3, 'last_k_count': 1, 'smoothed_k': 3}
+        # this_k=2 but not yet 4-in-a-row, so smoothed_k should stay carried 3
+        # (which IS present) rather than flip.
+        _, sm = group_k_smoother_update(prev, {2: 0.9, 3: 0.1})
+        assert sm == 3
+
+    def test_empty_silhouettes_raises(self):
+        """The contract guarantees smoothed_k is a key of silhouettes_by_k —
+        impossible for an empty dict, so fail fast instead of returning None."""
+        with pytest.raises(ValueError, match="non-empty"):
+            group_k_smoother_update({}, {})
+
+    def test_tie_break_higher_k_wins(self):
+        # Clojure max-key returns the LAST maximal arg; keys iterate ascending
+        # -> higher k wins ties (conversation.clj:461).
+        _, sm = group_k_smoother_update({}, {2: 0.5, 3: 0.5})
+        assert sm == 3
+
+    def test_tie_break_higher_k_wins_among_partial_ties(self):
+        _, sm = group_k_smoother_update({}, {2: 0.5, 3: 0.5, 4: 0.2})
+        assert sm == 3
+        _, sm = group_k_smoother_update({}, {2: 0.2, 3: 0.5, 4: 0.5})
+        assert sm == 4
+
+
+# ---------------------------------------------------------------------------
+# Pipeline integration: smoother threaded across chained update_votes
+# ---------------------------------------------------------------------------
+
+class _SilStub:
+    """Deterministic silhouette stub keyed on CALL INDEX, not label counts, so
+    it is robust to empty clusters. The group loop calls silhouette once per k
+    in ascending order (k=2 then k=3) every tick, so call 2*t is the k=2 call
+    of tick t and call 2*t+1 is the k=3 call. Returns 1.0 for the tick's
+    preferred k and 0.0 otherwise."""
+
+    def __init__(self, prefs):
+        self.prefs = prefs
+        self.n = 0
+
+    def __call__(self, X, labels):
+        idx = self.n
+        self.n += 1
+        tick = idx // 2
+        this_call_k = 2 if (idx % 2 == 0) else 3
+        pref = self.prefs[min(tick, len(self.prefs) - 1)]
+        return 1.0 if this_call_k == pref else 0.0
+
+
+def _many_ptpt_votes(n_ptpts=18, n_cmnts=8):
+    """18 DISTINCT ternary vote rows over 8 comments (3 group signatures + 5
+    unique bits), so base k-means yields 18 singleton base clusters and
+    max-k = min(5, 2 + 18//12) = 3 -> group clusterings for k in {2, 3}."""
+    votes = []
+    for i in range(n_ptpts):
+        g = i % 3
+        for j in range(n_cmnts):
+            if j < 3:
+                v = 1.0 if j == g else -1.0
+            else:
+                v = 1.0 if ((i >> (j - 3)) & 1) else -1.0
+            votes.append({'pid': f'p{i}', 'tid': f'c{j}', 'vote': v})
+    return {'votes': votes}
+
+
+# A repeat of p0's c0 vote (g=0 -> j==0 -> +1): unchanged matrix, but still
+# triggers a recompute tick.
+_REPEAT_VOTE = {'votes': [{'pid': 'p0', 'tid': 'c0', 'vote': 1.0}]}
+
+
+class TestSmootherPipeline:
+
+    def _setup(self, monkeypatch, mode, prefs):
+        monkeypatch.delenv(PCA_IMPL_ENV_VAR, raising=False)  # default powerit
+        monkeypatch.setenv(ENGINE_MODE_ENV_VAR, mode)
+        stub = _SilStub(prefs)
+        monkeypatch.setattr(conv_mod, 'calculate_silhouette_sklearn', stub)
+        return stub
+
+    def test_legacy_no_flicker_then_switch_after_four(self, monkeypatch):
+        # this_k schedule: 2,2,3,2,3,3,3,3,3  (the isolated 3 at tick 2 is brief
+        # noise; 3 becomes stable from tick 4 -> switch on the 4th consecutive).
+        prefs = [2, 2, 3, 2, 3, 3, 3, 3, 3]
+        self._setup(monkeypatch, 'clojure-legacy', prefs)
+
+        conv = Conversation('smooth').update_votes(_many_ptpt_votes())
+        # Sanity: the two-tick group clusterings really exist for k in {2,3}.
+        assert set(conv.group_clusterings.keys()) == {2, 3}
+
+        smoothed = [conv.group_k_smoother['smoothed_k']]
+        for _ in range(1, len(prefs)):
+            conv = conv.update_votes(_REPEAT_VOTE)
+            smoothed.append(conv.group_k_smoother['smoothed_k'])
+
+        assert smoothed == [2, 2, 2, 2, 2, 2, 2, 3, 3], smoothed
+        # Brief alternation (tick 2) did not flip; switch happened only on the
+        # 4th consecutive this_k=3 (tick 7), not the 3rd (tick 6).
+        assert smoothed[6] == 2 and smoothed[7] == 3
+        # group_clusters is picked from the smoothed k and is never None/empty.
+        assert conv.group_clusters, "group_clusters must be populated"
+
+    def test_improved_mode_leaves_smoother_inert(self, monkeypatch):
+        prefs = [3, 3, 3, 3]
+        self._setup(monkeypatch, 'improved', prefs)
+        conv = Conversation('smooth').update_votes(_many_ptpt_votes())
+        # Improved mode never touches the smoother/clusterings state.
+        assert conv.group_k_smoother == {}
+        assert conv.group_clusterings == {}
+        # But still produces group clusters via the untouched best_k path.
+        assert conv.group_clusters, "group_clusters must be populated"