pybroker.scope module

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

class ColumnScope(df: DataFrame)

Bases: object

Caches and retrieves column data queried from pandas.DataFrame.

Parameters:

df – pandas.DataFrame containing the column data.

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[Any, dtype[_ScalarType_co]] | 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[Any, dtype[_ScalarType_co]] | 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.

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[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).

class ModelInputScope(col_scope: ColumnScope, ind_scope: IndicatorScope, models: Mapping[ModelSymbol, TrainedModel])

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:

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

class PendingOrder(id: int, type: Literal['buy', 'sell'], symbol: str, created: datetime64, exec_date: datetime64, shares: Decimal, limit_price: Decimal | None, fill_price: int | float | floating | Decimal | PriceType | Callable[[str, BarData], int | float | Decimal])

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:

numpy.datetime64

exec_date

Date the order will be executed.

Type:

numpy.datetime64

shares

Number of shares to be bought or sold.

Type:

decimal.Decimal

limit_price

Limit price to use for the order.

Type:

decimal.Decimal | None

fill_price

Price that the order will be filled at.

Type:

int | float | numpy.floating | decimal.Decimal | pybroker.common.PriceType | Callable[[str, pybroker.common.BarData], int | float | decimal.Decimal]

class PendingOrderScope

Bases: object

Stores PendingOrders

add(type: Literal['buy', 'sell'], symbol: str, created: datetime64, exec_date: datetime64, shares: Decimal, limit_price: Decimal | None, fill_price: int | float | floating | Decimal | PriceType | Callable[[str, BarData], int | float | Decimal]) → 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.

Returns:

ID of the PendingOrder.

contains(order_id: int) → bool

Returns whether a PendingOrder exists with order_id.

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

Returns an Iterable of PendingOrders.

remove(order_id: int) → bool

Removes a PendingOrder with order_id`.

remove_all(symbol: str | None = None)

Removes all PendingOrders.

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[Any, dtype[_ScalarType_co]]

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])

Bases: object

Retrieves most recent prices.

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

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.

freeze_data_cols()

Prevents additional data columns from being registered.

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_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.

param(name: str, value: ~typing.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_indicator(indicator)

Stores pybroker.indicator.Indicator in static scope.

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.

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.

param(name: str, value: ~typing.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.

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

Unregisters names of user-defined data columns.