]> git.immae.eu Git - perso/Immae/Projets/Cryptomonnaies/Cryptoportfolio/Trader.git/blobdiff - store.py
Fix vanishing orders
[perso/Immae/Projets/Cryptomonnaies/Cryptoportfolio/Trader.git] / store.py
index d9038b7fa32d9adb46d5a57acc5d312268bc6a77..b3ada4567e38a28613efd0159dafd33ece105a0d 100644 (file)
--- a/store.py
+++ b/store.py
@@ -1,20 +1,34 @@
+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):
         self.market = market
         self.verbose_print = verbose_print
 
+        self.print_logs = []
         self.logs = []
 
+    def merge(self, other_report):
+        self.logs += other_report.logs
+        self.logs.sort(key=lambda x: x["date"])
+
+        self.print_logs += other_report.print_logs
+        self.print_logs.sort(key=lambda x: x[0])
+
     def print_log(self, message):
-        message = str(message)
+        now = datetime.now()
+        message = "{:%Y-%m-%d %H:%M:%S}: {}".format(now, str(message))
+        self.print_logs.append([now, message])
         if self.verbose_print:
             print(message)
 
@@ -22,12 +36,22 @@ class ReportStore:
         hash_["date"] = datetime.now()
         self.logs.append(hash_)
 
+    @staticmethod
+    def default_json_serial(obj):
+        if isinstance(obj, (datetime, date)):
+            return obj.isoformat()
+        return str(obj)
+
     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)
+        return json.dumps(self.logs, default=self.default_json_serial, indent="  ")
+
+    def to_json_array(self):
+        for log in (x.copy() for x in self.logs):
+            yield (
+                    log.pop("date"),
+                    log.pop("type"),
+                    json.dumps(log, default=self.default_json_serial, indent="  ")
+                    )
 
     def set_verbose(self, verbose_print):
         self.verbose_print = verbose_print
@@ -213,7 +237,7 @@ class BalanceStore:
 
     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():
@@ -301,4 +325,165 @@ class TradeStore:
         for order in self.all_orders(state="open"):
             order.get_status()
 
+class NoopLock:
+    def __enter__(self, *args):
+        pass
+    def __exit__(self, *args):
+        pass
+
+class LockedVar:
+    def __init__(self, value):
+        self.lock = NoopLock()
+        self.val = value
+
+    def start_lock(self):
+        import threading
+        self.lock = threading.Lock()
+
+    def set(self, value):
+        with self.lock:
+            self.val = value
+
+    def get(self, key=None):
+        with self.lock:
+            if key is not None and isinstance(self.val, dict):
+                return self.val.get(key)
+            else:
+                return self.val
+
+    def __getattr__(self, key):
+        with self.lock:
+            return getattr(self.val, key)
+
+class Portfolio:
+    URL = "https://cryptoportfolio.io/wp-content/uploads/portfolio/json/cryptoportfolio.json"
+    data = LockedVar(None)
+    liquidities = LockedVar({})
+    last_date = LockedVar(None)
+    report = LockedVar(ReportStore(None))
+    worker = None
+    worker_started = False
+    worker_notify = None
+    callback = None
+
+    @classmethod
+    def start_worker(cls, poll=30):
+        import threading
+
+        cls.worker = threading.Thread(name="portfolio", daemon=True,
+                target=cls.wait_for_notification, kwargs={"poll": poll})
+        cls.worker_notify = threading.Event()
+        cls.callback = threading.Event()
+
+        cls.last_date.start_lock()
+        cls.liquidities.start_lock()
+        cls.report.start_lock()
+
+        cls.worker_started = True
+        cls.worker.start()
+
+    @classmethod
+    def is_worker_thread(cls):
+        if cls.worker is None:
+            return False
+        else:
+            import threading
+            return cls.worker == threading.current_thread()
+
+    @classmethod
+    def wait_for_notification(cls, poll=30):
+        if not cls.is_worker_thread():
+            raise RuntimeError("This method needs to be ran with the worker")
+        while cls.worker_started:
+            cls.worker_notify.wait()
+            cls.worker_notify.clear()
+            cls.report.print_log("Fetching cryptoportfolio")
+            cls.get_cryptoportfolio(refetch=True)
+            cls.callback.set()
+            time.sleep(poll)
+
+    @classmethod
+    def notify_and_wait(cls):
+        cls.callback.clear()
+        cls.worker_notify.set()
+        cls.callback.wait()
+
+    @classmethod
+    def wait_for_recent(cls, delta=4, poll=30):
+        cls.get_cryptoportfolio()
+        while cls.last_date.get() is None or datetime.now() - cls.last_date.get() > timedelta(delta):
+            if cls.worker is None:
+                time.sleep(poll)
+                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.get(liquidity)
+        return liquidities[cls.last_date.get()]
+
+    @classmethod
+    def get_cryptoportfolio(cls, refetch=False):
+        if cls.data.get() is not None and not refetch:
+            return
+        if cls.worker is not None and not cls.is_worker_thread():
+            cls.notify_and_wait()
+            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.set(r.json(parse_int=D, parse_float=D))
+            cls.parse_cryptoportfolio()
+        except (JSONDecodeError, SimpleJSONDecodeError):
+            cls.data.set(None)
+            cls.last_date.set(None)
+            cls.liquidities.set({})
+
+    @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):
+            if "weights" not in portfolio_hash:
+                return {}
+            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.get("portfolio_1"))
+        medium_liquidity = parse_weights(cls.data.get("portfolio_2"))
+
+        cls.liquidities.set({
+            "medium": medium_liquidity,
+            "high":   high_liquidity,
+            })
+        cls.last_date.set(max(
+            max(medium_liquidity.keys(), default=datetime(1, 1, 1)),
+            max(high_liquidity.keys(), default=datetime(1, 1, 1))
+            ))
+