diff --git a/delphi/polismath/conversation/conversation.py b/delphi/polismath/conversation/conversation.py index fa497e180..1c66a3ff4 100644 --- a/delphi/polismath/conversation/conversation.py +++ b/delphi/polismath/conversation/conversation.py @@ -27,12 +27,61 @@ 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.pca_kmeans_rep.legacy_kmeans import ( + _NamedData as _LegacyNamedData, + kmeans as legacy_kmeans, +) from polismath.utils.engine_mode import resolve_engine_mode, ENGINE_MODE_LEGACY # Configure logging logger = logging.getLogger(__name__) + +def _base_clusters_to_legacy(base_clusters: Optional[List[Dict[str, Any]]]) -> Optional[List[Dict[str, Any]]]: + """Convert stored base clusters ({id, center: list, members: pids}) into the + legacy_kmeans warm-start form ({id, members, center: np.ndarray}). + + Returns None for empty/None input so the first tick cold-starts via + init-clusters (Clojure: a falsey :last-clusters -> init-clusters, + clusters.clj:305-307). PR-C warm-start plumbing for + :last-clusters (:base-clusters conv) (conversation.clj:409). + """ + if not base_clusters: + return None + return [ + {'id': c['id'], + 'members': list(c['members']), + 'center': np.asarray(c['center'], dtype=float)} + for c in base_clusters + ] + + +def _labels_from_id_clusters(row_names: List[Any], + clusters: List[Dict[str, Any]]) -> np.ndarray: + """Label array aligned with ``row_names`` for silhouette scoring: the index + (in ``clusters``) of the cluster that contains each name. + + Used at the group level to score a legacy (id-carrying) clustering with the + same ``calculate_silhouette_sklearn`` the improved path uses, so the smoother + sees comparable silhouettes. Every base cluster is assigned to exactly one + group cluster by ``cluster-step``; a name that (degenerately) appears in none + gets its own singleton label so it never silently merges into label 0. + """ + label_by_name: Dict[Any, int] = {} + for label, c in enumerate(clusters): + for m in c['members']: + label_by_name[m] = label + next_label = len(clusters) + labels = [] + for name in row_names: + if name in label_by_name: + labels.append(label_by_name[name]) + else: + labels.append(next_label) + next_label += 1 + return np.array(labels) + # Set up default logging only if root logger is not configured # This prevents duplicate handlers when logging is configured externally if not logging.root.handlers: @@ -711,6 +760,7 @@ def _get_clean_matrix(self, raw: bool = False) -> pd.DataFrame: return pd.DataFrame(matrix_data, index=source.index, columns=source.columns) def _compute_clusters(self, + prev_base_clusters: Optional[List[Dict[str, Any]]] = None, prev_group_clusterings: Optional[Dict[Any, Any]] = None, prev_group_k_smoother: Optional[Dict[str, Any]] = None) -> None: """ @@ -720,14 +770,16 @@ def _compute_clusters(self, Level 2: Group clusters (base clusters → 2-5 groups with silhouette-based k selection) Args: - prev_group_clusterings: The previous tick's per-k group clusterings - dict (k -> clustering tuple), or None. Captured and threaded by - recompute() but NOT consumed yet: the per-k k-means warm start - that will use it (Clojure :group-clusterings → :last-clusters, - conversation.clj:441-442) lands with the Clojure-exact k-means - lineage port (PR-C; see SEQUENTIAL_BITS_PORT_SPEC.md §2.3). - Until then, legacy-mode group clusterings are recomputed cold - each tick. Ignored in the default 'improved' mode. + prev_base_clusters: The previous tick's base clusters + (list of {id, center, members}), or None. Consumed ONLY in + 'clojure-legacy' engine mode as the base-level k-means warm start + (Clojure :last-clusters (:base-clusters conv), conversation.clj:409). + Ignored in the default 'improved' mode. + prev_group_clusterings: The previous tick's per-k group clusterings, + or None. In 'clojure-legacy' mode this is {k: [id-carrying cluster + dicts]} — the warm start for per-k group k-means (Clojure + :last-clusters (last-clusterings k), conversation.clj:441). Ignored + in 'improved' mode (where it is never even written, so it stays {}). prev_group_k_smoother: The previous tick's group-k-smoother state {last_k, last_k_count, smoothed_k}, or None. Consumed ONLY in 'clojure-legacy' mode (conversation.clj:457). Ignored in @@ -740,14 +792,22 @@ def _compute_clusters(self, # Configuration (matching Clojure defaults) BASE_K = 100 MAX_K = 5 - BASE_ITERS = 100 - GROUP_ITERS = 100 + BASE_ITERS = 100 # Clojure :base-iters (conversation.clj:147) + GROUP_ITERS = 100 # improved-mode group iterations (unchanged) + # Legacy-mode group iterations: Clojure passes :cluster-iters — a key + # kmeans IGNORES — so the group level runs kmeans' DEFAULT max-iters of + # 20, not :group-iters (clusters.clj:303, conversation.clj:443). + GROUP_LEGACY_ITERS = 20 # Check if we have projections if not self.proj: self.base_clusters = [] self.group_clusters = [] self.subgroup_clusters = {} + # P6a: no projections == the degenerate/empty conv. Clojure's + # conv-update SHORT-CIRCUITS a truly-empty conv (conversation.clj:807-811) + # and computes nothing, so the group-k smoother state is intentionally + # LEFT FROZEN here (no advance) — faithful to Clojure, not a divergence. logger.info(f"Clustering completed in {time.time() - start_time:.2f}s (no projections)") return @@ -772,28 +832,53 @@ def _compute_clusters(self, # Adjust BASE_K if we have fewer participants actual_base_k = min(BASE_K, len(in_conv_pids_list)) - logger.info(f"Computing base clusters with k={actual_base_k}...") - base_labels, base_centers, base_member_lists = kmeans_sklearn( - base_proj_values, - k=actual_base_k, - max_iters=BASE_ITERS - ) + legacy_mode = resolve_engine_mode() == ENGINE_MODE_LEGACY - # Convert to dictionary format with participant IDs as members - base_clusters = [] - for cluster_id, (center, member_indices) in enumerate(zip(base_centers, base_member_lists)): - # Map indices back to participant IDs - member_pids = [in_conv_pids_list[idx] for idx in member_indices] - base_clusters.append({ - 'id': cluster_id, - 'center': center.tolist(), - 'members': member_pids - }) + logger.info(f"Computing base clusters with k={actual_base_k}...") + if legacy_mode: + # PR-C: base-level warm start with lineage. Clojure threads the prior + # tick's base clusters into k-means as :last-clusters + # (conversation.clj:403-410 -> clusters.clj:301-312 -> clean-start- + # clusters), so base-cluster ids are STABLE across ticks, new ids + # strictly increase, and merges keep the larger side's id. The ported + # legacy_kmeans keys clusters to the current data by member NAME + # (participant id), which is what lets prior members be recentered or + # dropped. base-iters = 100 (conversation.clj:147). + base_data = _LegacyNamedData(in_conv_pids_list, base_proj_values) + last_base = _base_clusters_to_legacy(prev_base_clusters) + legacy_base = legacy_kmeans( + base_data, actual_base_k, + last_clusters=last_base, weights=None, max_iters=BASE_ITERS) + legacy_base.sort(key=lambda c: c['id']) # Clojure sort-by :id (conversation.clj:406) + base_clusters = [ + {'id': c['id'], + 'center': np.asarray(c['center'], dtype=float).tolist(), + 'members': list(c['members'])} + for c in legacy_base + ] + else: + # Improved (default): cold recompute, byte-for-byte unchanged. + base_labels, base_centers, base_member_lists = kmeans_sklearn( + base_proj_values, + k=actual_base_k, + max_iters=BASE_ITERS + ) - # Keep base clusters in k-means ID order (matching Clojure's sort-by :id) - # Do NOT sort by size or reassign IDs — that would change the encounter - # order of centers used in group clustering's first-k-distinct initialization. - base_clusters.sort(key=lambda c: c['id']) + # Convert to dictionary format with participant IDs as members + base_clusters = [] + for cluster_id, (center, member_indices) in enumerate(zip(base_centers, base_member_lists)): + # Map indices back to participant IDs + member_pids = [in_conv_pids_list[idx] for idx in member_indices] + base_clusters.append({ + 'id': cluster_id, + 'center': center.tolist(), + 'members': member_pids + }) + + # Keep base clusters in k-means ID order (matching Clojure's sort-by :id) + # Do NOT sort by size or reassign IDs — that would change the encounter + # order of centers used in group clustering's first-k-distinct initialization. + base_clusters.sort(key=lambda c: c['id']) logger.info(f"Created {len(base_clusters)} base clusters") @@ -811,6 +896,22 @@ def _compute_clusters(self, else: self.group_clusters = [] self.subgroup_clusters = {} + # P6a: Clojure has NO <2-base-cluster guard. Its max-k-fn is + # (min max-max-k (+ 2 (int (/ n 12)))) -> ALWAYS >= 2 + # (conversation.clj:273-279), so on a degenerate tick with a NON-empty + # conv (we are past the `if not self.proj` empty short-circuit above) + # the Clojure graph still clusters at k=2 and feeds this_k=2 to the + # group-k smoother, ADVANCING its {last_k, last_k_count, smoothed_k} + # state. Mirror that in legacy mode (silhouette sentinel 0.0 -> this_k=2) + # instead of FREEZING the smoother memory — which self-corrected within + # <=4 ticks but diverged from Clojure meanwhile. Improved mode carries + # no smoother state, so it is unaffected. + if legacy_mode: + new_smoother_state, _ = group_k_smoother_update( + prev_group_k_smoother or {}, {2: 0.0}) + self.group_k_smoother = new_smoother_state + logger.info(f"Legacy degenerate-tick smoother advance: " + f"state={new_smoother_state}") return # Prepare base cluster centers and weights @@ -823,78 +924,122 @@ def _compute_clusters(self, logger.info(f"Computing group clusters with k range 2-{max_k}...") - # Try different k values and compute silhouette scores - best_k = 2 - best_score = -1 - group_clusterings = {} - - for k in range(2, max_k + 1): - group_labels, group_centers, group_member_lists = kmeans_sklearn( - base_centers_array, - k=k, - max_iters=GROUP_ITERS, - weights=base_weights - ) - - # Calculate silhouette score - score = calculate_silhouette_sklearn(base_centers_array, group_labels) - group_clusterings[k] = (group_labels, group_centers, group_member_lists, score) - - logger.info(f" k={k}: silhouette={score:.4f}") - - if score > best_score: - best_score = score - best_k = k - - logger.info(f"Selected k={best_k} with silhouette={best_score:.4f}") - - # 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} + if legacy_mode: + # PR-C: group-level warm start with lineage + weighted recentering. + # Clojure clusters the BASE-CLUSTER CENTERS (base-clusters-proj), + # weighted by base-cluster member counts (:weights base-clusters- + # weights, conversation.clj:433-445), warm-starting each per-k + # clustering from the prior tick's k-clustering (:last-clusters + # (last-clusterings k), conversation.clj:441). + # + # Clojure passes :cluster-iters (a key kmeans does NOT destructure, + # clusters.clj:303), so the group level actually runs kmeans' DEFAULT + # max-iters (20), NOT :group-iters (100). We reproduce that + # (GROUP_LEGACY_ITERS below); well-separated data converges long + # before either bound, so on real conversations it is inert. + base_ids = [c['id'] for c in base_clusters] + base_weights_by_id = {c['id']: len(c['members']) for c in base_clusters} + group_data = _LegacyNamedData(base_ids, base_centers_array) + prev_gc = prev_group_clusterings or {} + + legacy_group_clusterings: Dict[int, List[Dict[str, Any]]] = {} + silhouettes_by_k: Dict[int, float] = {} + for k in range(2, max_k + 1): + gc = legacy_kmeans( + group_data, k, + last_clusters=prev_gc.get(k), + weights=base_weights_by_id, + max_iters=GROUP_LEGACY_ITERS) + gc.sort(key=lambda c: c['id']) # Clojure sort-by :id (conversation.clj:437) + legacy_group_clusterings[k] = gc + # Score with the SAME silhouette the improved path uses, on the + # legacy assignment, so the smoother sees comparable numbers. + labels = _labels_from_id_clusters(base_ids, gc) + score = calculate_silhouette_sklearn(base_centers_array, labels) + silhouettes_by_k[k] = score + logger.info(f" k={k}: silhouette={score:.4f}") + + # Group-K smoother (PR-D): damps K flicker (K only switches after + # :group-k-buffer=4 consecutive ticks agree) with Clojure's max-key + # HIGHER-k-wins tie-break, threading {last_k, last_k_count, + # smoothed_k}. self.group_clusterings holds the id-carrying cluster + # dicts (legacy value type) — the warm start read next tick. new_smoother_state, selected_k = group_k_smoother_update( prev_group_k_smoother or {}, silhouettes_by_k) - self.group_clusterings = group_clusterings + self.group_clusterings = legacy_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}") + logger.info(f"Legacy group-k-smoother: smoothed_k={selected_k} " + f"state={new_smoother_state}") + + # Build production-form group_clusters from the selected clustering. + # Members are base-cluster ids; ids carry the group-cluster lineage. + selected = legacy_group_clusterings[selected_k] + group_clusters = [ + {'id': c['id'], + 'center': np.asarray(c['center'], dtype=float).tolist(), + 'members': list(c['members'])} + for c in selected + ] + group_clusters.sort(key=lambda c: c['id']) else: - selected_k = best_k + # Improved (default): cold recompute + best_k selection, byte-for-byte + # unchanged. Clear any legacy warm-start state a prior clojure-legacy + # tick may have left on this instance: improved mode is stateless + # across ticks (no stale memory retained after a mode switch), and a + # later switch back to legacy warm-starts cold — same as a fresh + # Clojure worker boot. No-op in pure improved runs (both init to {}). + self.group_clusterings = {} + self.group_k_smoother = {} + best_k = 2 + best_score = -1 + group_clusterings = {} + + for k in range(2, max_k + 1): + group_labels, group_centers, group_member_lists = kmeans_sklearn( + base_centers_array, + k=k, + max_iters=GROUP_ITERS, + weights=base_weights + ) - # 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 = [] - for cluster_id, (center, member_indices) in enumerate(zip(group_centers, group_member_lists)): - # Members are base cluster IDs (not participant IDs!) - member_base_cluster_ids = [base_clusters[idx]['id'] for idx in member_indices] - group_clusters.append({ - 'id': cluster_id, - 'center': center.tolist(), - 'members': member_base_cluster_ids - }) + # Calculate silhouette score + score = calculate_silhouette_sklearn(base_centers_array, group_labels) + group_clusterings[k] = (group_labels, group_centers, group_member_lists, score) + + logger.info(f" k={k}: silhouette={score:.4f}") + + if score > best_score: + best_score = score + best_k = k + + logger.info(f"Selected k={best_k} with silhouette={best_score:.4f}") + + selected_k = best_k - # Keep group clusters in k-means ID order (matching Clojure's - # sort-by :id, conversation.clj:437). Do NOT sort by size or - # reassign IDs: Clojure assigns group ids by first-k-distinct - # encounter order over base-cluster centers (init-clusters, - # clusters.clj:55-64) and never re-orders by size. The former - # size-descending re-sort here was the root cause of the gid 0↔1 - # label swap vs Clojure blobs (S3-4 trace, 2026-06-11: identical - # memberships modulo label permutation on vw-cold_start). Mirrors - # the identical rule at the base-cluster level above. - group_clusters.sort(key=lambda c: c['id']) + # Use the selected clustering. 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 = [] + for cluster_id, (center, member_indices) in enumerate(zip(group_centers, group_member_lists)): + # Members are base cluster IDs (not participant IDs!) + member_base_cluster_ids = [base_clusters[idx]['id'] for idx in member_indices] + group_clusters.append({ + 'id': cluster_id, + 'center': center.tolist(), + 'members': member_base_cluster_ids + }) + + # Keep group clusters in k-means ID order (matching Clojure's + # sort-by :id, conversation.clj:437). Do NOT sort by size or + # reassign IDs: Clojure assigns group ids by first-k-distinct + # encounter order over base-cluster centers (init-clusters, + # clusters.clj:55-64) and never re-orders by size. The former + # size-descending re-sort here was the root cause of the gid 0↔1 + # label swap vs Clojure blobs (S3-4 trace, 2026-06-11: identical + # memberships modulo label permutation on vw-cold_start). Mirrors + # the identical rule at the base-cluster level above. + group_clusters.sort(key=lambda c: c['id']) logger.info(f"Created {len(group_clusters)} group clusters") @@ -1194,6 +1339,7 @@ def recompute(self) -> 'Conversation': # In 'improved' mode (default) these are IGNORED and behavior is # unchanged; only 'clojure-legacy' mode consumes them. prev_pca = result.pca + prev_base_clusters = getattr(result, 'base_clusters', []) prev_group_clusterings = getattr(result, 'group_clusterings', {}) prev_group_k_smoother = getattr(result, 'group_k_smoother', {}) @@ -1202,6 +1348,7 @@ def recompute(self) -> 'Conversation': # Compute clusters result._compute_clusters( + prev_base_clusters=prev_base_clusters, prev_group_clusterings=prev_group_clusterings, prev_group_k_smoother=prev_group_k_smoother, ) diff --git a/delphi/polismath/pca_kmeans_rep/legacy_kmeans.py b/delphi/polismath/pca_kmeans_rep/legacy_kmeans.py new file mode 100644 index 000000000..3fbfe6e9f --- /dev/null +++ b/delphi/polismath/pca_kmeans_rep/legacy_kmeans.py @@ -0,0 +1,468 @@ +""" +Faithful port of the Clojure ``polismath.math.clusters`` k-means WITH lineage. + +This is the warm-start k-means Clojure actually threads across conv-update ticks +(``:last-clusters (:base-clusters conv)`` -> ``kmeans`` -> ``clean-start-clusters``, +math/src/polismath/math/conversation.clj:403-410 and clusters.clj:301-312). Its +defining property is CLUSTER-IDENTITY LINEAGE: cluster ids are stable across +ticks, new ids strictly increase, merges keep the larger side's id, and vanished +members are dropped. Base-cluster ids feed group-level clustering and the +serialized blob, so lineage propagates downstream in sequential runs. + +This is a DIFFERENT algorithm from the off-production ``clusters.py`` warm start +(split-largest / merge-closest, clusters.py:302-364), which is NOT a port of the +Clojure ``clean-start-clusters``. That module is intentionally left untouched; +this one is the faithful port and is wired only into the ``clojure-legacy`` +engine mode (see ``polismath.utils.engine_mode``). + +Data model (mirrors Clojure's named-matrix + cluster maps): + + - A clustering is a ``list`` of ``dict`` clusters ``{'id': int, + 'members': list, 'center': np.ndarray (1-D float)}``. ``members`` are ROW + NAMES (participant ids at base level; base-cluster ids at group level), + exactly as Clojure's ``:members`` hold row names of the named matrix. + - Input data is a ``_NamedData(row_names, matrix)`` pair: ``matrix[i]`` is the + row vector for ``row_names[i]``. This reproduces named-matrix lookups + (``get-row-by-name``, ``rowname-subset``) that key clusters to the current + data by NAME — the mechanism that lets a prior tick's members be matched + against (or dropped from) the current tick's rows. + - ``weights`` is either ``None`` (base level) or a ``dict`` mapping row name -> + weight (group level, ``:weights base-clusters-weights``, + conversation.clj:444; the weight of a base cluster is its member count). + +Every public function cites the Clojure source it ports. Clojure is +authoritative; where a Clojure quirk is load-bearing it is reproduced and +flagged in the docstring. +""" + +from typing import Any, Dict, List, Mapping, Optional, Sequence + +import numpy as np + +# Reuse the EXACT first-k-distinct helper the production cold path uses +# (clusters.py:551) so that a COLD legacy clustering initialises from the same +# seed rows as ``kmeans_sklearn``'s ``use_first_k_init`` branch. Sharing this is +# what keeps the base-level cold-start invariant tight (see module tests). +from polismath.pca_kmeans_rep.clusters import _get_first_k_distinct_centers + +# Clojure ``same-clustering?`` default tolerance (clusters.clj:71). +SAME_CLUSTERING_THRESHOLD = 0.01 +# Clojure ``kmeans`` default ``max-iters`` (clusters.clj:303). +DEFAULT_MAX_ITERS = 20 + + +class _NamedData: + """A named matrix: row names aligned 1:1 with rows of a float matrix. + + Reproduces the subset of ``polismath.math.named-matrix`` used by k-means: + ``rownames`` (order-preserving), ``get-row-by-name`` (named_matrix.clj:268), + and membership (used to emulate ``safe-rowname-subset``'s drop-missing + behaviour, named_matrix.clj:258-265). + """ + + def __init__(self, row_names: Sequence[Any], matrix: np.ndarray): + self.row_names: List[Any] = list(row_names) + self.matrix: np.ndarray = np.asarray(matrix, dtype=float) + if self.matrix.ndim != 2 or self.matrix.shape[0] != len(self.row_names): + raise ValueError( + "matrix must be 2-D with one row per name " + f"(got shape {self.matrix.shape} for {len(self.row_names)} names)") + # Last-write-wins on duplicate names would corrupt lookups; Clojure's + # index-hash also de-dups names, but our callers pass unique names. + self._by_name: Dict[Any, np.ndarray] = { + name: self.matrix[i] for i, name in enumerate(self.row_names)} + + def get_row(self, name: Any) -> np.ndarray: + """Row vector for ``name`` (Clojure ``get-row-by-name``).""" + return self._by_name[name] + + def __contains__(self, name: Any) -> bool: + return name in self._by_name + + def n_distinct_rows(self) -> int: + """Count of distinct rows (Clojure ``(count (distinct (matrix/rows ...)))`` + used for ``possible-clusters``, clusters.clj:249). NaN-safe to match the + production first-k-distinct helper.""" + distinct: List[np.ndarray] = [] + for row in self.matrix: + if not any(np.array_equal(row, u, equal_nan=True) for u in distinct): + distinct.append(row) + return len(distinct) + + +def _euclidean(a: np.ndarray, b: np.ndarray) -> float: + """``matrix/distance`` (L2). Clojure uses core.matrix euclidean distance.""" + return float(np.linalg.norm(np.asarray(a, dtype=float) - np.asarray(b, dtype=float))) + + +def weighted_mean(rows: Sequence[np.ndarray], + weights: Optional[Sequence[float]] = None) -> np.ndarray: + """Mean (or weighted mean) of row vectors — Clojure ``weighted-mean`` + (clusters.clj:89-126, matrix branch). + + Clojure computes ``(count w)/(sum w) * sum_i(w_i * row_i)`` then takes the + plain per-row mean, which algebraically equals ``sum_i(w_i row_i)/sum_i(w_i)`` + = ``np.average(rows, weights=w, axis=0)``. Unweighted -> arithmetic mean. + """ + arr = np.asarray(rows, dtype=float) + if weights is None: + return np.mean(arr, axis=0) + return np.average(arr, axis=0, weights=np.asarray(weights, dtype=float)) + + +def _cluster_weights(members: Sequence[Any], + hm_weights: Optional[Mapping[Any, float]]) -> Optional[List[float]]: + """Per-member weight seq for a cluster — Clojure ``cluster-weights`` + (clusters.clj:133-139). ``None`` when ``hm_weights`` is falsey.""" + if not hm_weights: + return None + return [hm_weights[m] for m in members] + + +def init_clusters(data: _NamedData, k: int) -> List[Dict[str, Any]]: + """First ``k`` distinct rows in encounter order, ids ``0..k-1``, empty members. + + Port of Clojure ``init-clusters`` (clusters.clj:55-65). Reuses + ``_get_first_k_distinct_centers`` (clusters.py:551) so the seed rows are + byte-identical to the production cold path's init. May return fewer than + ``k`` clusters when the data has fewer than ``k`` distinct rows (``take k`` + semantics). + """ + centers = _get_first_k_distinct_centers(data.matrix, k) + return [ + {'id': i, 'members': [], 'center': np.asarray(center, dtype=float)} + for i, center in enumerate(centers) + ] + + +def same_clustering(clusters1: List[Dict[str, Any]], + clusters2: List[Dict[str, Any]], + threshold: float = SAME_CLUSTERING_THRESHOLD) -> bool: + """Whether two clusterings' SORTED centers are pairwise within ``threshold``. + + Port of Clojure ``same-clustering?`` (clusters.clj:68-76). Two Clojure + quirks are reproduced deliberately: + + - Centers are SORTED (order-independent comparison). Clojure sorts vectors + with ``compare`` = lexicographic; we sort by the center tuple. + - Clojure zips the two sorted center seqs with ``utils/zip``, which is + ``interleave``-based and TRUNCATES to the shorter seq (utils.clj:78-83). + So it does NOT require equal lengths — if one clustering has fewer + clusters, only the common prefix of sorted centers is compared. We + replicate that (``zip`` truncation), rather than the stricter + ``len != len -> False`` used elsewhere (clusters.py:133). + """ + c1 = sorted((np.asarray(c['center'], dtype=float) for c in clusters1), + key=lambda v: tuple(v.tolist())) + c2 = sorted((np.asarray(c['center'], dtype=float) for c in clusters2), + key=lambda v: tuple(v.tolist())) + return all(_euclidean(x, y) < threshold for x, y in zip(c1, c2)) + + +def cluster_step(data: _NamedData, + clusters: List[Dict[str, Any]], + weights: Optional[Mapping[Any, float]] = None) -> List[Dict[str, Any]]: + """One Lloyd step: reassign every row to its nearest center, drop empty + clusters, recenter. + + Port of Clojure ``cluster-step`` (clusters.clj:142-158): + + 1. Clear members (keep id + center) — ``cleared-clusters``. + 2. ``reduce add-to-closest`` over the rows in row order: each row joins the + nearest cluster by ``matrix/distance`` to its center. Ties resolve to + the LATER cluster in the current cluster order (Clojure ``min-key`` + returns the last of equal-keyed args, clusters.clj:44-52). Exact ties + are measure-zero on real float projections; the rule is fixed for + reproducibility and to match Clojure's array-map order for k<=8. + 3. Drop clusters that received no members (``filter > 0``). k can shrink. + 4. Recenter each surviving cluster on the rows it captured, weighted by + ``cluster-weights`` (clusters.clj:154-158). + + Cluster ORDER of the result follows the input cluster order (non-empty + only). Clojure's ``(into {} ...)`` is an array-map for <=8 clusters + (insertion/id order) but a hash-map for >8 (hash order); the only observable + effect of order is the assignment tie-break above, so this deterministic + order matches Clojure except on measure-zero exact ties in large clusterings. + """ + n = len(clusters) + if n == 0: + return [] + centers = [np.asarray(c['center'], dtype=float) for c in clusters] + members: List[List[Any]] = [[] for _ in range(n)] + positions: List[List[np.ndarray]] = [[] for _ in range(n)] + + for name, row in zip(data.row_names, data.matrix): + best_idx = 0 + best_dist = _euclidean(row, centers[0]) + for j in range(1, n): + d = _euclidean(row, centers[j]) + # ``<=`` => ties go to the LATER cluster (Clojure min-key semantics). + if d <= best_dist: + best_dist = d + best_idx = j + members[best_idx].append(name) + positions[best_idx].append(row) + + out: List[Dict[str, Any]] = [] + for j in range(n): + if not members[j]: + continue # drop empty cluster + w = _cluster_weights(members[j], weights) + out.append({ + 'id': clusters[j]['id'], + 'members': members[j], + 'center': weighted_mean(positions[j], w), + }) + return out + + +def _recenter_center(data: _NamedData, + members: Sequence[Any], + weights: Optional[Mapping[Any, float]]) -> Optional[np.ndarray]: + """Center from members that still exist in ``data`` (weighted). Returns + ``None`` if no member survives — the caller decides drop-vs-keep. + + Shared core of Clojure ``recenter-clusters`` / ``safe-recenter-clusters`` + (clusters.clj:161-191): both subset members to those present in the current + data (``rowname-subset`` / ``safe-rowname-subset`` drop missing names, + named_matrix.clj:135-141, 258-265) and take the weighted mean. + """ + surviving = [m for m in members if m in data] + if not surviving: + return None + rows = [data.get_row(m) for m in surviving] + w = _cluster_weights(surviving, weights) + return weighted_mean(rows, w) + + +def safe_recenter_clusters(data: _NamedData, + clusters: List[Dict[str, Any]], + weights: Optional[Mapping[Any, float]] = None) -> List[Dict[str, Any]]: + """Recenter each cluster on its surviving members; DROP clusters whose + members all vanished; if EVERY cluster vanishes, fall back to one big + cluster. + + Port of Clojure ``safe-recenter-clusters`` (clusters.clj:171-191). + + - Only ``:center`` is updated; ``:members`` keep their full prior list + (vanished names included). They are re-subset on every later recenter + and flushed by the first ``cluster-step`` in the k-means loop, so the + FINAL clustering never carries a vanished member (Clojure identical). + - Fallback id is ``(inc (apply max -1 (map :id clusters)))`` over the + ORIGINAL clusters (clusters.clj:188) — ``-1`` floor makes it 0 when + empty. + """ + out: List[Dict[str, Any]] = [] + for clst in clusters: + center = _recenter_center(data, clst['members'], weights) + if center is None: + continue # all members vanished -> drop (nil, removed) + out.append({'id': clst['id'], 'members': list(clst['members']), 'center': center}) + + if not out: + # Everything vanished: one cluster of all current rows (clusters.clj:187-190). + max_id = max((c['id'] for c in clusters), default=-1) + return [{ + 'id': max_id + 1, + 'members': list(data.row_names), + 'center': _recenter_center(data, data.row_names, weights), + }] + return out + + +def recenter_clusters(data: _NamedData, + clusters: List[Dict[str, Any]], + weights: Optional[Mapping[Any, float]] = None) -> List[Dict[str, Any]]: + """Recenter each cluster on its surviving members (no dropping). + + Port of Clojure ``recenter-clusters`` (clusters.clj:161-168). If a cluster's + members have all vanished mid-loop (only reachable in a degenerate split + edge), its center is kept unchanged rather than becoming NaN — a defensive, + idempotent belt on a measure-zero path that Clojure never exercises on real + data (most-distal never extracts a singleton's only point; :dist would be 0). + """ + out: List[Dict[str, Any]] = [] + for clst in clusters: + center = _recenter_center(data, clst['members'], weights) + if center is None: + out.append({'id': clst['id'], 'members': list(clst['members']), + 'center': np.asarray(clst['center'], dtype=float)}) + else: + out.append({'id': clst['id'], 'members': list(clst['members']), 'center': center}) + return out + + +def merge_clusters(clst1: Dict[str, Any], clst2: Dict[str, Any]) -> Dict[str, Any]: + """Merge two clusters, keeping the LARGER cluster's id. + + Port of Clojure ``merge-clusters`` (clusters.clj:194-199): + + - ``new-id`` = id of ``(max-key #(count (:members %)) clst1 clst2)``. On a + member-count TIE, Clojure ``max-key`` returns the LAST arg, i.e. + ``clst2`` — reproduced here. + - members concatenated (``clst1`` then ``clst2``). + - center = size-weighted mean of the two centers (weights = member counts). + """ + n1, n2 = len(clst1['members']), len(clst2['members']) + new_id = clst1['id'] if n1 > n2 else clst2['id'] # tie -> clst2 (max-key last) + return { + 'id': new_id, + 'members': list(clst1['members']) + list(clst2['members']), + 'center': weighted_mean( + [np.asarray(clst1['center'], dtype=float), + np.asarray(clst2['center'], dtype=float)], + weights=[n1, n2]), + } + + +def uniqify_clusters(clusters: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """Merge clusters that have IDENTICAL centers. + + Port of Clojure ``uniqify-clusters`` (clusters.clj:220-227): fold left; for + each cluster, if an already-accumulated cluster has an exactly-equal center, + ``merge-clusters`` the two in place (at the incumbent's position); else + append. Center equality is exact (``=`` on vectors) — reproduced with + ``np.array_equal``. + """ + acc: List[Dict[str, Any]] = [] + for clst in clusters: + match_idx = None + for i, existing in enumerate(acc): + if np.array_equal(np.asarray(existing['center'], dtype=float), + np.asarray(clst['center'], dtype=float)): + match_idx = i + break + if match_idx is not None: + acc[match_idx] = merge_clusters(acc[match_idx], clst) + else: + acc.append(clst) + return acc + + +def most_distal(data: _NamedData, clusters: List[Dict[str, Any]]) -> Dict[str, Any]: + """The data point whose distance to its NEAREST center is greatest. + + Port of Clojure ``most-distal`` (clusters.clj:202-217). For each row: find + ``(min over clusters of (distance, cluster-id))`` — its nearest center. Then + across rows take the ``max`` by that distance. Tie behaviour mirrors Clojure: + + - inner ``min-key`` on distance -> nearest cluster ties resolve to the + LATER cluster in ``clusters`` order; + - outer ``max-key`` on distance -> farthest row ties resolve to the LATER + row in ``data`` row order. + + Returns ``{'dist', 'clst_id', 'id'}`` where ``id`` is the row name. + """ + best_dist = None + best_clst_id = None + best_name = None + for name, row in zip(data.row_names, data.matrix): + # nearest cluster (ties -> later cluster) + near_dist = _euclidean(row, np.asarray(clusters[0]['center'], dtype=float)) + near_id = clusters[0]['id'] + for clst in clusters[1:]: + d = _euclidean(row, np.asarray(clst['center'], dtype=float)) + if d <= near_dist: + near_dist = d + near_id = clst['id'] + # farthest row (ties -> later row) + if best_dist is None or near_dist >= best_dist: + best_dist = near_dist + best_clst_id = near_id + best_name = name + return {'dist': best_dist, 'clst_id': best_clst_id, 'id': best_name} + + +def clean_start_clusters(data: _NamedData, + clusters: List[Dict[str, Any]], + k: int, + weights: Optional[Mapping[Any, float]] = None) -> List[Dict[str, Any]]: + """Prepare a prior clustering as the seed for a new k-means round. + + Port of Clojure ``clean-start-clusters`` (clusters.clj:230-277). Three + phases when prior clusters exist: + + 1. ``safe-recenter-clusters`` — recenter on surviving members, drop dead + clusters, big-cluster fallback if all die. + 2. ``uniqify-clusters`` — merge identical-center clusters. + 3. Split loop — while ``min(k, #distinct-rows) > #clusters``: recenter, + find the most-distal row; if its distance > 0, pull it out into a NEW + singleton cluster with id ``(inc (max ids))`` and repeat; else stop. + + With no prior clusters, defers to ``init-clusters`` (the warm path is never + used to build from scratch, clusters.clj:274-277). + """ + if not clusters: + return init_clusters(data, k) + + clusters = safe_recenter_clusters(data, clusters, weights) + clusters = uniqify_clusters(clusters) + possible = min(k, data.n_distinct_rows()) + + while True: + clusters = recenter_clusters(data, clusters, weights) + if possible <= len(clusters): + return clusters + outlier = most_distal(data, clusters) + if outlier['dist'] is None or outlier['dist'] <= 0: + return clusters + outlier_id = outlier['id'] + # Remove the outlier from whichever cluster(s) hold it. + clusters = [ + {'id': c['id'], + 'members': [m for m in c['members'] if m != outlier_id], + 'center': c['center']} + for c in clusters + ] + new_id = max(c['id'] for c in clusters) + 1 # (inc (max ids)) + clusters = clusters + [{ + 'id': new_id, + 'members': [outlier_id], + 'center': np.asarray(data.get_row(outlier_id), dtype=float), + }] + + +def kmeans(data: _NamedData, + k: int, + last_clusters: Optional[List[Dict[str, Any]]] = None, + weights: Optional[Mapping[Any, float]] = None, + max_iters: int = DEFAULT_MAX_ITERS) -> List[Dict[str, Any]]: + """K-means with lineage — Clojure ``kmeans`` (clusters.clj:301-312). + + Seed = ``clean-start-clusters`` when ``last_clusters`` is given (warm start + with id lineage), else ``init-clusters`` (cold, first-k-distinct). Then + iterate ``cluster-step`` until ``same-clustering?`` or ``max_iters`` is + exhausted. Clojure ALWAYS runs at least one ``cluster-step`` (the ``(= iter + 0)`` check happens AFTER computing ``new-clusters``), so ``max_iters=0`` + still performs a single reassignment. + + Args: + data: ``_NamedData`` — rows keyed by name (pids at base level, + base-cluster ids at group level). + k: target cluster count (``base-k``=100 at base level; 2..max-k at group + level). + last_clusters: the previous tick's clustering (id-carrying dicts) or + ``None`` for a cold start. + weights: ``None`` (base) or ``{name: weight}`` (group, + ``base-clusters-weights``). + max_iters: iteration cap. Clojure passes ``:base-iters``=100 at base + level; at group level it passes the MISNAMED ``:cluster-iters`` key + which ``kmeans`` ignores, so the group level actually runs the + default (see ``DEFAULT_MAX_ITERS`` = 20). Callers must pass the value + the Clojure code EFFECTIVELY uses. + + Returns: + List of cluster dicts ``{'id', 'members', 'center'}``. NOT sorted — the + caller applies ``sort-by :id`` (conversation.clj:406, 437). + """ + if data.matrix.shape[0] == 0: + return [] + clusters = (clean_start_clusters(data, last_clusters, k, weights) + if last_clusters else init_clusters(data, k)) + iters = max_iters + while True: + new_clusters = cluster_step(data, clusters, weights) + if iters == 0 or same_clustering(clusters, new_clusters): + return new_clusters + clusters = new_clusters + iters -= 1 diff --git a/delphi/tests/test_base_cluster_lineage.py b/delphi/tests/test_base_cluster_lineage.py new file mode 100644 index 000000000..dc701340a --- /dev/null +++ b/delphi/tests/test_base_cluster_lineage.py @@ -0,0 +1,237 @@ +#!/usr/bin/env python3 +""" +Integration tests for base-cluster lineage + warm start in 'clojure-legacy' +mode (PR-C). + +Clojure threads the previous tick's clusters back into k-means as +``:last-clusters`` at BOTH levels (conversation.clj:403-410 base, +conversation.clj:433-445 group), giving clusters STABLE ids across ticks. The +pre-PR Python legacy branch recomputed clusters COLD every tick (kmeans_sklearn +with no warm start), so no lineage was threaded. This module verifies: + + 1. Cold first tick: legacy base + group partitions equal improved mode + (the cold-start invariant — measured identical on vw and synthetic data). + 2. Warm-start threading: in legacy mode the ported legacy_kmeans is called + with the prior tick's clusters as last_clusters (base and per-k group); in + improved mode it is never called. + 3. self.group_clusterings holds id-carrying cluster dicts (legacy value type), + not the (labels, centers, member_lists, silhouette) tuple. + 4. Base-cluster ids are stable across chained update_votes and new + participants receive strictly larger ids (lineage). +""" + +import os +import sys + +import numpy as np +import pytest + +sys.path.append(os.path.abspath(os.path.dirname(__file__))) + +import polismath.conversation.conversation as conv_mod +from polismath.conversation.conversation import Conversation +from polismath.utils.engine_mode import ENGINE_MODE_ENV_VAR +from polismath.pca_kmeans_rep.pca import PCA_IMPL_ENV_VAR + + +def _many_ptpt_votes(n_ptpts=18, n_cmnts=8): + """18 distinct ternary vote rows (3 group signatures + unique bits) -> base + k-means yields singleton base clusters and group clusterings for k in {2,3} + (mirrors the fixture in test_group_k_smoother.py).""" + 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} + + +def _partition(clusters): + return sorted(sorted(str(m) for m in c['members']) for c in clusters) + + +def _pid_to_base_id(conv): + return {str(m): c['id'] for c in conv.base_clusters for m in c['members']} + + +# --------------------------------------------------------------------------- +# 1. Cold-start invariance gate (base + group) +# --------------------------------------------------------------------------- + +class TestColdStartInvariance: + + def _run(self, monkeypatch, mode): + monkeypatch.delenv(PCA_IMPL_ENV_VAR, raising=False) # default (powerit) both modes + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, mode) + return Conversation('cold').update_votes(_many_ptpt_votes()) + + def test_legacy_base_is_clojure_faithful_singletons(self, monkeypatch): + # base-k (=100) >= n_ptpts, so every DISTINCT projection becomes its own + # base cluster. This synthetic set has near-duplicate projections; legacy + # (init-clusters on exact-distinct rows + cluster-step) keeps each point + # as its own singleton with NO empty clusters, matching Clojure exactly. + # sklearn's Lloyd instead collapses a near-duplicate pair and leaves an + # empty cluster — so legacy and improved legitimately DIVERGE on + # near-duplicate projections (base cold identity holds only when all + # projections are distinct, e.g. vw; see TestVwColdStartInvariance). This + # test pins the Clojure-faithful legacy side. + leg = self._run(monkeypatch, 'clojure-legacy') + assert all(c['members'] for c in leg.base_clusters) # no empty clusters + assert all(len(c['members']) == 1 for c in leg.base_clusters) # singletons + + def test_group_clustering_is_deterministic_in_legacy(self, monkeypatch): + # NOTE (semantic finding): the GROUP level runs real k-means (k<= 2 # one per k in {2,3} + assert all(c['last_is_none'] for c in group_calls) # tick 1: all cold + + spy.calls.clear() + conv.update_votes({'votes': [{'pid': 'p0', 'tid': 'c0', 'vote': 1.0}]}) + group_calls = [c for c in spy.calls if c['level'] == 'group'] + assert len(group_calls) >= 2 + # tick 2: each per-k group clustering warm-started from tick-1's k-clustering + assert all(c['last_is_none'] is False for c in group_calls) + + def test_group_clusterings_are_id_carrying_dicts(self, monkeypatch): + monkeypatch.delenv(PCA_IMPL_ENV_VAR, raising=False) + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, 'clojure-legacy') + conv = Conversation('x').update_votes(_many_ptpt_votes()) + assert set(conv.group_clusterings.keys()) == {2, 3} + for k, clustering in conv.group_clusterings.items(): + assert isinstance(clustering, list) + for c in clustering: + assert set(c.keys()) >= {'id', 'members', 'center'} + + +# --------------------------------------------------------------------------- +# 4. Base-cluster id lineage across ticks +# --------------------------------------------------------------------------- + +class TestBaseIdLineage: + + def _votes(self, indexed_pids, n_cmnts=6): + """Each (global_index, pid) votes on ALL comments (so threshold + min(7,n)=n qualifies everyone), with a signature keyed on the GLOBAL + index so every pid is distinct -> singleton base clusters.""" + votes = [] + for idx, pid in indexed_pids: + for j in range(n_cmnts): + v = 1.0 if ((idx >> j) & 1) else -1.0 + votes.append({'pid': pid, 'tid': f'c{j}', 'vote': v}) + return {'votes': votes} + + def test_ids_stable_and_new_participant_gets_larger_id(self, monkeypatch): + monkeypatch.delenv(PCA_IMPL_ENV_VAR, raising=False) + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, 'clojure-legacy') + + conv = Conversation('lineage').update_votes( + self._votes([(i, f'p{i}') for i in range(5)])) + m1 = _pid_to_base_id(conv) + assert len(m1) == 5 # 5 singleton base clusters + + # Add one brand-new participant p5 with a distinct signature (index 5). + conv = conv.update_votes(self._votes([(5, 'p5')])) + m2 = _pid_to_base_id(conv) + + # Existing participants keep their base-cluster ids (lineage). + for pid in [f'p{i}' for i in range(5)]: + assert m2[pid] == m1[pid], (pid, m1[pid], m2.get(pid)) + # The new participant gets a strictly larger id (new lineage id). + assert m2['p5'] > max(m1.values()) + + +# --------------------------------------------------------------------------- +# 5. vw real-data cold-start invariance (base AND group), the documented gate. +# --------------------------------------------------------------------------- + +class TestVwColdStartInvariance: + """On vw (67 in-conv participants -> 67 singleton base clusters; groups for + k=2..5), cold legacy clustering was measured bit-identical to improved mode + at BOTH levels. Locked in here; skips if the committed vw dataset is absent. + """ + + def _vw_conv(self, monkeypatch, mode): + try: + from polismath.replay.real_data import load_export_votes + ds = load_export_votes('vw') + except (ImportError, FileNotFoundError): + pytest.skip('vw dataset unavailable') + monkeypatch.delenv(PCA_IMPL_ENV_VAR, raising=False) + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, mode) + votes = [{'pid': v.pid, 'tid': v.tid, 'vote': v.sign, 'created': v.t_ms} + for v in ds.votes] + return Conversation('vw').update_votes({'votes': votes}) + + def test_vw_base_and_group_identical_across_modes(self, monkeypatch): + imp = self._vw_conv(monkeypatch, 'improved') + leg = self._vw_conv(monkeypatch, 'clojure-legacy') + assert _partition(imp.base_clusters) == _partition(leg.base_clusters) + assert _partition(imp.group_clusters) == _partition(leg.group_clusters) diff --git a/delphi/tests/test_engine_mode.py b/delphi/tests/test_engine_mode.py index 2600d3823..56c39fb6f 100644 --- a/delphi/tests/test_engine_mode.py +++ b/delphi/tests/test_engine_mode.py @@ -8,13 +8,23 @@ - 'clojure-legacy' : threads warm-start state across ticks, matching Clojure (PCA :start-vectors, group-k-smoother). -On the FIRST tick (cold start) the two modes MUST coincide bit-for-bit, because -Clojure's warm-start state is empty on the first tick (no previous comps, no -smoother state). This module asserts: +On the FIRST tick (cold start) the two modes MUST coincide, because Clojure's +warm-start state is empty on the first tick (no previous comps, no smoother +state). This module asserts: 1. Flag resolution semantics (default, valid, invalid, case/whitespace, read-at-call-time) — mirrors tests/test_powerit_pca.py::TestPcaImplFlag. 2. Cold-start invariance: a single-shot vw pipeline run is identical under - both modes (guards commits 2 and 3 from diverging on the first tick). + both modes (guards the warm-start commits from diverging on the first tick). + + Since PR-C the legacy clustering computes cluster CENTERS via the ported + Clojure weighted-mean (np.average) instead of sklearn's centroid. On vw the + cold clustering STRUCTURE is bit-identical across modes (same base/group + memberships, ids, counts, and all downstream repness/priorities/group-votes) + but the center COORDINATES differ at floating-point precision (~1e-13: + e.g. group center y 2.0147429038868094 vs 2.01474290388681). The invariance + check therefore compares numbers with a tight tolerance and everything else + (ids, memberships, strings) exactly — a real structural regression (a moved + participant, a relabelled cluster) still fails. """ import os @@ -90,10 +100,56 @@ def _strip_volatile(d): return d +# Tolerance for cluster-center coordinates (see module docstring): PR-C's ported +# weighted-mean and sklearn's centroid agree to ~1e-13 on identical memberships; +# 1e-6 is far below any real structural divergence yet absorbs the float noise. +_COLD_IDENTITY_TOL = 1e-6 + + +def _almost_equal(a, b, path='', tol=_COLD_IDENTITY_TOL): + """Deep equality that tolerates float noise in numbers but is EXACT on + everything else (dict keys, list lengths, strings, ints such as cluster ids). + + Returns (ok, message). + """ + # bool is an int subclass — treat it as exact, not numeric-tolerant. + if isinstance(a, bool) or isinstance(b, bool): + return (a == b, f"{path}: {a!r} != {b!r}") + # int-vs-int compares EXACTLY (ids, counts): a relative tolerance would + # accept e.g. two large cluster ids that differ. Mixed int/float (0 vs 0.0 + # from a JSON round-trip) still takes the tolerant branch below. + if isinstance(a, int) and isinstance(b, int): + return (a == b, f"{path}: {a!r} != {b!r} (int exact)") + if isinstance(a, (int, float)) and isinstance(b, (int, float)): + if abs(float(a) - float(b)) <= tol * max(1.0, abs(a), abs(b)): + return (True, '') + return (False, f"{path}: {a!r} != {b!r} (>|tol|)") + if isinstance(a, dict) and isinstance(b, dict): + if set(a) != set(b): + return (False, f"{path}: dict keys differ {set(a) ^ set(b)}") + for k in a: + ok, msg = _almost_equal(a[k], b[k], f"{path}.{k}", tol) + if not ok: + return (ok, msg) + return (True, '') + if isinstance(a, (list, tuple)) and isinstance(b, (list, tuple)): + if len(a) != len(b): + return (False, f"{path}: length {len(a)} != {len(b)}") + for i, (x, y) in enumerate(zip(a, b)): + ok, msg = _almost_equal(x, y, f"{path}[{i}]", tol) + if not ok: + return (ok, msg) + return (True, '') + return (a == b, f"{path}: {a!r} != {b!r}") + + class TestColdStartInvariance: - """A single-shot (first-tick) pipeline run must be byte-identical under - both engine modes. This is the hard gate protecting 'improved' mode from - any drift introduced by the legacy warm-start plumbing.""" + """A single-shot (first-tick) pipeline run must be identical under both + engine modes UP TO floating-point cluster-center coordinates (see module + docstring). Structure — every id, membership, count, and all downstream + outputs — must be bit-identical. This is the hard gate protecting 'improved' + mode from any structural drift introduced by the legacy warm-start plumbing. + """ def _recompute_to_dict(self, monkeypatch, mode): from common_utils import create_test_conversation @@ -107,8 +163,30 @@ def _recompute_to_dict(self, monkeypatch, mode): def test_vw_cold_run_identical_across_modes(self, monkeypatch): improved = self._recompute_to_dict(monkeypatch, ENGINE_MODE_IMPROVED) legacy = self._recompute_to_dict(monkeypatch, ENGINE_MODE_LEGACY) - assert improved == legacy, ( - "Cold-start (first-tick) vw run diverged between 'improved' and " - "'clojure-legacy' engine modes; warm-start state must be empty on " - "tick 1 so the two modes must coincide bit-for-bit." + ok, msg = _almost_equal(improved, legacy) + assert ok, ( + "Cold-start (first-tick) vw run diverged STRUCTURALLY between " + "'improved' and 'clojure-legacy' engine modes (beyond float-level " + f"cluster centers): {msg}" ) + + def test_vw_cold_structure_bit_identical_ignoring_centers(self, monkeypatch): + """Belt-and-braces: with cluster CENTER coordinates dropped, the two cold + blobs are EXACTLY equal — proving the ~1e-13 divergence is confined to + center coordinates and nothing structural moved.""" + improved = _drop_centers(self._recompute_to_dict(monkeypatch, ENGINE_MODE_IMPROVED)) + legacy = _drop_centers(self._recompute_to_dict(monkeypatch, ENGINE_MODE_LEGACY)) + assert improved == legacy + + +def _drop_centers(d): + """Recursively drop cluster-center coordinate fields ('center', 'x', 'y') + so the remaining structure (ids, members, counts, downstream) is compared + exactly. base-clusters are folded to {id, members, x, y, count}; group + clusters carry {id, members, center}.""" + if isinstance(d, dict): + return {k: _drop_centers(v) for k, v in d.items() + if k not in ('center', 'x', 'y')} + if isinstance(d, list): + return [_drop_centers(v) for v in d] + return d diff --git a/delphi/tests/test_group_k_smoother.py b/delphi/tests/test_group_k_smoother.py index 992288587..de5681075 100644 --- a/delphi/tests/test_group_k_smoother.py +++ b/delphi/tests/test_group_k_smoother.py @@ -195,6 +195,20 @@ def test_legacy_no_flicker_then_switch_after_four(self, monkeypatch): # group_clusters is picked from the smoothed k and is never None/empty. assert conv.group_clusters, "group_clusters must be populated" + def test_mode_switch_to_improved_clears_legacy_state(self, monkeypatch): + """After a clojure-legacy tick populated the warm-start state, a tick + under improved mode must CLEAR it (not silently retain stale memory).""" + prefs = [2, 2] + self._setup(monkeypatch, 'clojure-legacy', prefs) + conv = Conversation('smooth').update_votes(_many_ptpt_votes()) + assert conv.group_clusterings and conv.group_k_smoother + + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, 'improved') + conv = conv.update_votes(_REPEAT_VOTE) + assert conv.group_k_smoother == {} + assert conv.group_clusterings == {} + 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) @@ -204,3 +218,45 @@ def test_improved_mode_leaves_smoother_inert(self, monkeypatch): assert conv.group_clusterings == {} # But still produces group clusters via the untouched best_k path. assert conv.group_clusters, "group_clusters must be populated" + + +def _degenerate_votes(n_ptpts=20, n_cmts=8): + """All participants vote identically -> a single base cluster (degenerate).""" + return {'votes': [{'pid': f'p{i}', 'tid': f'c{t}', 'vote': 1.0} + for i in range(n_ptpts) for t in range(n_cmts)]} + + +class TestDegenerateTickSmoother: + """P6a: on a <2-base-cluster degenerate tick with a NON-empty conv, Clojure's + max-k-fn is still >= 2 (conversation.clj:273-279), so its graph feeds this_k=2 + to the group-k smoother and ADVANCES it. Legacy mode must mirror that instead + of freezing the smoother memory. Improved mode carries no smoother state.""" + + def _mode(self, monkeypatch, mode): + monkeypatch.delenv(PCA_IMPL_ENV_VAR, raising=False) + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, mode) + + def test_legacy_degenerate_tick_advances_smoother(self, monkeypatch): + self._mode(monkeypatch, 'clojure-legacy') + conv = Conversation('deg').update_votes(_degenerate_votes()) + assert len(conv.base_clusters) < 2, "scenario must be degenerate" + # Smoother ADVANCED (this_k=2), not frozen at {}. + assert conv.group_k_smoother.get('last_k') == 2 + assert conv.group_k_smoother.get('smoothed_k') == 2 + assert conv.group_k_smoother.get('last_k_count') == 1 + + def test_legacy_degenerate_tick_accumulates_count_across_ticks(self, monkeypatch): + self._mode(monkeypatch, 'clojure-legacy') + conv = Conversation('deg').update_votes(_degenerate_votes()) + # A second still-degenerate tick keeps this_k=2 -> consecutive count grows + # (this is precisely the smoother advance Clojure performs each tick). + conv = conv.update_votes({'votes': [{'pid': 'p0', 'tid': 'c0', 'vote': 1.0}]}) + assert len(conv.base_clusters) < 2 + assert conv.group_k_smoother.get('last_k') == 2 + assert conv.group_k_smoother.get('last_k_count') == 2 + + def test_improved_degenerate_tick_leaves_smoother_inert(self, monkeypatch): + self._mode(monkeypatch, 'improved') + conv = Conversation('deg').update_votes(_degenerate_votes()) + assert len(conv.base_clusters) < 2 + assert conv.group_k_smoother == {} # improved carries no smoother state diff --git a/delphi/tests/test_legacy_kmeans.py b/delphi/tests/test_legacy_kmeans.py new file mode 100644 index 000000000..18e178636 --- /dev/null +++ b/delphi/tests/test_legacy_kmeans.py @@ -0,0 +1,292 @@ +#!/usr/bin/env python3 +""" +Unit tests for the faithful Clojure k-means port (PR-C, legacy_kmeans.py). + +Every expected value is hand-derived from the Clojure rules in +math/src/polismath/math/clusters.clj (cited per test), on tiny synthetic +matrices — NOT recomputed from the code under test. Covers the lineage +semantics that make this a DIFFERENT algorithm from clusters.py's warm start: +first-k-distinct cold init, drop-vanished, (inc max-id) new ids, +merge-keeps-larger-id, identical-center merge, most-distal split, weighted +group-level recentering, and stable ids across a warm-start chain. +""" + +import os +import sys + +import numpy as np +import pytest + +sys.path.append(os.path.abspath(os.path.dirname(__file__))) + +from polismath.pca_kmeans_rep.legacy_kmeans import ( + _NamedData, + weighted_mean, + init_clusters, + same_clustering, + cluster_step, + safe_recenter_clusters, + recenter_clusters, + merge_clusters, + uniqify_clusters, + most_distal, + clean_start_clusters, + kmeans, +) + + +def _nd(names, rows): + return _NamedData(names, np.array(rows, dtype=float)) + + +def _by_id(clusters): + return {c['id']: c for c in clusters} + + +# --------------------------------------------------------------------------- +# weighted_mean (clusters.clj:89-126) +# --------------------------------------------------------------------------- + +class TestWeightedMean: + def test_unweighted_is_arithmetic_mean(self): + m = weighted_mean([[0.0, 0.0], [2.0, 4.0]]) + np.testing.assert_allclose(m, [1.0, 2.0]) + + def test_weighted_is_sum_w_row_over_sum_w(self): + # (1*[0,0] + 3*[3,0]) / 4 = [9/4, 0] + m = weighted_mean([[0.0, 0.0], [3.0, 0.0]], weights=[1, 3]) + np.testing.assert_allclose(m, [2.25, 0.0]) + + +# --------------------------------------------------------------------------- +# init_clusters (clusters.clj:55-65) +# --------------------------------------------------------------------------- + +class TestInitClusters: + def test_first_k_distinct_encounter_order(self): + data = _nd(['a', 'b', 'c', 'd'], [[0, 0], [1, 1], [0, 0], [2, 2]]) + clusters = init_clusters(data, 3) + # Distinct rows in encounter order: [0,0], [1,1], [2,2] -> ids 0,1,2. + assert [c['id'] for c in clusters] == [0, 1, 2] + np.testing.assert_allclose(clusters[0]['center'], [0, 0]) + np.testing.assert_allclose(clusters[1]['center'], [1, 1]) + np.testing.assert_allclose(clusters[2]['center'], [2, 2]) + assert all(c['members'] == [] for c in clusters) + + def test_fewer_distinct_than_k(self): + data = _nd(['a', 'b', 'c'], [[0, 0], [0, 0], [1, 1]]) + clusters = init_clusters(data, 5) + assert [c['id'] for c in clusters] == [0, 1] # only 2 distinct rows + + +# --------------------------------------------------------------------------- +# same_clustering (clusters.clj:68-76) — sorted centers, zip-truncation +# --------------------------------------------------------------------------- + +class TestSameClustering: + def test_true_when_centers_match_within_threshold(self): + a = [{'id': 0, 'members': [], 'center': np.array([0.0, 0.0])}, + {'id': 1, 'members': [], 'center': np.array([5.0, 5.0])}] + b = [{'id': 9, 'members': [], 'center': np.array([5.001, 5.0])}, + {'id': 8, 'members': [], 'center': np.array([0.0, 0.0])}] + assert same_clustering(a, b) is True # sorted centers, <0.01 apart + + def test_false_when_a_center_moved(self): + a = [{'id': 0, 'members': [], 'center': np.array([0.0, 0.0])}] + b = [{'id': 0, 'members': [], 'center': np.array([0.5, 0.0])}] + assert same_clustering(a, b) is False + + def test_zip_truncates_to_shorter(self): + # Clojure utils/zip is interleave-based -> truncates; only the common + # prefix of SORTED centers is compared (clusters.clj:72-76, utils.clj:78). + a = [{'id': 0, 'members': [], 'center': np.array([0.0, 0.0])}] + b = [{'id': 0, 'members': [], 'center': np.array([0.0, 0.0])}, + {'id': 1, 'members': [], 'center': np.array([9.0, 9.0])}] + assert same_clustering(a, b) is True # extra cluster in b ignored + + +# --------------------------------------------------------------------------- +# cluster_step (clusters.clj:142-158) — assign, drop empty, recenter +# --------------------------------------------------------------------------- + +class TestClusterStep: + def test_assign_drop_empty_and_recenter(self): + data = _nd(['a', 'b', 'c', 'd'], [[0, 0], [0, 1], [10, 10], [10, 11]]) + clusters = init_clusters(data, 2) # centers [0,0], [0,1] + stepped = cluster_step(data, clusters) + by = _by_id(stepped) + # a->c0 (dist 0); b->c1 (dist 0); c,d closer to c1 -> c1 gets b,c,d. + assert set(by[0]['members']) == {'a'} + assert set(by[1]['members']) == {'b', 'c', 'd'} + np.testing.assert_allclose(by[0]['center'], [0, 0]) + np.testing.assert_allclose(by[1]['center'], [20 / 3, 22 / 3]) + + def test_empty_cluster_is_dropped(self): + # Two init centers, but all points identical -> one cluster empties out. + data = _nd(['a', 'b'], [[0, 0], [0, 0]]) + clusters = [{'id': 0, 'members': [], 'center': np.array([0.0, 0.0])}, + {'id': 1, 'members': [], 'center': np.array([9.0, 9.0])}] + stepped = cluster_step(data, clusters) + assert [c['id'] for c in stepped] == [0] # id 1 got no members, dropped + + def test_weighted_recentering(self): + # Group-level style: weights by name. Two points assigned to one cluster. + data = _nd([0, 1], [[0.0, 0.0], [0.0, 2.0]]) + clusters = [{'id': 7, 'members': [], 'center': np.array([0.0, 1.0])}] + stepped = cluster_step(data, clusters, weights={0: 1, 1: 3}) + # weighted mean y = (1*0 + 3*2)/4 = 1.5 (vs unweighted 1.0) + np.testing.assert_allclose(stepped[0]['center'], [0.0, 1.5]) + + +# --------------------------------------------------------------------------- +# safe_recenter_clusters (clusters.clj:171-191) — drop vanished +# --------------------------------------------------------------------------- + +class TestSafeRecenter: + def test_drops_cluster_whose_members_all_vanished(self): + clusters = [ + {'id': 0, 'members': ['a', 'b'], 'center': np.array([0.0, 0.5])}, + {'id': 1, 'members': ['c', 'd'], 'center': np.array([10.0, 10.5])}, + ] + # New data: c and d are gone; a, b remain; e is new. + data = _nd(['a', 'b', 'e'], [[0, 0], [0, 1], [5, 5]]) + out = safe_recenter_clusters(data, clusters) + assert [c['id'] for c in out] == [0] # cluster 1 dropped + np.testing.assert_allclose(out[0]['center'], [0.0, 0.5]) + + def test_all_vanished_fallback_one_big_cluster_inc_max_id(self): + clusters = [{'id': 4, 'members': ['x'], 'center': np.array([0.0, 0.0])}] + data = _nd(['y', 'z'], [[1, 1], [3, 3]]) # x gone + out = safe_recenter_clusters(data, clusters) + assert len(out) == 1 + assert out[0]['id'] == 5 # (inc (max 4)) + assert set(out[0]['members']) == {'y', 'z'} + np.testing.assert_allclose(out[0]['center'], [2.0, 2.0]) + + +# --------------------------------------------------------------------------- +# merge_clusters / uniqify_clusters (clusters.clj:194-227) +# --------------------------------------------------------------------------- + +class TestMerge: + def test_merge_keeps_larger_id_and_weighted_center(self): + big = {'id': 3, 'members': ['a', 'a2'], 'center': np.array([1.0, 1.0])} + small = {'id': 8, 'members': ['b'], 'center': np.array([4.0, 4.0])} + merged = merge_clusters(big, small) + assert merged['id'] == 3 # larger member count keeps its id + assert merged['members'] == ['a', 'a2', 'b'] + # weighted by counts: (2*[1,1] + 1*[4,4]) / 3 = [2,2] + np.testing.assert_allclose(merged['center'], [2.0, 2.0]) + + def test_merge_tie_keeps_second_arg_id(self): + c1 = {'id': 3, 'members': ['a'], 'center': np.array([0.0, 0.0])} + c2 = {'id': 8, 'members': ['b'], 'center': np.array([2.0, 2.0])} + merged = merge_clusters(c1, c2) + # Clojure max-key returns the LAST of equal-keyed args -> c2's id. + assert merged['id'] == 8 + + def test_uniqify_merges_identical_centers_keeps_larger(self): + clusters = [ + {'id': 0, 'members': ['a', 'a2'], 'center': np.array([1.0, 1.0])}, + {'id': 1, 'members': ['b'], 'center': np.array([1.0, 1.0])}, + {'id': 2, 'members': ['c'], 'center': np.array([9.0, 9.0])}, + ] + out = uniqify_clusters(clusters) + by = _by_id(out) + assert set(by.keys()) == {0, 2} # 0 and 1 merged, 1's id gone (0 larger) + assert by[0]['members'] == ['a', 'a2', 'b'] + + +# --------------------------------------------------------------------------- +# most_distal (clusters.clj:202-217) +# --------------------------------------------------------------------------- + +class TestMostDistal: + def test_farthest_point_from_nearest_center(self): + clusters = [{'id': 0, 'members': ['a', 'b'], 'center': np.array([0.0, 0.0])}] + data = _nd(['a', 'b', 'c'], [[0, 0], [0, 0], [3, 4]]) + out = most_distal(data, clusters) + assert out['id'] == 'c' + assert out['clst_id'] == 0 + assert out['dist'] == pytest.approx(5.0) + + +# --------------------------------------------------------------------------- +# clean_start_clusters (clusters.clj:230-277) — split loop, new ids +# --------------------------------------------------------------------------- + +class TestCleanStart: + def test_split_creates_new_cluster_with_inc_max_id(self): + # One surviving cluster (id 5) + a distal new point -> split to 2. + clusters = [{'id': 5, 'members': ['a'], 'center': np.array([0.0, 0.0])}] + data = _nd(['a', 'z'], [[0, 0], [9, 9]]) + out = clean_start_clusters(data, clusters, k=2) + by = _by_id(out) + assert set(by.keys()) == {5, 6} # new cluster id = inc(max(5)) + assert by[6]['members'] == ['z'] + np.testing.assert_allclose(by[6]['center'], [9, 9]) + + def test_no_split_when_enough_clusters(self): + clusters = [ + {'id': 0, 'members': ['a'], 'center': np.array([0.0, 0.0])}, + {'id': 1, 'members': ['b'], 'center': np.array([9.0, 9.0])}, + ] + data = _nd(['a', 'b'], [[0, 0], [9, 9]]) + out = clean_start_clusters(data, clusters, k=2) + assert {c['id'] for c in out} == {0, 1} # already at possible=2 + + +# --------------------------------------------------------------------------- +# kmeans end-to-end (clusters.clj:301-312) +# --------------------------------------------------------------------------- + +class TestKmeansEndToEnd: + def test_cold_two_well_separated_groups(self): + data = _nd(['a', 'b', 'c', 'd'], [[0, 0], [0, 1], [10, 10], [10, 11]]) + out = kmeans(data, k=2, max_iters=100) + by = _by_id(out) + assert set(by[0]['members']) == {'a', 'b'} + assert set(by[1]['members']) == {'c', 'd'} + # Compare center vectors AS-IS: sorting coordinates would mask an + # x/y axis swap. + np.testing.assert_allclose(by[0]['center'], [0.0, 0.5]) + np.testing.assert_allclose(by[1]['center'], [10.0, 10.5]) + + def test_cold_singletons_when_k_equals_n_distinct(self): + # k == n distinct rows -> each point its own cluster, ids by encounter. + data = _nd(['a', 'b', 'c'], [[0, 0], [5, 5], [9, 1]]) + out = kmeans(data, k=3, max_iters=100) + by = _by_id(out) + assert by[0]['members'] == ['a'] + assert by[1]['members'] == ['b'] + assert by[2]['members'] == ['c'] + + def test_warm_start_preserves_ids_adds_new_participant(self): + # Tick 1: two groups -> ids {0,1}. Tick 2: same points + a far new point, + # k bumped to 3. Old ids 0,1 persist; the new group gets a strictly + # larger id (lineage). + d1 = _nd(['a', 'b', 'c', 'd'], [[0, 0], [0, 1], [10, 10], [10, 11]]) + t1 = kmeans(d1, k=2, max_iters=100) + assert {c['id'] for c in t1} == {0, 1} + + d2 = _nd(['a', 'b', 'c', 'd', 'e'], + [[0, 0], [0, 1], [10, 10], [10, 11], [100, 100]]) + t2 = kmeans(d2, k=3, last_clusters=t1, max_iters=100) + ids = {c['id'] for c in t2} + assert {0, 1}.issubset(ids) # lineage preserved + assert max(ids) >= 2 # new cluster id strictly larger + by = _by_id(t2) + # 'e' is the lone far point -> its own new cluster. + e_cluster = next(c for c in t2 if 'e' in c['members']) + assert e_cluster['id'] >= 2 + assert e_cluster['members'] == ['e'] + + def test_warm_start_weighted_group_level(self): + # Group-level cold k-means with member-count weights; weighted mean must + # pull the c0 center toward the heavier member (clusters.clj:154-158). + data = _nd([0, 1, 2], [[0.0, 0.0], [0.0, 2.0], [10.0, 10.0]]) + out = kmeans(data, k=2, weights={0: 1, 1: 3, 2: 1}, max_iters=100) + by = _by_id(out) + c0 = next(c for c in out if set(c['members']) == {0, 1}) + # weighted center y = (1*0 + 3*2)/4 = 1.5, NOT unweighted 1.0 + np.testing.assert_allclose(c0['center'], [0.0, 1.5])