pybroker.model module

Contains model related functionality.

class CachedModel(model: Any, input_cols: tuple[str] | None, lag_columns: tuple[str, ...] | None = None)[source]

Bases: NamedTuple

Stores cached model data.

input_cols

Names of the columns to be used as input for the model when making predictions.

Type:: tuple[str] | None

lag_columns

Names of the columns that lag features were built from at training time, in feature block order. None when the model was not trained with lags, and when loading a model cached before this field existed.

Type:: tuple[str, …] | None

model: Any: Trained model instance.

class IntervalBoundModel(source: ModelSource, intervals: frozenset[TimeframeInterval])[source]

Bases: NamedTuple

A ModelSource bound to one or more compression intervals, returned by ModelSource.intervals() and passed to the models parameter of pybroker.strategy.Strategy.add_execution().

intervals: frozenset[int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str]: Normalized TimeframeIntervals the model is trained on. May include the literal 'base' for the base timeframe.

source: ModelSource: The bound ModelSource.

class LagSeriesKey(symbol: str, column: str, lag: int, interval: str | None = None)[source]

Bases: object

Internal cache key for a full-history lagged series.

column: str

interval: str | None = None

lag: int

symbol: str

class ModelInput(columns: tuple[str, ...], arrays: dict[str, ndarray], dates: ndarray, lag_features: ndarray | None = None, lags: int | None = None, lag_columns: tuple[str, ...] | None = None)[source]

Bases: object

Internal numpy-backed model input with optional lag feature metadata.

Not part of the public API. User-facing code receives pandas.DataFrame instances materialized via to_dataframe(), with any lag feature matrix passed explicitly as a separate numpy.ndarray argument.

arrays: dict[str, ndarray]

columns: tuple[str, ...]

dates: ndarray

drop_lag_warmup() → ModelInput[source]

Drops the leading rows whose lag features are not yet defined.

Only each symbol’s warmup region is trimmed. Dropping every row with a NaN lag feature would also remove rows from the middle of the series whenever a lag column is sparse – an event column registered with pybroker.scope.register_columns() is NaN except on the bars it fires, and an all-NaN volume column is routine for index, FX and CFD series. Either silently shrinks the training set, or empties it.

Pooled input stacks one block per symbol, so each block carries its own warmup at its own offset. Trimming only the front of the matrix would leave every block after the first with its NaN rows intact, which sklearn.fit rejects outright.

empty() → bool[source]

lag_features: ndarray | None = None

lag_warmup_len() → int[source]

Returns how many leading rows have undefined lag features.

These rows cannot be handed to an estimator: sklearn rejects NaN outright, and a NaN-tolerant predict_fn quietly produces a prediction from features that do not exist yet.

lags: int | None = None

select_columns(columns: tuple[str, ...]) → ModelInput[source]: Returns a view restricted to columns.

slice(end_index: int | None = None) → ModelInput[source]: Returns a row slice sharing backing array memory.

slice_range(start: int, end_index: int | None = None) → ModelInput[source]: Returns a row range sharing backing array memory.

to_dataframe() → DataFrame[source]: Materializes a DataFrame of the input columns.

class ModelLoader(name: str, load_fn: Callable[[...], Any | tuple[Any, Iterable[str]]], indicator_names: Iterable[str], input_data_fn: Callable[[DataFrame], DataFrame] | None, predict_fn: Callable[[Any, DataFrame | NDArray], NDArray] | None, pooled: bool, kwargs: dict[str, Any], lags: int | None = None, lag_cols: tuple[str, ...] = (), per_bar: bool = False)[source]

Bases: ModelSource

Loads a pre-trained model.

Parameters:

name – Name of model.
load_fn – Callable[[symbol: str, train_start_date: datetime, train_end_date: datetime, ...], DataFrame] used to load and return a pre-trained model. This is expected to return either a trained model instance, or a tuple containing a trained model instance and a Iterable of column names to to be used as input for the model when making predictions.
indicator_names – Iterable of names of pybroker.indicator.Indicators used as features of the model.
input_data_fn – Callable[[DataFrame], DataFrame] for preprocessing input data passed to the model when making predictions. If set, input_data_fn will be called with a pandas.DataFrame containing all test data.
predict_fn – Callable[[Model, DataFrame], ndarray] that overrides calling the model’s default predict function. If set, predict_fn will be called with the trained model and a pandas.DataFrame containing all test data. When lags is set, it is instead called with the lag feature matrix (numpy.ndarray) in place of the DataFrame.
pooled – If True, the model is trained once per execution using combined multi-symbol data. Defaults to False.
kwargs – dict of kwargs to pass to load_fn.

