Codex CI R5 P0: thread raw data (not working_data) on MPD Conley path

P0 Methodology [new in R5] — `MultiPeriodDiD.fit` built
`_conley_coords_arr` / `_conley_time_arr` / `_conley_unit_arr` from
`working_data`, which is absorb-demeaned. If a caller listed any of
those columns (coords, time, or unit) in `absorb`, the Conley helper
would silently partition the within-period spatial sandwich on
residualized values instead of the true geography/period labels and
misstate Conley SEs.

This is the same raw-data contract the local R1 fix established for
DifferenceInDifferences (`estimators.py:L538-L561`); MultiPeriodDiD
shared the same pattern but wasn't updated in that pass — codex CI R5
caught the holistic-fix omission.

Now reads from `data` in MPD, mirroring DiD + TWFE
(`twfe.py:L370-L371`). FWL composability holds: the meat is computed
on demeaned scores but the kernel grid uses raw coords + time/unit.

Adds `test_mpd_conley_with_absorb_uses_raw_coords_and_time` regression:
monkeypatches `_compute_conley_vcov` to capture the time + coords args,
asserts the helper sees raw integer time labels (0..T-1, not absorb-
demeaned floats) AND n_units distinct (lat, lon) pairs (not collapsed
to one demeaned mean per unit).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: igerber

diff_diff/estimators.py

@@ -1666,21 +1666,30 @@ def fit(  # type: ignore[override]
         _fit_vcov_type = self._resolve_effective_vcov_type(effective_cluster_ids)
 
         # Resolve Conley arrays from column names (init-time) plus the
-        # estimator's `time` / `unit` columns. When vcov_type != "conley",
-        # these are silently ignored downstream (Phase 1 / 2 convention).
+        # estimator's `time` / `unit` columns. CRITICAL: read from the
+        # ORIGINAL `data` frame, NOT `working_data` — if absorb is used
+        # with overlapping covariates (e.g. lat/lon or time listed in
+        # both `absorb` and `conley_coords`/`time`), `working_data` has
+        # those columns demeaned and the Conley helper would silently
+        # partition the spatial sandwich on residualized inputs.
+        # Mirrors the DiD/TWFE contract at `estimators.py::DifferenceInDifferences.fit`
+        # and `twfe.py::TwoWayFixedEffects.fit` (FWL composability: the meat
+        # is computed on demeaned scores but the kernel grid uses the raw
+        # coords + time/unit). When vcov_type != "conley", these are silently
+        # ignored downstream (Phase 1 / 2 convention).
         if _fit_vcov_type == "conley":
             _conley_coords_arr: Optional[np.ndarray] = np.column_stack(
                 [
-                    working_data[self.conley_coords[0]].values.astype(np.float64),
-                    working_data[self.conley_coords[1]].values.astype(np.float64),
+                    data[self.conley_coords[0]].values.astype(np.float64),
+                    data[self.conley_coords[1]].values.astype(np.float64),
                 ]
             )
             # Preserve the original time-label dtype (int, datetime64, pd.Period,
             # string). `_compute_conley_vcov` normalizes to dense 0..T-1 codes
             # internally; float coercion here would break datetime64 / Period /
             # string encodings before the normalizer runs.
-            _conley_time_arr: Optional[np.ndarray] = np.asarray(working_data[time].values)
-            _conley_unit_arr: Optional[np.ndarray] = working_data[unit].values
+            _conley_time_arr: Optional[np.ndarray] = np.asarray(data[time].values)
+            _conley_unit_arr: Optional[np.ndarray] = data[unit].values
         else:
             _conley_coords_arr = None
             _conley_time_arr = None

tests/test_conley_vcov.py

@@ -1291,6 +1291,92 @@ def _spy(*args, **kwargs):
         )
         assert set(uniques.tolist()) == {0, 1}, f"Expected raw integer labels 0/1, got {uniques}"
 
