Coverage for django_napse/core/models/bots/bot.py: 72%
139 statements
« prev ^ index » next coverage.py v7.4.3, created at 2024-03-12 13:49 +0000
1from __future__ import annotations
3import uuid
4from typing import Optional
6from django.db import models
8from django_napse.core.models.connections.connection import Connection
9from django_napse.core.models.modifications import ArchitectureModification, ConnectionModification, StrategyModification
10from django_napse.core.models.orders.order import Order, OrderBatch
11from django_napse.core.models.wallets.wallet import SpaceSimulationWallet, SpaceWallet
12from django_napse.utils.errors import BotError
15class Bot(models.Model):
16 uuid = models.UUIDField(unique=True, editable=False, default=uuid.uuid4)
18 name = models.CharField(max_length=100, default="Bot")
20 created_at = models.DateTimeField(auto_now_add=True)
21 active = models.BooleanField(default=True)
23 strategy = models.OneToOneField("Strategy", on_delete=models.CASCADE, related_name="bot")
25 def __str__(self):
26 return f"BOT {self.pk=}"
28 def save(self, *args, **kwargs):
29 if self.is_in_simulation and self.is_in_fleet:
30 error_msg = "Bot is in simulation and fleet."
31 raise BotError.InvalidSetting(error_msg)
32 return super().save(*args, **kwargs)
34 def info(self, verbose=True, beacon=""):
35 string = ""
36 string += f"{beacon}Bot {self.pk}:\n"
37 string += f"{beacon}Args:\n"
38 string += f"{beacon}\t{self.name=}\n"
40 string += f"{beacon}Strategy:\n"
41 new_beacon = beacon + "\t"
42 string += f"{beacon}\t{self.strategy.info(verbose=False, beacon=new_beacon)}\n"
44 if verbose: # pragma: no cover
45 print(string)
46 return string
48 @property
49 def is_in_simulation(self):
50 return hasattr(self, "simulation")
52 @property
53 def is_in_fleet(self):
54 # return hasattr(self, "bot_in_cluster")
55 return hasattr(self, "link")
57 @property
58 def is_templace(self):
59 """Is self a template bot of a cluster?"""
60 return hasattr(self, "cluster")
62 @property
63 def is_free(self):
64 """Is self not in a simulation, not in a fleet and not a cluster's template bot?"""
65 return not self.is_in_simulation and not self.is_in_fleet and not self.is_templace
67 @property
68 def testing(self):
69 if self.is_in_simulation:
70 return True
71 if self.is_in_fleet:
72 return self.bot_in_cluster.cluster.fleet.testing
73 error_msg = "Bot is not in simulation or fleet."
74 raise BotError.InvalidSetting(error_msg)
76 @property
77 def fleet(self):
78 if self.is_in_fleet:
79 return self.link.cluster.fleet
80 return None
82 @property
83 def space(self):
84 if self.is_in_simulation:
85 return self.simulation.space
86 if self.is_in_fleet:
87 error_msg = "Bot is in a fleet and therefore doesn't have a space."
88 raise BotError.NoSpace(error_msg)
89 error_msg = "Bot is not in simulation or fleet."
90 raise BotError.InvalidSetting(error_msg)
92 @property
93 def exchange_account(self):
94 if self.is_in_simulation:
95 return self.simulation.space.exchange_account
96 if self.is_in_fleet:
97 return self.link.cluster.fleet.exchange_account
98 error_msg = "Bot is not in simulation or fleet."
99 raise BotError.InvalidSetting(error_msg)
101 @property
102 def _strategy(self):
103 return self.strategy.find()
105 @property
106 def architecture(self):
107 return self._strategy.architecture.find()
109 @property
110 def controllers(self):
111 return self.architecture.controllers_dict()
113 @property
114 def orders(self):
115 connections = self.connections.select_related("orders").all()
116 return [connection.orders for connection in connections]
118 def hibernate(self):
119 if not self.active:
120 error_msg = "Bot is already hibernating."
121 raise BotError.InvalidSetting(error_msg)
122 self.active = False
123 self.save()
125 def get_connections(self):
126 return list(self.connections.all())
128 def get_connection_data(self):
129 return {connection: connection.to_dict() for connection in self.get_connections()}
131 def get_orders(self, data: Optional[dict] = None, no_db_data: Optional[dict] = None):
132 if not self.active:
133 error_msg = "Bot is hibernating."
134 raise BotError.InvalidSetting(error_msg)
136 orders = self._get_orders(data=data, no_db_data=no_db_data)
137 batches = {}
138 order_objects = []
139 for order in orders:
140 controller = order["controller"]
141 batches[controller] = OrderBatch.objects.create(controller=controller)
142 for order_dict in orders:
143 controller = order_dict.pop("controller")
144 strategy_modifications = order_dict.pop("StrategyModifications")
145 connection_modifications = order_dict.pop("ConnectionModifications")
146 architecture_modifications = order_dict.pop("ArchitectureModifications")
147 order = Order.objects.create(batch=batches[controller], **order)
148 order_objects.append(order)
149 for modification in strategy_modifications:
150 StrategyModification.objects.create(order=order, **modification)
151 for modification in connection_modifications:
152 ConnectionModification.objects.create(order=order, **modification)
153 for modification in architecture_modifications:
154 ArchitectureModification.objects.create(order=order, **modification)
155 for batch in batches.values():
156 batch.set_status_ready()
158 return order_objects, batches
160 def _get_orders(self, data: Optional[dict] = None, no_db_data: Optional[dict] = None):
161 """Get orders of the bot."""
162 return self.architecture._get_orders(data=data, no_db_data=no_db_data) # noqa: SLF001
164 def connect_to_wallet(self, wallet: SpaceSimulationWallet | SpaceWallet) -> Connection:
165 """Connect the bot to a (sim)space's wallet."""
166 connection = Connection.objects.create(owner=wallet, bot=self)
167 for plugin in self._strategy.plugins.all():
168 plugin.connect(connection)
169 self._strategy.connect(connection)
170 return connection
172 def copy(self) -> Bot:
173 """Copy the instance bot."""
174 return self.__class__.objects.create(
175 name=f"Copy of {self.name}",
176 strategy=self.strategy.copy(),
177 )
179 def value(self, space=None) -> float: # noqa: ANN001
180 """Return value market of the bot, depending of space containerization."""
181 if space is None:
182 return sum([connection.wallet.value_market() for connection in self.connections.all()])
183 connection = Connection.objects.get(owner=space.wallet, bot=self)
184 return connection.wallet.value_market()
186 def get_stats(self, space=None) -> dict[str, str | int | float]: # noqa: ANN001
187 """Some bot's statistics used for KPI dashboards."""
188 return {
189 "value": self.value(space),
190 "profit": 0,
191 "delta_30": 0, # TODO: Need history
192 }