Merge pull request #12 from globophobe/feature/top-n

Release v0.1.4
globophobe · Dec 27, 2022 · 2210295 · 2210295
2 parents dab71b2 + 1591ca6
commit 2210295
Show file tree

Hide file tree

Showing 4 changed files with 678 additions and 313 deletions.
diff --git a/cryptofeed_werks/trades/candles.py b/cryptofeed_werks/trades/candles.py
@@ -3,14 +3,18 @@
 
 from cryptofeed.backends.aggregate import AggregateCallback
 
+from .constants import NOTIONAL
 from .window import WindowMixin
 
 
 class CandleCallback(WindowMixin, AggregateCallback):
-    def __init__(self, *args, window_seconds: int = 60, **kwargs) -> None:
+    def __init__(
+        self, *args, window_seconds: int = 60, top_n: Optional[int] = None, **kwargs
+    ) -> None:
         super().__init__(*args, **kwargs)
         self.window_seconds = window_seconds
         self.window = {}
+        self.top_n = top_n
         self.trades = {}
 
     async def __call__(self, trade: dict, timestamp: float) -> Tuple[dict, float]:
@@ -43,7 +47,7 @@ def aggregate(self, trades: List[dict], is_late: bool = False) -> Optional[dict]
         """Aggregate."""
         first_trade = trades[0]
         prices = self.get_prices(trades)
-        return {
+        candle = {
             "exchange": first_trade["exchange"],
             "symbol": first_trade["symbol"],
             "timestamp": self.get_start(first_trade["timestamp"]),
@@ -58,6 +62,9 @@ def aggregate(self, trades: List[dict], is_late: bool = False) -> Optional[dict]
             "totalBuyTicks": sum([t["totalBuyTicks"] for t in trades]),
             "totalTicks": sum([t["totalTicks"] for t in trades]),
         }
+        if self.top_n:
+            candle["topN"] = self.get_top_n(trades)
+        return candle
 
     def get_prices(self, trades: List[dict]) -> List[Decimal]:
         """Get prices."""
@@ -68,3 +75,22 @@ def get_prices(self, trades: List[dict]) -> List[Decimal]:
                 if value:
                     prices.append(value)
         return prices
+
+    def get_top_n(self, trades: List[dict]) -> List[dict]:
+        """Get top N."""
+        filtered = [t for t in trades if "uid" in t]
+        filtered.sort(key=lambda t: t[NOTIONAL], reverse=True)
+        top_n = filtered[: self.top_n]
+        for trade in top_n:
+            for key in list(trade):
+                if key not in (
+                    "timestamp",
+                    "price",
+                    "volume",
+                    "notional",
+                    "tickRule",
+                    "ticks",
+                ):
+                    del trade[key]
+        top_n.sort(key=lambda t: t["timestamp"])
+        return top_n
diff --git a/main.py b/main.py
@@ -39,12 +39,16 @@ async def candles(candle: dict, timestamp: float) -> None:
     print(candle)
 
 
-def get_callback(exchange, min_volume=1_000, window_seconds=60):
+def get_callback(
+    exchange: any, min_volume: int = 1_000, window_seconds: int = 60, top_n: int = 10
+):
     if exchange == Bitflyer:
         min_volume *= 100
     elif exchange == Upbit:
         min_volume *= 1000
-    candle_callback = CandleCallback(candles, window_seconds=window_seconds)
+    candle_callback = CandleCallback(
+        candles, window_seconds=window_seconds, top_n=top_n
+    )
     return SignificantTradeCallback(
         candle_callback,
         significant_trade_filter=min_volume,