feat(exchanges): consolidate data

exchanges/exchange_manager.py

@@ -1,9 +1,11 @@
 import json
+from typing import Tuple
 
 import ccxt
 import logging
 
 import pandas as pd
+from pandas import DataFrame
 
 from exchanges.fetchers.binance_fetcher import BinanceFetcher
 from exchanges.handlers.merge import MergeMarketHandler
@@ -27,7 +29,7 @@ def fetch_binance_symbols(self):
         binance_option_symbols = self.binance_fetcher.fetch_options_symbols()
         return binance_option_symbols
 
-    def load_specific_pairs(self) -> pd.DataFrame:
+    def load_specific_pairs(self) -> tuple[DataFrame, DataFrame] | None | DataFrame:
         try:
             if self.exchange_id == "binance":
                 binance_option_symbols = self.fetch_binance_symbols()
@@ -63,7 +65,7 @@ def filter_markets(self, markets_df: pd.DataFrame) -> dict:
                 filtered_markets[pair][market_type] = symbols
         return filtered_markets
 
-    def handle_market_type(self, loaded_markets: dict) -> pd.DataFrame:
+    def handle_market_type(self, loaded_markets: dict) -> tuple[DataFrame, DataFrame] | None:
         dataframe = None
         for pair in self.pairs_to_load:
             future_symbols = loaded_markets.get(pair, {}).get("future", [])

exchanges/fetchers/binance_fetcher.py

@@ -98,6 +98,18 @@ def fetch_mark_price_options():
 
         return mark_prices_options_df
 
+    @staticmethod
+    def fetch_mark_price_options():
+        mark_prices_options = BinanceFetcher.get_response(
+            BINANCE_API_OPTIONS_URL + "/eapi/v1/mark"
+        )
+        mark_prices_options_df = pd.DataFrame(mark_prices_options)
+        mark_prices_options_df = mark_prices_options_df.loc[
+            mark_prices_options_df["symbol"].str.contains("BTC-")
+        ]
+
+        return mark_prices_options_df
+
     @staticmethod
     def fetch_spot_price(symbol: str = "BTCUSDT"):
         spot_price = BinanceFetcher.get_response(

exchanges/fetchers/future_fetcher.py

@@ -3,6 +3,7 @@
 import numpy as np
 import pandas as pd
 
+
 class FutureFetcher:
     def __init__(self, exchange):
         self.exchange = exchange
@@ -11,16 +12,20 @@ def fetch_future_market_symbols(self, symbol: str) -> list[str]:
         load_markets = self.exchange.load_markets()
         load_markets_df = pd.DataFrame(load_markets).transpose()
         future_symbols = load_markets_df[
-            (load_markets_df["future"] == True) &
-            (load_markets_df["symbol"].str.contains(f"{symbol}/USD")) &
-            (load_markets_df["symbol"].str.contains(f":{symbol}"))
+            (load_markets_df["future"] == True)
+            & (load_markets_df["symbol"].str.contains(f"{symbol}/USD"))
+            & (load_markets_df["symbol"].str.contains(f":{symbol}"))
         ].index.to_list()
         return future_symbols
 
     def fetch_future_orderbook(self, symbol: str) -> dict:
         order_book = self.exchange.fetch_order_book(symbol)
-        bids_df = pd.DataFrame(order_book["bids"], columns=["price", "quantity"]).astype({"price": "float"})
-        asks_df = pd.DataFrame(order_book["asks"], columns=["price", "quantity"]).astype({"price": "float"})
+        bids_df = pd.DataFrame(
+            order_book["bids"], columns=["price", "quantity"]
+        ).astype({"price": "float"})
+        asks_df = pd.DataFrame(
+            order_book["asks"], columns=["price", "quantity"]
+        ).astype({"price": "float"})
         best_bid = bids_df["price"].max()
         best_ask = asks_df["price"].min()
 
@@ -59,12 +64,16 @@ def fetch_implied_interest_rate(self, symbol: str) -> dict:
             "symbol": symbol,
             "expiry": expiry_str,
             "implied_interest_rate": implied_interest_rate,
+            "days_to_expiry": days_to_expiry,
+            "years_to_expiry": years_to_expiry,
         }
 
     def fetch_all_implied_interest_rates(self, symbols: list[str]) -> pd.DataFrame:
         data = [self.fetch_implied_interest_rate(symbol) for symbol in symbols]
         rates_data = pd.DataFrame(data)
 
         rates_data["expiry"] = pd.to_datetime(rates_data["expiry"], format="%y%m%d")
