finmlkit.bar.logic module

This module contains the logic for generating time, tick, volume, and dollar bar. These functions return the open indices of the bar in the raw trades data.

finmlkit.bar.logic._cusum_bar_indexer(timestamps: ndarray[tuple[int, ...], dtype[int64]], prices: ndarray[tuple[int, ...], dtype[float64]], sigma: ndarray[tuple[int, ...], dtype[float64]], sigma_floor: float, sigma_mult: float) → List[source]

Determine CUSUM bar open indices using a symmetric CUSUM filter on successive price changes (López de Prado, 2018).

A new bar starts whenever the cumulative sum of price changes exceeds +sigma*lambda or –sigma*lambda.

Parameters:

  • timestamps – timestamps of the trades.

  • prices – Trade prices.

  • sigma – Threshold vector for CUSUM (e.g. calculated EWMS volatility or constant).

  • sigma_floor – Minimum value for sigma to avoid division by zero.

  • sigma_mult – sigma multiplier for the CUSUM filter (threshold will be lambda_mult*sigma).

Returns:

close_indices

finmlkit.bar.logic._dollar_bar_indexer(prices: ndarray[tuple[int, ...], dtype[int64]], volumes: ndarray[tuple[int, ...], dtype[float64]], threshold: float) → List[source]

Determine the dollar bar open indices using cumulative dollar value.

Parameters:

  • prices – Trade prices.

  • volumes – Trade volumes.

  • threshold – Dollar value threshold for opening a new bar.

Returns:

close_indices: Timestamps at which each dollar bar opens.

Note

The first trade is always the start of a bar. A new bar is opened when the cumulative dollar value (price × volume) meets or exceeds the threshold.

finmlkit.bar.logic._imbalance_bar_indexer(timestamps: ndarray[tuple[int, ...], dtype[int64]], prices: ndarray[tuple[int, ...], dtype[float64]], volumes: ndarray[tuple[int, ...], dtype[float64]], threshold: float) → Tuple[ndarray[tuple[int, ...], dtype[int64]], ndarray[tuple[int, ...], dtype[int64]]][source]

Determine the imbalance bar open indices based on cumulative imbalance.

Parameters:

  • timestamps – Raw trade timestamps.

  • prices – Trade prices.

  • volumes – Trade volumes.

  • threshold – Imbalance threshold for opening a new bar.

Returns:

A tuple of open timestamps and indices for imbalance bars.

Raises:

NotImplementedError – Always raised as this function is not yet implemented.

finmlkit.bar.logic._run_bar_indexer(timestamps: ndarray[tuple[int, ...], dtype[int64]], prices: ndarray[tuple[int, ...], dtype[float64]], volumes: ndarray[tuple[int, ...], dtype[float64]], threshold: float) → Tuple[ndarray[tuple[int, ...], dtype[int64]], ndarray[tuple[int, ...], dtype[int64]]][source]

Determine the run bar open indices using cumulative run activity.

Parameters:

  • timestamps – Raw trade timestamps.

  • prices – Trade prices.

  • volumes – Trade volumes.

  • threshold – Run threshold for opening a new bar.

Returns:

A tuple of open timestamps and indices for run bars.

Raises:

NotImplementedError – Always raised as this function is not yet implemented.

finmlkit.bar.logic._tick_bar_indexer(timestamps: ndarray[tuple[int, ...], dtype[int64]], threshold: int) → List[source]

Determine the tick bar open indices in the raw trades timestamp array.

Parameters:

  • timestamps – Raw trade timestamps.

  • threshold – The tick count threshold for opening a new bar.

Returns:

close_indices: Timestamps at which each tick bar opens.

Note

The first trade is always the start of a bar. A new bar is opened every time the tick count reaches the specified threshold.

finmlkit.bar.logic._time_bar_indexer(timestamps: ndarray[tuple[int, ...], dtype[int64]], interval_seconds: float) → Tuple[ndarray[tuple[int, ...], dtype[int64]], ndarray[tuple[int, ...], dtype[int64]]][source]

Determine the time bar open indices in the raw trades timestamp array.

Parameters:

  • timestamps – Raw sorted trade timestamps in nanoseconds.

  • interval_seconds – Length of the time bar in seconds.

Returns:

A tuple of: - bar_close_ts: Timestamps at which each bar closes. - bar_close_indices: Indices in the trade data corresponding to bar closings.

Note

The first bar is aligned to the ceiling of the first timestamp, ensuring consistent bar boundaries. Duplicate indices may occur if a bar interval contains no trades (empty bars).

finmlkit.bar.logic._volume_bar_indexer(volumes: ndarray[tuple[int, ...], dtype[float64]], threshold: float) → List[source]

Determine the volume bar open indices using cumulative volume. :param volumes: Trade volumes. :param threshold: Volume bucket threshold for opening a new bar. :returns: close_indices: Timestamps at which each volume bar opens.

Note

The first trade is always the start of a bar. A new bar is opened when the cumulative trade volume meets or exceeds the threshold.