perpSnapshot

Unique endpoint allowing to get entire perps positioning of all Hyperliquid traders.

💧New endpoint - this endpoint is not a part of original Hyperliquid API and is added by us for builder convenience.

⚠️ This is an add-on endpoint - access has to be purchased separately.

Overview

perpSnapshot endpoint allows you to access the entire positioning of all traders on Hyperliquid across all assets, including entries and liquidations.

There are two ways to access the data:

Metadata Endpoint (Fast): Check for updates without downloading data
Snapshots Endpoint (Heavy): Download actual market snapshot data

Efficient polling pattern

To minimize bandwidth and server load, you should:

Poll the metadata endpoint to check for updates
Only call the snapshots endpoint when data has changed
Cache the snapshot ID locally to avoid unnecessary downloads

Dependencies and best practices

Dependencies

JavaScript

zstd: For decompression (e.g., fzstd, @gera2ld/zstd)
msgpack: For decoding (e.g., @msgpack/msgpack)

Python

zstd: For decompression (e.g., zstandard)
msgpack: For decoding (e.g., msgpack)
aiohttp: For async HTTP requests

✅ Best Practices

Always poll metadata first - Don't skip the lightweight metadata check.
Cache snapshots locally - Avoid unnecessary downloads of large files.
Handle errors gracefully - Network issues are common; implement retry logic.
Use appropriate poll intervals - 5-10 seconds is usually sufficient.
Request specific markets - Don't use ['ALL'] unless you truly need all data.
Monitor bandwidth usage - Snapshots can be large (1-13MB+).

Error handling

404: No snapshots are available yet. Wait and retry.
500: A server error occurred. Retry with exponential backoff.
Network errors: Implement retry logic with jitter to avoid overwhelming the server.
Parse errors: Log the issue and continue using cached data if available.

Perp Snapshot Metadata (Fast)

Endpoint: POST /info
Purpose: Check if snapshots have been updated
Response Time: ~1ms

Request

{
  "type": "perpSnapshotTimestamp"
}

Response

{
  "snapshot_id": "20240915_state_123456789",
  "timestamp": 1694764800
}

Perp Snapshots (Heavy)

Endpoint: POST /info
Purpose: Download market snapshot data
Response Time: ~100ms+

Request

{
  "type": "perpSnapshots",
  "market_names": ["BTC", "FARTCOIN"]  // or ["ALL"] for all core markets
}

Query Options (HIP-3 Multi-Dex)

Specific markets:

["BTC", "ETH"] → Defaults to the main dex
["xyz:BTC", "vntl:ETH"] → Specify DEX with dex:market format

All markets patterns:

["ALL"] → All markets from the main dex
["ALL:xyz"] → All markets from xyz DEX
["ALL:ALL_DEXES"] → All markets from all available DEXes (takes precedence over everything)

Mixed queries:

["ALL", "ALL:xyz", "BTC"] → All from main dex + all from xyz (BTC ignored, covered by main dex)
["ALL:xyz", "vntl:ETH"] → All from xyz + specific ETH from vntls

Rules:

ALL:ALL_DEXES takes precedence - if present, returns everything from all dexes
Multiple ALL patterns allowed - one per dex (e.g., ["ALL", "ALL:xyz"] gets all from both)
Specific markets are filtered - if a dex is covered by an ALL pattern, specific markets from that dex are ignored
No duplicates - each dex's ALL pattern is deduplicated

Response Headers

Single Market: x-payload-format: msgpack, Content-Encoding: zstd
Multiple Markets: x-payload-format: multi-zstd, x-compression: inner-zstd

Data Format

Single Market Response

Format: Raw zstd-compressed MessagePack data

Decompressed Structure:

[
  "20240915_state_123456789",       // Snapshot ID
  "BTC",                            // Market ID
  [                                 // Positions array
    [1.5, 1500.0, -2.5, 1000.0, 0.0, 5.0, 950.0, 10000.0],
    [/* more position tuples */]
  ],
  [                                 // Addresses array
    "0x1234567890abcdef...",
    "0xfedcba0987654321...",
    /* more addresses */
  ]
}

ℹ️ Position Tuple Format The array p contains tuples with the following 8 values in order:

size
notional_size
funding_pnl
entry_price
leverage_type_flag // 0 is cross and 1 is isolated
leverage_multiplier
liquidation_price  // Adjusted for unified/PM cross-collateral
account_value      // Reflects unified or PM total account value

