feed.go raw

   1  package market
   2  
   3  import (
   4  	"context"
   5  	"encoding/json"
   6  	"fmt"
   7  	"strconv"
   8  	"time"
   9  
  10  	"github.com/coder/websocket"
  11  )
  12  
  13  // Feed connects to an exchange WebSocket API and streams order book
  14  // updates as OrderBook snapshots. Each exchange has a different wire
  15  // format; the Feed normalizes them into the common OrderBook type.
  16  type Feed struct {
  17  	URL    string
  18  	Venue  Venue
  19  	Asset  string
  20  	conn   *websocket.Conn
  21  	Books  chan *OrderBook
  22  	Errors chan error
  23  }
  24  
  25  // NewFeed creates a feed for a specific asset on a specific venue.
  26  func NewFeed(url string, venue Venue, asset string) *Feed {
  27  	return &Feed{
  28  		URL:    url,
  29  		Venue:  venue,
  30  		Asset:  asset,
  31  		Books:  make(chan *OrderBook, 64),
  32  		Errors: make(chan error, 8),
  33  	}
  34  }
  35  
  36  // Connect establishes the WebSocket connection and sends the
  37  // subscription message.
  38  func (f *Feed) Connect(ctx context.Context) error {
  39  	conn, _, err := websocket.Dial(ctx, f.URL, nil)
  40  	if err != nil {
  41  		return fmt.Errorf("dial %s: %w", f.URL, err)
  42  	}
  43  	conn.SetReadLimit(1 << 20) // 1MB
  44  	f.conn = conn
  45  	return nil
  46  }
  47  
  48  // Listen reads messages from the exchange and parses them into
  49  // OrderBook snapshots. Call this in a goroutine.
  50  func (f *Feed) Listen(ctx context.Context) {
  51  	defer close(f.Books)
  52  	for {
  53  		_, data, err := f.conn.Read(ctx)
  54  		if err != nil {
  55  			select {
  56  			case f.Errors <- err:
  57  			default:
  58  			}
  59  			return
  60  		}
  61  
  62  		ob, err := f.parseMessage(data)
  63  		if err != nil {
  64  			continue // skip unparseable messages
  65  		}
  66  		if ob != nil {
  67  			select {
  68  			case f.Books <- ob:
  69  			default: // drop if consumer is slow
  70  			}
  71  		}
  72  	}
  73  }
  74  
  75  // Close disconnects the feed.
  76  func (f *Feed) Close() {
  77  	if f.conn != nil {
  78  		f.conn.Close(websocket.StatusNormalClosure, "done")
  79  	}
  80  }
  81  
  82  // parseMessage tries to parse an exchange message into an OrderBook.
  83  // Supports a generic JSON format that covers multiple exchanges.
  84  func (f *Feed) parseMessage(data []byte) (*OrderBook, error) {
  85  	// Try generic order book format: {"bids": [[price, qty], ...], "asks": [[price, qty], ...]}
  86  	var msg struct {
  87  		Bids [][]json.Number `json:"bids"`
  88  		Asks [][]json.Number `json:"asks"`
  89  	}
  90  	if err := json.Unmarshal(data, &msg); err != nil {
  91  		return nil, err
  92  	}
  93  
  94  	if len(msg.Bids) == 0 && len(msg.Asks) == 0 {
  95  		return nil, nil // not an order book message
  96  	}
  97  
  98  	ob := &OrderBook{
  99  		Asset: f.Asset,
 100  		Venue: f.Venue,
 101  		Time:  time.Now(),
 102  	}
 103  
 104  	for _, level := range msg.Bids {
 105  		if len(level) < 2 {
 106  			continue
 107  		}
 108  		price, _ := strconv.ParseFloat(string(level[0]), 64)
 109  		volume, _ := strconv.ParseFloat(string(level[1]), 64)
 110  		if price > 0 {
 111  			ob.Bids = append(ob.Bids, OrderEntry{
 112  				Price:  price,
 113  				Volume: volume,
 114  				Side:   Bid,
 115  				Venue:  f.Venue,
 116  				Time:   ob.Time,
 117  			})
 118  		}
 119  	}
 120  
 121  	for _, level := range msg.Asks {
 122  		if len(level) < 2 {
 123  			continue
 124  		}
 125  		price, _ := strconv.ParseFloat(string(level[0]), 64)
 126  		volume, _ := strconv.ParseFloat(string(level[1]), 64)
 127  		if price > 0 {
 128  			ob.Asks = append(ob.Asks, OrderEntry{
 129  				Price:  price,
 130  				Volume: volume,
 131  				Side:   Ask,
 132  				Venue:  f.Venue,
 133  				Time:   ob.Time,
 134  			})
 135  		}
 136  	}
 137  
 138  	return ob, nil
 139  }
 140