Coverage for source/agent/strategies/performance_testing_strategy_handler.py: 100%

1# agent/strategies/performance_testing_strategy_handler.py 

2 

3# global imports 

4import numpy as np 

5import random 

6from typing import Any 

7 

8# local imports 

9from source.agent import PerformanceTestable, TestingStrategyHandlerBase 

10from source.environment import TradingEnvironment 

11 

12class PerformanceTestingStrategyHandler(TestingStrategyHandlerBase): 

13 """ 

14 Implements a performance testing strategy handler for agents. It provides 

15 functionalities for evaluating the performance of agents in a trading environment. 

16 """ 

17 

18 # global class constants 

19 PLOTTING_KEY: str = 'performance_testing' 

20 

21 def evaluate(self, testable_agent: PerformanceTestable, environment: TradingEnvironment, 

22 env_length_range: tuple[int, int]) -> tuple[list[str], list[dict[str, Any]]]: 

23 """ 

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

25 

26 Parameters: 

27 testable_agent (PerformanceTestable): The agent to evaluate. 

28 environment (TradingEnvironment): The trading environment to use for evaluation. 

29 env_length_range (tuple[int, int]): A tuple specifying the range of environment lengths to consider. 

30 

31 Returns: 

32 (tuple[list[str], list[dict[str, Any]]]): A tuple containing the keys and data collected during evaluation. 

33 """ 

34 

35 history = {} 

36 assets_values = [] 

37 reward_values = [] 

38 iterations = [] 

39 done = False 

40 

41 environment.reset(env_length_range[0]) 

42 state = environment.state 

43 current_iteration = environment.current_iteration 

44 trading_data = environment.get_trading_data() 

45 current_assets = trading_data.current_budget + trading_data.currently_invested 

46 iterations.append(current_iteration) 

47 assets_values.append(current_assets) 

48 reward_values.append(0) 

49 

50 while(not done): 

51 next_action = testable_agent.perform(state) 

52 state, reward, done, info = environment.step(next_action) 

53 

54 iterations.append(environment.current_iteration) 

55 assets_values.append(info['current_budget'] + info['currently_invested']) 

56 reward_values.append(reward) 

57 

58 if environment.current_iteration == env_length_range[1]: 

59 done = True 

60 

61 solvency_coefficient = round((assets_values[-1] - assets_values[0]) / (iterations[-1] - iterations[0]), 3) 

62 assets_values = (np.array(assets_values) / assets_values[0]).tolist() 

63 currency_prices = environment.get_data_for_iteration(['close'], iterations[0], iterations[-1]) 

64 currency_prices = (np.array(currency_prices) / currency_prices[0]).tolist() 

65 

66 history['assets_values'] = assets_values 

67 history['reward_values'] = reward_values 

68 history['currency_prices'] = currency_prices 

69 history['iterations'] = iterations 

70 history['solvency_coefficient'] = solvency_coefficient 

71 history['trading_points_per_year'] = environment.get_number_of_trading_points_per_year() 

72 

73 return [self.PLOTTING_KEY], [history]