polymarket-real-time-client-go

Polymarket Real-time Data Client (Go)

High-performance Go client for Polymarket's real-time WebSocket data streams

⚠️ Disclaimer

This is an unofficial client library developed by Discoco Labs. This project is not affiliated with, endorsed by, or sponsored by Polymarket. It is an independent implementation for accessing Polymarket's publicly available WebSocket API.

🚀 Features

• 🎯 Go-Idiomatic API: Pure channel-based design, no callbacks needed
• ⚡ High Performance: Zero-copy channels with configurable buffering
• 🔄 Auto Reconnection: Smart exponential backoff with configurable retry policies
• 🛡️ Type Safety: Strongly typed messages and filters
• 📡 Real-time Streams: Live trading data, prices, and market updates
• 🔧 Production Ready: Built-in error handling, logging, and monitoring
• 🎛️ Flexible Filtering: Type-safe subscription filtering system
• 🔐 Authentication: Support for CLOB API keys and Gamma wallet authentication

📦 Installation

go get github.com/Discoco-Labs/polymarket-real-time-client-go

⚡ Quick Start

package main

import (
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/Discoco-Labs/polymarket-real-time-client-go/client"
)

func main() {
    // ✅ Go-idiomatic: Use channels, not callbacks!
    config := &client.ClientConfig{
        Host:         client.DefaultHost,
        PingInterval: 30 * time.Second,
        ChannelConfig: &client.ChannelConfig{
            MessageBufferSize: 1000,
            ErrorBufferSize:   100,
        },
    }

    rtClient := client.NewRealTimeDataClient(config)
    
    if err := rtClient.Connect(); err != nil {
        log.Fatal("Connection failed:", err)
    }
    defer rtClient.Close()

    // Subscribe to data streams
    subscriptions := []client.Subscription{
        client.CryptoPriceSubscription("ethusdt"),
        client.ActivitySubscription(),
    }
    
    if err := rtClient.Subscribe(subscriptions); err != nil {
        log.Fatal("Subscribe failed:", err)
    }

    // Setup graceful shutdown
    interrupt := make(chan os.Signal, 1)
    signal.Notify(interrupt, os.Interrupt, syscall.SIGTERM)

    fmt.Println("🎯 Starting Go-idiomatic message processor...")

    // ✅ Process messages using Go channels and select
    for {
        select {
        case msg := <-rtClient.Messages():
            // Handle incoming messages
            switch msg.Topic {
            case client.TopicCryptoPrices:
                fmt.Printf("💰 Price: %s\n", string(msg.Payload))
            case client.TopicActivity:
                fmt.Printf("📈 Activity: %s:%s\n", msg.Topic, msg.Type)
            default:
                fmt.Printf("📨 Message: %s:%s\n", msg.Topic, msg.Type)
            }
            
        case err := <-rtClient.Errors():
            // Handle errors
            log.Printf("❌ Error: %v", err)
            
        case status := <-rtClient.ConnectionEvents():
            // Monitor connection status
            fmt.Printf("🔗 Status: %s\n", status)

        case <-interrupt:
            fmt.Println("\n🛑 Shutting down...")
            return
        }
    }
}

📡 Available Data Streams

🔧 Configuration

Configuration Options

config := &client.ClientConfig{
    // Connection settings
    Host:           client.DefaultHost,
    PingInterval:   30 * time.Second,
    ConnectTimeout: 30 * time.Second,
    AutoReconnect:  true,
    
    // Channel configuration (recommended)
    ChannelConfig: &client.ChannelConfig{
        MessageBufferSize:         1000,
        ErrorBufferSize:          100,
        ConnectionEventBufferSize: 10,
        DropMessagesOnFullBuffer:  true,
    },
}

Advanced Configuration

