stable/freqtrade/exchange/okx.py

106 lines
3.4 KiB
Python
Raw Normal View History

2021-11-02 18:49:53 +00:00
import logging
from typing import Dict, List, Tuple
2021-11-02 18:49:53 +00:00
import ccxt
from freqtrade.enums import MarginMode, TradingMode
from freqtrade.exceptions import DDosProtection, OperationalException, TemporaryError
from freqtrade.exchange import Exchange
2022-02-15 06:04:50 +00:00
from freqtrade.exchange.common import retrier
2021-11-02 18:49:53 +00:00
logger = logging.getLogger(__name__)
2022-02-08 18:45:39 +00:00
class Okx(Exchange):
"""Okx exchange class.
2021-11-09 10:31:54 +00:00
Contains adjustments needed for Freqtrade to work with this exchange.
2021-11-02 18:49:53 +00:00
"""
_ft_has: Dict = {
2022-01-06 13:31:23 +00:00
"ohlcv_candle_limit": 300,
2021-12-19 13:48:59 +00:00
"mark_ohlcv_timeframe": "4h",
"funding_fee_timeframe": "8h",
"can_fetch_multiple_tiers": False,
2021-11-02 18:49:53 +00:00
}
_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),
]
@retrier
def _lev_prep(
self,
pair: str,
leverage: float,
side: str # buy or sell
):
if self.trading_mode != TradingMode.SPOT:
if self.margin_mode is None:
raise OperationalException(
f"{self.name}.margin_mode must be set for {self.trading_mode.value}"
)
try:
self._api.set_leverage(
leverage=leverage,
symbol=pair,
params={
"mgnMode": self.margin_mode.value,
# "posSide": "net"",
})
except ccxt.DDoSProtection as e:
raise DDosProtection(e) from e
except (ccxt.NetworkError, ccxt.ExchangeError) as e:
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
2022-02-07 09:44:37 +00:00
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')
2022-02-07 09:44:37 +00:00
pair_tiers = self._leverage_tiers[pair]
return pair_tiers[-1]['max'] / leverage
2022-02-07 10:18:02 +00:00
2022-02-15 06:04:50 +00:00
@retrier
2022-02-07 10:18:02 +00:00
def load_leverage_tiers(self) -> Dict[str, List[Dict]]:
2022-02-14 23:34:59 +00:00
# * This is slow(~45s) on Okex, must make 90-some api calls to load all linear swap markets
if self.trading_mode == TradingMode.FUTURES:
markets = self.markets
symbols = []
for symbol, market in markets.items():
if (self.market_is_future(market)
and market['quote'] == self._config['stake_currency']):
symbols.append(symbol)
2022-02-14 23:34:59 +00:00
tiers: Dict[str, List[Dict]] = {}
# Be verbose here, as this delays startup by ~1 minute.
logger.info(
f"Initializing leverage_tiers for {len(symbols)} markets. "
"This will take about a minute.")
2022-02-15 05:59:10 +00:00
for symbol in sorted(symbols):
res = self._api.fetch_leverage_tiers(symbol)
tiers[symbol] = res[symbol]
logger.info(f"Done initializing {len(symbols)} markets.")
return tiers
else:
return {}