python - 使用 pandas dataframe 加速迭代过程

Question

我有一个大型 Pandas 数据框 df_gen，其中包含 10000 个客户的时间序列数据。数据与能源使用有关。这是它的缩小版

In[1]: df_gen   
Out[2]: 
                         10053802  10053856  10053898  10058054
2013-01-01 00:00:00     0.196     1.493     0.332     0.278
2013-01-01 00:30:00     0.155     1.497     0.336     0.275
2013-01-01 01:00:00     0.109     1.487       NaN     0.310
2013-01-01 01:30:00     0.703     1.479     0.331     0.272
2013-01-01 02:00:00     0.389     1.533     0.293     0.313

我有一个填充缺失数据的过程:对于在特定时间戳中缺失数据的特定客户 ID，找到在整个数据集中具有最相似数据的时间戳，并使用它来填补空白。

使用此方法的原因是能源使用取决于外部因素，例如外部温度，因此，例如在炎热的日子里，很多顾客都开着空调。如果我们找到大多数其他客户的能源使用情况与缺失数据点的日期和时间相似的日期和时间，那么这是填充缺失数据的好地方。

它使用一个函数通过计算每一行的方差来识别数据与缺失数据的时间戳最匹配的时间戳:

def best_ts(df,ts_null,null_row):
# finds the timestamp for which the load is closest to the missing load at ts_null across the dataset df
# null_row is the row with the null data to be filled
var_df = pd.Series(index=df.index)
var_df.fillna(value=0, inplace=True)
if pd.isnull(null_row).all():
        logging.info('No customer data at all for %s ',str(ts_null))
var_df = ((df-null_row).fillna(value=0)**2).sum(axis=1)
smallest = var_df.idxmin()
return smallest

脚本然后为每个客户和每个时间戳迭代，当它找到空数据时，它调用 best_ts 并从该时间戳开始填充:

for id in df_gen.columns:
    for ts in df_gen.index:
        if pd.isnull(df_gen.loc[ts,id]):
        # slice df to remove rows that have no filling data for this customer and use this to fill from
        fill_ts = best_ts(df_gen[df_gen[id].notnull()],ts, df_gen.loc[ts])
        df_gen.loc[ts].fillna(df_gen.loc[fill_ts], inplace=True)

实例使用上面的示例 df，当找到 NaN 数据时，将向 best_ts 传递 3 个参数:删除缺失数据行的 df、缺失数据的时间戳，以及作为 Pandas 系列的缺失数据行

In: df_gen[df_gen[id].notnull()]
Out: 
                     10053802  10053856  10053898  10058054
2013-01-01 00:00:00     0.196     1.493     0.332     0.278
2013-01-01 00:30:00     0.155     1.497     0.336     0.275
2013-01-01 01:30:00     0.703     1.479     0.331     0.272
2013-01-01 02:00:00     0.389     1.533     0.293     0.313

In: ts
Out: 

datetime.datetime(2013, 1, 1, 1, 0)

In: df_gen.loc[ts]
Out: 
10053802    0.109
10053856    1.487
10053898      NaN
10058054    0.310

在该函数中，使用与数据帧相同的 DateTimeIndex 创建了一个 pandas 系列 var_df。每个值都是方差，即每个客户的能量值与时间戳 ts 的能量值之差的平方和。

例如 var_df 中的第一个值由 ((0.196-0.109)^2 + (1.493-1.487)^2 + 0 + (0.278-0.310)^2) = 0.008629 给出

In: var_df
Out: 
2013-01-01 00:00:00    0.008629
2013-01-01 00:30:00    0.003441
2013-01-01 01:30:00    0.354344
2013-01-01 02:00:00    0.080525
dtype: float64

所以时间戳 2013-01-01 00:30:00 是最“像”缺失数据时间的时间，所以选择这个时间来填充缺失数据。

所以填充的数据框看起来像这样:

In: df_gen
Out: 
                     10053802  10053856  10053898  10058054
2013-01-01 00:00:00     0.196     1.493     0.332     0.278
2013-01-01 00:30:00     0.155     1.497     0.336     0.275
2013-01-01 01:00:00     0.109     1.487     0.336     0.310
2013-01-01 01:30:00     0.703     1.479     0.331     0.272
2013-01-01 02:00:00     0.389     1.533     0.293     0.313

(注意:在这个小例子中，“最佳”时间戳恰好是紧接在缺失数据之前的时间戳，但在完整数据集中，它可能是一年中 17519 个时间戳中的任何一个。)

