diff --git a/delphi/polismath/conversation/conversation.py b/delphi/polismath/conversation/conversation.py index dc2a65a1e..a837a446f 100644 --- a/delphi/polismath/conversation/conversation.py +++ 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 diff --git a/delphi/polismath/pca_kmeans_rep/pca.py b/delphi/polismath/pca_kmeans_rep/pca.py index 1b217bfca..04bc8016b 100644 --- a/delphi/polismath/pca_kmeans_rep/pca.py +++ b/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) diff --git a/delphi/tests/test_pca_warm_start.py b/delphi/tests/test_pca_warm_start.py new file mode 100644 index 000000000..ff3e7389a --- /dev/null +++ b/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'])