generic_features

Classes

`FeatureType`	Enum for feature types.
`FillStrategy`	Enum for fill strategies used in Feature cleaning.
`OperationType`	Enum for rolling operations.
`Feature`	Represents a single feature with its name and value.
`Candle`	Represents a candle feature with OHLCV data.
`MovingWindow`	Represents a moving average feature by Days.
`ATR`	Average True Range (ATR) feature.
`RSI`	Relative Strength Index (RSI) feature.
`MACD`	Moving Average Convergence Divergence (MACD) feature.
`BollingerBands`	Bollinger Bands feature.
`MSTD`	Moving Standard Deviation feature.
`OBV`	On-Balance Volume (OBV) feature.
`Stochastic`	Stochastic Oscillator feature.
`Turbulence`	Computes a market turbulence index using multiple return periods and proper covariance calculation.

Module Contents

class generic_features.FeatureType

Bases: str, enum.Enum

Enum for feature types.

CANDLE = 'candle'

MOVING_WINDOW = 'moving_window'

RSI = 'rsi'

MACD = 'macd'

BOLLINGER_BANDS = 'bollinger_bands'

STOCHASTIC = 'stochastic'

PIOTROSKI = 'piotroski'

TURBULENCE = 'turbulence'

ATR = 'atr'

MSTD = 'mstd'

OBV = 'obv'

class generic_features.FillStrategy

Bases: str, enum.Enum

Enum for fill strategies used in Feature cleaning.

INTERPOLATE = 'interpolate'

ZERO = 'zero'

DROP = 'drop'

BACKWARD_FILL = 'bfill'

class generic_features.OperationType

Bases: str, enum.Enum

Enum for rolling operations.

MEAN = 'mean'

STD = 'std'

__call__(df, column, window)

Apply the rolling operation to a DataFrame column.

Parameters:

df (pd.DataFrame) – Input DataFrame.
column (str) – Name of the column to operate on.
window (int) – Rolling window size.

Returns:

Result of the rolling operation.

Return type:

pd.Series

class generic_features.Feature(**data)

Bases: pydantic.BaseModel

Represents a single feature with its name and value. This is an abstract base class for all features. Subclasses should implement the to_df method to convert feature data into a DataFrame. It also provides methods for cleaning columns, getting feature names, and a factory method to create instances based on a dictionary input. The Feature class is not meant to be instantiated directly; instead, it serves as a base class for specific feature implementations like Candle, MovingWindow, RSI, etc. Subclasses must define the TYPE class variable to specify their feature type. Subclasses can also define additional fields specific to their feature type.

type: FeatureType

name: str = None

enabled: bool = None

source: str | None = None

fill_strategy: FillStrategy = None

_registry: ClassVar[Dict[FeatureType, Type[Feature]]]

shape(data)

Shape the data to ensure all symbols have the same timestamps. Missing timestamps for symbols are filled with NaNs.

Parameters:: data (pd.DataFrame) – Input DataFrame with MultiIndex (timestamp, symbol).
Returns:: Shaped DataFrame where all symbols have identical timestamps.
Return type:: pd.DataFrame

clean_columns(cols, df)

Clean the DataFrame by filling NaN values in specified columns with given strategy. Filling is applied per symbol to avoid cross-contamination between symbols. :param cols: List of column names to clean. :type cols: List[str] :param df: DataFrame to clean. :type df: pd.DataFrame

Returns:

Cleaned DataFrame with NaN values handled according to the fill strategy.

Return type:

pd.DataFrame

Parameters:

cols (List[str])
df (pandas.DataFrame)

get_feature_names()

Get the names of the features provided by this instance. :returns: List of feature names. :rtype: List[str]

Return type:: List[str]

to_df(df, data)

Convert the feature data to a DataFrame. :param df: DataFrame to which the feature will be added. :type df: pd.DataFrame :param data: Data to convert into a DataFrame. :type data: Any

Returns:

DataFrame with the feature data added.

Return type:

pd.DataFrame

Raises:

NotImplementedError – If the method is not implemented in the subclass.

Parameters:

df (pandas.DataFrame)
data (Any)

classmethod __init_subclass__(**kwargs): Register subclasses of Feature in the _registry dictionary.

classmethod factory(data, fill_strategy)

