Bump urllib3 from 1.26.16 to 1.26.18

.dockerignore

@@ -2,6 +2,10 @@
 .idea
 .git
 venv
+.venv
 nginx
+.pytest_cache
+htmlcov
 .gitignore
 .pre-commit-config.yaml
+.coverage

.envs/.local/.producer

@@ -17,3 +17,17 @@ LINK-USDT
 FTM-USDT
 PEPE-USDT
 "
+
+COINBASE_COINS="
+BTC-USD
+ETH-USD
+LINK-USD
+"
+
+KRAKEN_COINS="
+BTC/USD
+ETH/USD ETH/BTC
+LINK/USD
+FTM/USD
+PEPE/USD
+"

.envs/.local/.web

@@ -1,6 +1,7 @@
 DEBUG=True
 
 DB_PATH=/app/exchange_radar/web/data/db.json
+DB_TABLE_MAX_ROWS=64
 
 TRADES_SOCKET_URL=ws://127.0.0.1:9000/trades/{coin}
 TRADES_WHALES_SOCKET_URL=ws://127.0.0.1:9000/trades/{coin}/whales
@@ -12,4 +13,7 @@ TRADES_WHALES_HOST_URL=http://127.0.0.1:9000/feed/{coin}/whales
 TRADES_DOLPHINS_HOST_URL=http://127.0.0.1:9000/feed/{coin}/dolphins
 TRADES_OCTOPUSES_HOST_URL=http://127.0.0.1:9000/feed/{coin}/octopuses
 
-DB_TABLE_MAX_ROWS=64
+BINANCE=LTO,ETH,BTC,LINK,FTM,PEPE
+COINBASE=ETH,BTC,LINK
+KRAKEN=ETH,BTC,LINK,FTM,PEPE
+KUCOIN=LTO,ETH,BTC,LINK,FTM,PEPE

Makefile

@@ -3,10 +3,10 @@ pre-commit:
 
 pre-commit-verify:
 	@docker-compose -f local.yml run --rm tests bash -c \
-    "git config --global --add safe.directory /app && \
-     git init && \
-     git add . && \
-     pre-commit run --color=always --show-diff-on-failure --all-files"
+"git config --global --add safe.directory /app && \
+git init && \
+git add . && \
+pre-commit run --color=always --show-diff-on-failure --all-files"
 
 tests:
 	@docker-compose -f local.yml run --rm tests pytest
@@ -19,7 +19,7 @@ tests-shell:
 
 coverage:
 	@docker-compose -f local.yml run --rm tests bash -c \
-	"coverage run -m pytest -v --tb=short -p no:warnings && \
-	 coverage report && \
-	 coverage html"
+"coverage run -m pytest -v --tb=short -p no:warnings && \
+coverage report && \
+coverage html"
 	open htmlcov/index.html

README.md

@@ -1,10 +1,14 @@
 # exchange-radar
 Get real-time trades from various crypto exchanges.
 
-So far the following crypto exchanges are implemented:
+So far the following crypto exchanges are supported:
 - Binance
+- Coinbase
+- Kraken
 - Kucoin
 
+Currently, these are the top 4 exchanges by highest volume.
+
 ### Build & Run
 As simple as:
 

compose/producer/coinbase/Dockerfile

