Approach: convert timestamps to dates, outer-merge predictions and ground truth on (id, timestamp) to capture missed predictions/labels, compute daily counts of TP/FP/FN, reindex to a complete daily range to handle missing days, then compute a 90-day rolling sum and derive precision/recall. This uses vectorized pandas ops and a fixed-size rolling window (90 days at daily frequency) for efficiency.python
import pandas as pd
def rolling_precision_recall(preds: pd.DataFrame, truth: pd.DataFrame, window_days: int = 90) -> pd.DataFrame:
"""
Compute daily rolling precision and recall over the past `window_days`.
Inputs:
preds: DataFrame with columns ['id', 'timestamp', 'pred'] where pred is 0/1 (or boolean)
truth: DataFrame with columns ['id', 'timestamp', 'label'] where label is 0/1 (or boolean)
window_days: integer window size in days (default 90)
Returns:
DataFrame indexed by date (daily) with columns:
tp, fp, fn, precision, recall, precision_rolling, recall_rolling
precision_rolling / recall_rolling are computed over the trailing window_days up to that date.
Notes/assumptions:
- timestamps can be any pandas-parsable datetimes; matching is done on exact timestamp and id.
- missing preds or labels are treated as 0 (i.e., no prediction / no positive label).
"""
# Normalize timestamps and create date
preds = preds.copy()
truth = truth.copy()
preds['timestamp'] = pd.to_datetime(preds['timestamp'])
truth['timestamp'] = pd.to_datetime(truth['timestamp'])
# Outer merge on id + timestamp to capture missed items
merged = pd.merge(preds, truth, on=['id', 'timestamp'], how='outer', suffixes=('_pred', '_label'))
# Fill missing preds/labels with 0
merged['pred'] = merged.get('pred', merged.get('pred_pred')).fillna(0).astype(int)
merged['label'] = merged.get('label', merged.get('label_label')).fillna(0).astype(int)
# Aggregate by date
merged['date'] = merged['timestamp'].dt.floor('D')
agg = merged.groupby('date').agg(
tp = pd.NamedAgg(column='pred', aggfunc=lambda x: int(((x==1) & (merged.loc[x.index,'label']==1)).sum())),
fp = pd.NamedAgg(column='pred', aggfunc=lambda x: int(((x==1) & (merged.loc[x.index,'label']==0)).sum())),
fn = pd.NamedAgg(column='label', aggfunc=lambda x: int(((x==1) & (merged.loc[x.index,'pred']==0)).sum())),
)
# Ensure daily index continuity
idx = pd.date_range(agg.index.min(), agg.index.max(), freq='D')
daily = agg.reindex(idx, fill_value=0)
daily.index.name = 'date'
# Rolling sums over window_days (since index is daily, window size = window_days)
rolling = daily.rolling(window=window_days, min_periods=1).sum()
# Compute rolling precision and recall
rolling['precision_rolling'] = rolling['tp'] / (rolling['tp'] + rolling['fp']).replace(0, pd.NA)
rolling['recall_rolling'] = rolling['tp'] / (rolling['tp'] + rolling['fn']).replace(0, pd.NA)
# Also keep per-day precision/recall if desired
daily['precision'] = daily['tp'] / (daily['tp'] + daily['fp']).replace(0, pd.NA)
daily['recall'] = daily['tp'] / (daily['tp'] + daily['fn']).replace(0, pd.NA)
result = daily.join(rolling[['precision_rolling','recall_rolling']])
return result
Complexity notes:- Time: O(n log n) dominated by the merge and groupby (n = total rows in preds + truth). Sorting inside groupby/rolling over daily index is O(d) where d = number of days.- Space: O(n) for merged table and O(d) for daily aggregated index (d << n typically).- Efficient because operations are vectorized; avoid per-day loops.