Coverage for django_napse/core/models/bots/architecture.py: 71%

1from typing import Optional 

2 

3from django.db import models 

4 

5from django_napse.core.models.bots.managers import ArchitectureManager 

6from django_napse.utils.constants import ORDER_LEEWAY_PERCENTAGE, PLUGIN_CATEGORIES, SIDES 

7from django_napse.utils.errors.orders import OrderError 

8from django_napse.utils.findable_class import FindableClass 

9 

10 

11class Architecture(models.Model, FindableClass): 

12 """Define how the bot will get data.""" 

13 

14 objects = ArchitectureManager() 

15 

16 def __str__(self) -> str: 

17 return f"ARCHITECHTURE {self.pk}" 

18 

19 @property 

20 def variables(self) -> dict[str, any]: 

21 """Returns variables of the architecture.""" 

22 self = self.find(self.pk) 

23 variables = {} 

24 for variable in self._meta.get_fields(): 

25 if variable.name.startswith("variable_"): 

26 variables[variable.name[8:]] = getattr(self, variable.name) 

27 return variables 

28 

29 def get_candles(self): # pragma: no cover # noqa: ANN201, D102 

30 if self.__class__ == Architecture: 

31 error_msg = "get_candles not implemented for the Architecture base class, please implement it in a subclass." 

32 else: 

33 error_msg = f"get_candles not implemented for the Architecture base class, please implement it in the {self.__class__} class." 

34 raise NotImplementedError(error_msg) 

35 

36 def copy(self): # pragma: no cover # noqa: ANN201, D102 

37 if self.__class__ == Architecture: 

38 error_msg = "copy not implemented for the Architecture base class, please implement it in a subclass." 

39 else: 

40 error_msg = f"copy not implemented for the Architecture base class, please implement it in the {self.__class__} class." 

41 raise NotImplementedError(error_msg) 

42 

43 def controllers_dict(self): # pragma: no cover # noqa: ANN201, D102 

44 if self.__class__ == Architecture: 

45 error_msg = "controllers_dict not implemented for the Architecture base class, please implement it in a subclass." 

46 else: 

47 error_msg = f"controllers_dict not implemented for the Architecture base class, please implement it in the {self.__class__} class." 

48 raise NotImplementedError(error_msg) 

49 

50 def accepted_tickers(self): # pragma: no cover # noqa: ANN201, D102 

51 if self.__class__ == Architecture: 

52 error_msg = "accepted_tickers not implemented for the Architecture base class, please implement it in a subclass." 

53 else: 

54 error_msg = f"accepted_tickers not implemented for the Architecture base class, please implement it in the {self.__class__} class." 

55 raise NotImplementedError(error_msg) 

56 

57 def accepted_investment_tickers(self): # pragma: no cover # noqa: ANN201, D102 

58 if self.__class__ == Architecture: 

59 error_msg = "accepted_investment_tickers not implemented for the Architecture base class, please implement it in a subclass." 

60 else: 

61 error_msg = ( 

62 f"accepted_investment_tickers not implemented for the Architecture base class, please implement it in the {self.__class__} class." 

63 ) 

64 raise NotImplementedError(error_msg) 

65 

66 def get_extras(self): # noqa 

67 return {} 

68 

69 def skip(self, data: dict) -> bool: # noqa 

70 return False 

71 

72 def strategy_modifications(self, order: dict, data) -> list[dict]: # noqa 

73 """Return modifications.""" 

74 return [] 

75 

76 def connection_modifications(self, order: dict, data) -> list[dict]: # noqa 

77 """Return modifications.""" 

78 return [] 

79 

80 def architecture_modifications(self, order: dict, data) -> list[dict]: # noqa 

81 """Return modifications.""" 

82 return [] 

83 

84 def prepare_data(self) -> dict[str, dict[str, any]]: 

85 """Return candles data.""" 

86 return { 

87 "candles": {controller: self.get_candles(controller) for controller in self.controllers_dict().values()}, 

88 "extras": self.get_extras(), 

89 } 

90 

91 def prepare_db_data(self) -> dict[str, any]: 

92 """Return the data that is needed to give orders.""" 

93 return { 

94 "strategy": self.strategy.find(), 

95 "config": self.strategy.find().config.find().settings, 

96 "architecture": self.find(), 

97 "controllers": self.controllers_dict(), 

98 "connections": self.strategy.bot.get_connections(), 

99 "connection_data": self.strategy.bot.get_connection_data(), 

100 "plugins": {category: self.strategy.plugins.filter(category=category) for category in PLUGIN_CATEGORIES}, 

101 } 

102 

103 def _get_orders(self, data: dict, no_db_data: Optional[dict] = None) -> list[dict]: 

104 data = data or self.prepare_data() 

105 no_db_data = no_db_data or self.prepare_db_data() 

106 strategy = no_db_data["strategy"] 

107 connections = no_db_data["connections"] 

108 architecture = no_db_data["architecture"] 

109 all_orders = [] 

110 for connection in connections: 

111 new_data = {**data, **no_db_data, "connection": connection} 

112 if architecture.skip(data=new_data): 

113 continue 

114 for plugin in no_db_data["plugins"][PLUGIN_CATEGORIES.PRE_ORDER]: 

115 plugin.apply(data=new_data) 

116 orders = strategy.give_order(data=new_data) 

117 for order in orders: 

118 for plugin in no_db_data["plugins"][PLUGIN_CATEGORIES.POST_ORDER]: 

119 plugin.apply(data={**new_data, "order": order}) 

120 order["StrategyModifications"] += architecture.strategy_modifications(order=order, data=new_data) 

121 order["ConnectionModifications"] += architecture.connection_modifications(order=order, data=new_data) 

122 order["ArchitectureModifications"] += architecture.architecture_modifications(order=order, data=new_data) 

123 required_amount = {} 

124 for order in orders: 

125 required_amount[order["asked_for_ticker"]] = required_amount.get(order["asked_for_ticker"], 0) + order["asked_for_amount"] 

126 

127 if order["side"] == SIDES.KEEP and order["asked_for_amount"] != 0: 

128 error_msg = f"Order on {order['pair']} has a side of KEEP but an amount of {order['asked_for_amount']}." 

129 raise OrderError.ProcessError(error_msg) 

130 if order["side"] != SIDES.KEEP and order["asked_for_amount"] == 0: 

131 error_msg = f"Order on {order['pair']} has a side of {order['side']} but an amount of 0." 

132 raise OrderError.ProcessError(error_msg) 

133 for ticker, amount in required_amount.items(): 

134 if amount > no_db_data["connection_data"][connection]["wallet"]["currencies"].get(ticker, {"amount": 0})["amount"] / ( 

135 1 + (ORDER_LEEWAY_PERCENTAGE + 1) / 100 

136 ): 

137 available = no_db_data["connection_data"][connection]["wallet"]["currencies"].get(ticker, {"amount": 0})["amount"] / ( 

138 1 + (ORDER_LEEWAY_PERCENTAGE + 1) / 100 

139 ) 

140 for order in [_order for _order in orders if _order["asked_for_ticker"] == ticker]: 

141 order["asked_for_amount"] *= available / amount 

142 all_orders += orders 

143 return all_orders