fisk/core/exchanges/oanda.py

from oandapyV20 import API
from oandapyV20.endpoints import accounts, orders, positions, pricing, trades

from core.exchanges import BaseExchange, common
from core.util import logs

log = logs.get_logger("oanda")


class OANDAExchange(BaseExchange):
    def call_method(self, request):
        self.client.request(request)
        response = request.response
        if isinstance(response, list):
            response = {"itemlist": response}
        return response

    def connect(self):
        self.client = API(access_token=self.account.api_secret)
        self.account_id = self.account.api_key

    def get_account(self):
        r = accounts.AccountDetails(self.account_id)
        return self.call(r)

    def get_instruments(self):
        r = accounts.AccountInstruments(accountID=self.account_id)
        response = self.call(r)
        return response

    def get_currencies(self, currencies):
        params = {"instruments": ",".join(currencies)}
        r = pricing.PricingInfo(accountID=self.account_id, params=params)
        response = self.call(r)
        return response

    def get_supported_assets(self, response=None):
        if not response:
            response = self.get_instruments()
        return [x["name"] for x in response["itemlist"]]

    def get_balance(self, return_usd=False):
        r = accounts.AccountSummary(self.account_id)
        response = self.call(r)
        balance = float(response["balance"])
        currency = response["currency"]
        balance_usd = common.to_currency("sell", self.account, balance, currency, "USD")

        common.get_balance_hook(
            self.account.user.id,
            self.account.user.username,
            self.account.id,
            self.account.name,
            balance_usd,
        )
        if return_usd:
            return balance_usd
        return balance

    def get_market_value(self, symbol):
        raise NotImplementedError

    def post_trade(self, trade):
        if trade.direction == "sell":
            amount = -trade.amount
        else:
            amount = trade.amount
        data = {
            "order": {
                # "price": "1.5000", - added later
                "timeInForce": trade.time_in_force.upper(),
                "instrument": trade.symbol,
                "units": str(amount),
                "type": trade.type.upper(),
                "positionFill": "DEFAULT",
            }
        }
        if trade.stop_loss is not None:
            data["order"]["stopLossOnFill"] = {
                "timeInForce": "GTC",
                "price": str(trade.stop_loss),
            }
        if trade.take_profit is not None:
            data["order"]["takeProfitOnFill"] = {"price": str(trade.take_profit)}
        if trade.price is not None:
            if trade.type == "limit":
                data["order"]["price"] = str(trade.price)
            elif trade.type == "market":
                data["order"]["priceBound"] = str(trade.price)
        if trade.trailing_stop_loss is not None:
            data["order"]["trailingStopLossOnFill"] = {
                "distance": str(trade.trailing_stop_loss),
                "timeInForce": "GTC",
            }
        r = orders.OrderCreate(self.account_id, data=data)
        response = self.call(r)
        trade.response = response
        trade.status = "posted"
        trade.order_id = str(int(response["id"]) + 1)
        trade.client_order_id = response["requestID"]
        trade.save()
        return response

    def get_trade_precision(self, symbol):
        instruments = self.account.instruments
        if not instruments:
            log.error("No instruments found")
            return None
        # Extract the information for the symbol
        instrument = self.extract_instrument(instruments, symbol)
        if not instrument:
            log.error(f"Symbol not found: {symbol}")
            return None
        # Get the required precision
        try:
            trade_precision = instrument["tradeUnitsPrecision"]
            return trade_precision
        except KeyError:
            log.error(f"Precision not found for {symbol} from {instrument}")
            return None

    def close_trade(self, trade_id, units=None, symbol=None):
        """
        Close a trade.
        """
        if not units:
            r = trades.TradeClose(accountID=self.account_id, tradeID=trade_id)
            return self.call(r)
        else:
            trade_precision = self.get_trade_precision(symbol)
            if trade_precision is None:
                log.error(f"Unable to get trade precision for {symbol}")
                return None
            units = round(units, trade_precision)
            data = {
                "units": str(units),
            }
            r = trades.TradeClose(
                accountID=self.account_id, tradeID=trade_id, data=data
            )
            return self.call(r)

    def get_trade(self, trade_id):
        # OANDA is off by one...
        r = trades.TradeDetails(accountID=self.account_id, tradeID=trade_id)
        return self.call(r)

    def update_trade(self, trade):
        raise NotImplementedError
        # r = orders.OrderReplace(
        #     accountID=self.account_id, orderID=trade.order_id, data=data
        # )
        # self.client.request(r)
        # return r.response

    def cancel_trade(self, trade_id):
        raise NotImplementedError

    def get_position_info(self, symbol):
        r = positions.PositionDetails(self.account_id, symbol)
        response = self.call(r)
        response["account"] = self.account.name
        response["account_id"] = self.account.id
        return response

    def get_all_positions(self):
        items = []
        r = positions.OpenPositions(accountID=self.account_id)
        response = self.call(r)

        for item in response["itemlist"]:
            item["account"] = self.account.name
            item["account_id"] = self.account.id
            item["unrealized_pl"] = float(item["unrealized_pl"])
            items.append(item)
        return items

    def get_all_open_trades(self):
        r = trades.OpenTrades(accountID=self.account_id)
        return self.call(r)["itemlist"]

    def close_position(self, side, symbol):
        data = {
            f"{side}Units": "ALL",
        }
        r = positions.PositionClose(
            accountID=self.account_id, instrument=symbol, data=data
        )
        response = self.call(r)
        return response

    def close_all_positions(self):
        all_positions = self.get_all_positions()
        responses = []
        for position in all_positions:
            side = position["side"]
            symbol = position["symbol"]
            data = {
                f"{side}Units": "ALL",
            }
            r = positions.PositionClose(
                accountID=self.account_id, instrument=symbol, data=data
            )
            response = self.call(r)
            responses.append(response)
        return responses