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

1# environment/broker.py 

2 

3import copy 

4 

5from .order import Order 

6 

7class Broker(): 

8 """ 

9 Responsible for managing and placing orders. It implements 

10 real-environment behavior of orders price actions.  

11 """ 

12 

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

14 """ 

15 Class constructor. 

16 

17 Parameters: 

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

19 leverage trading. 

20 """ 

21 

22 self.__leverage: int = leverage 

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

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

25 

26 def get_leverage(self) -> int: 

27 """ 

28 Leverage getter. 

29 

30 Returns: 

31 (int): Copy of leverage coefficient. 

32 """ 

33 

34 return copy.copy(self.__leverage) 

35 

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

37 """ 

38 Current orders getter. 

39 

40 Returns: 

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

42 """ 

43 

44 return copy.copy(self.__current_orders) 

45 

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

47 """ 

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

49 

50 Parameters: 

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

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

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

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

55 to expectations. 

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

57 to expectations. 

58 """ 

59 

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

61 

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

63 """ 

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

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

66 the order is closed. 

67 

68 Parameters: 

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

70 

71 Returns: 

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

73 """ 

74 

75 self.__recently_closed_orders.clear() 

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

77 sell_trade_coefficient = 2 - buy_trade_coefficient 

78 

79 for order in self.__current_orders: 

80 if order.is_buy_order: 

81 order.current_value = order.current_value * buy_trade_coefficient 

82 else: 

83 order.current_value = order.current_value * sell_trade_coefficient 

84 

85 order_ratio = order.current_value / order.initial_value 

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

87 self.__recently_closed_orders.append(order) 

88 

89 for order in self.__recently_closed_orders: 

90 self.__current_orders.remove(order) 

91 

92 return self.__recently_closed_orders 

93 

94 def reset(self) -> None: 

95 """ 

96 Resets broker by clearing the currently ongoing and recently closed 

97 lists of orders.  

98 """ 

99 

100 self.__current_orders.clear() 

101 self.__recently_closed_orders.clear()