高效迭代数据集计算滚动窗口：比特币1秒数据转OHLCV优化

核心优化思路 #

原方案的低效根源在于重复计算：每次遇到unique_tick=True的行，都要重新切片3600条秒级数据并分组聚合生成分钟OHLCV，而相邻的unique_tick往往共享大量重叠的分钟数据，导致大量冗余计算。

正确的优化路径是：

1. 先一次性将所有秒级数据转换为全局的分钟级OHLCV数据（仅做一次）
2. 针对每个unique_tick，直接从预处理好的分钟数据中提取对应的60分钟窗口

具体实现步骤 #

步骤1：预处理全局分钟级OHLCV

将秒级数据的时间戳转换为datetime类型，用pandas的重采样功能一次性生成所有分钟的OHLCV，这是向量化操作，比循环切片聚合效率高几个数量级：

import pandas as pd

# 将id（毫秒时间戳）转换为datetime索引
klinelog['timestamp'] = pd.to_datetime(klinelog['id'], unit='ms')
klinelog.set_index('timestamp', inplace=True)

# 一次性生成所有分钟级OHLCV
minute_ohlcv = klinelog.resample('1min').agg({
    'close': ['first', 'max', 'min', 'last'],
    'volume': 'sum',
    'id': 'min'
})

# 整理列名，方便后续使用
minute_ohlcv.columns = ['open', 'high', 'low', 'close', 'volume', 'id']
minute_ohlcv = minute_ohlcv.dropna()  # 移除无数据的分钟

步骤2：快速提取每个unique_tick的60分钟窗口

先筛选出所有unique_tick=True的记录，然后针对每条记录的时间，计算其60分钟窗口的时间范围，直接从预处理好的minute_ohlcv中查询：

# 获取所有unique_tick的时间点
unique_ticks = klinelog[klinelog['unique_tick']].copy().reset_index()

# 定义函数，根据起始时间获取60分钟窗口的OHLCV
def get_60min_window(start_time):
    end_time = start_time + pd.Timedelta(minutes=60)
    return minute_ohlcv.loc[start_time:end_time]

# 批量处理所有unique_tick的窗口
unique_ticks['60min_window'] = unique_ticks['timestamp'].apply(get_60min_window)

额外优化点 #

• 如果不需要保留每个窗口的完整DataFrame，可以直接在get_60min_window中对窗口数据做进一步聚合（比如计算窗口内的统计值），减少内存占用
• 若数据量极大（数年数据），可以考虑用Dask进行并行处理，或者将预处理后的分钟数据存储为Parquet格式，后续直接读取使用，避免重复预处理

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