Repository: narwhals-dev/narwhals
Pull request #1451 (2024-11-27 10:05:53)

feat: add `Series|Expr.rolling_var` and `Series|Expr.rolling_std`
## What type of PR is this? (check all applicable)

- [ ] 💾 Refactor
- [x] ✨ Feature
- [ ] 🐛 Bug Fix
- [ ] 🔧 Optimization
- [ ] 📝 Documentation
- [ ] ✅ Test
- [ ] 🐳 Other

## Related issues

- Related issue #1254 

## Checklist

- [x] Code follows style guide (ruff)
- [x] Tests added
- [x] Documented the changes

## If you have comments or can explain your changes, please do so below

Opening as draft, I still need to figure out how to deal with null/nan's to match polars behavior.

Edit: Scary size diff, yet 25%+ is just the copying of methods to the stable api, and another good 10% are docstrings examples

FBruzzesi: Checking performances: on 1M floats, pyarrow is ~2x slower than pandas 🤔

Patch:
diff --git a/docs/api-reference/expr.md b/docs/api-reference/expr.md
index 523a45f0..fef26d1e 100644
--- a/docs/api-reference/expr.md
+++ b/docs/api-reference/expr.md
@@ -48,7 +48,9 @@
         - quantile
         - replace_strict
         - rolling_mean
+        - rolling_std
         - rolling_sum
+        - rolling_var
         - round
         - sample
         - shift
diff --git a/docs/api-reference/series.md b/docs/api-reference/series.md
index 818e71b5..d5a446dc 100644
--- a/docs/api-reference/series.md
+++ b/docs/api-reference/series.md
@@ -56,7 +56,9 @@
         - rename
         - replace_strict
         - rolling_mean
+        - rolling_std
         - rolling_sum
+        - rolling_var
         - round
         - sample
         - scatter
diff --git a/narwhals/_arrow/expr.py b/narwhals/_arrow/expr.py
index 845181bd..b960ffa4 100644
--- a/narwhals/_arrow/expr.py
+++ b/narwhals/_arrow/expr.py
@@ -511,6 +511,40 @@ class ArrowExpr(CompliantExpr[ArrowSeries]):
             center=center,
         )
 
+    def rolling_var(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None,
+        center: bool,
+        ddof: int,
+    ) -> Self:
+        return reuse_series_implementation(
+            self,
+            "rolling_var",
+            window_size=window_size,
+            min_periods=min_periods,
+            center=center,
+            ddof=ddof,
+        )
+
+    def rolling_std(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None,
+        center: bool,
+        ddof: int,
+    ) -> Self:
+        return reuse_series_implementation(
+            self,
+            "rolling_std",
+            window_size=window_size,
+            min_periods=min_periods,
+            center=center,
+            ddof=ddof,
+        )
+
     @property
     def dt(self: Self) -> ArrowExprDateTimeNamespace:
         return ArrowExprDateTimeNamespace(self)
diff --git a/narwhals/_arrow/series.py b/narwhals/_arrow/series.py
index f97bc604..2f29ab9d 100644
--- a/narwhals/_arrow/series.py
+++ b/narwhals/_arrow/series.py
@@ -13,6 +13,7 @@ from narwhals._arrow.utils import cast_for_truediv
 from narwhals._arrow.utils import floordiv_compat
 from narwhals._arrow.utils import narwhals_to_native_dtype
 from narwhals._arrow.utils import native_to_narwhals_dtype
+from narwhals._arrow.utils import pad_series
 from narwhals._arrow.utils import parse_datetime_format
 from narwhals.utils import Implementation
 from narwhals.utils import generate_temporary_column_name
@@ -935,27 +936,12 @@ class ArrowSeries(CompliantSeries):
         min_periods: int | None,
         center: bool,
     ) -> Self:
-        import pyarrow as pa
         import pyarrow.compute as pc
 
         min_periods = min_periods if min_periods is not None else window_size
-        if center:
-            offset_left = window_size // 2
-            offset_right = offset_left - (
-                window_size % 2 == 0
-            )  # subtract one if window_size is even
-
-            native_series = self._native_series
-
-            pad_left = pa.array([None] * offset_left, type=native_series.type)
-            pad_right = pa.array([None] * offset_right, type=native_series.type)
-            padded_arr = self._from_native_series(
-                pa.concat_arrays([pad_left, native_series.combine_chunks(), pad_right])
-            )
-        else:
-            padded_arr = self
+        padded_series, offset = pad_series(self, window_size=window_size, center=center)
 
-        cum_sum = padded_arr.cum_sum(reverse=False).fill_null(
+        cum_sum = padded_series.cum_sum(reverse=False).fill_null(
             value=None, strategy="forward", limit=None
         )
         rolling_sum = (
@@ -965,7 +951,7 @@ class ArrowSeries(CompliantSeries):
             else cum_sum
         )
 
-        valid_count = padded_arr.cum_count(reverse=False)
+        valid_count = padded_series.cum_count(reverse=False)
         count_in_window = valid_count - valid_count.shift(window_size).fill_null(
             value=0, strategy=None, limit=None
         )
@@ -977,9 +963,7 @@ class ArrowSeries(CompliantSeries):
                 None,
             )
         )
-        if center:
-            result = result[offset_left + offset_right :]
-        return result
+        return result[offset:]
 
     def rolling_mean(
         self: Self,
@@ -988,27 +972,12 @@ class ArrowSeries(CompliantSeries):
         min_periods: int | None,
         center: bool,
     ) -> Self:
-        import pyarrow as pa
         import pyarrow.compute as pc
 
         min_periods = min_periods if min_periods is not None else window_size
-        if center:
-            offset_left = window_size // 2
-            offset_right = offset_left - (
-                window_size % 2 == 0
-            )  # subtract one if window_size is even
+        padded_series, offset = pad_series(self, window_size=window_size, center=center)
 
-            native_series = self._native_series
-
-            pad_left = pa.array([None] * offset_left, type=native_series.type)
-            pad_right = pa.array([None] * offset_right, type=native_series.type)
-            padded_arr = self._from_native_series(
-                pa.concat_arrays([pad_left, native_series.combine_chunks(), pad_right])
-            )
-        else:
-            padded_arr = self
-
-        cum_sum = padded_arr.cum_sum(reverse=False).fill_null(
+        cum_sum = padded_series.cum_sum(reverse=False).fill_null(
             value=None, strategy="forward", limit=None
         )
         rolling_sum = (
@@ -1018,7 +987,7 @@ class ArrowSeries(CompliantSeries):
             else cum_sum
         )
 
-        valid_count = padded_arr.cum_count(reverse=False)
+        valid_count = padded_series.cum_count(reverse=False)
         count_in_window = valid_count - valid_count.shift(window_size).fill_null(
             value=0, strategy=None, limit=None
         )
@@ -1033,9 +1002,74 @@ class ArrowSeries(CompliantSeries):
             )
             / count_in_window
         )
