alpaca.go raw
1 package market
2
3 import (
4 "context"
5 "fmt"
6 "sync"
7 "time"
8
9 "github.com/alpacahq/alpaca-trade-api-go/v3/marketdata"
10 "github.com/alpacahq/alpaca-trade-api-go/v3/marketdata/stream"
11 )
12
13 // AlpacaClient wraps the Alpaca market data API for ingesting OHLCV bars.
14 // It handles both historical (REST) and real-time (WebSocket) bar data,
15 // converting everything into the organism's native Bar type.
16 type AlpacaClient struct {
17 mu sync.Mutex
18
19 dataClient *marketdata.Client
20 streamClient *stream.CryptoClient
21 Symbols []string // e.g., ["BTC/USD", "ETH/USD"]
22 Bars chan *Bar
23 Errors chan error
24 connected bool
25 }
26
27 // AlpacaConfig holds the configuration for connecting to Alpaca.
28 type AlpacaConfig struct {
29 APIKey string // APCA_API_KEY_ID
30 APISecret string // APCA_API_SECRET_KEY
31 Symbols []string
32 // Paper controls whether trading goes to paper-api or live api.
33 // Market data always goes to data.alpaca.markets regardless.
34 Paper bool
35 }
36
37 // NewAlpacaClient creates a client connected to Alpaca's market data API.
38 func NewAlpacaClient(cfg AlpacaConfig) *AlpacaClient {
39 opts := marketdata.ClientOpts{
40 APIKey: cfg.APIKey,
41 APISecret: cfg.APISecret,
42 }
43
44 return &AlpacaClient{
45 dataClient: marketdata.NewClient(opts),
46 Symbols: cfg.Symbols,
47 Bars: make(chan *Bar, 256),
48 Errors: make(chan error, 16),
49 }
50 }
51
52 // GetHistoricalBars fetches OHLCV bars for a symbol between start and end.
53 // TimeFrame controls the bar resolution (1 minute, 1 hour, 1 day, etc.).
54 func (ac *AlpacaClient) GetHistoricalBars(
55 symbol string,
56 tf marketdata.TimeFrame,
57 start, end time.Time,
58 ) ([]*Bar, error) {
59 req := marketdata.GetCryptoBarsRequest{
60 TimeFrame: tf,
61 Start: start,
62 End: end,
63 }
64
65 cryptoBars, err := ac.dataClient.GetCryptoBars(symbol, req)
66 if err != nil {
67 return nil, fmt.Errorf("get crypto bars %s: %w", symbol, err)
68 }
69
70 bars := make([]*Bar, len(cryptoBars))
71 for i, cb := range cryptoBars {
72 bars[i] = &Bar{
73 Symbol: symbol,
74 Open: cb.Open,
75 High: cb.High,
76 Low: cb.Low,
77 Close: cb.Close,
78 Volume: cb.Volume,
79 VWAP: cb.VWAP,
80 TradeCount: cb.TradeCount,
81 Timestamp: cb.Timestamp,
82 }
83 }
84 return bars, nil
85 }
86
87 // GetMultiHistoricalBars fetches bars for multiple symbols at once.
88 func (ac *AlpacaClient) GetMultiHistoricalBars(
89 tf marketdata.TimeFrame,
90 start, end time.Time,
91 ) (map[string][]*Bar, error) {
92 req := marketdata.GetCryptoBarsRequest{
93 TimeFrame: tf,
94 Start: start,
95 End: end,
96 }
97
98 cryptoBars, err := ac.dataClient.GetCryptoMultiBars(ac.Symbols, req)
99 if err != nil {
100 return nil, fmt.Errorf("get crypto multi bars: %w", err)
101 }
102
103 result := make(map[string][]*Bar, len(cryptoBars))
104 for sym, cbs := range cryptoBars {
105 bars := make([]*Bar, len(cbs))
106 for i, cb := range cbs {
107 bars[i] = &Bar{
108 Symbol: sym,
109 Open: cb.Open,
110 High: cb.High,
111 Low: cb.Low,
112 Close: cb.Close,
113 Volume: cb.Volume,
114 VWAP: cb.VWAP,
115 TradeCount: cb.TradeCount,
116 Timestamp: cb.Timestamp,
117 }
118 }
119 result[sym] = bars
120 }
121 return result, nil
122 }
123
124 // StreamBars connects to Alpaca's WebSocket and streams real-time bars
125 // for all configured symbols. Bars are sent to the Bars channel.
126 // Call this in a goroutine. Cancel the context to stop streaming.
127 func (ac *AlpacaClient) StreamBars(ctx context.Context, apiKey, apiSecret string) error {
128 ac.mu.Lock()
129 if ac.connected {
130 ac.mu.Unlock()
131 return fmt.Errorf("already streaming")
132 }
133 ac.mu.Unlock()
134
135 cryptoClient := stream.NewCryptoClient(
136 marketdata.US,
137 stream.WithCredentials(apiKey, apiSecret),
138 )
139
140 if err := cryptoClient.Connect(ctx); err != nil {
141 return fmt.Errorf("stream connect: %w", err)
142 }
143
144 ac.mu.Lock()
145 ac.streamClient = cryptoClient
146 ac.connected = true
147 ac.mu.Unlock()
148
149 err := cryptoClient.SubscribeToBars(func(cb stream.CryptoBar) {
150 bar := &Bar{
151 Symbol: cb.Symbol,
152 Open: cb.Open,
153 High: cb.High,
154 Low: cb.Low,
155 Close: cb.Close,
156 Volume: cb.Volume,
157 VWAP: cb.VWAP,
158 TradeCount: cb.TradeCount,
159 Timestamp: cb.Timestamp,
160 }
161 select {
162 case ac.Bars <- bar:
163 default: // drop if consumer is slow
164 }
165 }, ac.Symbols...)
166 if err != nil {
167 return fmt.Errorf("subscribe bars: %w", err)
168 }
169
170 // Wait for termination or context cancellation.
171 select {
172 case err := <-cryptoClient.Terminated():
173 ac.mu.Lock()
174 ac.connected = false
175 ac.mu.Unlock()
176 if err != nil {
177 return fmt.Errorf("stream terminated: %w", err)
178 }
179 return nil
180 case <-ctx.Done():
181 ac.mu.Lock()
182 ac.connected = false
183 ac.mu.Unlock()
184 return ctx.Err()
185 }
186 }
187
188 // IsConnected returns whether the streaming connection is active.
189 func (ac *AlpacaClient) IsConnected() bool {
190 ac.mu.Lock()
191 defer ac.mu.Unlock()
192 return ac.connected
193 }
194