import logging from datetime import timedelta from typing import Any, Dict, List, Optional, Tuple import ccxt from freqtrade.constants import BuySell from freqtrade.enums import CandleType, MarginMode, PriceType, TradingMode from freqtrade.exceptions import ( DDosProtection, OperationalException, RetryableOrderError, TemporaryError, ) from freqtrade.exchange import Exchange, date_minus_candles from freqtrade.exchange.common import retrier from freqtrade.misc import safe_value_fallback2 from freqtrade.util import dt_now, dt_ts logger = logging.getLogger(__name__) class Okx(Exchange): """Okx exchange class. Contains adjustments needed for Freqtrade to work with this exchange. """ _ft_has: Dict = { "ohlcv_candle_limit": 100, # Warning, special case with data prior to X months "mark_ohlcv_timeframe": "4h", "funding_fee_timeframe": "8h", "stoploss_order_types": {"limit": "limit"}, "stoploss_on_exchange": True, "trades_has_history": False, # Endpoint doesn't have a "since" parameter } _ft_has_futures: Dict = { "tickers_have_quoteVolume": False, "stop_price_type_field": "slTriggerPxType", "stop_price_type_value_mapping": { PriceType.LAST: "last", PriceType.MARK: "index", PriceType.INDEX: "mark", }, } _supported_trading_mode_margin_pairs: List[Tuple[TradingMode, MarginMode]] = [ # TradingMode.SPOT always supported and not required in this list # (TradingMode.MARGIN, MarginMode.CROSS), # (TradingMode.FUTURES, MarginMode.CROSS), (TradingMode.FUTURES, MarginMode.ISOLATED), ] net_only = True _ccxt_params: Dict = {"options": {"brokerId": "ffb5405ad327SUDE"}} def ohlcv_candle_limit( self, timeframe: str, candle_type: CandleType, since_ms: Optional[int] = None ) -> int: """ Exchange ohlcv candle limit OKX has the following behaviour: * 300 candles for up-to-date data * 100 candles for historic data * 100 candles for additional candles (not futures or spot). :param timeframe: Timeframe to check :param candle_type: Candle-type :param since_ms: Starting timestamp :return: Candle limit as integer """ if candle_type in (CandleType.FUTURES, CandleType.SPOT) and ( not since_ms or since_ms > (date_minus_candles(timeframe, 300).timestamp() * 1000) ): return 300 return super().ohlcv_candle_limit(timeframe, candle_type, since_ms) @retrier def additional_exchange_init(self) -> None: """ Additional exchange initialization logic. .api will be available at this point. Must be overridden in child methods if required. """ try: if self.trading_mode == TradingMode.FUTURES and not self._config["dry_run"]: accounts = self._api.fetch_accounts() self._log_exchange_response("fetch_accounts", accounts) if len(accounts) > 0: self.net_only = accounts[0].get("info", {}).get("posMode") == "net_mode" except ccxt.DDoSProtection as e: raise DDosProtection(e) from e except (ccxt.OperationFailed, ccxt.ExchangeError) as e: raise TemporaryError( f"Error in additional_exchange_init due to {e.__class__.__name__}. Message: {e}" ) from e except ccxt.BaseError as e: raise OperationalException(e) from e def _get_posSide(self, side: BuySell, reduceOnly: bool): if self.net_only: return "net" if not reduceOnly: # Enter return "long" if side == "buy" else "short" else: # Exit return "long" if side == "sell" else "short" def _get_params( self, side: BuySell, ordertype: str, leverage: float, reduceOnly: bool, time_in_force: str = "GTC", ) -> Dict: params = super()._get_params( side=side, ordertype=ordertype, leverage=leverage, reduceOnly=reduceOnly, time_in_force=time_in_force, ) if self.trading_mode == TradingMode.FUTURES and self.margin_mode: params["tdMode"] = self.margin_mode.value params["posSide"] = self._get_posSide(side, reduceOnly) return params def __fetch_leverage_already_set(self, pair: str, leverage: float, side: BuySell) -> bool: try: res_lev = self._api.fetch_leverage( symbol=pair, params={ "mgnMode": self.margin_mode.value, "posSide": self._get_posSide(side, False), }, ) self._log_exchange_response("get_leverage", res_lev) already_set = all(float(x["lever"]) == leverage for x in res_lev["data"]) return already_set except ccxt.BaseError: # Assume all errors as "not set yet" return False @retrier def _lev_prep(self, pair: str, leverage: float, side: BuySell, accept_fail: bool = False): if self.trading_mode != TradingMode.SPOT and self.margin_mode is not None: try: res = self._api.set_leverage( leverage=leverage, symbol=pair, params={ "mgnMode": self.margin_mode.value, "posSide": self._get_posSide(side, False), }, ) self._log_exchange_response("set_leverage", res) except ccxt.DDoSProtection as e: raise DDosProtection(e) from e except (ccxt.OperationFailed, ccxt.ExchangeError) as e: already_set = self.__fetch_leverage_already_set(pair, leverage, side) if not already_set: raise TemporaryError( f"Could not set leverage due to {e.__class__.__name__}. Message: {e}" ) from e except ccxt.BaseError as e: raise OperationalException(e) from e def get_max_pair_stake_amount(self, pair: str, price: float, leverage: float = 1.0) -> float: if self.trading_mode == TradingMode.SPOT: return float("inf") # Not actually inf, but this probably won't matter for SPOT if pair not in self._leverage_tiers: return float("inf") pair_tiers = self._leverage_tiers[pair] return pair_tiers[-1]["maxNotional"] / leverage def _get_stop_params(self, side: BuySell, ordertype: str, stop_price: float) -> Dict: params = super()._get_stop_params(side, ordertype, stop_price) if self.trading_mode == TradingMode.FUTURES and self.margin_mode: params["tdMode"] = self.margin_mode.value params["posSide"] = self._get_posSide(side, True) return params def _convert_stop_order(self, pair: str, order_id: str, order: Dict) -> Dict: if ( order.get("status", "open") == "closed" and (real_order_id := order.get("info", {}).get("ordId")) is not None ): # Once a order triggered, we fetch the regular followup order. order_reg = self.fetch_order(real_order_id, pair) self._log_exchange_response("fetch_stoploss_order1", order_reg) order_reg["id_stop"] = order_reg["id"] order_reg["id"] = order_id order_reg["type"] = "stoploss" order_reg["status_stop"] = "triggered" return order_reg order = self._order_contracts_to_amount(order) order["type"] = "stoploss" return order def fetch_stoploss_order(self, order_id: str, pair: str, params: Optional[Dict] = None) -> Dict: if self._config["dry_run"]: return self.fetch_dry_run_order(order_id) try: params1 = {"stop": True} order_reg = self._api.fetch_order(order_id, pair, params=params1) self._log_exchange_response("fetch_stoploss_order", order_reg) return self._convert_stop_order(pair, order_id, order_reg) except ccxt.OrderNotFound: pass params2 = {"stop": True, "ordType": "conditional"} for method in ( self._api.fetch_open_orders, self._api.fetch_closed_orders, self._api.fetch_canceled_orders, ): try: orders = method(pair, params=params2) orders_f = [order for order in orders if order["id"] == order_id] if orders_f: order = orders_f[0] return self._convert_stop_order(pair, order_id, order) except ccxt.BaseError: pass raise RetryableOrderError(f"StoplossOrder not found (pair: {pair} id: {order_id}).") def get_order_id_conditional(self, order: Dict[str, Any]) -> str: if order.get("type", "") == "stop": return safe_value_fallback2(order, order, "id_stop", "id") return order["id"] def cancel_stoploss_order( self, order_id: str, pair: str, params: Optional[Dict] = None ) -> Dict: params1 = {"stop": True} # 'ordType': 'conditional' # return self.cancel_order( order_id=order_id, pair=pair, params=params1, ) def _fetch_orders_emulate(self, pair: str, since_ms: int) -> List[Dict]: orders = [] orders = self._api.fetch_closed_orders(pair, since=since_ms) if since_ms < dt_ts(dt_now() - timedelta(days=6, hours=23)): # Regular fetch_closed_orders only returns 7 days of data. # Force usage of "archive" endpoint, which returns 3 months of data. params = {"method": "privateGetTradeOrdersHistoryArchive"} orders_hist = self._api.fetch_closed_orders(pair, since=since_ms, params=params) orders.extend(orders_hist) orders_open = self._api.fetch_open_orders(pair, since=since_ms) orders.extend(orders_open) return orders