Pandas滑动窗口，揪出刷单黄牛党！

import random
import pandas as pd

def random_timestamp(hour, ymd='2021-12-01'):
    """
    生成年月日固定，分秒随机的时间戳
    :param ymd:
    :param hour: 传入小时
    :return:
    """

    def tstr():
        tms = set([random.randrange(1, 59, 2) for _ in range(20)])
        return [f'2020-12-01 {hour}:{ms}:{ms}' for ms in tms]

    return [pd.Timestamp(ws) for ws in tstr()]

series = random_timestamp(9)

[Timestamp('2020-12-01 09:01:01'), Timestamp('2020-12-01 09:35:35'), Timestamp('2020-12-01 09:05:05'), Timestamp('2020-12-01 09:07:07'), Timestamp('2020-12-01 09:09:09'), Timestamp('2020-12-01 09:11:11'), Timestamp('2020-12-01 09:13:13'), Timestamp('2020-12-01 09:19:19'), Timestamp('2020-12-01 09:53:53'), Timestamp('2020-12-01 09:55:55'), Timestamp('2020-12-01 09:23:23'), Timestamp('2020-12-01 09:21:21'), Timestamp('2020-12-01 09:29:29'), Timestamp('2020-12-01 09:31:31')]

import pandas as pd

def is_outlier(time_series: pd.Series, outlier_duration=20, outlier_count=5):
    """
    outlier_duration分钟内连续交易outlier_count次认为是异常
    :param time_series:
    :param outlier_duration:
    :param outlier_count:
    :return:
    """
    # 排序
    time_series2 = time_series.sort_values()
    # 基于最早时间的时间跨度
    time_delta = (time_series2 - time_series2.iloc[0]) / np.timedelta64(1, 's') / 60.
    # 求得长度为outlier_count的滑动窗口的时间跨度
    time_cumsum = time_delta.rolling(outlier_count).sum()
     # 若outlier_count次交易的时间小于outlier_duration，就是所定义的刷单，返回True
    if (time_cumsum <= outlier_duration).any():
        return True
    # 不存在刷单
    return False

series = random_timestamp(9) # hour 等于9点
print(series)

result = is_outlier(pd.Series(series))
print(result)