fix(celery): now works asynchronously

Makefile

@@ -32,10 +32,10 @@ up:
 	make makemigrations && make migrate && make runserver
 
 clean:
-	rm tests/test_app/db.sqlite3
+	rm tests/test_app/db.sqlite3 && rm dump.rdb
 
 celery:
-	source .venv/bin/activate && watchfiles --filter python celery.__main__.main --args "-A tests.test_app worker --beat -l INFO" --verbosity info
+	source .venv/bin/activate && watchfiles --filter python celery.__main__.main --args "-A tests.test_app worker --beat --scheduler django -l INFO" --verbosity info
 
 shell:
 	source .venv/bin/activate && python tests/test_app/manage.py shell

django_napse/core/management/commands/create_dca.py

@@ -19,7 +19,7 @@ def handle(self, *args, **options):  # noqa
             exchange_account=exchange_account,
             base="BTC",
             quote="USDT",
-            interval="1m",
+            interval="15m",
         )
         architecture = SinglePairArchitecture.objects.create(constants={"controller": controller})
         strategy = DCAStrategy.objects.create(config=config, architecture=architecture)

django_napse/core/models/bots/architecture.py

@@ -2,18 +2,30 @@
 
 from django.db import models
 
+from django_napse.core.models.bots.controller import Controller
 from django_napse.core.models.bots.managers import ArchitectureManager
-from django_napse.core.models.wallets.currency import CurrencyPydantic
+from django_napse.core.pydantic.candle import CandlePydantic
+from django_napse.core.pydantic.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
 
 if TYPE_CHECKING:
-    from django_napse.core.models.bots.controller import Controller
     from django_napse.core.models.bots.plugin import Plugin
     from django_napse.core.models.bots.strategy import Strategy
     from django_napse.core.models.connections.connection import Connection
