我有一個參差不齊（意思是不規(guī)則頻率）的時間索引 DataFrame，我想對其執(zhí)行時間加權(quán)滾動平均值，以保持 DataFrame 的原始索引。假設(shè)記錄的值在被另一個值取代之前是有效的。實(shí)現(xiàn)這一點(diǎn)的一種方法是將參差不齊的 DataFrame 上采樣到統(tǒng)一頻率，然后進(jìn)行滾動平均：import pandas as pdimport numpy as npdef time_weighted_average_using_upsampling(df: pd.DataFrame, avg_window: str) -> pd.DataFrame: # Leads to high memory usage original_index = df.index.copy() avg = ( df.resample("1s") .ffill() .rolling(avg_window, closed="left", min_periods=int(avg_window[0]))) .mean() .reindex(original_index) ) return avgif __name__ == "__main__": df = pd.DataFrame( {"A": [0, 1, 2, 3, 4, 5]}, index=[ pd.Timestamp("20130101 09:00:00"), pd.Timestamp("20130101 09:00:02"), pd.Timestamp("20130101 09:00:03"), pd.Timestamp("20130101 09:00:05"), pd.Timestamp("20130101 09:00:06"), pd.Timestamp("20130101 09:00:10"), ], ) expected_avg = pd.DataFrame( {"A": [np.nan, np.nan, 1 / 3, 5 / 3, 7 / 3, 4]}, index=[ pd.Timestamp("20130101 09:00:00"), pd.Timestamp("20130101 09:00:02"), pd.Timestamp("20130101 09:00:03"), pd.Timestamp("20130101 09:00:05"), pd.Timestamp("20130101 09:00:06"), pd.Timestamp("20130101 09:00:10"), ], ) pd.testing.assert_frame_equal( time_weighted_average_using_upsampling(df=df, avg_window="3s"), expected_avg )這個問題是上采樣破壞了參差不齊的 df 提供的稀疏表示的目的。稀疏表示具有內(nèi)存效率，而上采樣版本則不然。這就引出了一個問題：如何在不必對整個 df 進(jìn)行上采樣的情況下獲得上面顯示的結(jié)果？
2 回答
開滿天機(jī)
TA貢獻(xiàn)1786條經(jīng)驗 獲得超13個贊
這是一個替代方案，而不是對整個數(shù)據(jù)幀進(jìn)行上采樣，您可以首先檢查兩行之間的時間差異大于間隙的位置。然后將 3s 刪除到具有這些特定新時間戳的聯(lián)合的間隙和reindexdf 的行。創(chuàng)建這些行后，您可以groupby使用添加新索引的位置，resample每組 1 秒，最后rolling使用您所做的方法。Reindex最后有df。
rule = 3
rolling_win = f'{rule}s'
sparse = df.index.to_series().diff().dt.total_seconds().ge(rule)
new_timestamps = df.index[sparse] - pd.Timedelta(seconds=rule)
print(new_timestamps)
#DatetimeIndex(['2013-01-01 09:00:07'], dtype='datetime64[ns]', freq=None)
#reindex with the new
df_ = df.reindex(df.index.union(new_timestamps))
#perform first the resample 1s per group, then clean the dataframe to do the rolling.mean
#finally reindex like original df
df_ = (df_.groupby(df_.index.isin(new_timestamps).cumsum())
.resample("1s").ffill()
.reset_index(level=0, drop=True).ffill()
.rolling(rolling_win, closed="left", min_periods=rule)\
.mean()
.reindex(df.index)
)
print(df_)
A
2013-01-01 09:00:00 NaN
2013-01-01 09:00:02 NaN
2013-01-01 09:00:03 0.333333
2013-01-01 09:00:05 1.666667
2013-01-01 09:00:06 2.333333
2013-01-01 09:00:10 4.000000
在這種情況下，它并不是很有趣，因為差距實(shí)際上很小，但如果差距很大，那么它就會變得有用。
編輯或其他選項，可能更好，union所有索引都是從您刪除 1s、2s、3s 的原始索引中創(chuàng)建的，...（取決于規(guī)則）?，F(xiàn)在你只有滾動 so和reindex所需的索引。最后結(jié)果一樣ffillrolling.mean
from functools import reduce
rule = 3
rolling_win = f'{rule}s'
idx = df.index
df_ = (df.reindex(reduce(lambda x, y: x.union(y),
[idx - pd.Timedelta(seconds=i)
for i in range(0, rule+1)]))
.ffill()
.rolling(rolling_win, closed="left", min_periods=rule)\
.mean()
.reindex(df.index)
)
撒科打諢
TA貢獻(xiàn)1934條經(jīng)驗 獲得超2個贊
啟發(fā)的兩種可能的解決方案：
def time_weighted_average_using_local_upsampling(df: pd.DataFrame, avg_window: str) -> pd.DataFrame:
"""Uses second resolution up-sampling only on smaller windows at a time."""
original_index = df.index.copy()
avg = (
df.reindex(df.index.union(df.index.shift(periods=-1, freq=avg_window)), method="ffill")
.rolling(avg_window, closed="both", min_periods=2)
.apply(lambda x: x.resample("1s").ffill()[:-1].mean(skipna=False))
.reindex(original_index)
)
return avg
def time_weighted_average_using_index_weighting(df: pd.DataFrame, avg_window: str) -> pd.DataFrame:
"""Uses weighting by duration, by ensuring every window has a point at the start."""
original_index = df.index.copy()
avg = (
df.reindex(df.index.union(df.index.shift(periods=-1, freq=avg_window)), method="ffill")
.rolling(avg_window, closed="both", min_periods=2)
.apply(lambda x: np.average(x[:-1], weights=x.index.to_series().diff()[1:].dt.seconds))
.reindex(original_index)
)
return avg
第一個一次對單個滾動窗口進(jìn)行上采樣，而后者實(shí)際上通過確保在我們關(guān)心的窗口開始處始終有一個可用點(diǎn)來進(jìn)行參差不齊的時間加權(quán)平均。這是通過包括按窗口長度移動的原始索引來完成的。
我還沒有衡量相關(guān)案例的表現(xiàn)。
編輯： 我決定在大約 100,000 行的第二個分辨率數(shù)據(jù)集上測試函數(shù)，并使用 20 分鐘的窗口（?。﹥煞N變體都慢得令人難以忍受，但我認(rèn)為我有一個新的贏家：
def time_weighted_average_using_index_weighting2(df: pd.DataFrame, avg_window: str) -> pd.DataFrame:
"""Uses weighting by duration, by ensuring every window has a point at the start."""
original_index = df.index.copy()
avg = df.reindex(df.index.union(df.index.shift(periods=-1, freq=avg_window)), method="ffill")
avg = (
avg.multiply(avg.index.to_series().diff().dt.seconds.shift(-1), axis=0)
.divide(pd.Timedelta(avg_window).seconds)
.rolling(avg_window, closed="left")
.sum()
.reindex(original_index)
)
avg[~((avg.index - pd.Timedelta(avg_window)) >= original_index[0])] = np.nan
return avg
這個在滾動之前預(yù)先加權(quán)，因此我們不用使用.sum()而不是apply(). 這轉(zhuǎn)化為巨大的速度提升。無論平均窗口的大小如何，我們最多也可以將索引加倍。
添加回答
舉報
0/150
提交
取消