ENH: Rolling rank

asv_bench/benchmarks/rolling.py

@@ -180,6 +180,24 @@ def time_quantile(self, constructor, window, dtype, percentile, interpolation):
         self.roll.quantile(percentile, interpolation=interpolation)
 
 
+class Rank:
+    params = (
+        ["DataFrame", "Series"],
+        [10, 1000],
+        ["int", "float"],
+        [True, False],
+    )
+    param_names = ["constructor", "window", "dtype", "percentile"]
+
+    def setup(self, constructor, window, dtype, percentile):
+        N = 10 ** 5
+        arr = np.random.random(N).astype(dtype)
+        self.roll = getattr(pd, constructor)(arr).rolling(window)
+
+    def time_rank(self, constructor, window, dtype, percentile):
+        self.roll.rank(percentile)
+
+
 class PeakMemFixedWindowMinMax:
 
     params = ["min", "max"]

pandas/_libs/src/skiplist.h

@@ -183,7 +183,7 @@ PANDAS_INLINE double skiplist_get(skiplist_t *skp, int i, int *ret) {
 PANDAS_INLINE int skiplist_insert(skiplist_t *skp, double value) {
     node_t *node, *prevnode, *newnode, *next_at_level;
     int *steps_at_level;
-    int size, steps, level;
+    int size, steps, level, rank = 0;
     node_t **chain;
 
     chain = skp->tmp_chain;
@@ -197,6 +197,7 @@ PANDAS_INLINE int skiplist_insert(skiplist_t *skp, double value) {
         next_at_level = node->next[level];
         while (_node_cmp(next_at_level, value) >= 0) {
             steps_at_level[level] += node->width[level];
+            rank += node->width[level];
             node = next_at_level;
             next_at_level = node->next[level];
         }
@@ -230,7 +231,7 @@ PANDAS_INLINE int skiplist_insert(skiplist_t *skp, double value) {
 
     ++(skp->size);
 
-    return 1;
+    return rank;
 }
 
 PANDAS_INLINE int skiplist_remove(skiplist_t *skp, double value) {

pandas/_libs/window/aggregations.pyi

@@ -63,6 +63,13 @@ def roll_quantile(
     quantile: float,  # float64_t
     interpolation: Literal["linear", "lower", "higher", "nearest", "midpoint"],
 ) -> np.ndarray: ...  # np.ndarray[float]
+def roll_rank(
+    values: np.ndarray,
+    start: np.ndarray,
+    end: np.ndarray,
+    minp: int,
+    percentile: bool,
+) -> np.ndarray: ...  # np.ndarray[float]
 def roll_apply(
     obj: object,
     start: np.ndarray,  # np.ndarray[np.int64]

pandas/_libs/window/aggregations.pyx

@@ -795,7 +795,7 @@ def roll_median_c(const float64_t[:] values, ndarray[int64_t] start,
                     val = values[j]
                     if notnan(val):
                         nobs += 1
-                        err = skiplist_insert(sl, val) != 1
+                        err = skiplist_insert(sl, val) == -1
                         if err:
                             break
 
@@ -806,7 +806,7 @@ def roll_median_c(const float64_t[:] values, ndarray[int64_t] start,
                     val = values[j]
                     if notnan(val):
                         nobs += 1
-                        err = skiplist_insert(sl, val) != 1
+                        err = skiplist_insert(sl, val) == -1
                         if err:
                             break
 
@@ -1138,6 +1138,81 @@ def roll_quantile(const float64_t[:] values, ndarray[int64_t] start,
 
     return output
 
+def roll_rank(const float64_t[:] values, ndarray[int64_t] start,
+              ndarray[int64_t] end, int64_t minp, bint percentile) -> np.ndarray:
+    """
+    O(N log(window)) implementation using skip list
+
+    derived from roll_quantile
+    """
+    cdef:
+        Py_ssize_t i, j, s, e, N = len(values), idx
+        int rank = 0
+        int64_t nobs = 0, win
+        float64_t val
+        skiplist_t *skiplist
+        float64_t[::1] output = None
+
+    is_monotonic_increasing_bounds = is_monotonic_increasing_start_end_bounds(
+        start, end
+    )
+    # we use the Fixed/Variable Indexer here as the
+    # actual skiplist ops outweigh any window computation costs
+    output = np.empty(N, dtype=np.float64)
+
+    win = (end - start).max()
+    if win == 0:
+        output[:] = NaN
+        return output
+    skiplist = skiplist_init(<int>win)
+    if skiplist == NULL:
+        raise MemoryError("skiplist_init failed")
+
+    with nogil:
+        for i in range(N):
+            s = start[i]
+            e = end[i]
+
+            if i == 0 or not is_monotonic_increasing_bounds:
+                if not is_monotonic_increasing_bounds:
+                    nobs = 0
+                    skiplist_destroy(skiplist)
+                    skiplist = skiplist_init(<int>win)
+
+                # setup
+                for j in range(s, e):
+                    val = values[j]
+                    if notnan(val):
+                        nobs += 1
+                        rank = skiplist_insert(skiplist, val)
+                        if rank == -1:
+                            raise MemoryError("skiplist_insert failed")
+
+            else:
+                # calculate deletes
+                for j in range(start[i - 1], s):
+                    val = values[j]
+                    if notnan(val):
+                        skiplist_remove(skiplist, val)
+                        nobs -= 1
+
+                # calculate adds
+                for j in range(end[i - 1], e):
+                    val = values[j]
+                    if notnan(val):
+                        nobs += 1
+                        rank = skiplist_insert(skiplist, val)
+                        if rank == -1:
+                            raise MemoryError("skiplist_insert failed")
+            if nobs >= minp:
+                output[i] = <float64_t>(rank + 1) / nobs if percentile else rank + 1
+            else:
+                output[i] = NaN
+
+    skiplist_destroy(skiplist)
+
+    return np.asarray(output)
+
 
 def roll_apply(object obj,
                ndarray[int64_t] start, ndarray[int64_t] end,

pandas/core/window/rolling.py

@@ -1409,6 +1409,14 @@ def quantile(self, quantile: float, interpolation: str = "linear", **kwargs):
 
         return self._apply(window_func, name="quantile", **kwargs)
 
+    def rank(self, pct: bool = False, **kwargs):
+        window_func = partial(
+            window_aggregations.roll_rank,
+            percentile=pct,
+        )
+
+        return self._apply(window_func, name="rank", **kwargs)
+
     def cov(
         self,
         other: DataFrame | Series | None = None,

pandas/tests/window/test_rolling.py

@@ -1500,3 +1500,25 @@ def test_rolling_numeric_dtypes():
         dtype="float64",
     )
     tm.assert_frame_equal(result, expected)
+
+
+@pytest.mark.parametrize("window", [1, 3, 10, 1000])
+def test_rank(window):
+    length = 1000
+    ser = Series(data=np.random.rand(length))
+
+    expected = ser.rolling(window).apply(lambda x: x.rank().iloc[-1])
+    result = ser.rolling(window).rank()
+
+    tm.assert_series_equal(result, expected)
+
+
+@pytest.mark.parametrize("window", [1, 3, 10, 1000])
+def test_percentile_rank(window):
+    length = 1000
+    ser = Series(data=np.random.rand(length))
+
+    expected = ser.rolling(window).apply(lambda x: x.rank(pct=True).iloc[-1])
+    result = ser.rolling(window).rank(pct=True)
+
+    tm.assert_series_equal(result, expected)