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

Question

我试图创建一个索引循环FIFO(先进先出)缓冲区,用于保存烛台图表的最后90分钟,该图表用于按分钟汇总的pandas数据帧中的一组15个资产(即window_size = 150)实时显示在客户端应用程序(蜡烛棒图).它将分别为每个(1m)时间步长的每个资产维持近,开,高,低和音量特征.各个烛台将由websocket更新,其中最新的时间间隔将根据价格(烛台)变化进行更新.在pandas中表示此数据结构的最有效机制是什么,客户端应用程序需要输出形状[5,15,90],如as_frame中所示,表示[close,open,high,low,volume] 15个资产和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

我怎样才能最有效地在熊猫中实现上述问题？如果在给定另一个数据结构的情况下不存在更优化/更高效的解决方案？

Answer 1

添加数据应该像使用appenda一样简单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)