fractal.core.base.strategy package

Submodules

fractal.core.base.strategy.logger module

class fractal.core.base.strategy.logger.BaseLogger(base_artifacts_path: str | None = None, class_name: str = None)[source]

Bases: ABC

abstract property base_artifacts_path: str

abstract property datasets_path: str

abstract debug(message: str)[source]

abstract property logs_path: str

class fractal.core.base.strategy.logger.DefaultLogger(base_artifacts_path: str | None = None, class_name: str = None)[source]

Bases: BaseLogger

property base_artifacts_path: str

property datasets_path: str

debug(message: str)[source]: Log a debug message.

property logs_path: str

fractal.core.base.strategy.observation module

class fractal.core.base.strategy.observation.Observation(timestamp: datetime, states: Dict[str, GlobalState])[source]

Bases: object

Observation is a snapshot of the entities states at a given timestamp.

fractal.core.base.strategy.result module

class fractal.core.base.strategy.result.StrategyMetrics(accumulated_return: float, apy: float, sharpe: float, max_drawdown: float)[source]

Bases: object

Default metrics of the strategy.

accumulated_return: float

apy: float

max_drawdown: float

sharpe: float

class fractal.core.base.strategy.result.StrategyResult(timestamps: List[datetime], internal_states: List[Dict[str, InternalState]], global_states: List[Dict[str, GlobalState]], balances: List[Dict[str, float]])[source]

Bases: object

Result of the strategy running. It contains the timestamps, internal states, global states of all entities and total balances

get_metrics(data: pd.DataFrame) -> StrategyMetrics

get_default_metrics() → StrategyMetrics[source]

to_dataframe() → pd.DataFrame[source]

balances: List[Dict[str, float]]

get_default_metrics() → StrategyMetrics[source]

Calculate default metrics of the strategy.

Returns:: Metrics of the strategy.
Return type:: StrategyMetrics

get_metrics(data: DataFrame, notional_price: str | float | None = None) → StrategyMetrics[source]

Calculate metrics of the strategy by StrategyResult data. StrategyResult data can be generated by to_dataframe() method.

Parameters:

data (pd.DataFrame) – DataFrame with the result.
notional_price (Optional[str | float], optional) – Notional price of the asset. If it is None, the notional price is 1. If it is a string, the notional price is column name of the DataFrame {entity}_{state_name} For example, if notional_price=’SPOT_price’, the notional price is price of global_state of SPOT. If it is a float, the notional price is the value of the float.

Returns:

Metrics of the strategy.

Return type:

StrategyMetrics

global_states: List[Dict[str, GlobalState]]

internal_states: List[Dict[str, InternalState]]

timestamps: List[datetime]

to_dataframe() → DataFrame[source]

Convert the result to a DataFrame.

Returns:: DataFrame with the result.
Return type:: pd.DataFrame

fractal.core.base.strategy.strategy module

class fractal.core.base.strategy.strategy.ActionToTake(entity_name, action)

Bases: NamedTuple

action: Action: Alias for field number 1

entity_name: str: Alias for field number 0

class fractal.core.base.strategy.strategy.BaseStrategy(*args, params: BaseStrategyParams | Dict | None = None, debug: bool = False, **kwargs)[source]

Bases: ABC

Base class for strategies.

Strategies are responsible for predicting the next action to take based on the current state of the entities.

Abstract Methods:: set_up(*args, **kwargs): Register entities, load models, execute some initial actions. Important Note: This method is called while initializing the strategy. In child classes, if implemented, also should be called parent’s set_up. predict(*args, **kwargs) -> List[Action]: Predict the next action to take.

get_all_available_entities() → Dict[str, Type[BaseEntity]][source]: Get all available entities.

get_entity(entity_name: str) → BaseEntity[source]: Get an entity by name.

property logger: BaseLogger

property params: Dict

abstract predict(*args, **kwargs) → List[ActionToTake][source]

Predict the next action to take based on the current state of the entities.

Returns:: List of actions to take.
Return type:: List[Action]

register_entity(entity: NamedEntity) → None[source]: Register an entity for the strategy. It is used to keep track of the entities in the strategy. Each entity should have a unique name in entities registry of the strategy.

run(observations: List[Observation]) → StrategyResult[source]: Run the strategy on a sequence of observations. Execute self.step for each observation.

set_params(params: BaseStrategyParams | Dict) → None[source]

Set parameters for the strategy.

Parameters:

params (BaseStrategyParams | Dict) – Parameters for the strategy.
passed (If dict is)
BaseStrategyParams (it will be converted to)
values. (with the fields of the dictionary keys and)

abstract set_up(*args, **kwargs)[source]: Set up the strategy. Register entities, load models, etc.

step(observation: Observation)[source]: Take a step in the simulation by observations.

