func (c *client) GetTradesAsync(symbol string, params GetTradesParams) <-chan TradeItem {
	ch := make(chan TradeItem)
	go func() {
		defer close(ch)
		u, err := url.Parse(fmt.Sprintf("%s/v2/stocks/%s/trades", c.opts.BaseURL, symbol))
		if err != nil {
			ch <- TradeItem{Error: err}
			return
		}
		q := u.Query()
		setBaseQuery(q, params.Start, params.End, params.Feed, c.opts.Feed)
		received := 0
		for params.TotalLimit == 0 || received < params.TotalLimit {
			setQueryLimit(q, params.TotalLimit, params.PageLimit, received)
			u.RawQuery = q.Encode()
			resp, err := c.get(u)
			if err != nil {
				ch <- TradeItem{Error: err}
				return
			}
			var tradeResp tradeResponse
			if err = unmarshal(resp, &tradeResp); err != nil {
				ch <- TradeItem{Error: err}
				return
			}
			for _, trade := range tradeResp.Trades {
				ch <- TradeItem{Trade: trade}
			}
			if tradeResp.NextPageToken == nil {
				return
			}
			q.Set("page_token", *tradeResp.NextPageToken)
			received += len(tradeResp.Trades)
		}
	}()
	return ch
}
GetTradesAsync returns a channel that will be populated with the historical trades for the given symbol.
package marketdata
import (
	"compress/gzip"
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"net/http"
	"net/url"
	"os"
	"sort"
	"strings"
	"time"
)
type Client interface {
	GetTrades(symbol string, params GetTradesParams) ([]Trade, error)
	GetTradesAsync(symbol string, params GetTradesParams) <-chan TradeItem
	GetMultiTrades(symbols []string, params GetTradesParams) (map[string][]Trade, error)
	GetMultiTradesAsync(symbols []string, params GetTradesParams) <-chan MultiTradeItem
	GetQuotes(symbol string, params GetQuotesParams) ([]Quote, error)
	GetQuotesAsync(symbol string, params GetQuotesParams) <-chan QuoteItem
	GetMultiQuotes(symbols []string, params GetQuotesParams) (map[string][]Quote, error)
	GetMultiQuotesAsync(symbols []string, params GetQuotesParams) <-chan MultiQuoteItem
	GetBars(symbol string, params GetBarsParams) ([]Bar, error)
	GetBarsAsync(symbol string, params GetBarsParams) <-chan BarItem
	GetMultiBars(symbols []string, params GetBarsParams) (map[string][]Bar, error)
	GetMultiBarsAsync(symbols []string, params GetBarsParams) <-chan MultiBarItem
	GetLatestBar(symbol string) (*Bar, error)
	GetLatestBars(symbols []string) (map[string]Bar, error)
	GetLatestTrade(symbol string) (*Trade, error)
	GetLatestTrades(symbols []string) (map[string]Trade, error)
	GetLatestQuote(symbol string) (*Quote, error)
	GetLatestQuotes(symbols []string) (map[string]Quote, error)
	GetSnapshot(symbol string) (*Snapshot, error)
	GetSnapshots(symbols []string) (map[string]*Snapshot, error)
	GetCryptoTrades(symbol string, params GetCryptoTradesParams) ([]CryptoTrade, error)
	GetCryptoTradesAsync(symbol string, params GetCryptoTradesParams) <-chan CryptoTradeItem
	GetCryptoQuotes(symbol string, params GetCryptoQuotesParams) ([]CryptoQuote, error)
	GetCryptoQuotesAsync(symbol string, params GetCryptoQuotesParams) <-chan CryptoQuoteItem
	GetCryptoBars(symbol string, params GetCryptoBarsParams) ([]CryptoBar, error)
	GetCryptoBarsAsync(symbol string, params GetCryptoBarsParams) <-chan CryptoBarItem
	GetCryptoMultiBars(symbols []string, params GetCryptoBarsParams) (map[string][]CryptoBar, error)
	GetCryptoMultiBarsAsync(symbols []string, params GetCryptoBarsParams) <-chan CryptoMultiBarItem
	GetLatestCryptoTrade(symbol, exchange string) (*CryptoTrade, error)
	GetLatestCryptoQuote(symbol, exchange string) (*CryptoQuote, error)
	GetLatestCryptoXBBO(symbol string, exchanges []string) (*CryptoXBBO, error)
}
type ClientOpts struct {
	ApiKey     string
	ApiSecret  string
	OAuth      string
	BaseURL    string
	Timeout    time.Duration
	RetryLimit int
	RetryDelay time.Duration
	Feed       string
}
type client struct {
	opts ClientOpts
	do func(c *client, req *http.Request) (*http.Response, error)
}
func NewClient(opts ClientOpts) Client {
	if opts.ApiKey == "" {
		opts.ApiKey = os.Getenv("APCA_API_KEY_ID")
	}
	if opts.ApiSecret == "" {
		opts.ApiSecret = os.Getenv("APCA_API_SECRET_KEY")
	}
	if opts.OAuth == "" {
		opts.OAuth = os.Getenv("APCA_API_OAUTH")
	}
	if opts.BaseURL == "" {
		if s := os.Getenv("APCA_API_DATA_URL"); s != "" {
			opts.BaseURL = s
		} else {
			opts.BaseURL = "https://data.alpaca.markets"
		}
	}
	if opts.RetryLimit == 0 {
		opts.RetryLimit = 10
	}
	if opts.RetryDelay == 0 {
		opts.RetryDelay = time.Second
	}
	return &client{
		opts: opts,
		do: defaultDo,
	}
}
var DefaultClient = NewClient(ClientOpts{})
func defaultDo(c *client, req *http.Request) (*http.Response, error) {
	if c.opts.OAuth != "" {
		req.Header.Set("Authorization", "Bearer "+c.opts.OAuth)
	} else {
		req.Header.Set("APCA-API-KEY-ID", c.opts.ApiKey)
		req.Header.Set("APCA-API-SECRET-KEY", c.opts.ApiSecret)
	}
	client := &http.Client{
		Timeout: c.opts.Timeout,
	}
	var resp *http.Response
	var err error
	for i := 0; ; i++ {
		resp, err = client.Do(req)
		if err != nil {
			return nil, err
		}
		if resp.StatusCode != http.StatusTooManyRequests {
			break
		}
		if i >= c.opts.RetryLimit {
			break
		}
		time.Sleep(c.opts.RetryDelay)
	}
	if err = verify(resp); err != nil {
		return nil, err
	}
	return resp, nil
}
func setBaseQuery(q url.Values, start, end time.Time, feed, defaultFeed string) {
	if !start.IsZero() {
		q.Set("start", start.Format(time.RFC3339))
	}
	if !end.IsZero() {
		q.Set("end", end.Format(time.RFC3339))
	}
	if feed != "" {
		q.Set("feed", feed)
	} else {
		if defaultFeed != "" {
			q.Set("feed", feed)
		}
	}
}
func setCryptoBaseQuery(q url.Values, start, end time.Time, exchanges []string) {
	if !start.IsZero() {
		q.Set("start", start.Format(time.RFC3339))
	}
	if !end.IsZero() {
		q.Set("end", end.Format(time.RFC3339))
	}
	if len(exchanges) > 0 {
		q.Set("exchanges", strings.Join(exchanges, ","))
	}
}
const v2MaxLimit = 10000
func setQueryLimit(q url.Values, totalLimit int, pageLimit int, received int) {
	limit := 0  
	if pageLimit != 0 {
		limit = pageLimit
	}
	if totalLimit != 0 {
		remaining := totalLimit - received
		if remaining <= 0 {  
			return
		}
		if (limit == 0 || limit > remaining) && remaining <= v2MaxLimit {
			limit = remaining
		}
	}
	if limit != 0 {
		q.Set("limit", fmt.Sprintf("%d", limit))
	}
}
type GetTradesParams struct {
	Start time.Time
	End time.Time
	TotalLimit int
	PageLimit int
	Feed string
}
func (c *client) GetTrades(symbol string, params GetTradesParams) ([]Trade, error) {
	trades := make([]Trade, 0)
	for item := range c.GetTradesAsync(symbol, params) {
		if err := item.Error; err != nil {
			return nil, err
		}
		trades = append(trades, item.Trade)
	}
	return trades, nil
}