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 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 

29 Returns: 

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

31 """ 

32 

33 history = {} 

34 assets_values = [] 

35 reward_values = [] 

36 infos = [] 

37 iterations = [] 

38 done = False 

39 

40 state = environment.state 

41 current_iteration = environment.current_iteration 

42 trading_data = environment.get_trading_data() 

43 current_assets = trading_data.current_budget + trading_data.currently_invested 

44 iterations.append(current_iteration) 

45 assets_values.append(current_assets) 

46 reward_values.append(0) 

47 infos.append({}) 

48 

49 while(not done): 

50 next_action = testable_agent.perform(state) 

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

52 

53 if current_assets != info['current_budget'] + info['currently_invested'] or done: 

54 current_iteration = environment.current_iteration 

55 current_assets = info['current_budget'] + info['currently_invested'] 

56 iterations.append(current_iteration) 

57 assets_values.append(current_assets) 

58 reward_values.append(reward) 

59 infos.append(info) 

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['infos'] = infos 

70 history['iterations'] = iterations 

71 history['solvency_coefficient'] = solvency_coefficient 

72 

73 return [self.PLOTTING_KEY], [history]