Merge pull request #44 from drift-labs:goldhaxx/DATA-75/add-btc-perp-to-user-retention-report

add volume metrics

Author: goldhaxx

backend/api/user_retention_explorer_api.py

@@ -65,18 +65,25 @@ def log_current_identity():
 REGION   = os.environ.get("AWS_REGION", "eu-west-1")
 S3_OUTPUT = os.environ.get("ATHENA_S3_OUTPUT", "s3://mainnet-beta-data-ingestion-bucket/athena/")
 
+class UserVolumeDetails(BaseModel):
+    user_address: str
+    initial_selected_market_volume: float = 0.0
+    initial_other_market_volume: float = 0.0
+    selected_market_volume_14d: float = 0.0
+    other_market_volume_14d: float = 0.0
+    selected_market_volume_28d: float = 0.0
+    other_market_volume_28d: float = 0.0
+
 class RetentionExplorerItem(BaseModel):
     market: str
     category: List[str]
     start_date: str
-    new_traders: int
-    new_traders_list: List[str]
-    retained_users_14d: int
+    new_traders_count: int
+    retained_14d_count: int
+    retained_28d_count: int
     retention_ratio_14d: float
-    retained_users_14d_list: List[str]
-    retained_users_28d: int
     retention_ratio_28d: float
-    retained_users_28d_list: List[str]
+    user_data: List[UserVolumeDetails]
 
 UTC = tz.tzutc()
 
@@ -98,42 +105,93 @@ def partition_pred(parts: Set[Tuple[str, str, str]]) -> str:
 def sql_new_traders(mkt_idx: int, start_dt: datetime) -> str:
     parts = partition_pred(partition_tuples(start_dt, NEW_TRADER_WINDOW_DAYS))
     return f"""
-        SELECT "user",
-               MIN(slot) AS first_slot,
-               MIN(ts)   AS first_ts
-        FROM   eventtype_orderrecord
-        WHERE  ({parts})
-          AND  "order".marketindex = {mkt_idx}
-          AND  ("order".orderid = 0 OR "order".orderid = 1)
-        GROUP  BY "user"
+        WITH potential_new_traders AS (
+            -- First, find users who made their first trade in the target market within the window
+            SELECT "user", MIN(ts) as first_ts
+            FROM eventtype_orderrecord
+            WHERE ({parts})
+              AND "order".marketindex = {mkt_idx}
+            GROUP BY "user"
+        ),
+        all_trades AS (
+            -- Get all trades for these potential new traders across all time
+            -- This is needed to ensure the trade in the window was truly their first one
+            SELECT "user", MIN(ts) as first_ever_ts
+            FROM eventtype_orderrecord
+            WHERE "user" IN (SELECT "user" FROM potential_new_traders)
+            GROUP BY "user"
+        ),
+        true_new_traders AS (
+            -- Filter to users whose first-ever trade matches their first trade in the window
+            SELECT pnt."user"
+            FROM potential_new_traders pnt
+            JOIN all_trades at ON pnt."user" = at."user" AND pnt.first_ts = at.first_ever_ts
+        )
+        -- Finally, join with newuserrecord to ensure it's their first subaccount (subaccountid=0)
+        SELECT tnt."user"
+        FROM true_new_traders tnt
+        JOIN eventtype_newuserrecord nur ON tnt."user" = nur."user"
+        WHERE nur.subaccountid = 0
+"""
+
+# --- New Helper Function for SQL ---
+# Define a scaling factor for quote asset amounts, assuming 6 decimal places (e.g., USDC)
+QUOTE_ASSET_SCALE_FACTOR = 1e6
+
+def sql_users_volume(
+    users: List[str],
+    start_dt: datetime,
+    end_dt: datetime,
+    partition_func: callable,
+    market_index: Optional[int] = None,
+    exclude_market_index: Optional[int] = None
+) -> str:
     """
+    Generates SQL to calculate the trading volume for a list of users in a given time period.
+    Volume is calculated from `quoteassetamountfilled` in `eventtype_traderecord`.
+    """
+    if not users:
+        return ""
+
+    user_list = "', '".join(users)
+    
+    market_filter = ""
+    if market_index is not None:
+        market_filter = f"AND marketindex = {market_index}"
+    if exclude_market_index is not None:
+        market_filter = f"AND marketindex <> {exclude_market_index}"
+
+    start_ts = int(start_dt.timestamp())
+    # end_dt is exclusive, so we don't subtract one day
+    end_ts = int(end_dt.timestamp())
 
