[perso/Immae/Projets/Cryptomonnaies/Cryptoportfolio/Trader.git] / market.py

from ccxt import ExchangeError, NotSupported, RequestTimeout, InvalidNonce
import ccxt_wrapper as ccxt
import time
import psycopg2
import redis
from store import *
from cachetools.func import ttl_cache
from datetime import datetime
import datetime
from retry import retry
import portfolio

class Market:
    debug = False
    ccxt = None
    report = None
    trades = None
    balances = None

    def __init__(self, ccxt_instance, args, **kwargs):
        self.args = args
        self.debug = args.debug
        self.ccxt = ccxt_instance
        self.ccxt._market = self
        self.report = ReportStore(self, verbose_print=(not args.quiet))
        self.trades = TradeStore(self)
        self.balances = BalanceStore(self)
        self.processor = Processor(self)

        for key in ["user_id", "market_id", "pg_config", "redis_config"]:
            setattr(self, key, kwargs.get(key, None))

        self.report.log_market(self.args)

    @classmethod
    def from_config(cls, config, args, **kwargs):
        config["apiKey"] = config.pop("key", None)

        ccxt_instance = ccxt.poloniexE(config)

        return cls(ccxt_instance, args, **kwargs)

    def store_report(self):
        self.report.merge(Portfolio.report)
        date = datetime.datetime.now()
        if self.args.report_path is not None:
            self.store_file_report(date)
        if self.pg_config is not None and self.args.report_db:
            self.store_database_report(date)
        if self.redis_config is not None and self.args.report_redis:
            self.store_redis_report(date)

    def store_file_report(self, date):
        try:
            report_file = "{}/{}_{}".format(self.args.report_path, date.isoformat(), self.user_id)
            with open(report_file + ".json", "w") as f:
                f.write(self.report.to_json())
            with open(report_file + ".log", "w") as f:
                f.write("\n".join(map(lambda x: x[1], self.report.print_logs)))
        except Exception as e:
            print("impossible to store report file: {}; {}".format(e.__class__.__name__, e))

    def store_database_report(self, date):
        try:
            report_query = 'INSERT INTO reports("date", "market_config_id", "debug") VALUES (%s, %s, %s) RETURNING id;'
            line_query = 'INSERT INTO report_lines("date", "report_id", "type", "payload") VALUES (%s, %s, %s, %s);'
            connection = psycopg2.connect(**self.pg_config)
            cursor = connection.cursor()
            cursor.execute(report_query, (date, self.market_id, self.debug))
            report_id = cursor.fetchone()[0]
            for date, type_, payload in self.report.to_json_array():
                cursor.execute(line_query, (date, report_id, type_, payload))

            connection.commit()
            cursor.close()
            connection.close()
        except Exception as e:
            print("impossible to store report to database: {}; {}".format(e.__class__.__name__, e))

    def store_redis_report(self, date):
        try:
            conn = redis.Redis(**self.redis_config)
            for type_, log in self.report.to_json_redis():
                key = "/cryptoportfolio/{}/{}/{}".format(self.market_id, date.isoformat(), type_)
                conn.set(key, log, ex=31*24*60*60)
                key = "/cryptoportfolio/{}/latest/{}".format(self.market_id, type_)
                conn.set(key, log)
        except Exception as e:
            print("impossible to store report to redis: {}; {}".format(e.__class__.__name__, e))

    def process(self, actions, before=False, after=False):
        try:
            for action in actions:
                if bool(before) is bool(after):
                    self.processor.process(action, steps="all")
                elif before:
                    self.processor.process(action, steps="before")
                elif after:
                    self.processor.process(action, steps="after")
        except Exception as e:
            self.report.log_error("market_process", exception=e)
        finally:
            self.store_report()

    @retry((RequestTimeout, InvalidNonce), tries=5)
    def move_balances(self):
        needed_in_margin = {} 
        moving_to_margin = {}

        for currency, balance in self.balances.all.items():
            needed_in_margin[currency] = balance.margin_in_position - balance.margin_pending_gain
        for trade in self.trades.pending:
            needed_in_margin.setdefault(trade.base_currency, 0)
            if trade.trade_type == "short":
                needed_in_margin[trade.base_currency] -= trade.delta
        for currency, needed in needed_in_margin.items():
            current_balance = self.balances.all[currency].margin_available
            moving_to_margin[currency] = (needed - current_balance)
            delta = moving_to_margin[currency].value
            action = "Moving {} from exchange to margin".format(moving_to_margin[currency])

            if self.debug and delta != 0:
                self.report.log_debug_action(action)
                continue
            try:
                if delta > 0:
                    self.ccxt.transfer_balance(currency, delta, "exchange", "margin")
                elif delta < 0:
                    self.ccxt.transfer_balance(currency, -delta, "margin", "exchange")
            except (RequestTimeout, InvalidNonce) as e:
                self.report.log_error(action, message="Retrying", exception=e)
                self.report.log_move_balances(needed_in_margin, moving_to_margin)
                self.balances.fetch_balances()
                raise e
        self.report.log_move_balances(needed_in_margin, moving_to_margin)

        self.balances.fetch_balances()

    @ttl_cache(ttl=3600)
    def fetch_fees(self):
        return self.ccxt.fetch_fees()

    @ttl_cache(maxsize=20, ttl=5)
    def get_tickers(self, refresh=False):
        try:
            return self.ccxt.fetch_tickers()
        except NotSupported:
            return None

    @ttl_cache(maxsize=20, ttl=5)
    def get_ticker(self, c1, c2, refresh=False):
        def invert(ticker):
            return {
                    "inverted": True,
                    "average": (1/ticker["bid"] + 1/ticker["ask"]) / 2,
                    "original": ticker,
                    }
        def augment_ticker(ticker):
            ticker.update({
                "inverted": False,
                "average": (ticker["bid"] + ticker["ask"] ) / 2,
                })
            return ticker

        tickers = self.get_tickers()
        if tickers is None:
            try:
                ticker = augment_ticker(self.ccxt.fetch_ticker("{}/{}".format(c1, c2)))
            except ExchangeError:
                try:
                    ticker = invert(augment_ticker(self.ccxt.fetch_ticker("{}/{}".format(c2, c1))))
                except ExchangeError:
                    ticker = None
        else:
            if "{}/{}".format(c1, c2) in tickers:
                ticker = augment_ticker(tickers["{}/{}".format(c1, c2)])
            elif "{}/{}".format(c2, c1) in tickers:
                ticker = invert(augment_ticker(tickers["{}/{}".format(c2, c1)]))
            else:
                ticker = None
        return ticker

    def follow_orders(self, sleep=None):
        if sleep is None:
            sleep = 7 if self.debug else 30
            if self.debug:
                self.report.log_debug_action("Set follow_orders tick to {}s".format(sleep))
        tick = 0
        self.report.log_stage("follow_orders_begin")
        while len(self.trades.all_orders(state="open")) > 0:
            time.sleep(sleep)
            tick += 1
            open_orders = self.trades.all_orders(state="open")
            self.report.log_stage("follow_orders_tick_{}".format(tick))
            self.report.log_orders(open_orders, tick=tick)
            for order in open_orders:
                status = order.get_status()
                if status != "open":
                    self.report.log_order(order, tick, finished=True)
                else:
                    order.trade.update_order(order, tick)
                if status == "error_disappeared":
                    self.report.log_error("follow_orders",
                            message="{} disappeared, recreating it".format(order))
                    order.trade.prepare_order(
                            compute_value=order.trade.tick_actions_recreate(tick))

        self.report.log_stage("follow_orders_end")

    def prepare_trades(self, base_currency="BTC", liquidity="medium",
            compute_value="average", repartition=None, only=None):

        self.report.log_stage("prepare_trades",
                base_currency=base_currency, liquidity=liquidity,
                compute_value=compute_value, only=only,
                repartition=repartition)

        values_in_base = self.balances.in_currency(base_currency,
                compute_value=compute_value)
        total_base_value = sum(values_in_base.values())
        new_repartition = self.balances.dispatch_assets(total_base_value,
                liquidity=liquidity, repartition=repartition)
        self.trades.compute_trades(values_in_base, new_repartition, only=only)

    def print_tickers(self, base_currency="BTC"):
        if base_currency is not None:
            self.report.print_log("total:")
            self.report.print_log(sum(self.balances.in_currency(base_currency).values()))

