基于Python DataFrame计算带标签数据滚动时间窗口的欧氏距离

样本DataFrame #

timestamps labels
0  2023-08-01 00:00:00     A
1  2023-08-01 03:00:00     B
2  2023-08-01 06:00:00     C
3  2023-08-01 09:00:00     A
4  2023-08-01 12:00:00     B
5  2023-08-01 15:00:00     C
6  2023-08-01 18:00:00     A
7  2023-08-01 21:00:00     B

需求 #

1. 获取全数据集Top N（此处为10）高频标签的计数数组；
2. 在14D的滚动时间窗口内，获取窗口内Top N高频标签的计数数组（两个数组的第11位为1减去数组归一化和，可忽略）；
3. 计算两个数组的欧氏距离并随时间戳绘图。

现有问题 #

现有代码可实现需求，但处理超大数据集时，滚动窗口迭代效率极低，且需忽略首个时间窗口前的计算结果，求更高效的实现方法。

现有代码 #

import numpy as np
import pandas as pd
from scipy.spatial.distance import euclidean

def get_label_array(label_series, top_N=10):
    label_counts = label_series.value_counts().head(top_N)
    other_count = len(label_series) - label_counts.sum()
    
    label_arr = label_counts.values
    label_arr = np.append(label_arr, other_count)
    normalized_label_arr = label_arr / np.sum(label_arr)
    return normalized_label_arr

def calculate_rolling_dist(df, timestamps, time_window='14D', top_N=10):
    datetimestamps = pd.to_datetime(timestamps)
    df.set_index(datetimestamps, inplace=True)
    top_N_lft = get_label_array(df['labels'], top_N)
    rolling_result = []
    rolling_ts = []
    for window_start in df.index[:-1]:
        rolling_ts.append(window_start)   
        print(window_start)   
        window_end = window_start + pd.Timedelta(time_window)
        window_data = df.loc[window_start:window_end]
        if len(window_data) > 0:
            lft = get_label_array(window_data['labels'], top_N)
            if len(lft) >= 10:
                rolling_result.append(euclidean(top_N_lft,lft))
    return np.array(rolling_result), np.array(rolling_ts)

核心优化思路 #

抛弃逐行窗口迭代，利用Pandas的矢量化滚动操作和预计算Top N标签压缩时间复杂度，具体步骤如下：

步骤1：预计算全量Top N标签 #

先从全数据集提取固定的Top N高频标签，后续所有窗口统一基于这些标签统计，避免重复计算Top N：

def get_top_labels(label_series, top_N=10):
    return label_series.value_counts().head(top_N).index.tolist()

步骤2：生成标签的one-hot编码矩阵 #

将每个样本的标签转换为Top N标签对应的二进制列，同时生成"其他"标签列，方便后续滚动计数：

def prepare_label_matrix(df, top_labels):
    df = df.copy()
    for label in top_labels:
        df[f'label_{label}'] = (df['labels'] == label).astype(int)
    df['label_other'] = (~df['labels'].isin(top_labels)).astype(int)
    return df

步骤3：滚动窗口计数与归一化 #

用Pandas原生的时间窗口滚动求和，直接计算每个窗口内的标签计数，再逐行归一化得到数组：

def calculate_rolling_normalized(df, time_window='14D'):
    label_cols = [col for col in df.columns if col.startswith('label_')]
    # 滚动求和：窗口包含当前时间戳及之前14天的数据
    rolling_counts = df[label_cols].rolling(time_window, closed='right').sum()
    
    # 归一化处理，避免除以0
    row_sums = rolling_counts.sum(axis=1)
    row_sums[row_sums == 0] = np.nan
    rolling_normalized = rolling_counts.div(row_sums, axis=0)
    
    return rolling_normalized

步骤4：计算欧氏距离并过滤无效窗口 #

全量归一化数组是固定值，通过广播运算与滚动归一化数组逐行计算欧氏距离，自动过滤无数据的窗口：

