pybroker.scope module

Contains scopes that store data and object references used to execute a pybroker.strategy.Strategy.

class ColumnScope(store: SymbolArrayStore | DataFrame)

Bases: object

Caches and retrieves column data from a SymbolArrayStore.

Parameters:

store – Pre-built numpy column store, or a MultiIndex pandas.DataFrame (legacy convenience).

bar_data_from_data_columns(symbol: str, end_index: int) → BarData

Returns a new pybroker.common.BarData instance containing column data of default and custom data columns registered with StaticScope.

Parameters:

• symbol – Ticker symbol to query.

• end_index – Truncates column values (exclusive). If None, then column values are not truncated.

fetch(symbol: str, name: str, end_index: int | None = None) → ndarray[tuple[Any, ...], dtype[_ScalarT]] | None

Fetches a numpy.ndarray of column data for symbol.

Parameters:

• symbol – Ticker symbol to query.

• name – Name of column to query.

• end_index – Truncates column values (exclusive). If None, then column values are not truncated.

Returns:

numpy.ndarray of column data for every bar until end_index (when specified).

fetch_dict(symbol: str, names: Iterable[str], end_index: int | None = None) → dict[str, ndarray[tuple[Any, ...], dtype[_ScalarT]] | None]

Fetches a dict of column data for symbol.

Parameters:

• symbol – Ticker symbol to query.

• names – Names of columns to query.

• end_index – Truncates column values (exclusive). If None, then column values are not truncated.

Returns:

dict mapping column names to numpy.ndarrays of column values.

fetch_value(symbol: str, name: str, end_index: int) → float | None

Returns the scalar value at end_index - 1 without slicing.

property store: SymbolArrayStore

property symbols: frozenset[str]

Symbols held by the underlying store.

unique_dates() → ndarray[tuple[Any, ...], dtype[datetime64]]

Returns sorted unique dates across every symbol in the store.

class IndicatorScope(indicator_data: Mapping[IndicatorSymbol, Series], filter_dates: Sequence[datetime64])

Bases: object

Caches and retrieves pybroker.indicator.Indicator data.

Parameters:

• indicator_data – Mapping of pybroker.common.IndicatorSymbol pairs to pandas.Series of pybroker.indicator.Indicator values.

• filter_dates – Filters pybroker.indicator.Indicator data on Sequence of dates.

fetch(symbol: str, name: str, end_index: int | None = None) → ndarray[tuple[Any, ...], dtype[float64]]

Fetches pybroker.indicator.Indicator data.

Parameters:

• symbol – Ticker symbol to query.

• name – Name of pybroker.indicator.Indicator to query.

• end_index – Truncates the array of pybroker.indicator.Indicator data returned (exclusive). If None, then indicator data is not truncated.

Returns:

numpy.ndarray of pybroker.indicator.Indicator data for every bar until end_index (when specified).

fetch_full(symbol: str, name: str) → ndarray[tuple[Any, ...], dtype[float64]]

Fetches the full indicator array without truncation.

fetch_history(symbol: str, name: str, dates: ndarray[tuple[Any, ...], dtype[Any]]) → ndarray[tuple[Any, ...], dtype[float64]] | None

Aligns full-history indicator values to dates.

fetch() masks base timeframe indicators to filter_dates, so it cannot serve data from before the current window. Lag features need history that reaches back into the train window, which this reads from the unfiltered series.

Returns:

numpy.ndarray of values aligned to dates, or None when the indicator is not registered for symbol.

fetch_value(symbol: str, name: str, end_index: int) → float

Returns the scalar value at end_index - 1 without slicing.

has_indicator(symbol: str, name: str) → bool

Whether pybroker.indicator.Indicator data is registered for symbol.

class IntervalScope(interval_data: IntervalData, ind_scope: IndicatorScope, models: Mapping[ModelSymbol, TrainedModel] | None = None, test_dates: Sequence[datetime64] | None = None)

Bases: object

Serves compressed bar and indicator data through alignment maps.

clear_cache()

Drops every cached array.