config := &client.ClientConfig{
    Host:         client.DefaultHost,
    PingInterval: 3 * time.Second,
    AutoReconnect: true,
    
    // Reconnection settings
    ReconnectConfig: &client.ReconnectConfig{
        MaxAttempts:       10,
        InitialBackoff:    500 * time.Millisecond,
        MaxBackoff:        30 * time.Second,
        BackoffMultiplier: 2.0,
    },
    
    // Rate limiting
    RateLimitConfig: &client.RateLimitConfig{
        MaxMessagesPerSecond: 1000,
        BurstSize:           100,
        DropBehavior:        "log",
    },
    
    // Logging
    LogLevel:             "info",
    EnableStructuredLogs: true,
    
    // Callbacks
    OnConnect: onConnect,
    OnMessage: onMessage,
    OnStatus:  onStatusChange,
    OnError:   onError,
}

🔐 Authentication

CLOB API Authentication

subscription := client.Subscription{
    Topic: client.TopicCLOBUser,
    Type:  client.MessageTypeOrder,
    		ClobAuth: &client.ClobAPIKeyCreds{
        Key:        "your-api-key",
        Secret:     "your-api-secret",
        Passphrase: "your-passphrase",
    },
}

Gamma Wallet Authentication

subscription := client.Subscription{
    Topic: client.TopicComments,
    Type:  client.MessageTypeAll,
    GammaAuth: &client.GammaAuth{
        Address: "your-wallet-address",
    },
}

🎯 Type-Safe Message Filtering

Quick Start with Convenience Functions

// Crypto price filtering
btcSub := client.CryptoPriceSubscription("btcusdt")
ethSub := client.CryptoPriceSubscription("ethusdt")

// Activity filtering by event
f1Sub := client.ActivitySubscriptionForEvent("will-george-russell-win-the-2025-f1-hungarian-grand-prix")

// CLOB market filtering by asset IDs
clobSub := client.CLOBMarketSubscriptionForAssets(
    "109681959945973300464568698402968596289258214226684818748321941747028805721376",
    "23580794846556055407327394522946009947656180095466036451628554845007074532052",
)

// Authenticated CLOB user data
userSub := client.CLOBUserSubscription("api-key", "api-secret", "passphrase")

Builder Pattern for Complex Filters

// Build custom subscriptions with fluent API
subscription := client.NewSubscription(client.TopicActivity, client.MessageTypeAll).
    WithEventSlugFilter("presidential-election-2024").
    Build()

// Comments with entity filtering
commentSub := client.NewSubscription(client.TopicComments, client.MessageTypeAll).
    WithParentEntityFilter(12345, "Event").
    Build()

// CLOB market with multiple assets
marketSub := client.NewSubscription(client.TopicCLOBMarket, client.MessageTypeAll).
    WithAssetIDsFilter("asset-id-1", "asset-id-2", "asset-id-3").
    Build()

Available Filter Types

Legacy JSON Filter Support

// Still supported for backward compatibility
subscription := client.Subscription{
    Topic:   client.TopicCryptoPrices,
    Type:    client.MessageTypeAll,
    Filters: `{"symbol":"ethusdt"}`,  // Manual JSON string
}

📊 Type-Safe Message Processing

func onMessage(c client.RealTimeDataClient, msg *client.Message) {
    switch msg.Topic {
    case client.TopicActivity:
        if msg.Type == client.MessageTypeTrades {
            var trade client.TradePayload
            if err := json.Unmarshal(msg.Payload, &trade); err == nil {
                fmt.Printf("Trade: %s %d shares at $%.4f\n", 
                    trade.Side, trade.Size, trade.Price)
            }
        }
        
    case client.TopicCryptoPrices:
        var price client.CryptoPricePayload
        if err := json.Unmarshal(msg.Payload, &price); err == nil {
            fmt.Printf("Price Update: %s = $%.4f\n", 
                price.Symbol, price.Value)
        }
        
    case client.TopicRFQ:
        switch msg.Type {
        case client.MessageTypeRequestCreated:
            var req client.RequestPayload
            if err := json.Unmarshal(msg.Payload, &req); err == nil {
                fmt.Printf("New RFQ: %s %.2f->%.2f at $%.4f\n",
                    req.Side, req.SizeIn, req.SizeOut, req.Price)
            }
        }
    }
}

