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 

5from typing import Any 

6 

7# local imports 

8from source.agent import PerformanceTestable, TestingStrategyHandlerBase 

9from source.environment import TradingEnvironment 

10 

11class PerformanceTestingStrategyHandler(TestingStrategyHandlerBase): 

12 """ 

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

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

15 """ 

16 

17 # global class constants 

18 PLOTTING_KEY: str = 'performance_testing' 

19 

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

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

22 """ 

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

24 

25 Parameters: 

26 testable_agent (PerformanceTestable): The agent to evaluate. 

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

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

29 

30 Returns: 

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

32 """ 

33 

34 history = {} 

35 assets_values = [] 

36 reward_values = [] 

37 iterations = [] 

38 done = False 

39 

40 environment.reset(env_length_range[0]) 

41 state = environment.state 

42 current_iteration = environment.current_iteration 

43 trading_data = environment.get_trading_data() 

44 current_assets = trading_data.current_budget + trading_data.currently_invested 

45 iterations.append(current_iteration) 

46 assets_values.append(current_assets) 

47 reward_values.append(0) 

48 

49 while(not done): 

50 next_action = testable_agent.perform(state) 

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

52 

53 iterations.append(environment.current_iteration) 

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

55 reward_values.append(reward) 

56 

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

58 done = True 

59 

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

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

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

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

64 

65 history['assets_values'] = assets_values 

66 history['reward_values'] = reward_values 

67 history['currency_prices'] = currency_prices 

68 history['iterations'] = iterations 

69 history['solvency_coefficient'] = solvency_coefficient 

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

71 

72 return [self.PLOTTING_KEY], [history]