+import time
+import requests
import portfolio
import simplejson as json
from decimal import Decimal as D, ROUND_DOWN
-from datetime import date, datetime
+from datetime import date, datetime, timedelta
+import inspect
+from json import JSONDecodeError
+from simplejson.errors import JSONDecodeError as SimpleJSONDecodeError
-__all__ = ["BalanceStore", "ReportStore", "TradeStore"]
+__all__ = ["Portfolio", "BalanceStore", "ReportStore", "TradeStore"]
class ReportStore:
def __init__(self, market, verbose_print=True):
if isinstance(obj, (datetime, date)):
return obj.isoformat()
return str(obj)
- return json.dumps(self.logs, default=default_json_serial)
+ 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):
+ 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))
+ self.print_log("[Stage] {} {}".format(stage, ", ".join(args_str)))
self.add_log({
"type": "stage",
"stage": stage,
+ "args": args,
})
def log_balances(self, tag=None):
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
})
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({
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":
def dispatch_assets(self, amount, liquidity="medium", repartition=None):
if repartition is None:
- repartition = portfolio.Portfolio.repartition(self.market, liquidity=liquidity)
+ 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]
- if currency not in self.all:
- self.all[currency] = portfolio.Balance(currency, {})
+ self.all.setdefault(currency, portfolio.Balance(currency, {}))
self.market.report.log_dispatch(amount, amounts, liquidity, repartition)
return amounts
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
def prepare_orders(self, only=None, compute_value="default"):
orders = []
- for trade in self.all:
+ 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)
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()))
+
projects / perso / Immae / Projets / Cryptomonnaies / Cryptoportfolio / Trader.git / blobdiff