""" Provide a generic structure to support window functions, similar to how we have a Groupby object. """ from __future__ import annotations import copy from datetime import timedelta from functools import partial import inspect from typing import ( TYPE_CHECKING, Any, Concatenate, Literal, Self, cast, final, overload, ) import numpy as np from pandas._libs.tslibs import ( BaseOffset, Timedelta, to_offset, ) import pandas._libs.window.aggregations as window_aggregations from pandas.compat._optional import import_optional_dependency from pandas.errors import DataError from pandas.util._decorators import set_module from pandas.core.dtypes.common import ( ensure_float64, is_arrow_temporal_dtype, is_bool, is_integer, is_numeric_dtype, needs_i8_conversion, ) from pandas.core.dtypes.dtypes import ArrowDtype from pandas.core.dtypes.generic import ( ABCDataFrame, ABCSeries, ) from pandas.core.dtypes.missing import notna from pandas.core._numba import executor from pandas.core.algorithms import factorize from pandas.core.apply import ( ResamplerWindowApply, reconstruct_func, ) from pandas.core.arrays import ExtensionArray from pandas.core.base import SelectionMixin import pandas.core.common as com from pandas.core.indexers.objects import ( BaseIndexer, FixedWindowIndexer, GroupByIndexer, VariableWindowIndexer, ) from pandas.core.indexes.api import ( DatetimeIndex, Index, MultiIndex, PeriodIndex, TimedeltaIndex, ) from pandas.core.reshape.concat import concat from pandas.core.util.numba_ import ( get_jit_arguments, maybe_use_numba, prepare_function_arguments, ) from pandas.core.window.common import ( flex_binary_moment, zsqrt, ) from pandas.core.window.numba_ import ( generate_manual_numpy_nan_agg_with_axis, generate_numba_apply_func, generate_numba_table_func, ) if TYPE_CHECKING: from collections.abc import Callable from collections.abc import ( Hashable, Iterator, Sized, ) from pandas._typing import ( ArrayLike, NDFrameT, QuantileInterpolation, P, TimeUnit, T, WindowingRankType, npt, ) from pandas import ( DataFrame, Series, ) from pandas.core.generic import NDFrame from pandas.core.groupby.ops import BaseGrouper from pandas.core.arrays.datetimelike import dtype_to_unit class BaseWindow(SelectionMixin["NDFrame"]): """Provides utilities for performing windowing operations.""" _attributes: list[str] = [] exclusions: frozenset[Hashable] = frozenset() _on: Index def __init__( self, obj: NDFrame, window=None, min_periods: int | None = None, center: bool | None = False, win_type: str | None = None, on: str | Index | None = None, closed: str | None = None, step: int | None = None, method: str = "single", *, selection=None, ) -> None: self.obj = obj self.on = on self.closed = closed self.step = step self.window = window self.min_periods = min_periods self.center = center self.win_type = win_type self.method = method self._win_freq_i8: int | None = None if self.on is None: self._on = self.obj.index elif isinstance(self.on, Index): self._on = self.on elif isinstance(self.obj, ABCDataFrame) and self.on in self.obj.columns: self._on = Index(self.obj[self.on]) else: raise ValueError( f"invalid on specified as {self.on}, " "must be a column (of DataFrame), an Index or None" ) self._selection = selection self._validate() def _validate(self) -> None: if self.center is not None and not is_bool(self.center): raise ValueError("center must be a boolean") if self.min_periods is not None: if not is_integer(self.min_periods): raise ValueError("min_periods must be an integer") if self.min_periods < 0: raise ValueError("min_periods must be >= 0") if is_integer(self.window) and self.min_periods > self.window: raise ValueError( f"min_periods {self.min_periods} must be <= window {self.window}" ) if self.closed is not None and self.closed not in [ "right", "both", "left", "neither", ]: raise ValueError("closed must be 'right', 'left', 'both' or 'neither'") if not isinstance(self.obj, (ABCSeries, ABCDataFrame)): raise TypeError(f"invalid type: {type(self)}") if isinstance(self.window, BaseIndexer): # Validate that the passed BaseIndexer subclass has # a get_window_bounds with the correct signature. get_window_bounds_signature = inspect.signature( self.window.get_window_bounds ).parameters.keys() expected_signature = inspect.signature( BaseIndexer().get_window_bounds ).parameters.keys() if get_window_bounds_signature != expected_signature: raise ValueError( f"{type(self.window).__name__} does not implement " f"the correct signature for get_window_bounds" ) if self.method not in ["table", "single"]: raise ValueError("method must be 'table' or 'single") if self.step is not None: if not is_integer(self.step): raise ValueError("step must be an integer") if self.step < 0: raise ValueError("step must be >= 0") def _check_window_bounds( self, start: np.ndarray, end: np.ndarray, num_vals: int ) -> None: if len(start) != len(end): raise ValueError( f"start ({len(start)}) and end ({len(end)}) bounds must be the " f"same length" ) if len(start) != (num_vals + (self.step or 1) - 1) // (self.step or 1): raise ValueError( f"start and end bounds ({len(start)}) must be the same length " f"as the object ({num_vals}) divided by the step ({self.step}) " f"if given and rounded up" ) def _slice_axis_for_step(self, index: Index, result: Sized | None = None) -> Index: """ Slices the index for a given result and the preset step. """ return ( index if result is None or len(result) == len(index) else index[:: self.step] ) def _validate_numeric_only(self, name: str, numeric_only: bool) -> None: """ Validate numeric_only argument, raising if invalid for the input. Parameters ---------- name : str Name of the operator (kernel). numeric_only : bool Value passed by user. """ if ( self._selected_obj.ndim == 1 and numeric_only and not is_numeric_dtype(self._selected_obj.dtype) ): raise NotImplementedError( f"{type(self).__name__}.{name} does not implement numeric_only" ) def _make_numeric_only(self, obj: NDFrameT) -> NDFrameT: """Subset DataFrame to numeric columns. Parameters ---------- obj : DataFrame Returns ------- obj subset to numeric-only columns. """ result = obj.select_dtypes(include=["number"], exclude=["timedelta"]) return result def _create_data(self, obj: NDFrameT, numeric_only: bool = False) -> NDFrameT: """ Split data into blocks & return conformed data. """ # filter out the on from the object if self.on is not None and not isinstance(self.on, Index) and obj.ndim == 2: obj = obj.reindex(columns=obj.columns.difference([self.on], sort=False)) if obj.ndim > 1 and numeric_only: obj = self._make_numeric_only(obj) return obj def _gotitem(self, key, ndim, subset=None): """ Sub-classes to define. Return a sliced object. Parameters ---------- key : str / list of selections ndim : {1, 2} requested ndim of result subset : object, default None subset to act on """ # create a new object to prevent aliasing if subset is None: subset = self.obj # we need to make a shallow copy of ourselves # with the same groupby kwargs = {attr: getattr(self, attr) for attr in self._attributes} selection = self._infer_selection(key, subset) new_win = type(self)(subset, selection=selection, **kwargs) return new_win def __getattr__(self, attr: str): if attr in self._internal_names_set: return object.__getattribute__(self, attr) if attr in self.obj: return self[attr] raise AttributeError( f"'{type(self).__name__}' object has no attribute '{attr}'" ) def _dir_additions(self): return self.obj._dir_additions() def __repr__(self) -> str: """ Provide a nice str repr of our rolling object. """ attrs_list = ( f"{attr_name}={getattr(self, attr_name)}" for attr_name in self._attributes if getattr(self, attr_name, None) is not None and attr_name[0] != "_" ) attrs = ",".join(attrs_list) return f"{type(self).__name__} [{attrs}]" def __iter__(self) -> Iterator[Any]: obj = self._selected_obj.set_axis(self._on) obj = self._create_data(obj) indexer = self._get_window_indexer() start, end = indexer.get_window_bounds( num_values=len(obj), min_periods=self.min_periods, center=self.center, closed=self.closed, step=self.step, ) self._check_window_bounds(start, end, len(obj)) for s, e in zip(start, end, strict=True): result = obj.iloc[slice(s, e)] yield result def _prep_values(self, values: ArrayLike) -> np.ndarray: """Convert input to numpy arrays for Cython routines""" if needs_i8_conversion(values.dtype) or is_arrow_temporal_dtype(values.dtype): # GH#66445 ArrowDtype timestamps/durations are not covered by # needs_i8_conversion, so without the second check they fell through # to the ExtensionArray branch below and were silently converted to # float64 counts instead of raising like their NumPy counterparts. raise NotImplementedError( f"ops for {type(self).__name__} for this " f"dtype {values.dtype} is not implemented" ) # GH #12373 : rolling functions error on float32 data # make sure the data is coerced to float64 try: if isinstance(values, ExtensionArray): values = values.to_numpy(np.float64, na_value=np.nan) else: values = ensure_float64(values) except (ValueError, TypeError) as err: raise TypeError(f"cannot handle this type -> {values.dtype}") from err # Convert inf to nan for C funcs inf = np.isinf(values) if inf.any(): values = np.where(inf, np.nan, values) return values def _insert_on_column(self, result: DataFrame, obj: DataFrame) -> None: # if we have an 'on' column we want to put it back into # the results in the same location from pandas import Series if self.on is not None and not self._on.equals(obj.index): name = self._on.name extra_col = Series(self._on, index=self.obj.index, name=name, copy=False) if name in result.columns: # TODO: sure we want to overwrite results? result[name] = extra_col elif name in result.index.names: pass elif name in self._selected_obj.columns: # insert in the same location as we had in _selected_obj old_cols = self._selected_obj.columns new_cols = result.columns old_loc = old_cols.get_loc(name) overlap = new_cols.intersection(old_cols[:old_loc]) new_loc = len(overlap) result.insert(new_loc, name, extra_col) else: # insert at the end result[name] = extra_col @property def _index_array(self) -> npt.NDArray[np.int64] | None: # TODO: why do we get here with e.g. MultiIndex? if isinstance(self._on, (PeriodIndex, DatetimeIndex, TimedeltaIndex)): return self._on.asi8 elif isinstance(self._on.dtype, ArrowDtype) and self._on.dtype.kind in "mM": return self._on.to_numpy(dtype=np.int64) return None def _resolve_output(self, out: DataFrame, obj: DataFrame) -> DataFrame: """Validate and finalize result.""" if out.shape[1] == 0 and obj.shape[1] > 0: raise DataError("No numeric types to aggregate") if out.shape[1] == 0: return obj.astype("float64") self._insert_on_column(out, obj) return out def _get_window_indexer(self) -> BaseIndexer: """ Return an indexer class that will compute the window start and end bounds """ 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, center=self.center, ) return FixedWindowIndexer(window_size=self.window) def _apply_series( self, homogeneous_func: Callable[..., ArrayLike], name: str | None = None ) -> Series: """ Series version of _apply_columnwise """ obj = self._create_data(self._selected_obj) if name == "count": # GH 12541: Special case for count where we support date-like types obj = notna(obj).astype(int) try: values = self._prep_values(obj._values) except (TypeError, NotImplementedError) as err: raise DataError("No numeric types to aggregate") from err result = homogeneous_func(values) index = self._slice_axis_for_step(obj.index, result) return obj._constructor(result, index=index, name=obj.name) def _apply_columnwise( self, homogeneous_func: Callable[..., ArrayLike], name: str, numeric_only: bool = False, ) -> DataFrame | Series: """ Apply the given function to the DataFrame broken down into homogeneous sub-frames. """ self._validate_numeric_only(name, numeric_only) if self._selected_obj.ndim == 1: return self._apply_series(homogeneous_func, name) obj = self._create_data(self._selected_obj, numeric_only) if name == "count": # GH 12541: Special case for count where we support date-like types obj = notna(obj).astype(int) obj._mgr = obj._mgr.consolidate() taker = [] res_values = [] for i, arr in enumerate(obj._iter_column_arrays()): # GH#42736 operate column-wise instead of block-wise # As of 2.0, hfunc will raise for nuisance columns try: arr = self._prep_values(arr) except (TypeError, NotImplementedError) as err: raise DataError( f"Cannot aggregate non-numeric type: {arr.dtype}" ) from err res = homogeneous_func(arr) res_values.append(res) taker.append(i) index = self._slice_axis_for_step( obj.index, res_values[0] if len(res_values) > 0 else None ) df = type(obj)._from_arrays( res_values, index=index, columns=obj.columns.take(taker), verify_integrity=False, ) return self._resolve_output(df, obj) def _apply_tablewise( self, homogeneous_func: Callable[..., ArrayLike], name: str | None = None, numeric_only: bool = False, ) -> DataFrame | Series: """ Apply the given function to the DataFrame across the entire object """ if self._selected_obj.ndim == 1: raise ValueError("method='table' not applicable for Series objects.") obj = self._create_data(self._selected_obj, numeric_only) values = self._prep_values(obj.to_numpy()) result = homogeneous_func(values) index = self._slice_axis_for_step(obj.index, result) columns = ( obj.columns if result.shape[1] == len(obj.columns) else obj.columns[:: self.step] ) out = obj._constructor(result, index=index, columns=columns) return self._resolve_output(out, obj) def _apply_pairwise( self, target: DataFrame | Series, other: DataFrame | Series | None, pairwise: bool | None, func: Callable[[DataFrame | Series, DataFrame | Series], DataFrame | Series], numeric_only: bool, ) -> DataFrame | Series: """ Apply the given pairwise function given 2 pandas objects (DataFrame/Series) """ target = self._create_data(target, numeric_only) if other is None: other = target # only default unset pairwise = True if pairwise is None else pairwise elif not isinstance(other, (ABCDataFrame, ABCSeries)): raise ValueError("other must be a DataFrame or Series") elif other.ndim == 2 and numeric_only: other = self._make_numeric_only(other) return flex_binary_moment(target, other, func, pairwise=bool(pairwise)) def _apply( self, func: Callable[..., Any], name: str, numeric_only: bool = False, numba_args: tuple[Any, ...] = (), **kwargs, ): """ Rolling statistical measure using supplied function. Designed to be used with passed-in Cython array-based functions. Parameters ---------- func : callable function to apply name : str, numba_args : tuple args to be passed when func is a numba func **kwargs additional arguments for rolling function and window function Returns ------- y : type of input """ window_indexer = self._get_window_indexer() min_periods = ( self.min_periods if self.min_periods is not None else window_indexer.window_size ) def homogeneous_func(values: np.ndarray): # calculation function if values.size == 0: return values.copy() def calc(x): start, end = window_indexer.get_window_bounds( num_values=len(x), min_periods=min_periods, center=self.center, closed=self.closed, step=self.step, ) self._check_window_bounds(start, end, len(x)) return func(x, start, end, min_periods, *numba_args) with np.errstate(all="ignore"): result = calc(values) return result if self.method == "single": return self._apply_columnwise(homogeneous_func, name, numeric_only) else: return self._apply_tablewise(homogeneous_func, name, numeric_only) def _numba_apply( self, func: Callable[..., Any], engine_kwargs: dict[str, bool] | None = None, **func_kwargs, ): window_indexer = self._get_window_indexer() min_periods = ( self.min_periods if self.min_periods is not None else window_indexer.window_size ) obj = self._create_data(self._selected_obj) values = self._prep_values(obj.to_numpy()) if values.ndim == 1: values = values.reshape(-1, 1) start, end = window_indexer.get_window_bounds( num_values=len(values), min_periods=min_periods, center=self.center, closed=self.closed, step=self.step, ) self._check_window_bounds(start, end, len(values)) # For now, map everything to float to match the Cython impl # even though it is wrong # TODO: Could preserve correct dtypes in future # xref #53214 dtype_mapping = executor.float_dtype_mapping aggregator = executor.generate_shared_aggregator( func, dtype_mapping, is_grouped_kernel=False, **get_jit_arguments(engine_kwargs), ) result = aggregator( values.T, start=start, end=end, min_periods=min_periods, **func_kwargs ).T index = self._slice_axis_for_step(obj.index, result) if obj.ndim == 1: result = result.squeeze() out = obj._constructor(result, index=index, name=obj.name) return out else: columns = self._slice_axis_for_step(obj.columns, result.T) out = obj._constructor(result, index=index, columns=columns) return self._resolve_output(out, obj) def aggregate(self, func=None, *args, **kwargs): relabeling, func, columns, order = reconstruct_func(func, **kwargs) result = ResamplerWindowApply(self, func, args=args, kwargs=kwargs).agg() if isinstance(result, ABCDataFrame) and relabeling: result = result.iloc[:, order] result.columns = columns if result is None: return self.apply(func, raw=False, args=args, kwargs=kwargs) return result agg = aggregate class BaseWindowGroupBy(BaseWindow): """ Provide the groupby windowing facilities. """ _grouper: BaseGrouper _as_index: bool _attributes: list[str] = ["_grouper"] def __init__( self, obj: DataFrame | Series, *args, _grouper: BaseGrouper, _as_index: bool = True, **kwargs, ) -> None: from pandas.core.groupby.ops import BaseGrouper if not isinstance(_grouper, BaseGrouper): raise ValueError("Must pass a BaseGrouper object.") self._grouper = _grouper self._as_index = _as_index # GH 32262: It's convention to keep the grouping column in # groupby., but unexpected to users in # groupby.rolling. obj = obj.drop(columns=self._grouper.names, errors="ignore") # GH 15354 if kwargs.get("step") is not None: raise NotImplementedError("step not implemented for groupby") super().__init__(obj, *args, **kwargs) def _apply( self, func: Callable[..., Any], name: str, numeric_only: bool = False, numba_args: tuple[Any, ...] = (), **kwargs, ) -> DataFrame | Series: result = super()._apply( func, name, numeric_only, numba_args, **kwargs, ) # Reconstruct the resulting MultiIndex # 1st set of levels = group by labels # 2nd set of levels = original DataFrame/Series index grouped_object_index = self.obj.index grouped_index_name = [*grouped_object_index.names] groupby_keys = copy.copy(self._grouper.names) result_index_names = groupby_keys + grouped_index_name drop_columns = [ key for key in self._grouper.names if key not in self.obj.index.names or key is None ] if len(drop_columns) != len(groupby_keys): # Our result will have still kept the column in the result result = result.drop(columns=drop_columns, errors="ignore") codes = self._grouper.codes levels = copy.copy(self._grouper.levels) group_indices = self._grouper.indices.values() if group_indices: indexer = np.concatenate(list(group_indices)) else: indexer = np.array([], dtype=np.intp) codes = [c.take(indexer) for c in codes] # if the index of the original dataframe needs to be preserved, append # this index (but reordered) to the codes/levels from the groupby if grouped_object_index is not None: idx = grouped_object_index.take(indexer) if not isinstance(idx, MultiIndex): idx = MultiIndex.from_arrays([idx]) codes.extend(list(idx.codes)) levels.extend(list(idx.levels)) result_index = MultiIndex( levels, codes, names=result_index_names, verify_integrity=False ) result.index = result_index if not self._as_index: result = result.reset_index(level=list(range(len(groupby_keys)))) return result def _apply_pairwise( self, target: DataFrame | Series, other: DataFrame | Series | None, pairwise: bool | None, func: Callable[[DataFrame | Series, DataFrame | Series], DataFrame | Series], numeric_only: bool, ) -> DataFrame | Series: """ Apply the given pairwise function given 2 pandas objects (DataFrame/Series) """ # Manually drop the grouping column first target = target.drop(columns=self._grouper.names, errors="ignore") result = super()._apply_pairwise(target, other, pairwise, func, numeric_only) # 1) Determine the levels + codes of the groupby levels if other is not None and not all( len(group) == len(other) for group in self._grouper.indices.values() ): # GH 42915 # len(other) != len(any group), so must reindex (expand) the result # from flex_binary_moment to a "transform"-like result # per groupby combination old_result_len = len(result) result = concat( [ result.take(gb_indices).reindex(result.index) for gb_indices in self._grouper.indices.values() ] ) gb_pairs = ( com.maybe_make_list(pair) for pair in self._grouper.indices.keys() ) groupby_codes = [] groupby_levels = [] # e.g. [[1, 2], [4, 5]] as [[1, 4], [2, 5]] for gb_level_pair in map(list, zip(*gb_pairs, strict=True)): labels = np.repeat(np.array(gb_level_pair), old_result_len) codes, levels = factorize(labels) groupby_codes.append(codes) groupby_levels.append(levels) else: # pairwise=True or len(other) == len(each group), so repeat # the groupby labels by the number of columns in the original object groupby_codes = self._grouper.codes # error: Incompatible types in assignment (expression has type # "List[Index]", variable has type "List[Union[ndarray, Index]]") groupby_levels = self._grouper.levels # type: ignore[assignment] group_indices = self._grouper.indices.values() if group_indices: indexer = np.concatenate(list(group_indices)) else: indexer = np.array([], dtype=np.intp) if target.ndim == 1: repeat_by = 1 else: repeat_by = len(target.columns) groupby_codes = [ np.repeat(c.take(indexer), repeat_by) for c in groupby_codes ] # 2) Determine the levels + codes of the result from super()._apply_pairwise if isinstance(result.index, MultiIndex): result_codes = list(result.index.codes) result_levels = list(result.index.levels) result_names = list(result.index.names) else: idx_codes, idx_levels = factorize(result.index) result_codes = [idx_codes] result_levels = [idx_levels] result_names = [result.index.name] # 3) Create the resulting index by combining 1) + 2) result_codes = groupby_codes + result_codes result_levels = groupby_levels + result_levels result_names = self._grouper.names + result_names result_index = MultiIndex( result_levels, result_codes, names=result_names, verify_integrity=False ) result.index = result_index return result def _create_data(self, obj: NDFrameT, numeric_only: bool = False) -> NDFrameT: """ Split data into blocks & return conformed data. """ # Ensure the object we're rolling over is monotonically sorted relative # to the groups # GH 36197 if not obj.empty: groupby_order = np.concatenate(list(self._grouper.indices.values())).astype( np.int64 ) obj = obj.take(groupby_order) return super()._create_data(obj, numeric_only) def _gotitem(self, key, ndim, subset=None): # we are setting the index on the actual object # here so our index is carried through to the selected obj # when we do the splitting for the groupby if self.on is not None: # GH 43355 subset = self.obj.set_index(self._on) return super()._gotitem(key, ndim, subset=subset) @set_module("pandas.api.typing") class Window(BaseWindow): """ Provide rolling window calculations. Parameters ---------- window : int, timedelta, str, offset, or BaseIndexer subclass Interval of the moving window. If an integer, the delta between the start and end of each window. The number of points in the window depends on the ``closed`` argument. If a timedelta, str, or offset, the time period of each window. Each window will be a variable sized based on the observations included in the time-period. This is only valid for datetimelike indexes. To learn more about the offsets & frequency strings, please see :ref:`this link`. If a BaseIndexer subclass, the window boundaries based on the defined ``get_window_bounds`` method. Additional rolling keyword arguments, namely ``min_periods``, ``center``, ``closed`` and ``step`` will be passed to ``get_window_bounds``. min_periods : int, default None Minimum number of observations in window required to have a value; otherwise, result is ``np.nan``. For a window that is specified by an offset, ``min_periods`` will default to 1. For a window that is specified by an integer, ``min_periods`` will default to the size of the window. center : bool, default False If False, set the window labels as the right edge of the window index. If True, set the window labels as the center of the window index. win_type : str, default None If ``None``, all points are evenly weighted. If a string, it must be a valid `scipy.signal window function `__. Certain Scipy window types require additional parameters to be passed in the aggregation function. The additional parameters must match the keywords specified in the Scipy window type method signature. on : str, optional For a DataFrame, a column label or Index level on which to calculate the rolling window, rather than the DataFrame's index. For integer ``window`` values, the window bounds are based on the number of observations and are not calculated using the values of the ``on`` column. The ``on`` column is excluded from the aggregation, but is included in the result when its values differ from the object's index. closed : str, default None Determines the inclusivity of points in the window If ``'right'``, uses the window (first, last] meaning the last point is included in the calculations. If ``'left'``, uses the window [first, last) meaning the first point is included in the calculations. If ``'both'``, uses the window [first, last] meaning all points in the window are included in the calculations. If ``'neither'``, uses the window (first, last) meaning the first and last points in the window are excluded from calculations. () and [] are referencing open and closed set notation respetively. Default ``None`` (``'right'``). step : int, default None Evaluate the window at every ``step`` result, equivalent to slicing as ``[::step]``. ``window`` must be an integer. Using a step argument other than None or 1 will produce a result with a different shape than the input. method : str {'single', 'table'}, default 'single' Execute the rolling operation per single column or row (``'single'``) or over the entire object (``'table'``). This argument is only implemented when specifying ``engine='numba'`` in the method call. Returns ------- pandas.api.typing.Window or pandas.api.typing.Rolling An instance of Window is returned if ``win_type`` is passed. Otherwise, an instance of Rolling is returned. See Also -------- expanding : Provides expanding transformations. ewm : Provides exponential weighted functions. Notes ----- See :ref:`Windowing Operations ` for further usage details and examples. Examples -------- >>> df = pd.DataFrame({"B": [0, 1, 2, np.nan, 4]}) >>> df B 0 0.0 1 1.0 2 2.0 3 NaN 4 4.0 **window** Rolling sum with a window length of 2 observations. >>> df.rolling(2).sum() B 0 NaN 1 1.0 2 3.0 3 NaN 4 NaN Rolling sum with a window span of 2 seconds. >>> df_time = pd.DataFrame( ... {"B": [0, 1, 2, np.nan, 4]}, ... index=[ ... pd.Timestamp("20130101 09:00:00"), ... pd.Timestamp("20130101 09:00:02"), ... pd.Timestamp("20130101 09:00:03"), ... pd.Timestamp("20130101 09:00:05"), ... pd.Timestamp("20130101 09:00:06"), ... ], ... ) >>> df_time B 2013-01-01 09:00:00 0.0 2013-01-01 09:00:02 1.0 2013-01-01 09:00:03 2.0 2013-01-01 09:00:05 NaN 2013-01-01 09:00:06 4.0 >>> df_time.rolling("2s").sum() B 2013-01-01 09:00:00 0.0 2013-01-01 09:00:02 1.0 2013-01-01 09:00:03 3.0 2013-01-01 09:00:05 NaN 2013-01-01 09:00:06 4.0 Rolling sum with forward looking windows with 2 observations. >>> indexer = pd.api.indexers.FixedForwardWindowIndexer(window_size=2) >>> df.rolling(window=indexer, min_periods=1).sum() B 0 1.0 1 3.0 2 2.0 3 4.0 4 4.0 **min_periods** Rolling sum with a window length of 2 observations, but only needs a minimum of 1 observation to calculate a value. >>> df.rolling(2, min_periods=1).sum() B 0 0.0 1 1.0 2 3.0 3 2.0 4 4.0 **center** Rolling sum with the result assigned to the center of the window index. >>> df.rolling(3, min_periods=1, center=True).sum() B 0 1.0 1 3.0 2 3.0 3 6.0 4 4.0 >>> df.rolling(3, min_periods=1, center=False).sum() B 0 0.0 1 1.0 2 3.0 3 3.0 4 6.0 **step** Rolling sum with a window length of 2 observations, minimum of 1 observation to calculate a value, and a step of 2. >>> df.rolling(2, min_periods=1, step=2).sum() B 0 0.0 2 3.0 4 4.0 **win_type** Rolling sum with a window length of 2, using the Scipy ``'gaussian'`` window type. ``std`` is required in the aggregation function. >>> df.rolling(2, win_type="gaussian").sum(std=3) B 0 NaN 1 0.986207 2 2.958621 3 NaN 4 NaN **on** Rolling sum with a window length of 2 days. >>> df = pd.DataFrame( ... { ... "A": [ ... pd.to_datetime("2020-01-01"), ... pd.to_datetime("2020-01-01"), ... pd.to_datetime("2020-01-02"), ... ], ... "B": [1, 2, 3], ... }, ... index=pd.date_range("2020", periods=3), ... ) >>> df A B 2020-01-01 2020-01-01 1 2020-01-02 2020-01-01 2 2020-01-03 2020-01-02 3 >>> df.rolling("2D", on="A").sum() A B 2020-01-01 2020-01-01 1.0 2020-01-02 2020-01-01 3.0 2020-01-03 2020-01-02 6.0 """ _attributes = [ "window", "min_periods", "center", "win_type", "on", "closed", "step", "method", ] def _validate(self) -> None: super()._validate() if not isinstance(self.win_type, str): raise ValueError(f"Invalid win_type {self.win_type}") signal = import_optional_dependency( "scipy.signal.windows", extra="Scipy is required to generate window weight." ) self._scipy_weight_generator = getattr(signal, self.win_type, None) if self._scipy_weight_generator is None: raise ValueError(f"Invalid win_type {self.win_type}") if isinstance(self.window, BaseIndexer): raise NotImplementedError( "BaseIndexer subclasses not implemented with win_types." ) if not is_integer(self.window) or self.window < 0: raise ValueError("window must be an integer 0 or greater") if self.method != "single": raise NotImplementedError("'single' is the only supported method type.") def _center_window(self, result: np.ndarray, offset: int) -> np.ndarray: """ Center the result in the window for weighted rolling aggregations. """ if offset > 0: lead_indexer = [slice(offset, None)] result = np.copy(result[tuple(lead_indexer)]) return result def _apply( self, func: Callable[[np.ndarray, int, int], np.ndarray], name: str, numeric_only: bool = False, numba_args: tuple[Any, ...] = (), **kwargs, ): """ Rolling with weights statistical measure using supplied function. Designed to be used with passed-in Cython array-based functions. Parameters ---------- func : callable function to apply name : str, numeric_only : bool, default False Whether to only operate on bool, int, and float columns numba_args : tuple unused **kwargs additional arguments for scipy windows if necessary Returns ------- y : type of input """ # "None" not callable [misc] window = self._scipy_weight_generator( # type: ignore[misc] self.window, **kwargs ) offset = (len(window) - 1) // 2 if self.center else 0 def homogeneous_func(values: np.ndarray): # calculation function if values.size == 0: return values.copy() def calc(x): additional_nans = np.full(offset, np.nan) x = np.concatenate((x, additional_nans)) return func( x, window, self.min_periods if self.min_periods is not None else len(window), ) with np.errstate(all="ignore"): # Our weighted aggregations return memoryviews result = np.asarray(calc(values)) if self.center: result = self._center_window(result, offset) return result return self._apply_columnwise(homogeneous_func, name, numeric_only)[ :: self.step ] def aggregate(self, func=None, *args, **kwargs): """ Aggregate using one or more operations over the specified axis. Parameters ---------- func : function, str, list or dict Function to use for aggregating the data. If a function, must either work when passed a Series/DataFrame or when passed to Series/DataFrame.apply. Accepted combinations are: - function - string function name - list of functions and/or function names, e.g. ``[np.sum, 'mean']`` - dict of axis labels -> functions, function names or list of such. *args Positional arguments to pass to `func`. **kwargs Keyword arguments to pass to `func`. Returns ------- scalar, Series or DataFrame The return can be: * scalar : when Series.agg is called with single function * Series : when DataFrame.agg is called with a single function * DataFrame : when DataFrame.agg is called with several functions See Also -------- DataFrame.aggregate : Similar DataFrame method. Series.aggregate : Similar Series method. Notes ----- The aggregation operations are always performed over an axis, either the index (default) or the column axis. This behavior is different from `numpy` aggregation functions (`mean`, `median`, `prod`, `sum`, `std`, `var`), where the default is to compute the aggregation of the flattened array, e.g., ``numpy.mean(arr_2d)`` as opposed to ``numpy.mean(arr_2d, axis=0)``. `agg` is an alias for `aggregate`. Use the alias. Functions that mutate the passed object can produce unexpected behavior or errors and are not supported. See :ref:`gotchas.udf-mutation` for more details. A passed user-defined-function will be passed a Series for evaluation. If ``func`` defines an index relabeling, ``axis`` must be ``0`` or ``index``. Examples -------- >>> df = pd.DataFrame({"A": [1, 2, 3], "B": [4, 5, 6], "C": [7, 8, 9]}) >>> df A B C 0 1 4 7 1 2 5 8 2 3 6 9 >>> df.rolling(2, win_type="boxcar").agg("mean") A B C 0 NaN NaN NaN 1 1.5 4.5 7.5 2 2.5 5.5 8.5 """ result = ResamplerWindowApply(self, func, args=args, kwargs=kwargs).agg() if result is None: # these must apply directly result = func(self) # pyright: ignore[reportOptionalCall] return result agg = aggregate def sum(self, numeric_only: bool = False, **kwargs): """ Calculate the rolling weighted window sum. The window type is determined by the ``win_type`` parameter specified in :meth:`DataFrame.rolling` or :meth:`Series.rolling`. Additional keyword arguments are passed to the SciPy window function. Parameters ---------- numeric_only : bool, default False Include only float, int, boolean columns. **kwargs Keyword arguments to configure the ``SciPy`` weighted window type. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.sum : Aggregating sum for Series. DataFrame.sum : Aggregating sum for DataFrame. Examples -------- >>> ser = pd.Series([0, 1, 5, 2, 8]) To get an instance of :class:`~pandas.core.window.rolling.Window` we need to pass the parameter `win_type`. >>> type(ser.rolling(2, win_type="gaussian")) In order to use the `SciPy` Gaussian window we need to provide the parameters `M` and `std`. The parameter `M` corresponds to 2 in our example. We pass the second parameter `std` as a parameter of the following method (`sum` in this case): >>> ser.rolling(2, win_type="gaussian").sum(std=3) 0 NaN 1 0.986207 2 5.917243 3 6.903450 4 9.862071 dtype: float64 """ window_func = window_aggregations.roll_weighted_sum # error: Argument 1 to "_apply" of "Window" has incompatible type # "Callable[[ndarray, ndarray, int], ndarray]"; expected # "Callable[[ndarray, int, int], ndarray]" return self._apply( window_func, # type: ignore[arg-type] name="sum", numeric_only=numeric_only, **kwargs, ) def mean(self, numeric_only: bool = False, **kwargs): """ Calculate the rolling weighted window mean. The window type is determined by the ``win_type`` parameter specified in :meth:`DataFrame.rolling` or :meth:`Series.rolling`. Additional keyword arguments are passed to the SciPy window function. Parameters ---------- numeric_only : bool, default False Include only float, int, boolean columns. **kwargs Keyword arguments to configure the ``SciPy`` weighted window type. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.mean : Aggregating mean for Series. DataFrame.mean : Aggregating mean for DataFrame. Examples -------- >>> ser = pd.Series([0, 1, 5, 2, 8]) To get an instance of :class:`~pandas.core.window.rolling.Window` we need to pass the parameter `win_type`. >>> type(ser.rolling(2, win_type="gaussian")) In order to use the `SciPy` Gaussian window we need to provide the parameters `M` and `std`. The parameter `M` corresponds to 2 in our example. We pass the second parameter `std` as a parameter of the following method: >>> ser.rolling(2, win_type="gaussian").mean(std=3) 0 NaN 1 0.5 2 3.0 3 3.5 4 5.0 dtype: float64 """ window_func = window_aggregations.roll_weighted_mean # error: Argument 1 to "_apply" of "Window" has incompatible type # "Callable[[ndarray, ndarray, int], ndarray]"; expected # "Callable[[ndarray, int, int], ndarray]" return self._apply( window_func, # type: ignore[arg-type] name="mean", numeric_only=numeric_only, **kwargs, ) def var(self, ddof: int = 1, numeric_only: bool = False, **kwargs): """ Calculate the rolling weighted window variance. The window type is determined by the ``win_type`` parameter specified in :meth:`DataFrame.rolling` or :meth:`Series.rolling`. Additional keyword arguments are passed to the SciPy window function. Parameters ---------- ddof : int, default 1 Delta Degrees of Freedom. The divisor used in calculations is ``N - ddof``, where ``N`` represents the number of elements. numeric_only : bool, default False Include only float, int, boolean columns. **kwargs Keyword arguments to configure the ``SciPy`` weighted window type. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.var : Aggregating var for Series. DataFrame.var : Aggregating var for DataFrame. Examples -------- >>> ser = pd.Series([0, 1, 5, 2, 8]) To get an instance of :class:`~pandas.core.window.rolling.Window` we need to pass the parameter `win_type`. >>> type(ser.rolling(2, win_type="gaussian")) In order to use the `SciPy` Gaussian window we need to provide the parameters `M` and `std`. The parameter `M` corresponds to 2 in our example. We pass the second parameter `std` as a parameter of the following method: >>> ser.rolling(2, win_type="gaussian").var(std=3) 0 NaN 1 0.5 2 8.0 3 4.5 4 18.0 dtype: float64 """ window_func = partial(window_aggregations.roll_weighted_var, ddof=ddof) kwargs.pop("name", None) return self._apply(window_func, name="var", numeric_only=numeric_only, **kwargs) def std(self, ddof: int = 1, numeric_only: bool = False, **kwargs): """ Calculate the rolling weighted window standard deviation. The window type is determined by the ``win_type`` parameter specified in :meth:`DataFrame.rolling` or :meth:`Series.rolling`. Additional keyword arguments are passed to the SciPy window function. Parameters ---------- ddof : int, default 1 Delta Degrees of Freedom. The divisor used in calculations is ``N - ddof``, where ``N`` represents the number of elements. numeric_only : bool, default False Include only float, int, boolean columns. **kwargs Keyword arguments to configure the ``SciPy`` weighted window type. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.std : Aggregating std for Series. DataFrame.std : Aggregating std for DataFrame. Examples -------- >>> ser = pd.Series([0, 1, 5, 2, 8]) To get an instance of :class:`~pandas.core.window.rolling.Window` we need to pass the parameter `win_type`. >>> type(ser.rolling(2, win_type="gaussian")) In order to use the `SciPy` Gaussian window we need to provide the parameters `M` and `std`. The parameter `M` corresponds to 2 in our example. We pass the second parameter `std` as a parameter of the following method: >>> ser.rolling(2, win_type="gaussian").std(std=3) 0 NaN 1 0.707107 2 2.828427 3 2.121320 4 4.242641 dtype: float64 """ return zsqrt( self.var(ddof=ddof, name="std", numeric_only=numeric_only, **kwargs) ) class RollingAndExpandingMixin(BaseWindow): def count(self, numeric_only: bool = False): window_func = window_aggregations.roll_sum return self._apply(window_func, name="count", numeric_only=numeric_only) def apply( self, func: Callable[..., Any], raw: bool = False, engine: Literal["cython", "numba"] | None = None, engine_kwargs: dict[str, bool] | None = None, args: tuple[Any, ...] | None = None, kwargs: dict[str, Any] | None = None, ): if args is None: args = () if kwargs is None: kwargs = {} if not is_bool(raw): raise ValueError("raw parameter must be `True` or `False`") numba_args: tuple[Any, ...] = () if maybe_use_numba(engine): if raw is False: raise ValueError("raw must be `True` when using the numba engine") numba_args, kwargs = prepare_function_arguments( func, args, kwargs, num_required_args=1 ) if self.method == "single": apply_func = generate_numba_apply_func( func, **get_jit_arguments(engine_kwargs) ) else: apply_func = generate_numba_table_func( func, **get_jit_arguments(engine_kwargs) ) elif engine in ("cython", None): if engine_kwargs is not None: raise ValueError("cython engine does not accept engine_kwargs") apply_func = self._generate_cython_apply_func(args, kwargs, raw, func) else: raise ValueError("engine must be either 'numba' or 'cython'") return self._apply( apply_func, name="apply", numba_args=numba_args, ) def _generate_cython_apply_func( self, args: tuple[Any, ...], kwargs: dict[str, Any], raw: bool | np.bool_, function: Callable[..., Any], ) -> Callable[[np.ndarray, np.ndarray, np.ndarray, int], np.ndarray]: from pandas import Series window_func = partial( window_aggregations.roll_apply, args=args, kwargs=kwargs, raw=bool(raw), function=function, ) def apply_func(values, begin, end, min_periods, raw=raw): if not raw: # GH 45912 values = Series(values, index=self._on, copy=False) return window_func(values, begin, end, min_periods) return apply_func @overload def pipe( self, func: Callable[Concatenate[Self, P], T], *args: P.args, **kwargs: P.kwargs, ) -> T: ... @overload def pipe( self, func: tuple[Callable[..., T], str], *args: Any, **kwargs: Any, ) -> T: ... def pipe( self, func: Callable[Concatenate[Self, P], T] | tuple[Callable[..., T], str], *args: Any, **kwargs: Any, ) -> T: return com.pipe(self, func, *args, **kwargs) def sum( self, numeric_only: bool = False, engine: Literal["cython", "numba"] | None = None, engine_kwargs: dict[str, bool] | None = None, ): if maybe_use_numba(engine): if self.method == "table": func = generate_manual_numpy_nan_agg_with_axis(np.nansum) return self.apply( func, raw=True, engine=engine, engine_kwargs=engine_kwargs, ) else: from pandas.core._numba.kernels import sliding_sum return self._numba_apply(sliding_sum, engine_kwargs) window_func = window_aggregations.roll_sum return self._apply(window_func, name="sum", numeric_only=numeric_only) def max( self, numeric_only: bool = False, engine: Literal["cython", "numba"] | None = None, engine_kwargs: dict[str, bool] | None = None, ): if maybe_use_numba(engine): if self.method == "table": func = generate_manual_numpy_nan_agg_with_axis(np.nanmax) return self.apply( func, raw=True, engine=engine, engine_kwargs=engine_kwargs, ) else: from pandas.core._numba.kernels import sliding_min_max return self._numba_apply(sliding_min_max, engine_kwargs, is_max=True) window_func = window_aggregations.roll_max return self._apply(window_func, name="max", numeric_only=numeric_only) def min( self, numeric_only: bool = False, engine: Literal["cython", "numba"] | None = None, engine_kwargs: dict[str, bool] | None = None, ): if maybe_use_numba(engine): if self.method == "table": func = generate_manual_numpy_nan_agg_with_axis(np.nanmin) return self.apply( func, raw=True, engine=engine, engine_kwargs=engine_kwargs, ) else: from pandas.core._numba.kernels import sliding_min_max return self._numba_apply(sliding_min_max, engine_kwargs, is_max=False) window_func = window_aggregations.roll_min return self._apply(window_func, name="min", numeric_only=numeric_only) def mean( self, numeric_only: bool = False, engine: Literal["cython", "numba"] | None = None, engine_kwargs: dict[str, bool] | None = None, ): if maybe_use_numba(engine): if self.method == "table": func = generate_manual_numpy_nan_agg_with_axis(np.nanmean) return self.apply( func, raw=True, engine=engine, engine_kwargs=engine_kwargs, ) else: from pandas.core._numba.kernels import sliding_mean return self._numba_apply(sliding_mean, engine_kwargs) window_func = window_aggregations.roll_mean return self._apply(window_func, name="mean", numeric_only=numeric_only) def median( self, numeric_only: bool = False, engine: Literal["cython", "numba"] | None = None, engine_kwargs: dict[str, bool] | None = None, ): if maybe_use_numba(engine): if self.method == "table": func = generate_manual_numpy_nan_agg_with_axis(np.nanmedian) else: func = np.nanmedian return self.apply( func, raw=True, engine=engine, engine_kwargs=engine_kwargs, ) window_func = window_aggregations.roll_median_c return self._apply(window_func, name="median", numeric_only=numeric_only) def std( self, ddof: int = 1, numeric_only: bool = False, engine: Literal["cython", "numba"] | None = None, engine_kwargs: dict[str, bool] | None = None, ): if maybe_use_numba(engine): if self.method == "table": raise NotImplementedError("std not supported with method='table'") from pandas.core._numba.kernels import sliding_var return zsqrt(self._numba_apply(sliding_var, engine_kwargs, ddof=ddof)) window_func = window_aggregations.roll_var def zsqrt_func(values, begin, end, min_periods): return zsqrt(window_func(values, begin, end, min_periods, ddof=ddof)) return self._apply( zsqrt_func, name="std", numeric_only=numeric_only, ) def var( self, ddof: int = 1, numeric_only: bool = False, engine: Literal["cython", "numba"] | None = None, engine_kwargs: dict[str, bool] | None = None, ): if maybe_use_numba(engine): if self.method == "table": raise NotImplementedError("var not supported with method='table'") from pandas.core._numba.kernels import sliding_var return self._numba_apply(sliding_var, engine_kwargs, ddof=ddof) window_func = partial(window_aggregations.roll_var, ddof=ddof) return self._apply( window_func, name="var", numeric_only=numeric_only, ) def skew(self, numeric_only: bool = False): window_func = window_aggregations.roll_skew return self._apply( window_func, name="skew", numeric_only=numeric_only, ) def sem(self, ddof: int = 1, numeric_only: bool = False): # Raise here so error message says sem instead of std self._validate_numeric_only("sem", numeric_only) return self.std(numeric_only=numeric_only, ddof=ddof) / ( self.count(numeric_only=numeric_only) ).pow(0.5) def kurt(self, numeric_only: bool = False): window_func = window_aggregations.roll_kurt return self._apply( window_func, name="kurt", numeric_only=numeric_only, ) def first(self, numeric_only: bool = False): window_func = window_aggregations.roll_first return self._apply( window_func, name="first", numeric_only=numeric_only, ) def last(self, numeric_only: bool = False): window_func = window_aggregations.roll_last return self._apply( window_func, name="last", numeric_only=numeric_only, ) def quantile( self, q: float, interpolation: QuantileInterpolation = "linear", numeric_only: bool = False, ): if q == 1.0: window_func = window_aggregations.roll_max elif q == 0.0: window_func = window_aggregations.roll_min else: window_func = partial( window_aggregations.roll_quantile, quantile=q, interpolation=interpolation, ) return self._apply(window_func, name="quantile", numeric_only=numeric_only) def rank( self, method: WindowingRankType = "average", ascending: bool = True, pct: bool = False, numeric_only: bool = False, ): window_func = partial( window_aggregations.roll_rank, method=method, ascending=ascending, percentile=pct, ) return self._apply(window_func, name="rank", numeric_only=numeric_only) def nunique( self, numeric_only: bool = False, ): window_func = partial( window_aggregations.roll_nunique, ) return self._apply(window_func, name="nunique", numeric_only=numeric_only) def cov( self, other: DataFrame | Series | None = None, pairwise: bool | None = None, ddof: int = 1, numeric_only: bool = False, ): if self.step is not None: raise NotImplementedError("step not implemented for cov") self._validate_numeric_only("cov", numeric_only) from pandas import Series def cov_func(x, y): x_array = self._prep_values(x) y_array = self._prep_values(y) window_indexer = self._get_window_indexer() min_periods = ( self.min_periods if self.min_periods is not None else window_indexer.window_size ) start, end = window_indexer.get_window_bounds( num_values=len(x_array), min_periods=min_periods, center=self.center, closed=self.closed, step=self.step, ) self._check_window_bounds(start, end, len(x_array)) result = window_aggregations.roll_cov( x_array, y_array, start, end, min_periods, ddof ) return Series(result, index=x.index, name=x.name, copy=False) return self._apply_pairwise( self._selected_obj, other, pairwise, cov_func, numeric_only ) def corr( self, other: DataFrame | Series | None = None, pairwise: bool | None = None, ddof: int = 1, numeric_only: bool = False, ): if self.step is not None: raise NotImplementedError("step not implemented for corr") self._validate_numeric_only("corr", numeric_only) from pandas import Series def corr_func(x, y): x_array = self._prep_values(x) y_array = self._prep_values(y) window_indexer = self._get_window_indexer() min_periods = ( self.min_periods if self.min_periods is not None else window_indexer.window_size ) start, end = window_indexer.get_window_bounds( num_values=len(x_array), min_periods=min_periods, center=self.center, closed=self.closed, step=self.step, ) self._check_window_bounds(start, end, len(x_array)) result = window_aggregations.roll_corr( x_array, y_array, start, end, min_periods ) return Series(result, index=x.index, name=x.name, copy=False) return self._apply_pairwise( self._selected_obj, other, pairwise, corr_func, numeric_only ) @set_module("pandas.api.typing") class Rolling(RollingAndExpandingMixin): _attributes: list[str] = [ "window", "min_periods", "center", "win_type", "on", "closed", "step", "method", ] def _validate(self) -> None: super()._validate() # we allow rolling on a datetimelike index if ( self.obj.empty or isinstance(self._on, (DatetimeIndex, TimedeltaIndex, PeriodIndex)) or (isinstance(self._on.dtype, ArrowDtype) and self._on.dtype.kind in "mM") ) and isinstance(self.window, (str, BaseOffset, timedelta)): self._validate_datetimelike_monotonic() # this will raise ValueError on non-fixed freqs try: freq = to_offset(self.window) except (TypeError, ValueError) as err: raise ValueError( f"passed window {self.window} is not " "compatible with a datetimelike index" ) from err if isinstance(self._on, PeriodIndex): # error: Incompatible types in assignment (expression has type # "float", variable has type "Optional[int]") self._win_freq_i8 = freq.nanos / ( # type: ignore[assignment] self._on.freq.nanos / self._on.freq.n ) else: try: unit = dtype_to_unit(self._on.dtype) # type: ignore[arg-type] except TypeError: # if not a datetime dtype, eg for empty dataframes unit = "ns" unit = cast("TimeUnit", unit) self._win_freq_i8 = Timedelta(freq.nanos).as_unit(unit)._value # min_periods must be an integer if self.min_periods is None: self.min_periods = 1 if self.step is not None: raise NotImplementedError( "step is not supported with frequency windows" ) elif isinstance(self.window, BaseIndexer): # Passed BaseIndexer subclass should handle all other rolling kwargs pass elif not is_integer(self.window) or self.window < 0: raise ValueError("window must be an integer 0 or greater") def _validate_datetimelike_monotonic(self) -> None: """ Validate self._on is monotonic (increasing or decreasing) and has no NaT values for frequency windows. """ if self._on.hasnans: self._raise_monotonic_error("values must not have NaT") if not (self._on.is_monotonic_increasing or self._on.is_monotonic_decreasing): self._raise_monotonic_error("values must be monotonic") def _raise_monotonic_error(self, msg: str): on = self.on if on is None: on = "index" raise ValueError(f"{on} {msg}") def aggregate(self, func=None, *args, **kwargs): """ Aggregate using one or more operations over the specified axis. This method allows combining multiple aggregation functions (e.g. ``'sum'``, ``'mean'``) in a single call, returning a result for each function applied to each rolling window. Parameters ---------- func : function, str, list or dict Function to use for aggregating the data. If a function, must either work when passed a Series/Dataframe or when passed to Series/Dataframe.apply. Accepted combinations are: - function - string function name - list of functions and/or function names, e.g. ``[np.sum, 'mean']`` - dict of axis labels -> functions, function names or list of such. *args Positional arguments to pass to `func`. **kwargs Keyword arguments to pass to `func`. Returns ------- scalar, Series or DataFrame The return can be: * scalar : when Series.agg is called with single function * Series : when DataFrame.agg is called with a single function * DataFrame : when DataFrame.agg is called with several functions See Also -------- Series.rolling : Calling object with Series data. DataFrame.rolling : Calling object with DataFrame data. Notes ----- The aggregation operations are always performed over an axis, either the index (default) or the column axis. This behavior is different from `numpy` aggregation functions (`mean`, `median`, `prod`, `sum`, `std`, `var`), where the default is to compute the aggregation of the flattened array, e.g., ``numpy.mean(arr_2d)`` as opposed to ``numpy.mean(arr_2d, axis=0)``. `agg` is an alias for `aggregate`. Use the alias. Functions that mutate the passed object can produce unexpected behavior or errors and are not supported. See :ref:`gotchas.udf-mutation` for more details. A passed user-defined-function will be passed a Series for evaluation. If ``func`` defines an index relabeling, ``axis`` must be ``0`` or ``index``. Examples -------- >>> df = pd.DataFrame({"A": [1, 2, 3], "B": [4, 5, 6], "C": [7, 8, 9]}) >>> df A B C 0 1 4 7 1 2 5 8 2 3 6 9 >>> df.rolling(2).sum() A B C 0 NaN NaN NaN 1 3.0 9.0 15.0 2 5.0 11.0 17.0 >>> df.rolling(2).agg({"A": "sum", "B": "min"}) A B 0 NaN NaN 1 3.0 4.0 2 5.0 5.0 """ return super().aggregate(func, *args, **kwargs) agg = aggregate def count(self, numeric_only: bool = False): """ Calculate the rolling count of non NaN observations. This is useful for identifying windows with missing data, as it counts only non-NaN entries within each window. Parameters ---------- numeric_only : bool, default False Include only float, int, boolean columns. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.count : Aggregating count for Series. DataFrame.count : Aggregating count for DataFrame. Examples -------- >>> s = pd.Series([2, 3, np.nan, 10]) >>> s.rolling(2).count() 0 NaN 1 2.0 2 1.0 3 1.0 dtype: float64 >>> s.rolling(3).count() 0 NaN 1 NaN 2 2.0 3 2.0 dtype: float64 >>> s.rolling(4).count() 0 NaN 1 NaN 2 NaN 3 3.0 dtype: float64 """ return super().count(numeric_only) def apply( self, func: Callable[..., Any], raw: bool = False, engine: Literal["cython", "numba"] | None = None, engine_kwargs: dict[str, bool] | None = None, args: tuple[Any, ...] | None = None, kwargs: dict[str, Any] | None = None, ): """ Calculate the rolling custom aggregation function. Applies an arbitrary function to each rolling window. This is useful when the built-in rolling methods do not cover the desired computation. Parameters ---------- func : function Must produce a single value from an ndarray input if ``raw=True`` or a single value from a Series if ``raw=False``. Can also accept a Numba JIT function with ``engine='numba'`` specified. raw : bool, default False * ``False`` : passes each row or column as a Series to the function. * ``True`` : the passed function will receive ndarray objects instead. Pandas-only attributes such as ``.iloc`` or ``.index`` are not available on ndarrays and will raise ``AttributeError`` if used inside ``func``. If you are just applying a NumPy reduction function this will achieve much better performance. engine : str, default None * ``'cython'`` : Runs rolling apply through C-extensions from cython. * ``'numba'`` : Runs rolling apply through JIT compiled code from numba. Only available when ``raw`` is set to ``True``. * ``None`` : Defaults to ``'cython'`` or globally setting ``compute.use_numba``. engine_kwargs : dict, default None * For ``'cython'`` engine, there are no accepted ``engine_kwargs`` * For ``'numba'`` engine, the engine can accept ``nogil`` and ``parallel`` dictionary keys. The values must either be ``True`` or ``False``. The default ``engine_kwargs`` for the ``'numba'`` engine is ``{'nogil': False, 'parallel': False}`` and will be applied to both the ``func`` and the ``apply`` rolling aggregation. args : tuple, default None Positional arguments to be passed into func. kwargs : dict, default None Keyword arguments to be passed into func. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.apply : Aggregating apply for Series. DataFrame.apply : Aggregating apply for DataFrame. Notes ----- When ``raw=False``, the :class:`Series` passed to ``func`` is indexed by the column or :class:`Index` used to compute the rolling window. If :meth:`DataFrame.rolling` was called with ``on=col``, the index of the passed :class:`Series` will be the values of ``col`` rather than the original :class:`DataFrame` index. When ``on`` is not specified, the index of the passed :class:`Series` is the original index of the input. Examples -------- >>> ser = pd.Series([1, 6, 5, 4]) >>> ser.rolling(2).apply(lambda s: s.sum() - s.min()) 0 NaN 1 6.0 2 6.0 3 5.0 dtype: float64 By default ``func`` is applied to each column independently. Use ``method="table"`` (which requires ``engine="numba"`` and ``raw=True``) to instead pass the whole :class:`DataFrame` window to ``func`` as a single 2D ndarray, so the function can combine multiple columns. Here a rolling dot product of columns ``A`` and ``B`` is computed; the scalar result is broadcast across the columns. >>> df = pd.DataFrame({"A": [1, 2, 3, 4, 5], "B": [5, 4, 3, 2, 1]}) >>> def dot(window): ... return (window[:, 0] * window[:, 1]).sum() >>> df.rolling(3, method="table").apply( ... dot, engine="numba", raw=True ... ) # doctest: +SKIP A B 0 NaN NaN 1 NaN NaN 2 22.0 22.0 3 25.0 25.0 4 22.0 22.0 """ return super().apply( func, raw=raw, engine=engine, engine_kwargs=engine_kwargs, args=args, kwargs=kwargs, ) @overload def pipe( self, func: Callable[Concatenate[Self, P], T], *args: P.args, **kwargs: P.kwargs, ) -> T: ... @overload def pipe( self, func: tuple[Callable[..., T], str], *args: Any, **kwargs: Any, ) -> T: ... @final def pipe( self, func: Callable[Concatenate[Self, P], T] | tuple[Callable[..., T], str], *args: Any, **kwargs: Any, ) -> T: """ Apply a ``func`` with arguments to this Rolling object and return its result. Use `.pipe` when you want to improve readability by chaining together functions that expect Series, DataFrames, GroupBy, Rolling, Expanding or Resampler objects. Instead of writing >>> h = lambda x, arg2, arg3: x + 1 - arg2 * arg3 >>> g = lambda x, arg1: x * 5 / arg1 >>> f = lambda x: x**4 >>> df = pd.DataFrame( ... {"A": [1, 2, 3, 4]}, index=pd.date_range("2012-08-02", periods=4) ... ) >>> h(g(f(df.rolling("2D")), arg1=1), arg2=2, arg3=3) # doctest: +SKIP You can write >>> ( ... df.rolling("2D").pipe(f).pipe(g, arg1=1).pipe(h, arg2=2, arg3=3) ... ) # doctest: +SKIP which is much more readable. Parameters ---------- func : callable or tuple of (callable, str) Function to apply to this Rolling object or, alternatively, a `(callable, data_keyword)` tuple where `data_keyword` is a string indicating the keyword of `callable` that expects the Rolling object. *args : iterable, optional Positional arguments passed into `func`. **kwargs : dict, optional A dictionary of keyword arguments passed into `func`. Returns ------- Rolling The original object with the function `func` applied. See Also -------- Series.pipe : Apply a function with arguments to a series. DataFrame.pipe: Apply a function with arguments to a dataframe. apply : Apply function to each group instead of to the full Rolling object. Notes ----- See more `here `__. Examples -------- >>> df = pd.DataFrame( ... {"A": [1, 2, 3, 4]}, index=pd.date_range("2012-08-02", periods=4) ... ) >>> df A 2012-08-02 1 2012-08-03 2 2012-08-04 3 2012-08-05 4 To get the difference between each rolling 2-day window's maximum and minimum value in one pass, you can do >>> df.rolling("2D").pipe(lambda x: x.max() - x.min()) A 2012-08-02 0.0 2012-08-03 1.0 2012-08-04 1.0 2012-08-05 1.0 """ return super().pipe(func, *args, **kwargs) def sum( self, numeric_only: bool = False, engine: Literal["cython", "numba"] | None = None, engine_kwargs: dict[str, bool] | None = None, ): """ Calculate the rolling sum. This is equivalent to applying ``numpy.sum`` over each rolling window. Missing values (NaN) are excluded from the calculation. Parameters ---------- numeric_only : bool, default False Include only float, int, boolean columns. engine : str, default None * ``'cython'`` : Runs the operation through C-extensions from cython. * ``'numba'`` : Runs the operation through JIT compiled code from numba. * ``None`` : Defaults to ``'cython'`` or globally setting ``compute.use_numba`` engine_kwargs : dict, default None * For ``'cython'`` engine, there are no accepted ``engine_kwargs`` * For ``'numba'`` engine, the engine can accept ``nogil`` and ``parallel`` dictionary keys. The values must either be ``True`` or ``False``. The default ``engine_kwargs`` for the ``'numba'`` engine is ``{'nogil': False, 'parallel': False}``. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.sum : Aggregating sum for Series. DataFrame.sum : Aggregating sum for DataFrame. Notes ----- See :ref:`window.numba_engine` and :ref:`enhancingperf.numba` for extended documentation and performance considerations for the Numba engine. Examples -------- >>> s = pd.Series([1, 2, 3, 4, 5]) >>> s 0 1 1 2 2 3 3 4 4 5 dtype: int64 >>> s.rolling(3).sum() 0 NaN 1 NaN 2 6.0 3 9.0 4 12.0 dtype: float64 >>> s.rolling(3, center=True).sum() 0 NaN 1 6.0 2 9.0 3 12.0 4 NaN dtype: float64 For DataFrame, each sum is computed column-wise. >>> df = pd.DataFrame({"A": s, "B": s**2}) >>> df A B 0 1 1 1 2 4 2 3 9 3 4 16 4 5 25 >>> df.rolling(3).sum() A B 0 NaN NaN 1 NaN NaN 2 6.0 14.0 3 9.0 29.0 4 12.0 50.0 """ return super().sum( numeric_only=numeric_only, engine=engine, engine_kwargs=engine_kwargs, ) def max( self, numeric_only: bool = False, *args, engine: Literal["cython", "numba"] | None = None, engine_kwargs: dict[str, bool] | None = None, **kwargs, ): """ Calculate the rolling maximum. This is equivalent to applying ``numpy.max`` over each rolling window. Parameters ---------- numeric_only : bool, default False Include only float, int, boolean columns. *args : iterable, optional Positional arguments passed into ``func``. engine : str, default None * ``'cython'`` : Runs the operation through C-extensions from cython. * ``'numba'`` : Runs the operation through JIT compiled code from numba. * ``None`` : Defaults to ``'cython'`` or globally setting ``compute.use_numba`` engine_kwargs : dict, default None * For ``'cython'`` engine, there are no accepted ``engine_kwargs`` * For ``'numba'`` engine, the engine can accept ``nogil`` and ``parallel`` dictionary keys. The values must either be ``True`` or ``False``. The default ``engine_kwargs`` for the ``'numba'`` engine is ``{'nogil': False, 'parallel': False}``. **kwargs : mapping, optional A dictionary of keyword arguments passed into ``func``. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.max : Aggregating max for Series. DataFrame.max : Aggregating max for DataFrame. Notes ----- See :ref:`window.numba_engine` and :ref:`enhancingperf.numba` for extended documentation and performance considerations for the Numba engine. Examples -------- >>> ser = pd.Series([1, 2, 3, 4]) >>> ser.rolling(2).max() 0 NaN 1 2.0 2 3.0 3 4.0 dtype: float64 """ return super().max( numeric_only=numeric_only, engine=engine, engine_kwargs=engine_kwargs, ) def min( self, numeric_only: bool = False, engine: Literal["cython", "numba"] | None = None, engine_kwargs: dict[str, bool] | None = None, ): """ Calculate the rolling minimum. This is equivalent to applying ``numpy.min`` over each rolling window. Parameters ---------- numeric_only : bool, default False Include only float, int, boolean columns. engine : str, default None * ``'cython'`` : Runs the operation through C-extensions from cython. * ``'numba'`` : Runs the operation through JIT compiled code from numba. * ``None`` : Defaults to ``'cython'`` or globally setting ``compute.use_numba`` engine_kwargs : dict, default None * For ``'cython'`` engine, there are no accepted ``engine_kwargs`` * For ``'numba'`` engine, the engine can accept ``nogil`` and ``parallel`` dictionary keys. The values must either be ``True`` or ``False``. The default ``engine_kwargs`` for the ``'numba'`` engine is ``{'nogil': False, 'parallel': False}``. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.min : Aggregating min for Series. DataFrame.min : Aggregating min for DataFrame. Notes ----- See :ref:`window.numba_engine` and :ref:`enhancingperf.numba` for extended documentation and performance considerations for the Numba engine. Examples -------- Performing a rolling minimum with a window size of 3. >>> s = pd.Series([4, 3, 5, 2, 6]) >>> s.rolling(3).min() 0 NaN 1 NaN 2 3.0 3 2.0 4 2.0 dtype: float64 """ return super().min( numeric_only=numeric_only, engine=engine, engine_kwargs=engine_kwargs, ) def mean( self, numeric_only: bool = False, engine: Literal["cython", "numba"] | None = None, engine_kwargs: dict[str, bool] | None = None, ): """ Calculate the rolling mean. This is equivalent to applying ``numpy.mean`` over each rolling window. Parameters ---------- numeric_only : bool, default False Include only float, int, boolean columns. engine : str, default None * ``'cython'`` : Runs the operation through C-extensions from cython. * ``'numba'`` : Runs the operation through JIT compiled code from numba. * ``None`` : Defaults to ``'cython'`` or globally setting ``compute.use_numba`` engine_kwargs : dict, default None * For ``'cython'`` engine, there are no accepted ``engine_kwargs`` * For ``'numba'`` engine, the engine can accept ``nogil`` and ``parallel`` dictionary keys. The values must either be ``True`` or ``False``. The default ``engine_kwargs`` for the ``'numba'`` engine is ``{'nogil': False, 'parallel': False}``. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.mean : Aggregating mean for Series. DataFrame.mean : Aggregating mean for DataFrame. Notes ----- See :ref:`window.numba_engine` and :ref:`enhancingperf.numba` for extended documentation and performance considerations for the Numba engine. Examples -------- The below examples will show rolling mean calculations with window sizes of two and three, respectively. >>> s = pd.Series([1, 2, 3, 4]) >>> s.rolling(2).mean() 0 NaN 1 1.5 2 2.5 3 3.5 dtype: float64 >>> s.rolling(3).mean() 0 NaN 1 NaN 2 2.0 3 3.0 dtype: float64 """ return super().mean( numeric_only=numeric_only, engine=engine, engine_kwargs=engine_kwargs, ) def median( self, numeric_only: bool = False, engine: Literal["cython", "numba"] | None = None, engine_kwargs: dict[str, bool] | None = None, ): """ Calculate the rolling median. This is equivalent to applying ``numpy.median`` over each rolling window. Parameters ---------- numeric_only : bool, default False Include only float, int, boolean columns. engine : str, default None * ``'cython'`` : Runs the operation through C-extensions from cython. * ``'numba'`` : Runs the operation through JIT compiled code from numba. * ``None`` : Defaults to ``'cython'`` or globally setting ``compute.use_numba`` engine_kwargs : dict, default None * For ``'cython'`` engine, there are no accepted ``engine_kwargs`` * For ``'numba'`` engine, the engine can accept ``nogil`` and ``parallel`` dictionary keys. The values must either be ``True`` or ``False``. The default ``engine_kwargs`` for the ``'numba'`` engine is ``{'nogil': False, 'parallel': False}``. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.median : Aggregating median for Series. DataFrame.median : Aggregating median for DataFrame. Notes ----- See :ref:`window.numba_engine` and :ref:`enhancingperf.numba` for extended documentation and performance considerations for the Numba engine. Examples -------- Compute the rolling median of a series with a window size of 3. >>> s = pd.Series([0, 1, 2, 3, 4]) >>> s.rolling(3).median() 0 NaN 1 NaN 2 1.0 3 2.0 4 3.0 dtype: float64 """ return super().median( numeric_only=numeric_only, engine=engine, engine_kwargs=engine_kwargs, ) def std( self, ddof: int = 1, numeric_only: bool = False, engine: Literal["cython", "numba"] | None = None, engine_kwargs: dict[str, bool] | None = None, ): """ Calculate the rolling standard deviation. This is equivalent to applying ``numpy.std`` (with the specified ``ddof``) over each rolling window. Parameters ---------- ddof : int, default 1 Delta Degrees of Freedom. The divisor used in calculations is ``N - ddof``, where ``N`` represents the number of elements. numeric_only : bool, default False Include only float, int, boolean columns. engine : str, default None * ``'cython'`` : Runs the operation through C-extensions from cython. * ``'numba'`` : Runs the operation through JIT compiled code from numba. * ``None`` : Defaults to ``'cython'`` or globally setting ``compute.use_numba`` engine_kwargs : dict, default None * For ``'cython'`` engine, there are no accepted ``engine_kwargs`` * For ``'numba'`` engine, the engine can accept ``nogil`` and ``parallel`` dictionary keys. The values must either be ``True`` or ``False``. The default ``engine_kwargs`` for the ``'numba'`` engine is ``{'nogil': False, 'parallel': False}``. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- numpy.std : Equivalent method for NumPy array. Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.std : Aggregating std for Series. DataFrame.std : Aggregating std for DataFrame. Notes ----- The default ``ddof`` of 1 used in :meth:`Series.std` is different than the default ``ddof`` of 0 in :func:`numpy.std`. A minimum of one period is required for the rolling calculation. Examples -------- >>> s = pd.Series([5, 5, 6, 7, 5, 5, 5]) >>> s.rolling(3).std() 0 NaN 1 NaN 2 0.577350 3 1.000000 4 1.000000 5 1.154701 6 0.000000 dtype: float64 """ return super().std( ddof=ddof, numeric_only=numeric_only, engine=engine, engine_kwargs=engine_kwargs, ) def var( self, ddof: int = 1, numeric_only: bool = False, engine: Literal["cython", "numba"] | None = None, engine_kwargs: dict[str, bool] | None = None, ): """ Calculate the rolling variance. This is equivalent to applying ``numpy.var`` (with the specified ``ddof``) over each rolling window. Parameters ---------- ddof : int, default 1 Delta Degrees of Freedom. The divisor used in calculations is ``N - ddof``, where ``N`` represents the number of elements. numeric_only : bool, default False Include only float, int, boolean columns. engine : str, default None * ``'cython'`` : Runs the operation through C-extensions from cython. * ``'numba'`` : Runs the operation through JIT compiled code from numba. * ``None`` : Defaults to ``'cython'`` or globally setting ``compute.use_numba`` engine_kwargs : dict, default None * For ``'cython'`` engine, there are no accepted ``engine_kwargs`` * For ``'numba'`` engine, the engine can accept ``nogil`` and ``parallel`` dictionary keys. The values must either be ``True`` or ``False``. The default ``engine_kwargs`` for the ``'numba'`` engine is ``{'nogil': False, 'parallel': False}``. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- numpy.var : Equivalent method for NumPy array. Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.var : Aggregating var for Series. DataFrame.var : Aggregating var for DataFrame. Notes ----- The default ``ddof`` of 1 used in :meth:`Series.var` is different than the default ``ddof`` of 0 in :func:`numpy.var`. A minimum of one period is required for the rolling calculation. Examples -------- >>> s = pd.Series([5, 5, 6, 7, 5, 5, 5]) >>> s.rolling(3).var() 0 NaN 1 NaN 2 0.333333 3 1.000000 4 1.000000 5 1.333333 6 0.000000 dtype: float64 """ return super().var( ddof=ddof, numeric_only=numeric_only, engine=engine, engine_kwargs=engine_kwargs, ) def skew(self, numeric_only: bool = False): """ Calculate the rolling unbiased skewness. This is equivalent to applying ``scipy.stats.skew`` over each rolling window. A minimum of three periods is required. Parameters ---------- numeric_only : bool, default False Include only float, int, boolean columns. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- scipy.stats.skew : Third moment of a probability density. Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.skew : Aggregating skew for Series. DataFrame.skew : Aggregating skew for DataFrame. Notes ----- A minimum of three periods is required for the rolling calculation. Examples -------- >>> ser = pd.Series([1, 5, 2, 7, 15, 6]) >>> ser.rolling(3).skew().round(6) 0 NaN 1 NaN 2 1.293343 3 -0.585583 4 0.670284 5 1.652317 dtype: float64 """ return super().skew(numeric_only=numeric_only) def sem(self, ddof: int = 1, numeric_only: bool = False): """ Calculate the rolling standard error of mean. This is computed as the rolling standard deviation divided by the square root of the rolling count. A minimum of one period is required. Parameters ---------- ddof : int, default 1 Delta Degrees of Freedom. The divisor used in calculations is ``N - ddof``, where ``N`` represents the number of elements. numeric_only : bool, default False Include only float, int, boolean columns. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.sem : Aggregating sem for Series. DataFrame.sem : Aggregating sem for DataFrame. Notes ----- A minimum of one period is required for the calculation. Examples -------- >>> s = pd.Series([0, 1, 2, 3]) >>> s.rolling(2, min_periods=1).sem() 0 NaN 1 0.5 2 0.5 3 0.5 dtype: float64 """ # Raise here so error message says sem instead of std self._validate_numeric_only("sem", numeric_only) return self.std(numeric_only=numeric_only, ddof=ddof) / ( self.count(numeric_only) ).pow(0.5) def kurt(self, numeric_only: bool = False): """ Calculate the rolling Fisher's definition of kurtosis without bias. This is equivalent to applying ``scipy.stats.kurtosis`` (with ``bias=False``) over each rolling window. A minimum of four periods is required. Parameters ---------- numeric_only : bool, default False Include only float, int, boolean columns. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- scipy.stats.kurtosis : Reference SciPy method. Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.kurt : Aggregating kurt for Series. DataFrame.kurt : Aggregating kurt for DataFrame. Notes ----- A minimum of four periods is required for the calculation. Examples -------- The example below will show a rolling calculation with a window size of four matching the equivalent function call using `scipy.stats`. >>> arr = [1, 2, 3, 4, 999] >>> import scipy.stats >>> print(f"{scipy.stats.kurtosis(arr[:-1], bias=False):.6f}") -1.200000 >>> print(f"{scipy.stats.kurtosis(arr[1:], bias=False):.6f}") 3.999946 >>> s = pd.Series(arr) >>> s.rolling(4).kurt() 0 NaN 1 NaN 2 NaN 3 -1.200000 4 3.999946 dtype: float64 """ return super().kurt(numeric_only=numeric_only) def first(self, numeric_only: bool = False): """ Calculate the rolling First (left-most) element of the window. Return the first observed value in each rolling window. This is useful for tracking the starting value of a window as it slides forward. Parameters ---------- numeric_only : bool, default False Include only float, int, boolean columns. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- GroupBy.first : Similar method for GroupBy objects. Rolling.last : Method to get the last element in each window. Examples -------- The example below will show a rolling calculation with a window size of three. >>> s = pd.Series(range(5)) >>> s.rolling(3).first() 0 NaN 1 NaN 2 0.0 3 1.0 4 2.0 dtype: float64 """ return super().first(numeric_only=numeric_only) def last(self, numeric_only: bool = False): """ Calculate the rolling Last (right-most) element of the window. Return the last observed value in each rolling window. This is useful for tracking the most recent value of a window as it slides forward. Parameters ---------- numeric_only : bool, default False Include only float, int, boolean columns. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- GroupBy.last : Similar method for GroupBy objects. Rolling.first : Method to get the first element in each window. Examples -------- The example below will show a rolling calculation with a window size of three. >>> s = pd.Series(range(5)) >>> s.rolling(3).last() 0 NaN 1 NaN 2 2.0 3 3.0 4 4.0 dtype: float64 """ return super().last(numeric_only=numeric_only) def quantile( self, q: float, interpolation: QuantileInterpolation = "linear", numeric_only: bool = False, ): """ Calculate the rolling quantile. Return the ``q``-th quantile of the values in each rolling window. The ``interpolation`` parameter controls how quantiles falling between two data points are handled. Parameters ---------- q : float Quantile to compute. 0 <= quantile <= 1. interpolation : {'linear', 'lower', 'higher', 'midpoint', 'nearest'} This optional parameter specifies the interpolation method to use, when the desired quantile lies between two data points `i` and `j`: * linear: `i + (j - i) * fraction`, where `fraction` is the fractional part of the index surrounded by `i` and `j`. * lower: `i`. * higher: `j`. * nearest: `i` or `j` whichever is nearest. * midpoint: (`i` + `j`) / 2. numeric_only : bool, default False Include only float, int, boolean columns. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.quantile : Aggregating quantile for Series. DataFrame.quantile : Aggregating quantile for DataFrame. Examples -------- >>> s = pd.Series([1, 2, 3, 4]) >>> s.rolling(2).quantile(0.4, interpolation="lower") 0 NaN 1 1.0 2 2.0 3 3.0 dtype: float64 >>> s.rolling(2).quantile(0.4, interpolation="midpoint") 0 NaN 1 1.5 2 2.5 3 3.5 dtype: float64 """ return super().quantile( q=q, interpolation=interpolation, numeric_only=numeric_only, ) def rank( self, method: WindowingRankType = "average", ascending: bool = True, pct: bool = False, numeric_only: bool = False, ): """ Calculate the rolling rank. Compute the rank of each value within each rolling window. Ties are resolved using the ``method`` parameter. Parameters ---------- method : {'average', 'min', 'max'}, default 'average' How to rank the group of records that have the same value (i.e. ties): * average: average rank of the group * min: lowest rank in the group * max: highest rank in the group ascending : bool, default True Whether or not the elements should be ranked in ascending order. pct : bool, default False Whether or not to display the returned rankings in percentile form. numeric_only : bool, default False Include only float, int, boolean columns. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.rank : Aggregating rank for Series. DataFrame.rank : Aggregating rank for DataFrame. Examples -------- >>> s = pd.Series([1, 4, 2, 3, 5, 3]) >>> s.rolling(3).rank() 0 NaN 1 NaN 2 2.0 3 2.0 4 3.0 5 1.5 dtype: float64 >>> s.rolling(3).rank(method="max") 0 NaN 1 NaN 2 2.0 3 2.0 4 3.0 5 2.0 dtype: float64 >>> s.rolling(3).rank(method="min") 0 NaN 1 NaN 2 2.0 3 2.0 4 3.0 5 1.0 dtype: float64 """ return super().rank( method=method, ascending=ascending, pct=pct, numeric_only=numeric_only, ) def nunique( self, numeric_only: bool = False, ): """ Calculate the rolling nunique. .. versionadded:: 3.0.0 Parameters ---------- numeric_only : bool, default False Include only float, int, boolean columns. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.nunique : Aggregating nunique for Series. DataFrame.nunique : Aggregating nunique for DataFrame. Examples -------- >>> s = pd.Series([1, 4, 2, np.nan, 3, 3, 4, 5]) >>> s.rolling(3).nunique() 0 NaN 1 NaN 2 3.0 3 NaN 4 NaN 5 NaN 6 2.0 7 3.0 dtype: float64 """ return super().nunique( numeric_only=numeric_only, ) def cov( self, other: DataFrame | Series | None = None, pairwise: bool | None = None, ddof: int = 1, numeric_only: bool = False, ): """ Calculate the rolling sample covariance. Compute the sample covariance between the calling object and ``other`` over each rolling window. If ``other`` is not supplied, pairwise covariances of the columns are computed. Parameters ---------- other : Series or DataFrame, optional If not supplied then will default to self and produce pairwise output. pairwise : bool, default None If False then only matching columns between self and other will be used and the output will be a DataFrame. If True then all pairwise combinations will be calculated and the output will be a MultiIndexed DataFrame in the case of DataFrame inputs. In the case of missing elements, only complete pairwise observations will be used. ddof : int, default 1 Delta Degrees of Freedom. The divisor used in calculations is ``N - ddof``, where ``N`` represents the number of elements. numeric_only : bool, default False Include only float, int, boolean columns. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.cov : Aggregating cov for Series. DataFrame.cov : Aggregating cov for DataFrame. Examples -------- >>> ser1 = pd.Series([1, 2, 3, 4]) >>> ser2 = pd.Series([1, 4, 5, 8]) >>> ser1.rolling(2).cov(ser2) 0 NaN 1 1.5 2 0.5 3 1.5 dtype: float64 """ return super().cov( other=other, pairwise=pairwise, ddof=ddof, numeric_only=numeric_only, ) def corr( self, other: DataFrame | Series | None = None, pairwise: bool | None = None, ddof: int = 1, numeric_only: bool = False, ): """ Calculate the rolling correlation. Compute the Pearson correlation coefficient between the calling object and ``other`` over each rolling window. If ``other`` is not supplied, pairwise correlations of the columns are computed. Parameters ---------- other : Series or DataFrame, optional If not supplied then will default to self and produce pairwise output. pairwise : bool, default None If False then only matching columns between self and other will be used and the output will be a DataFrame. If True then all pairwise combinations will be calculated and the output will be a MultiIndexed DataFrame in the case of DataFrame inputs. In the case of missing elements, only complete pairwise observations will be used. ddof : int, default 1 Delta Degrees of Freedom. The divisor used in calculations is ``N - ddof``, where ``N`` represents the number of elements. numeric_only : bool, default False Include only float, int, boolean columns. Returns ------- Series or DataFrame Return type is the same as the original object with ``np.float64`` dtype. See Also -------- cov : Similar method to calculate covariance. numpy.corrcoef : NumPy Pearson's correlation calculation. Series.rolling : Calling rolling with Series data. DataFrame.rolling : Calling rolling with DataFrames. Series.corr : Aggregating corr for Series. DataFrame.corr : Aggregating corr for DataFrame. Notes ----- This function uses Pearson's definition of correlation (https://en.wikipedia.org/wiki/Pearson_correlation_coefficient). When `other` is not specified, the output will be self correlation (e.g. all 1's), except for :class:`~pandas.DataFrame` inputs with `pairwise` set to `True`. Function will return ``NaN`` for correlations of equal valued sequences; this is the result of a 0/0 division error. When `pairwise` is set to `False`, only matching columns between `self` and `other` will be used. When `pairwise` is set to `True`, the output will be a MultiIndex DataFrame with the original index on the first level, and the `other` DataFrame columns on the second level. In the case of missing elements, only complete pairwise observations will be used. Examples -------- The below example shows a rolling calculation with a window size of four matching the equivalent function call using :meth:`numpy.corrcoef`. >>> v1 = [3, 3, 3, 5, 8] >>> v2 = [3, 4, 4, 4, 8] >>> np.corrcoef(v1[:-1], v2[:-1]) array([[1. , 0.33333333], [0.33333333, 1. ]]) >>> np.corrcoef(v1[1:], v2[1:]) array([[1. , 0.9169493], [0.9169493, 1. ]]) >>> s1 = pd.Series(v1) >>> s2 = pd.Series(v2) >>> s1.rolling(4).corr(s2) 0 NaN 1 NaN 2 NaN 3 0.333333 4 0.916949 dtype: float64 The below example shows a similar rolling calculation on a DataFrame using the pairwise option. >>> matrix = np.array( ... [[51.0, 35.0], [49.0, 30.0], [47.0, 32.0], [46.0, 31.0], [50.0, 36.0]] ... ) >>> np.corrcoef(matrix[:-1, 0], matrix[:-1, 1]) array([[1. , 0.6263001], [0.6263001, 1. ]]) >>> np.corrcoef(matrix[1:, 0], matrix[1:, 1]) array([[1. , 0.55536811], [0.55536811, 1. ]]) >>> df = pd.DataFrame(matrix, columns=["X", "Y"]) >>> df X Y 0 51.0 35.0 1 49.0 30.0 2 47.0 32.0 3 46.0 31.0 4 50.0 36.0 >>> df.rolling(4).corr(pairwise=True) X Y 0 X NaN NaN Y NaN NaN 1 X NaN NaN Y NaN NaN 2 X NaN NaN Y NaN NaN 3 X 1.000000 0.626300 Y 0.626300 1.000000 4 X 1.000000 0.555368 Y 0.555368 1.000000 """ return super().corr( other=other, pairwise=pairwise, ddof=ddof, numeric_only=numeric_only, ) Rolling.__doc__ = Window.__doc__ @set_module("pandas.api.typing") class RollingGroupBy(BaseWindowGroupBy, Rolling): """ Provide a rolling groupby implementation. """ _attributes = Rolling._attributes + BaseWindowGroupBy._attributes def _get_window_indexer(self) -> GroupByIndexer: """ Return an indexer class that will compute the window start and end bounds Returns ------- GroupByIndexer """ rolling_indexer: type[BaseIndexer] indexer_kwargs: dict[str, Any] | None = None index_array = self._index_array if isinstance(self.window, BaseIndexer): rolling_indexer = type(self.window) indexer_kwargs = self.window.__dict__.copy() assert isinstance(indexer_kwargs, dict) # for mypy # We'll be using the index of each group later indexer_kwargs.pop("index_array", None) window = self.window elif self._win_freq_i8 is not None: rolling_indexer = VariableWindowIndexer # error: Incompatible types in assignment (expression has type # "int", variable has type "BaseIndexer") window = self._win_freq_i8 # type: ignore[assignment] else: rolling_indexer = FixedWindowIndexer window = self.window window_indexer = GroupByIndexer( index_array=index_array, window_size=window, groupby_indices=self._grouper.indices, window_indexer=rolling_indexer, indexer_kwargs=indexer_kwargs, ) return window_indexer def _validate_datetimelike_monotonic(self) -> None: """ Validate that each group in self._on is monotonic """ # GH 46061 if self._on.hasnans: self._raise_monotonic_error("values must not have NaT") for group_indices in self._grouper.indices.values(): group_on = self._on.take(group_indices) if not ( group_on.is_monotonic_increasing or group_on.is_monotonic_decreasing ): on = "index" if self.on is None else self.on raise ValueError( f"Each group within {on} must be monotonic. " f"Sort the values in {on} first." )