-        if center:
-            result = result[offset_left + offset_right :]
-        return result
+        return result[offset:]
+
+    def rolling_var(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None,
+        center: bool,
+        ddof: int,
+    ) -> Self:
+        import pyarrow.compute as pc  # ignore-banned-import
+
+        min_periods = min_periods if min_periods is not None else window_size
+        padded_series, offset = pad_series(self, window_size=window_size, center=center)
+
+        cum_sum = padded_series.cum_sum(reverse=False).fill_null(
+            value=None, strategy="forward", limit=None
+        )
+        rolling_sum = (
+            cum_sum
+            - cum_sum.shift(window_size).fill_null(value=0, strategy=None, limit=None)
+            if window_size != 0
+            else cum_sum
+        )
+
+        cum_sum_sq = (
+            padded_series.__pow__(2)
+            .cum_sum(reverse=False)
+            .fill_null(value=None, strategy="forward", limit=None)
+        )
+        rolling_sum_sq = (
+            cum_sum_sq
+            - cum_sum_sq.shift(window_size).fill_null(value=0, strategy=None, limit=None)
+            if window_size != 0
+            else cum_sum_sq
+        )
+
+        valid_count = padded_series.cum_count(reverse=False)
+        count_in_window = valid_count - valid_count.shift(window_size).fill_null(
+            value=0, strategy=None, limit=None
+        )
+
+        result = self._from_native_series(
+            pc.if_else(
+                (count_in_window >= min_periods)._native_series,
+                (rolling_sum_sq - (rolling_sum**2 / count_in_window))._native_series,
+                None,
+            )
+        ) / self._from_native_series(
+            pc.max_element_wise((count_in_window - ddof)._native_series, 0)
+        )
+
+        return result[offset:]
+
+    def rolling_std(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None,
+        center: bool,
+        ddof: int,
+    ) -> Self:
+        return (
+            self.rolling_var(
+                window_size=window_size, min_periods=min_periods, center=center, ddof=ddof
+            )
+            ** 0.5
+        )
 
     def __iter__(self: Self) -> Iterator[Any]:
         yield from (
diff --git a/narwhals/_arrow/utils.py b/narwhals/_arrow/utils.py
index f9a8fa89..8c895b81 100644
--- a/narwhals/_arrow/utils.py
+++ b/narwhals/_arrow/utils.py
@@ -502,3 +502,39 @@ def _parse_time_format(arr: pa.Array) -> str:
         if pc.all(matches.is_valid()).as_py():
             return time_fmt
     return ""
+
+
+def pad_series(
+    series: ArrowSeries, *, window_size: int, center: bool
+) -> tuple[ArrowSeries, int]:
+    """Pad series with None values on the left and/or right side, depending on the specified parameters.
+
+    Arguments:
+        series: The input ArrowSeries to be padded.
+        window_size: The desired size of the window.
+        center: Specifies whether to center the padding or not.
+
+    Returns:
+        A tuple containing the padded ArrowSeries and the offset value.
+    """
+    import pyarrow as pa  # ignore-banned-import
+
+    if center:
+        offset_left = window_size // 2
+        offset_right = offset_left - (
+            window_size % 2 == 0
+        )  # subtract one if window_size is even
+
+        native_series = series._native_series
+
+        pad_left = pa.array([None] * offset_left, type=native_series.type)
+        pad_right = pa.array([None] * offset_right, type=native_series.type)
+        padded_arr = series._from_native_series(
+            pa.concat_arrays([pad_left, native_series.combine_chunks(), pad_right])
+        )
+        offset = offset_left + offset_right
+    else:
+        padded_arr = series
+        offset = 0
+
+    return padded_arr, offset
diff --git a/narwhals/_dask/expr.py b/narwhals/_dask/expr.py
index ac3b3c5b..0e898579 100644
--- a/narwhals/_dask/expr.py
+++ b/narwhals/_dask/expr.py
@@ -928,6 +928,64 @@ class DaskExpr(CompliantExpr["dask_expr.Series"]):
             returns_scalar=False,
         )
 
+    def rolling_var(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None,
+        center: bool,
+        ddof: int,
+    ) -> Self:
+        def func(
+            _input: dask_expr.Series,
+            _window: int,
+            _min_periods: int | None,
+            _center: bool,  # noqa: FBT001
+            _ddof: int,
+        ) -> dask_expr.Series:
+            return _input.rolling(
+                window=_window, min_periods=_min_periods, center=_center
+            ).var(ddof=ddof)
+
+        return self._from_call(
+            func,
+            "rolling_var",
+            window_size,
+            min_periods,
+            center,
+            ddof,
+            returns_scalar=False,
+        )
+
+    def rolling_std(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None,
+        center: bool,
+        ddof: int,
+    ) -> Self:
+        def func(
+            _input: dask_expr.Series,
+            _window: int,
+            _min_periods: int | None,
+            _center: bool,  # noqa: FBT001
+            _ddof: int,
+        ) -> dask_expr.Series:
+            return _input.rolling(
+                window=_window, min_periods=_min_periods, center=_center
+            ).std(ddof=ddof)
+
+        return self._from_call(
+            func,
+            "rolling_std",
+            window_size,
+            min_periods,
+            center,
+            ddof,
+            returns_scalar=False,
+        )
+
 
 class DaskExprStringNamespace:
     def __init__(self, expr: DaskExpr) -> None:
diff --git a/narwhals/_pandas_like/expr.py b/narwhals/_pandas_like/expr.py
index 96af851e..8d507fb4 100644
--- a/narwhals/_pandas_like/expr.py
+++ b/narwhals/_pandas_like/expr.py
@@ -589,6 +589,40 @@ class PandasLikeExpr(CompliantExpr[PandasLikeSeries]):
             center=center,
         )
 
+    def rolling_var(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None,
+        center: bool,
+        ddof: int,
+    ) -> Self:
+        return reuse_series_implementation(
+            self,
+            "rolling_var",
+            window_size=window_size,
+            min_periods=min_periods,
+            center=center,
+            ddof=ddof,
+        )
+
+    def rolling_std(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None,
+        center: bool,
+        ddof: int,
+    ) -> Self:
+        return reuse_series_implementation(
+            self,
+            "rolling_std",
+            window_size=window_size,
+            min_periods=min_periods,
+            center=center,
+            ddof=ddof,
+        )
+
     @property
     def str(self: Self) -> PandasLikeExprStringNamespace:
         return PandasLikeExprStringNamespace(self)
diff --git a/narwhals/_pandas_like/series.py b/narwhals/_pandas_like/series.py
index f8d2e2cf..a8b57c5c 100644
--- a/narwhals/_pandas_like/series.py
+++ b/narwhals/_pandas_like/series.py
@@ -1059,6 +1059,32 @@ class PandasLikeSeries(CompliantSeries):
         ).mean()
         return self._from_native_series(result)
 