__call__(symbol: str, train_start_date: datetime, train_end_date: datetime) → Any | tuple[Any, Iterable[str]][source]

Loads pre-trained model.

Parameters:

symbol – Ticker symbol for loading the pre-trained model.
train_start_date – Start date of training window.
train_end_date – End date of training window.

Returns:

Pre-trained model.

class ModelSource(name: str, indicator_names: Iterable[str], input_data_fn: Callable[[DataFrame], DataFrame] | None, predict_fn: Callable[[Any, DataFrame | NDArray], NDArray] | None, pooled: bool, kwargs: dict[str, Any], lags: int | None = None, lag_cols: tuple[str, ...] = (), per_bar: bool = False)[source]

Bases: object

Base class of a model source. A model source provides a model instance either by training one or by loading a pre-trained model.

Parameters:

name – Name of model.
indicator_names – Iterable of names of pybroker.indicator.Indicators used as features of the model.
input_data_fn – Callable[[DataFrame], DataFrame] for preprocessing input data passed to the model when making predictions. If set, input_data_fn will be called with a pandas.DataFrame containing all test data.
predict_fn – Callable[[Model, DataFrame], ndarray] that overrides calling the model’s default predict function. If set, predict_fn will be called with the trained model and a pandas.DataFrame containing all test data. When lags is set, it is instead called with the lag feature matrix (numpy.ndarray) in place of the DataFrame.
lags – Number of lagged values to include for each column in lag_cols, producing a (n_rows, len(lag_cols) * (lags + 1)) feature matrix that is passed to train_fn as lag_train/lag_test and to predict_fn in place of the input DataFrame, rather than added as columns.
lag_cols – Columns to compute lagged values for. Defaults to the data columns of the training data, excluding date and symbol; indicators are lagged only when named here.
per_bar – If True, predict_fn is called once per bar with input truncated to rows up to and including the current bar.
pooled – If True, the model is trained once per execution using combined multi-symbol data. Defaults to False.
kwargs – dict of additional kwargs.

intervals(*intervals: int | Literal['daily', 'weekly', 'monthly', 'quarterly', 'yearly'] | str) → IntervalBoundModel[source]

Binds this model to one or more compression intervals for use with pybroker.strategy.Strategy.add_execution().

A bound model is trained on exactly the listed intervals’ compressed bars, together with any indicators registered on it. Binding replaces the default base-timeframe training; include the literal 'base' in intervals to also train the model on the base timeframe. Per-interval predictions are read with pybroker.context.IntervalContext.preds(). Bound intervals are automatically made available through pybroker.context.ExecContext.interval() without also declaring them in the intervals parameter of add_execution():

trend = pybroker.model("trend", train_fn, indicators=[sma_10])
strategy.add_execution(
    fn, "SPY", models=trend.intervals("base", "weekly")
)

Only trainable models support interval binding. Calling this on a pretrained model (a ModelLoader) raises ValueError.

Parameters:: intervals – One or more TimeframeIntervals to train this model on, each strictly coarser than the base bar spacing of the backtest data, or the literal 'base' for the base timeframe.
Returns:: IntervalBoundModel binding this model to intervals.

prepare_input_data(df: DataFrame) → DataFrame[source]: Prepares a pandas.DataFrame of input data for passing to a model when making predictions. If set, the input_data_fn is used to preprocess the input data. If False, then indicator columns in df are used as input features.

class ModelTrainer(name: str, train_fn: Callable[[...], Any | tuple[Any, Iterable[str]]], indicator_names: Iterable[str], input_data_fn: Callable[[DataFrame], DataFrame] | None, predict_fn: Callable[[Any, DataFrame | NDArray], NDArray] | None, pooled: bool, kwargs: dict[str, Any], lags: int | None = None, lag_cols: tuple[str, ...] = (), per_bar: bool = False)[source]

Bases: ModelSource

Trains a model.

Parameters:

name – Name of model.
train_fn – When pooled is False, Callable[[symbol: str, train_data: DataFrame, test_data: DataFrame, ...], DataFrame]. When pooled is True, Callable[[symbols: Sequence[str], train_data: DataFrame, test_data: DataFrame, ...], DataFrame]. When lags is set, train_fn is additionally called with lag_train= and lag_test= keyword arguments holding the lag feature matrices aligned one row per train_data/test_data row, and must accept both. This is expected to return either a trained model instance, or a tuple containing a trained model instance and a Iterable of column names to to be used as input for the model when making predictions.
indicator_names – Iterable of names of pybroker.indicator.Indicators used as features of the model.
input_data_fn – Callable[[DataFrame], DataFrame] for preprocessing input data passed to the model when making predictions. If set, input_data_fn will be called with a pandas.DataFrame containing all test data.
predict_fn – Callable[[Model, DataFrame], ndarray] that overrides calling the model’s default predict function. If set, predict_fn will be called with the trained model and a pandas.DataFrame containing all test data. When lags is set, it is instead called with the lag feature matrix (numpy.ndarray) in place of the DataFrame.
pooled – If True, the model is trained once per execution using combined multi-symbol data. Defaults to False.
kwargs – dict of kwargs to pass to train_fn.

__call__(symbol: str, train_data: DataFrame, test_data: DataFrame, *, lag_train: NDArray | None = None, lag_test: NDArray | None = None) → Any | tuple[Any, Iterable[str]][source]

Trains model per symbol.

Parameters:

symbol – Ticker symbol of model (models are trained per symbol).
train_data – Train data.
test_data – Test data.
lag_train – Lag feature matrix aligned one row per train_data row. Passed to train_fn as lag_train= when the model is registered with lags.
lag_test – Lag feature matrix aligned one row per test_data row. Passed to train_fn as lag_test= when the model is registered with lags.

Returns:

Trained model.

train_pooled(symbols: Sequence[str], train_data: DataFrame, test_data: DataFrame, *, lag_train: NDArray | None = None, lag_test: NDArray | None = None) → Any | tuple[Any, Iterable[str]][source]

Trains model using combined multi-symbol data.

Parameters:

symbols – Ticker symbols of the pooled group, sorted in ascending order to match the order that symbol blocks appear in train_data and test_data. A listed symbol can have no rows in a frame, such as when lags drops all of its rows with the lag warmup.
train_data – Train data containing a symbol column.
test_data – Test data containing a symbol column.
lag_train – Lag feature matrix aligned one row per train_data row. Passed to train_fn as lag_train= when the model is registered with lags.
lag_test – Lag feature matrix aligned one row per test_data row. Passed to train_fn as lag_test= when the model is registered with lags.

Returns:

Trained model.

class ModelsMixin[source]

Bases: object

Mixin implementing model related functionality.

train_models(model_syms: Iterable[ModelSymbol], train_data: DataFrame, test_data: DataFrame, indicator_data: Mapping[IndicatorSymbol, Series], cache_date_fields: CacheDateFields, parallel_models: bool = False, pooled_model_groups: Mapping[tuple[str, int], frozenset[str]] | None = None, interval_data: IntervalData | None = None, *, history_store: SymbolArrayStore | None = None, train_store: SymbolArrayStore | None = None, test_store: SymbolArrayStore | None = None, lookahead: int = 1) → dict[ModelSymbol, TrainedModel][source]

Trains models for the provided pybroker.common.ModelSymbol pairs.

Parameters:

model_syms – Iterable of pybroker.common.ModelSymbol pairs of models to train.
train_data – pandas.DataFrame of training data.
test_data – pandas.DataFrame of test data.
indicator_data – Mapping of pybroker.common.IndicatorSymbol pairs to pandas.Series of pybroker.indicator.Indicator values.
cache_date_fields – Date fields used to key cache data.
parallel_models – If True, ModelTrainer models are trained in parallel using multiple processes. Defaults to False.
pooled_model_groups – Mapping of (model_name, execution_id) pairs to frozenset[str] of symbols for pooled training. Defaults to None.
lookahead – Number of bars in the future of the target prediction, expressed in the bars of the timeframe each model is fitted on: a model bound to an interval holds out lookahead compressed bars between its train and test rows. Defaults to 1.

Returns:

dict mapping each pybroker.common.ModelSymbol pair to a pybroker.common.TrainedModel.

apply_lags_to_model_input(model_input: ModelInput, lag_columns: tuple[str, ...], lags: int, lag_cache: dict[LagSeriesKey, ndarray], symbol: str, history_dates: ndarray, interval: str | None = None) → ModelInput[source]: Attaches lag feature metadata to model_input.