此代码有效，但伙计，它太慢了!通过数据集大约需要 2 个月的时间!我希望通过避免嵌套迭代或加速函数来加快速度的建议。

Answer 1

看起来您的相似性指标正在计算每列之间的元素平方距离之和。一种方法，诚然有点笨拙(但利用了快速的 Pandas 操作)是:

1. 遍历每一列，并创建一个与原始数据框具有相同维度的新数据框，但其中每一列都是当前列的副本。
2. 使用df.subtract().pow(2).sum()计算相似度，忽略自减列，找到最小值的列名(即客户id)值(value)。
3. 用匹配列中的相应值更新当前列中的缺失值。

以下是草稿，但它可能足以适应您的用例。此实现的一个重要假设是每个客户只能丢失一个数据点。该代码应该可以推广到每个客户的多个缺失数据点，只需做一些工作。因此，在测试此代码时，请确保随机生成的 df 每列仅缺少一个数据点。 (通常是这样，但并非总是如此。)

生成样本数据

dates = pd.date_range('20170101', periods=10, freq='D')
ids = [10006414, 10006572, 10006630, 10006664, 10006674]
values = np.random.random(size=len(dates)*len(ids)).reshape(10,5)
df = pd.DataFrame(values, index=dates, columns=ids)

# insert random missing data
nan_size = 4
for _ in range(nan_size):
    nan_row = np.random.randint(0, df.shape[0])
    nan_col = np.random.randint(0, df.shape[1])
    df.iloc[nan_row, nan_col] = np.nan

执行匹配插值

def get_closest(customer, dims):
    cust = customer.name
    nrow = dims[0]
    ncol = dims[1]
    replace_row = df.index[df[cust].isnull()]
    # make data frame full of cust data
    df2 = pd.DataFrame(np.repeat(df.loc[:,cust], ncol).values.reshape(nrow,ncol), 
                       index=dates, columns=ids)
    replace_col = (df.subtract(df2)
                     .pow(2)
                     .sum()
                     .replace({0:np.nan}) # otherwise 0 will go to top of sort
                     .sort_values()
                     .index[0] # index here is matching customer id
                  )
    customer[replace_row] = df.ix[replace_row, replace_col]
    return customer

print(df.apply(get_closest, axis='rows', args=(df.shape,)))

更新
根据 OP 的澄清，目标是进行逐行比较(即找到最相似的时间戳)而不是逐列比较(即找到最相似的客户)。下面是 get_closest() 的更新版本，它进行逐行比较，并顺利处理多个缺失值。

我还添加了一个报告功能，它将打印包含所有客户缺失条目的每个时间戳，以及用于估算缺失值的时间戳。报告默认关闭，只需将 True 作为 apply() 中的第二个 args 条目传递即可将其打开。

更新 2
更新后的行式 get_closest() 现在考虑了边缘情况，其中最近的时间戳也有需要插补的客户列的 NaN 值。现在，该函数将搜索最近的时间戳，该时间戳也具有需要估算的缺失值的可用数据。

示例数据:

            10006414  10006572  10006630  10006664  10006674
2017-01-01  0.374593  0.982585  0.059732  0.513149  0.251808
2017-01-02  0.269229  0.998531  0.523589  0.780806  0.033106
2017-01-03  0.261173  0.828637  0.638376  0.314944  0.737646
2017-01-04  0.786112  0.101750  0.286983  0.242778  0.341717
2017-01-05  0.230358  0.387392  0.918353  0.206100       NaN
2017-01-06  0.715966  0.206121  0.153461  0.894511  0.765227
2017-01-07  0.095002  0.169697  0.465624  0.109404  0.212315
2017-01-08  0.474712       NaN  0.471861  0.773374  0.454295
2017-01-09       NaN  0.201928  0.228018  0.173968  0.248485
2017-01-10  0.542635       NaN  0.132974  0.692073  0.201721

ROW-WISE get_closest()

def get_closest(row, dims, report=False):
    if row.isnull().sum():
        ts_with_nan = row.name
        nrow, ncol = dims
        df2 = pd.DataFrame(np.tile(df.loc[ts_with_nan], nrow).reshape(nrow,ncol), 
                           index=df.index, columns=df.columns)
        most_similar_ts = (df.subtract(df2, axis='rows', fill_value=0)
                             .pow(2)
                             .sum(axis=1, skipna=True)
                             .sort_values()
                          )
        # remove current row from matched indices
        most_similar_ts = most_similar_ts[most_similar_ts.index != ts_with_nan] 
        # narrow down to only columns where replacements would occur
        match_vals = df.ix[most_similar_ts.index, df.loc[ts_with_nan].isnull()]
        # select only rows where all values are non-empty
        all_valid = match_vals.notnull().all(axis=1)
        # take the timestamp index of the first row of match_vals[all_valid]
        best_match = match_vals[all_valid].head(1).index[0]
        if report:
            print('MISSING VALUES found at timestamp: {}'.format(ts_with_nan.strftime('%Y-%m-%d %H:%M:%S')))
            print('            REPLACEMENT timestamp: {}'.format(best_match.strftime('%Y-%m-%d %H:%M:%S')))

        # replace missing values with matched data
        return row.fillna(df.loc[best_match])

    return row

