Pandas：带有日期条件的 SQL SelfJoin

Question

我经常在关系数据库中使用 SQL 执行的一个查询是将表连接回自身，并根据相同 id 的记录在时间上向后或向前汇总每一行。

例如，假设 table1 为列 'ID'、'Date'、'Var1'

在 SQL 中，我可以对过去 3 个月的每条记录求和 var1，如下所示：

Select a.ID, a.Date, sum(b.Var1) as sum_var1
from table1 a
left outer join table1 b
on a.ID = b.ID
and months_between(a.date,b.date) <0
and months_between(a.date,b.date) > -3

有没有办法在 Pandas 中做到这一点？

Answer 1

看来你需要GroupBy + rolling。以与用 SQL 编写的完全相同的方式实现逻辑可能会很昂贵，因为它会涉及重复循环。让我们以一个示例数据框为例：

        Date  ID  Var1
0 2015-01-01   1     0
1 2015-02-01   1     1
2 2015-03-01   1     2
3 2015-04-01   1     3
4 2015-05-01   1     4
5 2015-01-01   2     5
6 2015-02-01   2     6
7 2015-03-01   2     7
8 2015-04-01   2     8
9 2015-05-01   2     9

您可以添加一个列，该列按组回顾并在固定时期内对变量求和。首先使用pd.Series.rolling定义一个函数：

def lookbacker(x):
    """Sum over past 70 days"""
    return x.rolling('70D').sum().astype(int)

然后将其应用于GroupBy 对象并提取值进行赋值：

df['Lookback_Sum'] = df.set_index('Date').groupby('ID')['Var1'].apply(lookbacker).values

print(df)

        Date  ID  Var1  Lookback_Sum
0 2015-01-01   1     0             0
1 2015-02-01   1     1             1
2 2015-03-01   1     2             3
3 2015-04-01   1     3             6
4 2015-05-01   1     4             9
5 2015-01-01   2     5             5
6 2015-02-01   2     6            11
7 2015-03-01   2     7            18
8 2015-04-01   2     8            21
9 2015-05-01   2     9            24

看来pd.Series.rolling 确实不适用于几个月，例如使用'2M'（2 个月）而不是'70D'（70 天）给出ValueError: <2 * MonthEnds> is a non-fixed frequency。这是有道理的，因为“月份”是不明确的，因为月份有不同的天数。

值得一提的另一点是您可以直接使用GroupBy + rolling，并且可能通过绕过apply 更有效，但这需要确保您的索引是单调的。例如，通过sort_index：

df['Lookback_Sum'] = df.set_index('Date').sort_index()\
                       .groupby('ID')['Var1'].rolling('70D').sum()\
                       .astype(int).values

Answer 2

我不认为pandas.DataFrame.rolling() 在几个月内支持滚动窗口聚合；目前，您必须指定固定天数或其他固定长度的时间段。

但正如@jpp 提到的，您可以使用 python 循环在日历月中指定的窗口大小上执行滚动聚合，其中每个窗口中的天数会有所不同，具体取决于您滚动的日历的哪个部分.

以下方法基于this SO answer 以及@jpp 的：

# Build some example data:
# 3 unique IDs, each with 365 samples, one sample per day throughout 2015
df = pd.DataFrame({'Date': pd.date_range('2015-01-01', '2015-12-31', freq='D'),
                   'Var1': list(range(365))})
df = pd.concat([df] * 3)
df['ID'] = [1]*365 + [2]*365 + [3]*365
df.head()
        Date  Var1  ID
0 2015-01-01     0   1
1 2015-01-02     1   1
2 2015-01-03     2   1
3 2015-01-04     3   1
4 2015-01-05     4   1

# Define a lookback function that mimics rolling aggregation,
# but uses DateOffset() slicing, rather than a window of fixed size.
# Use .count() here as a sanity check; you will need .sum()
def lookbacker(ser): 
    return pd.Series([ser.loc[d - pd.offsets.DateOffset(months=3):d].count() 
                      for d in ser.index])

# By default, groupby.agg output is sorted by key. So make sure to 
# sort df by (ID, Date) before inserting the flattened groupby result 
# into a new column
df.sort_values(['ID', 'Date'], inplace=True)
df.set_index('Date', inplace=True)
df['window_size'] = df.groupby('ID')['Var1'].apply(lookbacker).values

# Manually check the resulting window sizes
df.head()
            Var1  ID  window_size
