chore(exchanges): working with okx data

exchanges/fetchers/future_fetcher.py

@@ -9,9 +9,7 @@ class FutureFetcher:
     def __init__(self, exchange):
         self.exchange = exchange
 
-    def fetch_market_data_okx(
-        self, market_symbols: list[str]
-    ) -> pd.DataFrame:
+    def fetch_market_data_okx(self, market_symbols: list[str]) -> pd.DataFrame:
         print("Fetching data from OKX for futures")
         data_list = []
         try:
@@ -23,9 +21,7 @@ def fetch_market_data_okx(
 
         return pd.DataFrame(data_list)
 
-    def fetch_market_data_binance(
-        self, market_symbols: list[str]
-    ) -> pd.DataFrame:
+    def fetch_market_data_binance(self, market_symbols: list[str]) -> pd.DataFrame:
         data_list = []
         print(f"Market symbols: {market_symbols}")
         try:
@@ -38,9 +34,7 @@ def fetch_market_data_binance(
 
         return pd.DataFrame(data_list)
 
-    def fetch_market_data_deribit(
-        self, market_symbols: list[str]
-    ) -> pd.DataFrame:
+    def fetch_market_data_deribit(self, market_symbols: list[str]) -> pd.DataFrame:
         data_list = []
         try:
             all_tickers = self.exchange.fetch_tickers(market_symbols)

exchanges/fetchers/option_fetcher.py

@@ -9,168 +9,108 @@ class OptionFetcher:
     def __init__(self, exchange):
         self.exchange = exchange
 
-    def fetch_market_data_deribit(self, market_symbols: list[str]) -> pd.DataFrame:
+    def fetch_market_data(
+        self, market_symbols: list[str], exchange_name: str
+    ) -> pd.DataFrame:
         """
-        Fetches market data for a given list of market symbols from deribit exchange.
+        Fetches market data for a given list of market symbols from a specified exchange and processes it using pandas.
         Args:
-            market_symbols: A list of symbols in the format recognized by OKX (e.g., "BTC-USD-240628-42000-C")
+            market_symbols: A list of symbols in the format recognized by the exchange.
+            exchange_name: String representing the exchange name ('deribit', 'okx', 'binance').
         Returns:
-            pd.DataFrame: DataFrame with market data for each option contract.
+            pd.DataFrame: DataFrame with processed market data for each option contract.
         """
-        data_list = []  # Initialize an empty list to store data dictionaries
-
-        try:
-            all_tickers = self.exchange.fetch_tickers(market_symbols)
-            for symbol, ticker in all_tickers.items():
-                info = ticker.get("info", {})
-                bid_raw = ticker.get("bid", 0) if ticker.get("bid") is not None else 0
-                ask_raw = ticker.get("ask", 0) if ticker.get("ask") is not None else 0
-                underlying_price = float(info.get("underlying_price", 0))
-                bid = float(bid_raw) * underlying_price
-                ask = float(ask_raw) * underlying_price
-                mark_price_raw = float(info.get("mark_price", 0))
-                mark_price = mark_price_raw * underlying_price
-                timestamp = ticker.get("timestamp", 0)
-                datetime = pd.to_datetime(timestamp, unit="ms")
-                expiry = self.date_parser(symbol)
-                ytm = (expiry - datetime) / np.timedelta64(1, "Y")
-                estimated_delivery_price = float(
-                    info.get("estimated_delivery_price", 0)
-                )
-
-                data_dict = {
-                    "symbol": symbol,
-                    "bid_btc": bid_raw,
-                    "ask_btc": ask_raw,
-                    "underlying_price": underlying_price,
-                    "bid": bid,
-                    "ask": ask,
-                    "mark_price_btc": mark_price_raw,
-                    "mark_price": mark_price,
-                    "timestamp": timestamp,
-                    "datetime": datetime,
-                    "expiry": expiry,
-                    "YTM": ytm,
-                    "forward_price": estimated_delivery_price,
-                }
-                data_list.append(data_dict)
-
-        except Exception as e:
-            logging.error(f"Error fetching tickers: {e}")
-
-        return pd.DataFrame(data_list)
-
-    def fetch_market_data_okx(self, market_symbols: list[str]) -> pd.DataFrame:
-        """
-        Fetches market data for a given list of market symbols from OKX exchange.
-        Args:
-            market_symbols: A list of symbols in the format recognized by OKX (e.g., "BTC-USD-240628-42000-C")
-        Returns:
-            pd.DataFrame: DataFrame with market data for each option contract.
-        """
-        data_list = []  # Initialize an empty list to store data dictionaries
-
         try:
             all_tickers = self.exchange.fetch_tickers(market_symbols)
-            for symbol, ticker in all_tickers.items():
-                bid = (
-                    float(ticker.get("bid", 0)) if ticker.get("bid") is not None else 0
-                )
-                ask = (
-                    float(ticker.get("ask", 0)) if ticker.get("ask") is not None else 0
-                )
-                timestamp = ticker.get("timestamp", 0)
-                datetime = pd.to_datetime(timestamp, unit="ms")
-                expiry = self.date_parser(symbol)
-                ytm = (expiry - datetime) / np.timedelta64(1, "Y")
-
-                # Construct a dictionary for each symbol with the required data
-                data_dict = {
-                    "symbol": symbol,
-                    "bid": bid,
-                    "ask": ask,
-                    "timestamp": timestamp,
-                    "datetime": datetime,
-                    "expiry": expiry,
-                    "YTM": ytm,
-                }
-                data_list.append(data_dict)
-
-            with open("okx_data.json", "w") as f:
-                json.dump(data_list, f, indent=4)
-
-            with open("okx_raw_data.json", "w") as f:
-                json.dump(all_tickers, f, indent=4)
-
+            tickers_df = pd.DataFrame(all_tickers).transpose()
+            if exchange_name == "Deribit":
+                return self.process_deribit_data(tickers_df)
+            elif exchange_name == "OKX":
+                return self.process_okx_data(tickers_df)
+            elif exchange_name == "Binance":
+                return self.process_binance_data(tickers_df)
+            else:
+                logging.error(f"Unsupported exchange: {exchange_name}")
+                return pd.DataFrame()
         except Exception as e:
-            logging.error(f"Error fetching tickers from OKX: {e}")
-
-        return pd.DataFrame(data_list)
+            logging.error(f"Error fetching tickers from {exchange_name}: {e}")
+            return pd.DataFrame()
+
+    def process_deribit_data(self, df: pd.DataFrame) -> pd.DataFrame:
+        # Convert 'info' column to separate columns for easier manipulation
+        info_df = pd.json_normalize(df["info"])
+        df = df.reset_index(drop=True)
+
+        # Replace 'None' with 0.0 and convert to float for 'bid' and 'ask'
+        df["bid"] = pd.to_numeric(df["bid"], errors="coerce").fillna(0.0)
+        df["ask"] = pd.to_numeric(df["ask"], errors="coerce").fillna(0.0)
+
+        # Convert 'mark_price' from info_df to numeric and update in df
+        df["mark_price"] = pd.to_numeric(info_df["mark_price"], errors="coerce").fillna(
+            0.0
+        )
+
+        # Assuming info_df and df are aligned by index after pd.json_normalize
+        underlying_prices = pd.to_numeric(
+            info_df["underlying_price"], errors="coerce"
+        ).fillna(0.0)
+
+        # Adjust 'bid' and 'ask' based on 'underlying_prices'
+        df["bid"] *= underlying_prices
+        df["ask"] *= underlying_prices
+        df["mark_price"] *= underlying_prices
+
+        df["underlying_price"] = underlying_prices
+
+        # Convert timestamp to datetime
+        df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms")
+
+        # Efficient expiry date parsing (assuming this can be vectorized or is already efficient)
+        df["expiry"] = df["symbol"].apply(self.date_parser)
+
+        # Calculate YTM
+        df["YTM"] = (df["expiry"] - df["datetime"]) / np.timedelta64(1, "Y")
+
+        # Select and reorder the required columns
+        return df[
+            [
+                "symbol",
+                "bid",
+                "ask",
+                "mark_price",
+                "datetime",
+                "expiry",
+                "YTM",
+                "underlying_price",
+            ]
+        ]
 
-    def fetch_market_data_binance(self, market_symbols: list[str]) -> pd.DataFrame:
-        """
-        Fetches market data for a given list of market symbols from Binance exchange.
-        Args:
-            market_symbols: A list of symbols in the format recognized by Binance (e.g., "BTC-240628-29000-P")
-        Returns:
-            pd.DataFrame: DataFrame with market data for each option contract.
-        """
-        data_list = []  # Initialize an empty list to store data dictionaries
+    def process_okx_data(self, df: pd.DataFrame) -> pd.DataFrame:
+        df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms")
+        df["expiry"] = df["symbol"].apply(self.date_parser)
+        df["YTM"] = (df["expiry"] - df["datetime"]) / np.timedelta64(1, "Y")
+        return df[["symbol", "bid", "ask", "datetime", "expiry", "YTM"]]
 
-        try:
-            all_tickers = self.exchange.fetch_tickers(market_symbols)
-            for symbol, ticker in all_tickers.items():
-                info = ticker.get("info", {})
-                bid = float(info.get("bidPrice", 0))
-                ask = float(info.get("askPrice", 0))
-                exercise_price = float(info.get("exercisePrice", 0))
-                timestamp = ticker.get("timestamp", 0)  # Timestamp
-
-                # Construct a dictionary for each symbol with the required data
-                data_dict = {
-                    "symbol": symbol,
-                    "bid": bid,
-                    "ask": ask,
-                    "timestamp": timestamp,
-                    "underlying-asset": exercise_price,
-                }
-                data_list.append(data_dict)
-            with open("binance_data.json", "w") as f:
-                json.dump(data_list, f, indent=4)
-
-            with open("binance_raw_data.json", "w") as f:
-                json.dump(all_tickers, f, indent=4)
+    def process_binance_data(self, df: pd.DataFrame) -> pd.DataFrame:
+        # Assuming 'info' contains the required details
+        df["bid"] = df["info"].apply(lambda x: float(x.get("bidPrice", 0)))
+        df["ask"] = df["info"].apply(lambda x: float(x.get("askPrice", 0)))
+        df["timestamp"] = pd.to_datetime(df["timestamp"], unit="ms")
+        df["expiry"] = df["symbol"].apply(self.date_parser)
+        df["YTM"] = (df["expiry"] - df["timestamp"]) / np.timedelta64(1, "Y")
 
-        except Exception as e:
-            print(f"Error fetching tickers from Binance: {e}")
-
-        return pd.DataFrame(data_list)
+        return df[["symbol", "bid", "ask", "timestamp", "expiry", "YTM"]]
 
     @staticmethod
-    def date_parser(symbol):
-        # Define a list of possible date formats to try
-        date_formats = [
-            "%y%m%d",  # YYMMDD
-            "%d%b%y",  # DDMMMYY
-            "%Y%m%d",  # YYYYMMDD
-            "%m%d%Y",  # MMDDYYYY
-            "%d%m%Y",  # DDMMYYYY
-        ]
-
-        # Split the symbol based on common separators
+    def date_parser(symbol: str) -> pd.Timestamp:
+        date_formats = ["%y%m%d", "%d%b%y", "%Y%m%d", "%m%d%Y", "%d%m%Y"]
         parts = symbol.replace(":", "-").replace("/", "-").replace(".", "-").split("-")
-
-        # Loop through the parts to identify potential date segment
         for part in parts:
-            # Skip segments that are purely alphabetical, as they likely don't contain date info
             if part.isalpha():
                 continue
-            # Try to parse each segment with each date format until successful
             for date_format in date_formats:
                 try:
-                    date = pd.to_datetime(part, format=date_format)
-                    return date  # Return the parsed date as soon as a successful parse occurs
+                    return pd.to_datetime(part, format=date_format)
                 except ValueError:
-                    continue  # If parsing fails, try the next format
-
-        return pd.NaT  # Return Not-A-Time if no date could be parsed
+                    continue
+        return pd.NaT

exchanges/handlers/future.py

@@ -16,17 +16,11 @@ def __init__(self, exchange: str, market_types: List[str]):
 
     def handle(self, market_symbols: List[str]) -> pd.DataFrame:
         if str(self.exchange) == "Binance":
-            market_data = self.data_fetcher.fetch_market_data_binance(
-                market_symbols
-            )
+            market_data = self.data_fetcher.fetch_market_data_binance(market_symbols)
         elif str(self.exchange) == "OKX":
-            market_data = self.data_fetcher.fetch_market_data_okx(
-                market_symbols
-            )
+            market_data = self.data_fetcher.fetch_market_data_okx(market_symbols)
         elif str(self.exchange) == "Deribit":
-            market_data = self.data_fetcher.fetch_market_data_deribit(
-                market_symbols
-            )
+            market_data = self.data_fetcher.fetch_market_data_deribit(market_symbols)
         else:
             logger.error(f"Exchange not supported: {self.exchange}")
             return pd.DataFrame()

exchanges/handlers/merge.py

@@ -5,6 +5,7 @@
 
 from exchanges.handlers.future import FutureMarketHandler
 from exchanges.handlers.option import OptionMarketHandler
+from exchanges.processing import Processing
 
 
 class MergeMarketHandler:
@@ -13,16 +14,37 @@ def __init__(self, exchange, market_types):
         self.market_types = market_types
         self.option_market_handler = OptionMarketHandler(exchange, market_types)
         self.future_market_handler = FutureMarketHandler(exchange, market_types)
+        self.processing = Processing()
 
     def handle(
         self, options_market: List[str], future_market: List[str] | None
     ) -> pd.DataFrame:
         options_data = self.option_market_handler.handle(options_market)
+        options_data.to_json(
+            f"{self.exchange}_options_data.json", orient="records", indent=4
+        )
         if future_market:
             futures_data = self.future_market_handler.handle(future_market)
             merged_data = pd.concat([options_data, futures_data], ignore_index=True)
         else:
             merged_data = options_data
 
+        valid_quotes = self.processing.eliminate_invalid_quotes(merged_data)
+        implied_interest_rates = self.processing.calculate_implied_interest_rates(
+            valid_quotes
+        )
+
+        spreads = self.processing.calculate_spreads(valid_quotes)
+        remove_large_spreads = self.processing.remove_large_spreads(spreads)
+        spreads.to_json(
+            f"{self.exchange}_spreads.json", orient="records", indent=4
+        )
+        remove_large_spreads.to_json(
+            f"{self.exchange}_remove_large_spreads.json", orient="records", indent=4
+        )
+
+        implied_interest_rates.to_json(
+            f"{self.exchange}_implied_interest_rates.json", orient="records", indent=4
+        )
 
         return merged_data

exchanges/handlers/option.py

@@ -20,14 +20,8 @@ def __init__(self, exchange, market_types):
         self.processing = Processing()
 
     def handle(self, market_symbols: List[str]) -> pd.DataFrame:
-        if str(self.exchange) == "Deribit":
-            market_data = self.data_fetcher.fetch_market_data_deribit(market_symbols)
-        elif str(self.exchange) == "OKX":
-            market_data = self.data_fetcher.fetch_market_data_okx(market_symbols)
-        elif str(self.exchange) == "Binance":
-            market_data = self.data_fetcher.fetch_market_data_binance(market_symbols)
-        else:
-            logger.error(f"Exchange not supported: {self.exchange}")
-            return pd.DataFrame()
+        market_data = self.data_fetcher.fetch_market_data(
+            market_symbols, str(self.exchange)
+        )
 
         return market_data

exchanges/main.py

@@ -22,7 +22,7 @@ def main():
         )
         results = pd.DataFrame()
 
-        for manager in [okx]:
+        for manager in [deribit]:
             results = pd.concat(
                 [results, manager.load_specific_pairs()], ignore_index=True
             )