+    def rolling_var(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None,
+        center: bool,
+        ddof: int,
+    ) -> Self:
+        result = self._native_series.rolling(
+            window=window_size, min_periods=min_periods, center=center
+        ).var(ddof=ddof)
+        return self._from_native_series(result)
+
+    def rolling_std(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None,
+        center: bool,
+        ddof: int,
+    ) -> Self:
+        result = self._native_series.rolling(
+            window=window_size, min_periods=min_periods, center=center
+        ).std(ddof=ddof)
+        return self._from_native_series(result)
+
     def __iter__(self: Self) -> Iterator[Any]:
         yield from self._native_series.__iter__()
 
diff --git a/narwhals/_polars/expr.py b/narwhals/_polars/expr.py
index 4c6e59a4..6c50c99a 100644
--- a/narwhals/_polars/expr.py
+++ b/narwhals/_polars/expr.py
@@ -79,6 +79,48 @@ class PolarsExpr:
             )
         return self._from_native_expr(native_expr)
 
+    def rolling_var(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None,
+        center: bool,
+        ddof: int,
+    ) -> Self:
+        if self._backend_version < (1,):  # pragma: no cover
+            msg = "`rolling_var` not implemented for polars older than 1.0"
+            raise NotImplementedError(msg)
+
+        return self._from_native_expr(
+            self._native_expr.rolling_var(
+                window_size=window_size,
+                min_periods=min_periods,
+                center=center,
+                ddof=ddof,
+            )
+        )
+
+    def rolling_std(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None,
+        center: bool,
+        ddof: int,
+    ) -> Self:
+        if self._backend_version < (1,):  # pragma: no cover
+            msg = "`rolling_std` not implemented for polars older than 1.0"
+            raise NotImplementedError(msg)
+
+        return self._from_native_expr(
+            self._native_expr.rolling_std(
+                window_size=window_size,
+                min_periods=min_periods,
+                center=center,
+                ddof=ddof,
+            )
+        )
+
     def map_batches(
         self,
         function: Callable[..., Self],
diff --git a/narwhals/_polars/series.py b/narwhals/_polars/series.py
index 424c5bfb..d7088bb0 100644
--- a/narwhals/_polars/series.py
+++ b/narwhals/_polars/series.py
@@ -284,6 +284,48 @@ class PolarsSeries:
 
         return self._from_native_series(native_result)
 
+    def rolling_var(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None,
+        center: bool,
+        ddof: int,
+    ) -> Self:
+        if self._backend_version < (1,):  # pragma: no cover
+            msg = "`rolling_var` not implemented for polars older than 1.0"
+            raise NotImplementedError(msg)
+
+        return self._from_native_series(
+            self._native_series.rolling_var(
+                window_size=window_size,
+                min_periods=min_periods,
+                center=center,
+                ddof=ddof,
+            )
+        )
+
+    def rolling_std(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None,
+        center: bool,
+        ddof: int,
+    ) -> Self:
+        if self._backend_version < (1,):  # pragma: no cover
+            msg = "`rolling_std` not implemented for polars older than 1.0"
+            raise NotImplementedError(msg)
+
+        return self._from_native_series(
+            self._native_series.rolling_std(
+                window_size=window_size,
+                min_periods=min_periods,
+                center=center,
+                ddof=ddof,
+            )
+        )
+
     def sort(self: Self, *, descending: bool, nulls_last: bool) -> Self:
         if self._backend_version < (0, 20, 6):
             result = self._native_series.sort(descending=descending)
diff --git a/narwhals/expr.py b/narwhals/expr.py
index 41038670..f1e7215b 100644
--- a/narwhals/expr.py
+++ b/narwhals/expr.py
@@ -3601,6 +3601,193 @@ class Expr:
             )
         )
 