+        # expiry in human readable format
+        rates_data["expiry"] = rates_data["expiry"].dt.strftime("%Y-%m-%d")
 
         return rates_data

exchanges/fetchers/option_fetcher.py

@@ -1,6 +1,7 @@
 import logging
 from datetime import time, datetime
 
+import ccxt
 import numpy as np
 import pandas as pd
 import requests
@@ -61,97 +62,39 @@ def process_deribit_data(self, df: pd.DataFrame) -> pd.DataFrame:
         df["ask"] *= underlying_prices
         df["mark_price"] *= underlying_prices
 
-        df["bid_spread"] = np.maximum(df["mark_price"] - df["bid"], 0)
-        df["ask_spread"] = np.maximum(df["ask"] - df["mark_price"], 0)
-
-        df["underlying_price"] = underlying_prices
-
-        df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms")
-
-        df["expiry"] = df["symbol"].apply(self.date_parser)
-        df["strike_price"], df["option_type"] = zip(
-            *df["symbol"].apply(self.get_strike_price_and_option_type)
-        )
-        df["datetime_hum"] = df["datetime"].dt.strftime("%Y-%m-%d %H:%M:%S")
-        df["expiry_hum"] = df["expiry"].dt.strftime("%Y-%m-%d %H:%M:%S")
-        # 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",
-                "bid_spread",
-                "ask_spread",
-                "datetime",
-                "expiry",
-                "YTM",
-                "datetime_hum",
-                "expiry_hum",
-                "strike_price",
-                "option_type",
-                "underlying_price",
             ]
         ]
 
     def process_okx_data(self, df: pd.DataFrame) -> pd.DataFrame:
-        # Fetch mark prices from OKX API
         response = requests.get(
             "https://www.okx.com/api/v5/public/mark-price?instType=OPTION"
         )
         mark_prices = response.json()["data"]
         mark_prices_df = pd.DataFrame(mark_prices)
-
-        # Convert 'instId' in mark_prices_df to 'symbol' format
         mark_prices_df["symbol"] = mark_prices_df["instId"].apply(
             self.convert_inst_id_to_symbol
         )
-        # Continue with the rest of the process, assuming the rest of your method is correct
+        mark_prices_df.rename(columns={"markPx": "mark_price"}, inplace=True)
         df["underlying_price"] = self.exchange.fetch_ticker("BTC/USDT")["last"]
         df["bid"] *= df["underlying_price"]
         df["ask"] *= df["underlying_price"]
 
-        df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms")
-        df["expiry"] = df["symbol"].apply(self.date_parser)
-        df["datetime_hum"] = df["datetime"].dt.strftime("%Y-%m-%d %H:%M:%S")
-        df["expiry_hum"] = df["expiry"].dt.strftime("%Y-%m-%d %H:%M:%S")
-        df["strike_price"], df["option_type"] = zip(
-            *df["symbol"].apply(self.get_strike_price_and_option_type)
-        )
-        df["YTM"] = (df["expiry"] - df["datetime"]) / np.timedelta64(1, "Y")
-
-        # Merge the mark prices into the df based on the new 'symbol'
-        df = df.merge(mark_prices_df[["symbol", "markPx"]], on="symbol", how="left")
-
-        # Rename the 'markPx' column to 'mark_price' for clarity (optional)
-        df.rename(columns={"markPx": "mark_price"}, inplace=True)
-        df["mark_price"] = (
-            pd.to_numeric(df["mark_price"], errors="coerce").fillna(0.0)
-            * df["underlying_price"]
-        )
-        df["bid_spread"] = np.maximum(df["mark_price"] - df["bid"], 0)
-        df["ask_spread"] = np.maximum(df["ask"] - df["mark_price"], 0)
+        df = df.merge(mark_prices_df[["symbol", "mark_price"]], on="symbol", how="left")
+        df["mark_price"] = pd.to_numeric(df["mark_price"], errors="coerce").fillna(0.0)
+        df["mark_price"] *= df["underlying_price"]
 
