API: reimplement FixedWindowIndexer.get_window_bounds to fix groupby bug

pandas/core/window/indexers.py

@@ -78,17 +78,17 @@ 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 - 1) // 2
+        else:
+            offset = 0
+
+        end = np.arange(1 + offset, num_values + 1 + offset)
+        start = end - self.window_size
+
+        end = np.clip(end, 0, num_values)
+        start = np.clip(start, 0, 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]
         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
@@ -639,6 +633,8 @@ def calc(x):
                     return func(x, start, end, min_periods)
 
             else:
+                offset = calculate_center_offset(window) if center else 0	
+                additional_nans = np.array([np.nan] * offset)
 
                 def calc(x):
                     x = np.concatenate((x, additional_nans))
@@ -653,8 +649,8 @@ def calc(x):
 
             if use_numba_cache:
                 NUMBA_FUNC_CACHE[(kwargs["original_func"], "rolling_apply")] = func
-
-            if center:
+            
+            if center and is_weighted:	
                 result = self._center_window(result, window)
 
             return result
@@ -1439,7 +1435,8 @@ def apply(
             # Cython apply functions handle center, so don't need to use
             # _apply's center handling
             window = self._get_window()
-            offset = calculate_center_offset(window) if self.center else 0
+
+            offset = calculate_center_offset(window) if self.center and self.win_type else 0
             apply_func = self._generate_cython_apply_func(
                 args, kwargs, raw, offset, func
             )