Add labeler.py to root package (bfxapi). Remove List aliases in bfxapi/rest/typings.py. Update BfxRestInterface.py to use new standards.

2025-12-19 23:04:21 +01:00 · 2022-12-15 19:07:55 +01:00
parent 851184bf75
commit c9f86d6d03
4 changed files with 154 additions and 165 deletions
--- a/bfxapi/labeler.py
+++ b/bfxapi/labeler.py
@@ -0,0 +1,20 @@
+from typing import Generic, TypeVar, Iterable, Optional, List, Any
+
+T = TypeVar("T")
+
+class _Serializer(Generic[T]):
+    def __init__(self, name: str, labels: List[str], IGNORE: List[str] = [ "_PLACEHOLDER" ]):
+        self.name, self.__labels, self.__IGNORE = name, labels, IGNORE
+
+    def __serialize(self, *args: Any, skip: Optional[List[str]]) -> Iterable[T]:
+        labels = list(filter(lambda label: label not in (skip or list()), self.__labels))
+
+        if len(labels) > len(args):
+            raise Exception("<labels> and <*args> arguments should contain the same amount of elements.")
+
+        for index, label in enumerate(labels):
+            if label not in self.__IGNORE:
+                yield label, args[index]
+
+    def parse(self, *values: Any, skip: Optional[List[str]] = None) -> T:
+        return dict(self.__serialize(*values, skip=skip))
--- a/bfxapi/rest/BfxRestInterface.py
+++ b/bfxapi/rest/BfxRestInterface.py
@@ -104,7 +104,7 @@ class _RestPublicEndpoints(_Requests):
    def f_ticker(self, currency: str) -> FundingCurrencyTicker:
        return serializers.FundingCurrencyTicker.parse(*self._GET(f"ticker/f{currency}"), skip=["SYMBOL"])

