"""See `KucoinDepoFetcher` for documentation."""
import datetime
import logging
import time
from collections.abc import Iterator
from decimal import Decimal
import fin_defs
import kucoin.client
from fin_defs import AssetAmount
from .data import (
DepoFetcher,
DepoGroup,
DepoSingle,
DepositDetails,
DoubleRegister,
TradeOrderDetails,
WithdrawalDetails,
)
logger = logging.getLogger(__name__)
def parse_asset_from_ticker(ticker: str) -> fin_defs.Asset:
if ticker == 'KCS':
return fin_defs.CryptoCurrency('KCS', coingecko_id='kucoin-shares')
return fin_defs.WELL_KNOWN_SYMBOLS[ticker]
def order_from_json(order_details: dict):
(asset, base_asset) = [
parse_asset_from_ticker(a) for a in order_details['symbol'].split('-')
]
# Convert from kucoin results
if order_details['side'] == 'buy':
input_asset, output_asset = base_asset, asset
input_amount_final = Decimal(order_details['dealFunds'])
output_amount_final = Decimal(order_details['dealSize'])
else:
input_asset, output_asset = asset, base_asset
output_amount_final = Decimal(order_details['dealFunds'])
input_amount_final = Decimal(order_details['dealSize'])
fee_asset = parse_asset_from_ticker(order_details['feeCurrency'])
return TradeOrderDetails(
input=AssetAmount(input_asset, input_amount_final),
output=AssetAmount(output_asset, output_amount_final),
fee=AssetAmount(fee_asset, Decimal(order_details['fee'])),
executed_time=datetime.datetime.fromtimestamp(
order_details['createdAt'] / 1000,
tz=datetime.UTC,
),
order_id=order_details['id'],
raw_order_details=order_details,
)
class KucoinDepoFetcher(DepoFetcher):
"""`Depo` fetcher for [Kucoin](https://www.kucoin.com), the online crypto currency exchange.
Requirements for use:
- Account on [Kucoin](https://www.kucoin.com).
- Have performed Know Your Customer (KYC) for your account.
- Created API key from [settings menu](https://www.kucoin.com/account/api).
API key must have the **General Permission**. Unless you are using the
`place_market_order` functionality, the API key **should not have
any additional permissions**. Employ principle of least priviledge.
- Install [`python-kucoin`](https://python-kucoin.readthedocs.io/en/latest/) library.
Depository structure: An upper level depo split for each of the
sub-accounts (Funding, Trading, Margin, Futures...)
"""
def __init__(
self,
kucoin_key: str,
kucoin_secret: str,
kucoin_pass: str,
allow_trades: bool = False,
):
self.assert_param('kucoin_key', str, kucoin_key)
self.assert_param('kucoin_secret', str, kucoin_secret)
self.assert_param('kucoin_pass', str, kucoin_pass)
self.allow_trades = allow_trades
self.kucoin_client = kucoin.client.Client(
kucoin_key,
kucoin_secret,
kucoin_pass,
)
def get_depo(self) -> DepoGroup:
# We would ideally get timestamp from request,
# but this is fine for now.
now = datetime.datetime.now(tz=datetime.UTC)
# Assets are spread across account types, but we would like them
# clustered into different depos.
assets_by_account_type: dict[str, dict[fin_defs.Asset, Decimal]] = {}
for account_data in self.kucoin_client.get_accounts():
asset = parse_asset_from_ticker(account_data['currency'])
balance = Decimal(account_data['balance'])
assets_for_account_type = assets_by_account_type.setdefault(
account_data['type'],
{},
)
assets_for_account_type[asset] = (
assets_for_account_type.get(asset, Decimal(0)) + balance
)
del account_data, asset, balance, assets_for_account_type
return DepoGroup(
'Kucoin',
now,
[
DepoSingle('Kucoin ' + account_type, now, assets)
for account_type, assets in assets_by_account_type.items()
],
)
def place_market_order(
self,
input_: fin_defs.AssetAmount,
output_asset: fin_defs.Asset,
) -> TradeOrderDetails:
"""Executes a market order through the Kucoin market.
Will automatically determine the market based on the input and output
assets.
Requirements:
- Fetcher must have been created with `allow_trades=True`.
- API key used with fetcher must have **Spot Trading** permissions.
- Assets must be on trading account. Assets on funding accounts or
other accounts cannot be used.
Note:
- A fee will be paid to Kucoin, with the rate determined by your VIP
level and the asset being traded.
- The full `input_amount` may not be used. Inspect the resulting
`TradeOrderDetails` to see how much of the `input_amount` have been
used.
References:
- POST Market Order: <https://www.kucoin.com/docs/rest/spot-trading/orders/place-order>
- GET Market Order by id: <https://www.kucoin.com/docs/rest/spot-trading/orders/get-order-details-by-orderid>
"""
# Check requirements
if not self.allow_trades:
msg = 'KucoinDepoFetcher.allow_trades is not enabled: Cannot make trades'
raise PermissionError(msg)
if fin_defs.USDT not in [input_.asset, output_asset]:
msg = 'Non-USDT Markets are not supported'
raise NotImplementedError(msg)
# Convert arguments to kucoin client arguments
if input_.asset == fin_defs.USDT:
symbol: str = (
f'{output_asset.raw_short_name()}-{input_.asset.raw_short_name()}'
)
side: str = 'buy'
size = None
funds = str(input_.amount)
else:
symbol = f'{input_.asset.raw_short_name()}-{output_asset.raw_short_name()}'
side = 'sell'
size = str(input_.amount)
funds = None
# Place order
logger.info('Placing order: %s', str([symbol, side, size, funds]))
response = self.kucoin_client.create_market_order(
symbol=symbol,
side=side,
size=size,
funds=funds,
)
del symbol, side, size, funds, input_
# Determine order details
return self._get_order_details(response['orderId'])
def _get_withdrawals(self) -> list[WithdrawalDetails]:
raw_details = self.kucoin_client.get_withdrawals()
withdrawals = []
for item in raw_details['items']:
withdrawn_asset = parse_asset_from_ticker(item['currency'])
withdrawn_amount = Decimal(item['amount'])
fee_asset = withdrawn_asset # Assumed
fee_amount = Decimal(item['fee'])
executed_time = datetime.datetime.fromtimestamp(
item['createdAt'] / 1000,
tz=datetime.UTC,
)
withdrawals.append(
WithdrawalDetails(
AssetAmount(withdrawn_asset, withdrawn_amount),
AssetAmount(fee_asset, fee_amount),
executed_time,
item,
),
)
del item
return withdrawals
def _get_deposits(self) -> list[DepositDetails]:
raw_details = self.kucoin_client.get_deposits()
deposits = []
for item in raw_details['items']:
deposit_asset = parse_asset_from_ticker(item['currency'])
deposit_amount = Decimal(item['amount'])
fee_asset = deposit_asset # Assumed
fee_amount = Decimal(item['fee'])
executed_time = datetime.datetime.fromtimestamp(
item['createdAt'] / 1000,
tz=datetime.UTC,
)
deposits.append(
DepositDetails(
AssetAmount(deposit_asset, deposit_amount),
AssetAmount(fee_asset, fee_amount),
executed_time,
item,
),
)
del item
return deposits
def _get_historic_spot_orders(self) -> Iterator[TradeOrderDetails]:
end_time = datetime.datetime.now(tz=datetime.UTC)
for _weeks_back in range(20):
end_time = end_time - datetime.timedelta(days=7)
timestamp = int(end_time.timestamp() * 1000)
page_index = 1
while True:
raw_details = self.kucoin_client.get_orders(
end=timestamp, page=page_index,
)
yield from (order_from_json(item) for item in raw_details['items'])
page_index = raw_details['currentPage'] + 1
if page_index > raw_details['totalPage']:
break
del raw_details
del _weeks_back
def _get_double_registers(self) -> list[DoubleRegister]:
double_registers: list[DoubleRegister] = []
double_registers += self._get_deposits()
double_registers += self._get_withdrawals()
double_registers += self._get_historic_spot_orders()
double_registers.sort(key=lambda x: x.executed_time)
return double_registers
def _get_order_details(
self,
order_id: str,
) -> TradeOrderDetails:
"""Determine the order details for the order with the given id.
Retries the order a few times, as KuCoin might not have propagated the
order through their systems.
"""
order_details = self._get_order_with_retries(order_id, num_retries=10)
return order_from_json(order_details)
def _get_order_with_retries(
self,
order_id: str,
*,
num_retries: int,
sleep_between_tries: float = 1.0,
) -> dict:
"""Get the order details from KuCoin backend.
Retries the order a few times, as KuCoin might not have propagated the
order through their systems since it was sent.
"""
for _ in range(num_retries):
try:
return self.kucoin_client.get_order(order_id)
except kucoin.exceptions.KucoinAPIException: # noqa: PERF203
time.sleep(sleep_between_tries)
return self.kucoin_client.get_order(order_id)