-        # Select and return the desired columns, including the new 'mark_price'
         return df[
             [
                 "symbol",
                 "bid",
                 "ask",
                 "mark_price",
-                "bid_spread",
-                "ask_spread",
-                "underlying_price",
-                "datetime",
-                "expiry",
-                "datetime_hum",
-                "expiry_hum",
-                "strike_price",
-                "option_type",
-                "YTM",
             ]
         ]
 
@@ -160,72 +103,24 @@ def process_binance_data(self, df: pd.DataFrame) -> pd.DataFrame:
         df["bid"] = df["info"].apply(lambda x: float(x.get("bidPrice", 0)))
         df["ask"] = df["info"].apply(lambda x: float(x.get("askPrice", 0)))
 
-        now = datetime.now()
-        df["datetime"] = pd.to_datetime(now)
-
-        df["expiry"] = df["symbol"].apply(self.date_parser)
-        df["datetime_hum"] = df["datetime"].dt.strftime("%Y-%m-%d %H:%M:%S")
-        df["expiry_hum"] = df["expiry"].dt.strftime("%Y-%m-%d %H:%M:%S")
-        df["strike_price"], df["option_type"] = zip(
-            *df["symbol"].apply(self.get_strike_price_and_option_type)
-        )
-        df["YTM"] = (df["expiry"] - df["datetime"]) / np.timedelta64(1, "Y")
-        df["underlying_price"] = self.binance_fetcher.fetch_spot_price("BTCUSDT")
-
-        forward_prices_df = self.binance_fetcher.fetch_mark_price_futures()
-        forward_prices_df["expiry"] = pd.to_datetime(
-            forward_prices_df["expiry"], format="%y%m%d"
-        )
-        # find all dates between forward_prices_df["expiry"]) in df["expiry"]
-        filtered_dates_df = df[
-            df["expiry"].between(
-                forward_prices_df["expiry"].min(), forward_prices_df["expiry"].max()
-            )
-        ]
-        # Now We have in forward_prices_df:
-        # symbol  forward_price     expiry
-        # 0  BTCUSDT_240628       66702.85 2024-06-28
-        # 1  BTCUSDT_240329       63982.80 2024-03-29
-        # I want to calculate the forward price for each expiry date in df["expiry"] like for 2024-04-01
-        print(forward_prices_df)
-        filtered_dates_df.to_json("df.json", orient="records", indent=4)
+        mark_price = self.binance_fetcher.fetch_mark_price_options()
+        mark_price["symbol"] = mark_price["symbol"].apply(self.transform_symbol_format)
+        mark_price.rename(columns={"markPrice": "mark_price"}, inplace=True)
+        mark_price["mark_price"] = pd.to_numeric(
+            mark_price["mark_price"], errors="coerce"
+        ).fillna(0.0)
 
-        df = df.merge(forward_prices_df, on="symbol", how="left")
+        df = df.merge(mark_price, on="symbol", how="left")
 
         return df[
             [
                 "symbol",
                 "bid",
                 "ask",
                 "mark_price",
-                "underlying_price",
-                "forward_price",
-                "bid_spread",
-                "ask_spread",
-                "datetime",
-                "expiry",
-                "datetime_hum",
-                "expiry_hum",
-                "strike_price",
-                "option_type",
-                "YTM",
             ]
         ]
 
-    @staticmethod
-    def date_parser(symbol: str) -> Timestamp | Timestamp | NaTType:
-        date_formats = ["%y%m%d", "%d%b%y", "%Y%m%d", "%m%d%Y", "%d%m%Y"]
-        parts = symbol.replace(":", "-").replace("/", "-").replace(".", "-").split("-")
-        for part in parts:
-            if part.isalpha():
-                continue
-            for date_format in date_formats:
-                try:
-                    return pd.to_datetime(part, format=date_format)
-                except ValueError:
-                    continue
-        return pd.NaT
-
     @staticmethod
     def convert_inst_id_to_symbol(inst_id: str) -> str:
         parts = inst_id.split("-")
@@ -255,3 +150,12 @@ def get_strike_price_and_option_type(symbol: str) -> tuple[str, str]:
         strike_price = parts[-2]
         option_type = parts[-1]
         return strike_price, option_type
