API: reimplement FixedWindowIndexer.get_window_bounds to fix groupby bug

pandas/core/window/indexers.py

@@ -78,17 +78,19 @@ 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]
+        if center:
+            offset = self.window_size // 2
+        else:
+            offset = 0
+
+        end = np.arange(1 + offset, num_values + 1 + offset)
+        start = end - self.window_size
+
+        # end is exclusive, whereas start is inclusive
+        # thus the bounds for end should be 1 greater than the bounds for start
+        end = np.clip(end, 1, num_values)
+        start = np.clip(start, 0, num_values - 1)
 
-        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]
         return start, end
 
 

pandas/core/window/rolling.py

@@ -494,9 +494,7 @@ def _get_cython_func_type(self, func: str) -> Callable:
 
         Variable algorithms do not use window while fixed do.
         """
-        if self.is_freq_type or isinstance(self.window, BaseIndexer):
-            return self._get_roll_func(f"{func}_variable")
-        return partial(self._get_roll_func(f"{func}_fixed"), win=self._get_window())
+        return self._get_roll_func(f"{func}_variable")
 
     def _get_window_indexer(self, window: int) -> BaseIndexer:
         """
@@ -611,13 +609,9 @@ def homogeneous_func(values: np.ndarray):
             if values.size == 0:
                 return values.copy()
 
-            offset = calculate_center_offset(window) if center else 0
-            additional_nans = np.array([np.nan] * offset)
-
             if not is_weighted:
 
                 def calc(x):
-                    x = np.concatenate((x, additional_nans))
                     if not isinstance(self.window, BaseIndexer):
                         min_periods = calculate_min_periods(
                             window, self.min_periods, len(x), require_min_periods, floor
@@ -641,7 +635,6 @@ def calc(x):
             else:
 
                 def calc(x):
-                    x = np.concatenate((x, additional_nans))
                     return func(x, window, self.min_periods)
 
             with np.errstate(all="ignore"):
@@ -654,9 +647,6 @@ def calc(x):
             if use_numba_cache:
                 NUMBA_FUNC_CACHE[(kwargs["original_func"], "rolling_apply")] = func
 
-            if center:
-                result = self._center_window(result, window)
-
             return result
 
         return self._apply_blockwise(homogeneous_func)