+    def rolling_var(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None = None,
+        center: bool = False,
+        ddof: int = 1,
+    ) -> Self:
+        """Apply a rolling variance (moving variance) over the values.
+
+        !!! warning
+            This functionality is considered **unstable**. It may be changed at any point
+            without it being considered a breaking change.
+
+        A window of length `window_size` will traverse the values. The resulting values
+        will be aggregated to their variance.
+
+        The window at a given row will include the row itself and the `window_size - 1`
+        elements before it.
+
+        Arguments:
+            window_size: The length of the window in number of elements. It must be a
+                strictly positive integer.
+            min_periods: The number of values in the window that should be non-null before
+                computing a result. If set to `None` (default), it will be set equal to
+                `window_size`. If provided, it must be a strictly positive integer, and
+                less than or equal to `window_size`.
+            center: Set the labels at the center of the window.
+            ddof: Delta Degrees of Freedom; the divisor for a length N window is N - ddof.
+
+        Returns:
+            A new expression.
+
+        Examples:
+            >>> import narwhals as nw
+            >>> from narwhals.typing import IntoFrameT
+            >>> import pandas as pd
+            >>> import polars as pl
+            >>> import pyarrow as pa
+            >>> data = {"a": [1.0, 2.0, None, 4.0]}
+            >>> df_pd = pd.DataFrame(data)
+            >>> df_pl = pl.DataFrame(data)
+            >>> df_pa = pa.table(data)
+
+            We define a library agnostic function:
+
+            >>> def agnostic_rolling_var(df_native: IntoFrameT) -> IntoFrameT:
+            ...     df = nw.from_native(df_native)
+            ...     return df.with_columns(
+            ...         b=nw.col("a").rolling_var(window_size=3, min_periods=1)
+            ...     ).to_native()
+
+            We can then pass any supported library such as Pandas, Polars, or PyArrow to `func`:
+
+            >>> agnostic_rolling_var(df_pd)
+                 a    b
+            0  1.0  NaN
+            1  2.0  0.5
+            2  NaN  0.5
+            3  4.0  2.0
+
+            >>> agnostic_rolling_var(df_pl)  #  doctest:+SKIP
+            shape: (4, 2)
+            ┌──────┬──────┐
+            │ a    ┆ b    │
+            │ ---  ┆ ---  │
+            │ f64  ┆ f64  │
+            ╞══════╪══════╡
+            │ 1.0  ┆ null │
+            │ 2.0  ┆ 0.5  │
+            │ null ┆ 0.5  │
+            │ 4.0  ┆ 2.0  │
+            └──────┴──────┘
+
+            >>> agnostic_rolling_var(df_pa)  #  doctest:+ELLIPSIS
+            pyarrow.Table
+            a: double
+            b: double
+            ----
+            a: [[1,2,null,4]]
+            b: [[nan,0.5,0.5,2]]
+        """
+        window_size, min_periods = _validate_rolling_arguments(
+            window_size=window_size, min_periods=min_periods
+        )
+
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).rolling_var(
+                window_size=window_size, min_periods=min_periods, center=center, ddof=ddof
+            )
+        )
+
+    def rolling_std(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None = None,
+        center: bool = False,
+        ddof: int = 1,
+    ) -> Self:
+        """Apply a rolling standard deviation (moving standard deviation) over the values.
+
+        !!! warning
+            This functionality is considered **unstable**. It may be changed at any point
+            without it being considered a breaking change.
+
+        A window of length `window_size` will traverse the values. The resulting values
+        will be aggregated to their standard deviation.
+
+        The window at a given row will include the row itself and the `window_size - 1`
+        elements before it.
+
+        Arguments:
+            window_size: The length of the window in number of elements. It must be a
+                strictly positive integer.
+            min_periods: The number of values in the window that should be non-null before
+                computing a result. If set to `None` (default), it will be set equal to
+                `window_size`. If provided, it must be a strictly positive integer, and
+                less than or equal to `window_size`.
+            center: Set the labels at the center of the window.
+            ddof: Delta Degrees of Freedom; the divisor for a length N window is N - ddof.
+
+        Returns:
+            A new expression.
+
+        Examples:
+            >>> import narwhals as nw
+            >>> from narwhals.typing import IntoFrameT
+            >>> import pandas as pd
+            >>> import polars as pl
+            >>> import pyarrow as pa
+            >>> data = {"a": [1.0, 2.0, None, 4.0]}
+            >>> df_pd = pd.DataFrame(data)
+            >>> df_pl = pl.DataFrame(data)
+            >>> df_pa = pa.table(data)
+
+            We define a library agnostic function:
+
+            >>> def agnostic_rolling_std(df_native: IntoFrameT) -> IntoFrameT:
+            ...     df = nw.from_native(df_native)
+            ...     return df.with_columns(
+            ...         b=nw.col("a").rolling_std(window_size=3, min_periods=1)
+            ...     ).to_native()
+
+            We can then pass any supported library such as Pandas, Polars, or PyArrow to `func`:
+
+            >>> agnostic_rolling_std(df_pd)
+                 a         b
+            0  1.0       NaN
+            1  2.0  0.707107
+            2  NaN  0.707107
+            3  4.0  1.414214
+
+            >>> agnostic_rolling_std(df_pl)  #  doctest:+SKIP
+            shape: (4, 2)
+            ┌──────┬──────────┐
+            │ a    ┆ b        │
+            │ ---  ┆ ---      │
+            │ f64  ┆ f64      │
+            ╞══════╪══════════╡
+            │ 1.0  ┆ null     │
+            │ 2.0  ┆ 0.707107 │
+            │ null ┆ 0.707107 │
+            │ 4.0  ┆ 1.414214 │
+            └──────┴──────────┘
+
+            >>> agnostic_rolling_std(df_pa)  #  doctest:+ELLIPSIS
+            pyarrow.Table
+            a: double
+            b: double
+            ----
+            a: [[1,2,null,4]]
+            b: [[nan,0.7071067811865476,0.7071067811865476,1.4142135623730951]]
+        """
+        window_size, min_periods = _validate_rolling_arguments(
+            window_size=window_size, min_periods=min_periods
+        )
+
+        return self.__class__(
+            lambda plx: self._to_compliant_expr(plx).rolling_std(
+                window_size=window_size,
+                min_periods=min_periods,
+                center=center,
+                ddof=ddof,
+            )
+        )
+
     @property
     def str(self: Self) -> ExprStringNamespace[Self]:
         return ExprStringNamespace(self)
diff --git a/narwhals/series.py b/narwhals/series.py
index 8601d72c..ba7b1faa 100644
--- a/narwhals/series.py
+++ b/narwhals/series.py
@@ -3358,7 +3358,7 @@ class Series(Generic[IntoSeriesT]):
             center: Set the labels at the center of the window.
 
         Returns:
-            A new expression.
+            A new series.
 
         Examples:
             >>> import narwhals as nw
@@ -3451,7 +3451,7 @@ class Series(Generic[IntoSeriesT]):
             center: Set the labels at the center of the window.
 
         Returns:
-            A new expression.
+            A new series.
 
         Examples:
             >>> import narwhals as nw
@@ -3515,6 +3515,192 @@ class Series(Generic[IntoSeriesT]):
             )
         )
 
