Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
import configargparse
class Wallet:
"""
Class holding current status of wallet (assets and currencies)
"""
arg_parser = configargparse.get_argument_parser()
arg_parser.add('--wallet_currency', help='Wallet currency (separated by comma)')
arg_parser.add("--wallet_amount", help='Wallet amount (separated by comma)')
def __init__(self):
args = self.arg_parser.parse_known_args()[0]
currency = args.wallet_currency.replace(" ", "").split(',')
amount = args.wallet_amount.replace(" ", "").split(',')
amount = [float(i) for i in amount]
self.initial_balance = dict(zip(currency, amount))
self.current_balance = self.initial_balance.copy()
from abc import ABC
import configargparse
from sklearn.externals import joblib
from termcolor import colored
class ScikitBase(ABC):
"""
Base class for AI strategies
"""
arg_parser = configargparse.get_argument_parser()
arg_parser.add('-p', '--pipeline', help='trained model/pipeline (*.pkl file)', required=True)
arg_parser.add('-f', '--feature_names', help='List of features list pipeline (*.pkl file)')
pipeline = None
def __init__(self):
args = self.arg_parser.parse_known_args()[0]
super(ScikitBase, self).__init__()
self.pipeline = self.load_pipeline(args.pipeline)
if args.feature_names:
self.feature_names = self.load_pipeline(args.feature_names)
@staticmethod
def load_pipeline(pipeline_file):
"""
Loads scikit model/pipeline
"""
import pandas as pd
import configargparse
from termcolor import colored
from exchanges.base import Base
from json import JSONDecodeError
from poloniex import Poloniex, PoloniexError
from core.bots.enums import TradeMode
from strategies.enums import TradeState
from core.bots.enums import BuySellMode
class Polo(Base):
"""
Poloniex interface
"""
arg_parser = configargparse.get_argument_parser()
arg_parser.add('--polo_api_key', help='Poloniex API key')
arg_parser.add("--polo_secret", help='Poloniex secret key')
arg_parser.add("--polo_txn_fee", help='Poloniex txn. fee')
arg_parser.add("--polo_buy_order", help='Poloniex buy order type')
arg_parser.add("--polo_sell_order", help='Poloniex sell order type')
valid_candle_intervals = [300, 900, 1800, 7200, 14400, 86400]
def __init__(self):
super(Polo, self).__init__()
args = self.arg_parser.parse_known_args()[0]
api_key = args.polo_api_key
secret = args.polo_secret
self.transaction_fee = float(args.polo_txn_fee)
self.polo = Poloniex(api_key, secret)
self.buy_order_type = args.polo_buy_order
self.sell_order_type = args.polo_sell_order
def get_ops_args():
p = configargparse.get_argument_parser()
#dont use parse_args because that will error
#if there are extra arguments e.g. for plugins
args = p.parse_known_args()[0]
#output all configuration values, useful for debugging
p.print_values()
#WARNING
#this is a horrible hack and should be removed
#once lookup and queues are handled better
mrtarget.common.connection.default_host = args.redis_host
mrtarget.common.connection.default_port = args.redis_port
return args
def get_argparser():
return configargparse.get_argument_parser()
def __init__(self, first_init = False, fill_unkown_args = False, coded_configuration_paths=None):
self._initialized = False
self._options = None
self._parser = None
if first_init is True:
self._initialized = True
parser = configargparse.get_argument_parser(default_config_files=coded_configuration_paths)
options = self.add_all_args(parser, fill_unkown_args)
self._options = options
singleton_options = SingleTone(options)
self._parser = parser
else:
#parser = configargparse.get_argument_parser()
#options, junk = parser.parse_known_args()
try:
self._options = SingleTone.get_value()
except:
self._options = {"ExceptionInitializing":True}
import configargparse
from termcolor import colored
from abc import ABC, abstractmethod
from .enums import TradeState as ts
from strategies.enums import TradeState
class Base(ABC):
"""
Base class for all strategies
"""
arg_parser = configargparse.get_argument_parser()
action_request = ts.none
actions = []
def __init__(self):
super(Base, self).__init__()
args = self.arg_parser.parse_known_args()[0]
self.verbosity = args.verbosity
self.min_history_ticks = 5
self.group_by_field = 'pair'
def get_min_history_ticks(self):
"""
Returns min_history_ticks
"""
return self.min_history_ticks
def parse_args(args: list = None):
"""parse CLI args and add them to app.config
"""
parser = configargparse.get_argument_parser()
# server config
parser.add_argument('--debug',
type=lambda x: bool(strtobool(x)),
env_var='JUSSI_DEBUG',
default=False)
parser.add_argument('--debug_route',
type=lambda x: bool(strtobool(x)),
env_var='JUSSI_DEBUG_ROUTE',
default=False)
parser.add_argument('--server_host', type=str, env_var='JUSSI_SERVER_HOST',
default='0.0.0.0')
parser.add_argument('--server_port', type=int, env_var='JUSSI_SERVER_PORT',
default=9000)
parser.add_argument('--server_workers', type=int,
env_var='JUSSI_SERVER_WORKERS', default=os.cpu_count())
import os
import time
import logging
import pandas as pd
import configargparse
import core.common as common
from exchanges.exchange import Exchange
from termcolor import colored
class Blueprint:
"""
Main module for generating and handling datasets for AI. Application will generate datasets including
future target/output parameters.
"""
arg_parser = configargparse.get_argument_parser()
arg_parser.add('--days', help='Days to start blueprint from', default=30)
arg_parser.add('-f', '--features', help='Blueprints module name to be used to generated features', required=True)
arg_parser.add('--ticker_size', help='Size of the candle ticker (minutes)', default=5)
arg_parser.add('--pairs', help='Pairs to blueprint')
arg_parser.add('-v', '--verbosity', help='Verbosity', action='store_true')
arg_parser.add("--buffer_size", help="Maximum Buffer size (days)", default=30)
arg_parser.add("--output_dir", help="Output directory")
logger = logging.getLogger(__name__)
features_list = None
exchange = None
blueprint = None
out_dir = 'out/blueprints/'
def __init__(self):
args = self.arg_parser.parse_known_args()[0]
from .base import Base
from core.bots.enums import TradeMode
import time
import configargparse
class Paper(Base):
"""
Main class for Paper trading
"""
arg_parser = configargparse.get_argument_parser()
arg_parser.add('--use_real_wallet', help='Use/not use fictive wallet (only for paper simulation)',
action='store_true')
mode = TradeMode.paper
ticker_df = None
def __init__(self, wallet):
args = self.arg_parser.parse_known_args()[0]
super(Paper, self).__init__(self.mode)
self.use_real_wallet = args.use_real_wallet
if not self.use_real_wallet:
self.balance = wallet.copy()
def get_next(self, interval_in_min):
"""
Returns next state
Interval: Interval in minutes