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

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 

27 def get_leverage(self) -> int: 

28 """ 

29 Leverage getter. 

30 

31 Returns: 

32 (int): Copy of leverage coefficient. 

33 """ 

34 

35 return copy.copy(self.__leverage) 

36 

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

38 """ 

39 Current orders getter. 

40 

41 Returns: 

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

43 """ 

44 

45 return copy.copy(self.__current_orders) 

46 

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

48 """ 

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

50 

51 Parameters: 

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

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

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

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

56 to expectations. 

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

58 to expectations. 

59 """ 

60 

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

62 

63 def force_close_orders(self) -> list[Order]: 

64 """ 

65 Closes all currently ongoing orders and returns them. 

66 

67 Returns: 

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

69 """ 

70 

71 current_orders_copy = copy.copy(self.__current_orders) 

72 self.__current_orders.clear() 

73 

74 return current_orders_copy 

75 

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

77 """ 

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

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

80 the order is closed. 

81 

82 Parameters: 

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

84 

85 Returns: 

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

87 """ 

88 

89 recently_closed_orders = [] 

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

91 sell_trade_coefficient = 2 - buy_trade_coefficient 

92 

93 for order in self.__current_orders: 

94 if order.is_buy_order: 

95 order.current_value = order.current_value * buy_trade_coefficient 

96 else: 

97 order.current_value = order.current_value * sell_trade_coefficient 

98 

99 order_ratio = order.current_value / order.initial_value 

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

101 recently_closed_orders.append(order) 

102 

103 for order in recently_closed_orders: 

104 self.__current_orders.remove(order) 

105 

106 return recently_closed_orders 

107 

108 def reset(self) -> None: 

109 """ 

110 Resets broker by clearing the currently ongoing and recently closed 

111 lists of orders. 

112 """ 

113 

114 self.__current_orders.clear()