Coverage for django_napse/core/models/accounts/space.py: 69%

1import uuid 

2from datetime import datetime, timedelta 

3 

4from django.db import models 

5from django.utils.timezone import get_default_timezone 

6 

7from django_napse.core.models.accounts.managers import NapseSpaceManager 

8from django_napse.core.models.bots.bot import Bot 

9from django_napse.core.models.fleets.fleet import Fleet 

10from django_napse.core.models.orders.order import Order 

11from django_napse.core.models.transactions.credit import Credit 

12from django_napse.core.models.transactions.debit import Debit 

13from django_napse.utils.constants import EXCHANGE_TICKERS 

14from django_napse.utils.errors import SpaceError 

15from django_napse.utils.errors.exchange import ExchangeError 

16from django_napse.utils.errors.wallets import WalletError 

17 

18 

19class NapseSpace(models.Model): 

20 """Categorize and manage money. 

21 

22 Attributes: 

23 name: Name of the space. 

24 uuid: Unique identifier of the space. 

25 description: Description of the space. 

26 exchange_account: Exchange account of the space. 

27 created_at: Date of creation of the space. 

28 

29 

30 Examples: 

31 Create a space: 

32 ```python 

33 import django_napse.core.models import NapseSpace, ExchangeAccount 

34 

35 exchange_account: ExchangeAccount = ... 

36 space = NapseSpace.objects.create( 

37 name="Space", 

38 description="Space description", 

39 exchange_account=exchange_account, 

40 ) 

41 ``` 

42 """ 

43 

44 name = models.CharField(max_length=200) 

45 uuid = models.UUIDField( 

46 default=uuid.uuid4, 

47 editable=False, 

48 unique=True, 

49 ) 

50 description = models.TextField() 

51 exchange_account = models.ForeignKey("ExchangeAccount", on_delete=models.CASCADE, related_name="spaces") 

52 created_at = models.DateTimeField(auto_now_add=True) 

53 

54 objects = NapseSpaceManager() 

55 

56 class Meta: 

57 unique_together = ("name", "exchange_account") 

58 

59 def __str__(self): 

60 return f"SPACE: {self.name}" 

61 

62 def info(self, verbose: bool = True, beacon: str = "") -> str: 

63 """Info documentation. 

64 

65 Params: 

66 verbose: Print to console. 

67 beacon: Indentation for printing. 

68 """ 

69 string = "" 

70 string += f"{beacon}Space ({self.pk=}):\n" 

71 string += f"{beacon}Args:\n" 

72 string += f"{beacon}\t{self.name=}\n" 

73 string += f"{beacon}\t{self.uuid=}\n" 

74 string += f"{beacon}Exchange Account:\n" 

75 new_beacon = beacon + "\t" 

76 string += f"{self.exchange_account.info(verbose=False, beacon=new_beacon)}\n" 

77 

78 if verbose: # pragma: no cover 

79 print(string) 

80 return string 

81 

82 @property 

83 def testing(self) -> bool: 

84 """Testing property.""" 

85 return self.exchange_account.testing 

86 

87 @property 

88 def value(self) -> float: 

89 """Value market of space's wallet.""" 

90 connections = self.wallet.connections.all() 

91 return sum([connection.wallet.value_market() for connection in connections]) 

92 

93 @property 

94 def fleets(self) -> models.QuerySet: 

95 """Fleets of the space.""" 

96 connections = self.wallet.connections.all() 

97 return Fleet.objects.filter(clusters__links__bot__connections__in=connections).distinct() 

98 

99 @property 

100 def bots(self) -> models.QuerySet: 

101 """Bots of the space.""" 

102 connections = self.wallet.connections.all() 

103 return Bot.objects.filter(connections__in=connections) 

104 

105 def get_stats(self) -> dict[str, int | float | str]: 

106 """Statistics of space.""" 

107 order_count_30 = Order.objects.filter( 

108 connection__in=self.wallet.connections.all(), 

109 created_at__gt=datetime.now(tz=get_default_timezone()) - timedelta(days=30), 

110 ).count() 

111 

112 return { 

113 "value": self.value, 

114 "order_count_30": order_count_30, 

115 "delta_30": 0, # need history on space's value 

116 } 

117 

118 def delete(self) -> None: 

119 """Delete space.""" 

120 if self.testing: 

121 return super().delete() 

122 if self.value > 0: 

123 raise SpaceError.DeleteError 

124 return super().delete() 

125 

126 def invest(self, amount: float, ticker: str): 

127 """Invest in space.""" 

128 if ticker not in EXCHANGE_TICKERS.get("BINANCE"): 

129 error_msg: str = f"{ticker} is not available on {self.exchange_account.name} exchange." 

130 raise ExchangeError.UnavailableTicker(error_msg) 

131 

132 # Real invest 

133 if not self.testing: 

134 error_msg: str = "Investing for real is not available yet." 

135 raise NotImplementedError(error_msg) 

136 

137 # Testing invest 

138 Credit.objects.create(wallet=self.wallet, amount=amount, ticker=ticker) 

139 

140 def withdraw(self, amount: float, ticker: str): 

141 """Withdraw from space.""" 

142 if ticker not in [currency.ticker for currency in self.wallet.currencies.all()]: 

143 error_msg: str = f"{ticker} is not on your {self.name}(space)'s wallet." 

144 raise WalletError.UnavailableTicker(error_msg) 

145 

146 # Real withdraw 

147 if not self.testing: 

148 error_msg: str = "Withdrawing for real is not available yet." 

149 raise NotImplementedError(error_msg) 

150 

151 # Testing withdraw 

152 Debit.objects.create( 

153 wallet=self.wallet, 

154 amount=amount, 

155 ticker=ticker, 

156 )