+    def test_mpd_conley_with_absorb_uses_raw_coords_and_time(self, monkeypatch):
+        """MultiPeriodDiD + Conley + absorb=[<col>] must feed the Conley
+        helper the ORIGINAL coords/time/unit columns from `data`, not the
+        absorb-demeaned `working_data`. If a user lists the `time` column
+        (or coord columns) in `absorb`, working_data has those demeaned and
+        the Conley helper would partition the within-period spatial sandwich
+        on residualized floats. Mirrors the DiD raw-time contract.
+        Codex CI R5 P0.
+        """
+        import pandas as _pd
+
+        import diff_diff.linalg as linalg_module
+        from diff_diff import MultiPeriodDiD
+
+        rng = np.random.default_rng(seed=71)
+        n_units = 10
+        T = 4
+        rows = []
+        for u in range(n_units):
+            treated = u >= n_units // 2
+            lat = rng.uniform(-30, 30)
+            lon = rng.uniform(-100, 100)
+            for t in range(T):
+                effect = 1.0 if (treated and t >= 2) else 0.0
+                yv = 0.2 * t + effect + rng.normal(0, 0.3)
+                rows.append(
+                    {
+                        "unit": u,
+                        "time": t,
+                        "y": yv,
+                        "treated": int(treated),
+                        "lat": lat,
+                        "lon": lon,
+                    }
+                )
+        df = _pd.DataFrame(rows)
+
+        captured: dict = {"time_arg": None, "unit_arg": None, "coords_arg": None}
+        orig = linalg_module._compute_conley_vcov
+
+        def _spy(*args, **kwargs):
+            captured["time_arg"] = kwargs.get("time")
+            captured["unit_arg"] = kwargs.get("unit")
+            # coords is the 3rd positional arg to _compute_conley_vcov
+            if len(args) >= 3:
+                captured["coords_arg"] = args[2]
+            return orig(*args, **kwargs)
+
+        monkeypatch.setattr(linalg_module, "_compute_conley_vcov", _spy)
+        MultiPeriodDiD(
+            vcov_type="conley",
+            conley_coords=("lat", "lon"),
+            conley_cutoff_km=2000.0,
+            conley_lag_cutoff=1,
+        ).fit(
+            df,
+            outcome="y",
+            treatment="treated",
+            time="time",
+            unit="unit",
+            post_periods=[2, 3],
+            reference_period=0,
+            absorb=["unit"],
+        )
+        # Raw time labels span exactly 0..T-1 = 4 distinct values; demeaned
+        # absorb would collapse to per-unit means → ~n_units distinct values.
+        time_arg = np.asarray(captured["time_arg"])
+        uniques = np.unique(time_arg)
+        assert len(uniques) == T, (
+            f"Expected {T} unique time labels (raw 0..{T - 1}), got {len(uniques)}: "
+            f"{uniques[:5]} — absorb is leaking demeaned time into the Conley helper."
+        )
+        assert set(uniques.tolist()) == set(range(T))
+        # Raw coords are time-invariant within unit and span n_units distinct
+        # (lat, lon) pairs. If absorb=["unit"] leaked demeaned coords, all
+        # within-unit coord values would collapse to 0 (per-unit mean), giving
+        # only 1 distinct row across all observations.
+        coords_arg = np.asarray(captured["coords_arg"])
+        # Expect n_units distinct (lat, lon) pairs since each unit has its own
+        unique_coords = np.unique(coords_arr_view := coords_arg, axis=0)
+        del coords_arr_view
+        assert len(unique_coords) == n_units, (
+            f"Expected {n_units} unique coord pairs, got {len(unique_coords)} — "
+            "absorb=['unit'] is leaking demeaned coords into the Conley helper."
+        )
+
     def test_multi_period_did_with_conley_panel(self):
         """Phase 2 MultiPeriodDiD + vcov_type='conley' uses the block-decomposed
         sandwich (matches R conleyreg). Verifies that finite SEs are produced