class Processor:
    scenarios = {
            "wait_for_cryptoportfolio": [
                {
                    "name": "wait",
                    "number": 1,
                    "before": False,
                    "after": True,
                    "wait_for_recent": {},
                    },
                ],
            "print_balances": [
                {
                    "name": "print_balances",
                    "number": 1,
                    "fetch_balances": ["begin"],
                    "print_tickers": { "base_currency": "BTC" },
                    }
                ],
            "print_orders": [
                {
                    "name": "wait",
                    "number": 1,
                    "before": True,
                    "after": False,
                    "wait_for_recent": {},
                    },
                {
                    "name": "make_orders",
                    "number": 2,
                    "before": False,
                    "after": True,
                    "fetch_balances": ["begin"],
                    "prepare_trades": { "compute_value": "average" },
                    "prepare_orders": { "compute_value": "average" },
                    },
                ],
            "sell_needed": [
                {
                    "name": "wait",
                    "number": 0,
                    "before": False,
                    "after": True,
                    "wait_for_recent": {},
                    },
                {
                    "name": "sell",
                    "number": 1,
                    "before": False,
                    "after": True,
                    "fetch_balances": ["begin", "end"],
                    "prepare_trades": {},
                    "prepare_orders": { "only": "dispose", "compute_value": "average" },
                    "run_orders": {},
                    "follow_orders": {},
                    "close_trades": {},
                    },
                {
                    "name": "buy",
                    "number": 2,
                    "before": False,
                    "after": True,
                    "fetch_balances": ["begin", "end"],
                    "prepare_trades": { "only": "acquire" },
                    "prepare_orders": { "only": "acquire", "compute_value": "average" },
                    "move_balances": {},
                    "run_orders": {},
                    "follow_orders": {},
                    "close_trades": {},
                    },
                ],
            "sell_all": [
                {
                    "name": "all_sell",
                    "number": 1,
                    "before": True,
                    "after": False,
                    "fetch_balances": ["begin", "end"],
                    "prepare_trades": { "repartition": { "base_currency": (1, "long") } },
                    "prepare_orders": { "compute_value": "average" },
                    "run_orders": {},
                    "follow_orders": {},
                    "close_trades": {},
                    },
                {
                    "name": "wait",
                    "number": 2,
                    "before": False,
                    "after": True,
                    "wait_for_recent": {},
                    },
                {
                    "name": "all_buy",
                    "number": 3,
                    "before": False,
                    "after": True,
                    "fetch_balances": ["begin", "end"],
                    "prepare_trades": {},
                    "prepare_orders": { "compute_value": "average" },
                    "move_balances": {},
                    "run_orders": {},
                    "follow_orders": {},
                    "close_trades": {},
                    },
                ]
            }

    ordered_actions = [
            "wait_for_recent", "prepare_trades", "prepare_orders",
            "move_balances", "run_orders", "follow_orders",
            "close_trades", "print_tickers"]

    def __init__(self, market):
        self.market = market

    def select_steps(self, scenario, step):
        if step == "all":
            return scenario
        elif step == "before" or step == "after":
            return list(filter(lambda x: x.get(step, False), scenario))
        elif type(step) == int:
            return [scenario[step-1]]
        elif type(step) == str:
            return list(filter(lambda x: x["name"] == step, scenario))
        else:
            raise TypeError("Unknown step {}".format(step))

    def can_process(self, scenario_name):
        return scenario_name in self.scenarios

    def process(self, scenario_name, steps="all", **kwargs):
        if not self.can_process(scenario_name):
            raise TypeError("Unknown scenario {}".format(scenario_name))
        scenario = self.scenarios[scenario_name]
        selected_steps = []

        if type(steps) == str or type(steps) == int:
            selected_steps += self.select_steps(scenario, steps)
        else:
            for step in steps:
                selected_steps += self.select_steps(scenario, step)
        for step in selected_steps:
            self.process_step(scenario_name, step, kwargs)

    def process_step(self, scenario_name, step, kwargs):
        process_name = "process_{}__{}_{}".format(scenario_name, step["number"], step["name"])
        self.market.report.log_stage("{}_begin".format(process_name))
        if "begin" in step.get("fetch_balances", []):
            self.market.balances.fetch_balances(tag="{}_begin".format(process_name))

        for action in self.ordered_actions:
            if action in step:
                self.run_action(action, step[action], kwargs)

        if "end" in step.get("fetch_balances", []):
            self.market.balances.fetch_balances(tag="{}_end".format(process_name))
        self.market.report.log_stage("{}_end".format(process_name))

    def method_arguments(self, action):
        import inspect

        if action == "wait_for_recent":
            method = Portfolio.wait_for_recent
        elif action == "prepare_trades":
            method = self.market.prepare_trades
        elif action == "prepare_orders":
            method = self.market.trades.prepare_orders
        elif action == "move_balances":
            method = self.market.move_balances
        elif action == "run_orders":
            method = self.market.trades.run_orders
        elif action == "follow_orders":
            method = self.market.follow_orders
        elif action == "close_trades":
            method = self.market.trades.close_trades
        elif action == "print_tickers":
            method = self.market.print_tickers

        signature = inspect.getfullargspec(method)
        defaults = signature.defaults or []
        kwargs = signature.args[-len(defaults):]

        return [method, kwargs]

    def parse_args(self, action, default_args, kwargs):
        method, allowed_arguments = self.method_arguments(action)
        args = {k: v for k, v in {**default_args, **kwargs}.items() if k in allowed_arguments }

        if "repartition" in args and "base_currency" in args["repartition"]:
            r = args["repartition"]
            r[args.get("base_currency", "BTC")] = r.pop("base_currency")

        return method, args

    def run_action(self, action, default_args, kwargs):
        method, args = self.parse_args(action, default_args, kwargs)

        method(**args)
Commit	Line	Data
b47d7b54	1	from ccxt import ExchangeError, NotSupported, RequestTimeout, InvalidNonce
774c099c	2	import ccxt_wrapper as ccxt
f86ee140	3	import time
b4e0ba0b	4	import psycopg2
1593c7a9	5	import redis
f86ee140	6	from store import *
aca4d437	7	from cachetools.func import ttl_cache
1f117ac7	8	from datetime import datetime
e7d7c0e5	9	import datetime
337c8286	10	from retry import retry
1f117ac7	11	import portfolio
4c51aa71	12
f86ee140 IB	13	class Market:
	14	debug = False
	15	ccxt = None
	16	report = None
	17	trades = None
	18	balances = None
ecba1113	19
35667b31	20	def __init__(self, ccxt_instance, args, **kwargs):
07fa7a4b IB	21	self.args = args
07fa7a4b IB	22	self.debug = args.debug
f86ee140 IB	23	self.ccxt = ccxt_instance
f86ee140 IB	24	self.ccxt._market = self
07fa7a4b	25	self.report = ReportStore(self, verbose_print=(not args.quiet))
f86ee140 IB	26	self.trades = TradeStore(self)
f86ee140 IB	27	self.balances = BalanceStore(self)
1f117ac7 IB	28	self.processor = Processor(self)
1f117ac7 IB	29
1593c7a9	30	for key in ["user_id", "market_id", "pg_config", "redis_config"]:
35667b31	31	setattr(self, key, kwargs.get(key, None))
f86ee140	32
e7d7c0e5	33	self.report.log_market(self.args)
90d7423e	34
f86ee140	35	@classmethod
35667b31	36	def from_config(cls, config, args, **kwargs):
1f117ac7	37	config["apiKey"] = config.pop("key", None)
d24bb10c	38
f86ee140 IB	39	ccxt_instance = ccxt.poloniexE(config)
f86ee140 IB	40
35667b31	41	return cls(ccxt_instance, args, **kwargs)
1f117ac7 IB	42
1f117ac7 IB	43	def store_report(self):
9bde69bf	44	self.report.merge(Portfolio.report)
e7d7c0e5	45	date = datetime.datetime.now()
a42d6cc8	46	if self.args.report_path is not None:
b4e0ba0b	47	self.store_file_report(date)
a42d6cc8	48	if self.pg_config is not None and self.args.report_db:
b4e0ba0b	49	self.store_database_report(date)
1593c7a9 IB	50	if self.redis_config is not None and self.args.report_redis:
1593c7a9 IB	51	self.store_redis_report(date)
b4e0ba0b IB	52
b4e0ba0b IB	53	def store_file_report(self, date):
1f117ac7	54	try:
a42d6cc8	55	report_file = "{}/{}_{}".format(self.args.report_path, date.isoformat(), self.user_id)
b4e0ba0b IB	56	with open(report_file + ".json", "w") as f:
	57	f.write(self.report.to_json())
	58	with open(report_file + ".log", "w") as f:
	59	f.write("\n".join(map(lambda x: x[1], self.report.print_logs)))