Unified & Portfolio Margin accounts: For unified account users, account_value includes spot collateral across all dexes. For Portfolio Margin users, account_value reflects the per-settlement-token equity and liquidation_price is recomputed using PM available margin. Use the accountValueSnapshots endpoint to get the full PM breakdown (pm_ratio, avail per branch).

Multiple Markets Response

Format: Custom binary format with length prefixes

Binary Structure:

[market_count: 4 bytes][length1: 4 bytes][zstd_data1][length2: 4 bytes][zstd_data2]...

ℹ️ Position Tuple Format The array p contains tuples with the following 8 values in order:

size
notional_size
funding_pnl
entry_price
leverage_type_flag // 0 is cross and 1 is isolated
leverage_multiplier
liquidation_price  // Adjusted for unified/PM cross-collateral
account_value      // Reflects unified or PM total account value

Implementation examples

Note: Make sure to set HYDROMANCER_API_KEY in your .env file

Note: Make sure to download required packages

const axios = require('axios');
const { compress, decompress } = require('@mongodb-js/zstd');
const msgpack = require('msgpack-lite');
const fs = require('fs').promises;
require('dotenv').config();

class PerpSnapshotClient {
    constructor() {
        this.baseUrl = process.env.HYDROMANCER_API_URL || 'https://api.hydromancer.xyz';
        this.apiKey = process.env.HYDROMANCER_API_KEY;
        
        if (!this.apiKey || this.apiKey === 'INSERT_API_KEY_HERE') {
            throw new Error('API key not found. Please set HYDROMANCER_API_KEY in .env file');
        }
        
        this.cachedTimestamp = null;
        this.cachedSnapshots = {};
        this.outputFile = 'perp_snapshot.json';
    }

    async checkForUpdates() {
        const headers = {
            'Content-Type': 'application/json',
            'Authorization': `Bearer ${this.apiKey}`
        };

        try {
            const response = await axios.post(
                `${this.baseUrl}/info`,
                { type: 'perpSnapshotTimestamp' },
                { headers }
            );

            const currentTimestamp = response.data.timestamp || response.data.snapshot_id;
            const hasUpdates = this.cachedTimestamp !== currentTimestamp;
            
            if (hasUpdates) {
                console.log(`Perp snapshot updated: ${this.cachedTimestamp} -> ${currentTimestamp}`);
                this.cachedTimestamp = currentTimestamp;
            } else {
                console.log(`No perp snapshot updates (timestamp: ${currentTimestamp})`);
            }

            return hasUpdates;
        } catch (error) {
            throw new Error(`Timestamp request failed: ${error.response?.status} - ${error.response?.data || error.message}`);
        }
    }

    async pollSnapshots(marketNames = ['BTC', 'ETH']) {
        try {
            const hasUpdates = await this.checkForUpdates();
            
            if (!hasUpdates) {
                console.log('No updates available, using cached perp data');
                return this.cachedSnapshots;
            }

            console.log('New perp snapshots available, downloading...');
            const snapshots = await this.downloadSnapshots(marketNames);
            
            this.cachedSnapshots = snapshots;
            await this.saveToFile(snapshots);
            
            return snapshots;
        } catch (error) {
            console.error('Error polling perpetual snapshots:', error);
            throw error;
        }
    }

    async downloadSnapshots(marketNames) {
        const headers = {
            'Content-Type': 'application/json',
            'Authorization': `Bearer ${this.apiKey}`
        };

        try {
            const response = await axios.post(
                `${this.baseUrl}/info`,
                { type: 'perpSnapshots', market_names: marketNames },
                { 
                    headers,
                    responseType: 'arraybuffer'
                }
            );

            const payloadFormat = response.headers['x-payload-format'];
            const binaryData = Buffer.from(response.data);
            
            console.log(`Payload format: ${payloadFormat}`);
            console.log(`Downloaded ${binaryData.length} bytes`);
            
            if (payloadFormat === 'multi-zstd') {
                return await this.parseMultipleMarkets(binaryData);
            } else {
                // Single market - could be msgpack or zstd compressed
                return await this.parseSingleMarket(binaryData, marketNames[0]);
            }
        } catch (error) {
            throw new Error(`Perp snapshots request failed: ${error.response?.status} - ${error.response?.data || error.message}`);
        }
    }

