Usage Examples

go run main.go --api-key YOUR_API_KEY

# With custom user
go run main.go --api-key YOUR_API_KEY --user 0x...

# Build binary
go build -o spot_volume
./spot_volume --api-key YOUR_API_KEY

#!/usr/bin/env python3

# /// script
# dependencies = [
#   "requests",
# ]
# ///
"""
Fetch all fills for a Hyperliquid user and calculate total trading volume for spot coins.
Spot coins are identified by names starting with '@'.
"""
import argparse
import sys
from datetime import datetime
from typing import Set, Tuple
import requests

# API Configuration
API_URL = "https://api.allium.so/api/v1/developer/trading/hyperliquid/info/fills"
DEFAULT_USER = "0xc0142fc8aa609f324b44e414816ea549322afbe8"

# Start from a historical date (e.g., Hyperliquid launch)
START_TIME_MS = 1609459200000  # Jan 1, 2021

def fetch_fills(user: str, start_time: int, api_key: str, end_time: int | None = None) -> list:
    """Fetch fills from the API for a given time range."""
    headers = {
        "Content-Type": "application/json",
        "X-API-KEY": api_key,
    }
    
    payload = {
        "user": user,
        "type": "userFillsByTime",
        "startTime": start_time,
    }
    
    if end_time:
        payload["endTime"] = end_time
    
    print("  Making API request...", flush=True)
    try:
        response = requests.post(API_URL, json=payload, headers=headers, timeout=30)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        print(f"Error fetching fills: {e}", file=sys.stderr)
        if hasattr(e, 'response') and e.response is not None:
            print(f"Response: {e.response.text}", file=sys.stderr)
        sys.exit(1)

def calculate_spot_volume(user: str, api_key: str) -> None:
    """Calculate total trading volume for spot coins (coins starting with '@')."""
    
    seen_fills: Set[Tuple[int, str, int]] = set()  # (time, coin, tid)
    total_volume = 0.0
    spot_fill_count = 0
    total_fill_count = 0
    current_start_time = START_TIME_MS
    page_count = 0
    
    print(f"Fetching fills for user: {user}", flush=True)
    print(f"Starting from: {datetime.fromtimestamp(START_TIME_MS / 1000).isoformat()}", flush=True)
    print("-" * 80, flush=True)
    
    while True:
        page_count += 1
        print(f"\nPage {page_count}: Fetching fills from {datetime.fromtimestamp(current_start_time / 1000).isoformat()}...", flush=True)
        
        fills = fetch_fills(user, current_start_time, api_key)
        
        if not fills:
            print("No more fills found.", flush=True)
            break
        
        print(f"  Received {len(fills)} fills", flush=True)
        
        # Track new fills in this batch
        new_fills = 0
        new_spot_fills = 0
        page_volume = 0.0
        
        for fill in fills:
            # Create unique key for this fill
            fill_key = (fill.get("time"), fill.get("coin"), fill.get("tid"))
            
            # Skip if we've already seen this fill
            if fill_key in seen_fills:
                continue
            
            seen_fills.add(fill_key)
            new_fills += 1
            total_fill_count += 1
            
            coin = fill.get("coin", "")
            
            # Check if it's a spot coin (starts with '@')
            if coin.startswith("@"):
                sz = fill.get("sz", "0")
                px = fill.get("px", "0")
                try:
                    size = float(sz)
                    price = float(px)
                    volume = size * price  # Trading volume = size * price
                    total_volume += volume
                    page_volume += volume
                    spot_fill_count += 1
                    new_spot_fills += 1
                except (ValueError, TypeError):
                    print(f"  Warning: Could not parse size '{sz}' or price '{px}' for coin {coin}", file=sys.stderr)
        
        print(f"  New unique fills: {new_fills} (spot: {new_spot_fills}, volume: {page_volume:.4f})", flush=True)
        
        # Check if we need to continue paging
        if len(fills) < 2000:
            print("\nReceived less than 2000 fills, pagination complete.", flush=True)
            break
        
        if new_fills == 0:
            print("\nNo new fills found, all remaining fills already seen. Pagination complete.", flush=True)
            break
        
        # Use the timestamp of the last fill for the next page
        last_timestamp = fills[-1].get("time")
        current_start_time = last_timestamp
        
        print(f"  Continuing from timestamp: {datetime.fromtimestamp(last_timestamp / 1000).isoformat()}", flush=True)
    
    # Print results
    print("\n" + "=" * 80)
    print("RESULTS")
    print("=" * 80)
    print(f"Total fills processed: {total_fill_count:,}")
    print(f"Spot coin fills: {spot_fill_count:,}")
    print(f"Total spot coin trading volume: {total_volume:,.4f}")
    print(f"Pages fetched: {page_count}")
    print("=" * 80)

def main():
    parser = argparse.ArgumentParser(
        description="Calculate total trading volume for spot coins on Hyperliquid"
    )
    parser.add_argument(
        "--user",
        type=str,
        default=DEFAULT_USER,
        help=f"User wallet address (default: {DEFAULT_USER})"
    )
    parser.add_argument(
        "--api-key",
        type=str,
        required=True,
        help="Allium API key"
    )
    
    args = parser.parse_args()
    
    # Validate user address format (basic check)
    if not args.user.startswith("0x") or len(args.user) != 42:
        print("Error: Invalid user address format. Expected 0x followed by 40 hex characters.", file=sys.stderr)
        sys.exit(1)
    
    calculate_spot_volume(args.user, args.api_key)

if __name__ == "__main__":
    main()