Merge pull request #30 from drift-labs:goldhaxx/DPE-3192/oi-statistics-per-authority-and-user-account

goldhaxx · web-flow · commit 26429b67be0d · 2025-05-12T21:09:49.000-07:00
Enhance Open Interest API and Streamlit Page Functionality
diff --git a/backend/api/open_interest_api.py b/backend/api/open_interest_api.py
@@ -1,5 +1,6 @@
 from fastapi import APIRouter
 from driftpy.constants.numeric_constants import PRICE_PRECISION
+from driftpy.constants.config import mainnet_perp_market_configs
 
 from backend.state import BackendRequest
 
@@ -65,6 +66,142 @@ async def _get_open_interest_per_authority(request: BackendRequest) -> dict:
         "data": result_list, 
     }
 
+async def _get_open_interest_per_account(request: BackendRequest) -> dict:
+    vat = request.state.backend_state.vat
+    slot = request.state.backend_state.last_oracle_slot
+
+    oi_per_account = {}
+
+    for user_data in vat.users.values(): # user_data is of type UserMapItem (based on health.py it should have user_public_key)
+        user_account = user_data.get_user_account()
+        
+        if user_account is None:
+            # Optionally log this: print(f"Warning: Skipping user_data as get_user_account() returned None.")
+            continue
+
+        user_public_key = str(user_data.user_public_key) # Get user_public_key from user_data
+        authority = str(user_account.authority) # Get authority from user_account
+
+        current_oi_for_account = oi_per_account.get(user_public_key, {
+            'total_open_interest_usd': 0.0,
+            'user_public_key': user_public_key, # Store user_public_key in the dict
+            'authority': authority # Add authority field
+        })
+
+        for position in user_account.perp_positions:
+            if position.base_asset_amount == 0:
+                continue
+
+            market_index = position.market_index
+            oracle_price_data = vat.perp_oracles.get(market_index)
+            
+            if oracle_price_data is None:
+                print(f"Warning: Missing oracle price data for market_index {market_index} for user {user_public_key}. Skipping position.")
+                continue
+            
+            try:
+                oracle_price = float(oracle_price_data.price) / PRICE_PRECISION
+                # All perpetual markets use BASE_PRECISION (10^9) for base asset amounts.
+                decimals = 9
+
+                base_asset_amount_val = position.base_asset_amount
+                if base_asset_amount_val is None: # Should not happen with base_asset_amount == 0 check, but good for safety
+                    print(f"Warning: Position base_asset_amount is None for user {user_public_key}, market {market_index}. Skipping position.")
+                    continue
+
+                position_value_usd = (abs(base_asset_amount_val) / (10**decimals)) * oracle_price
+                current_oi_for_account['total_open_interest_usd'] += position_value_usd
+            except (TypeError, ValueError) as e:
+                base_val_repr = repr(getattr(position, 'base_asset_amount', 'N/A'))
+                oracle_price_repr = repr(getattr(oracle_price_data, 'price', 'N/A'))
+                print(f"Error calculating position_value_usd for user {user_public_key}, market {market_index}: {e}. Base: {base_val_repr}, OraclePriceRaw: {oracle_price_repr}. Skipping position.")
+                continue
+        
+        if current_oi_for_account['total_open_interest_usd'] > 0:
+             oi_per_account[user_public_key] = current_oi_for_account
+    
+    result_list = sorted(list(oi_per_account.values()), key=lambda x: x['total_open_interest_usd'], reverse=True)
+    
+    return {
+        "slot": slot,
+        "data": result_list,
+    }
+
+async def _get_open_positions_detailed(request: BackendRequest) -> dict:
+    vat = request.state.backend_state.vat
+    slot = request.state.backend_state.last_oracle_slot
+    
+    detailed_positions = []
+    decimals = 9 # Constant for perpetual markets base asset amount
+
+    for user_data in vat.users.values():
+        user_account = user_data.get_user_account()
+        
+        if user_account is None:
+            continue
+
+        user_public_key = str(user_data.user_public_key)
+        authority = str(user_account.authority)
+
+        for position in user_account.perp_positions:
+            if position.base_asset_amount == 0:
+                continue
+
+            market_index = position.market_index
+            oracle_price_data = vat.perp_oracles.get(market_index)
+            
+            if oracle_price_data is None:
+                print(f"Warning: Missing oracle price data for market_index {market_index} for user {user_public_key}. Skipping position detail.")
+                continue
+            
+            try:
+                base_asset_amount_val = position.base_asset_amount
+                if base_asset_amount_val is None:
+                    print(f"Warning: Position base_asset_amount is None for user {user_public_key}, market {market_index}. Skipping position detail.")
+                    continue
+
+                oracle_price = float(oracle_price_data.price) / PRICE_PRECISION
+                position_value_usd = (abs(base_asset_amount_val) / (10**decimals)) * oracle_price
+                base_asset_amount_ui = base_asset_amount_val / (10**decimals)
+
+                # Get market symbol
+                market_symbol = 'N/A'
+                try:
+                    market_symbol = mainnet_perp_market_configs[market_index].symbol
+                except (IndexError, AttributeError, KeyError) as symbol_e:
+                     print(f"Warning: Could not find symbol for market_index {market_index}. Error: {symbol_e}")
+
+                detailed_positions.append({
+                    'market_index': market_index,
+                    'market_symbol': market_symbol,
+                    'base_asset_amount_ui': base_asset_amount_ui,
+                    'position_value_usd': position_value_usd,
+                    'user_public_key': user_public_key,
+                    'authority': authority
+                })
+
+            except (TypeError, ValueError) as e:
+                base_val_repr = repr(getattr(position, 'base_asset_amount', 'N/A'))
+                oracle_price_repr = repr(getattr(oracle_price_data, 'price', 'N/A'))
+                print(f"Error calculating detailed position data for user {user_public_key}, market {market_index}: {e}. Base: {base_val_repr}, OraclePriceRaw: {oracle_price_repr}. Skipping position detail.")
+                continue
+                
+    # Sort by notional value descending
+    sorted_positions = sorted(detailed_positions, key=lambda x: x['position_value_usd'], reverse=True)
+    
+    return {
+        "slot": slot,
+        "data": sorted_positions,
+    }
+
 @router.get("/per-authority")
 async def get_open_interest_per_authority(request: BackendRequest):
     return await _get_open_interest_per_authority(request)