df.apply(get_closest, axis='columns', args=(df.shape, True)) # report=True

输出:

# MISSING VALUES found at timestamp: 2017-01-02 00:00:00
            # REPLACEMENT timestamp: 2017-01-09 00:00:00
# MISSING VALUES found at timestamp: 2017-01-07 00:00:00
            # REPLACEMENT timestamp: 2017-01-10 00:00:00
# MISSING VALUES found at timestamp: 2017-01-09 00:00:00
            # REPLACEMENT timestamp: 2017-01-03 00:00:00

print(df)
            10006414  10006572  10006630  10006664  10006674
2017-01-01  0.374593  0.982585  0.059732  0.513149  0.251808
2017-01-02  0.269229  0.998531  0.523589  0.780806  0.033106
2017-01-03  0.261173  0.828637  0.638376  0.314944  0.737646
2017-01-04  0.786112  0.101750  0.286983  0.242778  0.341717
2017-01-05  0.230358  0.387392  0.918353  0.206100  0.212315
2017-01-06  0.715966  0.206121  0.153461  0.894511  0.765227
2017-01-07  0.095002  0.169697  0.465624  0.109404  0.212315
2017-01-08  0.474712  0.201928  0.471861  0.773374  0.454295
2017-01-09  0.095002  0.201928  0.228018  0.173968  0.248485
2017-01-10  0.542635  0.201928  0.132974  0.692073  0.201721

除了这种逐行方法之外，我还在这个答案的开头保留了 get_closest() 的原始版本，因为我可以看到基于“最近的客户”的插补的值(value)，而不是“最近的时间戳”，它可能会在将来用作其他人的引用点。

更新 3
OP 提供了这个更新和最终的解决方案:

import pandas as pd
import numpy as np

# create dataframe of random data
dates = pd.date_range('20170101', periods=10, freq='D')
ids = [10006414, 10006572, 10006630, 10006664, 10006674]
values = np.random.random(size=len(dates)*len(ids)).reshape(10,5)
df = pd.DataFrame(values, index=dates, columns=ids)

# insert random missing data
nan_size = 20
for _ in range(nan_size):
    nan_row = np.random.randint(0, df.shape[0])
    nan_col = np.random.randint(0, df.shape[1])
    df.iloc[nan_row, nan_col] = np.nan

print ('Original df is ', df)
def get_closest(row, dims, report=False):
    if row.isnull().sum():
        ts_with_nan = row.name
        nrow, ncol = dims
        df2 = pd.DataFrame(np.tile(df.loc[ts_with_nan], nrow).reshape(nrow, ncol), index=df.index, columns=df.columns)
        most_similar_ts = (df.subtract(df2, axis='rows')
                           .pow(2)
                           .sum(axis=1, skipna=True)
                           .sort_values())
        # remove current row from matched indices
        most_similar_ts = most_similar_ts[most_similar_ts.index != ts_with_nan]
        if report:
            print('MISSING VALUES found at timestamp: {}'.format(ts_with_nan.strftime('%Y-%m-%d %H:%M:%S')))
        while row.isnull().sum():
            # narrow down to only columns where replacements would occur
            match_vals = df.ix[most_similar_ts.index, df.loc[ts_with_nan].isnull()]
            # fill from closest ts
            best_match = match_vals.head(1).index[0]
            row = row.fillna(df.loc[best_match])

            if report:
                print('            REPLACEMENT timestamp: {}'.format(best_match.strftime('%Y-%m-%d %H:%M:%S')))
            # Any customers with remaining NaNs in df.loc[ts_with_nan] also have NaNs in df.loc[best_match]
            # so remove this ts from the results and repeat the process
            most_similar_ts = most_similar_ts[most_similar_ts.index != best_match]
        return row


    return row

df_new = df.apply(get_closest, axis='columns', args=(df.shape, True))  # report=True
print ('Final df is ', df_new)