crypto-seller/crypto_seller/__init__.py

"""# Automatic Crypto Seller.

Description TODO.
"""

__all__ = ['__version__', 'run_auto_sell']

import datetime
import dataclasses
import random
from typing import Any
import logging
from decimal import Decimal
import time


import fin_defs
import fin_depo

from . import config
from ._version import __version__

logger = logging.getLogger(__name__)
logger.setLevel('INFO')

@dataclasses.dataclass
class AutoSellConfig:
    interval_range: tuple[datetime.timedelta, datetime.timedelta]
    input_asset: fin_defs.Asset
    input_amount_range: tuple[Decimal, Decimal]
    output_asset: fin_defs.Asset
    exit_when_empty: bool = True

def sample_from_range(rng: random.Random, range: tuple[Any, Any]) -> Any:
    multiplier = rng.random()
    if isinstance(range[0], Decimal):
        multiplier = Decimal(multiplier)
    return range[0] + (range[1] - range[0]) * multiplier

SELLER = fin_depo.defi_kucoin.KucoinDepoFetcher(
        kucoin_secret = config.KUCOIN_SECRET,
        kucoin_key = config.KUCOIN_KEY,
        kucoin_pass = config.KUCOIN_PASS,
)

def determine_available_tokens(token: fin_defs.Asset) -> Decimal:
    return SELLER.get_depo().get_amount_of_asset(token)

def sell_tokens(input_asset: fin_defs.Asset, input_amount: Decimal,
                output_asset: fin_defs.Asset) -> fin_depo.data.TradeOrderDetails:
    #return SELLER.place_market_order(input_asset, input_amount, output_asset)
    return fin_depo.data.TradeOrderDetails(
            input_asset= input_asset,
            input_amount= input_amount,
            output_asset= output_asset,
            output_amount= input_amount,
            fee_asset = input_asset,
            fee_amount = input_amount,
            order_id = 10001,
            raw_order_details = {'TEST': 1, 'DATA': 2},
    )

def run_auto_sell(config: AutoSellConfig):
    rng = random.SystemRandom()

    while True:
        # Check that account has tokens.
        input_amount_available = determine_available_tokens(config.input_asset)
        logger.info('Currently own %s %s', input_amount_available, config.input_asset)

        if input_amount_available > 0:
            amount_to_sell = sample_from_range(rng, config.input_amount_range)
            amount_to_sell = min(input_amount_available, amount_to_sell)
            logger.info('Attempting to sell %s %s', amount_to_sell, config.input_asset)

            order_details = sell_tokens(config.input_asset, amount_to_sell, config.output_asset)
            print(order_details)
            # TODO: Write order details to file.

            del amount_to_sell
        elif config.exit_when_empty:
            break

        # Time out
        time_to_sleep = sample_from_range(rng, config.interval_range)
        time_to_sleep_secs = time_to_sleep.total_seconds()
        logger.info('Sleeping %s (%d seconds)', time_to_sleep, time_to_sleep_secs)
        time.sleep(time_to_sleep_secs)

        del input_amount_available