Coverage for source/environment/broker.py: 100%

1# environment/broker.py 

2 

3# global imports 

4import copy 

5 

6# local imports 

7from source.environment import Order 

8 

9class Broker(): 

10 """ 

11 Responsible for managing and placing orders. It implements 

12 real-environment behavior of orders price actions. 

13 """ 

14 

15 def __init__(self, leverage: int = 1) -> None: 

16 """ 

17 Class constructor. 

18 

19 Parameters: 

20 leverage (int): Coefficient of multiplication used to simulate 

21 leverage trading. 

22 """ 

23 

24 self.__leverage: int = leverage 

25 self.__current_orders: list[Order] = [] 

26 self.__recently_closed_orders: list[Order] = [] 

27 

28 def get_leverage(self) -> int: 

29 """ 

30 Leverage getter. 

31 

32 Returns: 

33 (int): Copy of leverage coefficient. 

34 """ 

35 

36 return copy.copy(self.__leverage) 

37 

38 def get_current_orders(self) -> list[Order]: 

39 """ 

40 Current orders getter. 

41 

42 Returns: 

43 (list[Order]): Copy of the list of currently ongoing trades. 

44 """ 

45 

46 return copy.copy(self.__current_orders) 

47 

48 def place_order(self, amount: float, is_buy_order: bool, stop_loss: float, take_profit: float) -> None: 

49 """ 

50 Creates trade with given parameters and attach it to current broker's orders. 

51 

52 Parameters: 

53 amount (float): Amount of money assigned to order. 

54 is_buy_order (bool): Indicates whether order should be treated as buy (long) 

55 position - when true - or as sell (short) postion - when false 

56 stop_loss (float): Coefficient used to close order when stock behaves contrary 

57 to expectations. 

58 take_profit (float): Coefficient used to close order when stock behaves accordingly 

59 to expectations. 

60 """ 

61 

62 self.__current_orders.append(Order(amount, is_buy_order, stop_loss, take_profit)) 

63 

64 def update_orders(self, coefficient: float) -> list[Order]: 

65 """ 

66 Updates and closes orders. The current value of the order is multiplied by 

67 coefficient and if stop loss or take profit boundaries are crossed, then 

68 the order is closed. 

69 

70 Parameters: 

71 coefficient (float): Multiplyer that current order value is multiplied by. 

72 

73 Returns: 

74 (list[Order]): List of closed trades. 

75 """ 

76 

77 self.__recently_closed_orders.clear() 

78 buy_trade_coefficient = ((coefficient - 1) * self.__leverage) + 1 

79 sell_trade_coefficient = 2 - buy_trade_coefficient 

80 

81 for order in self.__current_orders: 

82 if order.is_buy_order: 

83 order.current_value = order.current_value * buy_trade_coefficient 

84 else: 

85 order.current_value = order.current_value * sell_trade_coefficient 

86 

87 order_ratio = order.current_value / order.initial_value 

88 if order_ratio >= order.take_profit or order_ratio <= order.stop_loss: 

89 self.__recently_closed_orders.append(order) 

90 

91 for order in self.__recently_closed_orders: 

92 self.__current_orders.remove(order) 

93 

94 return self.__recently_closed_orders 

95 

96 def reset(self) -> None: 

97 """ 

98 Resets broker by clearing the currently ongoing and recently closed 

99 lists of orders. 

100 """ 

101 

102 self.__current_orders.clear() 

103 self.__recently_closed_orders.clear()