DogsTailFarmer/martin-binance

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Searches for optimal parameters for a strategy under given conditions
"""
__author__ = "Jerry Fedorenko"
__copyright__ = "Copyright © 2024 Jerry Fedorenko aka VM"
__license__ = "MIT"
__version__ = "3.0.5"
__maintainer__ = "Jerry Fedorenko"
__contact__ = "https://github.com/DogsTailFarmer"


import importlib.util as iu
import logging.handlers
import stat
import sys
from decimal import Decimal
from pathlib import Path

import optuna
import ujson as json

from martin_binance import LOG_PATH, TRIAL_PARAMS

OPTIMIZER = Path(__file__).absolute()
try:
    OPTIMIZER.chmod(OPTIMIZER.stat().st_mode | stat.S_IEXEC)
except PermissionError:
    pass  # if executed in Docker environment

PARAMS_FLOAT = ['KBB']
STRATEGY = None


# noinspection PyUnusedLocal
def notify_exception(*args):
    pass  # Supress message from sys.excepthook


def any2str(_x) -> str:
    return f"{_x:.6f}".rstrip('0').rstrip('.')


def try_trade(mbs, skip_log, **kwargs):
    for key, value in kwargs.items():
        setattr(mbs.ex, key, value if isinstance(value, int) or key in PARAMS_FLOAT else Decimal(f"{value}"))
    mbs.ex.MODE = 'S'
    mbs.ex.SAVE_DS = False
    mbs.ex.LOGGING = not skip_log
    global STRATEGY
    STRATEGY = mbs.trade(STRATEGY)
    return float(mbs.ex.SESSION_RESULT.get('profit', 0)) + float(mbs.ex.SESSION_RESULT.get('free', 0))


def optimize(study_name, cli, n_trials, storage_name=None, _prm_best=None, skip_log=True, show_progress_bar=False):
    sys.excepthook = notify_exception
    optuna.logging.set_verbosity(optuna.logging.WARNING)

    # Load parameter definitions from JSON file
    with open(TRIAL_PARAMS) as f:
        param_defs = json.load(f)

    spec = iu.spec_from_file_location("strategy", cli)
    mbs = iu.module_from_spec(spec)
    spec.loader.exec_module(mbs)

    def objective(_trial):
        params = {}
        for param_name, param_props in param_defs.items():
            if param_props['type'] == 'int':
                params[param_name] = _trial.suggest_int(
                    param_name, *param_props['range'], step=param_props.get('step', 1)
                )
            elif param_props['type'] == 'float':
                params[param_name] = _trial.suggest_float(
                    param_name, *param_props['range'], step=param_props.get('step', 0.1)
                )
        return try_trade(mbs, skip_log, **params)

    # noinspection PyArgumentList
    _study = optuna.create_study(study_name=study_name, storage=storage_name, direction="maximize")

    if _prm_best:
        logger.info(f"Previous best params: {_prm_best}")
        _study.enqueue_trial(_prm_best)

    _study.optimize(objective, n_trials=n_trials, gc_after_trial=True, show_progress_bar=show_progress_bar)
    return _study


if __name__ == "__main__":
    logger = logging.getLogger('logger_S')
    logger.level = logging.INFO
    formatter = logging.Formatter(fmt="[%(asctime)s: %(levelname)s] %(message)s")
    #
    fh = logging.handlers.RotatingFileHandler(Path(LOG_PATH, sys.argv[6]), maxBytes=500000, backupCount=5)
    fh.setFormatter(formatter)
    fh.setLevel(logging.INFO)
    logger.addHandler(fh)
    #
    prm_best = json.loads(sys.argv[5])
    logger.info(f"Previous best params: {prm_best}")
    try:
        study = optimize(
            sys.argv[1],
            sys.argv[2],
            int(sys.argv[3]),
            storage_name=sys.argv[4],
            _prm_best=prm_best
        )
    except KeyboardInterrupt:
        pass  # ignore
    except Exception as ex:
        logger.info(f"optimizer: {ex}")
    else:
        new_value = round(study.best_value, ndigits=6)
        bp = {k: int(any2str(v)) if isinstance(v, int) else float(any2str(v)) for k, v in study.best_params.items()}

        logger.info(f"Optimal parameters: {bp} for get {new_value}")
        if new_value:
            logger.info(f"Importance parameters: {optuna.importance.get_param_importances(study)}")

        _value = round(study.get_trials()[0].value, ndigits=6)

        if not prm_best or new_value > _value:
            bp |= {'new_value': any2str(new_value), '_value': any2str(_value)}
            print(json.dumps(bp))
        else:
            print(json.dumps({}))