@@ -0,0 +1,47 @@
+ARG PYTHON_VERSION=3.11.3-slim-bullseye
+
+FROM python:${PYTHON_VERSION} as python
+
+ENV PYTHONUNBUFFERED=1 \
+    PYTHONDONTWRITEBYTECODE=1 \
+    PIP_NO_CACHE_DIR=off \
+    PIP_DISABLE_PIP_VERSION_CHECK=on \
+    PIP_DEFAULT_TIMEOUT=100 \
+    POETRY_HOME="/opt/poetry" \
+    POETRY_VIRTUALENVS_IN_PROJECT=true \
+    POETRY_NO_INTERACTION=1 \
+    PYSETUP_PATH="/app" \
+    VENV_PATH="/app/.venv"
+
+ENV PATH="$POETRY_HOME/bin:$VENV_PATH/bin:$PATH"
+
+
+FROM python as python-build-stage
+
+RUN apt-get update && apt-get install --no-install-recommends -y \
+    curl \
+    build-essential
+
+ENV POETRY_VERSION=1.5.1
+RUN curl -sSL https://install.python-poetry.org | python
+
+WORKDIR $PYSETUP_PATH
+
+COPY ./poetry.lock ./pyproject.toml ./
+RUN poetry install --only main --no-root
+
+
+FROM python as python-run-stage
+
+COPY --from=python-build-stage $POETRY_HOME $POETRY_HOME
+COPY --from=python-build-stage $PYSETUP_PATH $PYSETUP_PATH
+
+COPY ./compose/producer/coinbase/start /start
+RUN sed -i 's/\r$//g' /start
+RUN chmod +x /start
+
+WORKDIR $PYSETUP_PATH
+
+COPY . .
+
+RUN poetry install

compose/producer/coinbase/start

@@ -0,0 +1,7 @@
+#!/bin/bash
+
+set -o errexit
+set -o pipefail
+set -o nounset
+
+exec poetry run producer ${COINBASE_COINS} --exchange Coinbase

compose/producer/kraken/Dockerfile

@@ -0,0 +1,47 @@
+ARG PYTHON_VERSION=3.11.3-slim-bullseye
+
+FROM python:${PYTHON_VERSION} as python
+
+ENV PYTHONUNBUFFERED=1 \
+    PYTHONDONTWRITEBYTECODE=1 \
+    PIP_NO_CACHE_DIR=off \
+    PIP_DISABLE_PIP_VERSION_CHECK=on \
+    PIP_DEFAULT_TIMEOUT=100 \
+    POETRY_HOME="/opt/poetry" \
+    POETRY_VIRTUALENVS_IN_PROJECT=true \
+    POETRY_NO_INTERACTION=1 \
+    PYSETUP_PATH="/app" \
+    VENV_PATH="/app/.venv"
+
+ENV PATH="$POETRY_HOME/bin:$VENV_PATH/bin:$PATH"
+
+
+FROM python as python-build-stage
+
+RUN apt-get update && apt-get install --no-install-recommends -y \
+    curl \
+    build-essential
+
+ENV POETRY_VERSION=1.5.1
+RUN curl -sSL https://install.python-poetry.org | python
+
+WORKDIR $PYSETUP_PATH
+
+COPY ./poetry.lock ./pyproject.toml ./
+RUN poetry install --only main --no-root
+
+
+FROM python as python-run-stage
+
+COPY --from=python-build-stage $POETRY_HOME $POETRY_HOME
+COPY --from=python-build-stage $PYSETUP_PATH $PYSETUP_PATH
+
+COPY ./compose/producer/kraken/start /start
+RUN sed -i 's/\r$//g' /start
+RUN chmod +x /start
+
+WORKDIR $PYSETUP_PATH
+
+COPY . .
+
+RUN poetry install

compose/producer/kraken/start

@@ -0,0 +1,7 @@
+#!/bin/bash
+
+set -o errexit
+set -o pipefail
+set -o nounset
+
+exec poetry run producer ${KRAKEN_COINS} --exchange Kraken

exchange_radar/__init__.py

@@ -1,12 +1,14 @@
-__title__ = "exchange-radar"
-__description__ = "Get real-time trades from various crypto exchanges"
-__author__ = "Paulo Antunes"
-__copyright__ = "Copyright 2023, exchange-radar"
-__credits__ = [
-    "Paulo Antunes",
-]
-__license__ = "GPLv3"
-__maintainer__ = "Paulo Antunes"
-__email__ = "[email protected]"
-
-__version__ = "0.1.0"
+import tomllib
+
+with open("pyproject.toml", "rb") as f:
+    data = tomllib.load(f)
+
+toml = data["tool"]["poetry"]
+
+
+__title__ = toml["name"]
+__description__ = toml["description"]
+__authors__ = toml["authors"]
+__license__ = toml["license"]
+
+__version__ = toml["version"]

exchange_radar/producer/publisher.py

