Center rolling window for time offset

pandas/_libs/window/indexers.pyx

@@ -11,7 +11,7 @@ def calculate_variable_window_bounds(
     int64_t num_values,
     int64_t window_size,
     object min_periods,  # unused but here to match get_window_bounds signature
-    object center,  # unused but here to match get_window_bounds signature
+    bint center,
     object closed,
     const int64_t[:] index
 ):
@@ -30,7 +30,7 @@ def calculate_variable_window_bounds(
         ignored, exists for compatibility
 
     center : object
-        ignored, exists for compatibility
+        center the rolling window on the current observation
 
     closed : str
         string of side of the window that should be closed
@@ -43,7 +43,9 @@ def calculate_variable_window_bounds(
     (ndarray[int64], ndarray[int64])
     """
     cdef:
-        bint left_closed = False, right_closed = False
+        bint left_closed = False
+        bint right_closed = False
+        bint center_window = False
         ndarray[int64_t, ndim=1] start, end
         int64_t start_bound, end_bound, index_growth_sign = 1
         Py_ssize_t i, j
@@ -60,6 +62,8 @@ def calculate_variable_window_bounds(
 
     if index[num_values - 1] < index[0]:
         index_growth_sign = -1
+    if center:
+        center_window = True
 
     start = np.empty(num_values, dtype='int64')
     start.fill(-1)
@@ -74,14 +78,27 @@ def calculate_variable_window_bounds(
     # right endpoint is open
     else:
         end[0] = 0
+    if center_window:
+        for j in range(0, num_values+1):
+            if (index[j] == index[0] + index_growth_sign*window_size/2 and
+               right_closed):
+                end[0] = j+1
+                break
+            elif index[j] >= index[0] + index_growth_sign * window_size/2:
+                end[0] = j
+                break
 
     with nogil:
 
         # start is start of slice interval (including)
         # end is end of slice interval (not including)
         for i in range(1, num_values):
-            end_bound = index[i]
-            start_bound = index[i] - index_growth_sign * window_size
+            if center_window:
+                end_bound = index[i] + index_growth_sign * window_size/2
+                start_bound = index[i] - index_growth_sign * window_size/2
+            else:
+                end_bound = index[i]
+                start_bound = index[i] - index_growth_sign * window_size
 
             # left endpoint is closed
             if left_closed:
@@ -95,14 +112,27 @@ def calculate_variable_window_bounds(
                     start[i] = j
                     break
 
+            # for centered window advance the end bound until we are
+            # outside the constraint
+            if center_window:
+                for j in range(end[i - 1], num_values+1):
+                    if j == num_values:
+                        end[i] = j
+                    elif ((index[j] - end_bound) * index_growth_sign == 0 and
+                          right_closed):
+                        end[i] = j+1
+                        break
+                    elif (index[j] - end_bound) * index_growth_sign >= 0:
+                        end[i] = j
+                        break
             # end bound is previous end
             # or current index
-            if (index[end[i - 1]] - end_bound) * index_growth_sign <= 0:
+            elif (index[end[i - 1]] - end_bound) * index_growth_sign <= 0:
                 end[i] = i + 1
             else:
                 end[i] = end[i - 1]
 
             # right endpoint is open
-            if not right_closed:
+            if not right_closed and not center_window:
                 end[i] -= 1
     return start, end

pandas/core/window/rolling.py

@@ -334,9 +334,12 @@ def _get_window_indexer(self) -> BaseIndexer:
         """
         if isinstance(self.window, BaseIndexer):
             return self.window
+
         if self._win_freq_i8 is not None:
             return VariableWindowIndexer(
-                index_array=self._index_array, window_size=self._win_freq_i8
+                index_array=self._index_array,
+                window_size=self._win_freq_i8,
+                center=self.center,
             )
         return FixedWindowIndexer(window_size=self.window)
 
@@ -450,7 +453,34 @@ def homogeneous_func(values: np.ndarray):
             if values.size == 0:
                 return values.copy()
 
+            def _calculate_center_offset(window) -> int:
+                """
+                Calculate an offset necessary to have the window label to be centered.
+
+                Parameters
+                ----------
+                window: ndarray or int
+                    window weights or window
+
+                Returns
+                -------
+                int
+                """
+                if not is_integer(window):
+                    window = len(window)
+                return int((window - 1) / 2.0)
+
+            offset = (
+                _calculate_center_offset(self.window)
+                if self.center
+                and not isinstance(self._get_window_indexer(), VariableWindowIndexer)
+                else 0
+            )
+            additional_nans = np.array([np.nan] * offset)
+
             def calc(x):
+                x = np.concatenate((x, additional_nans))
+
                 start, end = window_indexer.get_window_bounds(
                     num_values=len(x),
                     min_periods=min_periods,
@@ -469,6 +499,15 @@ def calc(x):
             if numba_cache_key is not None:
                 NUMBA_FUNC_CACHE[numba_cache_key] = func
 
+            # TODO: is this equivalent to the above?
+            # if use_numba_cache:
+            #     NUMBA_FUNC_CACHE[(kwargs["original_func"], "rolling_apply")] = func
+
+            if self.center and not isinstance(
+                self._get_window_indexer(), VariableWindowIndexer
+            ):
+                result = self._center_window(result, self.window)
+
             return result
 
         if self.method == "single":
@@ -1880,20 +1919,14 @@ def validate(self):
         # we allow rolling on a datetimelike index
         if (
             self.obj.empty
+            # TODO: add "or self.is_datetimelike"?
             or isinstance(
                 self._on, (ABCDatetimeIndex, ABCTimedeltaIndex, ABCPeriodIndex)
             )
         ) and isinstance(self.window, (str, BaseOffset, timedelta)):
 
             self._validate_monotonic()
 
-            # we don't allow center
-            if self.center:
-                raise NotImplementedError(
-                    "center is not implemented for "
-                    "datetimelike and offset based windows"
-                )
-
             # this will raise ValueError on non-fixed freqs
             try:
                 freq = to_offset(self.window)