    async parseSingleMarket(binaryData, marketName) {
        try {
            // Try to decompress first (if it's zstd compressed)
            let decompressed;
            try {
                decompressed = await decompress(binaryData);
                console.log('Successfully decompressed single market data');
            } catch (e) {
                // If decompression fails, assume it's already uncompressed msgpack
                console.log('Data appears to be uncompressed msgpack');
                decompressed = binaryData;
            }
            
            // Decode msgpack
            const data = msgpack.decode(decompressed);
            
            // According to docs, the structure could be:
            // For perps: [identifier, market, positions, addresses]
            if (Array.isArray(data) && data.length >= 4) {
                const [identifier, market, positions, addresses] = data;
                return {
                    [market]: {
                        identifier: identifier,
                        market: market,
                        positions: positions || [],
                        addresses: addresses || [],
                        timestamp: this.cachedTimestamp
                    }
                };
            }
            
            // Or it could be the object format from docs: {i, m, p, a}
            if (data.i !== undefined && data.m !== undefined) {
                return {
                    [marketName || data.m]: {
                        identifier: data.i,
                        market: data.m,
                        positions: data.p || [],
                        addresses: data.a || [],
                        timestamp: this.cachedTimestamp
                    }
                };
            }
            
            throw new Error('Unknown single market data format');
        } catch (error) {
            console.error('Error parsing single market:', error);
            throw error;
        }
    }

    async parseMultipleMarkets(binaryData) {
        let offset = 0;
        
        // Read number of snapshots (4 bytes, little-endian)
        const count = binaryData.readUInt32LE(offset);
        offset += 4;
        
        console.log(`Number of perp snapshots: ${count}`);
        
        const markets = {};
        
        for (let i = 0; i < count; i++) {
            // Read length of this snapshot (4 bytes, little-endian)
            const length = binaryData.readUInt32LE(offset);
            offset += 4;
            
            console.log(`Snapshot ${i + 1}/${count}: ${length} bytes`);
            
            // Extract and decompress snapshot
            const zstdData = binaryData.slice(offset, offset + length);
            
            try {
                const decompressed = await decompress(zstdData);
                const data = msgpack.decode(decompressed);
                
                // Handle array format: [identifier, market, positions, addresses]
                if (Array.isArray(data) && data.length >= 4) {
                    const [identifier, market, positions, addresses] = data;
                    markets[market] = {
                        identifier: identifier,
                        market: market,
                        positions: positions || [],
                        addresses: addresses || [],
                        timestamp: this.cachedTimestamp
                    };
                    console.log(`Parsed market: ${market} with ${positions?.length || 0} positions`);
                }
                // Handle object format: {i, m, p, a}
                else if (data.i !== undefined && data.m !== undefined) {
                    markets[data.m] = {
                        identifier: data.i,
                        market: data.m,
                        positions: data.p || [],
                        addresses: data.a || [],
                        timestamp: this.cachedTimestamp
                    };
                    console.log(`Parsed market: ${data.m} with ${data.p?.length || 0} positions`);
                }
            } catch (error) {
                console.error(`Error parsing snapshot ${i + 1}:`, error);
            }
            
            offset += length;
        }
        
        return markets;
    }

    async saveToFile(snapshots) {
        const outputData = {
            timestamp: this.cachedTimestamp,
            markets: {}
        };
        
        for (const [market, data] of Object.entries(snapshots)) {
            const positions = data.positions || [];
            const addresses = data.addresses || [];
            
            // Create readable position list (first 100 positions)
            const positionList = [];
            const numPositions = Math.min(100, positions.length, addresses.length);
            
            for (let i = 0; i < numPositions; i++) {
                if (Array.isArray(positions[i]) && positions[i].length >= 8) {
                    positionList.push({
                        address: addresses[i],
                        size: positions[i][0],
                        notional_size: positions[i][1],
                        funding_pnl: positions[i][2],
                        entry_price: positions[i][3],
                        leverage_type_flag: positions[i][4],
                        leverage_multiplier: positions[i][5],
                        liquidation_price: positions[i][6],
                        account_value: positions[i][7]
                    });
                }
            }
            
            outputData.markets[market] = {
                identifier: data.identifier,
                total_positions: positions.length,
                positions: positionList
            };
        }
        
        await fs.writeFile(this.outputFile, JSON.stringify(outputData, null, 2));
        console.log(`Saved perp snapshot to ${this.outputFile}`);
    }
}

