finmlkit.feature.core.volume module

class finmlkit.feature.core.volume.VolumePro(window_size: Timedelta, n_bins: int = 27, va_pct: float = 68.34)[source]

Bases: object

Encapsulates numba functions for smoother calling and parameter setting.

__init__(window_size: Timedelta, n_bins: int = 27, va_pct: float = 68.34)[source]

Initialize the Volume Profile calculator with the given parameters.

Parameters:

  • window_size – Size of the rolling windows.

  • n_bins – Number of bins for price level bucketing.

  • va_pct – Value area percentage.

Note:

This sets the rolling window size, the bin size for price level bucketing, and the value area percentage used for determining the high and low value areas (HVA and LVA). Default values are window_size_sec=1800, n_bins=27, va_pct=68.34.

compute(bars: DataFrame, fp_data: FootprintData) → tuple[ndarray, ndarray, ndarray, ndarray][source]

Compute the volume profile parameters (POC, HVA, LVA) in a rolling window fashion.

Parameters:

  • bars – DataFrame containing dynamic bars (must include high and low columns).

  • fp_data – FootprintData object with price levels and volume information.

Returns:

Tuple of POC, HVA, and LVA prices, and volume percentage above POC, (as NumPy arrays)

Raises:

AssertionError – If bars and fp_data have different lengths.

Note:

The computation is performed in a rolling window fashion, using the set window size, bin size, and value area percentage. The bars DataFrame must contain ‘high’ and ‘low’ columns, and the footprint data must be aligned in length. Returned arrays represent the Point of Control (POC), High Value Area (HVA), and Low Value Area (LVA) prices for each bar. The computation replaces starting zeros with NaN to indicate insufficient data at the window start.

compute_range(bars: DataFrame, fp_data: FootprintData, start: str | int | Timestamp, end: str | int | Timestamp) → tuple[ndarray, ndarray, ndarray, ndarray, ndarray][source]

Compute the volume profile (POC, HVA, LVA) in a rolling window fashion for a given time range.

Parameters:

  • bars – DataFrame containing dynamic bars with high and low.

  • fp_data – FootprintData object containing volume profiles.

  • start – Start timestamp for slicing (str, int, or pd.Timestamp).

  • end – End timestamp for slicing (same type as start).

Returns:

Tuple of bar timestamps, POC, HVA, and LVA prices.

Raises:

AssertionError – If bars and fp_data lengths differ or start and end are of different types.

Note:

This method computes the rolling window volume profile (POC, HVA, LVA) for a specified time range. The range is set by the start and end timestamps, which must be of the same type. The method internally adjusts the start timestamp for window warm-up and aligns bar and footprint data by timestamp. Returns arrays for the bar timestamps and the computed POC, HVA, and LVA prices over the specified range.

reset_parameters(window_size_sec: int = None, n_bins: int = None, va_pct: float = None)[source]

Reset the parameters of the Volume Profile calculator.

Parameters:

  • window_size_sec – Optional new window size in seconds. If None, the existing value is retained.

  • n_bins – Optional number of bins for price level bucketing. If None, the existing value is retained.

  • va_pct – Optional new value area percentage. If None, the existing value is retained.

Note:

This method allows dynamic reconfiguration of the rolling window size, price bin size, or value area percentage for the Volume Profile calculations. Any parameter left as None will retain its prior value.

finmlkit.feature.core.volume.aggregate_footprint(ts: ndarray, highs: ndarray, lows: ndarray, price_levels: list[ndarray], buy_volumes: list[ndarray], sell_volumes: list[ndarray], start_ts: int, end_ts: int, price_tick: float) → tuple[ndarray, ndarray, ndarray][source]

Aggregate the volume footprint data within a specified time window to calculate the POC, HVA, and LVA.

Parameters:

  • ts – Array of timestamps (int64) for each bar in nanoseconds.

  • highs – High prices for each bar.

  • lows – Low prices for each bar.

  • price_levels – List of arrays of price levels for each bar.

  • buy_volumes – List of arrays of buy volumes for each price level.

  • sell_volumes – List of arrays of sell volumes for each price level.

  • start_ts – Start timestamp of the aggregation window.

  • end_ts – End timestamp of the aggregation window.

  • price_tick – Tick size used to discretize prices.

Returns:

Tuple of: - complete_price_levels: Aggregated price levels in integer tick units. - aligned_buy_volumes: Aggregated buy volumes aligned to tick grid. - aligned_sell_volumes: Aggregated sell volumes aligned to tick grid.

Raises:

AssertionError – If input lists/arrays are not aligned in length or empty.

finmlkit.feature.core.volume.bucket_price_levels(all_price_levels: ndarray, total_volumes: ndarray, n_bins: int) → tuple[ndarray, ndarray][source]

