Agent Module

class AgentHandler(model_blue_print: BluePrintBase, trading_environment: TradingEnvironment, learning_strategy_handler: LearningStrategyHandlerBase, testing_strategy_handler: TestingStrategyHandlerBase)

Bases: object

Implements agent handler that is responsible for training and testing the agent in the given trading environment using the specified learning and testing strategies. It is used as an wrapper around the agent to provide a convenient interface for TrainingHandler.

print_model_summary(print_function: ~typing.Optional[~typing.Callable] = <built-in function print>) → None

Prints a summary of the model architecture and parameters.

Parameters:

print_function (Optional[Callable]) – A function to print the summary. Defaults to print.

test_agent(repeat: int = 1) → tuple[dict[int, list[str]], dict[int, list[dict[str, Any]]]]

Tests the agent using the specified number of repetitions.

Parameters:

repeat (int) – The number of times to repeat the testing process.

Returns:

A tuple containing the keys and

report data from the testing process.

Return type:

(tuple[dict[int, list[str]], dict[int, list[dict[str, Any]]]])

train_agent(nr_of_steps: int, nr_of_episodes: int, callbacks: Optional[list[tensorflow.python.keras.callbacks.Callback]] = None, model_load_path: Optional[str] = None, model_save_path: Optional[str] = None) → tuple[list[str], list[dict]]

Trains the agent using the specified number of steps and episodes.

Parameters:

  • nr_of_steps (int) – The number of steps to train the agent.

  • nr_of_episodes (int) – The number of episodes to train the agent.

  • callbacks (Optional[list[Callback]]) – A list of callbacks to be used during training.

  • model_load_path (Optional[str]) – Path to load the pre-trained model from.

  • model_save_path (Optional[str]) – Path to save the trained model to.

Returns:

A tuple containing the keys and report data from

the training process.

Return type:

(tuple[list[str], list[dict]])

class AgentBase(model_adapter: ModelAdapterBase)

Bases: object

Implements base class for agents that can be trained and tested in a trading environment. It provides basic functionalities such as loading and saving models,printing model summaries.

load_model(model_path: str) → None

Loads the model from the specified path.

Parameters:

model_path (str) – The path to the model file.

print_summary(print_function: ~typing.Optional[~typing.Callable] = <built-in function print>) → None

Prints a summary of the model architecture and parameters.

Parameters:

print_function (Optional[Callable]) – A function to print the summary. Defaults to print.

save_model(model_path: str) → None

Saves the model to the specified path.

Parameters:

model_path (str) – The path to the model file.

class ClassificationLearningAgent(model_adapter: ModelAdapterBase)

Bases: AgentBase, ClassificationTestable

Implements a classification learning agent that can be trained and tested in a trading environment. It provides functionalities for fitting the model with classification data and making predictions.

classification_fit(input_data: ndarray, output_data: ndarray, validation_data: tuple[numpy.ndarray, numpy.ndarray], batch_size: int, epochs: int, callbacks: list[tensorflow.python.keras.callbacks.Callback]) → dict[str, Any]

Fits the model to the classification data. Parameters passed to the model adapter are dynamically determined based on the model’s requirements.

Parameters:

  • input_data (np.ndarray) – The input data for training.

  • output_data (np.ndarray) – The output data for training.

  • validation_data (tuple[np.ndarray, np.ndarray]) – The validation data (input, output).

  • batch_size (int) – The batch size to be used during training.

  • epochs (int) – The number of epochs to train the model.

  • callbacks (list[Callback]) – A list of callbacks to be used during training.

Returns:

A dictionary containing the training history and other relevant information.

Return type:

(dict[str, Any])

classify(data: DataFrame) → list[list[float]]

Classifies the input data using the trained model.

Parameters:

data (pd.DataFrame) – The input data to be classified.

Returns:

The predicted class probabilities for each input sample.

Return type:

(list[list[float]])

get_model_adapter_tag() → str

Returns the tag of the model adapter.

Returns:

The tag of the model adapter.

Return type:

(str)

load_model(model_path: str) → None

Loads the model from the specified path.

Parameters:

model_path (str) – The path to the model file.

print_summary(print_function: ~typing.Optional[~typing.Callable] = <built-in function print>) → None

Prints a summary of the model architecture and parameters.

Parameters:

print_function (Optional[Callable]) – A function to print the summary. Defaults to print.

save_model(model_path: str) → None

Saves the model to the specified path.

Parameters:

model_path (str) – The path to the model file.

class ReinforcementLearningAgent(model: Model, policy: Policy, optimizer: OptimizerV2)