Compressed data is immutable for the lifetime of a scope (a new one is built per walkforward window), and each cache is keyed independently of the current bar, so this is only for tearing a scope down – calling it per bar would rebuild model input and rerun predict on every bar.

completed_index(symbol: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, end_index: int) → int

fetch_bar(symbol: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, col: str, end_index: int) → ndarray[tuple[Any, ...], dtype[Any]]

fetch_indicator(symbol: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, base_name: str, end_index: int) → ndarray[tuple[Any, ...], dtype[float64]]

fetch_input(symbol: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, base_model_name: str, end_index: int) → DataFrame

fetch_preds(symbol: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str, base_model_name: str, end_index: int) → ndarray[tuple[Any, ...], dtype[_ScalarT]]

window_len(symbol: str, interval: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str) → int

Returns the compressed bar count visible in the current window.

completed is realigned to the walkforward test window by pybroker.interval.IntervalData.slice_for_test(), so its last entry is the newest compressed bar that completes within the window. Model input and predictions are capped here so user callbacks never see compressed bars belonging to a future window.

class ModelInputScope(col_scope: ColumnScope, ind_scope: IndicatorScope, models: Mapping[ModelSymbol, TrainedModel], history_col_scope: ColumnScope | None = None, test_dates: Sequence[datetime64] | None = None)

Bases: object

Caches and retrieves model input data.

Parameters:

• col_scope – ColumnScope.

• ind_scope – IndicatorScope.

• models – Mapping of pybroker.common.ModelSymbol pairs to pybroker.common.TrainedModels.

fetch(symbol: str, name: str, end_index: int | None = None) → DataFrame

Fetches model input data.

Parameters:

• symbol – Ticker symbol to query.

• name – Name of pybroker.model.ModelSource to query input data.

• end_index – Truncates the array of model input data returned (exclusive). If None, then model input data is not truncated.

Returns:

pandas.DataFrame of model input data for every bar until end_index (when specified).

fetch_model_input(symbol: str, name: str, end_index: int | None = None) → ModelInput

Fetches model input as internal pybroker.model.ModelInput (no DataFrame).

Parameters:

• symbol – Ticker symbol to query.

• name – Name of pybroker.model.ModelSource to query input data.

• end_index – Truncates the array of model input data returned (exclusive). If None, then model input data is not truncated.

Returns:

pybroker.model.ModelInput for every bar until end_index (when specified).

class PendingOrder(id: int, type: Literal['buy', 'sell'], symbol: str, created: np.datetime64, exec_date: np.datetime64, shares: Decimal, limit_price: Decimal | None, fill_price: int | float | np.floating | Decimal | PriceType | Callable[[str, BarData], int | float | Decimal], exec_bar: int, timeout_bars: int | None, stops: frozenset['Stop'] | None, exit_pos_type: Literal['long', 'short'] | None = None)

Bases: NamedTuple

Holds data for a pending order.

id

Unique ID.

Type:

int

type

Type of order, either buy or sell.

Type:

Literal[‘buy’, ‘sell’]

symbol

Ticker symbol of the order.

Type:

str

created

Date the order was created.

Type:

np.datetime64

exec_date

Date the order will be executed.

Type:

np.datetime64

shares

Number of shares to be bought or sold.

Type:

Decimal

limit_price

Limit price to use for the order.

Type:

Optional[Decimal]

fill_price

Price that the order will be filled at.

Type:

Union[int, float, np.floating, Decimal, PriceType, Callable[[str, BarData], Union[int, float, Decimal]]]

exec_bar

Symbol bar index when the order will first be attempted.

Type:

int

timeout_bars

Number of bars to retry after the first attempt. None for a single attempt, -1 for indefinite persistence, or a positive integer for a limited number of retry bars.

Type:

Optional[int]

stops

Stops to attach when the order is filled.

Type:

Optional[frozenset[‘Stop’]]

exit_pos_type

Type of the pybroker.portfolio.Position this order exits, either long or short, or None when the order is not an exit. An exit order is clamped at fill time to the shares still held, so it can only close a position, never flip one to the opposite side.

