API: reimplement FixedWindowIndexer.get_window_bounds

pandas/core/window/indexers.py

@@ -78,30 +78,16 @@ def get_window_bounds(
         closed: Optional[str] = None,
     ) -> Tuple[np.ndarray, np.ndarray]:
 
-        start_s = np.zeros(self.window_size, dtype="int64")
-        start_e = (
-            np.arange(self.window_size, num_values, dtype="int64")
-            - self.window_size
-            + 1
-        )
-        start = np.concatenate([start_s, start_e])[:num_values]
-
-        end_s = np.arange(self.window_size, dtype="int64") + 1
-        end_e = start_e + self.window_size
-        end = np.concatenate([end_s, end_e])[:num_values]
-
-        if center and self.window_size > 2:
-            offset = min((self.window_size - 1) // 2, num_values - 1)
-            start_s_buffer = np.roll(start, -offset)[: num_values - offset]
-            end_s_buffer = np.roll(end, -offset)[: num_values - offset]
+        if center:
+            offset = (self.window_size - 1) // 2
+        else:
+            offset = 0
 
-            start_e_buffer = np.arange(
-                start[-1] + 1, start[-1] + 1 + offset, dtype="int64"
-            )
-            end_e_buffer = np.array([end[-1]] * offset, dtype="int64")
+        end = np.arange(1 + offset, num_values + 1 + offset).astype("int64")
+        start = end - self.window_size
 
-            start = np.concatenate([start_s_buffer, start_e_buffer])
-            end = np.concatenate([end_s_buffer, end_e_buffer])
+        end = np.clip(end, 0, num_values)
+        start = np.clip(start, 0, num_values)
 
         return start, end
 

pandas/tests/window/test_grouper.py

@@ -297,6 +297,32 @@ def test_groupby_rolling_center_center(self):
         )
         tm.assert_frame_equal(result, expected)
 
+    @pytest.mark.parametrize("min_periods", [5, 4, 3])
+    def test_groupby_rolling_center_min_periods(self, min_periods):
+        df = pd.DataFrame({"group": ["A"] * 10 + ["B"] * 10, "data": range(20)})
+
+        window_size = 5
+        result = (
+            df.groupby("group")
+            .rolling(window_size, center=True, min_periods=min_periods)
+            .mean()
+        )
+        result = result.reset_index()[["group", "data"]]
+
+        grp_A_mean = [1.0, 1.5, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 7.5, 8.0]
+        grp_B_mean = [x + 10.0 for x in grp_A_mean]
+
+        num_nans = max(0, min_periods - 3)  # For window_size of 5
+        nans = [np.nan] * num_nans
+        grp_A_expected = nans + grp_A_mean[num_nans : 10 - num_nans] + nans
+        grp_B_expected = nans + grp_B_mean[num_nans : 10 - num_nans] + nans
+
+        expected = pd.DataFrame(
+            {"group": ["A"] * 10 + ["B"] * 10, "data": grp_A_expected + grp_B_expected}
+        )
+
+        tm.assert_frame_equal(result, expected)
+
     def test_groupby_subselect_rolling(self):
         # GH 35486
         df = DataFrame(