NaxosData API & SDK Reference

"""
Event-Driven Dual-Leg Prediction Market Arbitrage Backtester
Using NaxosData Synchronized L2 Feeds & Latency Modeling
"""
from dataclasses import dataclass
import math
from typing import List, Dict, Optional
from naxosdata import NaxosClient

@dataclass
class ExecutionResult:
    timestamp_ms: int
    pair_id: str
    leg1_venue: str
    leg1_filled: float
    leg1_vwap: float
    leg1_slippage_bps: int
    leg2_venue: str
    leg2_filled: float
    leg2_vwap: float
    leg2_slippage_bps: int
    unhedged_shares: float
    unwind_loss_usd: float
    net_payout_usd: float
    total_cost_usd: float
    net_profit_usd: float
    net_roi_pct: float

class EventDrivenArbBacktester:
    def __init__(self, api_key: str, latency_map_ms: Dict[str, float] = None):
        self.client = NaxosClient(api_key=api_key)
        # Empirical wire latency model
        self.latency_map_ms = latency_map_ms or {
            "POLYMARKET": 14.2,  # Polygon RPC commit & block inclusion
            "KALSHI": 8.2,       # Direct REST/WebSocket gateway
            "LIMITLESS": 12.0,   # Base L2 sequencer RPC
            "PREDICTIT": 22.0,   # Web API polling latency
            "DRIFT_BET": 15.5,   # Solana RPC slot commitment
            "SMARKETS": 11.0,    # Smarkets streaming gateway
        }

    def simulate_order_fill(self, asks: List[dict], target_shares: float):
        """Walks down resting L2 depth rungs to simulate fill size, VWAP, and slippage."""
        remaining = target_shares
        total_cost = 0.0
        filled = 0.0
        best_ask = asks[0]["price"] if asks else 0.0

        for rung in asks:
            available = rung["size"]
            price = rung["price"]
            fill_amount = min(remaining, available)
            total_cost += fill_amount * price
            filled += fill_amount
            remaining -= fill_amount
            if remaining <= 0:
                break

        vwap = (total_cost / filled) if filled > 0 else best_ask
        slippage_bps = int(((vwap - best_ask) / best_ask) * 10000) if best_ask > 0 else 0
        return filled, vwap, slippage_bps

    def replay_arbitrage_window(self, pair_id: str, target_usd: float = 2500):
        # 1. Fetch synchronized cross-venue comparison series from NaxosData
        pair = self.client.pairs.comparison(pair_id=pair_id)
        print(f"Loaded Pair: {pair_id} | Fungibility: {pair.fungibility_score}/100")

        # 2. Iterate millisecond tick frames through historical window
        ticks = self.client.orderbooks.timeseries(venue="POLYMARKET", market_id=pair.market_id_long, resolution="100ms")
        results: List[ExecutionResult] = []

        for tick in ticks:
            # Reconstruct exact synchronized L2 depth at tick timestamp
            snap_poly = self.client.orderbooks.snapshot(venue=pair.venue_long, market_id=pair.market_id_long, timestamp_ms=tick.timestamp_ms)
            
            # Kalshi order book at timestamp + network latency offset
            kalshi_ts = tick.timestamp_ms + int(self.latency_map_ms["KALSHI"])
            snap_kalshi = self.client.orderbooks.snapshot(venue=pair.venue_short, market_id=pair.market_id_short, timestamp_ms=kalshi_ts)

            # Leg 1: Sweep YES on Polymarket
            target_shares = target_usd / (snap_poly.best_ask + (1 - snap_kalshi.best_bid))
            leg1_filled, leg1_vwap, leg1_slip = self.simulate_order_fill(snap_poly.asks, target_shares)

            # Leg 2: Sweep NO on Kalshi (reciprocal book: NO asks = 1 - YES bids)
            kalshi_no_asks = [{"price": round(1 - b["price"], 3), "size": b["size"]} for b in snap_kalshi.bids]
            leg2_filled, leg2_vwap, leg2_slip = self.simulate_order_fill(kalshi_no_asks, leg1_filled)

            # Model Unhedged Leg Risk & Emergency Unwind
            unhedged = abs(leg1_filled - leg2_filled)
            unwind_loss = unhedged * 0.045 if unhedged > 0 else 0.0

            # Compute Net-of-Fee Arbitrage P&L with exact Kalshi quadratic taker fee
            kalshi_fee = math.ceil(0.07 * leg2_filled * leg2_vwap * (1.0 - leg2_vwap) * 100.0) / 100.0
            total_cost = (leg1_filled * leg1_vwap) + (leg2_filled * leg2_vwap) + kalshi_fee + unwind_loss
            guaranteed_payout = min(leg1_filled, leg2_filled) * 1.00
            net_profit = guaranteed_payout - total_cost
            net_roi = (net_profit / total_cost) * 100 if total_cost > 0 else 0

            results.append(ExecutionResult(
                timestamp_ms=tick.timestamp_ms,
                pair_id=pair_id,
                leg1_venue=snap_poly.venue,
                leg1_filled=leg1_filled,
                leg1_vwap=leg1_vwap,
                leg1_slippage_bps=leg1_slip,
                leg2_venue=snap_kalshi.venue,
                leg2_filled=leg2_filled,
                leg2_vwap=leg2_vwap,
                leg2_slippage_bps=leg2_slip,
                unhedged_shares=unhedged,
                unwind_loss_usd=unwind_loss,
                net_payout_usd=guaranteed_payout,
                total_cost_usd=total_cost,
                net_profit_usd=net_profit,
                net_roi_pct=net_roi
            ))

        return results

# Replay Fed Cut arbitrage window
sim = EventDrivenArbBacktester(api_key="nx_inst_live")
executions = sim.replay_arbitrage_window(pair_id="PAIR_FED_NOV26_25BPS", target_usd=2500)
profitable = [e for e in executions if e.net_profit_usd > 0]
print(f"Replay complete: {len(profitable)}/{len(executions)} arbitrage executions profitable net of fees & wire latency.")