-import time
-from datetime import datetime, timedelta
+from datetime import datetime
from decimal import Decimal as D, ROUND_DOWN
-from json import JSONDecodeError
-from simplejson.errors import JSONDecodeError as SimpleJSONDecodeError
from ccxt import ExchangeError, InsufficientFunds, ExchangeNotAvailable, InvalidOrder, OrderNotCached, OrderNotFound
from retry import retry
-import requests
# FIXME: correctly handle web call timeouts
-class Portfolio:
- URL = "https://cryptoportfolio.io/wp-content/uploads/portfolio/json/cryptoportfolio.json"
- liquidities = {}
- data = None
- last_date = None
-
- @classmethod
- def wait_for_recent(cls, market, delta=4):
- cls.repartition(market, refetch=True)
- while cls.last_date is None or datetime.now() - cls.last_date > timedelta(delta):
- time.sleep(30)
- market.report.print_log("Attempt to fetch up-to-date cryptoportfolio")
- cls.repartition(market, refetch=True)
-
- @classmethod
- def repartition(cls, market, liquidity="medium", refetch=False):
- cls.parse_cryptoportfolio(market, refetch=refetch)
- liquidities = cls.liquidities[liquidity]
- return liquidities[cls.last_date]
-
- @classmethod
- def get_cryptoportfolio(cls, market):
- try:
- r = requests.get(cls.URL)
- market.report.log_http_request(r.request.method,
- r.request.url, r.request.body, r.request.headers, r)
- except Exception as e:
- market.report.log_error("get_cryptoportfolio", exception=e)
- return
- try:
- cls.data = r.json(parse_int=D, parse_float=D)
- except (JSONDecodeError, SimpleJSONDecodeError):
- cls.data = None
-
- @classmethod
- def parse_cryptoportfolio(cls, market, refetch=False):
- if refetch or cls.data is None:
- cls.get_cryptoportfolio(market)
-
- 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()))
-
class Computation:
computations = {
"default": lambda x, y: x[y],