Type:

Optional[Literal[‘long’, ‘short’]]

class PendingOrderScope

Bases: object

Stores PendingOrders

add(type: Literal['buy', 'sell'], symbol: str, created: np.datetime64, exec_date: np.datetime64, shares: Decimal, limit_price: Decimal | None, fill_price: int | float | np.floating | Decimal | PriceType | Callable[[str, BarData], int | float | Decimal], exec_bar: int, timeout_bars: int | None, stops: frozenset['Stop'] | None = None, exit_pos_type: Literal['long', 'short'] | None = None) → int

Creates a PendingOrder.

Parameters:

• type – Type of order, either buy or sell.

• symbol – Ticker symbol of the order.

• created – Date the order was created.

• exec_date – Date the order will be executed.

• shares – Number of shares to be bought or sold.

• limit_price – Limit price to use for the order.

• fill_price – Price that the order will be filled at.

• exec_bar – Symbol bar index when the order will first be attempted.

• timeout_bars – Number of bars to retry after the first attempt.

• stops – Stops to attach when the order is filled.

• exit_pos_type – Type of the position this order exits, or None when the order is not an exit.

Returns:

ID of the PendingOrder.

advance_retry_bars(order_id: int) → None

Records that order_id was attempted on a bar.

contains(order_id: int) → bool

Returns whether a PendingOrder exists with order_id.

get(order_id: int) → PendingOrder | None

Returns a PendingOrder with order_id.

has_orders() → bool

Returns whether any pending orders exist.

mark_attempted(order_id: int) → None

Records that order_id has had its first fill attempt.

orders(symbol: str | None = None, order_id: int | None = None) → Iterable[PendingOrder]

Returns an Iterable of PendingOrders.

Parameters:

• symbol – Filter by ticker symbol.

• order_id – Filter by order ID.

remove(order_id: int) → bool