1f117ac7 IB	60	except Exception as e:
	61	print("impossible to store report file: {}; {}".format(e.__class__.__name__, e))
	62
b4e0ba0b IB	63	def store_database_report(self, date):
	64	try:
	65	report_query = 'INSERT INTO reports("date", "market_config_id", "debug") VALUES (%s, %s, %s) RETURNING id;'
	66	line_query = 'INSERT INTO report_lines("date", "report_id", "type", "payload") VALUES (%s, %s, %s, %s);'
	67	connection = psycopg2.connect(**self.pg_config)
	68	cursor = connection.cursor()
	69	cursor.execute(report_query, (date, self.market_id, self.debug))
	70	report_id = cursor.fetchone()[0]
	71	for date, type_, payload in self.report.to_json_array():
	72	cursor.execute(line_query, (date, report_id, type_, payload))
	73
	74	connection.commit()
	75	cursor.close()
	76	connection.close()
	77	except Exception as e:
	78	print("impossible to store report to database: {}; {}".format(e.__class__.__name__, e))
	79
1593c7a9 IB	80	def store_redis_report(self, date):
	81	try:
	82	conn = redis.Redis(**self.redis_config)
	83	for type_, log in self.report.to_json_redis():
	84	key = "/cryptoportfolio/{}/{}/{}".format(self.market_id, date.isoformat(), type_)
	85	conn.set(key, log, ex=312460*60)
	86	key = "/cryptoportfolio/{}/latest/{}".format(self.market_id, type_)
	87	conn.set(key, log)
	88	except Exception as e:
	89	print("impossible to store report to redis: {}; {}".format(e.__class__.__name__, e))
	90
