crypto-seller/crypto_seller/__main__.py

import argparse
import datetime
import json
import logging
import logging.handlers
import sys
import time
from decimal import Decimal
from pathlib import Path

import fin_defs
import fin_depo

from . import (
    AutoSellConfig,
    log_estimates,
    log_results,
    order_csv,
    run_auto_sell,
)
from . import logger as module_logger

logger = logging.getLogger(__name__)

################################################################################
# Constants #

PATH_FILE_LOG = 'log.txt'
PATH_FILE_TRADES = 'trades.csv'

################################################################################
# Application Setup #


def setup_logging(path_log_file: Path):
    """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 = """
Script to automatically and with little effort sell financial assets from
online accounts.
""".strip()

CLI_EPILOG = """
Author   :  Jon Michael Aanes (jonjmaa@gmail.com)
Website  :  https://gitfub.space/Jmaa/crypto-seller
License  :  MIT License (see website for full text)
""".strip()


def load_config(config_path: Path, path_trades_file: 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(
        prog='crypto_seller',
        description=CLI_DESCRIPTION,
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=CLI_EPILOG,
    )
    parser.add_argument(
        '--config',
        type=Path,
        dest='config_file',
        required=True,
        help='Trading configuration file',
    )
    parser.add_argument(
        '--output',
        type=Path,
        dest='output_directory',
        required=True,
        help='Directory for outputing logs and trades list',
    )
    parser.add_argument(
        '--wait-before-first',
        action='store_true',
        dest='wait_before_first',
        help='Skip the first sell-off round, and wait for the next.',
    )
    return parser.parse_args()


def main():
    """Initializes the program."""

    args = parse_args()

    # Setup output paths
    path_output = args.output_directory.absolute()
    path_output.mkdir(parents=True, exist_ok=True)
    path_log_file = path_output / 'log.txt'
    path_trades_file = path_output / 'trades.csv'
    setup_logging(path_log_file )

    logger.info('Initializing crypto_seller')

    auto_sell_config = load_config(args.config_file, path_trades_file)

    # Display estimates
    try:
        log_estimates(auto_sell_config)
    except KeyboardInterrupt:
        sys.exit(1)

    # Run auto sell
    results = run_auto_sell(
        auto_sell_config,
        initial_rounds_to_skip=(args.wait_before_first and 1) or 0,
    )

    # Display results
    logging.info('Sell-offs complete')
    log_results(results)


if __name__ == '__main__':
    main()