import uuid from django.contrib.auth.mixins import LoginRequiredMixin from django.http import HttpResponseBadRequest from django.shortcuts import render from django.urls import reverse from django.views import View from rest_framework.parsers import FormParser from two_factor.views.mixins import OTPRequiredMixin from core.exchanges import GenericAPIError from core.models import Account, Trade from core.util import logs log = logs.get_logger(__name__) def get_positions(user, account_id=None): items = [] accounts = Account.objects.filter(user=user) for account in accounts: try: positions = account.client.get_all_positions() except GenericAPIError: continue for item in positions: items.append(item) return items def annotate_positions(positions, user, return_order_ids=False): """ Annotate positions with trade information. If return_order_ids is True, yield a list of order_ids instead of a list of trades. :param positions: list of positions :param user: user :param return_order_ids: whether to return a generator of order_ids :return: None or list of order_ids :rtype: None or generator """ for item in positions: try: if "trade_ids" in item: if return_order_ids: for trade_id in Trade.objects.filter( user=user, order_id__in=item["trade_ids"] ): yield trade_id.order_id else: item["trades"] = Trade.objects.filter( user=user, order_id__in=item["trade_ids"] ) except Trade.DoesNotExist: pass class Positions(LoginRequiredMixin, OTPRequiredMixin, View): allowed_types = ["modal", "widget", "window", "page"] window_content = "window-content/objects.html" list_template = "partials/position-list.html" page_title = "Live positions from all exchanges" page_subtitle = "Manual trades are editable under 'Bot Trades' tab." def get(self, request, type, account_id=None): if type not in self.allowed_types: return HttpResponseBadRequest self.template_name = f"wm/{type}.html" unique = str(uuid.uuid4())[:8] items = get_positions(request.user, account_id) annotate_positions(items, request.user, return_order_ids=False) orig_type = type if type == "page": type = "modal" cast = { "type": orig_type, } if account_id: cast["account_id"] = account_id list_url = reverse("positions", kwargs={**cast}) context = { "title": f"Positions ({type})", "unique": unique, "window_content": self.window_content, "list_template": self.list_template, "items": items, "type": type, "page_title": self.page_title, "page_subtitle": self.page_subtitle, "list_url": list_url, "context_object_name_singular": "position", "context_object_name": "positions", } # Return partials for HTMX if self.request.htmx: if orig_type == "page": self.template_name = self.list_template else: context["window_content"] = self.list_template return render(request, self.template_name, context) class PositionAction(LoginRequiredMixin, OTPRequiredMixin, View): allowed_types = ["modal", "widget", "window", "page"] window_content = "window-content/view-position.html" parser_classes = [FormParser] def get(self, request, type, account_id, symbol): """ Get live information for a trade. """ if type not in self.allowed_types: return HttpResponseBadRequest() template_name = f"wm/{type}.html" unique = str(uuid.uuid4())[:8] account = Account.get_by_id(account_id, request.user) info = account.client.get_position_info(symbol) valid_trade_ids = list( annotate_positions([info], request.user, return_order_ids=True) ) # Remove some fields from the info dict del info["long"] del info["short"] if type == "page": type = "modal" context = { "title": f"Position info ({type})", "unique": unique, "window_content": self.window_content, "type": type, "items": info, "valid_trade_ids": valid_trade_ids, } return render(request, template_name, context) def delete(self, request, account_id, side, symbol): """ Close a position. """ template_name = "partials/notify.html" account = Account.get_by_id(account_id, request.user) try: api_response = account.client.close_position(side, symbol) except GenericAPIError as e: context = {"message": e, "class": "danger"} return render(request, template_name, context) if "longOrderCreateTransaction" in api_response: context = { "message": f"Long position closed on {symbol}", "class": "success", } elif "shortOrderCreateTransaction" in api_response: context = { "message": f"Short position closed on {symbol}", "class": "success", } response = render(request, template_name, context) response["HX-Trigger"] = "positionEvent" return response