+
+
+if __name__ == "__main__":
+    exchange = ccxt.okx()
+    option_fetcher = OptionFetcher(exchange)
+    market_symbols = ["BTC-240329-15000-P", "BTC-240329-20000-C"]
+    exchange_name = "OKX"
+    option_fetcher.fetch_market_data(market_symbols, exchange_name)
+    print("Market data fetched and processed successfully!")

exchanges/handlers/future.py

@@ -14,6 +14,6 @@ def __init__(self, exchange: str, market_types: List[str]):
         self.market_types = market_types
         self.data_fetcher = FutureFetcher(exchange)
 
-    def handle(self) -> dict:
+    def handle(self) -> pd.DataFrame:
         future_symbols = self.data_fetcher.fetch_future_market_symbols("BTC")
         return self.data_fetcher.fetch_all_implied_interest_rates(future_symbols)

exchanges/handlers/merge.py

@@ -18,12 +18,9 @@ def __init__(self, exchange, market_types):
 
     def handle(
         self, options_market: List[str], future_market: List[str] | None
-    ) -> pd.DataFrame:
-        # options_data = self.option_market_handler.handle(options_market)
-        implied_interest_rates = self.future_market_handler.handle()
-        # valid_quotes = self.processing.eliminate_invalid_quotes(options_data)
-        implied_interest_rates.to_json(
-            f"{self.exchange}_implied_interest_rates.json", orient="records", indent=4
-        )
+    ) -> tuple[pd.DataFrame, pd.DataFrame]:
+        options_data = self.option_market_handler.handle(options_market)
+        options_data = self.processing.eliminate_invalid_quotes(options_data)
+        future_data = self.future_market_handler.handle()
 
-        return implied_interest_rates
+        return options_data, future_data

exchanges/handlers/option.py

@@ -23,5 +23,6 @@ def handle(self, market_symbols: List[str]) -> pd.DataFrame:
         market_data = self.data_fetcher.fetch_market_data(
             market_symbols, str(self.exchange)
         )
+        market_data.to_json("option_market_data.json", orient="records", indent=4)
 
         return market_data

exchanges/main.py

@@ -23,30 +23,16 @@ def main():
         global_orderbook_options = pd.DataFrame()
         global_orderbook_futures = pd.DataFrame()
 
+        for manager in [binance, deribit]:
+            options, futures = manager.load_specific_pairs()
+            global_orderbook_options = pd.concat([global_orderbook_options, options]).reset_index(drop=True)
+            global_orderbook_futures = pd.concat([global_orderbook_futures, futures]).reset_index(drop=True)
 
-        for manager in [binance, deribit, okx]:
-            global_orderbook_futures = pd.concat(
-                [global_orderbook_futures, manager.load_specific_pairs()], ignore_index=True
-            )
-        global_orderbook_futures.to_json("global_orderbook.json", orient="records", indent=4)
-        yield_curve = Processing().calculate_yield_curve(global_orderbook_futures)
-        yield_curve.to_json("yield_curve.json", orient="records", indent=4)
-        # build graph
-
-        plt.figure(figsize=(10, 6))
-        plt.plot(yield_curve['expiry'], yield_curve['implied_interest_rate'], marker='o', linestyle='-',
-                 color='blue')
-        plt.title('BTC Futures Implied Interest Rate Yield Curve')
-        plt.xlabel('Expiry Date')
-        plt.ylabel('Average Implied Interest Rate (%)')
-        plt.grid(True)
-        plt.xticks(rotation=45)
-        plt.tight_layout()
-        plt.show()
-        # global_orderbook = pd.read_json("global_orderbook.json")
-        # process = Processing()
-        # x = process.process_global_orderbook(global_orderbook)
-        # x.to_json("global_orderbook_processed.json", orient="records", indent=4)
+        consolidated_options = Processing().consolidate_quotes(global_orderbook_options)
+
+        global_orderbook_futures.to_json("futures.json", orient="records", indent=4)
+        global_orderbook_options.to_json("options.json", orient="records", indent=4)
+        consolidated_options.to_json("consolidated_options.json", orient="records", indent=4)
 
     except Exception as e:
         logger.error(f"An unexpected error occurred in the main function: {e}")