@@ -10,8 +10,7 @@
 )
 
 from exchange_radar.producer.enums import Ranking
-from exchange_radar.producer.schemas.binance import BinanceTradeSchema
-from exchange_radar.producer.schemas.kucoin import KucoinTradeSchema
+from exchange_radar.producer.schemas.base import CustomBaseModel
 from exchange_radar.producer.settings import base as settings
 from exchange_radar.producer.utils import get_ranking
 
@@ -73,8 +72,8 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 producer_connection = ProducerConnection()
 
 
-def publish(data: BinanceTradeSchema | KucoinTradeSchema) -> None:
-    logger.info(f"PUB: {data.trade_time} {data.symbol}")
+def publish(data: type[CustomBaseModel]) -> None:
+    logger.info(f"PUB: {data.trade_time} {data.symbol}")  # noqa
 
     try:
         with ProducerChannel(queue_name=settings.RABBITMQ_TRADES_QUEUE_NAME) as channel:
@@ -88,11 +87,11 @@ def publish(data: BinanceTradeSchema | KucoinTradeSchema) -> None:
             )
 
         queue_name = None
-        if get_ranking(data) == Ranking.WHALE:
+        if get_ranking(data) == Ranking.WHALE:  # noqa
             queue_name = settings.RABBITMQ_TRADES_WHALES_QUEUE_NAME
-        elif get_ranking(data) == Ranking.DOLPHIN:
+        elif get_ranking(data) == Ranking.DOLPHIN:  # noqa
             queue_name = settings.RABBITMQ_TRADES_DOLPHIN_QUEUE_NAME
-        elif get_ranking(data) == Ranking.OCTOPUS:
+        elif get_ranking(data) == Ranking.OCTOPUS:  # noqa
             queue_name = settings.RABBITMQ_TRADES_OCTOPUS_QUEUE_NAME
 
         if queue_name is not None:
@@ -116,4 +115,4 @@ def publish(data: BinanceTradeSchema | KucoinTradeSchema) -> None:
     except Exception as error:
         logger.error(f"ERROR: {error}")
     else:
-        logger.info(f"PUB OK: {data.trade_time} {data.symbol}")
+        logger.info(f"PUB OK: {data.trade_time} {data.symbol}")  # noqa

exchange_radar/producer/schemas/base.py

@@ -38,10 +38,10 @@ def trade_time_after(cls, v) -> datetime:
     def message(self) -> str:
         return (
             f"{self.trade_time} | "  # noqa
-            f"{self.exchange.ljust(8, ' ')} | "  # noqa
-            f"{'{:.8f} {}'.format(self.price, self.currency).rjust(14 + 5, ' ')} | "  # noqa
+            f"<span class='{self.exchange.lower()}'>{self.exchange.ljust(8, ' ')}</span> | "  # noqa
+            f"{'{:.8f} {}'.format(self.price, self.currency.rjust(4)).rjust(14 + 5, ' ')} | "  # noqa
             f"{'{:.8f} {}'.format(self.quantity, self.trade_symbol).rjust(21 + 5, ' ')} | "  # noqa
-            f"{'{:.8f} {}'.format(self.total, self.currency).rjust(17 + 5, ' ')}"  # noqa
+            f"{'{:.8f} {}'.format(self.total, self.currency.rjust(4)).rjust(17 + 5, ' ')}"  # noqa
         )
 
     @computed_field

exchange_radar/producer/schemas/coinbase.py

@@ -0,0 +1,31 @@
+from datetime import datetime
+from functools import cached_property
+
+from pydantic import Field, computed_field, condecimal, field_validator
+
+from exchange_radar.producer.schemas.base import CustomBaseModel
+
+
+class CoinbaseTradeSchema(CustomBaseModel):
+    symbol: str = Field(alias="product_id")
+    price: condecimal(ge=0, decimal_places=8)
+    quantity: condecimal(ge=0, decimal_places=8) = Field(alias="size")
+    trade_time: datetime = Field(alias="time")
+    side: str = Field(exclude=True)
+
+    @field_validator("symbol")
+    def symbol_normalization(cls, v) -> str:
+        return v.replace("-", "")
+
+    @field_validator("trade_time")
+    def trade_time_normalization(cls, v) -> str:
+        return v.replace(tzinfo=None)
+
+    @computed_field
+    @cached_property
+    def is_seller(self) -> bool:
+        return True if self.side == "sell" else False
+
+    @computed_field
+    def exchange(self) -> str:
+        return "Coinbase"

