Market Data Streaming

service MarketDataSubscriptionAPI {
    // Server-side streaming - symbols fixed at subscription time
    rpc CreateMarketDataSubscription(CreateMarketDataSubscriptionRequest)
        returns (stream CreateMarketDataSubscriptionResponse);

    // Bidirectional streaming - dynamically add/remove symbols
    rpc BiDirectionalStreamMarketData(stream BiDirectionalStreamMarketDataRequest)
        returns (stream BiDirectionalStreamMarketDataResponse);
}

from polymarket.v1 import marketdatasubscription_pb2

# Subscribe to specific symbols
request = marketdatasubscription_pb2.CreateMarketDataSubscriptionRequest(
    symbols=["tec-nfl-sbw-2026-02-08-kc", "tec-nfl-sbw-2026-02-08-phi"],
    unaggregated=False,
    depth=10,
    snapshot_only=False
)

# Subscribe to all symbols
request = marketdatasubscription_pb2.CreateMarketDataSubscriptionRequest(
    symbols=[],  # Empty = all symbols
    depth=20
)

if response.HasField('heartbeat'):
    print(f"[{datetime.now().strftime('%H:%M:%S')}] Heartbeat received")

if response.HasField('update'):
    update = response.update
    print(f"Symbol: {update.symbol}")
    print(f"State: {update.state}")
    print(f"Bids: {len(update.bids)}")
    print(f"Offers: {len(update.offers)}")

{
  "symbol": "aec-nfl-buf-kc-2026-01-26",
  "state": "INSTRUMENT_STATE_EXPIRED",
  "stats": {
    "settlement_px": 1000,
    "settlement_set_time": "2026-01-27T03:42:11Z",
    "settlement_preliminary": false,
    "settlement_price_calculation_method": "SETTLEMENT_PRICE_CALCULATION_METHOD_EVENT_TIER_1",
    "settlement_price_calculation_text": "Buffalo Bills win"
  }
}

{
  "symbol": "aec-nfl-buf-kc-2026-01-26",
  "state": "INSTRUMENT_STATE_EXPIRED",
  "stats": {
    "settlement_px": 0,
    "settlement_preliminary": false,
    "settlement_price_calculation_method": "SETTLEMENT_PRICE_CALCULATION_METHOD_EVENT_TIER_1",
    "settlement_price_calculation_text": "Kansas City Chiefs win"
  }
}

from polymarket.v1 import refdata_pb2

# Get state name
state_name = refdata_pb2.InstrumentState.Name(update.state)
print(f"State: {state_name}")

import grpc
import requests
from datetime import datetime, timedelta
from typing import Optional
from polymarket.v1 import marketdatasubscription_pb2
from polymarket.v1 import marketdatasubscription_pb2_grpc
from polymarket.v1 import refdata_pb2