-    def tickers_history(self, symbols: List[str], start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> TickersHistories:
+    def tickers_history(self, symbols: List[str], start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[TickersHistory]:
        params = {
            "symbols": ",".join(symbols),
            "start": start, "end": end,
@@ -115,51 +115,51 @@ class _RestPublicEndpoints(_Requests):
        
        return [ serializers.TickersHistory.parse(*subdata) for subdata in data ]

-    def t_trades(self, pair: str, limit: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, sort: Optional[Sort] = None) -> TradingPairTrades:
+    def t_trades(self, pair: str, limit: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, sort: Optional[Sort] = None) -> List[TradingPairTrade]:
        params = { "limit": limit, "start": start, "end": end, "sort": sort }
        data = self._GET(f"trades/{'t' + pair}/hist", params=params)
        return [ serializers.TradingPairTrade.parse(*subdata) for subdata in data ]

-    def f_trades(self, currency: str, limit: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, sort: Optional[Sort] = None) -> FundingCurrencyTrades:
+    def f_trades(self, currency: str, limit: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, sort: Optional[Sort] = None) -> List[FundingCurrencyTrade]:
        params = { "limit": limit, "start": start, "end": end, "sort": sort }
        data = self._GET(f"trades/{'f' + currency}/hist", params=params)
        return [ serializers.FundingCurrencyTrade.parse(*subdata) for subdata in data ]

-    def t_book(self, pair: str, precision: Literal["P0", "P1", "P2", "P3", "P4"], len: Optional[Literal[1, 25, 100]] = None) -> TradingPairBooks:
+    def t_book(self, pair: str, precision: Literal["P0", "P1", "P2", "P3", "P4"], len: Optional[Literal[1, 25, 100]] = None) -> List[TradingPairBook]:
        return [ serializers.TradingPairBook.parse(*subdata) for subdata in self._GET(f"book/{'t' + pair}/{precision}", params={ "len": len }) ]

-    def f_book(self, currency: str, precision: Literal["P0", "P1", "P2", "P3", "P4"], len: Optional[Literal[1, 25, 100]] = None) -> FundingCurrencyBooks:
+    def f_book(self, currency: str, precision: Literal["P0", "P1", "P2", "P3", "P4"], len: Optional[Literal[1, 25, 100]] = None) -> List[FundingCurrencyBook]:
        return [ serializers.FundingCurrencyBook.parse(*subdata) for subdata in self._GET(f"book/{'f' + currency}/{precision}", params={ "len": len }) ]

-    def t_raw_book(self, pair: str, len: Optional[Literal[1, 25, 100]] = None) -> TradingPairRawBooks:
+    def t_raw_book(self, pair: str, len: Optional[Literal[1, 25, 100]] = None) -> List[TradingPairRawBook]:
        return [ serializers.TradingPairRawBook.parse(*subdata) for subdata in self._GET(f"book/{'t' + pair}/R0", params={ "len": len }) ]

-    def f_raw_book(self, currency: str, len: Optional[Literal[1, 25, 100]] = None) -> FundingCurrencyRawBooks:
+    def f_raw_book(self, currency: str, len: Optional[Literal[1, 25, 100]] = None) -> List[FundingCurrencyRawBook]:
        return [ serializers.FundingCurrencyRawBook.parse(*subdata) for subdata in self._GET(f"book/{'f' + currency}/R0", params={ "len": len }) ]

    def stats_hist(
        self, 
        resource: str,
        sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None
-    ) -> Stats:
+    ) -> List[Statistic]:
        params = { "sort": sort, "start": start, "end": end, "limit": limit }
        data = self._GET(f"stats1/{resource}/hist", params=params)
-        return [ serializers.Stat.parse(*subdata) for subdata in data ]
+        return [ serializers.Statistic.parse(*subdata) for subdata in data ]

    def stats_last(
        self, 
        resource: str,
        sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None
-    ) -> Stat:
+    ) -> Statistic:
        params = { "sort": sort, "start": start, "end": end, "limit": limit }
        data = self._GET(f"stats1/{resource}/last", params=params)
-        return serializers.Stat.parse(*data)
+        return serializers.Statistic.parse(*data)

    def candles_hist(
        self,
        resource: str,
        sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None
-    ) -> Candles:
+    ) -> List[Candle]:
        params = { "sort": sort, "start": start, "end": end, "limit": limit }
        data = self._GET(f"candles/{resource}/hist", params=params)
        return [ serializers.Candle.parse(*subdata) for subdata in data ]
@@ -173,7 +173,7 @@ class _RestPublicEndpoints(_Requests):
        data = self._GET(f"candles/{resource}/last", params=params)
        return serializers.Candle.parse(*data)

-    def derivatives_status(self, type: str, keys: List[str]) -> DerivativeStatuses:
+    def derivatives_status(self, type: str, keys: List[str]) -> List[DerivativesStatus]:
        params = { "keys": ",".join(keys) }

        data = self._GET(f"status/{type}", params=params)
@@ -184,14 +184,14 @@ class _RestPublicEndpoints(_Requests):
        self, 
        type: str, symbol: str,
        sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None
-    ) -> DerivativeStatuses: 
+    ) -> List[DerivativesStatus]: 
        params = { "sort": sort, "start": start, "end": end, "limit": limit }

        data = self._GET(f"status/{type}/{symbol}/hist", params=params)

        return [ serializers.DerivativesStatus.parse(*subdata, skip=[ "KEY" ]) for subdata in data ]

-    def liquidations(self, sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> Liquidations:
+    def liquidations(self, sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[Liquidation]:
        params = { "sort": sort, "start": start, "end": end, "limit": limit }

        data = self._GET("liquidations/hist", params=params)
@@ -202,7 +202,7 @@ class _RestPublicEndpoints(_Requests):
        self,
        resource: str,
        sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None
-    ) -> Leaderboards:
+    ) -> List[Leaderboard]:
        params = { "sort": sort, "start": start, "end": end, "limit": limit }
        data = self._GET(f"rankings/{resource}/hist", params=params)
        return [ serializers.Leaderboard.parse(*subdata) for subdata in data ]
@@ -216,12 +216,12 @@ class _RestPublicEndpoints(_Requests):
        data = self._GET(f"rankings/{resource}/last", params=params)
        return serializers.Leaderboard.parse(*data)

-    def funding_stats(self, symbol: str, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> FundingStats:
+    def funding_stats(self, symbol: str, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[FundingStatistic]:
        params = { "start": start, "end": end, "limit": limit }

        data = self._GET(f"funding/stats/{symbol}/hist", params=params)

-        return [ serializers.FundingStat.parse(*subdata) for subdata in data ]
+        return [ serializers.FundingStatistic.parse(*subdata) for subdata in data ]

    def conf(self, config: Config) -> Any:
        return self._GET(f"conf/{config}")[0]
--- a/bfxapi/rest/serializers.py
+++ b/bfxapi/rest/serializers.py
@@ -1,27 +1,6 @@
-from typing import Generic, TypeVar, Iterable, Optional, List, Any
-
 from . import typings

-from .exceptions import BfxRestException
-
-T = TypeVar("T")
-
-class _Serializer(Generic[T]):
-    def __init__(self, name: str, labels: List[str], IGNORE: List[str] = [ "_PLACEHOLDER" ]):
-        self.name, self.__labels, self.__IGNORE = name, labels, IGNORE
-
-    def __serialize(self, *args: Any, skip: Optional[List[str]]) -> Iterable[T]:
-        labels = list(filter(lambda label: label not in (skip or list()), self.__labels))
-
-        if len(labels) > len(args):
-            raise BfxRestException("<labels> and <*args> arguments should contain the same amount of elements.")
-
-        for index, label in enumerate(labels):
-            if label not in self.__IGNORE:
-                yield label, args[index]
-
-    def parse(self, *values: Any, skip: Optional[List[str]] = None) -> T:
-        return dict(self.__serialize(*values, skip=skip))
+from .. labeler import _Serializer

 #region Serializers definition for Rest Public Endpoints

@@ -120,7 +99,7 @@ FundingCurrencyRawBook = _Serializer[typings.FundingCurrencyRawBook]("FundingCur
    "AMOUNT"
 ])

-Stat = _Serializer[typings.Stat]("Stat", labels=[
+Statistic = _Serializer[typings.Statistic]("Statistic", labels=[
    "MTS",
    "VALUE"
 ])
@@ -189,7 +168,7 @@ Leaderboard = _Serializer[typings.Leaderboard]("Leaderboard", labels=[
    "TWITTER_HANDLE"
 ])

-FundingStat = _Serializer[typings.FundingStat]("FundingStat", labels=[
+FundingStatistic = _Serializer[typings.FundingStatistic]("FundingStatistic", labels=[
    "TIMESTAMP",
    "_PLACEHOLDER",
    "_PLACEHOLDER",
--- a/bfxapi/rest/typings.py
+++ b/bfxapi/rest/typings.py
@@ -2,140 +2,130 @@ from typing import Type, Tuple, List, Dict, TypedDict, Union, Optional, Any

 #region Type hinting for Rest Public Endpoints

-PlatformStatus = TypedDict("PlatformStatus", {
-    "OPERATIVE": int
-})
+class PlatformStatus(TypedDict):
+    OPERATIVE: int

-TradingPairTicker = TypedDict("TradingPairTicker", {
-    "SYMBOL": Optional[str],
-    "BID": float,
-    "BID_SIZE": float,
-    "ASK": float,
-    "ASK_SIZE": float,
-    "DAILY_CHANGE": float,
-    "DAILY_CHANGE_RELATIVE": float,
-    "LAST_PRICE": float,
-    "VOLUME": float,
-    "HIGH": float,
-    "LOW": float
-})
+class TradingPairTicker(TypedDict):
+    SYMBOL: Optional[str]
+    BID: float
+    BID_SIZE: float
+    ASK: float
+    ASK_SIZE: float
+    DAILY_CHANGE: float
+    DAILY_CHANGE_RELATIVE: float
+    LAST_PRICE: float
+    VOLUME: float
+    HIGH: float
+    LOW: float

-FundingCurrencyTicker = TypedDict("FundingCurrencyTicker", {
-    "SYMBOL": Optional[str],
-    "FRR": float,
-    "BID": float,
-    "BID_PERIOD": int,
-    "BID_SIZE": float,
-    "ASK": float,
-    "ASK_PERIOD": int,
-    "ASK_SIZE": float,
-    "DAILY_CHANGE": float,
-    "DAILY_CHANGE_RELATIVE": float,
-    "LAST_PRICE": float,
-    "VOLUME": float,
-    "HIGH": float,
-    "LOW": float,
-    "FRR_AMOUNT_AVAILABLE": float
-})
+class FundingCurrencyTicker(TypedDict):
+    SYMBOL: Optional[str]
+    FRR: float
+    BID: float
+    BID_PERIOD: int
+    BID_SIZE: float
+    ASK: float
+    ASK_PERIOD: int
+    ASK_SIZE: float
+    DAILY_CHANGE: float
+    DAILY_CHANGE_RELATIVE: float
+    LAST_PRICE: float
+    VOLUME: float
+    HIGH: float
+    LOW: float
+    FRR_AMOUNT_AVAILABLE: float

-TickersHistory = TypedDict("TickersHistory", {
-    "SYMBOL": str,
-    "BID": float,
-    "ASK": float,
-    "MTS": int
-})
+class TickersHistory(TypedDict):
+    SYMBOL: str
+    BID: float
+    ASK: float
+    MTS: int

-TickersHistories = List[TickersHistory]
+class TradingPairTrade(TypedDict):
+    ID: int 
+    MTS: int 
+    AMOUNT: float 
+    PRICE: float

-(TradingPairTrade, FundingCurrencyTrade) = (
-    TypedDict("TradingPairTrade", { "ID": int, "MTS": int, "AMOUNT": float, "PRICE": float }),
-    TypedDict("FundingCurrencyTrade", { "ID": int, "MTS": int, "AMOUNT": float, "RATE": float, "PERIOD": int })
-)
+class FundingCurrencyTrade(TypedDict):
+    ID: int 
+    MTS: int 
+    AMOUNT: float 
+    RATE: float 
+    PERIOD: int

-(TradingPairTrades, FundingCurrencyTrades) = (List[TradingPairTrade], List[FundingCurrencyTrade])
+class TradingPairBook(TypedDict):
+    PRICE: float 
+    COUNT: int 
+    AMOUNT: float
    
-(TradingPairBook, FundingCurrencyBook) = (
-    TypedDict("TradingPairBook", { "PRICE": float, "COUNT": int, "AMOUNT": float }),
-    TypedDict("FundingCurrencyBook", { "RATE": float, "PERIOD": int, "COUNT": int, "AMOUNT": float })
-)
+class FundingCurrencyBook(TypedDict):
+    RATE: float 
+    PERIOD: int 
+    COUNT: int 
+    AMOUNT: float
        
-(TradingPairBooks, FundingCurrencyBooks) = (List[TradingPairBook], List[FundingCurrencyBook])
+class TradingPairRawBook(TypedDict):
+    ORDER_ID: int
+    PRICE: float 
+    AMOUNT: float
            
-(TradingPairRawBook, FundingCurrencyRawBook) = (
-    TypedDict("TradingPairRawBook", { "ORDER_ID": int, "PRICE": float, "AMOUNT": float }),
-    TypedDict("FundingCurrencyRawBook", { "OFFER_ID": int, "PERIOD": int, "RATE": float, "AMOUNT": float }),
-)
+class FundingCurrencyRawBook(TypedDict):
+    OFFER_ID: int 
+    PERIOD: int 
+    RATE: float 
+    AMOUNT: float

-(TradingPairRawBooks, FundingCurrencyRawBooks) = (List[TradingPairRawBook], List[FundingCurrencyRawBook])
+class Statistic(TypedDict):
+    MTS: int
+    VALUE: float

-Stat = TypedDict("Stat", {
-    "MTS": int,
-    "VALUE": float
-})
+class Candle(TypedDict):
+    MTS: int
+    OPEN: float
+    CLOSE: float
+    HIGH: float
+    LOW: float
+    VOLUME: float

-Stats = List[Stat]
+class DerivativesStatus(TypedDict):
+    KEY: Optional[str]
+    MTS: int
+    DERIV_PRICE: float
+    SPOT_PRICE: float
+    INSURANCE_FUND_BALANCE: float
+    NEXT_FUNDING_EVT_TIMESTAMP_MS: int
+    NEXT_FUNDING_ACCRUED: float
+    NEXT_FUNDING_STEP: int
+    CURRENT_FUNDING: float
+    MARK_PRICE: float
+    OPEN_INTEREST: float
+    CLAMP_MIN: float
+    CLAMP_MAX: float

-Candle = TypedDict("Candle", {
-    "MTS": int,
-    "OPEN": float,
-    "CLOSE": float,
-    "HIGH": float,
-    "LOW": float,
-    "VOLUME": float
-})
+class Liquidation(TypedDict):
+    POS_ID: int
+    MTS: int
+    SYMBOL: str
+    AMOUNT: float
+    BASE_PRICE: float
+    IS_MATCH: int
+    IS_MARKET_SOLD: int
+    PRICE_ACQUIRED: float

-Candles = List[Candle]
+class Leaderboard(TypedDict):
+    MTS: int
+    USERNAME: str
+    RANKING: int
+    VALUE: float
+    TWITTER_HANDLE: Optional[str]

-DerivativesStatus = TypedDict("DerivativesStatus", {
-    "KEY": Optional[str],
-    "MTS": int,
-    "DERIV_PRICE": float,
-    "SPOT_PRICE": float,
-    "INSURANCE_FUND_BALANCE": float,
-    "NEXT_FUNDING_EVT_TIMESTAMP_MS": int,
-    "NEXT_FUNDING_ACCRUED": float,
-    "NEXT_FUNDING_STEP": int,
-    "CURRENT_FUNDING": float,
-    "MARK_PRICE": float,
-    "OPEN_INTEREST": float,
-    "CLAMP_MIN": float,
-    "CLAMP_MAX": float
-})
-
-DerivativeStatuses = List[DerivativesStatus]
-
-Liquidation = TypedDict("Liquidation", {
-    "POS_ID": int,
-    "MTS": int,
-    "SYMBOL": str,
-    "AMOUNT": float,
-    "BASE_PRICE": float,
-    "IS_MATCH": int,
-    "IS_MARKET_SOLD": int,
-    "PRICE_ACQUIRED": float
-})
-
-Liquidations = List[Liquidation]
-
-Leaderboard = TypedDict("Leaderboard", {
-    "MTS": int,
-    "USERNAME": str,
-    "RANKING": int,
-    "VALUE": float,
-    "TWITTER_HANDLE": Optional[str]
-})
-
-Leaderboards = List[Leaderboard]
-
-FundingStat = TypedDict("FundingStat", {
-    "TIMESTAMP": int,
-    "FRR": float,
-    "AVG_PERIOD": float,
-    "FUNDING_AMOUNT": float,
-    "FUNDING_AMOUNT_USED": float,
-    "FUNDING_BELOW_THRESHOLD": float
-})
-
-FundingStats = List[FundingStat]
+class FundingStatistic(TypedDict): 
+    TIMESTAMP: int
+    FRR: float
+    AVG_PERIOD: float
+    FUNDING_AMOUNT: float
+    FUNDING_AMOUNT_USED: float
+    FUNDING_BELOW_THRESHOLD: float

 #endregion