1f117ac7 IB	91	def process(self, actions, before=False, after=False):
1f117ac7 IB	92	try:
ceb7fc4c IB	93	for action in actions:
	94	if bool(before) is bool(after):
	95	self.processor.process(action, steps="all")
	96	elif before:
	97	self.processor.process(action, steps="before")
	98	elif after:
	99	self.processor.process(action, steps="after")
1f117ac7 IB	100	except Exception as e:
	101	self.report.log_error("market_process", exception=e)
	102	finally:
	103	self.store_report()
f86ee140	104
b47d7b54	105	@retry((RequestTimeout, InvalidNonce), tries=5)
f86ee140 IB	106	def move_balances(self):
	107	needed_in_margin = {}
	108	moving_to_margin = {}
	109
aca4d437 IB	110	for currency, balance in self.balances.all.items():
	111	needed_in_margin[currency] = balance.margin_in_position - balance.margin_pending_gain
	112	for trade in self.trades.pending:
	113	needed_in_margin.setdefault(trade.base_currency, 0)
f86ee140	114	if trade.trade_type == "short":
aca4d437	115	needed_in_margin[trade.base_currency] -= trade.delta
f86ee140	116	for currency, needed in needed_in_margin.items():
aca4d437	117	current_balance = self.balances.all[currency].margin_available
f86ee140 IB	118	moving_to_margin[currency] = (needed - current_balance)
f86ee140 IB	119	delta = moving_to_margin[currency].value
337c8286 IB	120	action = "Moving {} from exchange to margin".format(moving_to_margin[currency])
337c8286 IB	121
aca4d437	122	if self.debug and delta != 0:
337c8286	123	self.report.log_debug_action(action)
f86ee140	124	continue
337c8286 IB	125	try:
	126	if delta > 0:
	127	self.ccxt.transfer_balance(currency, delta, "exchange", "margin")
	128	elif delta < 0:
	129	self.ccxt.transfer_balance(currency, -delta, "margin", "exchange")