-def sql_retention_users_chunk(traders: List[str],
-                               mkt_idx: int,
-                               chunk_start: datetime,
-                               chunk_days: int) -> str:
-    chunk_end = chunk_start + timedelta(days=chunk_days)
-    start_ts = int(chunk_start.timestamp())
-    end_ts = int(chunk_end.timestamp())
-    from_date = chunk_start.strftime('%Y%m%d')
-    to_date = chunk_end.strftime('%Y%m%d')
-    trader_list = "', '".join(traders)
-
-    return f'''
-        WITH time_range AS (
-            SELECT 
-                {start_ts} AS from_ts,
-                {end_ts} AS to_ts,
-                '{from_date}' AS from_date,
-                '{to_date}' AS to_date
+    # Use partitions for traderecord based on the provided function
+    num_days = (end_dt - start_dt).days
+    parts = partition_func(partition_tuples(start_dt, num_days))
+
+    # This query sums up volume for users as both takers and makers.
+    return f"""
+        WITH user_trades AS (
+            SELECT
+                taker AS user,
+                CAST(quoteassetamountfilled AS DOUBLE) / {QUOTE_ASSET_SCALE_FACTOR} AS quote_volume
+            FROM eventtype_traderecord
+            WHERE ({parts}) AND CAST(ts AS BIGINT) >= {start_ts} AND CAST(ts AS BIGINT) < {end_ts}
+              AND taker IN ('{user_list}') {market_filter}
+            UNION ALL
+            SELECT
+                maker AS user,
+                CAST(quoteassetamountfilled AS DOUBLE) / {QUOTE_ASSET_SCALE_FACTOR} AS quote_volume
+            FROM eventtype_traderecord
+            WHERE ({parts}) AND CAST(ts AS BIGINT) >= {start_ts} AND CAST(ts AS BIGINT) < {end_ts}
+              AND maker IN ('{user_list}') {market_filter}
         )
-        SELECT DISTINCT "user"
-        FROM   eventtype_orderrecord, time_range
-        WHERE  CAST(ts AS INT) BETWEEN time_range.from_ts AND time_range.to_ts
-          AND  CONCAT(year, month, day) BETWEEN time_range.from_date AND time_range.to_date
-          AND  "order".marketindex <> {mkt_idx}
-          AND  "user" IN ('{trader_list}')
-    '''
+        SELECT
+            user,
+            SUM(quote_volume) AS total_volume
+        FROM user_trades
+        GROUP BY user
+    """
 
 async def calculate_retention_for_market(market_name: str, start_date_str: str) -> Dict[str, Any]:
     conn = None
@@ -143,61 +201,77 @@ async def calculate_retention_for_market(market_name: str, start_date_str: str)
         if not market_config:
             raise HTTPException(status_code=404, detail=f"Market '{market_name}' not found.")
 
+        mkt_idx = market_config["index"]
+
         logger.info(f"Connecting to Athena. S3 staging: {S3_OUTPUT}, Region: {REGION}, DB: {DATABASE}")
         conn = connect(s3_staging_dir=S3_OUTPUT, region_name=REGION, schema_name=DATABASE)
         logger.info("Successfully connected to Athena.")
         log_current_identity()
 
-        # 1. Find new traders for the given market and date
+        # 1. Find new traders
         logger.info(f"Scanning for new traders for {market_name} from {start_date_str}...")
-        q_new_traders = sql_new_traders(market_config["index"], start_date)
+        q_new_traders = sql_new_traders(mkt_idx, start_date)
         new_traders_df = pd.read_sql(q_new_traders, conn)
-        logger.info(f"Found {len(new_traders_df)} new traders for {market_name}.")
-        
         mkt_traders = new_traders_df["user"].tolist()
         new_traders_count = len(mkt_traders)
+
+        if not mkt_traders:
+            return {
+                "market": market_name, "category": market_config.get("category", []), "start_date": start_date_str,
+                "new_traders_count": 0, "retained_14d_count": 0, "retained_28d_count": 0,
+                "retention_ratio_14d": 0.0, "retention_ratio_28d": 0.0, "user_data": []
+            }
+
+        # --- Volume Calculation ---
+        traders_df = pd.DataFrame(mkt_traders, columns=['user'])
+
+        # Define time windows
+        initial_window_end = start_date + timedelta(days=NEW_TRADER_WINDOW_DAYS)
+        retention_14d_end = initial_window_end + timedelta(days=14)
+        retention_28d_end = initial_window_end + timedelta(days=28)
+
+        def get_and_merge_volume(
+            base_df: pd.DataFrame, column_name: str, start_dt: datetime, end_dt: datetime,
+            m_idx: Optional[int] = None, exclude_m_idx: Optional[int] = None
+        ) -> pd.DataFrame:
+            logger.info(f"Calculating volume for '{column_name}'...")
+            vol_sql = sql_users_volume(mkt_traders, start_dt, end_dt, partition_pred, m_idx, exclude_m_idx)
+            if not vol_sql: return base_df.assign(**{column_name: 0.0})
+            
+            vol_df = pd.read_sql(vol_sql, conn)
+            merged_df = base_df.merge(vol_df, on='user', how='left')
+            merged_df['total_volume'] = merged_df['total_volume'].fillna(0)
+            return merged_df.rename(columns={'total_volume': column_name})
+
+        # Calculate all volume metrics
+        traders_df = get_and_merge_volume(traders_df, 'initial_selected_market_volume', start_date, initial_window_end, m_idx=mkt_idx)
+        traders_df = get_and_merge_volume(traders_df, 'initial_other_market_volume', start_date, initial_window_end, exclude_m_idx=mkt_idx)
+        traders_df = get_and_merge_volume(traders_df, 'selected_market_volume_14d', initial_window_end, retention_14d_end, m_idx=mkt_idx)
+        traders_df = get_and_merge_volume(traders_df, 'other_market_volume_14d', initial_window_end, retention_14d_end, exclude_m_idx=mkt_idx)
+        traders_df = get_and_merge_volume(traders_df, 'selected_market_volume_28d', initial_window_end, retention_28d_end, m_idx=mkt_idx)
+        traders_df = get_and_merge_volume(traders_df, 'other_market_volume_28d', initial_window_end, retention_28d_end, exclude_m_idx=mkt_idx)
 
+        traders_df = traders_df.rename(columns={'user': 'user_address'})
+
+        # --- Summary Statistics ---
+        retained_14d_count = int((traders_df['other_market_volume_14d'] > 0).sum())
+        retained_28d_count = int((traders_df['other_market_volume_28d'] > 0).sum())
+        retention_ratio_14d = (retained_14d_count / new_traders_count) if new_traders_count > 0 else 0.0
+        retention_ratio_28d = (retained_28d_count / new_traders_count) if new_traders_count > 0 else 0.0
+
         result = {
             "market": market_name,
             "category": market_config.get("category", []),
             "start_date": start_date_str,
-            "new_traders": new_traders_count,
-            "new_traders_list": mkt_traders,
+            "new_traders_count": new_traders_count,
+            "retained_14d_count": retained_14d_count,
+            "retained_28d_count": retained_28d_count,
+            "retention_ratio_14d": round(retention_ratio_14d, 4),
+            "retention_ratio_28d": round(retention_ratio_28d, 4),
+            "user_data": traders_df.to_dict('records')
         }
-
-        # 2. Calculate retention for each window
-        if not mkt_traders:
-             for win in RETENTION_WINDOWS_DAYS:
-                result[f"retained_users_{win}d"] = 0
-                result[f"retention_ratio_{win}d"] = 0.0
-                result[f"retained_users_{win}d_list"] = []
-             return result
-
-        retention_period_start_dt = start_date
-        for win in RETENTION_WINDOWS_DAYS:
-            offset = 0
-            retained_set: Set[str] = set()
-            
-            while offset < win:
-                chunk_start_dt = retention_period_start_dt + timedelta(days=offset)
-                span = min(CHUNK_DAYS, win - offset)
-                if span <= 0: break
-
-                logger.info(f"Fetching retention for {market_name}, window {win}d, chunk: {chunk_start_dt.strftime('%Y-%m-%d')} for {span} days")
-                q_retention_chunk = sql_retention_users_chunk(mkt_traders, market_config["index"], chunk_start_dt, span)
-                retained_users_df = pd.read_sql(q_retention_chunk, conn)
-                retained_set.update(retained_users_df["user"].tolist())
-                offset += CHUNK_DAYS
-
-            retained_list = sorted(list(retained_set))
-            retained_count = len(retained_list)
-            retention_ratio = (retained_count / new_traders_count) if new_traders_count > 0 else 0.0
-
-            result[f"retained_users_{win}d"] = retained_count
-            result[f"retention_ratio_{win}d"] = round(retention_ratio, 4)
-            result[f"retained_users_{win}d_list"] = retained_list
 
-        logger.info(f"Successfully calculated retention for {market_name}.")
+        logger.info(f"Successfully calculated consolidated retention for {market_name}.")
         return result
 
     except Exception as e:
@@ -229,7 +303,6 @@ async def get_retention_for_market(
     """
     try:
         logger.info(f"Received request for /calculate: market='{market_name}', date='{start_date}'")
-        # Input validation for date format can be added here if needed
         result_data = await calculate_retention_for_market(market_name, start_date)
         return RetentionExplorerItem(**result_data)
     except HTTPException as http_exc: