pybroker.interval module

Multi-interval bar compression utilities.

This code is licensed under Apache 2.0 with Commons Clause license (see LICENSE for details).

BASE_INTERVAL: Final = 'base': Sentinel accepted by pybroker.indicator.Indicator.intervals() and pybroker.model.ModelSource.intervals() to request the base-timeframe variant in addition to the listed compression intervals.

class CompressedBars(open: ~numpy._typing._array_like.NDArray[~numpy.float64], high: ~numpy._typing._array_like.NDArray[~numpy.float64], low: ~numpy._typing._array_like.NDArray[~numpy.float64], close: ~numpy._typing._array_like.NDArray[~numpy.float64], volume: ~numpy._typing._array_like.NDArray[~numpy.float64], dates: ~numpy._typing._array_like.NDArray[~numpy.datetime64], custom: ~typing.Mapping[str, ~numpy._typing._array_like.NDArray[~numpy.float64]] = <factory>, vwap: ~numpy._typing._array_like.NDArray[~numpy.float64] | None = None)[source]

Bases: object

OHLCV and custom columns aggregated into compressed bars.

close: NDArray[float64]

custom: Mapping[str, NDArray[float64]]

dates: NDArray[datetime64]

high: NDArray[float64]

low: NDArray[float64]

open: NDArray[float64]

slice_by_dates(dates: Iterable[datetime64]) → CompressedBars[source]: Returns compressed bars restricted to dates.

volume: NDArray[float64]

vwap: NDArray[float64] | None = None

class CompressedSymbolData(bars: CompressedBars, completed: NDArray[int64], base_dates: NDArray[datetime64])[source]

Bases: object

Compressed bar data and alignment map for one symbol.

bars: CompressedBars

base_dates: NDArray[datetime64]

completed: NDArray[int64]

INTERVAL_NAME_SEPARATOR = '@': Separator reserved for interval bindings in indicator and model names.

class IntervalData(compressed: dict[tuple[str, int | ~typing.Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str], ~pybroker.interval.CompressedSymbolData] = <factory>)[source]

Bases: object

Compressed data keyed by (symbol, interval).

compressed: dict[tuple[str, int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str], CompressedSymbolData]

slice_for_test(test_symbol_dates: Mapping[str, NDArray[datetime64]]) → IntervalData[source]: Returns a copy with completed arrays aligned to test dates.

TimeframeInterval

Compression interval for multi-interval data.

int (n > 1): every n base bars (e.g. 5).
str duration: digits plus one unit letter — "5m", "1h", "30s", or "1d" (letters: s, m, h, d).
str calendar: "daily", "weekly", "monthly", "quarterly", or "yearly", aligned to calendar boundaries: weeks start on Monday, months on the first of the month, quarters in January, April, July, and October, and years on January 1.

alias of int | Literal[‘daily’, ‘weekly’, ‘monthly’, ‘quarterly’, ‘yearly’] | str

base_timeframe_to_seconds(base_timeframe: str) → float[source]: Converts a base timeframe string to seconds.

build_compressed_symbol_arrays(symbol: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, compressed: CompressedSymbolData, indicator_data: Mapping[IndicatorSymbol, Series], indicator_names: Iterable[str], custom_cols: Iterable[str]) → tuple[tuple[str, ...], dict[str, NDArray], NDArray[datetime64]][source]: Builds compressed-bar column arrays with base indicator names.

build_compressed_symbol_df(symbol: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, compressed: CompressedSymbolData, indicator_data: Mapping[IndicatorSymbol, Series], indicator_names: Iterable[str], custom_cols: Iterable[str]) → DataFrame[source]

Builds a compressed-bar DataFrame with base indicator column names.

Not used on the backtest hot path; prefer build_compressed_symbol_arrays().

compress(dates: NDArray[datetime64], open_: NDArray[float64], high: NDArray[float64], low: NDArray[float64], close: NDArray[float64], volume: NDArray[float64], interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, custom_cols: Mapping[str, NDArray[float64]] | None = None, vwap: NDArray[float64] | None = None) → tuple[CompressedBars, NDArray[int64]][source]

Compresses base bars into coarser interval bars.

Returns compressed bars and a completed alignment map where completed[t] is the index of the last completed compressed bar at base bar t, or -1 during warmup.

compress_bars(data: BarData | DataFrame, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, *, base_timeframe: str) → BarData[source]

Compresses base OHLCV bars to a coarser interval.

Parameters:

data – Single-symbol BarData or OHLCV pandas.DataFrame.
interval – Target compression interval.
base_timeframe – Declared base bar spacing (e.g. "1m", "1d").

Returns:

Compressed BarData.

compress_intervals_from_frame(df: DataFrame, symbol_intervals: Mapping[str, Iterable[int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str]], custom_cols: Iterable[str], base_bar_seconds: float) → IntervalData[source]

