fisk/core/exchanges/convert.py

from decimal import Decimal as D

from core.exchanges import GenericAPIError
from core.models import Account
from core.util import logs

log = logs.get_logger(__name__)
# Separate module to prevent circular import from
# models -> exchanges -> common -> models
# Since we need Account here to look up missing prices


def get_price(account, direction, symbol):
    """
    Get the price for a given symbol.
    :param account: Account object
    :param direction: direction of the trade
    :param symbol: symbol
    :return: price of bid for buys, price of ask for sells
    """
    if direction == "buy":
        price_index = "bids"
    elif direction == "sell":
        price_index = "asks"
    try:
        prices = account.client.get_currencies([symbol])
    except GenericAPIError as e:
        log.error(f"Error getting currencies: {e}")
        return None
    price = D(prices["prices"][0][price_index][0]["price"])
    return price


def side_to_direction(side, flip_direction=False):
    """
    Convert a side to a direction.
    :param side: Side, e.g. long, short
    :param flip_direction: Flip the direction
    :return: Direction, e.g. buy, sell
    """
    if side == "long":
        if flip_direction:
            return "sell"
        return "buy"
    elif side == "short":
        if flip_direction:
            return "buy"
        return "sell"
    else:
        return False


def direction_to_side(direction, flip_side=False):
    """
    Convert a direction to a side.
    :param direction: Direction, e.g. buy, sell
    :param flip_side: Flip the side
    :return: Side, e.g. long, short
    """
    if direction == "buy":
        if flip_side:
            return "short"
        return "long"
    elif direction == "sell":
        if flip_side:
            return "long"
        return "short"
    else:
        return False


def tp_price_to_percent(tp_price, side, current_price, current_units, unrealised_pl):
    """
    Determine the percent change of the TP price from the initial price.
    Positive values indicate a profit, negative values indicate a loss.
    """
    pl_per_unit = D(unrealised_pl) / D(current_units)
    if side == "long":
        initial_price = D(current_price) - pl_per_unit
    else:
        initial_price = D(current_price) + pl_per_unit
    # Get the percent change of the TP price from the initial price.
    change_percent = ((initial_price - D(tp_price)) / initial_price) * 100

    if side == "long":
        if D(tp_price) < initial_price:
            loss = True
        else:
            loss = False
    else:
        if D(tp_price) > initial_price:
            loss = True
        else:
            loss = False

    # if we are in loss on the short side, we want to show a negative
    if loss:
        change_percent = 0 - abs(change_percent)
    else:
        change_percent = abs(change_percent)

    return round(change_percent, 5)


def tp_percent_to_price(tp_percent, side, current_price, current_units, unrealised_pl):
    """
    Determine the price of the TP percent from the initial price.
    Negative values for tp_percent indicate a loss.
    """
    pl_per_unit = D(unrealised_pl) / D(current_units)
    if side == "long":
        initial_price = D(current_price) - pl_per_unit
    else:
        initial_price = D(current_price) + pl_per_unit

    # Get the percent change of the TP price from the initial price.
    change_percent = D(tp_percent) / 100

    # Get the price of the TP percent from the initial price.
    change_price = initial_price * abs(change_percent)
    # loss is true if tp_percent is:
    # - below initial_price for long
    # - above initial_price for short

    if D(tp_percent) < D(0):
        loss = True
    else:
        loss = False

    if side == "long":
        if loss:
            tp_price = D(initial_price) - change_price
        else:
            tp_price = D(initial_price) + change_price
    else:
        if loss:
            tp_price = D(initial_price) + change_price
        else:
            tp_price = D(initial_price) - change_price

    # if side == "long":
    #     tp_price = initial_price - change_price
    # else:
    #     tp_price = initial_price + change_price

    return round(tp_price, 5)


def sl_price_to_percent(sl_price, side, current_price, current_units, unrealised_pl):
    """
    Determine the percent change of the SL price from the initial price.
    Positive values indicate a loss, negative values indicate a profit.
    This may seem backwards, but it is important to note that by default,
    SL indicates a loss, and positive values should be expected.
    Negative values indicate a negative loss, so a profit.
    """

    pl_per_unit = D(unrealised_pl) / D(current_units)
    if side == "long":
        initial_price = D(current_price) - pl_per_unit
    else:
        initial_price = D(current_price) + pl_per_unit

    # initial_price = D(current_price) - pl_per_unit

    # Get the percent change of the SL price from the initial price.
    change_percent = ((initial_price - D(sl_price)) / initial_price) * 100

    # If the trade is long, the SL price will be higher than the initial price.
    # if side == "long":
    #     change_percent *= -1

    if side == "long":
        if D(sl_price) > initial_price:
            profit = True
        else:
            profit = False
    else:
        if D(sl_price) < initial_price:
            profit = True
        else:
            profit = False

    if profit:
        change_percent = 0 - abs(change_percent)
    else:
        change_percent = abs(change_percent)

    return round(change_percent, 5)