Date                             
2015-01-01     0   1            1
2015-01-02     1   1            2
2015-01-03     2   1            3
2015-01-04     3   1            4
2015-01-05     4   1            5

df.tail()
            Var1  ID  window_size
Date                             
2015-12-27   360   3           92
2015-12-28   361   3           92
2015-12-29   362   3           92
2015-12-30   363   3           92
2015-12-31   364   3           93

df[df.ID == 1].loc['2015-05-25':'2015-06-05']
            Var1  ID  window_size
Date                             
2015-05-25   144   1           90
2015-05-26   145   1           90
2015-05-27   146   1           90
2015-05-28   147   1           90
2015-05-29   148   1           91
2015-05-30   149   1           92
2015-05-31   150   1           93
2015-06-01   151   1           93
2015-06-02   152   1           93
2015-06-03   153   1           93
2015-06-04   154   1           93
2015-06-05   155   1           93

最后一列以天为单位给出回溯窗口大小，从该日期开始回溯，包括开始日期和结束日期。

在2016-05-31 之前查找“3 个月”会到达2015-02-31，但 2015 年 2 月只有 28 天。正如您在上述健全性检查中的序列 90, 91, 92, 93 中看到的那样，DateOffset 方法将 5 月的最后四天映射到 2 月的最后一天：

pd.to_datetime('2015-05-31') - pd.offsets.DateOffset(months=3)
Timestamp('2015-02-28 00:00:00')

pd.to_datetime('2015-05-30') - pd.offsets.DateOffset(months=3)
 Timestamp('2015-02-28 00:00:00')

pd.to_datetime('2015-05-29') - pd.offsets.DateOffset(months=3)
Timestamp('2015-02-28 00:00:00')

pd.to_datetime('2015-05-28') - pd.offsets.DateOffset(months=3)
Timestamp('2015-02-28 00:00:00')

我不知道这是否符合 SQL 的行为，但无论如何，您都需要对此进行测试并确定这是否适合您的情况。

Answer 3

你可以使用 lambda 来实现它。

table1['sum_var1'] = table1.apply(lambda row: findSum(row), axis=1)

我们应该为months_between编写一个等效的方法

完整的例子是

from datetime import datetime
import datetime as dt
import pandas as pd

def months_between(date1, date2):
    if date1.day == date2.day:
        return (date1.year - date2.year) * 12 + date1.month - date2.month
    # if both are last days
    if date1.month != (date1 + dt.timedelta(days=1)).month :
        if date2.month != (date2 + dt.timedelta(days=1)).month :
            return date1.month - date2.month
    return (date1 - date2).days / 31

def findSum(cRow):
    table1['month_diff'] = table1['Date'].apply(months_between, date2=cRow['Date'])
    filtered_table = table1[(table1["month_diff"] < 0) & (table1["month_diff"] > -3) & (table1['ID'] == cRow['ID'])]
    if filtered_table.empty:
        return 0
    return filtered_table['Var1'].sum()


table1 = pd.DataFrame(columns = ['ID', 'Date', 'Var1'])
table1.loc[len(table1)] = [1, datetime.strptime('2015-01-01','%Y-%m-%d'), 0]
table1.loc[len(table1)] = [1, datetime.strptime('2015-02-01','%Y-%m-%d'), 1]
table1.loc[len(table1)] = [1, datetime.strptime('2015-03-01','%Y-%m-%d'), 2]
table1.loc[len(table1)] = [1, datetime.strptime('2015-04-01','%Y-%m-%d'), 3]
table1.loc[len(table1)] = [1, datetime.strptime('2015-05-01','%Y-%m-%d'), 4]

table1.loc[len(table1)] = [2, datetime.strptime('2015-01-01','%Y-%m-%d'), 5]
table1.loc[len(table1)] = [2, datetime.strptime('2015-02-01','%Y-%m-%d'), 6]
table1.loc[len(table1)] = [2, datetime.strptime('2015-03-01','%Y-%m-%d'), 7]
table1.loc[len(table1)] = [2, datetime.strptime('2015-04-01','%Y-%m-%d'), 8]
table1.loc[len(table1)] = [2, datetime.strptime('2015-05-01','%Y-%m-%d'), 9]

table1['sum_var1'] = table1.apply(lambda row: findSum(row), axis=1)
table1.drop(columns=['month_diff'], inplace=True)
print(table1)