Skip to content

Commit 086d228

Browse files
feat: implement DataFrame.rolling with count, sum, mean, std
Co-authored-by: tswast <247555+tswast@users.noreply.github.com>
1 parent 9fbf114 commit 086d228

3 files changed

Lines changed: 157 additions & 0 deletions

File tree

leanframe/core/frame.py

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
from functools import reduce
2323
import operator
2424

25+
from leanframe.core.window import Rolling
2526
from leanframe.core.dtypes import convert_ibis_to_pandas
2627
from leanframe.core.indexing import (
2728
Index,
@@ -126,6 +127,19 @@ def dtypes(self) -> pd.Series:
126127
types = [convert_ibis_to_pandas(t) for t in self._data.schema().types]
127128
return pd.Series(types, index=names, name="dtypes")
128129

130+
def rolling(self, window: int, min_periods: int | None = None) -> Rolling:
131+
"""Provide rolling window calculations.
132+
133+
Args:
134+
window: Size of the moving window. This is the number of observations
135+
used for calculating the statistic.
136+
min_periods: Minimum number of observations in window required to have
137+
a value; otherwise, result is null. Defaults to window size.
138+
139+
Returns:
140+
A Rolling object.
141+
"""
142+
return Rolling(self, window=window, min_periods=min_periods)
129143
def __getitem__(self, key: str) -> DataFrame:
130144
"""Get a column.
131145

leanframe/core/window.py

Lines changed: 101 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,101 @@
1+
# Copyright 2025 Google LLC, LeanFrame Authors
2+
#
3+
# Licensed under the Apache License, Version 2.0 (the "License");
4+
# you may not use this file except in compliance with the License.
5+
# You may obtain a copy of the License at
6+
#
7+
# http://www.apache.org/licenses/LICENSE-2.0
8+
#
9+
# Unless required by applicable law or agreed to in writing, software
10+
# distributed under the License is distributed on an "AS IS" BASIS,
11+
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12+
# See the License for the specific language governing permissions and
13+
# limitations under the License.
14+
15+
"""Windowed operations on DataFrames."""
16+
17+
from __future__ import annotations
18+
19+
from typing import TYPE_CHECKING, Any
20+
21+
import ibis
22+
23+
if TYPE_CHECKING:
24+
from leanframe.core.frame import DataFrame
25+
26+
27+
class Rolling:
28+
"""Windowed operations on DataFrames."""
29+
30+
def __init__(
31+
self,
32+
obj: DataFrame,
33+
window: int,
34+
min_periods: int | None = None,
35+
):
36+
"""Initialize a Rolling object.
37+
38+
Args:
39+
obj: The DataFrame to apply rolling operations to.
40+
window: The size of the moving window.
41+
min_periods: Minimum number of observations in window required to
42+
have a value. Defaults to window size.
43+
"""
44+
self._obj = obj
45+
self._window = window
46+
self._min_periods = min_periods if min_periods is not None else window
47+
48+
def _apply_aggregation(self, op: str, **kwargs: Any) -> DataFrame:
49+
from leanframe.core.frame import DataFrame
50+
51+
t = self._obj._data
52+
53+
w = ibis.window(preceding=self._window - 1, following=0)
54+
55+
exprs = []
56+
for c in t.columns:
57+
col = t[c]
58+
59+
if op == "count":
60+
agg = col.count()
61+
elif op == "sum":
62+
agg = col.sum()
63+
elif op == "mean":
64+
agg = col.mean()
65+
elif op == "std":
66+
agg = col.std(how="sample")
67+
else:
68+
raise NotImplementedError(f"Unsupported operation: {op}")
69+
70+
agg_over = agg.over(w)
71+
72+
if op == "count":
73+
# For count, min_periods applies to the TOTAL number of rows in the window (including nulls)
74+
# We can trick Ibis into counting all rows by coalescing the column.
75+
valid_count = ibis.coalesce(col, 0).count().over(w)
76+
else:
77+
# For others, min_periods applies to the number of non-null observations
78+
valid_count = col.count().over(w)
79+
80+
final_expr = ibis.ifelse(
81+
valid_count >= self._min_periods, agg_over, ibis.null()
82+
).name(c)
83+
exprs.append(final_expr)
84+
85+
return DataFrame(t.select(exprs))
86+
87+
def count(self) -> DataFrame:
88+
"""Calculate the rolling count."""
89+
return self._apply_aggregation("count")
90+
91+
def sum(self) -> DataFrame:
92+
"""Calculate the rolling sum."""
93+
return self._apply_aggregation("sum")
94+
95+
def mean(self) -> DataFrame:
96+
"""Calculate the rolling mean."""
97+
return self._apply_aggregation("mean")
98+
99+
def std(self) -> DataFrame:
100+
"""Calculate the rolling standard deviation."""
101+
return self._apply_aggregation("std")

tests/unit/test_window.py

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
import numpy as np
2+
import pandas as pd
3+
import pandas.testing as pdt
4+
import ibis
5+
6+
from leanframe.core.frame import DataFrame
7+
8+
9+
def test_dataframe_rolling_aggregations():
10+
pdf = pd.DataFrame(
11+
{
12+
"a": [1.0, 2.0, 3.0, 4.0, 5.0],
13+
"b": [10.0, np.nan, 30.0, 40.0, 50.0],
14+
}
15+
)
16+
ldf = DataFrame(ibis.memtable(pdf))
17+
18+
# Test sum with default min_periods (min_periods=window)
19+
p_sum = pdf.rolling(window=3).sum()
20+
l_sum = ldf.rolling(window=3).sum().to_pandas()
21+
pdt.assert_frame_equal(p_sum, l_sum, check_dtype=False)
22+
23+
# Test sum with min_periods=1
24+
p_sum_min = pdf.rolling(window=3, min_periods=1).sum()
25+
l_sum_min = ldf.rolling(window=3, min_periods=1).sum().to_pandas()
26+
pdt.assert_frame_equal(p_sum_min, l_sum_min, check_dtype=False)
27+
28+
# Test count
29+
p_count = pdf.rolling(window=2).count()
30+
# Pandas count returns float if there's nan, int if not. Ibis returns int. Let pandas handle the check_dtype=False
31+
l_count = ldf.rolling(window=2).count().to_pandas()
32+
pdt.assert_frame_equal(p_count, l_count, check_dtype=False)
33+
34+
# Test mean
35+
p_mean = pdf.rolling(window=3, min_periods=2).mean()
36+
l_mean = ldf.rolling(window=3, min_periods=2).mean().to_pandas()
37+
pdt.assert_frame_equal(p_mean, l_mean, check_dtype=False)
38+
39+
# Test std
40+
p_std = pdf.rolling(window=3, min_periods=2).std()
41+
l_std = ldf.rolling(window=3, min_periods=2).std().to_pandas()
42+
pdt.assert_frame_equal(p_std, l_std, check_dtype=False)

0 commit comments

Comments
 (0)