b47d7b54	130	except (RequestTimeout, InvalidNonce) as e:
337c8286 IB	131	self.report.log_error(action, message="Retrying", exception=e)
	132	self.report.log_move_balances(needed_in_margin, moving_to_margin)
	133	self.balances.fetch_balances()
	134	raise e
f86ee140 IB	135	self.report.log_move_balances(needed_in_margin, moving_to_margin)
	136
	137	self.balances.fetch_balances()
	138
aca4d437	139	@ttl_cache(ttl=3600)
f86ee140	140	def fetch_fees(self):
aca4d437 IB	141	return self.ccxt.fetch_fees()
	142
	143	@ttl_cache(maxsize=20, ttl=5)
	144	def get_tickers(self, refresh=False):
	145	try:
	146	return self.ccxt.fetch_tickers()
	147	except NotSupported:
	148	return None
f86ee140	149
aca4d437	150	@ttl_cache(maxsize=20, ttl=5)
f86ee140 IB	151	def get_ticker(self, c1, c2, refresh=False):
	152	def invert(ticker):
	153	return {
	154	"inverted": True,
	155	"average": (1/ticker["bid"] + 1/ticker["ask"]) / 2,
	156	"original": ticker,
	157	}
	158	def augment_ticker(ticker):
	159	ticker.update({
	160	"inverted": False,
	161	"average": (ticker["bid"] + ticker["ask"] ) / 2,
	162	})
7192b2e1	163	return ticker
f86ee140	164
aca4d437 IB	165	tickers = self.get_tickers()
aca4d437 IB	166	if tickers is None:
f86ee140	167	try:
7192b2e1	168	ticker = augment_ticker(self.ccxt.fetch_ticker("{}/{}".format(c1, c2)))
f86ee140	169	except ExchangeError:
aca4d437	170	try:
7192b2e1	171	ticker = invert(augment_ticker(self.ccxt.fetch_ticker("{}/{}".format(c2, c1))))
aca4d437 IB	172	except ExchangeError:
	173	ticker = None
	174	else:
	175	if "{}/{}".format(c1, c2) in tickers:
7192b2e1	176	ticker = augment_ticker(tickers["{}/{}".format(c1, c2)])
aca4d437	177	elif "{}/{}".format(c2, c1) in tickers:
7192b2e1	178	ticker = invert(augment_ticker(tickers["{}/{}".format(c2, c1)]))
aca4d437 IB	179	else:
	180	ticker = None
	181	return ticker
f86ee140 IB	182
	183	def follow_orders(self, sleep=None):
	184	if sleep is None:
	185	sleep = 7 if self.debug else 30
	186	if self.debug:
	187	self.report.log_debug_action("Set follow_orders tick to {}s".format(sleep))
	188	tick = 0
	189	self.report.log_stage("follow_orders_begin")
	190	while len(self.trades.all_orders(state="open")) > 0:
	191	time.sleep(sleep)
	192	tick += 1
	193	open_orders = self.trades.all_orders(state="open")
	194	self.report.log_stage("follow_orders_tick_{}".format(tick))
	195	self.report.log_orders(open_orders, tick=tick)
	196	for order in open_orders:
7ba831c5 IB	197	status = order.get_status()
7ba831c5 IB	198	if status != "open":
f86ee140 IB	199	self.report.log_order(order, tick, finished=True)
	200	else:
	201	order.trade.update_order(order, tick)
7ba831c5 IB	202	if status == "error_disappeared":
	203	self.report.log_error("follow_orders",
	204	message="{} disappeared, recreating it".format(order))
	205	order.trade.prepare_order(
	206	compute_value=order.trade.tick_actions_recreate(tick))
	207
f86ee140 IB	208	self.report.log_stage("follow_orders_end")
f86ee140 IB	209
9db7d156 IB	210	def prepare_trades(self, base_currency="BTC", liquidity="medium",
	211	compute_value="average", repartition=None, only=None):
	212
7bd830a8 IB	213	self.report.log_stage("prepare_trades",
7bd830a8 IB	214	base_currency=base_currency, liquidity=liquidity,
9db7d156 IB	215	compute_value=compute_value, only=only,
9db7d156 IB	216	repartition=repartition)
f86ee140	217
9db7d156 IB	218	values_in_base = self.balances.in_currency(base_currency,
9db7d156 IB	219	compute_value=compute_value)
f86ee140	220	total_base_value = sum(values_in_base.values())
9db7d156 IB	221	new_repartition = self.balances.dispatch_assets(total_base_value,
	222	liquidity=liquidity, repartition=repartition)
	223	self.trades.compute_trades(values_in_base, new_repartition, only=only)
2308a1c4	224
ceb7fc4c	225	def print_tickers(self, base_currency="BTC"):
1f117ac7 IB	226	if base_currency is not None:
	227	self.report.print_log("total:")
	228	self.report.print_log(sum(self.balances.in_currency(base_currency).values()))
	229
	230	class Processor:
	231	scenarios = {
81d1db51 IB	232	"wait_for_cryptoportfolio": [
	233	{
	234	"name": "wait",
	235	"number": 1,
	236	"before": False,
	237	"after": True,
	238	"wait_for_recent": {},
	239	},
	240	],
ceb7fc4c IB	241	"print_balances": [
	242	{
	243	"name": "print_balances",
	244	"number": 1,
	245	"fetch_balances": ["begin"],
	246	"print_tickers": { "base_currency": "BTC" },
	247	}
	248	],
81d1db51 IB	249	"print_orders": [
	250	{
	251	"name": "wait",
	252	"number": 1,
ceb7fc4c IB	253	"before": True,
ceb7fc4c IB	254	"after": False,
81d1db51 IB	255	"wait_for_recent": {},
	256	},
	257	{
	258	"name": "make_orders",
	259	"number": 2,
	260	"before": False,
	261	"after": True,
	262	"fetch_balances": ["begin"],
	263	"prepare_trades": { "compute_value": "average" },
	264	"prepare_orders": { "compute_value": "average" },
	265	},
	266	],
1f117ac7 IB	267	"sell_needed": [
	268	{
	269	"name": "wait",
	270	"number": 0,
	271	"before": False,
	272	"after": True,
	273	"wait_for_recent": {},
	274	},
	275	{
	276	"name": "sell",
	277	"number": 1,
	278	"before": False,
	279	"after": True,
	280	"fetch_balances": ["begin", "end"],
	281	"prepare_trades": {},
	282	"prepare_orders": { "only": "dispose", "compute_value": "average" },
	283	"run_orders": {},
	284	"follow_orders": {},
	285	"close_trades": {},
	286	},
	287	{
	288	"name": "buy",
	289	"number": 2,
	290	"before": False,
	291	"after": True,
	292	"fetch_balances": ["begin", "end"],
	293	"prepare_trades": { "only": "acquire" },
	294	"prepare_orders": { "only": "acquire", "compute_value": "average" },
	295	"move_balances": {},
	296	"run_orders": {},
	297	"follow_orders": {},
	298	"close_trades": {},
	299	},
	300	],
	301	"sell_all": [
	302	{
	303	"name": "all_sell",
	304	"number": 1,
	305	"before": True,
	306	"after": False,
	307	"fetch_balances": ["begin", "end"],
	308	"prepare_trades": { "repartition": { "base_currency": (1, "long") } },
	309	"prepare_orders": { "compute_value": "average" },
	310	"run_orders": {},
	311	"follow_orders": {},
	312	"close_trades": {},
	313	},
	314	{
	315	"name": "wait",
	316	"number": 2,
	317	"before": False,
	318	"after": True,
	319	"wait_for_recent": {},
	320	},
	321	{
	322	"name": "all_buy",
	323	"number": 3,
	324	"before": False,
	325	"after": True,
	326	"fetch_balances": ["begin", "end"],
	327	"prepare_trades": {},
	328	"prepare_orders": { "compute_value": "average" },
	329	"move_balances": {},
	330	"run_orders": {},
331	"follow_orders": {},
332	"close_trades": {},
333	},
334	]
335	}
336
337	ordered_actions = [
338	"wait_for_recent", "prepare_trades", "prepare_orders",
339	"move_balances", "run_orders", "follow_orders",
ceb7fc4c	340	"close_trades", "print_tickers"]
1f117ac7 IB	341
	342	def __init__(self, market):
	343	self.market = market
	344
	345	def select_steps(self, scenario, step):
	346	if step == "all":
	347	return scenario
	348	elif step == "before" or step == "after":