class PolymarketStreamer:
    def __init__(self, base_url: str = "https://rest.preprod.polymarketexchange.com",
                 grpc_server: str = "grpc-api.preprod.polymarketexchange.com:443"):
        self.base_url = base_url
        self.grpc_server = grpc_server
        self.access_token: Optional[str] = None
        self.refresh_token: Optional[str] = None
        self.access_expiration: Optional[datetime] = None
        self.price_scales: dict = {}  # symbol -> price_scale cache

    def get_price_scale(self, symbol: str) -> int:
        """Get price_scale for symbol from cache. Populate via list_instruments API."""
        # WARNING: Replace this with actual API lookup. Do not rely on default.
        return self.price_scales.get(symbol, 1000)

    def login(self, auth0_domain: str, client_id: str, private_key_path: str, audience: str) -> dict:
        """Authenticate using Private Key JWT and store the access token."""
        import jwt
        import uuid
        from cryptography.hazmat.primitives import serialization

        # Load private key
        with open(private_key_path, 'rb') as f:
            private_key = serialization.load_pem_private_key(f.read(), password=None)

        # Create JWT assertion
        now = int(datetime.now().timestamp())
        claims = {
            "iss": client_id,
            "sub": client_id,
            "aud": f"https://{auth0_domain}/oauth/token",
            "iat": now,
            "exp": now + 300,
            "jti": str(uuid.uuid4()),
        }
        assertion = jwt.encode(claims, private_key, algorithm="RS256")

        # Exchange for access token
        response = requests.post(
            f"https://{auth0_domain}/oauth/token",
            json={
                "client_id": client_id,
                "client_assertion_type": "urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
                "client_assertion": assertion,
                "audience": audience,
                "grant_type": "client_credentials"
            }
        )
        response.raise_for_status()

        token_data = response.json()
        self.access_token = token_data["access_token"]
        # Set expiration with 30-second buffer (tokens expire in 180 seconds)
        expires_in = token_data.get("expires_in", 180)
        self.access_expiration = datetime.now() + timedelta(seconds=expires_in - 30)

        return token_data

    def stream_market_data(self, symbols: list, unaggregated: bool = False,
                          depth: int = 10, snapshot_only: bool = False):
        """Stream market data for the given symbols using gRPC."""
        if not self.access_token:
            raise ValueError("Not authenticated. Please login first.")

        # Create credentials
        credentials = grpc.ssl_channel_credentials()

        # Create channel
        channel = grpc.secure_channel(self.grpc_server, credentials)

        # Create stub
        stub = marketdatasubscription_pb2_grpc.MarketDataSubscriptionAPIStub(channel)

        # Create request
        request = marketdatasubscription_pb2.CreateMarketDataSubscriptionRequest(
            symbols=symbols,
            unaggregated=unaggregated,
            depth=depth,
            snapshot_only=snapshot_only
        )

        # Set up metadata with authorization
        metadata = [
            ('authorization', f'Bearer {self.access_token}')
        ]

        try:
            print(f"Starting market data stream for symbols: {symbols}")
            print(f"Parameters: unaggregated={unaggregated}, depth={depth}, snapshot_only={snapshot_only}")
            print("-" * 60)

            # Start streaming
            response_stream = stub.CreateMarketDataSubscription(request, metadata=metadata)

            for response in response_stream:
                self._process_market_data_response(response)

        except grpc.RpcError as e:
            print(f"gRPC error: {e.code()} - {e.details()}")
            raise
        except KeyboardInterrupt:
            print("\nStream interrupted by user")
        finally:
            channel.close()

    def _process_market_data_response(self, response):
        """Process and display market data response."""
        if response.HasField('heartbeat'):
            print(f"[{datetime.now().strftime('%H:%M:%S')}] Heartbeat received")

        elif response.HasField('update'):
            update = response.update
            print(f"\n[{datetime.now().strftime('%H:%M:%S')}] Market Update for {update.symbol}")

            # Display instrument state
            state_name = refdata_pb2.InstrumentState.Name(update.state)
            print(f"  State: {state_name}")

            # Get price_scale from instrument metadata (via list_instruments API)
            price_scale = self.get_price_scale(update.symbol)

            # Display order book
            if update.bids:
                print("  Bids:")
                for i, bid in enumerate(update.bids[:5]):  # Show top 5 bids
                    px = bid.px / price_scale  # Convert from price representation
                    qty = bid.qty
                    print(f"    [{i+1}] ${px:.4f} x {qty}")

            if update.offers:
                print("  Offers:")
                for i, offer in enumerate(update.offers[:5]):  # Show top 5 offers
                    px = offer.px / price_scale # Convert from price representation
                    qty = offer.qty
                    print(f"    [{i+1}] ${px:.4f} x {qty}")

            # Display stats if available
            if update.HasField('stats'):
                stats = update.stats
                print("  Stats:")
                if stats.HasField('last_trade_px'):
                    last_px = stats.last_trade_px / price_scale
                    print(f"    Last Trade: ${last_px:.4f}")
                if stats.HasField('open_px'):
                    open_px = stats.open_px / price_scale
                    print(f"    Open: ${open_px:.4f}")
                if stats.HasField('high_px'):
                    high_px = stats.high_px / price_scale
                    print(f"    High: ${high_px:.4f}")
                if stats.HasField('low_px'):
                    low_px = stats.low_px / price_scale
                    print(f"    Low: ${low_px:.4f}")
                if stats.HasField('shares_traded'):
                    print(f"    Shares Traded: {stats.shares_traded}")
                if stats.HasField('open_interest'):
                    print(f"    Open Interest: {stats.open_interest}")

            print("-" * 60)


