New exchange rates system (#350)

* New exchange rates system * Fix formatting issues * Adaptive rates fetching * Fix worker * Try fixing * Fix database access * Final fix * Fix tests * Fix tests * Fix final test * More fixes * Fix all tests * Fixes * Fix flaky test * Fixes * Finalize tests * Finalize
bitcart · Apr 2, 2023 · e53ad38 · e53ad38
1 parent 89d8b43
commit e53ad38
Show file tree

Hide file tree

Showing 41 changed files with 777 additions and 225 deletions.
diff --git a/.circleci/config.yml b/.circleci/config.yml
@@ -126,6 +126,13 @@ jobs:
 
       - install_dependencies
 
+      - run:
+          name: prepare worker
+          command: |
+            TEST=true make migrate
+            TEST=true python3 worker.py
+          background: true
+
       - run:
           name: prepare daemon
           command: |

diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
@@ -8,7 +8,7 @@ repos:
     hooks:
       - id: yesqa
   - repo: https://github.com/psf/black
-    rev: 23.1.0
+    rev: 23.3.0
     hooks:
       - id: black
   - repo: https://github.com/pycqa/isort
@@ -45,7 +45,7 @@ repos:
         args: ["--remove"]
       - id: detect-private-key
   - repo: https://github.com/pre-commit/mirrors-prettier
-    rev: v3.0.0-alpha.4
+    rev: v3.0.0-alpha.6
     hooks:
       - id: prettier
   - repo: local

diff --git a/Makefile b/Makefile
@@ -44,6 +44,6 @@ fulcrum:
 	tests/functional/bootstrap/start_fulcrum.sh
 
 functional:
-	BTC_LIGHTNING=true pytest tests/functional/ --cov-append -n 0 ${TEST_ARGS}
+	BTC_LIGHTNING=true FUNCTIONAL_TESTS=true pytest tests/functional/ --cov-append -n 0 ${TEST_ARGS}
 
 ci: checkformat lint test
diff --git a/api/crud/invoices.py b/api/crud/invoices.py
@@ -93,9 +93,7 @@ async def determine_network_fee(coin, wallet, invoice, store, divisibility):  #
             coin_no_contract = await settings.settings.get_coin(
                 wallet.currency, {"xpub": wallet.xpub, **wallet.additional_xpub_data}
             )
-            rate_no_contract = await utils.wallets.get_rate(
-                wallet, invoice.currency, store.default_currency, coin=coin_no_contract
-            )
+            rate_no_contract = await utils.wallets.get_rate(wallet, invoice.currency, coin=coin_no_contract)
             return fee * rate_no_contract
         return fee
 
@@ -131,7 +129,7 @@ async def _create_payment_method(invoice, wallet, product, store, discounts, pro
         return method
     symbol = await utils.wallets.get_wallet_symbol(wallet, coin)
     divisibility = await utils.wallets.get_divisibility(wallet, coin)
-    rate = await utils.wallets.get_rate(wallet, invoice.currency, store.default_currency)
+    rate = await utils.wallets.get_rate(wallet, invoice.currency, store=store)
     price, discount_id = match_discount(invoice.price, wallet, invoice, discounts, promocode)
     support_underpaid = getattr(coin, "support_underpaid", True)
     request_price = price * (1 - (Decimal(store.checkout_settings.underpaid_percentage) / 100)) if support_underpaid else price
@@ -237,12 +235,10 @@ async def create_method_for_wallet(invoice, wallet, discounts, store, product, p
 
 async def update_invoice_payments(invoice, wallets_ids, discounts, store, product, promocode, start_time):
     logger.info(f"Started adding invoice payments for invoice {invoice.id}")
-    query = text(
-        """SELECT wallets.*
+    query = text("""SELECT wallets.*
     FROM   wallets
     JOIN   unnest((:wallets_ids)::varchar[]) WITH ORDINALITY t(id, ord) USING (id)
-    ORDER  BY t.ord;"""
-    )
+    ORDER  BY t.ord;""")
     wallets = await db.db.all(query, wallets_ids=wallets_ids)
     randomize_selection = store.checkout_settings.randomize_wallet_selection
     if randomize_selection:

diff --git a/api/events.py b/api/events.py
@@ -90,5 +90,8 @@ async def start_listening(custom_event_handler=None):  # pragma: no cover
         "send_notification": {
             "params": {"store_id", "text"},
         },
+        "rates_action": {
+            "params": {"func", "args", "task_id"},
+        },
     }
 )
diff --git a/api/ext/exchanges/__init__.py b/api/ext/exchanges/__init__.py
diff --git a/api/ext/exchanges/base.py b/api/ext/exchanges/base.py
@@ -0,0 +1,68 @@
+import asyncio
+import time
+from abc import ABCMeta, abstractmethod
+from decimal import Decimal
+from typing import Dict, List
+
+from bitcart.coin import Coin
+
+from api.ext.fxrate import ExchangePair
+from api.logger import get_exception_message, get_logger
+
+logger = get_logger(__name__)
+
+REFRESH_TIME = 150
+EXCHANGE_ACTIVE_TIME = 12 * 60 * 60
+
+# Adaptive system: avoid refresh on call except for first time, then refresh in background
+# If exchange wasn't used for 12 hours, stop refreshing in background
+
+
+def get_inverse_dict(d):
+    return {str(ExchangePair(k).inverse()): 1 / v for k, v in d.items()}
+
+
+class BaseExchange(metaclass=ABCMeta):
+    def __init__(self, coins: List[Coin], contracts: Dict[str, list]):
+        self.coins = coins
+        self.contracts = contracts
+        self.quotes = {}
+        self.last_refresh = 0
+        self.last_called = 0
+        self.lock = asyncio.Lock()
+        asyncio.create_task(self.refresh_task())
+
+    async def _check_fresh(self, called=False):
+        async with self.lock:
+            cur_time = time.time()
+            if (called and (self.last_refresh == 0 or cur_time - self.last_called > EXCHANGE_ACTIVE_TIME)) or (
+                not called and cur_time - self.last_refresh > REFRESH_TIME
+            ):
+                try:
+                    await self.refresh()
+                    self.quotes.update(get_inverse_dict(self.quotes))
+                except Exception as e:
+                    logger.error(f"Failed refreshing exchange rates:\n{get_exception_message(e)}")
+                self.last_refresh = cur_time
+            if called:
+                self.last_called = cur_time
+
+    async def get_rate(self, pair=None):
+        await self._check_fresh(True)
+        if pair is None:
+            return self.quotes
+        return self.quotes.get(pair, Decimal("NaN"))
+
+    async def get_fiat_currencies(self):
+        await self._check_fresh(True)
+        return list(map(lambda x: x.split("_")[1], self.quotes))
+
+    @abstractmethod
+    async def refresh(self):
+        pass
+
+    async def refresh_task(self):
+        while True:
+            if time.time() - self.last_called <= EXCHANGE_ACTIVE_TIME:
+                await self._check_fresh()
+            await asyncio.sleep(REFRESH_TIME + 1)
diff --git a/api/ext/exchanges/coingecko.py b/api/ext/exchanges/coingecko.py
@@ -0,0 +1,113 @@
+import asyncio
+import json
+
+from api import settings, utils
+from api.ext.exchanges.base import BaseExchange
+
+
+async def fetch_delayed(*args, delay=1, **kwargs):
+    resp, data = await utils.common.send_request(*args, return_json=False)
+    if resp.status == 429:
+        if delay < 60:
+            await asyncio.sleep(delay)
+            return await fetch_delayed(*args, **kwargs, delay=delay * 2)
+        resp.raise_for_status()
+    data = json.loads(data)
+    if kwargs.get("return_json", True):
+        return data
+    return resp, data
+
+
+def find_by_coin(all_coins, coin):
+    coingecko_id = settings.settings.exchange_rates.coingecko_ids.get(coin.coin_name.lower())
+    if coingecko_id:
+        for currency in all_coins:
+            if currency.get("id", "").lower() == coingecko_id.lower():
+                return currency
+    for currency in all_coins:
+        if currency.get("name", "").lower() == coin.friendly_name.lower():
+            return currency
+    for currency in all_coins:
+        if currency.get("symbol", "").lower() == coin.coin_name.lower():
+            return currency
+
+
+def find_by_contract(all_coins, contract):
+    for currency in all_coins:
+        if contract in currency.get("platforms", {}).values():
+            return currency
+
+
+def find_id(all_coins, x):
+    for coin in all_coins:
+        if coin["id"] == x:
+            return coin["symbol"]
+
+
+class CoingeckoExchange(BaseExchange):
+    def __init__(self, coins, contracts):
+        super().__init__(coins, contracts)
+        self.coins_cache = {}
+
+    async def refresh(self):
+        vs_currencies = await fetch_delayed("GET", "https://api.coingecko.com/api/v3/simple/supported_vs_currencies")
+        if not self.coins_cache:
+            self.coins_cache = await fetch_delayed("GET", "https://api.coingecko.com/api/v3/coins/list?include_platform=true")
+        coins = []
+        for coin in self.coins.copy():
+            currency = find_by_coin(self.coins_cache, coin)
+            if currency:
+                coins.append(currency["id"])
+        for contracts in self.contracts.copy().values():
+            for contract in contracts:
+                currency = find_by_contract(self.coins_cache, contract)
+                if currency:
+                    coins.append(currency["id"])
+        data = await fetch_delayed(
+            "GET",
+            (
+                f"https://api.coingecko.com/api/v3/simple/price?ids={','.join(coins)}"
+                f"&vs_currencies={','.join(vs_currencies)}&precision=full"
+            ),
+        )
+        self.quotes = {
+            f"{find_id(self.coins_cache, k).upper()}_{k2.upper()}": utils.common.precise_decimal(v2)
+            for k, v in data.items()
+            for k2, v2 in v.items()
+        }
+
+
+def coingecko_based_exchange(name):
+    class CoingeckoBasedExchange(BaseExchange):
+        def __init__(self, coins, contracts):
+            super().__init__(coins, contracts)
+            self.coins_cache = {}
+
+        async def refresh(self):
+            if not self.coins_cache:
+                self.coins_cache = await fetch_delayed("GET", "https://api.coingecko.com/api/v3/coins/list")
+            coins = []
+            for coin in self.coins.copy():
+                currency = find_by_coin(self.coins_cache, coin)
+                if currency:
+                    coins.append(currency["id"])
+            self.quotes = await self.fetch_rates(coins)
+
+        async def fetch_rates(self, coins, page=1):
+            base_url = f"https://api.coingecko.com/api/v3/exchanges/{name}/tickers"
+            page = 1
+            resp, data = await fetch_delayed("GET", f"{base_url}?page={page}&coin_ids={','.join(coins)}", return_json=False)
+            result = {f"{x['base']}_{x['target']}": utils.common.precise_decimal(x["last"]) for x in data["tickers"]}
+            total = resp.headers.get("total")
+            per_page = resp.headers.get("per-page")
+            if page == 1 and total and per_page:
+                total = int(total)
+                per_page = int(per_page)
+                total_pages = total // per_page
+                if total % per_page != 0:
+                    total_pages += 1
+                for page in range(2, total_pages + 1):
+                    result.update(await self.fetch_rates(page=page))
+            return result
+
+    return CoingeckoBasedExchange
diff --git a/api/ext/exchanges/coinrules.py b/api/ext/exchanges/coinrules.py
@@ -0,0 +1,46 @@
+class BTC:
+    coingecko_id = "bitcoin"
+
+
+class BCH:
+    coingecko_id = "bitcoin-cash"
+
+
+class LTC:
+    coingecko_id = "litecoin"
+
+
+class BSTY:
+    coingecko_id = "globalboost"
+
+
+class XRG:
+    coingecko_id = "bitcoin"  # not working
+
+
+class ETH:
+    coingecko_id = "ethereum"
+
+
+class BNB:
+    coingecko_id = "binancecoin"
+
+
+class SBCH:
+    default_rule = "SBCH_X = BCH_X"
+
+
+class MATIC:
+    coingecko_id = "matic-network"
+
+
+class TRX:
+    coingecko_id = "tron"
+
+
+class GRS:
+    coingecko_id = "groestlcoin"
+
+
+class XMR:
+    coingecko_id = "monero"
diff --git a/api/ext/exchanges/fiat.py b/api/ext/exchanges/fiat.py
@@ -0,0 +1,10 @@
+from api import utils
+from api.ext.exchanges.base import BaseExchange
+
+
+class FiatExchange(BaseExchange):
+    async def refresh(self):
+        result = await utils.common.send_request(
+            "GET", "https://cdn.jsdelivr.net/gh/fawazahmed0/currency-api@1/latest/currencies/usd.json"
+        )
+        self.quotes = {f"USD_{k.upper()}": utils.common.precise_decimal(v) for k, v in result["usd"].items()}
diff --git a/api/ext/exchanges/kraken.py b/api/ext/exchanges/kraken.py
@@ -0,0 +1,10 @@
+from api import utils
+from api.ext.exchanges.base import BaseExchange
+
+
+class Kraken(BaseExchange):
+    async def refresh(self):
+        ccys = ["EUR", "USD", "CAD", "GBP", "JPY"]
+        pairs = ["XBT%s" % c for c in ccys]
+        json = await utils.common.send_request("GET", "https://api.kraken.com/0/public/Ticker?pair=%s" % ",".join(pairs))
+        self.quotes = {f"BTC_{k[-3:]}": utils.common.precise_decimal(str(v["c"][0])) for k, v in json["result"].items()}