Coverage for src/finbot/market/alpaca.py: 100%

1import logging 

2from datetime import datetime 

3from typing import Any, Literal 

4 

5import requests 

6 

7from finbot.market.cache import HistoricalDataCache 

8from finbot.market.data_provider import MarketDataProvider 

9from finbot.market.dataset import DatasetMetadata 

10from finbot.market.models import Bar 

11 

12logger = logging.getLogger(__name__) 

13 

14 

15class AlpacaMarketDataProvider(MarketDataProvider): 

16 BASE_URL = "https://data.alpaca.markets" 

17 

18 def __init__( 

19 self, 

20 api_key: str, 

21 api_secret: str, 

22 feed: str = "iex", 

23 cache: HistoricalDataCache | None = None, 

24 ): 

25 self.feed = feed 

26 self.cache = cache 

27 

28 self.session = requests.Session() 

29 self.session.headers.update({ 

30 "accept": "application/json", 

31 "APCA-API-KEY-ID": api_key, 

32 "APCA-API-SECRET-KEY": api_secret, 

33 }) 

34 

35 def get_latest_bar(self, symbol: str) -> Bar: 

36 logger.debug("Requesting latest Alpaca bar for %s using feed %s", symbol, self.feed) 

37 response = self.session.get( 

38 f"{self.BASE_URL}/v2/stocks/{symbol}/bars/latest", 

39 params={"feed": self.feed}, 

40 timeout=10, 

41 ) 

42 response.raise_for_status() 

43 

44 data = response.json()["bar"] 

45 return self._parse_bar(symbol, data) 

46 

47 def get_bars( 

48 self, 

49 symbols: list[str], 

50 start: datetime, 

51 end: datetime, 

52 timeframe: str = "1Min", 

53 adjustment: Literal["raw", "split", "dividend", "spin-off", "all"] = "all", 

54 ) -> dict[str, list[Bar]]: 

55 metadata = DatasetMetadata( 

56 source="alpaca", 

57 symbols=tuple(symbols), 

58 timeframe=timeframe, 

59 start=start, 

60 end=end, 

61 adjustment=adjustment, 

62 source_options=(("feed", self.feed),), 

63 ) 

64 logger.debug( 

65 "Historical data request: dataset=%s symbols=%s " 

66 "timeframe=%s start=%s end=%s adjustment=%s feed=%s", 

67 metadata.dataset_id, 

68 ",".join(metadata.symbols), 

69 metadata.timeframe, 

70 metadata.start.isoformat(), 

71 metadata.end.isoformat(), 

72 metadata.adjustment, 

73 self.feed, 

74 ) 

75 

76 if self.cache is not None: 

77 if self.cache.contains(metadata.dataset_id): 

78 logger.info("Using cached historical dataset %s", metadata.dataset_id) 

79 _, bars = self.cache.load(metadata.dataset_id) 

80 

81 logger.debug( 

82 "Loaded %d cached bars for dataset %s", 

83 sum(len(items) for items in bars.values()), metadata.dataset_id, 

84 ) 

85 return bars 

86 

87 logger.info( 

88 "Historical dataset %s not cached; downloading from Alpaca", metadata.dataset_id, 

89 ) 

90 else: 

91 logger.debug("Historical cache disabled; downloading data from Alpaca") 

92 

93 normalized_symbols = list(metadata.symbols) 

94 result: dict[str, list[Bar]] = {symbol: [] for symbol in normalized_symbols} 

95 

96 page_token = None 

97 page_number = 0 

98 

99 logger.debug( 

100 "Requesting Alpaca historical bars: symbols=%s " 

101 "timeframe=%s start=%s end=%s adjustment=%s feed=%s", 

102 ",".join(normalized_symbols), 

103 timeframe, 

104 start.isoformat(), 

105 end.isoformat(), 

106 adjustment, 

107 self.feed, 

108 ) 

109 

110 while True: 

111 page_number += 1 

112 params: dict[str, Any] = { 

113 "symbols": ",".join(normalized_symbols), 

114 "timeframe": timeframe, 

115 "start": start.isoformat(), 

116 "end": end.isoformat(), 

117 "limit": 10_000, 

118 "adjustment": adjustment, 

119 "feed": self.feed, 

120 "sort": "asc", 

121 } 

122 

123 if page_token: 

124 params["page_token"] = page_token 

125 

126 response = self.session.get( 

127 f"{self.BASE_URL}/v2/stocks/bars", 

128 params=params, 

129 timeout=30, 

130 ) 

131 response.raise_for_status() 

132 

133 payload: dict = response.json() 

134 

135 page_bar_count = sum(len(items) for items in payload["bars"].values()) 

136 logger.debug( 

137 "Received Alpaca historical page %d: %d bars", 

138 page_number, 

139 page_bar_count, 

140 ) 

141 

142 for symbol, bars in payload["bars"].items(): 

143 result[symbol].extend(self._parse_bar(symbol, bar) for bar in bars) 

144 

145 page_token = payload.get("next_page_token") 

146 if not page_token: break 

147 

148 total_bars = sum(len(items) for items in result.values()) 

149 logger.info("Downloaded %d historical bars from Alpaca", total_bars) 

150 

151 if self.cache is not None: 

152 self.cache.save(metadata, result) 

153 logger.info("Cached historical dataset %s", metadata.dataset_id) 

154 

155 return result 

156 

157 @staticmethod 

158 def _parse_bar(symbol: str, data: dict) -> Bar: 

159 return Bar( 

160 symbol=symbol, 

161 timestamp=datetime.fromisoformat( 

162 data["t"].replace("Z", "+00:00"), 

163 ), 

164 open=float(data["o"]), 

165 high=float(data["h"]), 

166 low=float(data["l"]), 

167 close=float(data["c"]), 

168 volume=float(data["v"]), 

169 )