1
0
crypto-seller/crypto_seller/__main__.py

169 lines
4.4 KiB
Python
Raw Normal View History

2024-07-22 13:48:55 +00:00
import argparse
2024-07-17 20:21:58 +00:00
import datetime
2024-07-22 21:02:25 +00:00
import json
2024-07-22 21:05:30 +00:00
import logging
import logging.handlers
2024-09-04 17:48:29 +00:00
import sys
2024-07-22 11:31:53 +00:00
import time
2024-07-22 11:36:52 +00:00
from decimal import Decimal
2024-07-22 13:48:55 +00:00
from pathlib import Path
2024-07-17 20:21:58 +00:00
import fin_defs
2024-07-22 12:53:48 +00:00
import fin_depo
2024-07-17 20:21:58 +00:00
2024-07-22 13:48:55 +00:00
from . import (
AutoSellConfig,
log_estimates,
2024-07-22 13:48:55 +00:00
log_results,
order_csv,
run_auto_sell,
)
from . import logger as module_logger
2024-07-17 20:21:58 +00:00
logger = logging.getLogger(__name__)
2024-07-22 13:48:55 +00:00
################################################################################
# Constants #
PATH_FILE_LOG = 'log.txt'
PATH_FILE_TRADES = 'trades.csv'
2024-07-22 13:48:55 +00:00
################################################################################
# Application Setup #
2024-07-22 11:36:52 +00:00
2024-07-22 21:05:30 +00:00
2024-12-02 16:34:18 +00:00
def setup_logging(path_log_file: Path):
2024-07-22 12:53:48 +00:00
"""Enables logging for the terminal and to a log file."""
2024-12-02 16:34:18 +00:00
path_log_file.parent.mkdir(parents=True, exist_ok=True)
file_handler = logging.handlers.WatchedFileHandler(filename=path_log_file)
file_handler.setFormatter(
logging.Formatter(
2024-07-22 11:36:52 +00:00
'%(levelname)s:%(asctime)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S',
),
)
2024-07-22 21:02:25 +00:00
stream_handler = logging.StreamHandler()
try:
import logging_color
2024-07-22 21:05:30 +00:00
stream_handler = logging_color.ColorStreamHandler()
2024-07-22 21:02:25 +00:00
except ImportError:
pass
logging.basicConfig(handlers=[stream_handler, file_handler])
logger.setLevel('INFO')
module_logger.setLevel('INFO')
2024-07-22 13:48:55 +00:00
2024-07-22 12:53:48 +00:00
CLI_DESCRIPTION = """
2024-09-04 15:53:06 +00:00
Script to automatically and with little effort sell financial assets from
online accounts.
2024-07-22 12:53:48 +00:00
""".strip()
2024-09-04 15:53:06 +00:00
CLI_EPILOG = """
Author : Jon Michael Aanes (jonjmaa@gmail.com)
Website : https://gitfub.space/Jmaa/crypto-seller
License : MIT License (see website for full text)
2024-07-22 12:53:48 +00:00
""".strip()
2024-07-22 13:48:55 +00:00
2024-12-02 16:34:18 +00:00
def load_config(config_path: Path, path_trades_file: Path) -> AutoSellConfig:
2024-07-22 21:02:25 +00:00
logger.info('Loading configuration')
from . import secrets_config
2024-07-22 11:21:41 +00:00
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,
2024-07-22 21:05:30 +00:00
allow_trades=True,
2024-07-22 11:21:41 +00:00
)
2024-07-22 21:02:25 +00:00
with open(config_path) as f:
json_config = json.load(f)
return AutoSellConfig(
2024-07-22 21:05:30 +00:00
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']),
),
2024-07-22 21:02:25 +00:00
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,
2024-07-22 11:36:52 +00:00
seller=seller_backend,
sleep=time.sleep,
2024-12-02 16:34:18 +00:00
log_order_to_csv=order_csv.CsvFileLogger(path_trades_file),
2024-07-17 20:21:58 +00:00
)
2024-07-22 21:02:25 +00:00
def parse_args():
2024-09-04 15:54:22 +00:00
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',
)
2024-09-04 15:54:22 +00:00
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.',
)
2024-07-22 21:02:25 +00:00
return parser.parse_args()
def main():
"""Initializes the program."""
args = parse_args()
2024-12-02 16:34:18 +00:00
# Setup output paths
2024-12-02 17:22:38 +00:00
path_output = args.output_directory.absolute()
2024-12-02 16:34:18 +00:00
path_output.mkdir(parents=True, exist_ok=True)
path_log_file = path_output / 'log.txt'
path_trades_file = path_output / 'trades.csv'
2024-12-02 17:29:59 +00:00
setup_logging(path_log_file)
2024-12-02 16:34:18 +00:00
logger.info('Initializing crypto_seller')
auto_sell_config = load_config(args.config_file, path_trades_file)
# Display estimates
2024-09-04 17:48:29 +00:00
try:
log_estimates(auto_sell_config)
except KeyboardInterrupt:
sys.exit(1)
# Run auto sell
2024-09-04 15:54:22 +00:00
results = run_auto_sell(
auto_sell_config,
2024-12-02 17:22:38 +00:00
initial_rounds_to_skip=(args.wait_before_first and 1) or 0,
2024-09-04 15:54:22 +00:00
)
# Display results
2024-07-22 12:12:05 +00:00
logging.info('Sell-offs complete')
log_results(results)
2024-07-17 20:21:58 +00:00
2024-07-17 20:21:58 +00:00
if __name__ == '__main__':
main()