Removes a PendingOrder with order_id`.

remove_all(symbol: str | None = None)

Removes all PendingOrders.

retry_bars(order_id: int) → int

Returns how many bars order_id has been retried for.

0 on the bar of its first attempt.

was_attempted(order_id: int) → bool

Returns whether order_id has had its first fill attempt.

class PredictionScope(models: Mapping[ModelSymbol, TrainedModel], input_scope: ModelInputScope)

Bases: object

Caches and retrieves model predictions.

Parameters:

• models – Mapping of pybroker.common.ModelSymbol pairs to pybroker.common.TrainedModels.

• input_scope – ModelInputScope.

fetch(symbol: str, name: str, end_index: int | None = None) → ndarray[tuple[Any, ...], dtype[_ScalarT]]

Fetches model predictions.

Parameters:

• symbol – Ticker symbol to query.

• name – Name of pybroker.model.ModelSource that made the predictions.

• end_index – Truncates the array of predictions returned (exclusive). If None, then predictions are not truncated.

Returns:

numpy.ndarray of model predictions for every bar until end_index (when specified).

class PriceScope(col_scope: ColumnScope, sym_end_index: Mapping[str, int], round_fill_price: bool)

Bases: object

Retrieves most recent prices.

fetch(symbol: str, price: int | float | floating | Decimal | PriceType | Callable[[str, BarData], int | float | Decimal]) → Decimal

fetch_bar_ohlc(symbol: str, date: datetime64) → tuple[float | None, float | None, float | None]

Returns (close, low, high) for symbol on date, or Nones.

Memoized per bar: both check_stops loops and capture_bar read this for every symbol on every bar, and each miss re-fetches the column dict. Keyed by date as well as symbol, like has_bar_on(), so a stale entry cannot answer for a later bar.

fetch_float(symbol: str, price: int | float | floating | Decimal | PriceType | Callable[[str, BarData], int | float | Decimal]) → float

Returns a bar price as float using the per-bar cache when possible.

has_bar(symbol: str) → bool

Returns whether symbol has a bar that can be priced.

False for a symbol absent from the current test window – one that stopped trading, or that a pybroker.common.SymbolSelector dropped – whose prices would otherwise raise.

has_bar_on(symbol: str, date: datetime64) → bool

Returns whether symbol’s current bar falls on date.

Stricter than has_bar(), which only reports that the symbol has traded at some point. When calendars are ragged, a symbol’s index is not advanced on a date it has no bar, so its “current” bar is an earlier one and pricing against it would use a stale price.

Memoized per bar: check_stops calls this once per symbol holding a stop on every bar, and each miss fetches the symbol’s whole date array. Keyed by date as well as symbol so a caller that does not call reset_bar() still reads a correct answer.

reset_bar() → None

Clears the per-bar OHLC cache. Call once at the start of each bar.

class StaticScope

Bases: object

A static registry of data and object references.

logger

pybroker.log.Logger

data_source_cache

diskcache.Cache that stores data retrieved from pybroker.data.DataSource.

data_source_cache_ns

Namespace set for data_source_cache.

indicator_cache

diskcache.Cache that stores pybroker.indicator.Indicator data.

indicator_cache_ns

Namespace set for indicator_cache.

model_cache

diskcache.Cache that stores trained models.

model_cache_ns

Namespace set for model_cache.

default_data_cols

Default data columns in pandas.DataFrame retrieved from a pybroker.data.DataSource.

custom_data_cols

User-defined data columns in pandas.DataFrame retrieved from a pybroker.data.DataSource.

property all_data_cols: frozenset[str]

All registered data column names. Unordered; use ordered_data_cols when iteration order is significant.

clear_params()

Clears all global parameters.

freeze_data_cols()

Prevents additional data columns from being registered.

get_hyperparam(name: str) → Any

Retrieves a hyperparam from static scope.

get_indicator(name: str)

Retrieves a pybroker.indicator.Indicator from static scope.

get_indicator_names(model_name: str) → tuple[str]

Returns a tuple[str] of all pybroker.indicator.Indicator names that are registered with pybroker.model.ModelSource having model_name.

get_model_source(name: str)

Retrieves a pybroker.model.ModelSource from static scope.

has_hyperparam(name: str) → bool

Whether a hyperparam is stored in static scope.

has_indicator(name: str) → bool

Whether pybroker.indicator.Indicator is stored in static scope.

has_model_source(name: str) → bool

Whether pybroker.model.ModelSource is stored in static scope.

classmethod instance() → StaticScope

Returns singleton instance.

iter_hyperparams() → Iterable[Any]

Iterates registered hyperparams.

property ordered_data_cols: tuple[str, ...]

All registered data column names in deterministic order. Iterating all_data_cols instead yields a process-dependent order, which makes column-order sensitive output such as model input data irreproducible across runs.

param(name: str, value: Any | None = <object object>) → Any | None

Get or set a global parameter.

register_custom_cols(names: str | Iterable[str], *args)

Registers user-defined column names.

set_hyperparam(hyperparam: Any) → None

Stores a pybroker.optimize.Hyperparam in static scope.

set_indicator(indicator)

Stores pybroker.indicator.Indicator in static scope.

classmethod set_instance(scope: StaticScope | None) → None

Replaces the singleton instance, or clears it when scope is None.

Used to install a scope that was pickled from another process, so that worker tasks see the caller’s registered indicators, model sources and params instead of an empty scope. Replacing wholesale (rather than merging) also keeps stale registrations from surviving in a worker that is reused across runs.

set_model_source(source)

Stores pybroker.model.ModelSource in static scope.

unfreeze_data_cols()

Allows additional data columns to be registered if pybroker.scope.StaticScope.freeze_data_cols() was called.

unregister_custom_cols(names: str | Iterable[str], *args)

Unregisters user-defined column names.

validate_registered_names(indicators: Iterable[str] | None = None, models: Iterable[str] | None = None)

Raises when an indicator used by a run or one of its models’ prediction columns shares a name with a data column or another registered source.

A colliding name is resolved differently by different consumers: model training reads the data column while prediction reads the indicator, and signals output silently overwrites one value with the other – so the collision is rejected outright.

Parameters:

• indicators – Indicator names the run uses. Defaults to every registered indicator.

• models – Model names the run uses. Defaults to every registered model.

class SymbolArrayStore(symbols: frozenset[str], sym_arrays: Mapping[str, Mapping[str, ndarray[tuple[Any, ...], dtype[_ScalarT]]]], backing: _StoreBacking | None = None)

Bases: object

Internal numpy-backed OHLCV/custom columns keyed by symbol.

backing: _StoreBacking | None = None

sym_arrays: Mapping[str, Mapping[str, ndarray[tuple[Any, ...], dtype[_ScalarT]]]]

symbols: frozenset[str]

unique_dates() → ndarray[tuple[Any, ...], dtype[datetime64]]

Returns sorted unique dates across every symbol.

clear_params()

Clears all global parameters.

column_scope_from_frame(df: DataFrame, sym_col: str = 'symbol', date_col: str = 'date') → ColumnScope

Creates a ColumnScope with upfront numpy extraction.

disable_logging()

Disables event logging.

disable_progress_bar()

Disables logging a progress bar.

enable_logging()

Enables event logging.

enable_progress_bar()

Enables logging a progress bar.

get_signals(symbols: Iterable[str], col_scope: ColumnScope, ind_scope: IndicatorScope, pred_scope: PredictionScope) → dict[str, DataFrame]

Retrieves dictionary of pandas.DataFrames containing bar data, indicator data, and model predictions for each symbol.

merge_symbol_array_stores(left: SymbolArrayStore, right: SymbolArrayStore) → SymbolArrayStore

Concatenates per-symbol column arrays from two stores.

param(name: str, value: Any | None = <object object>) → Any | None

Get or set a global parameter.

register_columns(names: str | Iterable[str], *args)

Registers names of user-defined data columns.

run_with_scope(scope: StaticScope, fn: Callable[[...], Any], *args: Any) → Any

Installs scope as this process’ scope, then runs fn.

StaticScope is a per-process singleton, so a worker process starts with an empty one and would not see the caller’s registered indicators, model sources, params or custom columns. Wrap work dispatched to pybroker.parallel.parallel() in this to ship the caller’s scope along with it. Running sequentially, scope is already the installed instance and this is a no-op.

slice_symbol_array_store_by_dates(store: SymbolArrayStore, selected_dates: Sequence[datetime64] | ndarray[tuple[Any, ...], dtype[datetime64]]) → SymbolArrayStore

Filters a store to rows whose dates are in selected_dates.

sym_data_from_store(store: SymbolArrayStore, data_cols: Iterable[str]) → dict[str, dict[str, ndarray[tuple[Any, ...], dtype[_ScalarT]] | None]]

Converts a SymbolArrayStore to per-symbol column arrays.

sym_exec_dates_from_store(store: SymbolArrayStore) → dict[str, frozenset[datetime64]]

Returns per-symbol test dates from a column store.

Symbols are walked in sorted order. SymbolArrayStore.symbols is a frozenset[str], so iterating it directly would seed this mapping in string-hash order, and that order decides which symbol is served first on each bar when calendars are ragged – making a capital-constrained backtest depend on PYTHONHASHSEED. Sorting also matches the sorted(test_syms) order the aligned-calendar path already uses.

symbol_array_store_from_flat_frame(df: DataFrame, sym_col: str = 'symbol', date_col: str = 'date', symbols: frozenset[str] | None = None) → SymbolArrayStore

Builds a store from a flat frame via numpy lex-sort and bin slicing.

symbol_array_store_from_frame(df: DataFrame, sym_col: str = 'symbol', date_col: str = 'date', symbols: frozenset[str] | None = None) → SymbolArrayStore

Builds a store from a flat or MultiIndex OHLCV frame.

symbol_array_store_from_indexed_df(df: DataFrame) → SymbolArrayStore

Builds a SymbolArrayStore from a sorted MultiIndex frame.

unregister_columns(names: str | Iterable[str], *args)

Unregisters names of user-defined data columns.