Volume Bar Aggregator script found from internet, there were some interesting discussions on reddit.

volume_bar_aggregator.py

#https://pastebin.com/jgr4ZBVK
#https://www.reddit.com/r/algotrading/comments/p4flbp/can_i_make_tick_volume_and_dollar_bars_from_ohlc/

class VolumeBarAggregator():

    class NumpyWrapper():
        def __init__(self, length):
            self.arr = np.array(dtype=float, )

    def __init__(self):
        self.threshold = 0
        self.index = 0

        self.volumes = []
        self.first_run = True

        self.date = None
        self.start_index = 0
        self.end_index = 0
        self.open = 0
        self.high = 0
        self.low = 0
        self.close = 0
        self.cum_volume = 0
        self.candle_count = 0

        self.arr_date = []
        self.arr_start_index = []
        self.arr_end_index = []
        self.arr_lab_start_index = []
        self.arr_lab_end_index = []
        self.arr_open = []
        self.arr_high = []
        self.arr_low = []
        self.arr_close = []
        self.arr_cum_volume = []
        self.arr_candle_count = []


    def convert(self, df, rolling_median_interval, rolling_median_window, rolling_median_multiplikator, keep_unfinished_canldes=False, embargo=0):
        data = df.to_json() + f"{rolling_median_interval}{rolling_median_window}{rolling_median_multiplikator}{embargo}{keep_unfinished_canldes}"
        hash = (hashlib.md5(data.encode('utf-8')).hexdigest())

        cache_path = os.path.join(gettempdir(), hash + ".csv")
        if False and os.path.isfile(cache_path):
            logger.info("Lade gecachte VolumeBar: " + cache_path)
            return file_utils.read_standard_csv(cache_path)
        else:
            df = df.reset_index(drop=True)
            df["rolling_median_volume"] = df.volume.rolling(rolling_median_window).median().shift(1) * rolling_median_multiplikator

            tmp = df[['date', 'open', 'high', 'low', 'close', 'volume', 'rolling_median_volume']].copy()
            date = tmp.columns.get_loc("date")
            open = tmp.columns.get_loc("open")
            high = tmp.columns.get_loc("high")
            low = tmp.columns.get_loc("low")
            close = tmp.columns.get_loc("close")
            volume = tmp.columns.get_loc("volume")
            rolling_median_volume = tmp.columns.get_loc("rolling_median_volume")
            numpy_df = tmp.to_numpy()

            l = len(numpy_df)
            last_index = l - 1
            tmp = int(l / 10)

            for i in tqdm.trange(l):
                #if i % tmp == 0:
                #    logger.info(f"VolumeBarAggregator: {i} / {l}")
                    #logger.info(f"{i} von {l}")

                # Volume Bars
                if rolling_median_interval == 0 or i % rolling_median_interval == 0:
                    self.threshold = numpy_df[i, rolling_median_volume]
                if np.nan_to_num(self.threshold) > 0:
                    if self.first_run:
                        append = True
                    else:
                        append = self.cum_volume > self.threshold

                    if append:
                        self.first_run = False

                        # Appenden
                        if self.date is not None:
                            self.arr_date.append(self.date)
                            self.arr_start_index.append(self.start_index)
                            self.arr_end_index.append(self.end_index)
                            self.arr_open.append(self.open)
                            self.arr_high.append(self.high)
                            self.arr_low.append(self.low)
                            self.arr_close.append(self.close)
                            self.arr_cum_volume.append(self.cum_volume)
                            self.arr_candle_count.append(self.candle_count)

                            # Damit Lookahead Bias nicht entsteht wenn man die fertigen Indikatoren wieder reappliziert.
                            # Embargo ist dazu da im Backtest Modus die verzögerung der Berechnung zu simulieren.
                            if len(self.arr_lab_end_index) > 0:
                                self.arr_lab_end_index[-1] = self.end_index + embargo
                            self.arr_lab_start_index.append(self.end_index + 1 + embargo)
                            self.arr_lab_end_index.append(last_index)

                        self.date = numpy_df[i, date]
                        self.start_index = i
                        self.end_index = i
                        self.open = numpy_df[i, open]
                        self.high = numpy_df[i, high]
                        self.low = numpy_df[i, low]
                        self.close = numpy_df[i, close]
                        self.cum_volume = numpy_df[i, volume]
                        self.candle_count = 1
                    else:
                        self.end_index = i
                        self.high = max(self.high, numpy_df[i, high])
                        self.low = min(self.low, numpy_df[i, low])
                        self.close = numpy_df[i, close]
                        self.cum_volume += numpy_df[i, volume]
                        self.candle_count += 1

            logger.info(f"Letzte Candle in VolumeDF: {self.arr_date[-1]}")

            if keep_unfinished_canldes:
                self.arr_date.append(self.date)
                self.arr_start_index.append(self.start_index)
                self.arr_end_index.append(self.end_index)

                if len(self.arr_lab_end_index) > 0:
                    self.arr_lab_end_index[-1] = last_index - 2
                self.arr_lab_start_index.append(last_index - 1)
                self.arr_lab_end_index.append(last_index)

                self.arr_open.append(self.open)
                self.arr_high.append(self.high)
                self.arr_low.append(self.low)
                self.arr_close.append(self.close)
                self.arr_cum_volume.append(self.cum_volume)
                self.arr_candle_count.append(self.candle_count)

            vol_bars = pd.DataFrame(
                {
                    "date": self.arr_date,
                    "start_index": self.arr_start_index,
                    "end_index": self.arr_end_index,
                    "lab_start_index": self.arr_lab_start_index,
                    "lab_end_index": self.arr_lab_end_index,
                    "open": self.arr_open,
                    "high": self.arr_high,
                    "low": self.arr_low,
                    "close": self.arr_close,
                    "cum_volume": self.arr_cum_volume,
                    "volume": self.arr_cum_volume,
                    "candle_count": self.arr_candle_count
                }
            )

            vol_bars = vol_bars.reset_index(drop=True)
            vol_bars.to_csv(cache_path)
            return vol_bars