stream

Documentation

Index

• type DurationUpdate
• type JupiterAPIStream
  • func (j *JupiterAPIStream) FetchPrice(ctx context.Context) (jr JupiterPriceResponse, err error)
  • func (j *JupiterAPIStream) FetchQuote(ctx context.Context) (jr JupiterPriceResponse, err error)
  • func (j *JupiterAPIStream) Name() string
  • func (j *JupiterAPIStream) Process(updates []RawUpdate[DurationUpdate[*JupiterPriceResponse]], ...) (results map[int][]ProcessedUpdate[QuoteResult], ...)
  • func (j *JupiterAPIStream) Run(parent context.Context) ([]RawUpdate[DurationUpdate[*JupiterPriceResponse]], error)
• type JupiterFee
• type JupiterMarketInfo
• type JupiterOpt
  • func WithJupiterAmount(amount float64) JupiterOpt
  • func WithJupiterInterval(interval time.Duration) JupiterOpt
  • func WithJupiterSlippage(slippage int64) JupiterOpt
  • func WithJupiterTicker(t *time.Ticker) JupiterOpt
  • func WithJupiterToken(mint string, decimals int) JupiterOpt
• type JupiterPriceInfo
• type JupiterPriceResponse
  • func (jr JupiterPriceResponse) Price(mint string) float64
• type JupiterQuoteResponse
• type JupiterRoute
• type ProcessedUpdate
• type QuoteResult
• type RawUpdate
  • func NewDurationUpdate[T any](start time.Time, data T) RawUpdate[DurationUpdate[T]]
  • func NewRawUpdate[T any](data T) RawUpdate[T]
• type SolanaRawUpdate
• type SolanaUpdate
  • func (s SolanaUpdate) IsRedundant() bool
• type Source
  • func NewAPIOrderbookStream(address, market, authHeader string) (Source[[]byte, TraderAPIUpdate], error)
  • func NewJupiterAPI(opts ...JupiterOpt) (Source[DurationUpdate[*JupiterPriceResponse], QuoteResult], error)
  • func NewSolanaOrderbookStream(ctx context.Context, rpcAddress string, wsAddress, marketAddr string) (Source[SolanaRawUpdate, SolanaUpdate], error)
  • func NewTraderHTTPPriceStream(opts ...TraderHTTPPriceOpt) (Source[DurationUpdate[*pb.GetPriceResponse], QuoteResult], error)
  • func NewTraderWSPrice(opts ...TraderWSPriceOpt) (Source[*pb.GetPricesStreamResponse, QuoteResult], error)
• type TraderAPIUpdate
  • func (s TraderAPIUpdate) IsRedundant() bool
• type TraderHTTPPriceOpt
  • func WithTraderHTTPClient(h *provider.HTTPClient) TraderHTTPPriceOpt
  • func WithTraderHTTPMint(m string) TraderHTTPPriceOpt
  • func WithTraderHTTPTicker(t *time.Ticker) TraderHTTPPriceOpt
• type TraderWSPriceOpt
  • func WithTraderWSClient(w *provider.WSClient) TraderWSPriceOpt
  • func WithTraderWSMint(m string) TraderWSPriceOpt

Types

type DurationUpdate

type DurationUpdate[T any] struct {
	Start time.Time
	Data  T
}

type JupiterAPIStream

type JupiterAPIStream struct {
	// contains filtered or unexported fields
}

func (*JupiterAPIStream) FetchPrice

func (j *JupiterAPIStream) FetchPrice(ctx context.Context) (jr JupiterPriceResponse, err error)

FetchPrice returns a price based off of swapping 0.0000001 USDC (the minimum possible unit). Trader API does 1 USDC.

func (*JupiterAPIStream) FetchQuote

func (j *JupiterAPIStream) FetchQuote(ctx context.Context) (jr JupiterPriceResponse, err error)

FetchQuote is used to specify 1 USDC instead of 0.000001

func (*JupiterAPIStream) Name

func (j *JupiterAPIStream) Name() string

func (*JupiterAPIStream) Process

func (j *JupiterAPIStream) Process(updates []RawUpdate[DurationUpdate[*JupiterPriceResponse]], removeDuplicates bool) (results map[int][]ProcessedUpdate[QuoteResult], duplicates map[int][]ProcessedUpdate[QuoteResult], err error)

func (*JupiterAPIStream) Run

func (j *JupiterAPIStream) Run(parent context.Context) ([]RawUpdate[DurationUpdate[*JupiterPriceResponse]], error)

type JupiterFee

type JupiterFee struct {
	Amount string  `json:"amount"`
	Mint   string  `json:"mint"`
	Pct    float64 `json:"pct"`
}

type JupiterMarketInfo

type JupiterMarketInfo struct {
	ID                 string     `json:"id"`
	Label              string     `json:"label"`
	InputMint          string     `json:"inputMint"`
	OutputMint         string     `json:"outputMint"`
	NotEnoughLiquidity bool       `json:"notEnoughLiquidity"`
	InAmount           string     `json:"inAmount"`
	OutAmount          string     `json:"outAmount"`
	PriceImpactPct     float64    `json:"priceImpactPct"`
	LpFee              JupiterFee `json:"lpFee"`
	PlatformFee        JupiterFee `json:"platformFee"`
}

type JupiterOpt

type JupiterOpt func(s *JupiterAPIStream)

func WithJupiterAmount

func WithJupiterAmount(amount float64) JupiterOpt

func WithJupiterInterval

