feat(histories): added WalletHistories

django_napse/core/models/bots/architecture.py

@@ -3,6 +3,7 @@
 from django.db import models
 
 from django_napse.core.models.bots.managers import ArchitectureManager
+from django_napse.core.models.wallets.currency import CurrencyPydantic
 from django_napse.utils.constants import ORDER_LEEWAY_PERCENTAGE, PLUGIN_CATEGORIES, SIDES
 from django_napse.utils.errors.orders import OrderError
 from django_napse.utils.findable_class import FindableClass
@@ -131,12 +132,14 @@ def _get_orders(self, data: dict, no_db_data: Optional[dict] = None) -> list[dic
                     error_msg = f"Order on {order['pair']} has a side of {order['side']} but an amount of 0."
                     raise OrderError.ProcessError(error_msg)
             for ticker, amount in required_amount.items():
-                if amount > no_db_data["connection_data"][connection]["wallet"]["currencies"].get(ticker, {"amount": 0})["amount"] / (
-                    1 + (ORDER_LEEWAY_PERCENTAGE + 1) / 100
-                ):
-                    available = no_db_data["connection_data"][connection]["wallet"]["currencies"].get(ticker, {"amount": 0})["amount"] / (
-                        1 + (ORDER_LEEWAY_PERCENTAGE + 1) / 100
-                    )
+                if amount > no_db_data["connection_data"][connection]["wallet"].currencies.get(
+                    ticker,
+                    CurrencyPydantic(ticker=ticker, amount=0, mbp=0),
+                ).amount / (1 + (ORDER_LEEWAY_PERCENTAGE + 1) / 100):
+                    available = no_db_data["connection_data"][connection]["wallet"].currencies.get(
+                        ticker,
+                        CurrencyPydantic(ticker=ticker, amount=0, mbp=0),
+                    ).amount / (1 + (ORDER_LEEWAY_PERCENTAGE + 1) / 100)
                     for order in [_order for _order in orders if _order["asked_for_ticker"] == ticker]:
                         order["asked_for_amount"] *= available / amount
             all_orders += orders

django_napse/core/models/bots/bot.py

@@ -1,16 +1,19 @@
 from __future__ import annotations
 
 import uuid
-from typing import Optional
+from typing import TYPE_CHECKING, Optional
 
 from django.db import models
 
 from django_napse.core.models.connections.connection import Connection
 from django_napse.core.models.modifications import ArchitectureModification, ConnectionModification, StrategyModification
 from django_napse.core.models.orders.order import Order, OrderBatch
-from django_napse.core.models.wallets.wallet import SpaceSimulationWallet, SpaceWallet
 from django_napse.utils.errors import BotError
 
+if TYPE_CHECKING:
+    from django_napse.core.models.wallets.space_simulation_wallet import SpaceSimulationWallet
+    from django_napse.core.models.wallets.space_wallet import SpaceWallet
+
 
 class Bot(models.Model):
     uuid = models.UUIDField(unique=True, editable=False, default=uuid.uuid4)

django_napse/core/models/bots/implementations/turbo_dca/strategy.py

@@ -6,6 +6,7 @@
 from django_napse.core.models.bots.implementations.turbo_dca.config import TurboDCABotConfig
 from django_napse.core.models.bots.plugins import LBOPlugin, MBPPlugin, SBVPlugin
 from django_napse.core.models.bots.strategy import Strategy
+from django_napse.core.models.wallets.currency import CurrencyPydantic
 from django_napse.utils.constants import SIDES
 
 
@@ -46,8 +47,16 @@ def give_order(self, data: dict) -> list[dict]:
             mbp = data["connection_data"][data["connection"]]["connection_specific_args"]["mbp"].get_value()
             lbo = data["connection_data"][data["connection"]]["connection_specific_args"]["lbo"].get_value()
             sbv = data["connection_data"][data["connection"]]["connection_specific_args"]["sbv"].get_value()
-            available_base = data["connection_data"][data["connection"]]["wallet"]["currencies"].get(controller.base, {"amount": 0})["amount"]
-            available_quote = data["connection_data"][data["connection"]]["wallet"]["currencies"].get(controller.quote, {"amount": 0})["amount"]
+            available_base = (
+                data["connection_data"][data["connection"]]["wallet"]
+                .currencies.get(controller.base, CurrencyPydantic(ticker=controller.base, amount=0, mbp=0))
+                .amount
+            )
+            available_quote = (
+                data["connection_data"][data["connection"]]["wallet"]
+                .currencies.get(controller.quote, CurrencyPydantic(ticker=controller.quote, amount=0, mbp=0))
+                .amount
+            )
             mbp = mbp if mbp is not None else math.inf
             sbv = sbv if sbv is not None else available_quote
             current_price = data["candles"][controller]["latest"]["close"]

django_napse/core/models/bots/plugins/sbv.py

@@ -1,5 +1,6 @@
 from django_napse.core.models.bots.plugin import Plugin
 from django_napse.core.models.connections.connection import ConnectionSpecificArgs
+from django_napse.core.models.wallets.currency import CurrencyPydantic
 from django_napse.utils.constants import PLUGIN_CATEGORIES, SIDES
 
 
@@ -10,12 +11,16 @@ def plugin_category(cls):
 
     def _apply(self, data: dict) -> dict:
         order = data["order"]
-        current_base_amout = data["connection_data"][data["connection"]]["wallet"]["currencies"].get(order["controller"].base, {"amount": 0})[
-            "amount"
-        ]
-        current_quote_amout = data["connection_data"][data["connection"]]["wallet"]["currencies"].get(order["controller"].quote, {"amount": 0})[
-            "amount"
-        ]
+        current_base_amout = (
+            data["connection_data"][data["connection"]]["wallet"]
+            .currencies.get(order["controller"].base, CurrencyPydantic(ticker=order["controller"].base, amount=0, mbp=0))
+            .amount
+        )
+        current_quote_amout = (
+            data["connection_data"][data["connection"]]["wallet"]
+            .currencies.get(order["controller"].quote, CurrencyPydantic(ticker=order["controller"].quote, amount=0, mbp=0))
+            .amount
+        )
         if data["connection_data"][data["connection"]]["connection_specific_args"]["sbv"].get_value() is None or order["side"] == SIDES.SELL:
             order["ConnectionModifications"] += [
                 {

django_napse/core/models/connections/connection.py

@@ -1,10 +1,15 @@
+from typing import TYPE_CHECKING
+
 from django.db import models
 
 from django_napse.core.models.connections.managers import ConnectionManager
 from django_napse.core.models.transactions.transaction import Transaction
 from django_napse.utils.constants import TRANSACTION_TYPES
 from django_napse.utils.usefull_functions import process_value_from_type
 
+if TYPE_CHECKING:
+    from django_napse.core.models.accounts.space import NapseSpace
+
 
 class Connection(models.Model):
     """Link between a bot & a wallet."""
@@ -53,7 +58,7 @@ def testing(self) -> bool:
         return self.space.testing
 
     @property
-    def space(self) -> "NapseSpace":  # noqa: F821
+    def space(self) -> "NapseSpace":
         """Return the space of the connection."""
         return self.owner.find().space
 

django_napse/core/models/histories/bot.py

@@ -4,4 +4,16 @@
 
 
 class BotHistory(History):
+    """A History for a Bot.
+
+    Use it to track the evolution of a bot over time.
+
+    This tracks the following fields:
+    TODO
+    """
+
     owner = models.OneToOneField("Bot", on_delete=models.CASCADE, related_name="history")
+
+    def generate_data_points(self) -> None:
+        """Create a new data point for the bot."""
+        return

django_napse/core/models/histories/history.py

@@ -5,19 +5,39 @@
 from django.db import models
 from django.utils.timezone import get_default_timezone
 
-from django_napse.utils.constants import HISTORY_DATAPOINT_FIELDS
+from django_napse.core.models.histories.managers.history_data_point import HistoryDataPointManager
+from django_napse.core.models.histories.managers.history_data_point_field import HistoryDataPointFieldManager
+from django_napse.utils.constants import HISTORY_DATAPOINT_FIELDS, HISTORY_DATAPOINT_FIELDS_WILDCARDS
 from django_napse.utils.errors import HistoryError
 from django_napse.utils.findable_class import FindableClass
 from django_napse.utils.usefull_functions import process_value_from_type
 
 
 class History(FindableClass, models.Model):
+    """A History is a collection of data points.
+
+    Use it to track the evolution of a value over time.
+
+    Create a child class to get started.
+
+    The child class should have a ForeignKey to the model you want to track (called `owner`).
+    """
+
     uuid = models.UUIDField(unique=True, editable=False, default=uuid.uuid4)
 
     def __str__(self) -> str:
         return f"HISTORY {self.uuid}"
 
-    def info(self, verbose=True, beacon=""):
+    def info(self, beacon: str = "", *, verbose: bool = True) -> str:
+        """Return a string with the model information.
+
+        Args:
+            beacon (str, optional): The prefix for each line. Defaults to "".
+            verbose (bool, optional): Whether to print the string. Defaults to True.
+
+        Returns:
+            str: The string with the history information.
+        """
         string = ""
         string += f"{beacon}History {self.pk}:\n"
         string += f"{beacon}\t{self.uuid=}\n"
@@ -28,23 +48,25 @@ def info(self, verbose=True, beacon=""):
             print(string)
         return string
 
-    def to_dataframe(self):
+    def to_dataframe(self) -> pd.DataFrame:
+        """Return a DataFrame containing the data points."""
         all_data_points = self.data_points.all()
         return pd.DataFrame([data_point.to_dict() for data_point in all_data_points])
 
     @property
-    def owner(self):
+    def owner(self) -> models.Model:
+        """Return the owner of the history."""
         return self.find().owner
 
     @classmethod
-    def get_or_create(cls, owner):
+    def get_or_create(cls, owner: models.Model) -> "History":
+        """Return the history of the owner if it exists, else create it."""
         if hasattr(owner, "history"):
             return owner.history
         return cls.objects.create(owner=owner)
 
     def delta(self, days: int = 30) -> float:
         """Return the value delta between today and {days} days ago."""
-        # TODO: TEST IT
         date = datetime.now(tz=get_default_timezone()) - timedelta(days=days)
         data_points = self.data_points.filter(created_at__date=date.date())
 
@@ -59,32 +81,93 @@ def delta(self, days: int = 30) -> float:
 
         return data_points
 
+    def generate_data_point(self) -> "HistoryDataPoint":
+        """Create a new data point.
+
+        This method should be implemented in the child class.
+        """
+        error_msg = "You must implement the generate_data_point method in your child class."
+        raise NotImplementedError(error_msg)
+
 
 class HistoryDataPoint(models.Model):
+    """A HistoryDataPoint is a collection of fields."""
+
     history = models.ForeignKey(History, on_delete=models.CASCADE, related_name="data_points")
     created_at = models.DateTimeField(auto_now_add=True)
 
+    objects = HistoryDataPointManager()
+
     def __str__(self) -> str:  # pragma: no cover
         return f"HISTORY DATA POINT {self.pk} {self.history.uuid}"
 
-    def to_dict(self):
+    def to_dict(self) -> dict:
+        """Return a dictionary containing the fields."""
         return {field.key: field.get_value() for field in self.fields.all()}
 
+    def info(self, beacon: str = "", *, verbose: bool = True) -> str:
+        """Return a string with the model information.
+
+        Args:
+            beacon (str, optional): The prefix for each line. Defaults to "".
+            verbose (bool, optional): Whether to print the string. Defaults to True.
+
+        Returns:
+            str: The string with the history information.
+        """
+        string = ""
+        string += f"{beacon}HistoryDataPoint {self.pk}:\n"
+        string += f"{beacon}Fields:\n"
+        for field in self.fields.all():
+            string += field.info(beacon=beacon + "\t", verbose=False)
+        if verbose:
+            print(string)
+        return string
+
 
 class HistoryDataPointField(models.Model):
+    """A HistoryDataPointField is a key-value pair with a target type."""
+
     history_data_point = models.ForeignKey(HistoryDataPoint, on_delete=models.CASCADE, related_name="fields")
     key = models.CharField(max_length=255)
     value = models.CharField(max_length=255)
     target_type = models.CharField(max_length=255)
 
+    objects = HistoryDataPointFieldManager()
+
     def __str__(self) -> str:  # pragma: no cover
         return f"HISTORY DATA POINT FIELD {self.pk} {self.history_data_point.pk}"
 
-    def save(self, *args, **kwargs):
-        if self.key not in HISTORY_DATAPOINT_FIELDS:
+    def save(self, *args, **kwargs):  # noqa
+        is_wildcard = False
+        for wildcard in HISTORY_DATAPOINT_FIELDS_WILDCARDS:
+            if self.key.startswith(wildcard):
+                is_wildcard = True
+                break
+        if self.key not in HISTORY_DATAPOINT_FIELDS and not is_wildcard:
             error_msg = f"Invalid key {self.key} for HistoryDataPointField"
             raise HistoryError.InvalidDataPointFieldKey(error_msg)
         return super().save(*args, **kwargs)
 
-    def get_value(self):
+    def info(self, beacon: str = "", *, verbose: bool = True) -> str:
+        """Return a string with the model information.
+
+        Args:
+            beacon (str, optional): The prefix for each line. Defaults to "".
+            verbose (bool, optional): Whether to print the string. Defaults to True.
+
+        Returns:
+            str: The string with the history information.
+        """
+        string = ""
+        string += f"{beacon}HistoryDataPointField {self.pk}:\n"
+        string += f"{beacon}\t{self.key=}\n"
+        string += f"{beacon}\t{self.value=}\n"
+        string += f"{beacon}\t{self.target_type=}\n"
+        if verbose:
+            print(string)
+        return string
+
+    def get_value(self) -> any:
+        """Return the value as the target type."""
         return process_value_from_type(value=self.value, target_type=self.target_type)

django_napse/core/models/histories/managers/__init__.py

@@ -0,0 +1 @@
+from .history_data_point import HistoryDataPointManager

django_napse/core/models/histories/managers/history_data_point.py

@@ -0,0 +1,25 @@
+from typing import TYPE_CHECKING, Optional
+
+from django.apps import apps
+from django.db import models
+
+if TYPE_CHECKING:
+    from django_napse.core.models.histories.history import History, HistoryDataPoint
+
+
+class HistoryDataPointManager(models.Manager):
+    """The manager for the HistoryDataPoint model."""
+
+    def create(
+        self,
+        history: "History",
+        points: Optional[dict] = None,
+    ) -> "HistoryDataPoint":
+        """Create a new data point for the history."""
+        HistoryDataPointField = apps.get_model("django_napse_core", "HistoryDataPointField")
+        points = points or {}
+        data_point = self.model(history=history)
+        data_point.save()
+        for key, value in points.items():
+            HistoryDataPointField.objects.create(history_data_point=data_point, key=key, value=value)
+        return data_point

django_napse/core/models/histories/managers/history_data_point_field.py

@@ -0,0 +1,21 @@
+from typing import TYPE_CHECKING
+
+from django.db import models
+
+if TYPE_CHECKING:
+    from django_napse.core.models.histories.history import HistoryDataPoint, HistoryDataPointField
+
+
+class HistoryDataPointFieldManager(models.Manager):
+    """The manager for the HistoryDataPointField model."""
+
+    def create(
+        self,
+        history_data_point: "HistoryDataPoint",
+        key: str,
+        value: str,
+    ) -> "HistoryDataPointField":
+        """Create a new data point field for a data point."""
+        data_point_field = self.model(history_data_point=history_data_point, key=key, value=str(value), target_type=type(value).__name__)
+        data_point_field.save()
+        return data_point_field