Address PR #456 R6 review (1 P2 perf + 1 P3 docs)

P2 perf: remove duplicate cohort distance pass. Previously the event-
study path computed cohort-by-unit distances twice on staggered panels:
once in _compute_nearest_treated_distance_staggered for d_it (running
min), then again in _compute_event_time_per_row to recover the per-row
spillover-trigger onset. On large staggered panels this doubled the
dominant spatial work.

Fix: thread d_bar into _compute_nearest_treated_distance_staggered as
an optional kwarg. When supplied, the cohort loop now ALSO computes
trigger_onset_per_unit (the first cohort whose treated units fall
within d_bar of unit i) and broadcasts it to rows. The helper's
return is now a 4-tuple (d_it, row_unit, row_time, trigger_onset_or_None).
_compute_event_time_per_row accepts an optional
precomputed_trigger_onset_per_row that, when supplied (as fit() now does
on the staggered event-study path), skips the redundant cohort loop.
Falls back to inline computation for unit-test callers.

Test callsites for _compute_nearest_treated_distance_staggered updated
to handle the new 4-tuple via `d_it, row_unit, row_time, _trigger = ...`.

P3 docs: llms-full.txt and api/spillover.rst now explicitly state
that event_study=True requires horizon_max>=1 or None (horizon_max=0
is rejected, with redirect to event_study=False for the aggregate
spec). The previous wording described horizon_max=0 as a meaningful
collapsed design, which contradicted the new R5 rejection.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: igerber

diff_diff/guides/llms-full.txt