apply_lags_to_model_input_pooled(model_input: ModelInput, lag_columns: tuple[str, ...], lags: int, lag_cache: dict[LagSeriesKey, ndarray], history_dates_by_symbol: dict[str, ndarray], symbols: Iterable[str], interval: str | None = None) → ModelInput[source]: Attaches lag feature metadata to pooled model_input.

apply_prepare_input_data(model_input: ModelInput, prepare_fn: Callable[[DataFrame], DataFrame]) → ModelInput[source]: Applies a DataFrame-only prepare function to model_input.

build_lag_feature_matrix(symbol: str, columns: tuple[str, ...], lags: int, row_dates: ndarray, history_dates: ndarray, lag_cache: dict[LagSeriesKey, ndarray], interval: str | None = None) → ndarray[source]

Builds a lag-expanded feature matrix from numpy arrays.

Returns a matrix of shape (len(row_dates), len(columns) * (lags + 1)), laid out as one contiguous block per column, each block holding the column’s current value followed by lags 1 through lags.

build_lag_feature_matrix_pooled(sym_col: ndarray, columns: tuple[str, ...], lags: int, row_dates: ndarray, history_dates_by_symbol: dict[str, ndarray], lag_cache: dict[LagSeriesKey, ndarray], symbols: Iterable[str], interval: str | None = None) → ndarray[source]: Builds a lag-expanded feature matrix for pooled multi-symbol data.

cached_stacked_lags(cache: dict[LagSeriesKey, ndarray], symbol: str, col: str, lags: int, interval: str | None = None) → ndarray | None[source]: Returns a cached stacked array deep enough for lags, else None.

compute_lag_series_cache(df: DataFrame, symbols: Iterable[str], columns: tuple[str, ...], lags: int) → dict[LagSeriesKey, ndarray][source]: Computes full-history lag arrays for daily/base bars.

history_date_offset(history_dates: ndarray, row_dates: ndarray) → int[source]: Returns the start index of row_dates inside history_dates.

merge_interval_lag_series_cache(cache: dict[LagSeriesKey, ndarray], symbols: Iterable[str], columns: tuple[str, ...], lags: int, interval: str, bars_by_symbol, arrays_by_symbol=None) → dict[LagSeriesKey, ndarray][source]

Adds full-history interval lag arrays into cache.

pybroker.interval.CompressedBars holds data columns only, so arrays_by_symbol supplies columns the bars cannot – indicator values in particular – and is consulted first when given.

merge_lag_series_cache(cache: dict[LagSeriesKey, ndarray], history_df: DataFrame, symbols: Iterable[str], columns: tuple[str, ...], lags: int, history_dates: dict[str, ndarray] | None = None) → dict[LagSeriesKey, ndarray][source]: Adds full-history lag arrays for columns into cache.

merge_lag_series_cache_from_arrays(cache: dict[LagSeriesKey, ndarray], symbol: str, columns: tuple[str, ...], lags: int, history_dates: ndarray, column_arrays: Mapping[str, ndarray]) → None[source]: Adds full-history lag arrays built from numpy column data.

merge_lag_series_cache_from_store(cache: dict[LagSeriesKey, ndarray], store: SymbolArrayStore, symbols: Iterable[str], columns: tuple[str, ...], lags: int, history_dates: dict[str, ndarray] | None = None, indicators: tuple[str, ...] = (), indicator_data: Mapping[IndicatorSymbol, Series] | None = None) → dict[LagSeriesKey, ndarray][source]

Adds full-history lag arrays from a pybroker.scope.SymbolArrayStore.

A pybroker.scope.SymbolArrayStore holds data columns only, so indicator values are aligned onto the store’s dates from indicator_data, which covers each symbol’s full history. Lagging an indicator would otherwise fail even though it is a column of the model input.

model(name: str, fn: Callable[[...], Any | tuple[Any, Iterable[str]]], indicators: Iterable[Indicator] | None = None, lags: int | None = None, lag_cols: Iterable[str | Indicator] | None = None, per_bar: bool = False, input_data_fn: Callable[[DataFrame], DataFrame] | None = None, predict_fn: Callable[[Any, DataFrame | NDArray], NDArray] | None = None, pretrained: bool = False, pooled: bool = False, **kwargs) → ModelSource[source]

Creates a ModelSource instance and registers it globally with name.

Parameters:

name – Name for referencing the model globally.
fn – Callable used to either train or load a model instance. If for training with pooled=False, then fn has signature Callable[[symbol: str, train_data: DataFrame, test_data: DataFrame, ...], DataFrame]. If for training with pooled=True, then fn has signature Callable[[symbols: Sequence[str], train_data: DataFrame, test_data: DataFrame, ...], DataFrame] where symbols contains the pooled symbols sorted in ascending order, and both frames contain a symbol column with each symbol’s rows grouped together in that order. A listed symbol can have no rows in a frame, such as when lags drops all of its rows with the lag warmup. If for loading, then fn has signature Callable[[symbol: str, train_start_date: datetime, train_end_date: datetime, ...], DataFrame]. When lags is set, a training fn is additionally called with lag_train= and lag_test= keyword arguments holding the prebuilt lag feature matrices, and must accept both. This is expected to return either a trained model instance, or a tuple containing a trained model instance and a Iterable of column names to to be used as input for the model when making predictions. When only a model instance is returned, columns from the training DataFrame are used for prediction. For pooled models, the symbol column is omitted from inferred prediction columns.
indicators – Iterable of pybroker.indicator.Indicators used as features of the model.
lags – Number of lagged values to include for each column in lag_cols. The lagged values are built into a feature matrix of shape (n_rows, len(lag_cols) * (lags + 1)): one contiguous block per column in lag_cols declaration order, each block holding the column’s current value followed by lags 1 through lags, where lag 1 is the value from the previous bar. The matrix is passed to a training fn as the lag_train and lag_test keyword arguments, aligned one row per train_data/test_data row, and to predict_fn (or the model’s default predict) in place of the input DataFrame. Because the current bar’s value is the first feature of each block, the intended prediction target is the next bar — using the current bar’s value as the target would leak it. Lag data is kept separate from model input rather than added as columns, so the input pandas.DataFrame is never widened or copied. Lagged values are computed from each symbol’s full history, so rows at the start of a test window use real values carried over from the preceding train window instead of NaN. Rows whose lags are undefined are dropped from training data only.
lag_cols – Column names and/or pybroker.indicator.Indicators to compute lagged values for. pybroker.indicator.Indicators passed here are added to indicators. Declaration order sets the order of the feature blocks. Defaults to the data columns of the training data, excluding date and symbol; indicators are lagged only when named here.
per_bar – If True, predict_fn is called once per bar with input truncated to rows up to and including the current bar, and must return a scalar prediction for that bar. With lags, the input is the lag feature matrix truncated the same way, with the current bar as its last row. Use for models that are refit or updated every bar, such as GARCH or state space models. Note this makes one model call per bar per symbol, which is far slower than a single vectorized predict_fn call over the whole test window. Requires predict_fn and is not supported with pooled=True.
input_data_fn – Callable[[DataFrame], DataFrame] for preprocessing input data passed to the model when making predictions. If set, input_data_fn will be called with a pandas.DataFrame containing all test data, including when per_bar=True. It must return one row per bar; adding or dropping rows would misalign predictions with bars and raises a ValueError. For models registered with lags, it shapes pybroker.context.ExecContext.input() only, since predictions are made from the lag feature matrix.
predict_fn – Callable[[Model, DataFrame], ndarray] that overrides calling the model’s default predict function. If set, predict_fn will be called with the trained model and a pandas.DataFrame containing all test data. When lags is set, it is instead called with the lag feature matrix (numpy.ndarray) in place of the DataFrame. When per_bar=True, predict_fn receives input rows up to and including the current bar and must return a scalar prediction.
pretrained – If True, then fn is used to load and return a pre-trained model. If False, fn is used to train and return a new model. Defaults to False.
pooled – If True, the model is trained once per execution using combined multi-symbol data. Defaults to False.
**kwargs – Additional arguments to pass to fn.

Returns:

ModelSource instance.

model_input_from_arrays(columns: tuple[str, ...], arrays: dict[str, ndarray], dates: ndarray) → ModelInput[source]: Builds a ModelInput from column arrays.

model_input_from_frame(df: DataFrame, columns: tuple[str, ...] | None = None, dates: ndarray | None = None) → ModelInput[source]: Builds a ModelInput from a DataFrame without copying columns.

shift_array(values: ndarray, lag: int) → ndarray[source]: Returns values shifted forward by lag bars with NaN warmup.

symbol_history_arrays(history_df: DataFrame, symbol: str, columns: tuple[str, ...]) → tuple[ndarray, dict[str, ndarray]][source]: Extracts sorted full-history date and column arrays for one symbol.