def compute_euclidean_distances(full_normalized, rolling_normalized):
    # 将全量数组广播为与滚动数组行数一致的矩阵
    full_arr = np.tile(full_normalized, (len(rolling_normalized), 1))
    # 逐行计算欧氏距离
    distances = np.sqrt(((full_arr - rolling_normalized.values)**2).sum(axis=1))
    # 包装为带时间戳的Series并过滤无效值
    dist_series = pd.Series(distances, index=rolling_normalized.index).dropna()
    return dist_series

完整整合代码 #

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

def get_top_labels(label_series, top_N=10):
    return label_series.value_counts().head(top_N).index.tolist()

def prepare_label_matrix(df, top_labels):
    df = df.copy()
    for label in top_labels:
        df[f'label_{label}'] = (df['labels'] == label).astype(int)
    df['label_other'] = (~df['labels'].isin(top_labels)).astype(int)
    return df

def calculate_rolling_normalized(df, time_window='14D'):
    label_cols = [col for col in df.columns if col.startswith('label_')]
    rolling_counts = df[label_cols].rolling(time_window, closed='right').sum()
    row_sums = rolling_counts.sum(axis=1)
    row_sums[row_sums == 0] = np.nan
    rolling_normalized = rolling_counts.div(row_sums, axis=0)
    return rolling_normalized

def compute_euclidean_distances(full_normalized, rolling_normalized):
    full_arr = np.tile(full_normalized, (len(rolling_normalized), 1))
    distances = np.sqrt(((full_arr - rolling_normalized.values)**2).sum(axis=1))
    dist_series = pd.Series(distances, index=rolling_normalized.index).dropna()
    return dist_series

def main(df, time_window='14D', top_N=10):
    # 处理时间戳索引并排序
    df['timestamps'] = pd.to_datetime(df['timestamps'])
    df = df.set_index('timestamps').sort_index()
    
    # 1. 生成全量归一化数组
    top_labels = get_top_labels(df['labels'], top_N)
    full_counts = [df['labels'].value_counts()[label] for label in top_labels]
    full_counts.append(len(df) - sum(full_counts))
    full_normalized = np.array(full_counts) / sum(full_counts)
    
    # 2. 计算滚动窗口归一化数组
    df = prepare_label_matrix(df, top_labels)
    rolling_normalized = calculate_rolling_normalized(df, time_window)
    
    # 3. 计算欧氏距离
    dist_series = compute_euclidean_distances(full_normalized, rolling_normalized)
    
    # 4. 绘图
    plt.figure(figsize=(12,6))
    dist_series.plot()
    plt.title(f'Euclidean Distance between Full Dataset and {time_window} Rolling Window')
    plt.xlabel('Timestamp')
    plt.ylabel('Distance')
    plt.grid(True)
    plt.show()
    
    return dist_series

# 测试样本数据
sample_df = pd.DataFrame({
    'timestamps': ['2023-08-01 00:00:00', '2023-08-01 03:00:00', '2023-08-01 06:00:00',
                   '2023-08-01 09:00:00', '2023-08-01 12:00:00', '2023-08-01 15:00:00',
                   '2023-08-01 18:00:00', '2023-08-01 21:00:00'],
    'labels': ['A', 'B', 'C', 'A', 'B', 'C', 'A', 'B']
})

# 执行
dist_result = main(sample_df, time_window='14D', top_N=3)

优化点说明 #

1. 时间复杂度优化：从原代码的O(n²)降至O(n)，完全避免逐行窗口遍历；
2. 统一标签维度：全量和窗口使用同一组Top标签，保证数组维度一致，无需额外补全；
3. 矢量化运算：one-hot矩阵和广播运算大幅提升数组计算效率；
4. 自动过滤无效窗口：通过dropna()直接去除无数据的窗口结果，无需额外判断逻辑。

内容的提问来源于stack exchange，提问作者beaglul