@@ -478,7 +478,7 @@ SpilloverDiD(
     alpha: float = 0.05,
     anticipation: int = 0,
     event_study: bool = False,           # Wave C: per-event-time × ring decomposition (Butts Table 2)
-    horizon_max: int | None = None,      # Bin event-times outside [-H,+H] into endpoint pools (event-study mode)
+    horizon_max: int | None = None,      # Bin event-times outside [-H,+H] into endpoint pools (event-study mode); H>=1 or None — H=0 rejected (use event_study=False for aggregate spec)
     rank_deficient_action: str = "warn",
 )
 ```

diff_diff/spillover.py

@@ -338,7 +338,8 @@ def _compute_nearest_treated_distance_staggered(
     coords: Tuple[str, str],
     metric: SpilloverMetric,
     first_treat_by_unit: Dict[Any, Any],
-) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:
+    d_bar: Optional[float] = None,
+) -> Tuple[np.ndarray, np.ndarray, np.ndarray, Optional[np.ndarray]]:
     """Return per-row nearest-treated distance for the staggered case.
 
     For each (unit, period) observation, find the minimum distance to any
@@ -361,6 +362,13 @@ def _compute_nearest_treated_distance_staggered(
     first_treat_by_unit : dict
         Mapping from unit identifier to onset time (or ``np.inf`` for
         never-treated). Generated by :func:`_extract_treatment_onsets`.
+    d_bar : float, optional
+        When supplied, the function additionally computes the per-row
+        **spillover-trigger onset** (earliest cohort onset whose treated
+        units fall within ``d_bar`` of unit ``i``) reusing the cohort
+        loop. Used by :func:`_compute_event_time_per_row` to avoid a
+        duplicate cohort pass on the event-study path
+        (PR #456 R6 performance fix).
 
     Notes
     -----
@@ -377,6 +385,11 @@ def _compute_nearest_treated_distance_staggered(
         Aligned unit identifier per row (for downstream broadcasting).
     row_time : ndarray of shape (n_rows,)
         Aligned time identifier per row.
+    trigger_onset_per_row : ndarray of shape (n_rows,) or None
+        ``None`` when ``d_bar`` is None. Otherwise: per-row earliest
+        cohort onset whose treated units fall within ``d_bar`` of the
+        row's unit, broadcast from per-unit. NaN for rows whose unit is
+        never within ``d_bar`` of any cohort.
     """
     unit_coords_df = (
         data[[unit, coords[0], coords[1]]].drop_duplicates(subset=[unit]).set_index(unit)
@@ -389,13 +402,16 @@ def _compute_nearest_treated_distance_staggered(
     row_time = np.asarray(data[time].values)
     n_rows = len(row_unit)
     d_it = np.full(n_rows, np.inf, dtype=np.float64)
+    trigger_onset_per_unit_pos: Optional[np.ndarray] = (
+        np.full(len(unit_index), np.nan, dtype=np.float64) if d_bar is not None else None
+    )
 
     # Determine the cohort onset times that exist in the data (excluding never-treated).
     unique_onsets = sorted({ft for ft in first_treat_by_unit.values() if np.isfinite(ft)})
     if not unique_onsets:
         # Degenerate: no treated units. Caller should have rejected this
         # in `_validate_spillover_inputs`, but defensively return inf.
-        return d_it, row_unit, row_time
+        return d_it, row_unit, row_time, None
 
     # Row's unit position. Invariant across cohort iterations — compute
     # once outside the loop.
@@ -426,7 +442,25 @@ def _compute_nearest_treated_distance_staggered(
         update_mask = affected_rows & (row_cohort_dist < d_it)
         d_it[update_mask] = row_cohort_dist[update_mask]
 
-    return d_it, row_unit, row_time
+        # Reuse this same cohort distance computation for the per-unit
+        # spillover-trigger onset when d_bar is supplied. The trigger is
+        # the FIRST cohort whose treated units fall within d_bar of unit
+        # i — once locked it persists for later cohort iterations. Using
+        # cumulative-treated distances here is fine: if a unit is in
+        # range of cohort c1, dists_to_cohort at onset=c1 already detects
+        # it; later iterations with extra treated units only shrink the
+        # distance, never grow it back above d_bar.
+        if trigger_onset_per_unit_pos is not None:
+            in_range_for_cohort = dists_to_cohort <= d_bar
+            not_yet_triggered = np.isnan(trigger_onset_per_unit_pos)
+            trigger_onset_per_unit_pos[in_range_for_cohort & not_yet_triggered] = onset
+
+    # Broadcast per-unit trigger to rows when computed.
+    if trigger_onset_per_unit_pos is not None:
+        trigger_onset_per_row = trigger_onset_per_unit_pos[row_pos]
+    else:
+        trigger_onset_per_row = None
+    return d_it, row_unit, row_time, trigger_onset_per_row
 
 
 def _compute_event_time_per_row(
@@ -439,6 +473,7 @@ def _compute_event_time_per_row(
     coords: Tuple[str, str],
     metric: SpilloverMetric,
     d_bar: float,
+    precomputed_trigger_onset_per_row: Optional[np.ndarray] = None,
 ) -> Tuple[np.ndarray, np.ndarray]:
     """Compute two event-time clocks per row for Wave C event-study mode.
 
@@ -475,6 +510,16 @@ def _compute_event_time_per_row(
     -------
     K_direct : ndarray of shape (n_rows,), float64 with NaN where undefined.
     K_spill : ndarray of shape (n_rows,), float64 with NaN where undefined.
+
+    Notes
+    -----
+    PR #456 R6 performance fix: when ``precomputed_trigger_onset_per_row``
+    is supplied (as :func:`_compute_nearest_treated_distance_staggered`
+    now optionally returns when called with ``d_bar=...``), the cohort
+    loop is skipped — K_spill is derived directly from the precomputed
+    trigger. The fallback (compute trigger inline) is kept for unit-test
+    callers and other code paths that don't have access to the staggered
+    distance helper's output.
     """
     n_rows = len(row_unit)
     row_time_arr = np.asarray(row_time, dtype=np.float64)
@@ -485,8 +530,19 @@ def _compute_event_time_per_row(
     direct_defined = np.isfinite(own_onsets)
     K_direct[direct_defined] = row_time_arr[direct_defined] - own_onsets[direct_defined]
 
-    # trigger_onset[i] = first effective_onset among cohorts whose treated
-    # units have d(i, treated_in_cohort) <= d_bar.
+    if precomputed_trigger_onset_per_row is not None:
+        # Fast path: reuse trigger onsets already computed by the staggered
+        # distance helper. Avoids a duplicate cohort loop.
+        row_trigger = np.asarray(precomputed_trigger_onset_per_row, dtype=np.float64)
+        K_spill = np.full(n_rows, np.nan, dtype=np.float64)
+        triggered = np.isfinite(row_trigger)
+        post_trigger = triggered & (row_time_arr >= row_trigger)
+        K_spill[post_trigger] = row_time_arr[post_trigger] - row_trigger[post_trigger]
+        return K_direct, K_spill
+
+    # Fallback path (test callers, etc.): compute trigger inline via own
+    # cohort loop. trigger_onset[i] = first effective_onset among cohorts
+    # whose treated units have d(i, treated_in_cohort) <= d_bar.
     unit_coords_df = (
         data[[unit, coords[0], coords[1]]].drop_duplicates(subset=[unit]).set_index(unit)
     )
@@ -2167,14 +2223,21 @@ def fit(
                 unit_coords_for_validation.shape[0],
             )
 
+        # Capture the spillover-trigger onsets alongside d_it on the
+        # staggered path so the event-study branch below can reuse them
+        # without redoing the cohort distance loop (PR #456 R6 perf fix).
+        trigger_onset_per_row_cached: Optional[np.ndarray] = None
         if is_staggered:
-            d_it_per_row, _, _ = _compute_nearest_treated_distance_staggered(
-                data,
-                unit=unit,
-                time=time,
-                coords=self.conley_coords,
-                metric=self.conley_metric,
-                first_treat_by_unit=effective_onsets,
+            d_it_per_row, _, _, trigger_onset_per_row_cached = (
+                _compute_nearest_treated_distance_staggered(
+                    data,
+                    unit=unit,
+                    time=time,
+                    coords=self.conley_coords,
+                    metric=self.conley_metric,
+                    first_treat_by_unit=effective_onsets,
+                    d_bar=self._effective_d_bar if self.event_study else None,
+                )
             )
         else:
             # Non-staggered: single common onset. Build d_i per unit once,
@@ -2398,6 +2461,10 @@ def fit(
                 ),
                 metric=self.conley_metric,
                 d_bar=self._effective_d_bar,
+                # PR #456 R6 perf fix: on the staggered path, reuse the
+                # trigger onsets computed during the d_it cohort loop
+                # instead of redoing the dense pairwise pass.
+                precomputed_trigger_onset_per_row=trigger_onset_per_row_cached,
             )
             # event_study=True without conley_coords requires fallback coords for
             # ring-trigger computation. The validator already requires either

docs/api/spillover.rst

@@ -195,7 +195,11 @@ planned as follow-up enhancements:
   period ``-1 - anticipation`` (TwoStageDiD parity). ``horizon_max``
   bins event-times into endpoint pools (no row drop — divergence
   from TwoStageDiD's filtering semantic, intentional per
-  ``feedback_no_silent_failures``). Scalar ``att`` becomes a
+  ``feedback_no_silent_failures``). ``horizon_max`` must be ``>=1`` or
+  ``None`` under ``event_study=True``; ``horizon_max=0`` is rejected
+  (the single bin ``k=0`` leaves no event-time pair to anchor the
+  reference period — for a single aggregate effect, use
+  ``event_study=False`` instead). Scalar ``att`` becomes a
   sample-share-weighted average of post-treatment ``tau_k`` with SE
   from linear-combination inference on the post-treatment vcov block.
   Per-event-time SEs share the same Wave B Gardner-GMM caveat

tests/test_spillover.py

@@ -359,7 +359,7 @@ class TestComputeNearestTreatedDistanceStaggered:
 
     def test_inf_pre_any_treatment(self, staggered_panel):
         df, ft = staggered_panel
-        d_it, row_unit, row_time = _compute_nearest_treated_distance_staggered(
+        d_it, row_unit, row_time, _trigger = _compute_nearest_treated_distance_staggered(
             df,
             unit="unit",
             time="time",
@@ -373,7 +373,7 @@ def test_inf_pre_any_treatment(self, staggered_panel):
 
     def test_cohort_a_active_at_t1(self, staggered_panel):
         df, ft = staggered_panel
-        d_it, row_unit, row_time = _compute_nearest_treated_distance_staggered(
+        d_it, row_unit, row_time, _trigger = _compute_nearest_treated_distance_staggered(
             df,
             unit="unit",
             time="time",
@@ -394,7 +394,7 @@ def test_cohort_a_active_at_t1(self, staggered_panel):
 
     def test_running_min_across_cohorts_at_t2(self, staggered_panel):
         df, ft = staggered_panel
-        d_it, row_unit, row_time = _compute_nearest_treated_distance_staggered(
+        d_it, row_unit, row_time, _trigger = _compute_nearest_treated_distance_staggered(
             df,
             unit="unit",
             time="time",