python - 如何使用 pandas 创建索引循环 FIFO 缓冲区

Question

我正在尝试创建一个索引循环 FIFO(先进先出)缓冲区，用于保存 pandas 数据框中一组 15 种 Assets 的烛台图表的最后 90 分钟，按分钟聚合(即 window_size=150)以显示在客户端应用程序上实时(烛台图)。它将分别为每 (1m) 时间步长的每个 Assets 维护收盘价、开盘价、最高价、最低价和成交量特征。单个烛台将由 websocket 更新，从而在每次价格(烛台)变化时更新最新的时间间隔。在 pandas 中表示此数据结构的最有效机制是什么，客户端应用程序需要输出形状 [5,15,90]，如 as_frame 所示，表示 [close, open, high, low, volume] 15 项 Assets 和 90 个区间。

因此数据将被表示为:

assets  time   close    open    high    low    volume
asset1  time1  0.001    0.002   0.003   0.001  0.001
        time2  0.001    0.001   0.003   0.001  0.001
        ...
        time90 ...

...

asset15 time1  0.001    0.002   0.003   0.001  0.001
        time2  0.001    0.001   0.003   0.001  0.001
        ...
        time90 ...

我已经用 python pandas 实现了一个简单的解决方案:

class Buffer():
    def __init__(self):
        self.cols = [
                'asset',
                'timestamp',
                'close',
                'high',
                'low'
        ];

        self.lvls = [
            'asset',
            'timestamp'
        ]

        self.frame = pd.DataFrame(
            columns=self.cols
        );

        self.frame.set_index(self.lvls)

    def add(
        self,
        entry
    ):
        ... what would be the most effective
        mechanism to add to the multi indexed
        dataframe given the entry/record 
        {
           "asset":"ASSET",
           "timestamp": 158090000, 
           "close":1.3, 
           "high":1.4, 
           "low":1.2, 
           "open":1.3, 
           "volume":134.5 
        } 
        such that the dataframe timestamp 
        index does not exceed the given 
        window size?

    def as_frame(
        self,
        assets,
        features,
        window_size
    ):
        outframe = self.frame.set_index(self.lvls)
        outframe = outframe.groupby(self.lvls).last()
        outlist = outframe.to_xarray().to_array()
        return outlist

我怎样才能最有效地在 pandas 中解决上述问题？如果给定另一种数据结构不存在更优/性能更好的解决方案？

Answer 1

添加数据应该像使用 append 和 dict 一样简单，过滤目标 DataFrame 中的列。

self.frame.append({k:v for k, v in entry.items() if k in self.cols})

根据你的问题，我认为你想保留一个 DataFrame，它在传入数据上保持一个连续的 150 秒窗口。

要过滤生成的帧，我会从最新添加的时间窗口计算“开始”时间戳，并使用它来过滤行，如下所示:

start_ts = entry['timestamp'] - 150
self.frame = self.frame[ self.frame['ts'] >= start_ts ]

根据 DataFrame 的大小，您可能会获得更好的性能:

self.frame.drop(self.frame[ self.frame['ts'] < start_ts ].index, inplace=True)