+    def rolling_var(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None = None,
+        center: bool = False,
+        ddof: int = 1,
+    ) -> Self:
+        """Apply a rolling variance (moving variance) over the values.
+
+        !!! warning
+            This functionality is considered **unstable**. It may be changed at any point
+            without it being considered a breaking change.
+
+        A window of length `window_size` will traverse the values. The resulting values
+        will be aggregated to their variance.
+
+        The window at a given row will include the row itself and the `window_size - 1`
+        elements before it.
+
+        Arguments:
+            window_size: The length of the window in number of elements. It must be a
+                strictly positive integer.
+            min_periods: The number of values in the window that should be non-null before
+                computing a result. If set to `None` (default), it will be set equal to
+                `window_size`. If provided, it must be a strictly positive integer, and
+                less than or equal to `window_size`.
+            center: Set the labels at the center of the window.
+            ddof: Delta Degrees of Freedom; the divisor for a length N window is N - ddof.
+
+        Returns:
+            A new series.
+
+        Examples:
+            >>> import narwhals as nw
+            >>> from narwhals.typing import IntoSeriesT
+            >>> import pandas as pd
+            >>> import polars as pl
+            >>> import pyarrow as pa
+            >>> data = [1.0, 3.0, 1.0, 4.0]
+            >>> s_pd = pd.Series(data)
+            >>> s_pl = pl.Series(data)
+            >>> s_pa = pa.chunked_array([data])
+
+            We define a library agnostic function:
+
+            >>> def agnostic_rolling_var(s_native: IntoSeriesT) -> IntoSeriesT:
+            ...     s = nw.from_native(s_native, series_only=True)
+            ...     return s.rolling_var(window_size=2, min_periods=1).to_native()
+
+            We can then pass any supported library such as Pandas, Polars, or PyArrow to `func`:
+
+            >>> agnostic_rolling_var(s_pd)
+            0    NaN
+            1    2.0
+            2    2.0
+            3    4.5
+            dtype: float64
+
+            >>> agnostic_rolling_var(s_pl)  # doctest:+NORMALIZE_WHITESPACE
+            shape: (4,)
+            Series: '' [f64]
+            [
+               null
+               2.0
+               2.0
+               4.5
+            ]
+
+            >>> agnostic_rolling_var(s_pa)  # doctest:+ELLIPSIS
+            <pyarrow.lib.ChunkedArray object at ...>
+            [
+              [
+                nan,
+                2,
+                2,
+                4.5
+              ]
+            ]
+        """
+        window_size, min_periods = _validate_rolling_arguments(
+            window_size=window_size, min_periods=min_periods
+        )
+
+        if len(self) == 0:  # pragma: no cover
+            return self
+
+        return self._from_compliant_series(
+            self._compliant_series.rolling_var(
+                window_size=window_size, min_periods=min_periods, center=center, ddof=ddof
+            )
+        )
+
+    def rolling_std(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None = None,
+        center: bool = False,
+        ddof: int = 1,
+    ) -> Self:
+        """Apply a rolling standard deviation (moving standard deviation) over the values.
+
+        !!! warning
+            This functionality is considered **unstable**. It may be changed at any point
+            without it being considered a breaking change.
+
+        A window of length `window_size` will traverse the values. The resulting values
+        will be aggregated to their standard deviation.
+
+        The window at a given row will include the row itself and the `window_size - 1`
+        elements before it.
+
+        Arguments:
+            window_size: The length of the window in number of elements. It must be a
+                strictly positive integer.
+            min_periods: The number of values in the window that should be non-null before
+                computing a result. If set to `None` (default), it will be set equal to
+                `window_size`. If provided, it must be a strictly positive integer, and
+                less than or equal to `window_size`.
+            center: Set the labels at the center of the window.
+            ddof: Delta Degrees of Freedom; the divisor for a length N window is N - ddof.
+
+        Returns:
+            A new series.
+
+        Examples:
+            >>> import narwhals as nw
+            >>> from narwhals.typing import IntoSeriesT
+            >>> import pandas as pd
+            >>> import polars as pl
+            >>> import pyarrow as pa
+            >>> data = [1.0, 3.0, 1.0, 4.0]
+            >>> s_pd = pd.Series(data)
+            >>> s_pl = pl.Series(data)
+            >>> s_pa = pa.chunked_array([data])
+
+            We define a library agnostic function:
+
+            >>> def agnostic_rolling_std(s_native: IntoSeriesT) -> IntoSeriesT:
+            ...     s = nw.from_native(s_native, series_only=True)
+            ...     return s.rolling_std(window_size=2, min_periods=1).to_native()
+
+            We can then pass any supported library such as Pandas, Polars, or PyArrow to `func`:
+
+            >>> agnostic_rolling_std(s_pd)
+            0         NaN
+            1    1.414214
+            2    1.414214
+            3    2.121320
+            dtype: float64
+
+            >>> agnostic_rolling_std(s_pl)  # doctest:+NORMALIZE_WHITESPACE
+            shape: (4,)
+            Series: '' [f64]
+            [
+               null
+               1.414214
+               1.414214
+               2.12132
+            ]
+
+            >>> agnostic_rolling_std(s_pa)  # doctest:+ELLIPSIS
+            <pyarrow.lib.ChunkedArray object at ...>
+            [
+              [
+                nan,
+                1.4142135623730951,
+                1.4142135623730951,
+                2.1213203435596424
+              ]
+            ]
+        """
+        window_size, min_periods = _validate_rolling_arguments(
+            window_size=window_size, min_periods=min_periods
+        )
+
+        if len(self) == 0:  # pragma: no cover
+            return self
+
+        return self._from_compliant_series(
+            self._compliant_series.rolling_std(
+                window_size=window_size, min_periods=min_periods, center=center, ddof=ddof
+            )
+        )
+
     def __iter__(self: Self) -> Iterator[Any]:
         yield from self._compliant_series.__iter__()
 
diff --git a/narwhals/stable/v1/__init__.py b/narwhals/stable/v1/__init__.py
index e5aa8f28..7be34a9e 100644
--- a/narwhals/stable/v1/__init__.py
+++ b/narwhals/stable/v1/__init__.py
@@ -671,7 +671,7 @@ class Series(NwSeries[Any]):
             center: Set the labels at the center of the window.
 
         Returns:
-            A new expression.
+            A new series.
 
         Examples:
             >>> import narwhals as nw
@@ -763,7 +763,7 @@ class Series(NwSeries[Any]):
             center: Set the labels at the center of the window.
 
         Returns:
-            A new expression.
+            A new series.
 
         Examples:
             >>> import narwhals as nw
@@ -826,6 +826,196 @@ class Series(NwSeries[Any]):
             center=center,
         )
 
