fisk/core/views/positions.py

from django.contrib.auth.mixins import LoginRequiredMixin
from django.shortcuts import render
from mixins.views import ObjectList, ObjectRead
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 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, ObjectList):
    list_template = "partials/position-list.html"
    page_title = "Live positions from all exchanges"
    page_subtitle = "Manual trades are editable under 'Bot Trades' tab."
    context_object_name_singular = "position"
    context_object_name = "positions"

    list_url_name = "positions"
    list_url_args = ["type", "account_id"]

    widget_options = 'gs-w="12" gs-h="1" gs-y="0" gs-x="0"'

    def get_queryset(self, **kwargs):
        account_id = kwargs.get("account_id", None)
        if account_id:
            self.extra_context["account_id"] = account_id
        items = []
        # Only get enabled accounts for positions
        accounts = Account.objects.filter(user=self.request.user, enabled=True)
        for account in accounts:
            try:
                positions = account.client.get_all_positions()
            except GenericAPIError:
                continue

            for item in positions:
                items.append(item)
        annotate_positions(items, self.request.user, return_order_ids=False)
        return items


class PositionAction(LoginRequiredMixin, OTPRequiredMixin, ObjectRead):
    detail_template = "partials/position-detail.html"

    context_object_name_singular = "position"
    context_object_name = "position info"

    detail_url_name = "position_action"
    detail_url_args = ["type", "account_id", "symbol"]

    def get_object(self, **kwargs):
        account_id = kwargs.get("account_id")
        symbol = kwargs.get("symbol")
        account = Account.get_by_id(account_id, self.request.user)
        info = account.client.get_position_info(symbol)
        del info["long"]
        del info["short"]
        valid_trade_ids = list(
            annotate_positions([info], self.request.user, return_order_ids=True)
        )
        self.extra_context = {"valid_trade_ids": valid_trade_ids}
        return info

    def delete(self, request, account_id, side, symbol):
        """
        Close a position.
        """
        template_name = "mixins/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