python-math #6: feat(math): PCA warm start in clojure-legacy engine mode (PR-B)

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.utils.engine_mode import resolve_engine_mode, ENGINE_MODE_LEGACY
 
 
 # Configure logging
@@ -356,6 +357,11 @@ def update_votes(self,
         # Column order: natsort is fine — column permutation doesn't affect PCA
         # eigenvalues/vectors (only reorders the component loadings), so it has
         # no effect on clustering k.
+        # NB: in clojure-legacy mode this column order is now LOAD-BEARING for
+        # PCA warm-start alignment — the previous tick's component loadings are
+        # threaded in positionally, so the ordering must be STABLE tick-to-tick.
+        # Safe while tids are append-only (natsort keeps prior columns' relative
+        # order and appends new ones); revisit if columns can ever be removed.
         all_cols = natsorted(existing_cols.union(new_cols))
 
         logger.info(f"[{time.time() - start_time:.2f}s] Found {len(new_rows)} new rows and {len(new_cols)} new columns")
@@ -615,7 +621,30 @@ def _compute_pca(self, n_components: int = 2,
             # Make a clean copy of the rating matrix
             clean_matrix = self._get_clean_matrix()
 
-            pca_results, proj_dict = pca_project_dataframe(clean_matrix, n_components)
+            # Engine-mode warm start (PR-B). In 'clojure-legacy' mode we thread
+            # the previous tick's unit components back in as the power-iteration
+            # start vectors (Clojure :start-vectors, conversation.clj:385) and
+            # require the power-iteration solver (sklearn cannot inject start
+            # vectors). In the default 'improved' mode nothing changes:
+            # start_vectors stays None and the solver is chosen purely by
+            # POLISMATH_PCA_IMPL, so this call is byte-identical to the pre-PR
+            # behavior.
+            start_vectors = None
+            require_powerit = False
+            if resolve_engine_mode() == ENGINE_MODE_LEGACY:
+                require_powerit = True
+                if prev_pca is not None and prev_pca.get('comps') is not None:
+                    # Only warm-start from real components; a missing/None
+                    # 'comps' (np.asarray(None) would be a size-1 object array,
+                    # a garbage seed) or empty/cold state (first tick) falls
+                    # through to the cold random draw.
+                    prev_comps = np.asarray(prev_pca['comps'])
+                    if prev_comps.size > 0:
+                        start_vectors = prev_comps
+
+            pca_results, proj_dict = pca_project_dataframe(
+                clean_matrix, n_components,
+                start_vectors=start_vectors, require_powerit=require_powerit)
 
             # Store results
             self.pca = pca_results

delphi/polismath/pca_kmeans_rep/pca.py

@@ -238,7 +238,10 @@ def powerit_pca(matrix: np.ndarray,
 
 
 def pca_project_dataframe(df: pd.DataFrame,
-                         n_comps: int = 2) -> Tuple[Dict[str, np.ndarray], Dict[str, np.ndarray]]:
+                         n_comps: int = 2,
+                         start_vectors: Optional[Sequence[np.ndarray]] = None,
+                         require_powerit: bool = False,
+                         ) -> Tuple[Dict[str, np.ndarray], Dict[str, np.ndarray]]:
     """
     Perform PCA on a DataFrame and project participants into PCA space.
 
@@ -251,6 +254,18 @@ def pca_project_dataframe(df: pd.DataFrame,
         df: DataFrame with participants as rows and comments as columns.
             Values are votes (float); NaN indicates missing/unseen.
         n_comps: Number of principal components to compute.
+        start_vectors: Optional per-component power-iteration warm start. In
+            Clojure-legacy engine mode this is the PREVIOUS tick's unit
+            components (Clojure :start-vectors, conversation.clj:385 ->
+            powerit-pca, pca.clj:98). `None` (the default) is the cold path and
+            is BYTE-IDENTICAL to the pre-PR behavior. Shorter-than-current start
+            vectors are 1-padded for new comments inside `_power_iteration`
+            (pca.clj:46-49). Only consumed by the power-iteration solver.
+        require_powerit: When True the caller mandates the power-iteration
+            solver (it is the only one that can be seeded). If
+            POLISMATH_PCA_IMPL=sklearn is set anyway, we warn and fall back to
+            power iteration rather than silently drop the warm start. `False`
+            (default) preserves the pre-PR solver-selection behavior exactly.
 
     Returns:
         Tuple of (pca_results, proj_dict) where:
@@ -330,6 +345,24 @@ def pca_project_dataframe(df: pd.DataFrame,
     # only the eigen-solver differs.
     impl = _resolve_impl_flag(PCA_IMPL_ENV_VAR, PCA_IMPL_DEFAULT, PCA_IMPL_CHOICES)
 
+    # Warm-start parity (PR-B): power iteration is the ONLY solver that can be
+    # seeded with the previous tick's components (Clojure :start-vectors,
+    # conversation.clj:385 -> pca.clj:98). sklearn's SVD has no start-vector
+    # hook, so when a warm start is supplied - or explicitly required by the
+    # 'clojure-legacy' engine mode - override POLISMATH_PCA_IMPL=sklearn back to
+    # powerit and warn. Running sklearn here would silently drop the warm start.
+    # When require_powerit / start_vectors are both absent (improved mode), this
+    # is a no-op and solver selection is exactly the pre-PR behavior.
+    if (require_powerit or start_vectors is not None) and impl == PCA_IMPL_SKLEARN:
+        logger.warning(
+            "%s=sklearn %s; falling back to the power-iteration PCA for the "
+            "Clojure-legacy path.",
+            PCA_IMPL_ENV_VAR,
+            "cannot inject the provided warm-start vectors"
+            if start_vectors is not None
+            else "cannot satisfy require_powerit (cold tick, no warm-start vectors)")
+        impl = PCA_IMPL_POWERIT
+
     # Perform PCA with error handling
     # TODO(julien): use function that compute projections and PCAs in one pass.
     try:
@@ -347,7 +380,10 @@ def pca_project_dataframe(df: pd.DataFrame,
             # Legacy/Clojure-parity solver (default). Comps are unit vectors;
             # projections are (X - center) @ compsᵀ, exactly like sklearn's
             # fit_transform convention.
-            pca_results = powerit_pca(matrix_data_no_nan, n_comps=n_comps)
+            # start_vectors warm-starts each component's power iteration
+            # (None == cold == pre-PR behavior; see the PR-B note above).
+            pca_results = powerit_pca(matrix_data_no_nan, n_comps=n_comps,
+                                      start_vectors=start_vectors)
             projections = ((matrix_data_no_nan - pca_results['center'])
                            @ pca_results['comps'].T)
 

delphi/tests/test_pca_warm_start.py

@@ -0,0 +1,227 @@
+#!/usr/bin/env python3
+"""
+Tests for PCA warm-start threading in 'clojure-legacy' engine mode (PR-B).
+
+Clojure warm-starts the power-iteration PCA with the PREVIOUS tick's
+post-normalization unit components (conversation.clj:381-387 passes
+:start-vectors (get-in conv [:pca :comps]) into powerit-pca, pca.clj:86-105).
+Python already supports start_vectors in powerit_pca, but no production caller
+passed them — every tick ran cold. This module verifies:
+
+  1. pca_project_dataframe threads start_vectors into powerit_pca, and refuses
+     to run sklearn when warm-start vectors are required (sklearn cannot inject
+     start vectors) — it warns and falls back to power iteration.
+  2. In 'clojure-legacy' mode, a second recompute tick feeds tick-1's comps to
+     powerit_pca as start_vectors; in 'improved' mode it stays None (cold).
+  3. Warm-started tick-2 comps stay close in angle to tick-1 (reduced jitter).
+  4. The cold FIRST tick is identical across the two modes (no prev state).
+"""
+
+import os
+import sys
+
+import numpy as np
+import pytest
+
+sys.path.append(os.path.abspath(os.path.dirname(__file__)))
+
+import polismath.pca_kmeans_rep.pca as pca_mod
+from polismath.pca_kmeans_rep.pca import (
+    pca_project_dataframe,
+    PCA_IMPL_ENV_VAR,
+)
+from polismath.utils.engine_mode import ENGINE_MODE_ENV_VAR
+from polismath.conversation.conversation import Conversation
+
+
+# ---------------------------------------------------------------------------
+# Synthetic two-group data helpers
+# ---------------------------------------------------------------------------
+
+def _two_group_votes(pids, tids, group_a):
+    """Group A agrees on the first half of tids, disagrees on the second half;
+    group B is the mirror image. Gives a clean 1-D PCA separation."""
+    votes = []
+    half = len(tids) // 2
+    for pid in pids:
+        in_a = pid in group_a
+        for j, tid in enumerate(tids):
+            first_half = j < half
+            # A: +1 on first half, -1 on second; B: mirror.
+            v = 1.0 if (first_half == in_a) else -1.0
+            votes.append({'pid': pid, 'tid': tid, 'vote': v})
+    return {'votes': votes}
+
+
+def _spy_powerit(monkeypatch):
+    """Wrap pca.powerit_pca to record the start_vectors of every call while
+    still running the real computation."""
+    recorded = []
+    real = pca_mod.powerit_pca
+
+    def spy(matrix, n_comps=2, iters=100, start_vectors=None):
+        recorded.append(start_vectors)
+        return real(matrix, n_comps=n_comps, iters=iters, start_vectors=start_vectors)
+
+    monkeypatch.setattr(pca_mod, 'powerit_pca', spy)
+    return recorded
+
+
+def _angle_deg(u, v):
+    u = np.asarray(u, dtype=float)
+    v = np.asarray(v, dtype=float)
+    c = np.dot(u, v) / (np.linalg.norm(u) * np.linalg.norm(v) + 1e-300)
+    return np.degrees(np.arccos(np.clip(abs(c), -1.0, 1.0)))
+
+
+# ---------------------------------------------------------------------------
+# 1. pca_project_dataframe: start_vectors threading + sklearn conflict
+# ---------------------------------------------------------------------------
+
+class TestDataframeStartVectors:
+
+    def _df(self):
+        import pandas as pd
+        rng = np.random.default_rng(0)
+        # 8 ptpts x 5 comments, two clear blocks.
+        block = np.vstack([np.ones((4, 5)), -np.ones((4, 5))])
+        block[:, 2] *= -1  # break perfect collinearity a bit
+        noise = rng.normal(scale=0.01, size=block.shape)
+        return pd.DataFrame(block + noise,
+                            index=[f'p{i}' for i in range(8)],
+                            columns=[f'c{j}' for j in range(5)])
+
+    def test_start_vectors_reach_powerit(self, monkeypatch):
+        monkeypatch.delenv(PCA_IMPL_ENV_VAR, raising=False)
+        recorded = _spy_powerit(monkeypatch)
+        df = self._df()
+        sv = np.ones((2, 5))
+        pca_project_dataframe(df, n_comps=2, start_vectors=sv)
+        assert len(recorded) == 1
+        assert recorded[0] is not None
+        np.testing.assert_array_equal(np.asarray(recorded[0]), sv)
+
+    def test_default_start_vectors_none(self, monkeypatch):
+        monkeypatch.delenv(PCA_IMPL_ENV_VAR, raising=False)
+        recorded = _spy_powerit(monkeypatch)
+        df = self._df()
+        pca_project_dataframe(df, n_comps=2)
+        assert recorded == [None]
+
+    def test_require_powerit_overrides_sklearn_with_warning(self, monkeypatch, caplog):
+        """When warm-start vectors are supplied but POLISMATH_PCA_IMPL=sklearn,
+        the solver must fall back to power iteration (sklearn cannot inject a
+        start vector) and log a warning."""
+        monkeypatch.setenv(PCA_IMPL_ENV_VAR, 'sklearn')
+        recorded = _spy_powerit(monkeypatch)
+        df = self._df()
+        sv = np.ones((2, 5))
+        with caplog.at_level('WARNING'):
+            pca_project_dataframe(df, n_comps=2, start_vectors=sv, require_powerit=True)
+        # powerit was actually called (sklearn branch would not touch it)
+        assert len(recorded) == 1
+        assert recorded[0] is not None
+        # ... and the fallback was announced (sklearn cannot inject warm start).
+        assert any(r.levelname == 'WARNING' and 'cannot inject the provided warm-start' in r.getMessage()
+                   for r in caplog.records)
+
+    def test_require_powerit_cold_warning_wording(self, monkeypatch, caplog):
+        """Same sklearn conflict on a COLD tick (require_powerit=True, no
+        start vectors): the warning must not claim vectors were supplied."""
+        monkeypatch.setenv(PCA_IMPL_ENV_VAR, 'sklearn')
+        recorded = _spy_powerit(monkeypatch)
+        df = self._df()
+        with caplog.at_level('WARNING'):
+            pca_project_dataframe(df, n_comps=2, require_powerit=True)
+        assert len(recorded) == 1
+        assert recorded[0] is None
+        msgs = [r.getMessage() for r in caplog.records if r.levelname == 'WARNING']
+        assert any('require_powerit' in m for m in msgs)
+        assert not any('provided warm-start' in m for m in msgs)
+
+    def test_improved_path_byte_identical(self, monkeypatch):
+        """No start_vectors + no require_powerit == exactly the pre-PR behavior."""
+        monkeypatch.delenv(PCA_IMPL_ENV_VAR, raising=False)
+        df = self._df()
+        base, _ = pca_project_dataframe(df, n_comps=2)
+        same, _ = pca_project_dataframe(df, n_comps=2, start_vectors=None,
+                                        require_powerit=False)
+        np.testing.assert_array_equal(base['comps'], same['comps'])
+        np.testing.assert_array_equal(base['center'], same['center'])
+
+
+# ---------------------------------------------------------------------------
+# 2/3/4. Two-tick chained recompute through the Conversation pipeline
+# ---------------------------------------------------------------------------
+
+class TestChainedWarmStart:
+
+    PIDS = [f'p{i}' for i in range(6)]
+    TIDS = [f'c{j}' for j in range(4)]
+    GROUP_A = {'p0', 'p1', 'p2'}
+
+    def _tick1(self):
+        return _two_group_votes(self.PIDS, self.TIDS, self.GROUP_A)
+
+    def _tick2_new_ptpt(self):
+        # A new participant votes on the SAME comments (no new column), so the
+        # warm-start vectors line up 1:1 with the current column set.
+        return _two_group_votes(['p6'], self.TIDS, self.GROUP_A)
+
+    def _run_two_ticks(self, monkeypatch, mode):
+        monkeypatch.delenv(PCA_IMPL_ENV_VAR, raising=False)
+        monkeypatch.setenv(ENGINE_MODE_ENV_VAR, mode)
+        recorded = _spy_powerit(monkeypatch)
+        conv0 = Conversation('warm')
+        conv1 = conv0.update_votes(self._tick1())
+        conv2 = conv1.update_votes(self._tick2_new_ptpt())
+        return conv1, conv2, recorded
+
+    def test_legacy_tick2_receives_tick1_comps(self, monkeypatch):
+        conv1, conv2, recorded = self._run_two_ticks(monkeypatch, 'clojure-legacy')
+        assert len(recorded) == 2
+        # Tick 1 is cold (no previous comps).
+        assert recorded[0] is None
+        # Tick 2 warm-starts from tick-1's comps.
+        assert recorded[1] is not None
+        np.testing.assert_allclose(np.asarray(recorded[1]),
+                                   np.asarray(conv1.pca['comps']))
+
+    def test_improved_tick2_receives_none(self, monkeypatch):
+        conv1, conv2, recorded = self._run_two_ticks(monkeypatch, 'improved')
+        assert len(recorded) == 2
+        assert recorded[0] is None
+        assert recorded[1] is None  # cold recompute every tick
+
+    def test_legacy_warm_comps_close_in_angle(self, monkeypatch):
+        conv1, conv2, _ = self._run_two_ticks(monkeypatch, 'clojure-legacy')
+        # Same column set across ticks, so comps are directly comparable.
+        # Two perfectly-separable groups are rank-1, so PC2 is a degenerate
+        # zero vector — skip components with ~no variance in either tick.
+        comps1 = np.asarray(conv1.pca['comps'])
+        comps2 = np.asarray(conv2.pca['comps'])
+        checked = 0
+        for i in range(len(comps1)):
+            if np.linalg.norm(comps1[i]) > 1e-8 and np.linalg.norm(comps2[i]) > 1e-8:
+                assert _angle_deg(comps1[i], comps2[i]) < 15.0
+                checked += 1
+        assert checked >= 1, "no non-degenerate component to compare"
+
+    def test_legacy_prev_pca_without_comps_falls_back_to_cold(self, monkeypatch):
+        """A prev_pca whose 'comps' is missing/None must NOT be turned into a
+        np.asarray(None) garbage seed — it falls back to the cold draw."""
+        monkeypatch.delenv(PCA_IMPL_ENV_VAR, raising=False)
+        monkeypatch.setenv(ENGINE_MODE_ENV_VAR, 'clojure-legacy')
+        recorded = _spy_powerit(monkeypatch)
+        conv = Conversation('warm').update_votes(self._tick1())
+        for degenerate in ({'center': None, 'comps': None}, {}):
+            conv._compute_pca(prev_pca=degenerate)
+            assert recorded[-1] is None, (
+                f"prev_pca={degenerate!r} must cold-start, not seed powerit"
+            )
+
+    def test_cold_first_tick_identical_across_modes(self, monkeypatch):
+        conv1_imp, _, _ = self._run_two_ticks(monkeypatch, 'improved')
+        conv1_leg, _, _ = self._run_two_ticks(monkeypatch, 'clojure-legacy')
+        np.testing.assert_array_equal(conv1_imp.pca['comps'], conv1_leg.pca['comps'])
+        np.testing.assert_array_equal(conv1_imp.pca['center'], conv1_leg.pca['center'])