Coverage for django_napse/core/models/accounts/exchange.py: 98%

1import uuid 

2 

3from django.db import models 

4 

5from django_napse.core.models.accounts.managers.exchange import ExchangeAccountManager 

6from django_napse.utils.constants import EXCHANGE_TICKERS 

7from django_napse.utils.errors import ExchangeAccountError 

8from django_napse.utils.findable_class import FindableClass 

9from django_napse.utils.trading.binance_controller import BinanceController 

10 

11 

12class Exchange(models.Model): 

13 name = models.CharField(max_length=200, unique=True) 

14 description = models.TextField() 

15 

16 def __str__(self): 

17 return f"EXCHANGE: {self.name}" 

18 

19 def info(self, verbose=True, beacon=""): 

20 string = "" 

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

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

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

24 if verbose: # pragma: no cover 

25 print(string) 

26 return string 

27 

28 def get_tickers(self): 

29 """Return a list of available tickers on the exchange.""" 

30 return EXCHANGE_TICKERS.get(self.name, []) 

31 

32 

33class ExchangeAccount(models.Model, FindableClass): 

34 uuid = models.UUIDField(unique=True, editable=False, default=uuid.uuid4) 

35 

36 exchange = models.ForeignKey( 

37 "Exchange", 

38 on_delete=models.CASCADE, 

39 related_name="accounts", 

40 ) 

41 name = models.CharField(max_length=200) 

42 testing = models.BooleanField(default=True) 

43 description = models.TextField() 

44 default = models.BooleanField(default=False) 

45 created_at = models.DateTimeField(auto_now_add=True) 

46 

47 objects = ExchangeAccountManager() 

48 

49 def __str__(self): 

50 return f"EXCHANGE ACCOUNT ({self.__class__.__name__}): {self.pk=} - {self.name=} - {self.testing=}" 

51 

52 def info(self, verbose=True, beacon=""): 

53 string = "" 

54 string += f"{beacon}{self.__class__.__name__} ({self.pk=}):\n" 

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

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

57 string += f"{beacon}\t{self.testing=}\n" 

58 string += f"{beacon}\t{self.description=}\n" 

59 exchange_str = self.exchange.info(verbose=False, beacon=f"{beacon}\t") 

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

61 string += f"{beacon}{exchange_str}\n" 

62 if verbose: # pragma: no cover 

63 print(string) 

64 return string 

65 

66 def ping(self): # pragma: no cover 

67 error_msg = f"ping() not implemented for {self.__class__.__name__}" 

68 raise NotImplementedError(error_msg) 

69 

70 def create_client(self): # pragma: no cover 

71 error_msg = f"create_client() not implemented for {self.__class__.__name__}" 

72 raise NotImplementedError(error_msg) 

73 

74 def exchange_controller(self): # pragma: no cover 

75 error_msg = f"exchange_controller() not implemented for {self.__class__.__name__}" 

76 raise NotImplementedError(error_msg) 

77 

78 def get_tickers(self): 

79 return self.exchange.get_tickers() 

80 

81 

82class BinanceAccount(ExchangeAccount): 

83 public_key = models.CharField(max_length=200) 

84 private_key = models.CharField(max_length=200) 

85 

86 class Meta: 

87 unique_together = ("public_key", "private_key") 

88 

89 def __str__(self): 

90 return "BINANCE " + super().__str__() 

91 

92 def ping(self): 

93 request = self.exchange_controller().get_info() 

94 if "error" in request: 

95 error_msg = f"Error pinging {self.exchange.name}. Check that your API keys are correct and have the correct permissions." 

96 raise ExchangeAccountError.APIPermissionError(error_msg) 

97 

98 def exchange_controller(self): 

99 return BinanceController(self.public_key, self.private_key) 

100 

101 

102EXCHANGE_ACCOUNT_DICT = { 

103 "BINANCE": BinanceAccount, 

104}