ceb7fc4c	349	return list(filter(lambda x: x.get(step, False), scenario))
1f117ac7 IB	350	elif type(step) == int:
	351	return [scenario[step-1]]
	352	elif type(step) == str:
	353	return list(filter(lambda x: x["name"] == step, scenario))
	354	else:
	355	raise TypeError("Unknown step {}".format(step))
	356
ceb7fc4c IB	357	def can_process(self, scenario_name):
	358	return scenario_name in self.scenarios
	359
1f117ac7	360	def process(self, scenario_name, steps="all", **kwargs):
ceb7fc4c IB	361	if not self.can_process(scenario_name):
ceb7fc4c IB	362	raise TypeError("Unknown scenario {}".format(scenario_name))
1f117ac7 IB	363	scenario = self.scenarios[scenario_name]
	364	selected_steps = []
	365
	366	if type(steps) == str or type(steps) == int:
	367	selected_steps += self.select_steps(scenario, steps)
	368	else:
	369	for step in steps:
	370	selected_steps += self.select_steps(scenario, step)
	371	for step in selected_steps:
	372	self.process_step(scenario_name, step, kwargs)
	373
	374	def process_step(self, scenario_name, step, kwargs):
	375	process_name = "process_{}__{}_{}".format(scenario_name, step["number"], step["name"])
	376	self.market.report.log_stage("{}_begin".format(process_name))
	377	if "begin" in step.get("fetch_balances", []):
	378	self.market.balances.fetch_balances(tag="{}_begin".format(process_name))
	379
	380	for action in self.ordered_actions:
	381	if action in step:
	382	self.run_action(action, step[action], kwargs)
	383
	384	if "end" in step.get("fetch_balances", []):
	385	self.market.balances.fetch_balances(tag="{}_end".format(process_name))
	386	self.market.report.log_stage("{}_end".format(process_name))
	387
	388	def method_arguments(self, action):
	389	import inspect
	390
	391	if action == "wait_for_recent":
ada1b5f1	392	method = Portfolio.wait_for_recent
1f117ac7 IB	393	elif action == "prepare_trades":
	394	method = self.market.prepare_trades
	395	elif action == "prepare_orders":
	396	method = self.market.trades.prepare_orders
	397	elif action == "move_balances":
	398	method = self.market.move_balances
	399	elif action == "run_orders":
	400	method = self.market.trades.run_orders
	401	elif action == "follow_orders":
	402	method = self.market.follow_orders
	403	elif action == "close_trades":
	404	method = self.market.trades.close_trades
ceb7fc4c IB	405	elif action == "print_tickers":
ceb7fc4c IB	406	method = self.market.print_tickers
1f117ac7 IB	407
	408	signature = inspect.getfullargspec(method)
	409	defaults = signature.defaults or []
	410	kwargs = signature.args[-len(defaults):]
	411
	412	return [method, kwargs]
	413
	414	def parse_args(self, action, default_args, kwargs):
	415	method, allowed_arguments = self.method_arguments(action)
	416	args = {k: v for k, v in {default_args, kwargs}.items() if k in allowed_arguments }
	417
	418	if "repartition" in args and "base_currency" in args["repartition"]:
	419	r = args["repartition"]
	420	r[args.get("base_currency", "BTC")] = r.pop("base_currency")
	421
	422	return method, args
	423
	424	def run_action(self, action, default_args, kwargs):
	425	method, args = self.parse_args(action, default_args, kwargs)
	426
ada1b5f1	427	method(**args)