Bucket the price levels and associated volumes using a fixed-size bin to reduce noise.

Parameters:

  • all_price_levels – Array of all price levels.

  • total_volumes – Corresponding total volumes.

  • n_bins – number of bins to create for bucketing the price levels.

Returns:

Tuple of: - binned_price_levels: Midpoints of each bucket. - binned_volumes: Aggregated volumes per bucket.

Raises:

  • AssertionError – If input arrays are empty or of mismatched length.

  • ValueError – If a price level falls outside defined bins.

finmlkit.feature.core.volume.calc_volume_percentage_above_poc(price_levels: ndarray, volumes: ndarray, poc_price: int) → float[source]

Calculate the percentage of volume above the Point of Control (POC) price level.

Parameters:

  • price_levels – Array of price levels.

  • volumes – Corresponding volumes.

  • poc_price – The Point of Control price level.

Returns:

Percentage of volume above POC (0-1 range).

finmlkit.feature.core.volume.comp_flow_acceleration(volumes: ndarray[tuple[int, ...], dtype[float64]], window: int, recent_periods: int) → ndarray[tuple[int, ...], dtype[float64]][source]

Calculate flow acceleration using Numba

Parameters:

  • volumes – volumes

  • window – window size (eg. 20)

  • recent_periods – Most recent periods to consider for acceleration calculation

Returns:

finmlkit.feature.core.volume.comp_poc_hva_lva(price_levels: ndarray, volumes: ndarray, va_pct=68.34) → tuple[int, int, int][source]

Calculate the POC (Point of Control), HVA (High Value Area), and LVA (Low Value Area) for a given volume profile.

Parameters:

  • price_levels – Price levels (sorted ascending).

  • volumes – Corresponding volumes.

  • va_pct – Value area percentage (default 68.34).

Returns:

Tuple of: - poc_price: Price with the highest volume. - hva_price: High value area bound. - lva_price: Low value area bound.

Raises:

AssertionError – If inputs are empty or mismatched in length.

finmlkit.feature.core.volume.trim_trailing_zeros(price_levels: ndarray, volumes: ndarray) → tuple[ndarray, ndarray][source]

Trim trailing zero volumes from a price-level volume profile.

Parameters:

  • price_levels – Array of price levels.

  • volumes – Corresponding volumes.

Returns:

Trimmed price levels and volumes.

finmlkit.feature.core.volume.volume_profile_developing(ts: ndarray, highs: ndarray, lows: ndarray, price_levels: list[ndarray], buy_volumes: list[ndarray], sell_volumes: list[ndarray], start_ts: int, end_ts: int, n_bins: int = None, price_tick: float = None, va_pct: float = 68.34) → tuple[ndarray, ndarray, ndarray, ndarray][source]

Compute a developing volume profile between two timestamps using cumulative aggregation.

Parameters:

  • ts – Nanosecond timestamps per bar.

  • highs – High prices per bar.

  • lows – Low prices per bar.

  • price_levels – List of price levels per bar.

  • buy_volumes – List of buy volumes per bar.

  • sell_volumes – List of sell volumes per bar.

  • start_ts – Start time in nanoseconds.

  • end_ts – End time in nanoseconds.

  • n_bins – Optional number of bins for bucketing price levels.

  • price_tick – Tick size for price bucketing.

  • va_pct – Value area percentage.

Returns:

Tuple of timestamps, POC, HVA, and LVA series for the range.

Raises:

AssertionError – If input arrays are empty or misaligned in length.

finmlkit.feature.core.volume.volume_profile_rolling(ts: ndarray, highs: ndarray, lows: ndarray, price_levels: list[ndarray], buy_volumes: list[ndarray], sell_volumes: list[ndarray], window_size_sec: float, n_bins: int = None, price_tick: float = None, va_pct: float = 68.34) → tuple[ndarray, ndarray, ndarray, ndarray][source]

Compute rolling volume profiles over a fixed-width time window.

Parameters:

  • ts – Nanosecond timestamps per bar.

  • highs – High prices per bar.

  • lows – Low prices per bar.

  • price_levels – List of price levels per bar.

  • buy_volumes – List of buy volumes per bar.

  • sell_volumes – List of sell volumes per bar.

  • window_size_sec – Width of the rolling time window.

  • n_bins – Optional number of bins for bucketing price levels.

  • price_tick – Price tick size for discretization.

  • va_pct – Value area percentage.

Returns:

Tuple of POC, HVA, LVA, and vp_pct_abv_poc price series aligned to input bars.

Raises:

AssertionError – If input arrays are empty or misaligned in length.

finmlkit.feature.core.volume.vpin(volume_buy, volume_sell, window)[source]