# Usage
if __name__ == "__main__":
    streamer = PolymarketStreamer()

    # Login using Private Key JWT
    streamer.login(
        auth0_domain="pmx-preprod.us.auth0.com",
        client_id="your_client_id",
        private_key_path="private_key.pem",
        audience="https://api.preprod.polymarketexchange.com"
    )

    # Stream market data
    streamer.stream_market_data(
        symbols=["tec-nfl-sbw-2026-02-08-kc"],
        depth=10
    )

message BiDirectionalStreamMarketDataRequest {
    oneof command {
        SubscribeCommand subscribe = 1;     // Add symbols
        UnsubscribeCommand unsubscribe = 2; // Remove symbols
        KeepAliveCommand keepalive = 7;     // Application-level keepalive (no-op)
    }
    bool unaggregated = 3;  // Options (read from first request only)
    int32 depth = 4;
    bool snapshot_only = 5;
    bool slow_consumer_skip_to_head = 6;
}

message SubscribeCommand {
    repeated string symbols = 1;
}

message UnsubscribeCommand {
    repeated string symbols = 1;
}

// Application-level keepalive for BiDirectionalStreamMarketData. Sending one
// puts a client-to-server frame on the wire without modifying subscription
// state; the server returns no response.
message KeepAliveCommand {}

message BiDirectionalStreamMarketDataResponse {
    oneof event {
        Heartbeat heartbeat = 1;
        MarketDataUpdate update = 2;
        SubscriptionAck subscription_ack = 3;
        SubscriptionError subscription_error = 4;
    }
}

message SubscriptionAck {
    repeated string symbols_added = 1;    // Symbols added in this operation
    repeated string symbols_removed = 2;  // Symbols removed in this operation
    repeated string active_symbols = 3;   // All currently active symbols
}

message SubscriptionError {
    string error_code = 1;     // e.g., "INVALID_SYMBOL", "ALREADY_SUBSCRIBED"
    string message = 2;        // Human-readable error message
    repeated string symbols = 3; // Symbols that caused the error
}

import grpc
import threading
import queue
from polymarket.v1 import marketdatasubscription_pb2
from polymarket.v1 import marketdatasubscription_pb2_grpc


def request_generator(request_queue):
    """Generator that yields requests from a queue."""
    while True:
        try:
            request = request_queue.get(timeout=0.1)
            if request is None:
                break
            yield request
        except queue.Empty:
            continue


def stream_bidi_market_data(stub, access_token):
    """Demonstrate bidirectional market data streaming."""
    request_queue = queue.Queue()

    # Set up metadata with authorization
    metadata = [('authorization', f'Bearer {access_token}')]

    # Start bidirectional stream
    response_stream = stub.BiDirectionalStreamMarketData(
        request_generator(request_queue),
        metadata=metadata
    )

    # Subscribe to initial symbol
    request_queue.put(marketdatasubscription_pb2.BiDirectionalStreamMarketDataRequest(
        subscribe=marketdatasubscription_pb2.SubscribeCommand(
            symbols=["tec-nfl-sbw-2026-02-08-kc"]
        ),
        depth=10
    ))

    # Process responses
    for response in response_stream:
        if response.HasField('subscription_ack'):
            ack = response.subscription_ack
            print(f"Subscription ACK - Active: {list(ack.active_symbols)}")
        elif response.HasField('subscription_error'):
            err = response.subscription_error
            print(f"Error: {err.error_code} - {err.message}")
        elif response.HasField('update'):
            update = response.update
            print(f"Update: {update.symbol} - {len(update.bids)} bids")
        elif response.HasField('heartbeat'):
            print("Heartbeat")

        # Dynamically add another symbol after receiving first update
        # request_queue.put(marketdatasubscription_pb2.BiDirectionalStreamMarketDataRequest(
        #     subscribe=marketdatasubscription_pb2.SubscribeCommand(
        #         symbols=["tec-nfl-sbw-2026-02-08-phi"]
        #     )
        # ))