exchange_radar/producer/schemas/kraken.py

@@ -0,0 +1,35 @@
+from datetime import datetime
+from functools import cached_property
+
+from pydantic import Field, computed_field, condecimal, field_validator
+
+from exchange_radar.producer.schemas.base import CustomBaseModel
+
+
+class KrakenTradeSchema(CustomBaseModel):
+    symbol: str
+    price: condecimal(ge=0, decimal_places=12)
+    quantity: condecimal(ge=0, decimal_places=9)
+    trade_time: datetime
+    side: str = Field(exclude=True)
+
+    @field_validator("symbol")
+    def symbol_normalization(cls, v) -> str:
+        if v[:4] == "XBT/":
+            v = v.replace("XBT/", "BTC/")
+        elif v[-4:] == "/XBT":
+            v = v.replace("/XBT", "/BTC")
+        return v.replace("/", "")
+
+    @field_validator("trade_time", mode="before")
+    def trade_time_before(cls, v) -> int:
+        return int(v[:10])
+
+    @computed_field
+    @cached_property
+    def is_seller(self) -> bool:
+        return True if self.side == "s" else False
+
+    @computed_field
+    def exchange(self) -> str:
+        return "Kraken"

exchange_radar/producer/settings/base.py

@@ -19,4 +19,4 @@
 RABBITMQ_CONNECTION_HEARTBEAT = env.int("RABBITMQ_CONNECTION_HEARTBEAT")
 RABBITMQ_BLOCKED_CONNECTION_TIMEOUT = env.int("RABBITMQ_BLOCKED_CONNECTION_TIMEOUT")
 
-CURRENCIES = ["USDT", "BTC", "ETH", "BUSD"]
+CURRENCIES = ["USDT", "BTC", "ETH", "BUSD", "USD"]

exchange_radar/producer/settings/exchanges.py

@@ -1,7 +1,11 @@
 from exchange_radar.producer.tasks.binance import BinanceTradesTask
+from exchange_radar.producer.tasks.coinbase import CoinbaseTradesTask
+from exchange_radar.producer.tasks.kraken import KrakenTradesTask
 from exchange_radar.producer.tasks.kucoin import KucoinTradesTask
 
 EXCHANGES = {
     "Binance": BinanceTradesTask,
     "Kucoin": KucoinTradesTask,
+    "Coinbase": CoinbaseTradesTask,
+    "Kraken": KrakenTradesTask,
 }

exchange_radar/producer/task.py

@@ -12,7 +12,7 @@ def __init__(self):
     async def task(self, symbols: tuple[str]):
         await asyncio.gather(*[self.process(s) for s in symbols])
 
-    async def process(self, symbol_or_symbols: str):
+    async def process(self, symbol_or_symbols: str | tuple):
         raise NotImplementedError("Task to be Processed")
 
     def start(self, symbols: tuple[str, ...]):

exchange_radar/producer/tasks/binance.py

@@ -10,11 +10,11 @@
 logger = logging.getLogger(__name__)
 
 
-ITER_SLEEP = 10 * 60.0
+ITER_SLEEP = 10.0
 
 
 class BinanceTradesTask(Task):
-    async def process(self, symbol_or_symbols: str):
+    async def process(self, symbol_or_symbols: str | tuple):
         while True:
             try:
                 async_client = await AsyncClient.create()
@@ -31,6 +31,6 @@ async def process(self, symbol_or_symbols: str):
                 res = await ts.recv()
                 try:
                     data = BinanceTradeSchema(**res)
-                    publish(data)
+                    publish(data)  # noqa
                 except Exception as error:
                     logger.error(f"ERROR: {error}")