class fractal.core.base.strategy.strategy.BaseStrategyParams(data: Dict | None = None)[source]

Bases: object

Base class for strategy parameters. It is used to store hyperparameters of the strategy. It can be initialized by a dictionary.

class fractal.core.base.strategy.strategy.NamedEntity(entity_name, entity)

Bases: NamedTuple

entity: BaseEntity: Alias for field number 1

entity_name: str: Alias for field number 0

Module contents

class fractal.core.base.strategy.ActionToTake(entity_name, action)

Bases: NamedTuple

action: Action: Alias for field number 1

entity_name: str: Alias for field number 0

class fractal.core.base.strategy.BaseStrategy(*args, params: BaseStrategyParams | Dict | None = None, debug: bool = False, **kwargs)[source]

Bases: ABC

Base class for strategies.

Strategies are responsible for predicting the next action to take based on the current state of the entities.

Abstract Methods:: set_up(*args, **kwargs): Register entities, load models, execute some initial actions. Important Note: This method is called while initializing the strategy. In child classes, if implemented, also should be called parent’s set_up. predict(*args, **kwargs) -> List[Action]: Predict the next action to take.

get_all_available_entities() → Dict[str, Type[BaseEntity]][source]: Get all available entities.

get_entity(entity_name: str) → BaseEntity[source]: Get an entity by name.

property logger: BaseLogger

property params: Dict

abstract predict(*args, **kwargs) → List[ActionToTake][source]

Predict the next action to take based on the current state of the entities.

Returns:: List of actions to take.
Return type:: List[Action]

register_entity(entity: NamedEntity) → None[source]: Register an entity for the strategy. It is used to keep track of the entities in the strategy. Each entity should have a unique name in entities registry of the strategy.

run(observations: List[Observation]) → StrategyResult[source]: Run the strategy on a sequence of observations. Execute self.step for each observation.

set_params(params: BaseStrategyParams | Dict) → None[source]

Set parameters for the strategy.

Parameters:

params (BaseStrategyParams | Dict) – Parameters for the strategy.
passed (If dict is)
BaseStrategyParams (it will be converted to)
values. (with the fields of the dictionary keys and)

abstract set_up(*args, **kwargs)[source]: Set up the strategy. Register entities, load models, etc.

step(observation: Observation)[source]: Take a step in the simulation by observations.

class fractal.core.base.strategy.BaseStrategyParams(data: Dict | None = None)[source]

Bases: object

Base class for strategy parameters. It is used to store hyperparameters of the strategy. It can be initialized by a dictionary.

class fractal.core.base.strategy.NamedEntity(entity_name, entity)

Bases: NamedTuple

entity: BaseEntity: Alias for field number 1

entity_name: str: Alias for field number 0

class fractal.core.base.strategy.Observation(timestamp: datetime, states: Dict[str, GlobalState])[source]

Bases: object

Observation is a snapshot of the entities states at a given timestamp.

class fractal.core.base.strategy.StrategyMetrics(accumulated_return: float, apy: float, sharpe: float, max_drawdown: float)[source]

Bases: object

Default metrics of the strategy.

accumulated_return: float

apy: float

max_drawdown: float

sharpe: float

class fractal.core.base.strategy.StrategyResult(timestamps: List[datetime], internal_states: List[Dict[str, InternalState]], global_states: List[Dict[str, GlobalState]], balances: List[Dict[str, float]])[source]

Bases: object

Result of the strategy running. It contains the timestamps, internal states, global states of all entities and total balances

get_metrics(data: pd.DataFrame) -> StrategyMetrics

get_default_metrics() → StrategyMetrics[source]

to_dataframe() → pd.DataFrame[source]

balances: List[Dict[str, float]]

get_default_metrics() → StrategyMetrics[source]

Calculate default metrics of the strategy.

Returns:: Metrics of the strategy.
Return type:: StrategyMetrics

get_metrics(data: DataFrame, notional_price: str | float | None = None) → StrategyMetrics[source]

Calculate metrics of the strategy by StrategyResult data. StrategyResult data can be generated by to_dataframe() method.

Parameters:

data (pd.DataFrame) – DataFrame with the result.
notional_price (Optional[str | float], optional) – Notional price of the asset. If it is None, the notional price is 1. If it is a string, the notional price is column name of the DataFrame {entity}_{state_name} For example, if notional_price=’SPOT_price’, the notional price is price of global_state of SPOT. If it is a float, the notional price is the value of the float.

Returns:

Metrics of the strategy.

Return type:

StrategyMetrics

global_states: List[Dict[str, GlobalState]]

internal_states: List[Dict[str, InternalState]]

timestamps: List[datetime]

to_dataframe() → DataFrame[source]

Convert the result to a DataFrame.

Returns:: DataFrame with the result.
Return type:: pd.DataFrame