func WithJupiterInterval(interval time.Duration) JupiterOpt

func WithJupiterSlippage

func WithJupiterSlippage(slippage int64) JupiterOpt

func WithJupiterTicker

func WithJupiterTicker(t *time.Ticker) JupiterOpt

func WithJupiterToken

func WithJupiterToken(mint string, decimals int) JupiterOpt

type JupiterPriceInfo

type JupiterPriceInfo struct {
	ID            string  `json:"id"`
	MintSymbol    string  `json:"mintSymbol"`
	VsToken       string  `json:"vsToken"`
	VsTokenSymbol string  `json:"vsTokenSymbol"`
	Price         float64 `json:"price"`
}

type JupiterPriceResponse

type JupiterPriceResponse struct {
	PriceInfo   map[string]JupiterPriceInfo `json:"data"`
	TimeTaken   float64                     `json:"timeTaken"`
	ContextSlot int                         `json:"contextSlot"` // might be removed from the API endpoint now
}

func (JupiterPriceResponse) Price

func (jr JupiterPriceResponse) Price(mint string) float64

type JupiterQuoteResponse

type JupiterQuoteResponse struct {
	Routes      []JupiterRoute `json:"data"`
	TimeTaken   float64        `json:"timeTaken"`
	ContextSlot uint64         `json:"contextSlot"`
}

type JupiterRoute

type JupiterRoute struct {
	InAmount             string              `json:"inAmount"`
	OutAmount            string              `json:"outAmount"`
	PriceImpactPct       float64             `json:"priceImpactPct"`
	MarketInfos          []JupiterMarketInfo `json:"marketInfos"`
	Amount               string              `json:"amount"`
	SlippageBps          int                 `json:"slippageBps"`
	OtherAmountThreshold string              `json:"otherAmountThreshold"`
	SwapMode             string              `json:"swapMode"`
}

type ProcessedUpdate

type ProcessedUpdate[T any] struct {
	Timestamp time.Time
	Slot      int
	Data      T
}

type QuoteResult

type QuoteResult struct {
	Elapsed   time.Duration
	BuyPrice  float64
	SellPrice float64
	Source    string
}

type RawUpdate

type RawUpdate[T any] struct {
	Timestamp time.Time
	Data      T
}

func NewDurationUpdate

func NewDurationUpdate[T any](start time.Time, data T) RawUpdate[DurationUpdate[T]]

func NewRawUpdate

func NewRawUpdate[T any](data T) RawUpdate[T]

type SolanaRawUpdate

type SolanaRawUpdate struct {
	Data *solanaws.AccountResult
	Side gserum.Side
}

type SolanaUpdate

type SolanaUpdate struct {
	Side   gserum.Side
	Orders []*pb.OrderbookItem
	// contains filtered or unexported fields
}

func (SolanaUpdate) IsRedundant

func (s SolanaUpdate) IsRedundant() bool

type Source

type Source[T any, R any] interface {
	// Name returns an identifier for the source for printing
	Name() string

	// Run collects stream updates for the context duration. Run should avoid doing any other work besides collecting updates to have accurate timestamps.
	Run(context.Context) ([]RawUpdate[T], error)

	// Process deserializes the messages received by Run into useful formats for comparison.
	Process(updates []RawUpdate[T], removeDuplicates bool) (results map[int][]ProcessedUpdate[R], duplicates map[int][]ProcessedUpdate[R], err error)
}

Source represents any streaming interface that provides timestamped updates for comparison.

func NewAPIOrderbookStream

func NewAPIOrderbookStream(address, market, authHeader string) (Source[[]byte, TraderAPIUpdate], error)

func NewJupiterAPI

func NewJupiterAPI(opts ...JupiterOpt) (Source[DurationUpdate[*JupiterPriceResponse], QuoteResult], error)

func NewSolanaOrderbookStream

func NewSolanaOrderbookStream(ctx context.Context, rpcAddress string, wsAddress, marketAddr string) (Source[SolanaRawUpdate, SolanaUpdate], error)

func NewTraderHTTPPriceStream

func NewTraderHTTPPriceStream(opts ...TraderHTTPPriceOpt) (Source[DurationUpdate[*pb.GetPriceResponse], QuoteResult], error)

func NewTraderWSPrice

func NewTraderWSPrice(opts ...TraderWSPriceOpt) (Source[*pb.GetPricesStreamResponse, QuoteResult], error)

type TraderAPIUpdate

type TraderAPIUpdate struct {
	Asks []*pb.OrderbookItem
	Bids []*pb.OrderbookItem
	// contains filtered or unexported fields
}

func (TraderAPIUpdate) IsRedundant

func (s TraderAPIUpdate) IsRedundant() bool

type TraderHTTPPriceOpt

type TraderHTTPPriceOpt func(s *traderHTTPPriceStream)

func WithTraderHTTPClient

func WithTraderHTTPClient(h *provider.HTTPClient) TraderHTTPPriceOpt

func WithTraderHTTPMint

func WithTraderHTTPMint(m string) TraderHTTPPriceOpt

func WithTraderHTTPTicker

func WithTraderHTTPTicker(t *time.Ticker) TraderHTTPPriceOpt

type TraderWSPriceOpt

type TraderWSPriceOpt func(s *tradeWSPrice)

func WithTraderWSClient

func WithTraderWSClient(w *provider.WSClient) TraderWSPriceOpt

func WithTraderWSMint

func WithTraderWSMint(m string) TraderWSPriceOpt