Compresses each symbol to the intervals declared for it.

Parameters:

df – Multi-symbol OHLCV frame.
symbol_intervals – Maps each symbol to the intervals it is compressed to. Symbols absent from the mapping are skipped, so a strategy only pays for the (symbol, interval) pairs its executions declare rather than the full symbol x interval cross product.
custom_cols – Custom data columns carried onto compressed bars.
base_bar_seconds – Bar spacing of the base feed, in seconds.

compress_symbol_df(sym_df: DataFrame, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, custom_cols: Iterable[str], base_bar_seconds: float, *, validate_dates: bool = True) → CompressedSymbolData[source]: Compresses a single-symbol DataFrame.

compress_symbol_from_frame(df: DataFrame, symbol: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, custom_cols: Iterable[str], base_bar_seconds: float, *, validate_dates: bool = True) → CompressedSymbolData[source]: Compresses one symbol from a multi-symbol frame without copying rows.

compress_symbol_intervals_from_frame(df: DataFrame, symbol: str, intervals: Iterable[int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str], custom_cols: Iterable[str], base_bar_seconds: float, *, validate_dates: bool = True, rows: NDArray[int64] | None = None) → dict[int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, CompressedSymbolData][source]

Compresses one symbol to multiple intervals with a single OHLCV extract.

rows optionally supplies this symbol’s precomputed row indices, so a caller compressing many symbols groups the frame once instead of scanning the symbol column per symbol.

compressed_bars_to_bar_data(bars: CompressedBars) → BarData[source]: Converts compressed OHLCV arrays to BarData.

format_interval(interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str) → str[source]: Returns a stable string representation of interval.

indicator_interval_name(base: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str) → str[source]: Returns the suffixed indicator name for an interval binding.

is_valid_interval(interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, base_bar_seconds: float) → bool[source]: Returns whether interval is valid for the base feed bar spacing.

lookahead_train_dates(bar_dates: NDArray[datetime64], train_dates: Iterable[datetime64], test_dates: Iterable[datetime64], lookahead: int) → tuple[NDArray[datetime64], int][source]

Trims compressed train bar dates so the train/test hold-out is lookahead compressed bars wide.

The walkforward split holds out lookahead bars of the base timeframe, but a model bound to an interval is fitted on compressed bars, so the hold-out must be re-measured in compressed-bar units: every kept train bar satisfies compressed_index <= first_test_compressed_index - lookahead.

Parameters:

bar_dates – Dates of the full compressed bar history for one symbol.
train_dates – Base-timeframe train window dates; compressed bars are selected by membership of their closing date.
test_dates – Base-timeframe test window dates.
lookahead – Number of compressed bars to hold out.

Returns:

(dates_to_select, n_dropped) — the train bar dates to keep and how many train compressed bars were dropped. With lookahead <= 1 the requested train dates are returned unchanged, which matches the one-bar gap that date membership already produces.

model_interval_name(base: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str) → str[source]: Returns the suffixed model name for an interval binding.

normalize_interval(interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str) → int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str[source]: Normalizes and validates a compression interval.

Normalizes one or more compression intervals into a frozenset, rejecting empty input and duplicates.

Parameters:

intervals – A single TimeframeInterval or an Iterable of them.
param – Parameter name used in error messages.
allow_base – If True, the literal 'base' passes through verbatim. Otherwise it is rejected like any other invalid interval.

parse_indicator_interval_name(name: str) → tuple[str, int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str | None][source]: Parses a suffixed indicator name into base name and interval.

parse_model_interval_name(name: str) → tuple[str, int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str | None][source]: Parses a suffixed model name into base name and interval.

slice_arrays_by_dates(columns: tuple[str, ...], arrays: Mapping[str, NDArray], dates: NDArray[datetime64], selected: Iterable[datetime64]) → tuple[tuple[str, ...], dict[str, NDArray], NDArray[datetime64]][source]: Filters column arrays to rows whose dates are in selected.

slice_compressed_df_by_dates(df: DataFrame, dates: Iterable[datetime64]) → DataFrame[source]

Filters a compressed DataFrame to rows whose dates are in dates.

Not used on the backtest hot path; prefer slice_arrays_by_dates().

symbol_dates_from_frame(df: DataFrame) → dict[str, NDArray[datetime64]][source]: Extracts per-symbol test dates from a multi-symbol frame.

validate_base_timeframe_data(df: DataFrame, base_bar_seconds: float) → None[source]: Raises if bar timestamps are inconsistent with base_bar_seconds.

validate_interval(interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, base_bar_seconds: float) → None[source]: Validates an interval against the base feed bar spacing.

validate_source_name(name: str, kind: str) → None[source]

Raises if name cannot be used as an indicator or model name.

Parameters:

name – Name being registered.
kind – 'indicator' or 'model', used in the error message.