基于毫秒级时间戳的Tick数据滑动窗口统计问题求助

需求说明 #

处理毫秒级时间戳的Tick数据集，需要每隔5分钟取一个时间点，提取该时间点前400秒内的所有数据，计算生成包含均值、最大值、最小值特征的新DataFrame。

现有Tick数据格式示例 #

Timestamp,Bid price,Ask price,Bid volume,Ask volume
2022-09-19 00:00:00:710,3876.152,3876.652,0.00044000000343658,0.000119999996968545
2022-09-19 00:00:00:761,3875.942,3876.652,0.00270000007003546,0.00270000007003546
2022-09-19 00:00:01:033,3875.949,3876.364,0.00044000000343658,0.000119999996968545
2022-09-19 00:00:01:084,3875.636,3876.367,0.00270000007003546,0.00270000007003546
2022-09-19 00:00:01:190,3875.851,3876.351,0.00270000007003546,3.99999998990097E-06
2022-09-19 00:00:01:241,3875.845,3876.501,0.00270000007003546,0.00270000007003546
2022-09-19 00:00:01:865,3875.936,3876.563,0.00270000007003546,3.99999998990097E-06
2022-09-19 00:00:01:925,3875.942,3876.661,0.00270000007003546,0.00270000007003546
2022-09-19 00:00:02:971,3876.034,3876.655,0.00044000000343658,0.000119999996968545
2022-09-19 00:00:03:022,3876.054,3876.761,0.00270000007003546,0.00270000007003546
2022-09-19 00:00:03:072,3876.142,3876.864,0.00270000007003546,0.00270000007003546
2022-09-19 00:00:06:035,3876.454,3876.848,0.00270000007003546,3.99999998990097E-06
2022-09-19 00:00:06:085,3876.451,3877.146,0.00270000007003546,0.00270000007003546
2022-09-19 00:00:07:302,3876.699,3877.103,0.00270000007003546,3.99999998990097E-06
2022-09-19 00:00:07:353,3876.651,3877.349,0.00270000007003546,0.00270000007003546
2022-09-19 00:00:08:147,3876.954,3877.303,0.00270000007003546,3.99999998990097E-06
2022-09-19 00:00:08:198,3876.942,3877.649,0.00270000007003546,0.00270000007003546
2022-09-19 00:00:09:018,3876.937,3877.558,0.00044000000343658,0.000119999996968545
2022-09-19 00:00:09:068,3876.842,3877.552,0.00270000007003546,0.00270000007003546
2022-09-19 00:00:09:602,3876.945,3877.569,0.00270000007003546,3.99999998990097E-06
2022-09-19 00:00:09:653,3876.933,3877.664,0.00270000007003546,0.00270000007003546
2022-09-19 00:00:09:763,3876.642,3877.352,0.00270000007003546,0.00270000007003546

已尝试的方法 #

方法1：使用resample但未满足需求 #

尝试用resample('5min')聚合，但该方法是对5分钟窗口内的数据计算，而非每个5分钟时间点前400秒的数据，代码如下：

import pandas as pd
import numpy as np
from scipy.signal import lombscargle
import matplotlib.pyplot as plt

df = pd.read_csv('SP500.csv')
df["Timestamp"] = pd.to_datetime(df["Timestamp"], format="%Y-%m-%d %H:%M:%S:%f")
df.set_index("Timestamp", inplace=True)

candles = df.resample('5min').agg({
    "MEAN":  "mean",
    "HIGH":  "max",
    "LOW":   "min",
})

方法2：手动生成时间序列但计算失效 #

手动创建间隔5分钟的时间序列DataFrame，但计算特征时结果未随时间变化，代码如下：

import pandas as pd
import numpy as np

df = pd.read_csv('SP500.csv')
df["Timestamp"] = pd.to_datetime(df["Timestamp"], format="%Y-%m-%d %H:%M:%S:%f")

# 生成间隔5分钟的时间序列
statistics = pd.DataFrame()
statistics['Timestamp'] = pd.date_range(start='19/09/2022', periods=646289, freq='5min')

# 计算Bid Mean时结果无变化
statistics['Bid Mean'] = df.loc[((df['Timestamp'] < statistics['Timestamp']) & (df['Timestamp'] > statistics['Timestamp'] - pd.Timedelta(seconds=400)))]['Bid price'].mean()

预期输出结构 #

新DataFrame需包含：

