116 lines
4.1 KiB
Python
116 lines
4.1 KiB
Python
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
|