🚦 Error Handling

func onError(err error) {
    if clientErr, ok := err.(*client.ClientError); ok {
        switch clientErr.Type {
        case client.ErrorTypeConnection:
            fmt.Printf("Connection error: %s\n", clientErr.Message)
            // Handle connection issues
            
        case client.ErrorTypeRateLimit:
            fmt.Printf("Rate limit exceeded: %s\n", clientErr.Message)
            // Reduce subscription load
            
        case client.ErrorTypeAuthentication:
            fmt.Printf("Auth error: %s\n", clientErr.Message)
            // Check credentials
            
        case client.ErrorTypeMessageParsing:
            fmt.Printf("Message parsing error: %s\n", clientErr.Message)
            // Handle malformed messages
        }
    }
}

📈 Performance Optimization

Connection Pooling

// Use a single client instance for multiple subscriptions
client := client.NewRealTimeDataClient(config)

// Subscribe to multiple streams efficiently
subscriptions := []client.Subscription{
    {Topic: client.TopicActivity, Type: client.MessageTypeTrades},
    {Topic: client.TopicCryptoPrices, Type: client.MessageTypePriceUpdate},
    {Topic: client.TopicCLOBMarket, Type: client.MessageTypeAggOrderbook},
}
client.Subscribe(subscriptions)

Rate Limiting

rateLimitConfig := &client.RateLimitConfig{
    MaxMessagesPerSecond: 500,  // Limit to 500 msgs/sec
    BurstSize:           50,    // Allow bursts of 50
    DropBehavior:        "log", // Log dropped messages
}

Memory Management

config := &client.ClientConfig{
    MessageBufferSize: 2000,  // Larger buffer for high-volume streams
    LogLevel:         "warn", // Reduce log verbosity in production
}

🏃‍♂️ Examples

The repository includes comprehensive examples:

Basic Example

go run examples/basic/main.go

Advanced Example (Multi-stream with Authentication)

go run examples/advanced/main.go

Optimized Example (Production Configuration)

go run examples/optimized/main.go

Type-Safe Filters Example

go run examples/filters/main.go

🧪 Testing

# Run all tests
go test ./...

# Run with coverage
go test -cover ./...

# Run benchmarks
go test -bench=. ./...

# Run specific test package
go test ./client

🔧 Development

Build

go build ./...

Lint

golangci-lint run

Format

go fmt ./...

📊 Monitoring

Structured Logging

config := &client.ClientConfig{
    LogLevel:             "info",
    EnableStructuredLogs: true,  // JSON formatted logs
}

Custom Metrics

func onMessage(c client.RealTimeDataClient, msg *client.Message) {
    // Custom metrics collection
    messageCounter.WithLabelValues(string(msg.Topic), string(msg.Type)).Inc()
    lastMessageTime.Set(float64(time.Now().Unix()))
}

🤝 Contributing

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Make your changes
4. Add tests for new functionality
5. Run tests (go test ./...)
6. Commit your changes (git commit -m 'Add amazing feature')
7. Push to the branch (git push origin feature/amazing-feature)
8. Open a Pull Request

Development Guidelines

• Follow Go conventions and best practices
• Add comprehensive tests for new features
• Update documentation for API changes
• Use structured logging for debugging
• Ensure backward compatibility

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

🔗 Links

• Polymarket API Documentation
• CLOB API Reference
• WebSocket API Specification

💬 Support

• Documentation: Polymarket Docs
• Issues: GitHub Issues
• Discord: Polymarket Community

Built with ❤️ by Discoco Labs for high-frequency trading and real-time market data# Test commit for hook validation