rtds

Documentation

Index

• Constants
• Variables
• type BaseEvent
• type ChainlinkPriceEvent
• type Client
  • func NewClient(url string) (Client, error)
  • func NewClientWithConfig(url string, cfg ClientConfig) (Client, error)
• type ClientConfig
  • func ClientConfigFromEnv() ClientConfig
  • func DefaultClientConfig() ClientConfig
• type ClobAuth
• type CommentEvent
• type CommentFilter
• type CommentProfile
• type CommentType
• type ConnectionState
• type ConnectionStateEvent
• type CryptoPriceEvent
• type EventType
• type LaggedError
  • func (e LaggedError) Error() string
• type OrdersMatchedEvent
• type RtdsMessage
• type Stream
  • func (s *Stream[T]) Close() error
• type Subscription
  • func (s Subscription) MarshalJSON() ([]byte, error)
• type SubscriptionAction
• type SubscriptionRequest

Constants

const (
	ProdURL = "wss://ws-live-data.polymarket.com"
)

Variables

var (
	ErrInvalidSubscription = sdkerrors.ErrInvalidSubscription
)

Use unified error definitions from pkg/errors

Types

type BaseEvent

type BaseEvent struct {
	Topic            EventType `json:"-"`
	MessageType      string    `json:"-"`
	MessageTimestamp int64     `json:"-"`
}

BaseEvent carries message metadata.

type ChainlinkPriceEvent

type ChainlinkPriceEvent struct {
	BaseEvent
	Symbol    string        `json:"symbol"`
	Timestamp int64         `json:"timestamp"`
	Value     types.Decimal `json:"value"`
}

ChainlinkPriceEvent is a Chainlink price update payload.

type Client

type Client interface {
	// Authenticate sets default CLOB credentials for authenticated comment streams.
	Authenticate(apiKey *auth.APIKey) Client
	// Deauthenticate clears any stored credentials for authenticated streams.
	Deauthenticate() Client

	SubscribeCryptoPricesStream(ctx context.Context, symbols []string) (*Stream[CryptoPriceEvent], error)
	SubscribeChainlinkPricesStream(ctx context.Context, feeds []string) (*Stream[ChainlinkPriceEvent], error)
	SubscribeCommentsStream(ctx context.Context, req *CommentFilter) (*Stream[CommentEvent], error)
	SubscribeOrdersMatchedStream(ctx context.Context) (*Stream[OrdersMatchedEvent], error)
	SubscribeRawStream(ctx context.Context, sub *Subscription) (*Stream[RtdsMessage], error)
	SubscribeCryptoPrices(ctx context.Context, symbols []string) (<-chan CryptoPriceEvent, error)
	SubscribeChainlinkPrices(ctx context.Context, feeds []string) (<-chan ChainlinkPriceEvent, error)
	SubscribeComments(ctx context.Context, req *CommentFilter) (<-chan CommentEvent, error)
	SubscribeOrdersMatched(ctx context.Context) (<-chan OrdersMatchedEvent, error)
	SubscribeRaw(ctx context.Context, sub *Subscription) (<-chan RtdsMessage, error)
	UnsubscribeCryptoPrices(ctx context.Context) error
	UnsubscribeChainlinkPrices(ctx context.Context) error
	UnsubscribeComments(ctx context.Context, commentType *CommentType) error
	UnsubscribeOrdersMatched(ctx context.Context) error
	UnsubscribeRaw(ctx context.Context, sub *Subscription) error
	ConnectionState() ConnectionState
	ConnectionStateStream(ctx context.Context) (*Stream[ConnectionStateEvent], error)
	SubscriptionCount() int
	Close() error
}

Client defines the RTDS WebSocket interface.

func NewClient

func NewClient(url string) (Client, error)

func NewClientWithConfig

func NewClientWithConfig(url string, cfg ClientConfig) (Client, error)

NewClientWithConfig creates an RTDS client with explicit reconnect/heartbeat settings.

type ClientConfig

type ClientConfig struct {
	Reconnect      bool
	ReconnectDelay time.Duration
	ReconnectMax   int
	PingInterval   time.Duration
}

ClientConfig controls RTDS WebSocket reconnect and heartbeat behavior.

func ClientConfigFromEnv

func ClientConfigFromEnv() ClientConfig

ClientConfigFromEnv keeps backward compatibility with the old env-driven initialization behavior.

func DefaultClientConfig

func DefaultClientConfig() ClientConfig

DefaultClientConfig returns deterministic defaults without reading environment variables.

type ClobAuth

type ClobAuth struct {
	Key        string `json:"key"`
	Secret     string `json:"secret"`
	Passphrase string `json:"passphrase"`
}

ClobAuth carries CLOB credentials for authenticated comment streams.

type CommentEvent

type CommentEvent struct {
	BaseEvent
	ID               string         `json:"id"`
	Body             string         `json:"body"`
	CreatedAt        time.Time      `json:"createdAt"`
	ParentCommentID  *string        `json:"parentCommentID,omitempty"`
	ParentEntityID   int64          `json:"parentEntityID"`
	ParentEntityType string         `json:"parentEntityType"`
	Profile          CommentProfile `json:"profile"`
	ReactionCount    int64          `json:"reactionCount,omitempty"`
	ReplyAddress     *types.Address `json:"replyAddress,omitempty"`
	ReportCount      int64          `json:"reportCount,omitempty"`
	UserAddress      types.Address  `json:"userAddress"`
}

