StreamFills - Real-time Fill Streaming

{
  "local_time": "2025-07-27T08:50:10.334741319",   // ISO-8601 timestamp when node processed fill (nanosecond precision)
  "block_time": "2025-07-27T08:50:10.273720809",   // ISO-8601 timestamp from block consensus
  "block_number": 676607012,                        // Block height containing this fill
  "events": [                                       // Array of [address, fill_data] pairs
    [
      "0x7839e2f2c375dd2935193f2736167514efff9916",  // User address (40 hex chars, lowercase)
      {
        "coin": "BTC",                              // Trading pair symbol
        "px": "118136.0",                           // Fill price (string)
        "sz": "0.00009",                            // Fill size (string)
        "side": "B",                                // "B" (buy) or "A" (sell/ask)
        "time": 1753606210273,                      // Fill timestamp (ms since Unix epoch)
        "startPosition": "-1.41864",                // Position size before fill (string)
        "dir": "Close Short",                       // Direction: "Open Long" | "Open Short" | "Close Long" | "Close Short"
        "closedPnl": "-0.003753",                   // Realized PnL from closing position (string, can be negative)
        "hash": "0xe7822040155eaa2e737e042854342401120052bbf063906ce8c8f3babe853a79",  // Transaction hash (64 hex)
        "oid": 121670079265,                        // Order ID (numeric)
        "crossed": false,                           // Whether order crossed the spread
        "fee": "-0.000212",                         // Trading fee (string, negative = rebate)
        "tid": 161270588369408,                     // Trade ID (unique identifier)
        "cloid": "0x09367b9f8541c581f95b02aaf05f1508",  // Client order ID (optional, 32 hex)
        "feeToken": "USDC",                         // Token used for fee payment
        "twapId": null,                             // TWAP order ID (optional, null if not a TWAP fill)
        "liquidation": null,                        // Liquidation details (optional, null for normal fills)
        "builder": "0x49ae63056b3a0be0b166813ee687309ab653c07c",  // Builder address (optional)
        "builderFee": "0.005528"                    // Builder fee amount (optional, string)
      }
    ]
    // ... more [address, fill_data] pairs
  ]
}

{
  "block_number": 859864054,
  "block_time": "2026-04-16T14:22:31.456000Z",
  "events": [
    [
      "0x7a3b1c9d2e4f5a6b7c8d9e0f1a2b3c4d5e6f7a8b",
      {
        "coin": "BTC",
        "px": "103850.0",
        "sz": "0.0234",
        "side": "A",
        "time": 1744813351456,
        "startPosition": "0.0234",
        "dir": "Close Long",
        "closedPnl": "-31.599",
        "hash": "0xab12cd34ef56ab12cd34ef56ab12cd34ef56ab12cd34ef56ab12cd34ef56ab12",
        "oid": 199445678901,
        "crossed": true,
        "fee": "0.073107",
        "tid": 884916789012345,
        "cloid": null,
        "feeToken": "USDC",
        "twapId": null,
        "liquidation": {
          "liquidatedUser": "0x7a3b1c9d2e4f5a6b7c8d9e0f1a2b3c4d5e6f7a8b",
          "markPx": "103842.5",
          "method": "market"
        }
      }
    ],
    [
      "0x1234abcd5678ef901234abcd5678ef901234abcd",
      {
        "coin": "BTC",
        "px": "103850.0",
        "sz": "0.0234",
        "side": "B",
        "time": 1744813351456,
        "startPosition": "-0.125",
        "dir": "Close Short",
        "closedPnl": "2.457",
        "hash": "0xab12cd34ef56ab12cd34ef56ab12cd34ef56ab12cd34ef56ab12cd34ef56ab12",
        "oid": 199445678902,
        "crossed": false,
        "fee": "-0.007311",
        "tid": 884916789012345,
        "cloid": "0x00000000000003849201784629103847",
        "feeToken": "USDC",
        "twapId": null,
        "liquidation": {
          "liquidatedUser": "0x7a3b1c9d2e4f5a6b7c8d9e0f1a2b3c4d5e6f7a8b",
          "markPx": "103842.5",
          "method": "market"
        }
      }
    ]
  ]
}

class TradeExecutionTracker {
    constructor(streamManager) {
        this.streamManager = streamManager;
        this.executedTrades = new Map();

        streamManager.on('fill', (fillData) => {
            this.processFill(fillData);
        });
    }

    processFill(fillData) {
        // Track each fill by order ID
        const orderId = fillData.oid;

        if (!this.executedTrades.has(orderId)) {
            this.executedTrades.set(orderId, {
                orderId,
                fills: [],
                totalFilled: 0,
                avgPrice: 0
            });
        }

        const trade = this.executedTrades.get(orderId);
        trade.fills.push(fillData);
        trade.totalFilled += parseFloat(fillData.sz);

        // Recalculate average price
        const totalValue = trade.fills.reduce(
            (sum, f) => sum + parseFloat(f.px) * parseFloat(f.sz), 0
        );
        trade.avgPrice = totalValue / trade.totalFilled;

        console.log(`Order ${orderId}: Filled ${trade.totalFilled} @ avg ${trade.avgPrice}`);
    }
}

class FillRateMonitor:
    def __init__(self):
        self.fills_per_minute = []
        self.current_minute_fills = 0
        self.last_minute = None

    def record_fill(self, fill_data):
        """Record a fill and track rates"""
        from datetime import datetime

        current_minute = datetime.now().replace(second=0, microsecond=0)

        if self.last_minute != current_minute:
            if self.last_minute is not None:
                self.fills_per_minute.append({
                    'minute': self.last_minute,
                    'count': self.current_minute_fills
                })
            self.last_minute = current_minute
            self.current_minute_fills = 0

        self.current_minute_fills += 1

    def get_average_rate(self, minutes=5):
        """Get average fills per minute over last N minutes"""
        recent = self.fills_per_minute[-minutes:]
        if not recent:
            return 0
        return sum(r['count'] for r in recent) / len(recent)

type PositionReconciler struct {
    client pb.HyperliquidL1GatewayClient
    positions map[string]float64
}

func (pr *PositionReconciler) ReconcileFills(ctx context.Context) {
    stream, err := pr.client.StreamFills(ctx, &pb.Position{})
    if err != nil {
        log.Fatal(err)
    }

    for {
        fill, err := stream.Recv()
        if err != nil {
            log.Printf("Stream error: %v", err)
            return
        }

        pr.updatePosition(fill.Data)
    }
}

func (pr *PositionReconciler) updatePosition(data []byte) {
    var fillData map[string]interface{}
    json.Unmarshal(data, &fillData)

    // Extract fill details and update position
    if coin, ok := fillData["coin"].(string); ok {
        size, _ := fillData["sz"].(float64)
        side, _ := fillData["side"].(string)

        if side == "B" {
            pr.positions[coin] += size
        } else {
            pr.positions[coin] -= size
        }

        log.Printf("Position %s: %f", coin, pr.positions[coin])
    }
}

class RobustFillStreamer {
    constructor(endpoint) {
        this.endpoint = endpoint;
        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));
    }
}