1
0

Added KucoinDepoFetcher.place_market_order
All checks were successful
Test Python / Test (push) Successful in 26s

This commit is contained in:
Jon Michael Aanes 2024-07-20 21:54:29 +02:00
parent 15386f0b4d
commit dfeb3a3294
Signed by: Jmaa
SSH Key Fingerprint: SHA256:Ab0GfHGCblESJx7JRE4fj4bFy/KRpeLhi41y4pF3sNA
3 changed files with 201 additions and 7 deletions

View File

@ -92,3 +92,25 @@ class DepoFetcher(abc.ABC):
msg = f'{self} expected {param_name} parameter of type {param_type}, but got: {param_value}'
raise TypeError(msg)
return param_value
@enforce_typing.enforce_types
@dataclasses.dataclass(frozen=True)
class TradeOrderDetails:
"""Information about an executed trade.
Includes both well-structured data about the order, and unstructured data
from the backend. The unstructured data might be needed for tax purposes.
"""
input_asset: Asset
input_amount: Decimal
output_asset: Asset
output_amount: Decimal
fee_asset: Asset
fee_amount: Decimal
order_id: object
raw_order_details: object

View File

@ -7,7 +7,7 @@ from decimal import Decimal
import fin_defs
import kucoin.client
from .data import DepoFetcher, DepoGroup, DepoSingle
from .data import DepoFetcher, DepoGroup, DepoSingle, TradeOrderDetails
logger = logging.getLogger(__name__)
@ -19,7 +19,8 @@ class KucoinDepoFetcher(DepoFetcher):
- 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**, and **should not have
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.
@ -27,11 +28,18 @@ class KucoinDepoFetcher(DepoFetcher):
sub-accounts (Funding, Trading, Margin, Futures...)
"""
def __init__(self, kucoin_key: str, kucoin_secret: str, kucoin_pass: str):
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.client = kucoin.client.Client(
self.allow_trades = allow_trades
self.kucoin_client = kucoin.client.Client(
kucoin_key,
kucoin_secret,
kucoin_pass,
@ -46,7 +54,7 @@ class KucoinDepoFetcher(DepoFetcher):
# clustered into different depos.
assets_by_account_type: dict[str, dict[fin_defs.Asset, Decimal]] = {}
for account_data in self.client.get_accounts():
for account_data in self.kucoin_client.get_accounts():
asset = fin_defs.WELL_KNOWN_SYMBOLS[account_data['currency']]
balance = Decimal(account_data['balance'])
assets_for_account_type = assets_by_account_type.setdefault(
@ -66,3 +74,81 @@ class KucoinDepoFetcher(DepoFetcher):
for account_type, assets in assets_by_account_type.items()
],
)
def place_market_order(
self,
input_asset: fin_defs.Asset,
input_amount: Decimal,
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 WIP
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.
"""
# Check requirements
if not self.allow_trades:
msg = 'KucoinDepoFetcher.allow_trades is not enabled: Cannot make trades'
raise PermissionError(msg)
assert fin_defs.USDT in [input_asset, output_asset], 'USDT markets only for now'
# Convert arguments to kucoin client arguments
if input_asset == fin_defs.USDT:
symbol: str = f'{output_asset}-{input_asset}'
side: str = 'buy'
size = None
funds = str(input_amount)
else:
symbol = f'{input_asset}-{output_asset}'
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_amount
# Determine order details
order_id = response['orderId']
order_details = self.kucoin_client.get_order(order_id)
# Convert from kucoin results
if input_asset == fin_defs.USDT:
input_amount_final = Decimal(order_details['dealFunds'])
output_amount_final = Decimal(order_details['dealSize'])
else:
output_amount_final = Decimal(order_details['dealFunds'])
input_amount_final = Decimal(order_details['dealSize'])
return TradeOrderDetails(
input_asset=input_asset,
input_amount=input_amount_final,
output_asset=output_asset,
output_amount=output_amount_final,
fee_asset=fin_defs.WELL_KNOWN_SYMBOLS[order_details['feeCurrency']],
fee_amount=Decimal(order_details['fee']),
order_id=order_id,
raw_order_details=order_details,
)

View File

@ -1,19 +1,25 @@
from decimal import Decimal
import fin_defs
import pytest
import requests
from fin_depo import defi_kucoin
from . import secrets
TEST_MARKET_ORDERS = False
needs_secrets = pytest.mark.skipif(
not secrets.KUCOIN_KEY,
reason='Secret kucoin_USERNAME required',
)
defi_kucoin.logger.setLevel('INFO')
@needs_secrets
def test_get_depo():
session = requests.Session()
"""Can inspect kucoin depository."""
fetcher = defi_kucoin.KucoinDepoFetcher(
secrets.KUCOIN_KEY,
secrets.KUCOIN_SECRET,
@ -22,3 +28,83 @@ def test_get_depo():
depo = fetcher.get_depo()
assert depo is not None
@needs_secrets
def test_place_market_order_requires_allow_trades():
"""Client fails if allow_trades is not enabled when doing market orders."""
fetcher = defi_kucoin.KucoinDepoFetcher(
secrets.KUCOIN_KEY,
secrets.KUCOIN_SECRET,
secrets.KUCOIN_PASS,
allow_trades=False,
)
with pytest.raises(PermissionError) as m:
fetcher.place_market_order(fin_defs.MPC, Decimal(1), fin_defs.USDT)
assert 'KucoinDepoFetcher.allow_trades is not enabled: Cannot make trades' in str(m)
@needs_secrets
def test_place_buy_side_order():
"""Client can place buy side market orders."""
fetcher = defi_kucoin.KucoinDepoFetcher(
secrets.KUCOIN_KEY,
secrets.KUCOIN_SECRET,
secrets.KUCOIN_PASS,
allow_trades=TEST_MARKET_ORDERS,
)
input_amount = Decimal('0.1')
order_details = fetcher.place_market_order(
fin_defs.USDT, input_amount, fin_defs.MPC,
)
assert order_details is not None
assert order_details.order_id is not None
assert order_details.raw_order_details is not None
assert order_details.input_asset == fin_defs.USDT
assert order_details.output_asset == fin_defs.MPC
assert order_details.input_amount <= input_amount
assert order_details.output_amount >= 0
assert order_details.input_amount != order_details.output_amount
assert order_details.fee_asset == fin_defs.USDT
assert order_details.fee_amount <= Decimal('0.0002')
@needs_secrets
def test_place_sell_side_order():
"""Client can place sell side market orders."""
fetcher = defi_kucoin.KucoinDepoFetcher(
secrets.KUCOIN_KEY,
secrets.KUCOIN_SECRET,
secrets.KUCOIN_PASS,
allow_trades=TEST_MARKET_ORDERS,
)
input_amount = Decimal('1')
order_details = fetcher.place_market_order(
fin_defs.MPC, input_amount, fin_defs.USDT,
)
assert order_details is not None
assert order_details.order_id is not None
assert order_details.raw_order_details is not None
assert order_details.input_asset == fin_defs.MPC
assert order_details.output_asset == fin_defs.USDT
assert order_details.input_amount <= input_amount
assert order_details.output_amount >= 0
assert order_details.input_amount != order_details.output_amount
assert order_details.fee_asset == fin_defs.USDT
assert order_details.fee_amount is not None