使用 pandas 對(duì)數(shù)據(jù)進(jìn)行移動(dòng)計(jì)算
假設(shè)有 10 天的銷(xiāo)售額數(shù)據(jù)，我們想每三天求一次總和，比如第五天的總和就是第三天 + 第四天 + 第五天的銷(xiāo)售額之和，這個(gè)時(shí)候該怎么做呢？
Series 對(duì)象有一個(gè) rolling 方法，專(zhuān)門(mén)用來(lái)做移動(dòng)計(jì)算，我們來(lái)看一下。
import pandas as pd
amount = pd.Series(
[100, 90, 110, 150, 110, 130, 80, 90, 100, 150])
print(amount.rolling(3).sum())
"""
0 NaN # NaN + NaN + 100
1 NaN # NaN + 100 + 90
2 300.0 # 100 + 90 + 110
3 350.0 # 90 + 110 + 150
4 370.0 # 110 + 150 + 110
5 390.0 # 150 + 110 + 130
6 320.0 # 110 + 130 + 80
7 300.0 # 130 + 80 + 90
8 270.0 # 80 + 90 + 100
9 340.0 # 90 + 100 + 150
dtype: float64
"""
結(jié)果和我們想要的是一樣的，amount.rolling(3) 相當(dāng)于創(chuàng)建了一個(gè)長(zhǎng)度為 3 的窗口，窗口從上到下依次滑動(dòng)，我們畫(huà)一張圖：
amount.rolling(3) 就做了類(lèi)似于圖中的事情，然后在其基礎(chǔ)上調(diào)用 sum，會(huì)將每個(gè)窗口里面的元素加起來(lái)，就得到上面代碼輸出的結(jié)果。另外窗口的大小可以任意，這里我們以 3 為例。
除了sum，還可以求平均值、求方差等等，可以進(jìn)行很多的操作，有興趣可以自己嘗試一下。當(dāng)然我們也可以自定義函數(shù)：
import pandas as pd
import numpy as np
amount = pd.Series(
[100, 90, 110, 150, 110, 130, 80, 90, 100, 150])
print(
# 調(diào)用 agg 方法，傳遞一個(gè)函數(shù)
# 參數(shù) x 就是每個(gè)窗口里面的元素組成的 Series 對(duì)象
amount.rolling(3).agg(lambda x: np.sum(x) * 2)
)
"""
0 NaN # (NaN + NaN + 100) * 2
1 NaN # (NaN + 100 + 90) * 2
2 600.0 # (100 + 90 + 110) * 2
3 700.0 # (90 + 110 + 150) * 2
4 740.0 # (110 + 150 + 110) * 2
5 780.0 # (150 + 110 + 130) * 2
6 640.0 # (110 + 130 + 80) * 2
7 600.0 # (130 + 80 + 90) * 2
8 540.0 # (80 + 90 + 100) * 2
9 680.0 # (90 + 100 + 150) * 2
dtype: float64
"""
agg 里面的函數(shù)的邏輯可以任意，但返回的必須是一個(gè)數(shù)值。
此外我們注意到，開(kāi)始的兩個(gè)元素為 NaN，這是因?yàn)?rolling(3) 表示從當(dāng)前位置往上篩選，總共篩選 3 個(gè)元素，圖上已經(jīng)畫(huà)的很清晰了。但如果我們希望元素不夠的時(shí)候有多少算多少，該怎么辦呢？比如：第一個(gè)窗口里面的元素之和就是第一個(gè)元素，第二個(gè)窗口里面的元素之和是第一個(gè)元素加上第二個(gè)元素。
import pandas as pd
amount = pd.Series(
[100, 90, 110, 150, 110, 130, 80, 90, 100, 150])
print(
# min_periods 表示窗口的最小觀測(cè)值
amount.rolling(3, min_periods=1).sum()
)
"""
0 100.0
1 190.0
2 300.0
3 350.0
4 370.0
5 390.0
6 320.0
7 300.0
8 270.0
9 340.0
dtype: float64
"""
添加一個(gè) min_periods 參數(shù)即可實(shí)現(xiàn)，這個(gè)參數(shù)表示窗口的最小觀測(cè)值，即：窗口里面元素的最小數(shù)量，默認(rèn)它和窗口的長(zhǎng)度相等。我們窗口長(zhǎng)度為 3，但指定了 min_periods 為 1，表示元素不夠也沒(méi)關(guān)系，只要有一個(gè)就行。
因此元素不夠的話，有幾個(gè)就算幾個(gè)。如果我們指定 min_periods 為 2 的話，那么會(huì)是什么結(jié)果呢？顯然第一個(gè)是 NaN，第二個(gè)還是 190.0，因?yàn)榇翱诶锩娴脑貍€(gè)數(shù)至少為 2。
import pandas as pd
amount = pd.Series(
[100, 90, 110, 150, 110, 130, 80, 90, 100, 150])
print(
# 窗口的最小觀測(cè)值為 2
amount.rolling(3, min_periods=2).sum()
)
"""
0 NaN
1 190.0
2 300.0
3 350.0
4 370.0
5 390.0
6 320.0
7 300.0
8 270.0
9 340.0
dtype: float64
"""
- 注意：min_periods必須小于等于窗口長(zhǎng)度，否則報(bào)錯(cuò)。
rolling 里面還有一個(gè) center 參數(shù)，默認(rèn)為 False。我們知道 rolling(3) 表示從當(dāng)前元素往上篩選，加上本身總共篩選 3 個(gè)。
但如果將 center 指定為 True 的話，那么會(huì)以當(dāng)前元素為中心，從兩個(gè)方向上進(jìn)行篩選。比如 rolling(3, center=True)，那么會(huì)往上選一個(gè)、往下選一個(gè)，再加上本身總共 3 個(gè)。所以示意圖會(huì)變成下面這樣：
我們來(lái)測(cè)試一下：
import pandas as pd
amount = pd.Series(
[100, 90, 110, 150, 110, 130, 80, 90, 100, 150])
print(
amount.rolling(3, center=True).sum()
)
"""
0 NaN
1 300.0
2 350.0
3 370.0
4 390.0
5 320.0
6 300.0
7 270.0
8 340.0
9 NaN
dtype: float64
"""
這里沒(méi)有指定 min_periods，最小觀測(cè)值和窗口長(zhǎng)度相等，所以 rolling(3, center=True)會(huì)使得開(kāi)頭出現(xiàn)一個(gè) NaN，結(jié)尾出現(xiàn)一個(gè) NaN。
這時(shí)候可能有人好奇了，如果窗口的長(zhǎng)度為奇數(shù)的話很簡(jiǎn)單，比如長(zhǎng)度為 9，那么往上選 4 個(gè)、往下選 4 個(gè)，加上本身正好 9 個(gè)。但如果窗口的長(zhǎng)度為偶數(shù)該怎么辦？比如長(zhǎng)度為 8，這個(gè)時(shí)候會(huì)往上選 4 個(gè)、往下選 3 個(gè)，加上本身正好 8 個(gè)。
另外我們還可以從上往下篩選，比如窗口長(zhǎng)度為 3，但我們是希望從當(dāng)前元素開(kāi)始往下篩選，加上本身總共篩選 3 個(gè)。
import pandas as pd
from pandas.api.indexers import FixedForwardWindowIndexer
amount = pd.Series(
[100, 90, 110, 150, 110, 130, 80, 90, 100, 150])
print(
amount.rolling(
FixedForwardWindowIndexer(window_size=3)).sum()
)
"""
0 300.0
1 350.0
2 370.0
3 390.0
4 320.0
5 300.0
6 270.0
7 340.0
8 NaN
9 NaN
dtype: float64
"""
通過(guò)類(lèi)FixedForwardWindowIndexer即可實(shí)現(xiàn)這一點(diǎn)，當(dāng)然此時(shí)就不可以指定 center 參數(shù)了。
調(diào)用 amount.rolling() 會(huì)返回一個(gè) Rolling 對(duì)象，再調(diào)用 Rolling 對(duì)象的 sum, max, min, mean, std 等方法即可對(duì)每個(gè)窗口求總和、最大值、最小值等等。當(dāng)然我們也可以調(diào)用 agg 方法，里面?zhèn)魅胍粋€(gè)函數(shù)，來(lái)自定義每個(gè)窗口的計(jì)算邏輯。然后重點(diǎn)是，agg 里面除了接收一個(gè)函數(shù)之外，還能接收一個(gè)列表，列表里面可以有多個(gè)函數(shù)，然后同時(shí)執(zhí)行多個(gè)操作。
import pandas as pd
import numpy as np
amount = pd.Series(
[100, 90, 110, 150, 110, 130, 80, 90, 100, 150])
print(
amount.rolling(3).agg(
[np.sum, np.mean, lambda x: np.sum(x) * 2])
)
# 執(zhí)行多個(gè)操作，那么會(huì)返回一個(gè) DataFrame
"""
sum mean <lambda>
0 NaN NaN NaN
1 NaN NaN NaN
2 300.0 100.000000 600.0
3 350.0 116.666667 700.0
4 370.0 123.333333 740.0
5 390.0 130.000000 780.0
6 320.0 106.666667 640.0
7 300.0 100.000000 600.0
8 270.0 90.000000 540.0
9 340.0 113.333333 680.0
"""
除了 Series 之外，DataFrame 也有 rolling 方法，功能和用法是一樣的，只不過(guò)后者可以同時(shí)作用于多列。但大部分情況下，我們都調(diào)用 Series 對(duì)象的 rolling 方法。
rolling 方法還有一個(gè)強(qiáng)大的功能，就是它可以對(duì)時(shí)間進(jìn)行移動(dòng)分析，因?yàn)?pandas 本身就誕生在金融領(lǐng)域，所以非常擅長(zhǎng)對(duì)時(shí)間的操作。
那么對(duì)時(shí)間進(jìn)行移動(dòng)分析的使用場(chǎng)景都有哪些呢？舉一個(gè)筆者在大四實(shí)習(xí)時(shí)所遇到的問(wèn)題吧，當(dāng)時(shí)在用 pandas 做審計(jì)，遇到過(guò)這樣一個(gè)需求：判斷是否存在 30 秒內(nèi)充值次數(shù)超過(guò) 1000 次的情況（也就是檢測(cè)是否存在同時(shí)大量充值的情況），如果有就把它們找出來(lái)。
因?yàn)槊恳淮纬渲刀紝?duì)應(yīng)一條記錄，每條記錄都有一個(gè)具體的時(shí)間，換句話說(shuō)就是要判斷是否存在某個(gè) 30 秒，在這其中出現(xiàn)了超過(guò) 1000 條的記錄。當(dāng)時(shí)剛實(shí)習(xí)，被這個(gè)問(wèn)題直接搞懵了，不過(guò)有了 rolling 方法就變得簡(jiǎn)單多了。
import pandas as pd
amount = pd.Series(
[100, 100, 100, 100, 100, 100, 100, 100, 100, 100],
index=pd.DatetimeIndex(
["2020-1-1", "2020-1-3", "2020-1-4", "2020-1-6",
"2020-1-7", "2020-1-9", "2020-1-12", "2020-1-13",
"2020-1-14", "2020-1-15"])
)
print(amount)
"""
2020-01-01 100
2020-01-03 100
2020-01-04 100
2020-01-06 100
2020-01-07 100
2020-01-09 100
2020-01-12 100
2020-01-13 100
2020-01-14 100
2020-01-15 100
dtype: int64
"""
# 這里我們還是算 3 天之內(nèi)的總和吧
# 為了簡(jiǎn)單直觀我們把值都改成100
print(amount.rolling("3D").sum())
"""
2020-01-01 100.0
2020-01-03 200.0
2020-01-04 200.0
2020-01-06 200.0
2020-01-07 200.0
2020-01-09 200.0
2020-01-12 100.0
2020-01-13 200.0
2020-01-14 300.0
2020-01-15 300.0
dtype: float64
"""
我們來(lái)分析一下，首先 rolling("3D") 表示篩選 3 天之內(nèi)的，而且如果是對(duì)時(shí)間進(jìn)行移動(dòng)分析的話，那么要求索引必須是 datetime 類(lèi)型。
- 先看 2020-01-01，它上面沒(méi)有記錄了，所以是100（此時(shí)就沒(méi)有NaN了）；
- 然后是 2020-01-03，由于上面的 2020-01-01 和它之間沒(méi)有超過(guò)3天，所以加起來(lái)總共是200；
- 再看 2020-01-12，由于它只能往上找 2020-01-10, 2020-01-11，然后加在一起。但它的上面是 2020-01-09，已經(jīng)超過(guò)3天了，所以結(jié)果是 100（就是它本身）；
- 最后看 2020-01-14，3 天之內(nèi)的話，應(yīng)該 2020-01-12, 2020-01-13，再加上自身的 2020-01-14，所以結(jié)果是300。2020-01-15 也是同理。
怎么樣，是不是很簡(jiǎn)單呢？回到筆者當(dāng)初的那個(gè)問(wèn)題上來(lái)，如果是找出 30 秒內(nèi)超過(guò) 1000 次的記錄的話，將交易時(shí)間設(shè)置為索引、直接 rolling("30S").count()。然后找出大于 1000 的記錄，說(shuō)明該條記錄往上的第 1000 條記錄的交易時(shí)間和該條記錄的交易時(shí)間之差的絕對(duì)值不超過(guò) 30 秒（記錄是按照交易時(shí)間排好序的）。
至于這 30 秒內(nèi)到底交易了多少次，直接將該條記錄的交易時(shí)間減去 30 秒，進(jìn)行篩選就行了。所以用 rolling 方法處理該問(wèn)題非常方便，但當(dāng)時(shí)不知道，傻了吧唧地寫(xiě) for 循環(huán)一條條遍歷。
另外，關(guān)于 pandas 中表示時(shí)間的符號(hào)估計(jì)有人還不太清楚，最主要的是容易和 Python datetime 在格式化時(shí)所使用的符號(hào)搞混，下面我們來(lái)區(qū)分一下。
感覺(jué)如何，是不是既好用，功能又強(qiáng)大呢？
