import time import requests import portfolio import simplejson as json from decimal import Decimal as D, ROUND_DOWN from datetime import date, datetime, timedelta import inspect from json import JSONDecodeError from simplejson.errors import JSONDecodeError as SimpleJSONDecodeError __all__ = ["Portfolio", "BalanceStore", "ReportStore", "TradeStore"] class ReportStore: def __init__(self, market, verbose_print=True): self.market = market self.verbose_print = verbose_print self.logs = [] def merge(self, other_report): self.logs += other_report.logs self.logs.sort(key=lambda x: x["date"]) def print_log(self, message): message = str(message) if self.verbose_print: print(message) def add_log(self, hash_): hash_["date"] = datetime.now() self.logs.append(hash_) def to_json(self): def default_json_serial(obj): if isinstance(obj, (datetime, date)): return obj.isoformat() return str(obj) return json.dumps(self.logs, default=default_json_serial, indent=" ") def set_verbose(self, verbose_print): self.verbose_print = verbose_print def log_stage(self, stage, **kwargs): def as_json(element): if callable(element): return inspect.getsource(element).strip() elif hasattr(element, "as_json"): return element.as_json() else: return element args = { k: as_json(v) for k, v in kwargs.items() } args_str = ["{}={}".format(k, v) for k, v in args.items()] self.print_log("-" * (len(stage) + 8)) self.print_log("[Stage] {} {}".format(stage, ", ".join(args_str))) self.add_log({ "type": "stage", "stage": stage, "args": args, }) def log_balances(self, tag=None): self.print_log("[Balance]") for currency, balance in self.market.balances.all.items(): self.print_log("\t{}".format(balance)) self.add_log({ "type": "balance", "tag": tag, "balances": self.market.balances.as_json() }) def log_tickers(self, amounts, other_currency, compute_value, type): values = {} rates = {} if callable(compute_value): compute_value = inspect.getsource(compute_value).strip() for currency, amount in amounts.items(): values[currency] = amount.as_json()["value"] rates[currency] = amount.rate self.add_log({ "type": "tickers", "compute_value": compute_value, "balance_type": type, "currency": other_currency, "balances": values, "rates": rates, "total": sum(amounts.values()).as_json()["value"] }) def log_dispatch(self, amount, amounts, liquidity, repartition): self.add_log({ "type": "dispatch", "liquidity": liquidity, "repartition_ratio": repartition, "total_amount": amount.as_json(), "repartition": { k: v.as_json()["value"] for k, v in amounts.items() } }) def log_trades(self, matching_and_trades, only): trades = [] for matching, trade in matching_and_trades: trade_json = trade.as_json() trade_json["skipped"] = not matching trades.append(trade_json) self.add_log({ "type": "trades", "only": only, "debug": self.market.debug, "trades": trades }) def log_orders(self, orders, tick=None, only=None, compute_value=None): if callable(compute_value): compute_value = inspect.getsource(compute_value).strip() self.print_log("[Orders]") self.market.trades.print_all_with_order(ind="\t") self.add_log({ "type": "orders", "only": only, "compute_value": compute_value, "tick": tick, "orders": [order.as_json() for order in orders if order is not None] }) def log_order(self, order, tick, finished=False, update=None, new_order=None, compute_value=None): if callable(compute_value): compute_value = inspect.getsource(compute_value).strip() if finished: self.print_log("[Order] Finished {}".format(order)) elif update == "waiting": self.print_log("[Order] {}, tick {}, waiting".format(order, tick)) elif update == "adjusting": self.print_log("[Order] {}, tick {}, cancelling and adjusting to {}".format(order, tick, new_order)) elif update == "market_fallback": self.print_log("[Order] {}, tick {}, fallbacking to market value".format(order, tick)) elif update == "market_adjust": self.print_log("[Order] {}, tick {}, market value, cancelling and adjusting to {}".format(order, tick, new_order)) self.add_log({ "type": "order", "tick": tick, "update": update, "order": order.as_json(), "compute_value": compute_value, "new_order": new_order.as_json() if new_order is not None else None }) def log_move_balances(self, needed, moving): self.add_log({ "type": "move_balances", "debug": self.market.debug, "needed": { k: v.as_json()["value"] if isinstance(v, portfolio.Amount) else v for k, v in needed.items() }, "moving": { k: v.as_json()["value"] if isinstance(v, portfolio.Amount) else v for k, v in moving.items() }, }) def log_http_request(self, method, url, body, headers, response): self.add_log({ "type": "http_request", "method": method, "url": url, "body": body, "headers": headers, "status": response.status_code, "response": response.text }) def log_error(self, action, message=None, exception=None): self.print_log("[Error] {}".format(action)) if exception is not None: self.print_log(str("\t{}: {}".format(exception.__class__.__name__, exception))) if message is not None: self.print_log("\t{}".format(message)) self.add_log({ "type": "error", "action": action, "exception_class": exception.__class__.__name__ if exception is not None else None, "exception_message": str(exception) if exception is not None else None, "message": message, }) def log_debug_action(self, action): self.print_log("[Debug] {}".format(action)) self.add_log({ "type": "debug_action", "action": action, }) class BalanceStore: def __init__(self, market): self.market = market self.all = {} def currencies(self): return self.all.keys() def in_currency(self, other_currency, compute_value="average", type="total"): amounts = {} for currency, balance in self.all.items(): other_currency_amount = getattr(balance, type)\ .in_currency(other_currency, self.market, compute_value=compute_value) amounts[currency] = other_currency_amount self.market.report.log_tickers(amounts, other_currency, compute_value, type) return amounts def fetch_balances(self, tag=None): all_balances = self.market.ccxt.fetch_all_balances() for currency, balance in all_balances.items(): if balance["exchange_total"] != 0 or balance["margin_total"] != 0 or \ currency in self.all: self.all[currency] = portfolio.Balance(currency, balance) self.market.report.log_balances(tag=tag) def dispatch_assets(self, amount, liquidity="medium", repartition=None): if repartition is None: repartition = Portfolio.repartition(liquidity=liquidity) sum_ratio = sum([v[0] for k, v in repartition.items()]) amounts = {} for currency, (ptt, trade_type) in repartition.items(): amounts[currency] = ptt * amount / sum_ratio if trade_type == "short": amounts[currency] = - amounts[currency] self.all.setdefault(currency, portfolio.Balance(currency, {})) self.market.report.log_dispatch(amount, amounts, liquidity, repartition) return amounts def as_json(self): return { k: v.as_json() for k, v in self.all.items() } class TradeStore: def __init__(self, market): self.market = market self.all = [] @property def pending(self): return list(filter(lambda t: t.pending, self.all)) def compute_trades(self, values_in_base, new_repartition, only=None): computed_trades = [] base_currency = sum(values_in_base.values()).currency for currency in self.market.balances.currencies(): if currency == base_currency: continue value_from = values_in_base.get(currency, portfolio.Amount(base_currency, 0)) value_to = new_repartition.get(currency, portfolio.Amount(base_currency, 0)) if value_from.value * value_to.value < 0: computed_trades.append(self.trade_if_matching( value_from, portfolio.Amount(base_currency, 0), currency, only=only)) computed_trades.append(self.trade_if_matching( portfolio.Amount(base_currency, 0), value_to, currency, only=only)) else: computed_trades.append(self.trade_if_matching( value_from, value_to, currency, only=only)) for matching, trade in computed_trades: if matching: self.all.append(trade) self.market.report.log_trades(computed_trades, only) def trade_if_matching(self, value_from, value_to, currency, only=None): trade = portfolio.Trade(value_from, value_to, currency, self.market) matching = only is None or trade.action == only return [matching, trade] def prepare_orders(self, only=None, compute_value="default"): orders = [] for trade in self.pending: if only is None or trade.action == only: orders.append(trade.prepare_order(compute_value=compute_value)) self.market.report.log_orders(orders, only, compute_value) def close_trades(self): for trade in self.all: trade.close() def print_all_with_order(self, ind=""): for trade in self.all: trade.print_with_order(ind=ind) def run_orders(self): orders = self.all_orders(state="pending") for order in orders: order.run() self.market.report.log_stage("run_orders") self.market.report.log_orders(orders) def all_orders(self, state=None): all_orders = sum(map(lambda v: v.orders, self.all), []) if state is None: return all_orders else: return list(filter(lambda o: o.status == state, all_orders)) def update_all_orders_status(self): for order in self.all_orders(state="open"): order.get_status() class Portfolio: URL = "https://cryptoportfolio.io/wp-content/uploads/portfolio/json/cryptoportfolio.json" liquidities = {} data = None last_date = None report = ReportStore(None) @classmethod def wait_for_recent(cls, delta=4): cls.get_cryptoportfolio() while cls.last_date is None or datetime.now() - cls.last_date > timedelta(delta): time.sleep(30) cls.report.print_log("Attempt to fetch up-to-date cryptoportfolio") cls.get_cryptoportfolio(refetch=True) @classmethod def repartition(cls, liquidity="medium"): cls.get_cryptoportfolio() liquidities = cls.liquidities[liquidity] return liquidities[cls.last_date] @classmethod def get_cryptoportfolio(cls, refetch=False): if cls.data is not None and not refetch: return try: r = requests.get(cls.URL) cls.report.log_http_request(r.request.method, r.request.url, r.request.body, r.request.headers, r) except Exception as e: cls.report.log_error("get_cryptoportfolio", exception=e) return try: cls.data = r.json(parse_int=D, parse_float=D) cls.parse_cryptoportfolio() except (JSONDecodeError, SimpleJSONDecodeError): cls.data = None cls.liquidities = {} @classmethod def parse_cryptoportfolio(cls): def filter_weights(weight_hash): if weight_hash[1][0] == 0: return False if weight_hash[0] == "_row": return False return True def clean_weights(i): def clean_weights_(h): if h[0].endswith("s"): return [h[0][0:-1], (h[1][i], "short")] else: return [h[0], (h[1][i], "long")] return clean_weights_ def parse_weights(portfolio_hash): weights_hash = portfolio_hash["weights"] weights = {} for i in range(len(weights_hash["_row"])): date = datetime.strptime(weights_hash["_row"][i], "%Y-%m-%d") weights[date] = dict(filter( filter_weights, map(clean_weights(i), weights_hash.items()))) return weights high_liquidity = parse_weights(cls.data["portfolio_1"]) medium_liquidity = parse_weights(cls.data["portfolio_2"]) cls.liquidities = { "medium": medium_liquidity, "high": high_liquidity, } cls.last_date = max(max(medium_liquidity.keys()), max(high_liquidity.keys()))