// Example usage
async function main() {
    const client = new PerpSnapshotClient();
    
    while (true) {
        try {
            const snapshots = await client.pollSnapshots(['FARTCOIN', 'HYPE']);
            
            for (const [market, data] of Object.entries(snapshots)) {
                console.log(`Market ${market}: ${data.positions?.length || 0} positions`);
            }
            
            console.log('Waiting 5 seconds...');
            await new Promise(resolve => setTimeout(resolve, 5000));
            
        } catch (error) {
            console.error('Polling error:', error.message);
            await new Promise(resolve => setTimeout(resolve, 5000));
        }
    }
}

if (require.main === module) {
    main().catch(console.error);
}

module.exports = PerpSnapshotClient;

import asyncio
import aiohttp
import struct
from typing import Optional, Dict, List, Any
import time
import json
import zstandard as zstd
import msgpack
import os
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()


class PerpSnapshotClient:
    def __init__(self):
        self.base_url = os.getenv('HYDROMANCER_API_URL', 'https://api.hydromancer.xyz')
        self.api_key = os.getenv('HYDROMANCER_API_KEY')
        
        if not self.api_key or self.api_key == 'INSERT_API_KEY_HERE':
            raise ValueError("API key not found. Please set HYDROMANCER_API_KEY in .env file")
            
        self.cached_timestamp: Optional[str] = None
        self.cached_snapshots: Dict[str, Any] = {}
        self.session: Optional[aiohttp.ClientSession] = None
        self.output_file = "perp_snapshot.json"

    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()

    async def check_for_updates(self) -> bool:
        """Check if snapshots have been updated since last poll"""
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}"
        }

        async with self.session.post(
            f'{self.base_url}/info',
            json={'type': 'perpSnapshotTimestamp'},
            headers=headers
        ) as response:
            if not response.ok:
                text = await response.text()
                raise Exception(f'Timestamp request failed: {response.status} - {text}')

            data = await response.json()
            current_timestamp = data.get('timestamp', data.get('snapshot_id'))
            
            has_updates = self.cached_timestamp != current_timestamp
            
            if has_updates:
                print(f'Perp snapshot updated: {self.cached_timestamp} -> {current_timestamp}')
                self.cached_timestamp = current_timestamp
            else:
                print(f'No perp snapshot updates (timestamp: {current_timestamp})')

            return has_updates

    async def poll_snapshots(self, market_names: List[str] = ["BTC", "ETH"]) -> Dict[str, Any]:
        """Efficiently poll for perpetual snapshots"""
        try:
            has_updates = await self.check_for_updates()
            
            if not has_updates:
                print('No updates available, using cached perp data')
                return self.cached_snapshots

            print('New perp snapshots available, downloading...')
            snapshots = await self.download_snapshots(market_names)
            
            self.cached_snapshots = snapshots
            await self.save_to_file(snapshots)
            
            return snapshots
        except Exception as error:
            print(f'Error polling perpetual snapshots: {error}')
            raise

    async def download_snapshots(self, market_names: List[str]) -> Dict[str, Any]:
        """Download perpetual market snapshots"""
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}"
        }

        async with self.session.post(
            f'{self.base_url}/info',
            json={'type': 'perpSnapshots', 'market_names': market_names},
            headers=headers
        ) as response:
            if not response.ok:
                text = await response.text()
                raise Exception(f'Perp snapshots request failed: {response.status} - {text}')

            payload_format = response.headers.get('x-payload-format')
            binary_data = await response.read()
            
            print(f"Payload format: {payload_format}")
            print(f"Downloaded {len(binary_data)} bytes")
            
            if payload_format == 'multi-zstd':
                return await self.parse_multiple_markets(binary_data)
            else:
                # Single market - could be msgpack or zstd compressed
                return await self.parse_single_market(binary_data, market_names[0])

    async def parse_single_market(self, binary_data: bytes, market_name: str) -> Dict[str, Any]:
        """Parse single market response"""
        try:
            # Try to decompress first (if it's zstd compressed)
            try:
                decompressed = self.decompress_zstd(binary_data)
                print('Successfully decompressed single market data')
            except:
                # If decompression fails, assume it's already uncompressed msgpack
                print('Data appears to be uncompressed msgpack')
                decompressed = binary_data
            
            # Decode msgpack
            data = msgpack.unpackb(decompressed, raw=False)
            
            # For perps: [identifier, market, positions, addresses]
            if isinstance(data, list) and len(data) >= 4:
                identifier, market, positions, addresses = data[:4]
                return {
                    market: {
                        'identifier': identifier,
                        'market': market,
                        'positions': positions or [],
                        'addresses': addresses or [],
                        'timestamp': self.cached_timestamp
                    }
                }
            
            # Or object format: {i, m, p, a}
            if isinstance(data, dict) and 'i' in data and 'm' in data:
                return {
                    market_name or data['m']: {
                        'identifier': data['i'],
                        'market': data['m'],
                        'positions': data.get('p', []),
                        'addresses': data.get('a', []),
                        'timestamp': self.cached_timestamp
                    }
                }
            
            raise ValueError('Unknown single market data format')
        except Exception as error:
            print(f'Error parsing single market: {error}')
            raise

    async def parse_multiple_markets(self, binary_data: bytes) -> Dict[str, Any]:
        """Parse multiple markets response (multi-zstd format)"""
        offset = 0
        
        # Read number of snapshots (4 bytes, little-endian)
        count = struct.unpack('<I', binary_data[offset:offset + 4])[0]
        offset += 4
        
        print(f"Number of perp snapshots: {count}")
        
        markets = {}
        
        for i in range(count):
            # Read length of this snapshot (4 bytes, little-endian)
            length = struct.unpack('<I', binary_data[offset:offset + 4])[0]
            offset += 4
            
            print(f"Snapshot {i + 1}/{count}: {length} bytes")
            
            # Extract and decompress snapshot
            zstd_data = binary_data[offset:offset + length]
            
            try:
                decompressed = self.decompress_zstd(zstd_data)
                data = msgpack.unpackb(decompressed, raw=False)
                
                # Handle array format: [identifier, market, positions, addresses]
                if isinstance(data, list) and len(data) >= 4:
                    identifier, market, positions, addresses = data[:4]
                    markets[market] = {
                        'identifier': identifier,
                        'market': market,
                        'positions': positions or [],
                        'addresses': addresses or [],
                        'timestamp': self.cached_timestamp
                    }
                    print(f"Parsed market: {market} with {len(positions) if positions else 0} positions")
                    
                # Handle object format: {i, m, p, a}
                elif isinstance(data, dict) and 'i' in data and 'm' in data:
                    market = data['m']
                    markets[market] = {
                        'identifier': data['i'],
                        'market': market,
                        'positions': data.get('p', []),
                        'addresses': data.get('a', []),
                        'timestamp': self.cached_timestamp
                    }
                    print(f"Parsed market: {market} with {len(data.get('p', [])) if data.get('p') else 0} positions")
                    
            except Exception as e:
                print(f"Error parsing snapshot {i + 1}: {e}")
                
            offset += length
        
        return markets

    def decompress_zstd(self, data: bytes) -> bytes:
        """Decompress zstd data"""
        dctx = zstd.ZstdDecompressor()
        return dctx.decompress(data)

    async def save_to_file(self, snapshots: Dict[str, Any]):
        """Save snapshots to JSON file"""
        output_data = {
            'timestamp': self.cached_timestamp,
            'markets': {}
        }
        
        for market, data in snapshots.items():
            positions = data.get('positions', [])
            addresses = data.get('addresses', [])
            
            position_list = []
            
            for i in range(len(positions)):
                if isinstance(positions[i], list) and len(positions[i]) >= 8:
                    position_list.append({
                        'address': addresses[i],
                        'size': positions[i][0],
                        'notional_size': positions[i][1],
                        'funding_pnl': positions[i][2],
                        'entry_price': positions[i][3],
                        'leverage_type_flag': positions[i][4],
                        'leverage_multiplier': positions[i][5],
                        'liquidation_price': positions[i][6],
                        'account_value': positions[i][7]
                    })
            
            output_data['markets'][market] = {
                'identifier': data.get('identifier'),
                'total_positions': len(positions),
                'positions': position_list
            }
        
        with open(self.output_file, 'w') as f:
            json.dump(output_data, f, indent=2)
        
        print(f"Saved perp snapshot to {self.output_file}")


async def main():
    """Example usage with different market configurations"""
    async with PerpSnapshotClient() as client:
        while True:
            try:
                snapshots = await client.poll_snapshots(['FARTCOIN', 'HYPE'])
                
                for market, data in snapshots.items():
                    print(f"Market {market}: {len(data.get('positions', []))} positions")
                
                print("Waiting 5 seconds...")
                await asyncio.sleep(5)
                
            except Exception as error:
                print(f'Polling error: {error}')
                await asyncio.sleep(5)


if __name__ == "__main__":
    asyncio.run(main())

PreviousMarket data NextperpTwapSnapshot

Last updated 2 months ago