]> git.immae.eu Git - perso/Immae/Projets/Cryptomonnaies/Cryptoportfolio/Trader.git/blobdiff - store.py
Merge branch 'dev'
[perso/Immae/Projets/Cryptomonnaies/Cryptoportfolio/Trader.git] / store.py
index 76cfec88f4648e55cd9f1b084204e46f5ad730b4..425f08d7ce85f322addb44219ed7ab0341926132 100644 (file)
--- a/store.py
+++ b/store.py
@@ -99,7 +99,7 @@ class ReportStore:
             "args": args,
             })
 
-    def log_balances(self, tag=None, tickers=None,
+    def log_balances(self, tag=None, checkpoint=None, tickers=None,
             ticker_currency=None, compute_value=None, type=None):
         self.print_log("[Balance]")
         for currency, balance in self.market.balances.all.items():
@@ -108,6 +108,7 @@ class ReportStore:
         log = {
                 "type": "balance",
                 "tag": tag,
+                "checkpoint": checkpoint,
                 "balances": self.market.balances.as_json()
                 }
 
@@ -303,7 +304,8 @@ class BalanceStore:
                 compute_value, type)
         return amounts
 
-    def fetch_balances(self, tag=None, add_portfolio=False, log_tickers=False,
+    def fetch_balances(self, tag=None, add_portfolio=False, liquidity="medium",
+            checkpoint=None, log_tickers=False, add_usdt=False,
             ticker_currency="BTC", ticker_compute_value="average", ticker_type="total"):
         all_balances = self.market.ccxt.fetch_all_balances()
         for currency, balance in all_balances.items():
@@ -311,15 +313,60 @@ class BalanceStore:
                     currency in self.all:
                 self.all[currency] = portfolio.Balance(currency, balance)
         if add_portfolio:
-            for currency in Portfolio.repartition(from_cache=True):
+            for currency in Portfolio.repartition(from_cache=True, liquidity=liquidity):
                 self.all.setdefault(currency, portfolio.Balance(currency, {}))
+        if add_usdt:
+            self.all.setdefault("USDT", portfolio.Balance("USDT", {}))
         if log_tickers:
             tickers = self.in_currency(ticker_currency, compute_value=ticker_compute_value, type=ticker_type)
-            self.market.report.log_balances(tag=tag,
+            self.market.report.log_balances(tag=tag, checkpoint=checkpoint,
                     tickers=tickers, ticker_currency=ticker_currency,
                     compute_value=ticker_compute_value, type=ticker_type)
         else:
-            self.market.report.log_balances(tag=tag)
+            self.market.report.log_balances(tag=tag, checkpoint=checkpoint)
+
+    def available_balances_for_repartition(self,
+            compute_value="average", base_currency="BTC",
+            liquidity="medium", repartition=None):
+        if repartition is None:
+            repartition = Portfolio.repartition(liquidity=liquidity)
+        base_currency_balance = self.all.get(base_currency)
+
+        if base_currency_balance is None:
+            total_base_value = portfolio.Amount(base_currency, 0)
+        else:
+            total_base_value = base_currency_balance.exchange_free + \
+                    base_currency_balance.margin_available - \
+                    base_currency_balance.margin_in_position
+
+        amount_in_position = {}
+
+        # Compute balances already in the target position
+        for currency, (ptt, trade_type) in repartition.items():
+            amount_in_position[currency] = portfolio.Amount(base_currency, 0)
+            balance = self.all.get(currency)
+            if currency != base_currency and balance is not None:
+                if trade_type == "short":
+                    amount = balance.margin_borrowed
+                else:
+                    amount = balance.exchange_free + balance.exchange_used
+                amount_in_position[currency] = amount.in_currency(base_currency,
+                    self.market, compute_value=compute_value)
+                total_base_value += amount_in_position[currency]
+
+        # recursively delete more-than-filled positions from the wanted
+        # repartition
+        did_delete = True
+        while did_delete:
+            did_delete = False
+            sum_ratio = sum([v[0] for k, v in repartition.items()])
+            current_base_value = total_base_value
+            for currency, (ptt, trade_type) in repartition.copy().items():
+                if amount_in_position[currency] > current_base_value * ptt / sum_ratio:
+                    did_delete = True
+                    del(repartition[currency])
+                    total_base_value -= amount_in_position[currency]
+        return repartition, total_base_value, amount_in_position
 
     def dispatch_assets(self, amount, liquidity="medium", repartition=None):
         if repartition is None:
@@ -452,13 +499,32 @@ class Portfolio:
     worker_started = False
     worker_notify = None
     callback = None
+    poll_started_at = None
+
+    @classmethod
+    def next_wait_time(cls):
+        now = datetime.datetime.now()
+        if cls.poll_started_at is None:
+            cls.poll_started_at = now
+        delta = now - cls.poll_started_at
+
+        if delta < datetime.timedelta(minutes=30):
+            return 30
+        elif delta < datetime.timedelta(hours=1):
+            return 60
+        elif delta < datetime.timedelta(hours=4):
+            return 5*60
+        elif delta < datetime.timedelta(days=1):
+            return 60*60
+        else:
+            raise Exception("Too long waiting")
 
     @classmethod
-    def start_worker(cls, poll=30):
+    def start_worker(cls):
         import threading
 
         cls.worker = threading.Thread(name="portfolio", daemon=True,
-                target=cls.wait_for_notification, kwargs={"poll": poll})
+                target=cls.wait_for_notification)
         cls.worker_notify = threading.Event()
         cls.callback = threading.Event()
 
@@ -479,7 +545,7 @@ class Portfolio:
             return cls.worker == threading.current_thread()
 
     @classmethod
-    def wait_for_notification(cls, poll=30):
+    def wait_for_notification(cls):
         if not cls.is_worker_thread():
             raise RuntimeError("This method needs to be ran with the worker")
         while cls.worker_started:
@@ -489,7 +555,10 @@ class Portfolio:
                 cls.report.print_log("[Worker] Fetching cryptoportfolio")
                 cls.get_cryptoportfolio(refetch=True)
                 cls.callback.set()
-                time.sleep(poll)
+                try:
+                    time.sleep(cls.next_wait_time())
+                except Exception:
+                    cls.stop_worker()
 
     @classmethod
     def stop_worker(cls):
@@ -503,11 +572,11 @@ class Portfolio:
         cls.callback.wait()
 
     @classmethod
-    def wait_for_recent(cls, delta=4, poll=30):
+    def wait_for_recent(cls, delta=4):
         cls.get_cryptoportfolio()
         while cls.last_date.get() is None or datetime.datetime.now() - cls.last_date.get() > datetime.timedelta(delta):
             if cls.worker is None:
-                time.sleep(poll)
+                time.sleep(cls.next_wait_time())
                 cls.report.print_log("Attempt to fetch up-to-date cryptoportfolio")
             cls.get_cryptoportfolio(refetch=True)
 
@@ -517,13 +586,16 @@ class Portfolio:
             cls.retrieve_cryptoportfolio()
         cls.get_cryptoportfolio()
         liquidities = cls.liquidities.get(liquidity)
-        return liquidities[cls.last_date.get()]
+        if liquidities is not None and cls.last_date.get() in liquidities:
+            return liquidities[cls.last_date.get()].copy()
 
     @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():
+            if not cls.worker_started:
+                raise Exception("Portfolio worker is down and no usable data is present")
             cls.notify_and_wait()
             return
         try:
@@ -537,7 +609,7 @@ class Portfolio:
             cls.data.set(r.json(parse_int=D, parse_float=D))
             cls.parse_cryptoportfolio()
             cls.store_cryptoportfolio()
-        except (JSONDecodeError, SimpleJSONDecodeError):
+        except (AssertionError, JSONDecodeError, SimpleJSONDecodeError):
             cls.data.set(None)
             cls.last_date.set(None)
             cls.liquidities.set({})
@@ -600,6 +672,8 @@ class Portfolio:
         high_liquidity = parse_weights(cls.data.get("portfolio_1"))
         medium_liquidity = parse_weights(cls.data.get("portfolio_2"))
 
+        assert len(high_liquidity) > 0
+        assert len(medium_liquidity) > 0
         cls.liquidities.set({
             "medium": medium_liquidity,
             "high":   high_liquidity,