+    def rolling_var(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None = None,
+        center: bool = False,
+        ddof: int = 1,
+    ) -> Self:
+        """Apply a rolling variance (moving variance) over the values.
+
+        !!! warning
+            This functionality is considered **unstable**. It may be changed at any point
+            without it being considered a breaking change.
+
+        A window of length `window_size` will traverse the values. The resulting values
+        will be aggregated to their variance.
+
+        The window at a given row will include the row itself and the `window_size - 1`
+        elements before it.
+
+        Arguments:
+            window_size: The length of the window in number of elements. It must be a
+                strictly positive integer.
+            min_periods: The number of values in the window that should be non-null before
+                computing a result. If set to `None` (default), it will be set equal to
+                `window_size`. If provided, it must be a strictly positive integer, and
+                less than or equal to `window_size`.
+            center: Set the labels at the center of the window.
+            ddof: Delta Degrees of Freedom; the divisor for a length N window is N - ddof.
+
+        Returns:
+            A new series.
+
+        Examples:
+            >>> import narwhals as nw
+            >>> from narwhals.typing import IntoSeriesT
+            >>> import pandas as pd
+            >>> import polars as pl
+            >>> import pyarrow as pa
+            >>> data = [1.0, 3.0, 1.0, 4.0]
+            >>> s_pd = pd.Series(data)
+            >>> s_pl = pl.Series(data)
+            >>> s_pa = pa.chunked_array([data])
+
+            We define a library agnostic function:
+
+            >>> def agnostic_rolling_var(s_native: IntoSeriesT) -> IntoSeriesT:
+            ...     s = nw.from_native(s_native, series_only=True)
+            ...     return s.rolling_var(window_size=2, min_periods=1).to_native()
+
+            We can then pass any supported library such as Pandas, Polars, or PyArrow to `func`:
+
+            >>> agnostic_rolling_var(s_pd)
+            0    NaN
+            1    2.0
+            2    2.0
+            3    4.5
+            dtype: float64
+
+            >>> agnostic_rolling_var(s_pl)  # doctest:+NORMALIZE_WHITESPACE
+            shape: (4,)
+            Series: '' [f64]
+            [
+               null
+               2.0
+               2.0
+               4.5
+            ]
+
+            >>> agnostic_rolling_var(s_pa)  # doctest:+ELLIPSIS
+            <pyarrow.lib.ChunkedArray object at ...>
+            [
+              [
+                nan,
+                2,
+                2,
+                4.5
+              ]
+            ]
+        """
+        from narwhals.exceptions import NarwhalsUnstableWarning
+        from narwhals.utils import find_stacklevel
+
+        msg = (
+            "`Series.rolling_var` is being called from the stable API although considered "
+            "an unstable feature."
+        )
+        warn(message=msg, category=NarwhalsUnstableWarning, stacklevel=find_stacklevel())
+        return super().rolling_var(
+            window_size=window_size,
+            min_periods=min_periods,
+            center=center,
+            ddof=ddof,
+        )
+
+    def rolling_std(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None = None,
+        center: bool = False,
+        ddof: int = 1,
+    ) -> Self:
+        """Apply a rolling standard deviation (moving standard deviation) over the values.
+
+        !!! warning
+            This functionality is considered **unstable**. It may be changed at any point
+            without it being considered a breaking change.
+
+        A window of length `window_size` will traverse the values. The resulting values
+        will be aggregated to their standard deviation.
+
+        The window at a given row will include the row itself and the `window_size - 1`
+        elements before it.
+
+        Arguments:
+            window_size: The length of the window in number of elements. It must be a
+                strictly positive integer.
+            min_periods: The number of values in the window that should be non-null before
+                computing a result. If set to `None` (default), it will be set equal to
+                `window_size`. If provided, it must be a strictly positive integer, and
+                less than or equal to `window_size`.
+            center: Set the labels at the center of the window.
+            ddof: Delta Degrees of Freedom; the divisor for a length N window is N - ddof.
+
+        Returns:
+            A new series.
+
+        Examples:
+            >>> import narwhals as nw
+            >>> from narwhals.typing import IntoSeriesT
+            >>> import pandas as pd
+            >>> import polars as pl
+            >>> import pyarrow as pa
+            >>> data = [1.0, 3.0, 1.0, 4.0]
+            >>> s_pd = pd.Series(data)
+            >>> s_pl = pl.Series(data)
+            >>> s_pa = pa.chunked_array([data])
+
+            We define a library agnostic function:
+
+            >>> def agnostic_rolling_std(s_native: IntoSeriesT) -> IntoSeriesT:
+            ...     s = nw.from_native(s_native, series_only=True)
+            ...     return s.rolling_std(window_size=2, min_periods=1).to_native()
+
+            We can then pass any supported library such as Pandas, Polars, or PyArrow to `func`:
+
+            >>> agnostic_rolling_std(s_pd)
+            0         NaN
+            1    1.414214
+            2    1.414214
+            3    2.121320
+            dtype: float64
+
+            >>> agnostic_rolling_std(s_pl)  # doctest:+NORMALIZE_WHITESPACE
+            shape: (4,)
+            Series: '' [f64]
+            [
+               null
+               1.414214
+               1.414214
+               2.12132
+            ]
+
+            >>> agnostic_rolling_std(s_pa)  # doctest:+ELLIPSIS
+            <pyarrow.lib.ChunkedArray object at ...>
+            [
+              [
+                nan,
+                1.4142135623730951,
+                1.4142135623730951,
+                2.1213203435596424
+              ]
+            ]
+        """
+        from narwhals.exceptions import NarwhalsUnstableWarning
+        from narwhals.utils import find_stacklevel
+
+        msg = (
+            "`Series.rolling_std` is being called from the stable API although considered "
+            "an unstable feature."
+        )
+        warn(message=msg, category=NarwhalsUnstableWarning, stacklevel=find_stacklevel())
+        return super().rolling_std(
+            window_size=window_size,
+            min_periods=min_periods,
+            center=center,
+            ddof=ddof,
+        )
+
 
 class Expr(NwExpr):
     def _l1_norm(self) -> Self:
@@ -1124,6 +1314,197 @@ class Expr(NwExpr):
             center=center,
         )
 