-
+DataType = dict[
+    Literal[
+        "candles",
+        "extras",
+    ],
+    Union[
+        dict[
+            Controller,
+            dict[Literal["current", "latest"], Union[CandlePydantic]],
+        ]
+    ],
+]
 DBDataType = dict[
     Literal[
         "strategy",
@@ -91,30 +103,30 @@ def accepted_investment_tickers(self):  # pragma: no cover  # noqa: ANN201, D102
             )
         raise NotImplementedError(error_msg)
 
-    def get_extras(self):  # noqa
+    def get_extras(self):
         return {}
 
-    def skip(self, data: dict) -> bool:  # noqa
+    def skip(self, data: dict) -> bool:
         return False
 
-    def strategy_modifications(self, order: dict, data) -> list[dict]:  # noqa
+    def strategy_modifications(self, order: dict, data) -> list[dict]:  # noqa: ARG002
         """Return modifications."""
         return []
 
-    def connection_modifications(self, order: dict, data) -> list[dict]:  # noqa
+    def connection_modifications(self, order: dict, data) -> list[dict]:  # noqa: ARG002
         """Return modifications."""
         return []
 
-    def architecture_modifications(self, order: dict, data) -> list[dict]:  # noqa
+    def architecture_modifications(self, order: dict, data) -> list[dict]:  # noqa: ARG002
         """Return modifications."""
         return []
 
-    def prepare_data(self) -> dict[str, dict[str, any]]:
-        """Return candles data."""
-        return {
-            "candles": {controller: self.get_candles(controller) for controller in self.controllers_dict().values()},
-            "extras": self.get_extras(),
-        }
+    # def prepare_data(self) -> dict[str, dict[str, any]]:
+    #     """Return candles data."""
+    #     return {
+    #         "candles": {controller: self.get_candles(controller) for controller in self.controllers_dict().values()},
+    #         "extras": self.get_extras(),
+    #     }
 
     def prepare_db_data(
         self,
@@ -130,13 +142,14 @@ def prepare_db_data(
             "plugins": {category: self.strategy.plugins.filter(category=category) for category in PLUGIN_CATEGORIES},
         }
 
-    def _get_orders(self, data: dict, no_db_data: DBDataType = None) -> list[dict]:
-        data = data or self.prepare_data()
+    def get_orders__no_db(self, data: DataType, no_db_data: DBDataType = None) -> list[dict]:
+        # data = data   or self.prepare_data()
         no_db_data = no_db_data or self.prepare_db_data()
         strategy = no_db_data["strategy"]
         connections = no_db_data["connections"]
         architecture = no_db_data["architecture"]
         all_orders = []
+
         for connection in connections:
             new_data = {**data, **no_db_data, "connection": connection}
             if architecture.skip(data=new_data):
@@ -151,6 +164,7 @@ def _get_orders(self, data: dict, no_db_data: DBDataType = None) -> list[dict]:
                 order["ConnectionModifications"] += architecture.connection_modifications(order=order, data=new_data)
                 order["ArchitectureModifications"] += architecture.architecture_modifications(order=order, data=new_data)
             required_amount = {}
+
             for order in orders:
                 required_amount[order["asked_for_ticker"]] = required_amount.get(order["asked_for_ticker"], 0) + order["asked_for_amount"]
 

django_napse/core/models/bots/architectures/single_pair.py

@@ -31,7 +31,7 @@ def controllers_dict(self):
         return {"main": self.controller}
 
     def skip(self, data: dict) -> bool:
-        if data["candles"][data["controllers"]["main"]]["current"]["open_time"] < self.variable_last_candle_date + timedelta(
+        if data["candles"][data["controllers"]["main"]]["current"].open_time < self.variable_last_candle_date + timedelta(
             milliseconds=interval_to_milliseconds(data["controllers"]["main"].interval),
         ):
             return True
@@ -41,7 +41,7 @@ def architecture_modifications(self, order: dict, data: dict):
         return [
             {
                 "key": "last_candle_date",
-                "value": str(data["candles"][data["controllers"]["main"]]["current"]["open_time"]),
+                "value": str(data["candles"][data["controllers"]["main"]]["current"].open_time),
                 "target_type": "datetime",
                 "ignore_failed_order": True,
             },

django_napse/core/models/bots/bot.py

@@ -11,6 +11,8 @@
 from django_napse.utils.errors import BotError
 
 if TYPE_CHECKING:
+    from django_napse.core.models.bots.architecture import Architecture, DataType, DBDataType
+    from django_napse.core.models.bots.strategy import Strategy
     from django_napse.core.models.wallets.space_simulation_wallet import SpaceSimulationWallet
     from django_napse.core.models.wallets.space_wallet import SpaceWallet
 
@@ -23,7 +25,7 @@ class Bot(models.Model):
     created_at = models.DateTimeField(auto_now_add=True)
     active = models.BooleanField(default=True)
 
-    strategy = models.OneToOneField("Strategy", on_delete=models.CASCADE, related_name="bot")
+    strategy: Strategy = models.OneToOneField("Strategy", on_delete=models.CASCADE, related_name="bot")
 
     def __str__(self) -> str:
         return f"BOT {self.pk=}"
@@ -115,7 +117,7 @@ def _strategy(self):
         return self.strategy.find()
 
     @property
-    def architecture(self):
+    def architecture(self) -> Architecture:
         return self._strategy.architecture.find()
 
     @property
@@ -140,12 +142,11 @@ def get_connections(self):
     def get_connection_data(self):
         return {connection: connection.to_dict() for connection in self.get_connections()}
 
-    def get_orders(self, data: Optional[dict] = None, no_db_data: Optional[dict] = None):
+    def get_orders(self, data: DataType, no_db_data: Optional[DBDataType] = None):
         if not self.active:
             error_msg = "Bot is hibernating."
             raise BotError.InvalidSetting(error_msg)
-
-        orders = self._get_orders(data=data, no_db_data=no_db_data)
+        orders = self.get_orders__no_db(data=data, no_db_data=no_db_data)
         batches = {}
         order_objects = []
         for order in orders:
@@ -164,14 +165,15 @@ def get_orders(self, data: Optional[dict] = None, no_db_data: Optional[dict] = N
                 ConnectionModification.objects.create(order=order, **modification)
             for modification in architecture_modifications:
                 ArchitectureModification.objects.create(order=order, **modification)
+
         for batch in batches.values():
             batch.set_status_ready()
 
         return order_objects, batches
 
-    def _get_orders(self, data: Optional[dict] = None, no_db_data: Optional[dict] = None):
+    def get_orders__no_db(self, data: DataType, no_db_data: Optional[DBDataType] = None) -> list[dict]:
         """Get orders of the bot."""
-        return self.architecture._get_orders(data=data, no_db_data=no_db_data)  # noqa: SLF001
+        return self.architecture.get_orders__no_db(data=data, no_db_data=no_db_data)
 
     def connect_to_wallet(self, wallet: SpaceSimulationWallet | SpaceWallet) -> Connection:
         """Connect the bot to a (sim)space's wallet."""

django_napse/core/models/bots/controller.py

@@ -1,10 +1,13 @@
+from __future__ import annotations
+
 import math
 from datetime import datetime, timedelta, timezone
 from typing import TYPE_CHECKING, Optional
 
 from django.db import models
 from requests.exceptions import ConnectionError, ReadTimeout, SSLError
 
+from django_napse.core.models.bots.bot import Bot
 from django_napse.core.models.bots.managers.controller import ControllerManager
 from django_napse.core.models.orders.order import Order, OrderBatch
 from django_napse.utils.constants import EXCHANGE_INTERVALS, EXCHANGE_PAIRS, ORDER_STATUS, SIDES, STABLECOINS
@@ -13,7 +16,8 @@
 
 if TYPE_CHECKING:
     from django_napse.core.models.accounts.exchange import Exchange, ExchangeAccount
-    from django_napse.core.models.bots.bot import Bot
+    from django_napse.core.models.bots.architecture import DataType
+    from django_napse.core.pydantic.candle import CandlePydantic
 
 
 class Controller(models.Model):
@@ -119,6 +123,17 @@ def exchange(self) -> "Exchange":
         """Return the exchange of the controller."""
         return self.exchange_account.exchange
 
+    @property
+    def single_pair_bots(self) -> list["Bot"]:
+        """Return the bots that are allowed to trade on the controller."""
+        bots = []
+
+        for bot in Bot.objects.filter(strategy__architecture__in=self.single_pair_architectures.all()):
+            fleet = bot.fleet
+            if fleet is not None and fleet.running and bot.active and not bot.is_in_simulation:
+                bots.append(bot)
+        return bots
+
     def update_variables(self) -> None:
         """If the variables are older than 1 minute, update them."""
         if self.last_settings_update is None or self.last_settings_update < datetime.now(tz=timezone.utc) - timedelta(minutes=1):
@@ -154,28 +169,37 @@ def update_variables_always(self) -> None:
         if self.pk:
             self.save()
 
-    def process_orders(self, no_db_data: Optional[dict] = None, *, testing: bool) -> list[Order]:
+    def process_orders__no_db(self, no_db_data: Optional[dict] = None, *, testing: bool) -> tuple[list[Order], list[OrderBatch]]:
         in_simulation = no_db_data is not None
         no_db_data = no_db_data or {
-            "buy_orders": Order.objects.filter(
-                order__batch__status=ORDER_STATUS.READY,
-                order__side=SIDES.BUY,
-                order__batch__controller=self,
-                order__testing=testing,
-            ),
-            "sell_orders": Order.objects.filter(
-                order__batch__status=ORDER_STATUS.READY,
-                order__side=SIDES.SELL,
-                order__batch__controller=self,
-                order__testing=testing,
-            ),
-            "keep_orders": Order.objects.filter(
-                order__batch__status=ORDER_STATUS.READY,
-                order__side=SIDES.KEEP,
-                order__batch__controller=self,
-                order__testing=testing,
-            ),
-            "batches": OrderBatch.objects.filter(status=ORDER_STATUS.READY, batch__controller=self),
+            "buy_orders": [
+                order
+                for order in Order.objects.filter(
+                    batch__status=ORDER_STATUS.READY,
+                    side=SIDES.BUY,
+                    batch__controller=self,
+                )
+                if order.testing == testing
+            ],
+            "sell_orders": [
+                order
+                for order in Order.objects.filter(
+                    batch__status=ORDER_STATUS.READY,
+                    side=SIDES.SELL,
+                    batch__controller=self,
+                )
+                if order.testing == testing
+            ],
+            "keep_orders": [
+                order
+                for order in Order.objects.filter(
+                    batch__status=ORDER_STATUS.READY,
+                    side=SIDES.KEEP,
+                    batch__controller=self,
+                )
+                if order.testing == testing
+            ],
+            "batches": OrderBatch.objects.filter(status=ORDER_STATUS.READY, controller=self),
             "exchange_controller": self.exchange_controller,
             "min_trade": self.min_trade,
             "price": self.get_price(),
@@ -203,12 +227,12 @@ def process_orders(self, no_db_data: Optional[dict] = None, *, testing: bool) ->
                 order.calculate_batch_share(total=aggregated_order["sell_amount"])
         for order in no_db_data["keep_orders"]:
             order.batch_share = 0
-
         receipt, executed_amounts_buy, executed_amounts_sell, fees_buy, fees_sell = no_db_data["exchange_controller"].submit_order(
             controller=self,
             aggregated_order=aggregated_order,
             testing=in_simulation or testing,
         )
+
         all_orders = []
         for order in no_db_data["buy_orders"]:
             order.calculate_exit_amounts(
@@ -233,16 +257,23 @@ def process_orders(self, no_db_data: Optional[dict] = None, *, testing: bool) ->
             all_orders.append(order)
 
         for batch in no_db_data["batches"]:
-            batch._set_status_post_process(receipt=receipt)
+            batch.set_status_post_process__no_db(receipt=receipt)
 
-        return all_orders
+        return all_orders, no_db_data["batches"]
 
-    def apply_orders(self, orders):
+    def apply_orders(self, orders: list[Order]) -> None:
         for order in orders:
             order.save()
             order.apply_swap()
 
-    def send_candles_to_bots(self, closed_candle, current_candle) -> list:
+    def apply_batches(self, batches: list[OrderBatch]) -> None:
+        for batch in batches:
+            batch.save()
+
+    def prepare_candles(self, closed_candle: CandlePydantic, current_candle: CandlePydantic) -> DataType:
+        return {"candles": {self: {"current": current_candle, "latest": closed_candle}}, "extras": {}}
+
+    def send_candles_to_bots(self, closed_candle: CandlePydantic, current_candle: CandlePydantic) -> list:
         """Scan all bots (that are allowed to trade) and get their orders.
 
         Args:
@@ -255,9 +286,8 @@ def send_candles_to_bots(self, closed_candle, current_candle) -> list:
         list: A list of orders.
         """
         orders = []
-        for bot in self.bots.all().filter(is_simulation=False, fleet__running=True, can_trade=True):
-            bot: "Bot"
-            orders = [*orders, bot.give_order(closed_candle, current_candle)]
+        for bot in self.single_pair_bots:
+            orders = [*orders, bot.get_orders(data=self.prepare_candles(closed_candle, current_candle))[0]]
         return orders
 
     @staticmethod

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

@@ -34,7 +34,7 @@ def give_order(self, data: dict) -> list[dict]:
         controller = data["controllers"]["main"]
         if (
             self.variable_last_buy_date is None
-            or data["candles"][controller]["current"]["open_time"] - self.variable_last_buy_date >= data["config"]["timeframe"]
+            or data["candles"][controller]["current"].open_time - self.variable_last_buy_date >= data["config"]["timeframe"]
         ):
             return [
                 {
@@ -43,7 +43,7 @@ def give_order(self, data: dict) -> list[dict]:
                     "StrategyModifications": [
                         {
                             "key": "last_buy_date",
-                            "value": str(data["candles"][controller]["current"]["open_time"]),
+                            "value": str(data["candles"][controller]["current"].open_time),
                             "target_type": "datetime",
                             "ignore_failed_order": False,
                         },
@@ -53,7 +53,7 @@ def give_order(self, data: dict) -> list[dict]:
                     "asked_for_amount": 20,
                     "asked_for_ticker": controller.quote,
                     "pair": controller.pair,
-                    "price": data["candles"][controller]["latest"]["close"],
+                    "price": data["candles"][controller]["latest"].close,
                     "side": SIDES.BUY,
                 },
             ]
@@ -67,7 +67,7 @@ def give_order(self, data: dict) -> list[dict]:
                 "asked_for_amount": 0,
                 "asked_for_ticker": controller.quote,
                 "pair": controller.pair,
-                "price": data["candles"][controller]["latest"]["close"],
+                "price": data["candles"][controller]["latest"].close,
                 "side": SIDES.KEEP,
             },
         ]

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

@@ -38,7 +38,7 @@ def give_order(self, data: dict) -> list[dict]:
                 "asked_for_amount": 0,
                 "asked_for_ticker": controller.quote,
                 "pair": controller.pair,
-                "price": data["candles"][controller]["latest"]["close"],
+                "price": data["candles"][controller]["latest"].close,
                 "side": SIDES.KEEP,
             },
         ]