Data API v2

.circleci/config.yml

@@ -2,9 +2,8 @@ version: 2
 jobs:
   build:
     docker:
-      - image: circleci/golang:1.10.3
+      - image: circleci/golang:1.14.13
     working_directory: /go/src/github.com/alpacahq/alpaca-trade-api-go
     steps:
       - checkout
-      - run: dep ensure
-      - run: go test -v -cover ./...
+      - run: go test -v -cover ./...

alpaca/entities.go

@@ -3,6 +3,7 @@ package alpaca
 import (
 	"time"
 
+	v2 "github.com/alpacahq/alpaca-trade-api-go/v2"
 	"github.com/shopspring/decimal"
 )
 
@@ -203,6 +204,24 @@ type Aggregates struct {
 	Results      []AggV2 `json:"results"`
 }
 
+type tradeResponse struct {
+	Symbol        string     `json:"symbol"`
+	NextPageToken *string    `json:"next_page_token"`
+	Trades        []v2.Trade `json:"trades"`
+}
+
+type quoteResponse struct {
+	Symbol        string     `json:"symbol"`
+	NextPageToken *string    `json:"next_page_token"`
+	Quotes        []v2.Quote `json:"quotes"`
+}
+
+type barResponse struct {
+	Symbol        string   `json:"symbol"`
+	NextPageToken *string  `json:"next_page_token"`
+	Bars          []v2.Bar `json:"bars"`
+}
+
 type CalendarDay struct {
 	Date  string `json:"date"`
 	Open  string `json:"open"`

alpaca/rest.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	"github.com/alpacahq/alpaca-trade-api-go/common"
+	v2 "github.com/alpacahq/alpaca-trade-api-go/v2"
 )
 
 const (
@@ -66,6 +67,11 @@ func defaultDo(c *Client, req *http.Request) (*http.Response, error) {
 	return resp, nil
 }
 
+const (
+	// v2MaxLimit is the maximum allowed limit parameter for all v2 endpoints
+	v2MaxLimit = 10000
+)
+
 func init() {
 	if s := os.Getenv("APCA_API_BASE_URL"); s != "" {
 		base = s
@@ -404,6 +410,189 @@ func (c *Client) GetLastTrade(symbol string) (*LastTradeResponse, error) {
 	return lastTrade, nil
 }
 
+// GetTrades returns a channel that will be populated with the trades for the given symbol
+// that happened between the given start and end times, limited to the given limit.
+func (c *Client) GetTrades(symbol string, start, end time.Time, limit int) <-chan v2.TradeItem {
+	ch := make(chan v2.TradeItem)
+
+	go func() {
+		defer close(ch)
+
+		u, err := url.Parse(fmt.Sprintf("%s/v2/stocks/%s/trades", dataURL, symbol))
+		if err != nil {
+			ch <- v2.TradeItem{Error: err}
+			return
+		}
+
+		q := u.Query()
+		q.Set("start", start.Format(time.RFC3339))
+		q.Set("end", end.Format(time.RFC3339))
+
+		total := 0
+		pageToken := ""
+		for {
+			actualLimit := limit - total
+			if actualLimit <= 0 {
+				return
+			}
+			if actualLimit > v2MaxLimit {
+				actualLimit = v2MaxLimit
+			}
+			q.Set("limit", fmt.Sprintf("%d", actualLimit))
+			q.Set("page_token", pageToken)
+			u.RawQuery = q.Encode()
+
+			resp, err := c.get(u)
+			if err != nil {
+				ch <- v2.TradeItem{Error: err}
+				return
+			}
+
+			var tradeResp tradeResponse
+			if err = unmarshal(resp, &tradeResp); err != nil {
+				ch <- v2.TradeItem{Error: err}
+				return
+			}
+
+			for _, trade := range tradeResp.Trades {
+				ch <- v2.TradeItem{Trade: trade}
+			}
+			if tradeResp.NextPageToken == nil {
+				return
+			}
+			pageToken = *tradeResp.NextPageToken
+			total += len(tradeResp.Trades)
+		}
+	}()
+
+	return ch
+}
+
+// GetQuotes returns a channel that will be populated with the quotes for the given symbol
+// that happened between the given start and end times, limited to the given limit.
+func (c *Client) GetQuotes(symbol string, start, end time.Time, limit int) <-chan v2.QuoteItem {
+	// NOTE: this method is very similar to GetTrades.
+	// With generics it would be almost trivial to refactor them to use a common base method,
+	// but without them it doesn't seem to be worth it
+	ch := make(chan v2.QuoteItem)
+
+	go func() {
+		defer close(ch)
+
+		u, err := url.Parse(fmt.Sprintf("%s/v2/stocks/%s/quotes", dataURL, symbol))
+		if err != nil {
+			ch <- v2.QuoteItem{Error: err}
+			return
+		}
+
+		q := u.Query()
+		q.Set("start", start.Format(time.RFC3339))
+		q.Set("end", end.Format(time.RFC3339))
+
+		total := 0
+		pageToken := ""
+		for {
+			actualLimit := limit - total
+			if actualLimit <= 0 {
+				return
+			}
+			if actualLimit > v2MaxLimit {
+				actualLimit = v2MaxLimit
+			}
+			q.Set("limit", fmt.Sprintf("%d", actualLimit))
+			q.Set("page_token", pageToken)
+			u.RawQuery = q.Encode()
+
+			resp, err := c.get(u)
+			if err != nil {
+				ch <- v2.QuoteItem{Error: err}
+				return
+			}
+
+			var quoteResp quoteResponse
+			if err = unmarshal(resp, &quoteResp); err != nil {
+				ch <- v2.QuoteItem{Error: err}
+				return
+			}
+
+			for _, quote := range quoteResp.Quotes {
+				ch <- v2.QuoteItem{Quote: quote}
+			}
+			if quoteResp.NextPageToken == nil {
+				return
+			}
+			pageToken = *quoteResp.NextPageToken
+			total += len(quoteResp.Quotes)
+		}
+	}()
+
+	return ch
+}
+
+// GetBars returns a channel that will be populated with the bars for the given symbol
+// between the given start and end times, limited to the given limit,
+// using the given and timeframe and adjustment.
+func (c *Client) GetBars(
+	symbol string, timeFrame v2.TimeFrame, adjustment v2.Adjustment,
+	start, end time.Time, limit int,
+) <-chan v2.BarItem {
+	ch := make(chan v2.BarItem)
+
+	go func() {
+		defer close(ch)
+
+		u, err := url.Parse(fmt.Sprintf("%s/v2/stocks/%s/bars", dataURL, symbol))
+		if err != nil {
+			ch <- v2.BarItem{Error: err}
+			return
+		}
+
+		q := u.Query()
+		q.Set("start", start.Format(time.RFC3339))
+		q.Set("end", end.Format(time.RFC3339))
+		q.Set("adjustment", string(adjustment))
+		q.Set("timeframe", string(timeFrame))
+
+		total := 0
+		pageToken := ""
+		for {
+			actualLimit := limit - total
+			if actualLimit <= 0 {
+				return
+			}
+			if actualLimit > v2MaxLimit {
+				actualLimit = v2MaxLimit
+			}
+			q.Set("limit", fmt.Sprintf("%d", actualLimit))
+			q.Set("page_token", pageToken)
+			u.RawQuery = q.Encode()
+
+			resp, err := c.get(u)
+			if err != nil {
+				ch <- v2.BarItem{Error: err}
+				return
+			}
+
+			var barResp barResponse
+			if err = unmarshal(resp, &barResp); err != nil {
+				ch <- v2.BarItem{Error: err}
+				return
+			}
+
+			for _, bar := range barResp.Bars {
+				ch <- v2.BarItem{Bar: bar}
+			}
+			if barResp.NextPageToken == nil {
+				return
+			}
+			pageToken = *barResp.NextPageToken
+			total += len(barResp.Bars)
+		}
+	}()
+
+	return ch
+}
+
 // CloseAllPositions liquidates all open positions at market price.
 func (c *Client) CloseAllPositions() error {
 	u, err := url.Parse(fmt.Sprintf("%s/%s/positions", base, apiVersion))
@@ -801,6 +990,28 @@ func GetLastTrade(symbol string) (*LastTradeResponse, error) {
 	return DefaultClient.GetLastTrade(symbol)
 }
 
+// GetTrades returns a channel that will be populated with the trades for the given symbol
+// that happened between the given start and end times, limited to the given limit.
+func GetTrades(symbol string, start, end time.Time, limit int) <-chan v2.TradeItem {
+	return DefaultClient.GetTrades(symbol, start, end, limit)
+}
+
+// GetQuotes returns a channel that will be populated with the quotes for the given symbol
+// that happened between the given start and end times, limited to the given limit.
+func GetQuotes(symbol string, start, end time.Time, limit int) <-chan v2.QuoteItem {
+	return DefaultClient.GetQuotes(symbol, start, end, limit)
+}
+
+// GetBars returns a channel that will be populated with the bars for the given symbol
+// between the given start and end times, limited to the given limit,
+// using the given and timeframe and adjustment.
+func GetBars(
+	symbol string, timeFrame v2.TimeFrame, adjustment v2.Adjustment,
+	start, end time.Time, limit int,
+) <-chan v2.BarItem {
+	return DefaultClient.GetBars(symbol, timeFrame, adjustment, start, end, limit)
+}
+
 // GetPosition returns the account's position for the
 // provided symbol using the default Alpaca client.
 func GetPosition(symbol string) (*Position, error) {

examples/v2stream/v2stream.go

@@ -0,0 +1,58 @@
+package main
+
+import (
+	"fmt"
+	"os"
+
+	"github.com/alpacahq/alpaca-trade-api-go/alpaca"
+	"github.com/alpacahq/alpaca-trade-api-go/common"
+	"github.com/alpacahq/alpaca-trade-api-go/v2/stream"
+)
+
+func main() {
+	// You can set your credentials here in the code, or (preferably) via the
+	// APCA_API_KEY_ID and APCA_API_SECRET_KEY environment variables
+	apiKey := "YOUR_API_KEY_HERE"
+	apiSecret := "YOUR_API_SECRET_HERE"
+	if common.Credentials().ID == "" {
+		os.Setenv(common.EnvApiKeyID, apiKey)
+	}
+	if common.Credentials().Secret == "" {
+		os.Setenv(common.EnvApiSecretKey, apiSecret)
+	}
+
+	stream.UseFeed("sip")
+
+	if err := stream.SubscribeTradeUpdates(tradeUpdateHandler); err != nil {
+		panic(err)
+	}
+
+	if err := stream.SubscribeTrades(tradeHandler, "AAPL"); err != nil {
+		panic(err)
+	}
+
+	if err := stream.SubscribeQuotes(quoteHandler, "MSFT"); err != nil {
+		panic(err)
+	}
+	if err := stream.SubscribeBars(barHandler, "IBM"); err != nil {
+		panic(err)
+	}
+
+	select {}
+}
+
+func tradeUpdateHandler(update alpaca.TradeUpdate) {
+	fmt.Println("trade update", update)
+}
+
+func tradeHandler(trade stream.Trade) {
+	fmt.Println("trade", trade)
+}
+
+func quoteHandler(quote stream.Quote) {
+	fmt.Println("quote", quote)
+}
+
+func barHandler(bar stream.Bar) {
+	fmt.Println("bar", bar)
+}

go.mod

@@ -4,12 +4,14 @@ go 1.14
 
 require (
 	github.com/cheekybits/is v0.0.0-20150225183255-68e9c0620927 // indirect
-	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/gorilla/websocket v1.4.0
+	github.com/gorilla/websocket v1.4.1
 	github.com/kr/pretty v0.1.0 // indirect
 	github.com/matryer/try v0.0.0-20161228173917-9ac251b645a2 // indirect
+	github.com/mitchellh/mapstructure v1.4.0
 	github.com/shopspring/decimal v1.2.0
-	github.com/stretchr/testify v1.4.0
+	github.com/stretchr/testify v1.6.1
+	github.com/vmihailenco/msgpack/v5 v5.1.4
 	gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 // indirect
 	gopkg.in/matryer/try.v1 v1.0.0-20150601225556-312d2599e12e
+	nhooyr.io/websocket v1.8.6
 )