StreamOrderbookSnapshots - Real-time Order Book Streaming

{
  "coin": "BTC",
  "side": "B",
  "limitPx": "84500.0",
  "sz": "0.5",
  "oid": 333003526755,
  "timestamp": 1772276628506,
  "triggerCondition": "N/A",
  "isTrigger": false,
  "triggerPx": "0.0",
  "children": [],
  "isPositionTpsl": false,
  "reduceOnly": false,
  "orderType": "Limit",
  "origSz": "0.5",
  "tif": "Alo",
  "cloid": "0x00000000000000000000000000000318"
}

class OrderFlowTracker:
    def __init__(self):
        self.previous_orders = {}  # oid -> order

    def track(self, snapshot):
        """Compare snapshots to detect order changes"""
        current_orders = {}

        for market in snapshot:
            coin = market[0]
            bids = market[1][0]
            asks = market[1][1]

            for order in bids + asks:
                current_orders[order['oid']] = order

        # Detect changes
        new_oids = set(current_orders) - set(self.previous_orders)
        removed_oids = set(self.previous_orders) - set(current_orders)

        for oid in new_oids:
            order = current_orders[oid]
            print(f'NEW: {order["coin"]} {order["side"]} '
                  f'{order["sz"]} @ {order["limitPx"]} '
                  f'({order["orderType"]})')

        for oid in removed_oids:
            order = self.previous_orders[oid]
            print(f'REMOVED: {order["coin"]} {order["side"]} '
                  f'{order["sz"]} @ {order["limitPx"]}')

        self.previous_orders = current_orders

function analyzeTriggerOrders(snapshot, targetCoin) {
    for (const market of snapshot) {
        if (market[0] !== targetCoin) continue;

        const allOrders = [...market[1][0], ...market[1][1]];
        const triggerOrders = allOrders.filter(o => o.isTrigger);

        // Group by trigger price
        const triggerLevels = {};
        for (const order of triggerOrders) {
            const px = order.triggerPx;
            if (!triggerLevels[px]) {
                triggerLevels[px] = { count: 0, totalSz: 0, types: [] };
            }
            triggerLevels[px].count++;
            triggerLevels[px].totalSz += parseFloat(order.sz);
            triggerLevels[px].types.push(order.orderType);
        }

        // Report significant trigger clusters
        const sorted = Object.entries(triggerLevels)
            .sort((a, b) => b[1].totalSz - a[1].totalSz);

        console.log(`\n${targetCoin} Trigger Order Clusters:`);
        for (const [px, data] of sorted.slice(0, 5)) {
            console.log(`  ${px}: ${data.count} orders, ` +
                `size ${data.totalSz.toFixed(2)}`);
        }
    }
}

func detectLargeOrders(data []byte, threshold float64) {
    var snapshot [][]interface{}

    if err := json.Unmarshal(data, &snapshot); err != nil {
        return
    }

    for _, market := range snapshot {
        coin := market[0].(string)
        sides := market[1].([]interface{})

        for _, side := range sides {
            orders := side.([]interface{})
            for _, o := range orders {
                order := o.(map[string]interface{})
                sz, _ := strconv.ParseFloat(order["sz"].(string), 64)
                if sz >= threshold {
                    log.Printf("LARGE ORDER: %s %s %.2f @ %s (%s)",
                        coin, order["side"], sz,
                        order["limitPx"], order["orderType"])
                }
            }
        }
    }
}

class RobustOrderbookStreamer {
    constructor(endpoint, apiKey) {
        this.endpoint = endpoint;
        this.apiKey = apiKey;
        this.maxRetries = 5;
        this.retryDelay = 1000;
        this.currentRetries = 0;
    }

    async startStreamWithRetry() {
        while (this.currentRetries < this.maxRetries) {
            try {
                await this.startStream();
                this.currentRetries = 0;
                this.retryDelay = 1000;
            } catch (error) {
                this.currentRetries++;
                console.error(`Stream attempt ${this.currentRetries} failed:`, error.message);

                if (this.currentRetries >= this.maxRetries) {
                    throw new Error('Max retry attempts exceeded');
                }

                // Exponential backoff
                await this.sleep(this.retryDelay);
                this.retryDelay *= 2;
            }
        }
    }

    sleep(ms) {
        return new Promise(resolve => setTimeout(resolve, ms));
    }
}