def keepalive_worker(request_queue, interval_seconds: int = 1800):
    """Periodically push a KeepAliveCommand onto the request queue (every 30 minutes
    by default) so the connection stays inside the AWS ALB idle window."""
    import time
    while True:
        time.sleep(interval_seconds)
        request_queue.put(marketdatasubscription_pb2.BiDirectionalStreamMarketDataRequest(
            keepalive=marketdatasubscription_pb2.KeepAliveCommand()
        ))

import (
    "context"
    polymarketv1 "github.com/polymarket/client-sample-code/go/gen/polymarket/v1"
)

func streamBidiMarketData(client polymarketv1.MarketDataSubscriptionAPIClient) error {
    ctx := context.Background()
    stream, err := client.BiDirectionalStreamMarketData(ctx)
    if err != nil {
        return err
    }

    // Subscribe to initial symbol
    err = stream.Send(&polymarketv1.BiDirectionalStreamMarketDataRequest{
        Command: &polymarketv1.BiDirectionalStreamMarketDataRequest_Subscribe{
            Subscribe: &polymarketv1.SubscribeCommand{
                Symbols: []string{"tec-nfl-sbw-2026-02-08-kc"},
            },
        },
        Depth: 10,
    })
    if err != nil {
        return err
    }

    // Process responses
    for {
        resp, err := stream.Recv()
        if err != nil {
            return err
        }

        switch {
        case resp.GetSubscriptionAck() != nil:
            ack := resp.GetSubscriptionAck()
            fmt.Printf("ACK - Active: %v\n", ack.ActiveSymbols)
        case resp.GetSubscriptionError() != nil:
            subErr := resp.GetSubscriptionError()
            fmt.Printf("Error: %s - %s\n", subErr.ErrorCode, subErr.Message)
        case resp.GetUpdate() != nil:
            update := resp.GetUpdate()
            fmt.Printf("Update: %s - %d bids\n", update.Symbol, len(update.Bids))

            // Dynamically add another symbol
            // stream.Send(&polymarketv1.BiDirectionalStreamMarketDataRequest{
            //     Command: &polymarketv1.BiDirectionalStreamMarketDataRequest_Subscribe{
            //         Subscribe: &polymarketv1.SubscribeCommand{
            //             Symbols: []string{"tec-nfl-sbw-2026-02-08-phi"},
            //         },
            //     },
            // })
        case resp.GetHeartbeat() != nil:
            fmt.Println("Heartbeat")
        }
    }
}

// Run alongside the receive loop above to keep the stream alive when the
// client is not actively (un)subscribing. Send every 30 minutes (1800s) so
// the connection stays inside the AWS ALB 3600s idle window.
func sendKeepalives(stream polymarketv1.MarketDataSubscriptionAPI_BiDirectionalStreamMarketDataClient) {
    ticker := time.NewTicker(30 * time.Minute)
    defer ticker.Stop()
    for range ticker.C {
        _ = stream.Send(&polymarketv1.BiDirectionalStreamMarketDataRequest{
            Command: &polymarketv1.BiDirectionalStreamMarketDataRequest_Keepalive{
                Keepalive: &polymarketv1.KeepAliveCommand{},
            },
        })
    }
}