CommentEvent is a comment stream payload.

type CommentFilter

type CommentFilter struct {
	Type    *CommentType
	Auth    *auth.APIKey
	Filters interface{}
}

CommentFilter configures the comments subscription.

type CommentProfile

type CommentProfile struct {
	BaseAddress           types.Address  `json:"baseAddress"`
	DisplayUsernamePublic bool           `json:"displayUsernamePublic,omitempty"`
	Name                  string         `json:"name"`
	ProxyWallet           *types.Address `json:"proxyWallet,omitempty"`
	Pseudonym             *string        `json:"pseudonym,omitempty"`
}

CommentProfile describes the comment author.

type CommentType

type CommentType string

CommentType enumerates comment event types.

const (
	CommentCreated  CommentType = "comment_created"
	CommentRemoved  CommentType = "comment_removed"
	ReactionCreated CommentType = "reaction_created"
	ReactionRemoved CommentType = "reaction_removed"
)

type ConnectionState

type ConnectionState string

ConnectionState represents RTDS connection status.

const (
	ConnectionDisconnected ConnectionState = "disconnected"
	ConnectionConnected    ConnectionState = "connected"
)

type ConnectionStateEvent

type ConnectionStateEvent struct {
	State    ConnectionState `json:"state"`
	Recorded int64           `json:"recorded"`
}

ConnectionStateEvent captures connection transitions.

type CryptoPriceEvent

type CryptoPriceEvent struct {
	BaseEvent
	Symbol    string        `json:"symbol"`
	Timestamp int64         `json:"timestamp"`
	Value     types.Decimal `json:"value"`
}

CryptoPriceEvent is a Binance price update payload.

type EventType

type EventType string

EventType represents RTDS topic categories.

const (
	CryptoPrice    EventType = "crypto_prices"
	ChainlinkPrice EventType = "crypto_prices_chainlink"
	Comments       EventType = "comments"
	Activity       EventType = "activity"
)

type LaggedError

type LaggedError struct {
	Count   int
	Topic   string
	MsgType string
}

LaggedError indicates the subscriber missed messages due to backpressure.

func (LaggedError) Error

func (e LaggedError) Error() string

type OrdersMatchedEvent

type OrdersMatchedEvent struct {
	BaseEvent
	Asset           string  `json:"asset"`
	Bio             string  `json:"bio"`
	ConditionID     string  `json:"conditionId"`
	EventSlug       string  `json:"eventSlug"`
	Icon            string  `json:"icon"`
	Name            string  `json:"name"`
	Outcome         string  `json:"outcome"`
	OutcomeIndex    int     `json:"outcomeIndex"`
	Price           float64 `json:"price"`
	ProfileImage    string  `json:"profileImage"`
	ProxyWallet     string  `json:"proxyWallet"`
	Pseudonym       string  `json:"pseudonym"`
	Side            string  `json:"side"`
	Size            float64 `json:"size"`
	Slug            string  `json:"slug"`
	Timestamp       int64   `json:"timestamp"`
	Title           string  `json:"title"`
	TransactionHash string  `json:"transactionHash"`
}

OrdersMatchedEvent is an activity stream payload for matched orders.

type RtdsMessage

type RtdsMessage struct {
	Topic     string          `json:"topic"`
	MsgType   string          `json:"type"`
	Timestamp int64           `json:"timestamp"`
	Payload   json.RawMessage `json:"payload"`
}

RtdsMessage is the raw RTDS message wrapper.

type Stream

type Stream[T any] struct {
	C   <-chan T
	Err <-chan error
	// contains filtered or unexported fields
}

Stream delivers messages and async errors for a subscription.

func (*Stream[T]) Close

func (s *Stream[T]) Close() error

Close stops the subscription and closes the stream.

type Subscription

type Subscription struct {
	Topic    string      `json:"topic"`
	MsgType  string      `json:"type"`
	Filters  interface{} `json:"filters,omitempty"`
	ClobAuth *ClobAuth   `json:"clob_auth,omitempty"`
}

Subscription describes a single RTDS subscription.

func (Subscription) MarshalJSON

func (s Subscription) MarshalJSON() ([]byte, error)

MarshalJSON customizes filters encoding to align with RTDS expectations. For non-Chainlink topics, JSON strings are parsed and sent as raw JSON. Chainlink filters are always serialized as a JSON string.

type SubscriptionAction

type SubscriptionAction string

SubscriptionAction indicates subscribe/unsubscribe.

const (
	SubscribeAction   SubscriptionAction = "subscribe"
	UnsubscribeAction SubscriptionAction = "unsubscribe"
)

type SubscriptionRequest

type SubscriptionRequest struct {
	Action        SubscriptionAction `json:"action"`
	Subscriptions []Subscription     `json:"subscriptions"`
}

SubscriptionRequest is the top-level RTDS subscribe/unsubscribe payload.