def sl_percent_to_price(sl_percent, side, current_price, current_units, unrealised_pl):
    """
    Determine the price of the SL percent from the initial price.
    """
    pl_per_unit = D(unrealised_pl) / D(current_units)
    if side == "long":
        initial_price = D(current_price) - pl_per_unit
    else:
        initial_price = D(current_price) + pl_per_unit

    # Get the percent change of the SL price from the initial price.
    change_percent = D(sl_percent) / 100

    # Get the price of the SL percent from the initial price.
    change_price = initial_price * abs(change_percent)

    if D(sl_percent) < D(0):
        profit = True
    else:
        profit = False

    if side == "long":
        if profit:
            sl_price = D(initial_price) + change_price
        else:
            sl_price = D(initial_price) - change_price
    else:
        if profit:
            sl_price = D(initial_price) - change_price
        else:
            sl_price = D(initial_price) + change_price

    return round(sl_price, 5)


def annotate_trade_tp_sl_percent(trade):
    """
    Annotate the trade with the TP and SL percent.
    This works on Trade objects, which will require an additional market
    lookup to get the current price.
    """
    if "current_price" in trade:
        current_price = trade["current_price"]
    else:
        account_id = trade["account_id"]
        account = Account.get_by_id_no_user_check(account_id)

        current_price = get_price(account, trade["direction"], trade["symbol"])
        trade["current_price"] = current_price

    current_units = trade["amount"]
    if "pl" in trade:
        unrealised_pl = trade["pl"]
    else:
        unrealised_pl = 0

    if "side" in trade:
        side = trade["side"]
        direction = side_to_direction(side)
        trade["direction"] = direction
    else:
        direction = trade["direction"]
        side = direction_to_side(direction)
        trade["side"] = side

    if "take_profit" in trade:
        if trade["take_profit"]:
            take_profit = trade["take_profit"]
            take_profit_percent = tp_price_to_percent(
                take_profit, trade["side"], current_price, current_units, unrealised_pl
            )

            trade["take_profit_percent"] = take_profit_percent

    if "stop_loss" in trade:
        if trade["stop_loss"]:
            stop_loss = trade["stop_loss"]
            stop_loss_percent = sl_price_to_percent(
                stop_loss, side, current_price, current_units, unrealised_pl
            )

            trade["stop_loss_percent"] = stop_loss_percent

    if "trailing_stop_loss" in trade:
        if trade["trailing_stop_loss"]:
            trailing_stop_loss = trade["trailing_stop_loss"]
            trailing_stop_loss_percent = sl_price_to_percent(
                trailing_stop_loss,
                trade["side"],
                current_price,
                current_units,
                unrealised_pl,
            )

            trade["trailing_stop_loss_percent"] = trailing_stop_loss_percent

    return trade


def open_trade_to_unified_format(trade):
    """
    Convert an open trade to a Trade-like format.
    """
    current_price = trade["price"]
    current_units = trade["currentUnits"]
    unrealised_pl = trade["unrealizedPL"]
    side = trade["side"]
    cast = {
        "id": trade["id"],
        "symbol": trade["symbol"],
        "amount": current_units,
        # For crossfilter
        "units": current_units,
        "side": side,
        "direction": side_to_direction(side),
        "state": trade["state"],
        "current_price": current_price,
        "pl": unrealised_pl,
    }
    if "openTime" in trade:
        cast["open_time"] = trade["openTime"]
    # Add some extra fields, sometimes we have already looked up the
    # prices and don't need to call convert_trades_to_usd
    # This is mostly for tests, but it can be useful in other places.
    if "take_profit_usd" in trade:
        cast["take_profit_usd"] = trade["take_profit_usd"]
    if "stop_loss_usd" in trade:
        cast["stop_loss_usd"] = trade["stop_loss_usd"]
    if "trailing_stop_loss_usd" in trade:
        cast["trailing_stop_loss_usd"] = trade["trailing_stop_loss_usd"]
    if "takeProfitOrder" in trade:
        if trade["takeProfitOrder"]:
            take_profit = trade["takeProfitOrder"]["price"]
            cast["take_profit"] = take_profit

    if "stopLossOrder" in trade:
        if trade["stopLossOrder"]:
            stop_loss = trade["stopLossOrder"]["price"]
            cast["stop_loss"] = stop_loss

    if "trailingStopLossOrder" in trade:
        if trade["trailingStopLossOrder"]:
            trailing_stop_loss = trade["trailingStopLossOrder"]["price"]
            cast["trailing_stop_loss"] = trailing_stop_loss

    return cast


def convert_trades(open_trades):
    """
    Convert a list of open trades into a list of Trade-like objects.
    """
    trades = []
    for trade in open_trades:
        if "currentUnits" in trade:  # Open trade
            cast = open_trade_to_unified_format(trade)
            cast = annotate_trade_tp_sl_percent(cast)
            trades.append(cast)
        else:
            cast = annotate_trade_tp_sl_percent(trade)
            trades.append(cast)

    return trades