+    def rolling_var(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None = None,
+        center: bool = False,
+        ddof: int = 1,
+    ) -> Self:
+        """Apply a rolling variance (moving variance) over the values.
+
+        !!! warning
+            This functionality is considered **unstable**. It may be changed at any point
+            without it being considered a breaking change.
+
+        A window of length `window_size` will traverse the values. The resulting values
+        will be aggregated to their variance.
+
+        The window at a given row will include the row itself and the `window_size - 1`
+        elements before it.
+
+        Arguments:
+            window_size: The length of the window in number of elements. It must be a
+                strictly positive integer.
+            min_periods: The number of values in the window that should be non-null before
+                computing a result. If set to `None` (default), it will be set equal to
+                `window_size`. If provided, it must be a strictly positive integer, and
+                less than or equal to `window_size`.
+            center: Set the labels at the center of the window.
+            ddof: Delta Degrees of Freedom; the divisor for a length N window is N - ddof.
+
+        Returns:
+            A new expression.
+
+        Examples:
+            >>> import narwhals as nw
+            >>> from narwhals.typing import IntoFrameT
+            >>> import pandas as pd
+            >>> import polars as pl
+            >>> import pyarrow as pa
+            >>> data = {"a": [1.0, 2.0, None, 4.0]}
+            >>> df_pd = pd.DataFrame(data)
+            >>> df_pl = pl.DataFrame(data)
+            >>> df_pa = pa.table(data)
+
+            We define a library agnostic function:
+
+            >>> def agnostic_rolling_var(df_native: IntoFrameT) -> IntoFrameT:
+            ...     df = nw.from_native(df_native)
+            ...     return df.with_columns(
+            ...         b=nw.col("a").rolling_var(window_size=3, min_periods=1)
+            ...     ).to_native()
+
+            We can then pass any supported library such as Pandas, Polars, or PyArrow to `func`:
+
+            >>> agnostic_rolling_var(df_pd)
+                 a    b
+            0  1.0  NaN
+            1  2.0  0.5
+            2  NaN  0.5
+            3  4.0  2.0
+
+            >>> agnostic_rolling_var(df_pl)  #  doctest:+SKIP
+            shape: (4, 2)
+            ┌──────┬──────┐
+            │ a    ┆ b    │
+            │ ---  ┆ ---  │
+            │ f64  ┆ f64  │
+            ╞══════╪══════╡
+            │ 1.0  ┆ null │
+            │ 2.0  ┆ 0.5  │
+            │ null ┆ 0.5  │
+            │ 4.0  ┆ 2.0  │
+            └──────┴──────┘
+
+            >>> agnostic_rolling_var(df_pa)  #  doctest:+ELLIPSIS
+            pyarrow.Table
+            a: double
+            b: double
+            ----
+            a: [[1,2,null,4]]
+            b: [[nan,0.5,0.5,2]]
+        """
+        from narwhals.exceptions import NarwhalsUnstableWarning
+        from narwhals.utils import find_stacklevel
+
+        msg = (
+            "`Expr.rolling_var` is being called from the stable API although considered "
+            "an unstable feature."
+        )
+        warn(message=msg, category=NarwhalsUnstableWarning, stacklevel=find_stacklevel())
+        return super().rolling_var(
+            window_size=window_size, min_periods=min_periods, center=center, ddof=ddof
+        )
+
+    def rolling_std(
+        self: Self,
+        window_size: int,
+        *,
+        min_periods: int | None = None,
+        center: bool = False,
+        ddof: int = 1,
+    ) -> Self:
+        """Apply a rolling standard deviation (moving standard deviation) over the values.
+
+        !!! warning
+            This functionality is considered **unstable**. It may be changed at any point
+            without it being considered a breaking change.
+
+        A window of length `window_size` will traverse the values. The resulting values
+        will be aggregated to their standard deviation.
+
+        The window at a given row will include the row itself and the `window_size - 1`
+        elements before it.
+
+        Arguments:
+            window_size: The length of the window in number of elements. It must be a
+                strictly positive integer.
+            min_periods: The number of values in the window that should be non-null before
+                computing a result. If set to `None` (default), it will be set equal to
+                `window_size`. If provided, it must be a strictly positive integer, and
+                less than or equal to `window_size`
+            center: Set the labels at the center of the window.
+            ddof: Delta Degrees of Freedom; the divisor for a length N window is N - ddof.
+
+        Returns:
+            A new expression.
+
+        Examples:
+            >>> import narwhals as nw
+            >>> from narwhals.typing import IntoFrameT
+            >>> import pandas as pd
+            >>> import polars as pl
+            >>> import pyarrow as pa
+            >>> data = {"a": [1.0, 2.0, None, 4.0]}
+            >>> df_pd = pd.DataFrame(data)
+            >>> df_pl = pl.DataFrame(data)
+            >>> df_pa = pa.table(data)
+
+            We define a library agnostic function:
+
+            >>> def agnostic_rolling_std(df_native: IntoFrameT) -> IntoFrameT:
+            ...     df = nw.from_native(df_native)
+            ...     return df.with_columns(
+            ...         b=nw.col("a").rolling_std(window_size=3, min_periods=1)
+            ...     ).to_native()
+
+            We can then pass any supported library such as Pandas, Polars, or PyArrow to `func`:
+
+            >>> agnostic_rolling_std(df_pd)
+                 a         b
+            0  1.0       NaN
+            1  2.0  0.707107
+            2  NaN  0.707107
+            3  4.0  1.414214
+
+            >>> agnostic_rolling_std(df_pl)  #  doctest:+SKIP
+            shape: (4, 2)
+            ┌──────┬──────────┐
+            │ a    ┆ b        │
+            │ ---  ┆ ---      │
+            │ f64  ┆ f64      │
+            ╞══════╪══════════╡
+            │ 1.0  ┆ null     │
+            │ 2.0  ┆ 0.707107 │
+            │ null ┆ 0.707107 │
+            │ 4.0  ┆ 1.414214 │
+            └──────┴──────────┘
+
+            >>> agnostic_rolling_std(df_pa)  #  doctest:+ELLIPSIS
+            pyarrow.Table
+            a: double
+            b: double
+            ----
+            a: [[1,2,null,4]]
+            b: [[nan,0.7071067811865476,0.7071067811865476,1.4142135623730951]]
+        """
+        from narwhals.exceptions import NarwhalsUnstableWarning
+        from narwhals.utils import find_stacklevel
+
+        msg = (
+            "`Expr.rolling_std` is being called from the stable API although considered "
+            "an unstable feature."
+        )
+        warn(message=msg, category=NarwhalsUnstableWarning, stacklevel=find_stacklevel())
+        return super().rolling_std(
+            window_size=window_size,
+            min_periods=min_periods,
+            center=center,
+            ddof=ddof,
+        )
+
 
 class Schema(NwSchema):
     """Ordered mapping of column names to their data type.


Test patch:
diff --git a/tests/expr_and_series/rolling_std_test.py b/tests/expr_and_series/rolling_std_test.py
new file mode 100644
index 00000000..3fdba949
--- /dev/null
+++ b/tests/expr_and_series/rolling_std_test.py
@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+from typing import Any
+
+import numpy as np
+import pytest
+
+import narwhals.stable.v1 as nw
+from tests.utils import POLARS_VERSION
+from tests.utils import Constructor
+from tests.utils import ConstructorEager
+from tests.utils import assert_equal_data
+
+data = {"a": [1.0, 2.0, 1.0, 3.0, 1.0, 4.0, 1.0]}
+
+kwargs_and_expected = (
+    {
+        "name": "x1",
+        "kwargs": {"window_size": 3},
+        "expected": np.sqrt([float("nan"), float("nan"), 1 / 3, 1, 4 / 3, 7 / 3, 3]),
+    },
+    {
+        "name": "x2",
+        "kwargs": {"window_size": 3, "min_periods": 1},
+        "expected": np.sqrt([float("nan"), 0.5, 1 / 3, 1.0, 4 / 3, 7 / 3, 3]),
+    },
+    {
+        "name": "x3",
+        "kwargs": {"window_size": 2, "min_periods": 1},
+        "expected": np.sqrt([float("nan"), 0.5, 0.5, 2.0, 2.0, 4.5, 4.5]),
+    },
+    {
+        "name": "x4",
+        "kwargs": {"window_size": 5, "min_periods": 1, "center": True},
+        "expected": np.sqrt([1 / 3, 11 / 12, 4 / 5, 17 / 10, 2.0, 2.25, 3]),
+    },
+    {
+        "name": "x5",
+        "kwargs": {"window_size": 4, "min_periods": 1, "center": True},
+        "expected": np.sqrt([0.5, 1 / 3, 11 / 12, 11 / 12, 2.25, 2.25, 3]),
+    },
+    {
+        "name": "x6",
+        "kwargs": {"window_size": 3, "ddof": 2},
+        "expected": np.sqrt([float("nan"), float("nan"), 2 / 3, 2.0, 8 / 3, 14 / 3, 6.0]),
+    },
+)
+
+
+@pytest.mark.filterwarnings(
+    "ignore:`Expr.rolling_std` is being called from the stable API although considered an unstable feature."
+)
+@pytest.mark.parametrize("kwargs_and_expected", kwargs_and_expected)
+def test_rolling_std_expr(
+    request: pytest.FixtureRequest,
+    constructor: Constructor,
+    kwargs_and_expected: dict[str, Any],
+) -> None:
+    name = kwargs_and_expected["name"]
+    kwargs = kwargs_and_expected["kwargs"]
+    expected = kwargs_and_expected["expected"]
+
+    if "dask" in str(constructor) or (
+        "polars" in str(constructor) and POLARS_VERSION < (1,)
+    ):
+        # TODO(FBruzzesi): Dask is raising the following error:
+        # NotImplementedError: Partition size is less than overlapping window size.
+        # Try using ``df.repartition`` to increase the partition size.
+        request.applymarker(pytest.mark.xfail)
+
+    df = nw.from_native(constructor(data))
+    result = df.select(nw.col("a").rolling_std(**kwargs).alias(name))
+
+    assert_equal_data(result, {name: expected})
+
+
+@pytest.mark.filterwarnings(
+    "ignore:`Series.rolling_std` is being called from the stable API although considered an unstable feature."
+)
+@pytest.mark.parametrize("kwargs_and_expected", kwargs_and_expected)
+def test_rolling_std_series(
+    request: pytest.FixtureRequest,
+    constructor_eager: ConstructorEager,
+    kwargs_and_expected: dict[str, Any],
+) -> None:
+    if "polars" in str(constructor_eager) and POLARS_VERSION < (1,):
+        request.applymarker(pytest.mark.xfail)
+
+    name = kwargs_and_expected["name"]
+    kwargs = kwargs_and_expected["kwargs"]
+    expected = kwargs_and_expected["expected"]
+
+    df = nw.from_native(constructor_eager(data), eager_only=True)
+    result = df.select(df["a"].rolling_std(**kwargs).alias(name))
+
+    assert_equal_data(result, {name: expected})
diff --git a/tests/expr_and_series/rolling_var_test.py b/tests/expr_and_series/rolling_var_test.py
new file mode 100644
index 00000000..32767c99
--- /dev/null
+++ b/tests/expr_and_series/rolling_var_test.py
@@ -0,0 +1,141 @@
+from __future__ import annotations
+
+import random
+from typing import Any
+
+import hypothesis.strategies as st
+import pandas as pd
+import polars as pl
+import pyarrow as pa
+import pytest
+from hypothesis import given
+
+import narwhals.stable.v1 as nw
+from tests.utils import PANDAS_VERSION
+from tests.utils import POLARS_VERSION
+from tests.utils import Constructor
+from tests.utils import ConstructorEager
+from tests.utils import assert_equal_data
+
+data = {"a": [1.0, 2.0, 1.0, 3.0, 1.0, 4.0, 1.0]}
+
+kwargs_and_expected = (
+    {
+        "name": "x1",
+        "kwargs": {"window_size": 3},
+        "expected": [float("nan"), float("nan"), 1 / 3, 1, 4 / 3, 7 / 3, 3],
+    },
+    {
+        "name": "x2",
+        "kwargs": {"window_size": 3, "min_periods": 1},
+        "expected": [float("nan"), 0.5, 1 / 3, 1.0, 4 / 3, 7 / 3, 3],
+    },
+    {
+        "name": "x3",
+        "kwargs": {"window_size": 2, "min_periods": 1},
+        "expected": [float("nan"), 0.5, 0.5, 2.0, 2.0, 4.5, 4.5],
+    },
+    {
+        "name": "x4",
+        "kwargs": {"window_size": 5, "min_periods": 1, "center": True},
+        "expected": [1 / 3, 11 / 12, 4 / 5, 17 / 10, 2.0, 2.25, 3],
+    },
+    {
+        "name": "x5",
+        "kwargs": {"window_size": 4, "min_periods": 1, "center": True},
+        "expected": [0.5, 1 / 3, 11 / 12, 11 / 12, 2.25, 2.25, 3],
+    },
+    {
+        "name": "x6",
+        "kwargs": {"window_size": 3, "ddof": 2},
+        "expected": [float("nan"), float("nan"), 2 / 3, 2.0, 8 / 3, 14 / 3, 6.0],
+    },
+)
+
+
+@pytest.mark.filterwarnings(
+    "ignore:`Expr.rolling_var` is being called from the stable API although considered an unstable feature."
+)
+@pytest.mark.parametrize("kwargs_and_expected", kwargs_and_expected)
+def test_rolling_var_expr(
+    request: pytest.FixtureRequest,
+    constructor: Constructor,
+    kwargs_and_expected: dict[str, Any],
+) -> None:
+    name = kwargs_and_expected["name"]
+    kwargs = kwargs_and_expected["kwargs"]
+    expected = kwargs_and_expected["expected"]
+
+    if "dask" in str(constructor) or (
+        "polars" in str(constructor) and POLARS_VERSION < (1,)
+    ):
+        # TODO(FBruzzesi): Dask is raising the following error:
+        # NotImplementedError: Partition size is less than overlapping window size.
+        # Try using ``df.repartition`` to increase the partition size.
+        request.applymarker(pytest.mark.xfail)
+
+    df = nw.from_native(constructor(data))
+    result = df.select(nw.col("a").rolling_var(**kwargs).alias(name))
+
+    assert_equal_data(result, {name: expected})
+
+
+@pytest.mark.filterwarnings(
+    "ignore:`Series.rolling_var` is being called from the stable API although considered an unstable feature."
+)
+@pytest.mark.parametrize("kwargs_and_expected", kwargs_and_expected)
+def test_rolling_var_series(
+    request: pytest.FixtureRequest,
+    constructor_eager: ConstructorEager,
+    kwargs_and_expected: dict[str, Any],
+) -> None:
+    if "polars" in str(constructor_eager) and POLARS_VERSION < (1,):
+        request.applymarker(pytest.mark.xfail)
+
+    name = kwargs_and_expected["name"]
+    kwargs = kwargs_and_expected["kwargs"]
+    expected = kwargs_and_expected["expected"]
+
+    df = nw.from_native(constructor_eager(data), eager_only=True)
+    result = df.select(df["a"].rolling_var(**kwargs).alias(name))
+
+    assert_equal_data(result, {name: expected})
+
+
+@given(  # type: ignore[misc]
+    center=st.booleans(),
+    values=st.lists(st.floats(-10, 10), min_size=5, max_size=10),
+)
+@pytest.mark.skipif(PANDAS_VERSION < (1,), reason="too old for pyarrow")
+@pytest.mark.skipif(POLARS_VERSION < (1,), reason="different null behavior")
+@pytest.mark.filterwarnings("ignore:.*:narwhals.exceptions.NarwhalsUnstableWarning")
+def test_rolling_var_hypothesis(center: bool, values: list[float]) -> None:  # noqa: FBT001
+    s = pd.Series(values)
+    window_size = random.randint(2, len(s))  # noqa: S311
+    min_periods = random.randint(2, window_size)  # noqa: S311
+    ddof = random.randint(0, min_periods - 1)  # noqa: S311
+    mask = random.sample(range(len(s)), 2)
+
+    s[mask] = None
+    df = pd.DataFrame({"a": s})
+    expected = (
+        s.rolling(window=window_size, center=center, min_periods=min_periods)
+        .var(ddof=ddof)
+        .to_frame("a")
+    )
+
+    result = nw.from_native(pa.Table.from_pandas(df)).select(
+        nw.col("a").rolling_var(
+            window_size, center=center, min_periods=min_periods, ddof=ddof
+        )
+    )
+    expected_dict = nw.from_native(expected, eager_only=True).to_dict(as_series=False)
+    assert_equal_data(result, expected_dict)
+
+    result = nw.from_native(pl.from_pandas(df)).select(
+        nw.col("a").rolling_var(
+            window_size, center=center, min_periods=min_periods, ddof=ddof
+        )
+    )
+    expected_dict = nw.from_native(expected, eager_only=True).to_dict(as_series=False)
+    assert_equal_data(result, expected_dict)