diff --git a/crypto_seller/__init__.py b/crypto_seller/__init__.py index 0114779..51d3960 100644 --- a/crypto_seller/__init__.py +++ b/crypto_seller/__init__.py @@ -51,7 +51,7 @@ The most relevant libraries for auditing are: - [`python-kucoin`](https://python-kucoin.readthedocs.io/en/latest/) is used by `fin_depo` for providing KuCoin support. -Todo +TODO ## Taxation @@ -65,6 +65,7 @@ the trades file (`output/trades.csv`). To help with tax reporting, it might be useful to sign up to tax oriented websites. For example, [CryptoSkat](https://cryptoskat.dk/) seems to be the most mature on the danish market, and does support KuCoin. + """ __all__ = ['__version__', 'run_auto_sell'] @@ -74,9 +75,10 @@ import datetime import logging import random import time +from collections.abc import Callable from decimal import Decimal from pathlib import Path -from typing import Any,Callable +from typing import Any import fin_defs import fin_depo @@ -96,28 +98,32 @@ PATH_TRADES_FILE = PATH_OUTPUT / 'trades.csv' # Main code # -@dataclasses.dataclass(frozen = True) +@dataclasses.dataclass(frozen=True) 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 seller: fin_depo.defi_kucoin.KucoinDepoFetcher - sleep: Callable[[float],None] + sleep: Callable[[float], None] 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 + def run_auto_sell(config: AutoSellConfig): rng = random.SystemRandom() while True: # Check that account has tokens. - input_amount_available = config.seller.get_depo().get_amount_of_asset(config.input_asset) + input_amount_available = config.seller.get_depo().get_amount_of_asset( + config.input_asset, + ) logger.info('Currently own %s %s', input_amount_available, config.input_asset) if input_amount_available > 0: @@ -125,8 +131,9 @@ def run_auto_sell(config: AutoSellConfig): 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 = config.seller.place_market_order( - config.input_asset, amount_to_sell, config.output_asset) + order_details = config.seller.place_market_order( + config.input_asset, amount_to_sell, config.output_asset, + ) print(order_details) # TODO: Write order details to file. diff --git a/crypto_seller/__main__.py b/crypto_seller/__main__.py index 98c525c..ebbbda7 100644 --- a/crypto_seller/__main__.py +++ b/crypto_seller/__main__.py @@ -1,23 +1,24 @@ import datetime import logging import logging.handlers -from decimal import Decimal import time +from decimal import Decimal import fin_defs -from . import config -from . import PATH_LOG_FILE, AutoSellConfig, run_auto_sell +from . import PATH_LOG_FILE, AutoSellConfig, config, run_auto_sell from . import logger as module_logger logger = logging.getLogger(__name__) + def setup_logging(): PATH_LOG_FILE.parent.mkdir(parents=True, exist_ok=True) file_handler = logging.handlers.WatchedFileHandler(filename=PATH_LOG_FILE) file_handler.setFormatter( logging.Formatter( - '%(levelname)s:%(asctime)s: %(message)s', datefmt='%Y-%m-%d %H:%M:%S', + '%(levelname)s:%(asctime)s: %(message)s', + datefmt='%Y-%m-%d %H:%M:%S', ), ) @@ -26,6 +27,7 @@ def setup_logging(): logger.setLevel('INFO') module_logger.setLevel('INFO') + def main(): setup_logging() logger.info('Initializing crypto_seller') @@ -42,8 +44,8 @@ def main(): input_asset=fin_defs.MPC, output_asset=fin_defs.USDT, exit_when_empty=True, - seller = seller_backend, - sleep = time.sleep + seller=seller_backend, + sleep=time.sleep, ) run_auto_sell(auto_sell_config) diff --git a/test/test_auto_sell.py b/test/test_auto_sell.py index d87f4a7..87b7c89 100644 --- a/test/test_auto_sell.py +++ b/test/test_auto_sell.py @@ -1,22 +1,30 @@ +import datetime +from decimal import Decimal import fin_defs -from decimal import Decimal import fin_depo -import datetime import crypto_seller -class SellerMock(fin_depo.data.DepoFetcher): +class SellerMock(fin_depo.data.DepoFetcher): def __init__(self, asset: fin_defs.Asset, initial_amount: Decimal): self.asset = asset self.amount_left = initial_amount def get_depo(self) -> fin_depo.data.Depo: - return fin_depo.data.DepoSingle('TestDepo', datetime.datetime.now(tz=datetime.UTC), {self.asset:self.amount_left}) + return fin_depo.data.DepoSingle( + 'TestDepo', + datetime.datetime.now(tz=datetime.UTC), + {self.asset: self.amount_left}, + ) - def place_market_order(self, input_asset: fin_defs.Asset, input_amount: - Decimal, output_asset: fin_defs.Asset) -> fin_depo.data.Depo: + def place_market_order( + self, + input_asset: fin_defs.Asset, + input_amount: Decimal, + output_asset: fin_defs.Asset, + ) -> fin_depo.data.Depo: assert input_amount <= self.amount_left, 'Attempt to sell too much' self.amount_left -= input_amount @@ -24,7 +32,7 @@ class SellerMock(fin_depo.data.DepoFetcher): input_asset=input_asset, input_amount=input_amount, output_asset=output_asset, - output_amount=input_amount, # TODO? + output_amount=input_amount, # TODO? fee_asset=input_asset, fee_amount=Decimal(0), order_id=10000000 - self.amount_left, @@ -33,15 +41,14 @@ class SellerMock(fin_depo.data.DepoFetcher): class SleepMock: - def __init__(self): self.time_slept = 0.0 def __call__(self, amount: float): self.time_slept = self.time_slept + amount -def test_auto_run(): +def test_auto_run(): sleep_mock = SleepMock() seller_mock = SellerMock(fin_defs.USDT, Decimal('1000')) @@ -51,12 +58,11 @@ def test_auto_run(): input_asset=fin_defs.USDT, output_asset=fin_defs.USD, exit_when_empty=True, - seller = seller_mock, - sleep = sleep_mock , + seller=seller_mock, + sleep=sleep_mock, ) crypto_seller.run_auto_sell(config) assert seller_mock.amount_left == 0 assert sleep_mock.time_slept == 1000 -