Factory method to create a Feature instance from a dictionary. :param data: Dictionary containing feature data, must include ‘type’. :type data: dict

Returns:

An instance of a subclass of Feature based on the ‘type’ field.

Return type:

Feature

Raises:

ValueError – If ‘type’ is missing or if the type is unknown.

Parameters:

data (dict)
fill_strategy (FillStrategy)

static _zscore(s)

Parameters:: s (pandas.Series)
Return type:: pandas.Series

static _robust_zscore(s)

Parameters:: s (pandas.Series)
Return type:: pandas.Series

static _minmax(s)

Parameters:: s (pandas.Series)
Return type:: pandas.Series

static _sigmoid_pct_change(s)

Parameters:: s (pandas.Series)
Return type:: pandas.Series

class generic_features.Candle(**data)

Bases: Feature

Represents a candle feature with OHLCV data.

get_feature_names()

Get the names of the features provided by this instance. :returns: List of feature names. :rtype: List[str]

Return type:: List[str]

get_raw_feature_names()

Return type:: List[str]

normalize(df)

Parameters:: df (pandas.DataFrame)
Return type:: pandas.DataFrame

to_df(df, data)

Convert the feature data to a DataFrame. :param df: DataFrame to which the feature will be added. :type df: pd.DataFrame :param data: Data to convert into a DataFrame. :type data: Any

Returns:

DataFrame with the feature data added.

Return type:

pd.DataFrame

Raises:

NotImplementedError – If the method is not implemented in the subclass.

Parameters:

df (pandas.DataFrame)
data (Any)

TYPE: ClassVar[FeatureType]

class generic_features.MovingWindow(**data)

Bases: Feature

Represents a moving average feature by Days. Note: This feature is readily available in vectorbt, but this implementation allows for custom operations and fields.

to_df(df, data)

Convert the feature data to a DataFrame. :param df: DataFrame to which the feature will be added. :type df: pd.DataFrame :param data: Data to convert into a DataFrame. :type data: Any

Returns:

DataFrame with the feature data added.

Return type:

pd.DataFrame

Raises:

NotImplementedError – If the method is not implemented in the subclass.

Parameters:

df (pandas.DataFrame)
data (Any)

TYPE: ClassVar[FeatureType]

period: int = None

field: str = None

operation: OperationType = None

class generic_features.ATR(**data)

Bases: Feature

Average True Range (ATR) feature.

TYPE: ClassVar[FeatureType]

period: int = None

field_high: str = None

field_low: str = None

field_close: str = None

get_feature_names()

Get the names of the features provided by this instance. :returns: List of feature names. :rtype: List[str]

Return type:: List[str]

get_raw_feature_names()

Return type:: List[str]

to_df(df, data)

Convert the feature data to a DataFrame. :param df: DataFrame to which the feature will be added. :type df: pd.DataFrame :param data: Data to convert into a DataFrame. :type data: Any

Returns:

DataFrame with the feature data added.

Return type:

pd.DataFrame

Raises:

NotImplementedError – If the method is not implemented in the subclass.

Parameters:

df (pandas.DataFrame)
data (Any)

class generic_features.RSI(**data)

Bases: Feature

Relative Strength Index (RSI) feature.

TYPE: ClassVar[FeatureType]

period: int = None

field: str = None

ewm: bool = None

to_df(df, data)

Convert the feature data to a DataFrame. :param df: DataFrame to which the feature will be added. :type df: pd.DataFrame :param data: Data to convert into a DataFrame. :type data: Any

Returns:

DataFrame with the feature data added.

Return type:

pd.DataFrame

Raises:

NotImplementedError – If the method is not implemented in the subclass.

Parameters:

df (pandas.DataFrame)
data (Any)

class generic_features.MACD(**data)

Bases: Feature

Moving Average Convergence Divergence (MACD) feature.

TYPE: ClassVar[FeatureType]

fast_period: int = None

slow_period: int = None

signal_period: int = None

field: str = None

ewm: bool = None

signal_ewm: bool = None

get_feature_names()

Get the names of the features provided by this instance. :returns: List of feature names. :rtype: List[str]

Return type:: List[str]

get_raw_feature_names()

Return type:: List[str]

normalize(df)

Parameters:: df (pandas.DataFrame)
Return type:: pandas.DataFrame

to_df(df, data)