Bases: AgentBase, PerformanceTestable

Implements a reinforcement learning agent using DQN that can be trained and tested in a trading environment. It provides functionalities for fitting the model with reinforcement learning data and performing actions based on observations.

load_model(model_path: str) → None

Loads the model weights from the specified file path.

Parameters:

model_path (str) – The path to the model weights file.

perform(observation: list[float]) → int

Performs the action for the agent based on the given observation.

Parameters:

observation (list[float]) – The observation data to use for the action.

Returns:

The result of the action.

Return type:

(int)

print_summary(print_function: ~typing.Optional[~typing.Callable] = <built-in function print>) → None

Prints a summary of the model architecture.

Parameters:

print_function (Optional[Callable]) – A function to use for printing the summary. Defaults to the print function.

reinforcement_learning_fit(environment: TradingEnvironment, nr_of_steps: int, steps_per_episode: int, callbacks: list[tensorflow.python.keras.callbacks.Callback]) → dict[str, Any]

Trains the reinforcement learning agent using the specified environment and parameters.

Parameters:

  • environment (TradingEnvironment) – The trading environment to use for training.

  • nr_of_steps (int) – The total number of steps to train the agent.

  • steps_per_episode (int) – The number of steps per episode.

  • callbacks (list[Callback]) – A list of Keras callbacks to use during training.

Returns:

The training history of the agent.

Return type:

(dict[str, Any])

save_model(model_path: str) → None

Saves the model weights to the specified file path.

Parameters:

model_path (str) – The path to the model weights file.

class ClassificationLearningStrategyHandler

Bases: LearningStrategyHandlerBase

Implements a learning strategy handler for classification tasks. It provides functionalities for creating agents and fitting models.

PLOTTING_KEY: str = 'asset_price_movement_summary'
PLOTTING_KEYS: str = ['price_movement_trend_class_summary', 'classification_learning']
create_agent(model_blue_print: BluePrintBase, trading_environment: TradingEnvironment) → AgentBase

Creates a classification learning agent. It dynamically determines the parameters needed for instantiation based on the model blueprint.

Parameters:

  • model_blue_print (BluePrintBase) – The model blueprint to be used for agent creation.

  • trading_environment (TradingEnvironment) – The trading environment in which the agent will operate.

Returns:

An instance of the classification learning agent created using the model blueprint and trading environment.

Return type:

(Agent)

fit(agent: ClassificationLearningAgent, trading_environment: TradingEnvironment, nr_of_steps: int, nr_of_episodes: int, callbacks: list[tensorflow.python.keras.callbacks.Callback]) → tuple[list[str], list[dict[str, Any]]]

Fits the classification learning agent to the trading environment.

Parameters:

  • agent (ClassificationLearningAgent) – The classification learning agent to fit.

  • environment (TradingEnvironment) – The trading environment to use.

  • nr_of_steps (int) – The number of training steps to perform.

  • nr_of_episodes (int) – The number of training episodes to perform.

  • callbacks (list[Callback]) – List of callbacks to use during training.

Raises:

TypeError – If the agent is not an instance of ClassificationLearningAgent.

Returns:

A tuple containing the keys and data collected during training.

Return type:

(list[str], list[dict[str, Any]])

class ClassificationTestable

Bases: ABC

Implements an abstract base class for classification testable agents.

abstract classify(data: DataFrame) → list[list[float]]

Classifies the input data using the trained model.

Parameters:

data (pd.DataFrame) – The input data to be classified.

Returns:

The predicted class probabilities for each input sample.

Return type:

(list[list[float]])

class ClassificationTestingStrategyHandler

Bases: TestingStrategyHandlerBase

Implements a testing strategy handler for classification tasks.

PLOTTING_KEY: str = 'classification_testing'
evaluate(testable_agent: ClassificationTestable, environment: TradingEnvironment) → tuple[list[str], list[dict[str, Any]]]

Evaluates the classification model using the given testable agent and trading environment.

Parameters:
Returns:

A tuple containing the keys and data collected during evaluation.

Return type:

(tuple[list[str], list[dict[str, Any]]])

class LearningStrategyHandlerBase

Bases: ABC

Implements a base class for learning strategy handlers. It provides an interface for creating agents, fitting them to the environment, and providing parameters needed for the model blueprint instantiation.

PLOTTING_KEY: str = 'asset_price_movement_summary'
abstract create_agent(model_blue_print: BluePrintBase, trading_environment: TradingEnvironment) → AgentBase

Creates an agent based on the provided model blueprint and trading environment.

