Question: How do I use polars' native API to do a rolling window on a datetime column, but filter out rows in the window based on the value of a column of the "current" row? My polars dataframe of financial transactions has the following schema:  For each transaction and a duration d, I want to:  grab the source_acct and its timestamp look back timestamp - d hours and get only rows whose source_acct or dest_acct matches the current source_acct sum up all txn as amount_in when the current source_acct is equal to a row's dest_acct do the same for amount_out but where the current src acct is the row's source_account including itself.  I tried this using map_rows but it's way too slow for a dataframe with 20M rows. I sort my df on the timestamp column, then run: def windowing(df: pl.DataFrame, window_in_hours: int): d = timedelta(hours=window_in_hours) def calculate_amt(row): acc_no, window_end = row[0], row[1] window_start = window_end - d acct_window_mask = ( (pl.col('timestamp') >= window_start) & (pl.col('timestamp') <= window_end) & (pl.col('dest_acct').eq(acc_no) | pl.col('source_acct').eq(acc_no)) ) window_txns = df.filter(acct_window_mask) amount_in = window_txns.filter(pl.col('dest_acct').eq(acc_no))['amount'].sum() amount_out = window_txns.filter(pl.col('source_acct').eq(acc_no))['amount'].sum() return (amount_in, amount_out) calculated_amounts = df.select(["source_acct", "timestamp", 'dest_acct', 'amount']).map_rows(calculate_amt) return df.with_columns( calculated_amounts['column_0'].alias('amount_in'), calculated_amounts['column_1'].alias('amount_out'), )  I've been trying to implement this using polars native API like .rolling() but I don't get how to do the filter step of comparing the current row's source account against the windowed transactions. Here's a sample: import polars as pl from datetime import timedelta data = { "timestamp": [ "2024-01-01 10:00:00", "2024-01-01 10:30:00", "2024-01-01 11:00:00", "2024-01-01 11:30:00", "2024-01-01 12:00:00" ], "source_acct": ["A", "B", "A", "C", "A"], "dest_acct": ["B", "A", "C", "A", "B"], "amount": [100, 150, 200, 300, 250] } df = pl.DataFrame(data).with_columns(pl.col("timestamp").str.to_datetime()) print(windowing(df, 1))  Expected output: ┌─────────────────────┬─────────────┬───────────┬────────┬───────────┬────────────┐ │ timestamp ┆ source_acct ┆ dest_acct ┆ amount ┆ amount_in ┆ amount_out │ │ --- ┆ --- ┆ --- ┆ --- ┆ --- ┆ --- │ │ datetime[μs] ┆ str ┆ str ┆ i64 ┆ i64 ┆ i64 │ ╞═════════════════════╪═════════════╪═══════════╪════════╪═══════════╪════════════╡ │ 2024-01-01 10:00:00 ┆ A ┆ B ┆ 100 ┆ 0 ┆ 100 │ │ 2024-01-01 10:30:00 ┆ B ┆ A ┆ 150 ┆ 100 ┆ 150 │ │ 2024-01-01 11:00:00 ┆ A ┆ C ┆ 200 ┆ 150 ┆ 300 │ │ 2024-01-01 11:30:00 ┆ C ┆ A ┆ 300 ┆ 200 ┆ 300 │ │ 2024-01-01 12:00:00 ┆ A ┆ B ┆ 250 ┆ 300 ┆ 450 │ └─────────────────────┴─────────────┴───────────┴────────┴───────────┴────────────┘ 

Answer: df_in = ( df.join_where( df, pl.col.source_acct == pl.col.dest_acct_right, pl.col.timestamp.dt.offset_by("-1h") <= pl.col.timestamp_right, pl.col.timestamp >= pl.col.timestamp_right ) .group_by("timestamp", "source_acct") .agg(amount_in = pl.col.amount_right.sum()) ) ( df .join(df_in, on=["timestamp","source_acct"], how="left") .with_columns( pl.col.amount_in.fill_null(0), pl.col.amount .rolling_sum_by("timestamp", "1h", closed="both") .over("source_acct") .alias("amount_out") ) )  shape: (5, 6) ┌─────────────────────┬─────────────┬───────────┬────────┬───────────┬────────────┐ │ timestamp ┆ source_acct ┆ dest_acct ┆ amount ┆ amount_in ┆ amount_out │ │ --- ┆ --- ┆ --- ┆ --- ┆ --- ┆ --- │ │ datetime[μs] ┆ str ┆ str ┆ i64 ┆ i64 ┆ i64 │ ╞═════════════════════╪═════════════╪═══════════╪════════╪═══════════╪════════════╡ │ 2024-01-01 10:00:00 ┆ A ┆ B ┆ 100 ┆ 0 ┆ 100 │ │ 2024-01-01 10:30:00 ┆ B ┆ A ┆ 150 ┆ 100 ┆ 150 │ │ 2024-01-01 11:00:00 ┆ A ┆ C ┆ 200 ┆ 150 ┆ 300 │ │ 2024-01-01 11:30:00 ┆ C ┆ A ┆ 300 ┆ 200 ┆ 300 │ │ 2024-01-01 12:00:00 ┆ A ┆ B ┆ 250 ┆ 300 ┆ 450 │ └─────────────────────┴─────────────┴───────────┴────────┴───────────┴────────────┘  Just for illustration, you can also do it with DuckDB sql: import duckdb duckdb.sql(""" select d.*, ( select coalesce(sum(tt.amount), 0) from df as tt where tt.dest_acct = d.source_acct and tt.timestamp between d.timestamp - interval 1 hour and d.timestamp ) as amount_in, sum(d.amount) over( partition by d.source_acct order by d.timestamp range between interval 1 hour preceding and current row ) as amount_out from df as d """).pl()