Convert the feature data to a DataFrame. :param df: DataFrame to which the feature will be added. :type df: pd.DataFrame :param data: Data to convert into a DataFrame. :type data: Any

Returns:

DataFrame with the feature data added.

Return type:

pd.DataFrame

Raises:

NotImplementedError – If the method is not implemented in the subclass.

Parameters:

df (pandas.DataFrame)
data (Any)

class generic_features.BollingerBands(**data)

Bases: Feature

Bollinger Bands feature.

TYPE: ClassVar[FeatureType]

period: int = None

field: str = None

ewm: bool = None

alpha: float = None

get_feature_names()

Get the names of the features provided by this instance. :returns: List of feature names. :rtype: List[str]

Return type:: List[str]

get_raw_feature_names()

Return type:: List[str]

normalize(df)

Parameters:: df (pandas.DataFrame)
Return type:: pandas.DataFrame

to_df(df, data)

Convert the feature data to a DataFrame. :param df: DataFrame to which the feature will be added. :type df: pd.DataFrame :param data: Data to convert into a DataFrame. :type data: Any

Returns:

DataFrame with the feature data added.

Return type:

pd.DataFrame

Raises:

NotImplementedError – If the method is not implemented in the subclass.

Parameters:

df (pandas.DataFrame)
data (Any)

class generic_features.MSTD(**data)

Bases: Feature

Moving Standard Deviation feature.

TYPE: ClassVar[FeatureType]

field: str = None

window: int = None

ewm: bool = None

to_df(df, data)

Convert the feature data to a DataFrame. :param df: DataFrame to which the feature will be added. :type df: pd.DataFrame :param data: Data to convert into a DataFrame. :type data: Any

Returns:

DataFrame with the feature data added.

Return type:

pd.DataFrame

Raises:

NotImplementedError – If the method is not implemented in the subclass.

Parameters:

df (pandas.DataFrame)
data (Any)

class generic_features.OBV(**data)

Bases: Feature

On-Balance Volume (OBV) feature.

TYPE: ClassVar[FeatureType]

field_close: str = None

field_volume: str = None

get_feature_names()

Get the names of the features provided by this instance. :returns: List of feature names. :rtype: List[str]

Return type:: List[str]

get_raw_feature_names()

Return type:: List[str]

normalize(df)

Parameters:: df (pandas.DataFrame)
Return type:: pandas.DataFrame

to_df(df, data)

Convert the feature data to a DataFrame. :param df: DataFrame to which the feature will be added. :type df: pd.DataFrame :param data: Data to convert into a DataFrame. :type data: Any

Returns:

DataFrame with the feature data added.

Return type:

pd.DataFrame

Raises:

NotImplementedError – If the method is not implemented in the subclass.

Parameters:

df (pandas.DataFrame)
data (Any)

class generic_features.Stochastic(**data)

Bases: Feature

Stochastic Oscillator feature.

TYPE: ClassVar[FeatureType]

period: int = None

field_high: str = None

field_low: str = None

field_close: str = None

k_window: int = None

d_window: int = None

ewm: bool = None

get_feature_names()

Get the names of the features provided by this instance. :returns: List of feature names. :rtype: List[str]

Return type:: List[str]

to_df(df, data)

Convert the feature data to a DataFrame. :param df: DataFrame to which the feature will be added. :type df: pd.DataFrame :param data: Data to convert into a DataFrame. :type data: Any

Returns:

DataFrame with the feature data added.

Return type:

pd.DataFrame

Raises:

NotImplementedError – If the method is not implemented in the subclass.

Parameters:

df (pandas.DataFrame)
data (Any)

class generic_features.Turbulence(**data)

Bases: Feature

Computes a market turbulence index using multiple return periods and proper covariance calculation. This measures how unusual current market conditions are compared to historical patterns. ! TODO pretty sure turbulence is broken

TYPE: ClassVar[FeatureType]

lookback: int = None

field: str = None

return_periods: List[int] = None

to_df(df, data=None)

Convert the feature data to a DataFrame. :param df: DataFrame to which the feature will be added. :type df: pd.DataFrame :param data: Data to convert into a DataFrame. :type data: Any

Returns:

DataFrame with the feature data added.

Return type:

pd.DataFrame

Raises:

NotImplementedError – If the method is not implemented in the subclass.

Parameters:

df (pandas.DataFrame)
data (Any)