• Timestamp列：间隔5分钟的时间点（如2022-09-19 00:00:00:000、2022-09-19 00:05:00:000）
• Mean列：对应时间点前400秒内Bid/Ask价格的均值
• High列：对应时间点前400秒内Bid/Ask价格的最大值
• Low列：对应时间点前400秒内Bid/Ask价格的最小值

疑问 #

1. 为何手动计算时statistics['Timestamp']未逐行变化，导致特征结果无差异？
2. 有没有更简洁高效的实现方式？
3. 如何在自定义函数中提取指定时间范围内的df子集？

解答 #

1. 手动计算失效原因

你写的代码是向量级别的广播运算，并非逐行遍历：

df.loc[((df['Timestamp'] < statistics['Timestamp']) & (df['Timestamp'] > statistics['Timestamp'] - pd.Timedelta(seconds=400)))]

这里statistics['Timestamp']是一个Series，和df['Timestamp']比较时会生成二维布尔矩阵，但loc无法处理这种多维索引，最终只会取所有时间点都满足的交集数据，计算出一个全局均值并赋值给整列，所以结果不会随时间变化。

2. 简洁高效的实现方式

推荐两种高效方案：

方案一：滚动窗口+目标时间序列对齐

import pandas as pd

# 读取并处理原始数据
df = pd.read_csv('SP500.csv')
df['Timestamp'] = pd.to_datetime(df['Timestamp'], format="%Y-%m-%d %H:%M:%S:%f")
df = df.set_index('Timestamp').sort_index()

# 生成5分钟间隔的目标时间序列
start_time = df.index.min().floor('5min')
end_time = df.index.max().ceil('5min')
target_timestamps = pd.date_range(start=start_time, end=end_time, freq='5min')

# 合并目标时间与原始数据，确保每个时间点都被覆盖
merged = pd.DataFrame(index=target_timestamps).join(df, how='left')

# 滚动窗口计算前400秒的统计值
window = '400S'
result = merged.rolling(window, closed='right').agg({
    'Bid price': ['mean', 'max', 'min'],
    'Ask price': ['mean', 'max', 'min']
}).dropna()

# 整理列名
result.columns = ['Bid_Mean', 'Bid_High', 'Bid_Low', 'Ask_Mean', 'Ask_High', 'Ask_Low']
result = result.reset_index().rename(columns={'index': 'Timestamp'})

方案二：merge_asof时间窗口匹配（适合大规模数据）

import pandas as pd

df = pd.read_csv('SP500.csv')
df['Timestamp'] = pd.to_datetime(df['Timestamp'], format="%Y-%m-%d %H:%M:%S:%f")
df = df.sort_values('Timestamp')

# 生成目标时间序列并计算窗口起始时间
target = pd.DataFrame({
    'Timestamp': pd.date_range(start='2022-09-19', end=df['Timestamp'].max(), freq='5min')
})
target['window_start'] = target['Timestamp'] - pd.Timedelta(seconds=400)

# 匹配每个目标时间点对应的窗口内数据
merged = pd.merge_asof(df, target, on='Timestamp', direction='backward')
merged = merged[merged['Timestamp'] >= merged['window_start']]

# 按目标时间点分组计算特征
result = merged.groupby('Timestamp_y').agg({
    'Bid price': ['mean', 'max', 'min'],
    'Ask price': ['mean', 'max', 'min']
}).reset_index()
result.columns = ['Timestamp', 'Bid_Mean', 'Bid_High', 'Bid_Low', 'Ask_Mean', 'Ask_High', 'Ask_Low']

3. 自定义函数中提取指定时间范围子集

可以通过时间索引切片实现，结合apply逐行处理：

def calculate_features(target_time, df, window_seconds=400):
    start_time = target_time - pd.Timedelta(seconds=window_seconds)
    # 利用时间索引切片提取窗口内数据
    subset = df.loc[start_time:target_time]
    if subset.empty:
        return pd.Series([np.nan, np.nan, np.nan])
    # 计算Bid/Ask合并后的特征
    mean_val = (subset['Bid price'] + subset['Ask price']).mean() / 2
    high_val = max(subset['Bid price'].max(), subset['Ask price'].max())
    low_val = min(subset['Bid price'].min(), subset['Ask price'].min())
    return pd.Series([mean_val, high_val, low_val], index=['Mean', 'High', 'Low'])

# 应用到目标时间序列
target_timestamps = pd.date_range(start='2022-09-19', end=df.index.max(), freq='5min')
result = pd.DataFrame({'Timestamp': target_timestamps})
result[['Mean', 'High', 'Low']] = result['Timestamp'].apply(lambda x: calculate_features(x, df))

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