from oandapyV20 import API from oandapyV20.endpoints import accounts, orders, positions, pricing, trades from core.exchanges import BaseExchange 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): r = accounts.AccountSummary(self.account_id) response = self.call(r) return float(response["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(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 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() # for position in all_positions["itemlist"]: # print("POS ITER", position) r = positions.PositionClose(accountID=self.account_id) response = self.call(r) return response