Parameters:

  • model_blue_print (BluePrintBase) – The model blueprint to be used for agent creation.

  • trading_environment (TradingEnvironment) – The trading environment in which the agent will operate.

Returns:

An instance of the agent created using the model blueprint and trading environment.

Return type:

(AgentBase)

abstract fit(agent: AgentBase, environment: TradingEnvironment, nr_of_steps: int, nr_of_episodes: int, callbacks: list[tensorflow.python.keras.callbacks.Callback]) → tuple[list[str], list[dict[str, Any]]]

Fits the agent to the environment using the specified number of steps and episodes. It also collects data for plotting summary statistics.

Parameters:

  • agent (AgentBase) – The agent to be fitted to the environment.

  • environment (TradingEnvironment) – The trading environment in which the agent will be trained.

  • nr_of_steps (int) – The number of steps to be taken during training.

  • nr_of_episodes (int) – The number of episodes to be run during training.

  • callbacks (list[Callback]) – A list of Keras callbacks to be used during training.

Returns:

A tuple containing a list of keys and a list of dictionaries with the data collected during training.

Return type:

(tuple[list[str], list[dict[str, Any]]])

class PerformanceTestable

Bases: ABC

Implements a performance testable interface for agents.

abstract perform(observation: list[float]) → int

Performs the action for the agent based on the given observation.

Parameters:

observation (list[float]) – The observation data to use for the action.

Returns:

The result of the action.

Return type:

(int)

class PerformanceTestingStrategyHandler

Bases: TestingStrategyHandlerBase

Implements a performance testing strategy handler for agents. It provides functionalities for evaluating the performance of agents in a trading environment.

PLOTTING_KEY: str = 'performance_testing'
evaluate(testable_agent: PerformanceTestable, environment: TradingEnvironment) → tuple[list[str], list[dict[str, Any]]]

Evaluates the performance of the given testable agent in the specified trading environment.

Parameters:
Returns:

A tuple containing the keys and data collected during evaluation.

Return type:

(tuple[list[str], list[dict[str, Any]]])

class ReinforcementLearningStrategyHandler(policy: ~rl.policy.Policy = <rl.policy.BoltzmannQPolicy object>, optimizer: ~tensorflow.python.keras.optimizer_v2.optimizer_v2.OptimizerV2 = <tensorflow.python.keras.optimizer_v2.adam.Adam object>)

Bases: LearningStrategyHandlerBase

Implements a reinforcement learning strategy handler. It provides methods for creating reinforcement learning agents, fitting them to the trading environment, and providing parameters needed for model instantiation coherent with reinforcement learning.

PLOTTING_KEY: str = 'reinforcement_learning'
create_agent(model_blue_print: BluePrintBase, trading_environment: TradingEnvironment) → AgentBase

Creates a reinforcement learning agent. It dynamically determines the parameters needed for instantiation based on the model blueprint.

Parameters:

  • model_blue_print (BluePrintBase) – The model blueprint to be used for agent creation.

  • trading_environment (TradingEnvironment) – The trading environment in which the agent will operate.

Raises:

TypeError – If the model adapter is not an instance of TFModelAdapter.

Returns:

An instance of the agent created using the model blueprint and trading environment.

Return type:

(AgentBase)

fit(agent: ReinforcementLearningAgent, trading_environment: TradingEnvironment, nr_of_steps: int, nr_of_episodes: int, callbacks: list[tensorflow.python.keras.callbacks.Callback]) → tuple[list[str], dict[str, Any]]

Fits the reinforcement learning agent to the trading environment.

Parameters:

  • agent (ReinforcementLearningAgent) – The reinforcement learning agent to fit.

  • trading_environment (TradingEnvironment) – The trading environment to use.

  • nr_of_steps (int) – The number of training steps to perform.

  • nr_of_episodes (int) – The number of training episodes to perform.

  • callbacks (list[Callback]) – List of callbacks to use during training.

Raises:

TypeError – If the agent is not an instance of ReinforcementLearningAgent.

Returns:

A tuple containing the keys and data collected during training.

Return type:

(list[str], dict[str, Any])

class TestingStrategyHandlerBase

Bases: ABC

Implements a base class for testing strategy handlers. It expects derived classes to implement functions for evaluating agents in a trading environment.

abstract evaluate(agent: AgentBase, environment: TradingEnvironment) → tuple[list[str], list[dict[str, Any]]]

Evaluates the performance of the given agent in the specified trading environment.

Parameters:
Returns:

A tuple containing the keys and data collected during evaluation.

Return type:

(tuple[list[str], list[dict[str, Any]]])