import argparse import datetime import json import logging import logging.handlers import time from decimal import Decimal from pathlib import Path import fin_defs import fin_depo from . import ( AutoSellConfig, log_results, order_csv, run_auto_sell, ) from . import logger as module_logger logger = logging.getLogger(__name__) ################################################################################ # Constants # PATH_OUTPUT = Path('./output').absolute() PATH_LOG_FILE = PATH_OUTPUT / 'log.txt' PATH_TRADES_FILE = PATH_OUTPUT / 'trades.csv' ################################################################################ # Application Setup # def setup_logging(): """Enables logging for the terminal and to a log file.""" 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', ), ) stream_handler = logging.StreamHandler() try: import logging_color stream_handler = logging_color.ColorStreamHandler() except ImportError: pass logging.basicConfig(handlers=[stream_handler, file_handler]) logger.setLevel('INFO') module_logger.setLevel('INFO') CLI_DESCRIPTION = """ Sells financial assets from an online account. """.strip() def load_config(config_path: Path) -> AutoSellConfig: logger.info('Loading configuration') from . import secrets_config seller_backend = fin_depo.defi_kucoin.KucoinDepoFetcher( kucoin_secret=secrets_config.KUCOIN_SECRET, kucoin_key=secrets_config.KUCOIN_KEY, kucoin_pass=secrets_config.KUCOIN_PASS, allow_trades=True, ) with open(config_path) as f: json_config = json.load(f) return AutoSellConfig( input_amount_range=( Decimal(json_config['input_amount_low']), Decimal(json_config['input_amount_high']), ), interval_range=( datetime.timedelta(minutes=json_config['interval_minutes_low']), datetime.timedelta(minutes=json_config['interval_minutes_high']), ), input_asset=fin_defs.WELL_KNOWN_SYMBOLS[json_config['input_asset']], output_asset=fin_defs.WELL_KNOWN_SYMBOLS[json_config['output_asset']], exit_when_empty=True, seller=seller_backend, sleep=time.sleep, log_order_to_csv=order_csv.CsvFileLogger(PATH_TRADES_FILE), ) def parse_args(): parser = argparse.ArgumentParser('crypto_seller', description=CLI_DESCRIPTION) parser.add_argument('--config', type=Path, dest='config_file', required=True) return parser.parse_args() def main(): """Initializes the program.""" setup_logging() logger.info('Initializing crypto_seller') args = parse_args() auto_sell_config = load_config(args.config_file) results = run_auto_sell(auto_sell_config) logging.info('Sell-offs complete') log_results(results) if __name__ == '__main__': main()