+
+@router.get("/per-account")
+async def get_open_interest_per_account(request: BackendRequest):
+    return await _get_open_interest_per_account(request)
+
+@router.get("/detailed-positions")
+async def get_open_positions_detailed(request: BackendRequest):
+    return await _get_open_positions_detailed(request)
diff --git a/src/page/open_interest_page.py b/src/page/open_interest_page.py
@@ -1,27 +1,83 @@
 import streamlit as st
 import pandas as pd
+import time # Added import
 from lib.api import fetch_api_data
 from utils import get_current_slot
 
+RETRY_DELAY_SECONDS = 5 # Define delay for auto-refresh
+
+def is_processing(result):
+    """Checks if the API result indicates backend processing."""
+    # Check if result is a dictionary and has the specific processing structure
+    return isinstance(result, dict) and result.get("result") == "processing"
+
+def has_error(result):
+    """Checks if the API result indicates an error."""
+    # Check if result is None or a dictionary with an error message
+    return result is None or (isinstance(result, dict) and result.get("result") == "error")
+
 def open_interest_page():
-    st.title("Open Interest per Authority")
+    st.title("Open Interest on Drift")
 
     try:
-        # Fetch data from the API endpoint directly
-        result = fetch_api_data(
+        # --- 1. Fetch all data --- 
+        result_authority = fetch_api_data(
             section="open-interest", 
             path="per-authority",
-            retry=True # Enable retry for cache misses/processing states
+            retry=False # Let the page handle retries via rerun
+        )
+        result_account = fetch_api_data(
+            section="open-interest",
+            path="per-account",
+            retry=False
+        )
+        result_detailed = fetch_api_data(
+            section="open-interest",
+            path="detailed-positions",
+            retry=False
         )
+
+        # --- 2. Check for processing state --- 
+        if is_processing(result_authority) or is_processing(result_account) or is_processing(result_detailed):
+            st.info(f"Backend is processing open interest data. Auto-refreshing in {RETRY_DELAY_SECONDS} seconds...")
+            # Optionally add more details about which endpoint is processing if needed
+            # e.g., if is_processing(result_authority): st.caption("Authority data processing...")
+            with st.spinner("Please wait..."):
+                time.sleep(RETRY_DELAY_SECONDS)
+            st.rerun() # Rerun the script to fetch again
+            return # Stop further execution in this run
+
+        # --- 3. Check for errors --- 
+        # Check critical data source first (e.g., authority for total OI metric)
+        if has_error(result_authority):
+            error_msg = result_authority['message'] if isinstance(result_authority, dict) else "Could not connect or fetch data."
+            st.error(f"Failed to fetch essential data (OI by Authority): {error_msg}")
+            if st.button("Retry Fetch"): 
+                # Optionally clear streamlit cache if fetch_api_data uses it
+                # st.cache_data.clear()
+                st.rerun()
+            return # Stop execution
+
+        # Handle potential errors in other non-critical fetches (optional: display warnings instead of stopping)
+        if has_error(result_account):
+            st.warning("Could not fetch OI by Account data. This table will be empty.")
+            # Proceed without account data
         
-        if result is None:
-            st.error("Failed to fetch data from the API.")
-            return
+        if has_error(result_detailed):
+            st.warning("Could not fetch Detailed Open Positions data. This table will be empty.")
+            # Proceed without detailed data
 
-        data = result.get("data", [])
-        slot = result.get("slot", "N/A")
-        current_slot = get_current_slot()
+        # --- 4. Extract data and slot (if all checks passed) --- 
+        # We know result_authority is valid here
+        data_authority = result_authority.get("data", [])
+        slot = result_authority.get("slot", "N/A")
+        
+        # Extract data for others, defaulting to empty list if fetch failed (handled by warnings above)
+        data_account = result_account.get("data", []) if isinstance(result_account, dict) else []
+        data_detailed = result_detailed.get("data", []) if isinstance(result_detailed, dict) else []
 
+        # --- 5. Display Slot Info --- 
+        current_slot = get_current_slot()
         if slot != "N/A" and current_slot:
             try:
                 slot_age = int(current_slot) - int(slot)
@@ -31,31 +87,84 @@ def open_interest_page():
         else:
             st.info(f"Slot information unavailable. Current slot: {current_slot}")
 
-        if not data:
-            st.warning("No open interest data found.")
-            return
-
-        df = pd.DataFrame(data)
-
-        if df.empty:
-            st.warning("No open interest data to display.")
-            return
-
-        # Rename columns for better readability
-        df.rename(columns={
-            'authority': 'User Authority',
-            'total_open_interest_usd': 'Total Open Interest (USD)'
-        }, inplace=True)
-        # Reorder columns
-        df = df[["User Authority", "Total Open Interest (USD)"]]
-        # Format USD column with dollar sign and commas
-        df["Total Open Interest (USD)"] = df["Total Open Interest (USD)"].apply(lambda x: f"${x:,.2f}")
+        # --- 6. Process and Prepare DataFrames --- 
+        df_authority = pd.DataFrame() 
+        if data_authority:
+            df_authority = pd.DataFrame(data_authority)
+            if not df_authority.empty:
+                df_authority.rename(columns={
+                    'authority': 'User Authority',
+                    'total_open_interest_usd': 'Total Open Interest (USD)'
+                }, inplace=True)
+                df_authority = df_authority[["User Authority", "Total Open Interest (USD)"]]
+                df_authority["Total Open Interest (USD)"] = df_authority["Total Open Interest (USD)"].apply(lambda x: f"${x:,.2f}")
         
-        st.metric("Total Authorities with Open Interest", len(df))
-        st.metric("Total Open Interest (USD)", f"{df['Total Open Interest (USD)'].str.replace('$','').str.replace(',','').astype(float).sum():,.2f}")
+        df_account = pd.DataFrame()
+        if data_account:
+            df_account = pd.DataFrame(data_account)
+            if not df_account.empty:
+                df_account.rename(columns={
+                    'user_public_key': 'User Account',
+                    'authority': 'Authority',
+                    'total_open_interest_usd': 'Total Open Interest (USD)'
+                }, inplace=True)
+                df_account = df_account[['User Account', 'Authority', 'Total Open Interest (USD)']]
+                df_account["Total Open Interest (USD)"] = df_account["Total Open Interest (USD)"].apply(lambda x: f"${x:,.2f}")
+
+        df_detailed = pd.DataFrame()
+        if data_detailed:
+            df_detailed = pd.DataFrame(data_detailed)
+            if not df_detailed.empty:
+                df_detailed.rename(columns={
+                    'market_index': 'Market Index',
+                    'market_symbol': 'Market Symbol',
+                    'base_asset_amount_ui': 'Base Asset Amount',
+                    'position_value_usd': 'Notional Value (USD)',
+                    'user_public_key': 'User Account',
+                    'authority': 'Authority'
+                }, inplace=True)
+                df_detailed = df_detailed[[
+                    'Market Index', 'Market Symbol', 'Base Asset Amount', 'Notional Value (USD)',
+                    'User Account', 'Authority'
+                ]]
+                df_detailed['Base Asset Amount'] = df_detailed['Base Asset Amount'].apply(lambda x: f"{x:,.4f}")
+                df_detailed['Notional Value (USD)'] = df_detailed['Notional Value (USD)'].apply(lambda x: f"${x:,.2f}")
+
+        # --- 7. Display Layout (Metrics and DataFrames) --- 
+        total_oi_usd = 0.0
+        if not df_authority.empty:
+             try:
+                # Calculate metric only if df_authority is valid and processed
+                total_oi_usd = df_authority['Total Open Interest (USD)'].str.replace('[$,]','', regex=True).astype(float).sum()
+             except Exception as calc_e:
+                 st.warning(f"Could not calculate Total OI metric: {calc_e}")
 
-        st.subheader("Open Interest Details")
-        st.dataframe(df, hide_index=True)
+        col1, col2 = st.columns(2)
+        with col1:
+            st.metric("Total Authorities with Open Interest", len(df_authority) if not df_authority.empty else 0)
+        with col2:
+            st.metric("Total Open Interest (USD)", f"${total_oi_usd:,.2f}")
+
+        col3, col4 = st.columns(2)
+        with col3:
+            st.subheader("OI by Authority")
+            if not df_authority.empty:
+                st.dataframe(df_authority, hide_index=True)
+            else:
+                st.info("Authority data not available or empty.")
+
+        with col4:
+            st.subheader("OI by Account")
+            if not df_account.empty:
+                st.dataframe(df_account, hide_index=True)
+            else:
+                st.info("Account data not available or empty.")
+
+        st.subheader("Detailed Open Positions")
+        if not df_detailed.empty:
+            st.dataframe(df_detailed, hide_index=True, use_container_width=True)
+        else:
+            st.info("Detailed position data not available or empty.")
 
     except Exception as e:
         st.error(f"An error occurred while displaying the page: {e}")
