Coverage for django_napse/simulations/models/simulations/simulation.py: 86%

1from django.db import models 

2 

3from django_napse.simulations.models.simulations.managers import SimulationDataPointManager, SimulationManager 

4from django_napse.utils.usefull_functions import process_value_from_type 

5 

6 

7class Simulation(models.Model): 

8 simulation_reference = models.UUIDField(unique=True, editable=False, null=True) 

9 space = models.ForeignKey("django_napse_core.NapseSpace", on_delete=models.CASCADE, null=True) 

10 bot = models.OneToOneField("django_napse_core.Bot", on_delete=models.CASCADE, null=True, related_name="simulation") 

11 

12 start_date = models.DateTimeField() 

13 end_date = models.DateTimeField() 

14 created_at = models.DateTimeField(auto_now_add=True) 

15 

16 objects = SimulationManager() 

17 

18 def __str__(self) -> str: 

19 return f"SIMULATION {self.pk}" 

20 

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

22 string = "" 

23 string += f"{beacon}Simulation {self.pk}:\n" 

24 string += f"{beacon}\t{self.bot=}\n" 

25 string += f"{beacon}\t{self.space=}\n" 

26 

27 string += f"{beacon}\t{self.start_date=}\n" 

28 string += f"{beacon}\t{self.end_date=}\n" 

29 string += f"{beacon}\t{self.created_at=}\n" 

30 

31 string += f"{beacon}Data points:\n" 

32 data_points = self.data_points.all().order_by("date") 

33 

34 if data_points.count() > 10: 

35 for data_point in data_points[:5]: 

36 string += f"{beacon}\t{data_point.to_str()}\n" 

37 string += f"{beacon}\t...\n" 

38 for data_point in data_points[data_points.count() - 5 :]: 

39 string += f"{beacon}\t{data_point.to_str()}\n" 

40 string += f"{beacon}\t({data_points.count()} data points)\n" 

41 

42 elif data_points.count() > 0: 

43 for data_point in data_points: 

44 string += f"{beacon}\t{data_point.to_str()}\n" 

45 

46 else: 

47 string += f"{beacon}\tNo data points\n" 

48 if verbose: 

49 print(string) 

50 return string 

51 

52 

53class SimulationDataPoint(models.Model): 

54 simulation = models.ForeignKey("Simulation", on_delete=models.CASCADE, related_name="data_points") 

55 date = models.DateTimeField() 

56 value = models.FloatField() 

57 action = models.CharField(max_length=10) 

58 amount = models.FloatField() 

59 ticker = models.CharField(max_length=10) 

60 

61 objects = SimulationDataPointManager() 

62 

63 def __str__(self): 

64 return f"SIMULATION DATA POINT {self.pk}" 

65 

66 def to_dict(self): 

67 extra_info = self.extra_info.all() 

68 return { 

69 "date": self.date, 

70 "value": self.value, 

71 "action": self.action, 

72 "amount": self.amount, 

73 "ticker": self.ticker, 

74 **{info.key: info.get_value() for info in extra_info}, 

75 } 

76 

77 def to_str(self): 

78 string = "" 

79 for key, value in self.to_dict().items(): 

80 string += f"{key}: {value}\t" 

81 return string 

82 

83 

84class SimulationDataPointExtraInfo(models.Model): 

85 data_point = models.ForeignKey("SimulationDataPoint", on_delete=models.CASCADE, related_name="extra_info") 

86 key = models.CharField(max_length=64) 

87 value = models.CharField(max_length=64) 

88 target_type = models.CharField(max_length=64) 

89 

90 def __str__(self): 

91 return f"SIMULATION DATA POINT EXTRA INFO {self.pk}" 

92 

93 def get_value(self): 

94 return process_value_from_type(value=self.value, target_type=self.target_type)