Coverage for src/finbot/execution/alpaca.py: 80%

1import logging

2import math

3import random

4import time

5from datetime import datetime

6from typing import Any

8import requests

10from finbot.execution.errors import (

11 BrokerAPIError,

12 BrokerConnectionError,

13 BrokerResponseError,

14 OrderConflictError,

15 TradingDisabledError,

16)

17from finbot.execution.models import BrokerPosition, Fill, Order, OrderSide, OrderState, OrderStatus

18from finbot.execution.state import DuplicateClientOrderError, ExecutionStateStore, PersistedOrder

19from finbot.logger import get_logger, log_event

20from finbot.portfolio.portfolio import Portfolio, Position

22_PAPER_BASE_URL = "https://paper-api.alpaca.markets"

24_ALPACA_ORDER_STATUS_MAP = {

25 "accepted": OrderStatus.OPEN,

26 "pending_new": OrderStatus.OPEN,

27 "new": OrderStatus.OPEN,

28 "accepted_for_bidding": OrderStatus.OPEN,

29 "pending_cancel": OrderStatus.OPEN,

30 "pending_replace": OrderStatus.OPEN,

31 "stopped": OrderStatus.OPEN,

32 "suspended": OrderStatus.OPEN,

33 "held": OrderStatus.OPEN,

34 "done_for_day": OrderStatus.OPEN,

35 "calculated": OrderStatus.OPEN,

36 "partially_filled": OrderStatus.PARTIALLY_FILLED,

37 "filled": OrderStatus.FILLED,

38 "canceled": OrderStatus.CANCELED,

39 "expired": OrderStatus.EXPIRED,

40 "rejected": OrderStatus.REJECTED,

41 "replaced": OrderStatus.CANCELED,

42}

44_RETRYABLE_SERVER_STATUS_CODES = frozenset({500, 502, 503, 504})

46_MAX_RETRIES = 3

47_INITIAL_BACKOFF_SECONDS = 0.5

48_MAX_BACKOFF_SECONDS = 8.0

49_RETRY_JITTER_SECONDS = 0.25

51logger = get_logger(__name__)

54class AlpacaPaperBroker:

55 """Broker adapter restricted to Alpaca paper trading"""

57 def __init__(

58 self,

59 api_key: str,

60 api_secret: str,

61 *,

62 trading_enabled: bool = False,

63 dry_run: bool = False,

64 state_store: ExecutionStateStore | None = None,

65 session: requests.Session | None = None,

66 timeout: float = 10.0,

67 ):

68 if not isinstance(api_key, str) or not api_key.strip():

69 raise ValueError("API key must be a non-empty string.")

70 if not isinstance(api_secret, str) or not api_secret.strip():

71 raise ValueError("API secret must be a non-empty string.")

72 if not math.isfinite(timeout) or timeout <= 0:

73 raise ValueError("Timeout must be greater than zero.")

75 self._api_key = api_key

76 self._api_secret = api_secret

77 self._trading_enabled = trading_enabled

78 self._dry_run = dry_run

79 self._state_store = state_store

80 self._session = session or requests.Session()

81 self._timeout = timeout

82 self._next_dry_run_order_id = 1

83 self._dry_run_orders: dict[str, OrderState] = {}

85 @property

86 def trading_enabled(self):

87 return self._trading_enabled

89 @property

90 def dry_run(self) -> bool:

91 return self._dry_run

93 def _get_state_store(self) -> ExecutionStateStore:

94 if self._state_store is None: 94 ↛ 95line 94 didn't jump to line 95 because the condition on line 94 was never true

95 self._state_store = ExecutionStateStore()

97 return self._state_store

99 @staticmethod

100 def _map_order_response(payload: dict[str, Any]) -> OrderState:

101 try:

102 order_id = payload["id"]

103 symbol = payload["symbol"]

104 side = OrderSide(payload["side"])

105 quantity = float(payload["qty"])

106 raw_status = payload["status"]

107 except (KeyError, TypeError, ValueError) as exc:

108 raise BrokerResponseError(

109 "Invalid order response from Alpaca paper trading API.",

110 ) from exc

111

112 try:

113 status = _ALPACA_ORDER_STATUS_MAP[raw_status]

114 except KeyError as exc:

115 raise BrokerResponseError(f"Unsupported Alpaca order status: {raw_status!r}.") from exc

116

117 client_order_id = payload.get("client_order_id")

118

119 try:

120 order = Order(

121 symbol=symbol,

122 side=side,

123 quantity=quantity,

124 client_order_id=client_order_id,

125 )

126 return OrderState(order_id=order_id, order=order, status=status)

127 except (TypeError, ValueError) as exc:

128 raise BrokerResponseError(

129 "Invalid order response from Alpaca paper trading API.",

130 ) from exc

131

132 def _request(

133 self,

134 method: str,

135 path: str,

136 *,

137 json: dict[str, Any] | None = None,

138 params: dict[str, Any] | None = None,

139 retryable: bool | None = None,

140 ) -> requests.Response:

141 method = method.upper()

142

143 if retryable is None: 143 ↛ 146line 143 didn't jump to line 146 because the condition on line 143 was always true

144 retryable = method == "GET"

145

146 url = f"{_PAPER_BASE_URL}/{path.lstrip("/")}"

147 request_kwargs: dict[str, Any] = {

148 "headers": {

149 "APCA-API-KEY-ID": self._api_key,

150 "APCA-API-SECRET-KEY": self._api_secret,

151 },

152 "json": json,

153 "timeout": self._timeout,

154 }

155

156 if params is not None:

157 request_kwargs["params"] = params

158

159 for attempt in range(_MAX_RETRIES + 1): 159 ↛ 238line 159 didn't jump to line 238 because the loop on line 159 didn't complete

160 try:

161 response = self._session.request(method, url, **request_kwargs)

162 except requests.RequestException as exc:

163 if not retryable or attempt >= _MAX_RETRIES:

164 log_event(

165 logger,

166 logging.ERROR,

167 "broker.request_failed",

168 method=method,

169 path=path,

170 reason="connection_error",

171 attempts=attempt + 1,

172 )

173 raise BrokerConnectionError(

174 "Failed to connect to Alpaca paper trading API.",

175 ) from exc

176

177 delay = self._retry_delay(attempt)

178

179 log_event(

180 logger,

181 logging.WARNING,

182 "broker.request_retry",

183 method=method,

184 path=path,

185 reason="connection_error",

186 retry_attempt=attempt + 1,

187 max_retries=_MAX_RETRIES,

188 delay_seconds=delay,

189 )

190

191 time.sleep(delay)

192 continue

193

194 if response.status_code < 400:

195 return response

196

197 rate_limited = response.status_code == 429

198 transient_server_error = (

199 retryable

200 and response.status_code in _RETRYABLE_SERVER_STATUS_CODES

201 )

202

203 if (rate_limited or transient_server_error) and attempt < _MAX_RETRIES:

204 delay = self._retry_delay(attempt, response)

205

206 log_event(

207 logger,

208 logging.WARNING,

209 "broker.request_retry",

210 method=method,

211 path=path,

212 reason="rate_limited" if rate_limited else "server_error",

213 status_code=response.status_code,

214 retry_attempt=attempt + 1,

215 max_retries=_MAX_RETRIES,

216 delay_seconds=delay,

217 )

218

219 time.sleep(delay)

220 continue

221

222 log_event(

223 logger,

224 logging.ERROR,

225 "broker.request_failed",

226 method=method,

227 path=path,

228 reason="rate_limited" if response.status_code == 429 else "api_error",

229 status_code=response.status_code,

230 attempts=attempt + 1,

231 )

232

233 raise BrokerAPIError(

234 self._response_error_message(response),

235 status_code=response.status_code,

236 )

237

238 raise AssertionError("Unreachable")

239

240 @staticmethod

241 def _response_error_message(response: requests.Response) -> str:

242 try:

243 payload = response.json()

244 except ValueError:

245 payload = None

246

247 if isinstance(payload, dict): 247 ↛ 252line 247 didn't jump to line 252 because the condition on line 247 was always true

248 message = payload.get("message")

249 if isinstance(message, str) and message.strip(): 249 ↛ 252line 249 didn't jump to line 252 because the condition on line 249 was always true

250 return message

251

252 if response.text:

253 return response.text

254

255 return f"Alpaca paper API returned HTTP {response.status_code}."

256

257 def _submit_dry_run_order(self, order: Order) -> OrderState:

258 order_id = f"dry-run-order-{self._next_dry_run_order_id}"

259 self._next_dry_run_order_id += 1

260

261 state = OrderState(order_id=order_id, order=order, status=OrderStatus.PENDING)

262 self._dry_run_orders[order_id] = state

263

264 return state

265

266 def submit_order(self, order: Order) -> OrderState:

267 if self._dry_run:

268 return self._submit_dry_run_order(order)

269

270 if not self._trading_enabled:

271 log_event(

272 logger,

273 logging.WARNING,

274 "order.submit_blocked",

275 reason="trading_disabled",

276 symbol=order.symbol,

277 side=order.side.value,

278 quantity=order.quantity,

279 client_order_id=order.client_order_id,

280 )

281

282 raise TradingDisabledError(

283 "Paper trading is disabled; in order to submit "

284 "orders, it has to be explicitly enabled.",

285 )

286

287 if order.client_order_id is None:

288 raise ValueError("Client order ID required for paper order submission")

289

290 store = self._get_state_store()

291

292 try:

293 store.reserve(order)

294 except DuplicateClientOrderError:

295 log_event(

296 logger,

297 logging.WARNING,

298 "order.duplicate_detected",

299 client_order_id=order.client_order_id,

300 symbol=order.symbol,

301 side=order.side.value,

302 quantity=order.quantity,

303 )

304

305 persisted = store.get_by_client_order_id(order.client_order_id)

306 assert persisted is not None

307

308 self._ensure_matching_order(order, persisted.order)

309

310 recovered = self._recover_persisted_order(persisted)

311

312 if recovered is not None: 312 ↛ 315line 312 didn't jump to line 315 because the condition on line 312 was always true

313 return recovered

314

315 payload: dict[str, Any] = {

316 "symbol": order.symbol,

317 "qty": order.quantity,

318 "side": order.side.value,

319 "type": "market",

320 "time_in_force": "day",

321 "client_order_id": order.client_order_id,

322 }

323

324 try:

325 response = self._request("POST", "/v2/orders", json=payload)

326 except (BrokerAPIError, BrokerConnectionError) as submit_error:

327 # the request still could have reached Alpaca, even if we didn't get a 'SUCCESS' answer

328 try:

329 recovered = self.get_order_by_client_order_id(order.client_order_id)

330 except (BrokerAPIError, BrokerConnectionError):

331 raise submit_error

332

333 if recovered is not None: 333 ↛ 349line 333 didn't jump to line 349 because the condition on line 333 was always true

334 self._ensure_matching_order(order, recovered.order)

335 store.update(recovered)

336

337 log_event(

338 logger,

339 logging.WARNING,

340 "order.recovered",

341 reason="ambiguous_submit",

342 client_order_id=order.client_order_id,

343 broker_order_id=recovered.order_id,

344 status=recovered.status.value,

345 )

346

347 return recovered

348

349 raise submit_error

350

351 try:

352 response_payload = response.json()

353 except ValueError as exc:

354 raise BrokerResponseError(

355 "Invalid order response from Alpaca paper trading API.",

356 ) from exc

357

358 state = self._map_order_response(response_payload)

359

360 self._ensure_matching_order(order, state.order)

361 store.update(state)

362

363 log_event(

364 logger,

365 logging.INFO,

366 "order.submitted",

367 client_order_id=order.client_order_id,

368 broker_order_id=state.order_id,

369 symbol=order.symbol,

370 side=order.side.value,

371 quantity=order.quantity,

372 status=state.status.value,

373 )

374

375 return state

376

377 def get_order(self, order_id: str) -> OrderState | None:

378 if self._dry_run:

379 return self._dry_run_orders.get(order_id)

380

381 try:

382 response = self._request("GET", f"/v2/orders/{order_id}")

383 except BrokerAPIError as exc:

384 if exc.status_code == 404: 384 ↛ 386line 384 didn't jump to line 386 because the condition on line 384 was always true

385 return None

386 raise

387

388 try:

389 payload = response.json()

390 except ValueError as exc:

391 raise BrokerResponseError(

392 "Invalid order response from Alpaca paper trading API.",

393 ) from exc

394

395 if not isinstance(payload, dict): 395 ↛ 396line 395 didn't jump to line 396 because the condition on line 395 was never true

396 raise BrokerResponseError("Invalid order response from Alpaca paper trading API.")

397

398 return self._map_order_response(payload)

399

400 def get_order_by_client_order_id(self, client_order_id: str) -> OrderState | None:

401 try:

402 response = self._request(

403 "GET", "/v2/orders:by_client_order_id",

404 params={"client_order_id": client_order_id},

405 )

406 except BrokerAPIError as exc:

407 if exc.status_code == 404:

408 return None

409 raise

410

411 try:

412 payload = response.json()

413 except ValueError:

414 raise BrokerResponseError("Invalid order response from Alpaca paper trading API")

415

416 return self._map_order_response(payload)

417

418 def _cancel_dry_run_order(self, order_id: str) -> OrderState:

419 state = self._dry_run_orders.get(order_id)

420

421 if state is None:

422 raise KeyError(f"Unknown order ID: {order_id}")

423

424 if state.status == OrderStatus.CANCELED:

425 raise ValueError(

426 f"Order {order_id} cannot be canceled from status {state.status.value}.",

427 )

428

429 canceled = OrderState(

430 order_id=state.order_id,

431 order=state.order,

432 status=OrderStatus.CANCELED,

433 )

434 self._dry_run_orders[order_id] = canceled

435

436 return canceled

437

438 def cancel_order(self, order_id: str) -> OrderState:

439 if self._dry_run:

440 return self._cancel_dry_run_order(order_id)

441

442 self._request("DELETE", f"/v2/orders/{order_id}")

443

444 state = self.get_order(order_id)

445

446 if state is None: 446 ↛ 447line 446 didn't jump to line 447 because the condition on line 446 was never true

447 raise BrokerResponseError("Alpaca order could not be retrieved after cancellation.")

448

449 return state

450

451 @staticmethod

452 def _map_fill_activity(payload: dict[str, Any]) -> Fill:

453 try:

454 activity_type = payload["activity_type"]

455 fill_id = payload["id"]

456 order_id = payload["order_id"]

457 symbol = payload["symbol"]

458 side = OrderSide(payload["side"])

459 quantity = float(payload["qty"])

460 price = float(payload["price"])

461 raw_timestamp = payload["transaction_time"]

462 except (KeyError, TypeError, ValueError) as exc:

463 raise BrokerResponseError(

464 "Invalid fill response from Alpaca paper trading API.",

465 ) from exc

466

467 if activity_type != "FILL":

468 raise BrokerResponseError(f"Unsupported Alpaca activity type: {activity_type!r}.")

469

470 if not isinstance(raw_timestamp, str): 470 ↛ 471line 470 didn't jump to line 471 because the condition on line 470 was never true

471 raise BrokerResponseError("Invalid fill response from Alpaca paper trading API.")

472

473 try:

474 timestamp = datetime.fromisoformat(raw_timestamp.replace("Z", "+00:00"))

475 except ValueError as exc:

476 raise BrokerResponseError(

477 "Invalid fill response from Alpaca paper trading API.",

478 ) from exc

479

480 try:

481 return Fill(

482 symbol=symbol,

483 side=side,

484 quantity=quantity,

485 price=price,

486 timestamp=timestamp,

487 fill_id=fill_id,

488 order_id=order_id,

489 )

490 except (TypeError, ValueError) as exc:

491 raise BrokerResponseError(

492 "Invalid fill response from Alpaca paper trading API.",

493 ) from exc

494

495 def get_fills(self, order_id: str) -> list[Fill]:

496 if self._dry_run:

497 return []

498

499 fills: list[Fill] = []

500 page_token: str | None = None

501

502 while True:

503 params: dict[str, Any] = {"order_id": order_id, "direction": "asc", "page_size": 100}

504

505 if page_token is not None:

506 params["page_token"] = page_token

507

508 response = self._request("GET", "/v2/account/activities/FILL", params=params)

509

510 try:

511 payload = response.json()

512 except ValueError as exc:

513 raise BrokerResponseError(

514 "Invalid fill response from Alpaca paper trading API.",

515 ) from exc

516

517 if not isinstance(payload, list): 517 ↛ 518line 517 didn't jump to line 518 because the condition on line 517 was never true

518 raise BrokerResponseError("Invalid fill response from Alpaca paper trading API.")

519 fills.extend(self._map_fill_activity(activity) for activity in payload)

520

521 if len(payload) < 100:

522 break

523

524 last_activity = payload[-1]

525 if not isinstance(last_activity, dict): 525 ↛ 526line 525 didn't jump to line 526 because the condition on line 525 was never true

526 raise BrokerResponseError("Invalid fill response from Alpaca paper trading API.")

527

528 next_page_token = last_activity.get("id")

529 if not isinstance(next_page_token, str) or not next_page_token: 529 ↛ 530line 529 didn't jump to line 530 because the condition on line 529 was never true

530 raise BrokerResponseError("Invalid fill response from Alpaca paper trading API.")

531

532 page_token = next_page_token

533

534 return fills

535

536 @staticmethod

537 def _ensure_matching_order(expected: Order, actual: Order) -> None:

538 if actual != expected:

539 raise OrderConflictError(

540 f"Client order ID {expected.client_order_id!r} "

541 "is already associated with a different order",

542 )

543

544 def _recover_persisted_order(self, persisted: PersistedOrder) -> OrderState | None:

545 order = persisted.order

546

547 if persisted.broker_order_id is not None: 547 ↛ 548line 547 didn't jump to line 548 because the condition on line 547 was never true

548 state = self.get_order(persisted.broker_order_id)

549

550 if state is not None:

551 self._ensure_matching_order(order, state.order)

552 self._get_state_store().update(state)

553 return state

554

555 # We already know locally for sure that a broker ID did exist; NEVER resubmit

556 return OrderState(

557 order_id=persisted.broker_order_id,

558 order=order,

559 status=persisted.status,

560 )

561

562 assert order.client_order_id is not None

563

564 state = self.get_order_by_client_order_id(order.client_order_id)

565

566 if state is None: 566 ↛ 567line 566 didn't jump to line 567 because the condition on line 566 was never true

567 return None

568

569 self._ensure_matching_order(order, state.order)

570 self._get_state_store().update(state)

571

572 return state

573

574 @staticmethod

575 def _map_position_response(payload: dict[str, Any]) -> BrokerPosition:

576 try:

577 symbol = payload["symbol"]

578 quantity = float(payload["qty"])

579 average_entry_price = float(payload["avg_entry_price"])

580 except (KeyError, TypeError, ValueError) as exc:

581 raise BrokerResponseError(

582 "Invalid position response from Alpaca paper trading API",

583 ) from exc

584

585 try:

586 return BrokerPosition(

587 symbol=symbol,

588 quantity=quantity,

589 average_entry_price=average_entry_price,

590 )

591 except (TypeError, ValueError) as exc:

592 raise BrokerResponseError(

593 "Invalid position response from Alpaca paper trading API",

594 ) from exc

595

596 def get_positions(self) -> list[BrokerPosition]:

597 response = self._request("GET", "/v2/positions")

598

599 try:

600 payload = response.json()

601 except ValueError as exc:

602 raise BrokerResponseError(

603 "Invalid position response from Alpaca paper trading API",

604 ) from exc

605

606 if not isinstance(payload, list): 606 ↛ 607line 606 didn't jump to line 607 because the condition on line 606 was never true

607 raise BrokerResponseError("Invalid position response from Alpaca paper trading API")

608

609 positions: list[BrokerPosition] = []

610

611 for item in payload:

612 if not isinstance(item, dict): 612 ↛ 613line 612 didn't jump to line 613 because the condition on line 612 was never true

613 raise BrokerResponseError("Invalid position response from Alpaca paper trading API")

614

615 positions.append(self._map_position_response(item))

616

617 return positions

618

619 @staticmethod

620 def _rate_limit_delay(response: requests.Response) -> float | None:

621 reset_at = response.headers.get("X-RateLimit-Reset")

622

623 if reset_at is not None:

624 try:

625 return max(0.0, float(reset_at) - time.time())

626 except ValueError:

627 pass

628

629 retry_after = response.headers.get("Retry-After")

630

631 if retry_after is not None:

632 try:

633 return max(0.0, float(retry_after))

634 except ValueError:

635 pass

636

637 return None

638

639 @classmethod

640 def _retry_delay(cls, attempt: int, response: requests.Response | None = None) -> float:

641 delay = min(_INITIAL_BACKOFF_SECONDS * (2 ** attempt), _MAX_BACKOFF_SECONDS)

642

643 if response is not None and response.status_code == 429:

644 rate_limit_delay = cls._rate_limit_delay(response)

645

646 if rate_limit_delay is not None:

647 delay = max(delay, rate_limit_delay)

648

649 return delay + random.uniform(0.0, _RETRY_JITTER_SECONDS)

650

651 def get_portfolio_snapshot(self) -> Portfolio:

652 response = self._request("GET", "/v2/account")

653

654 try:

655 payload = response.json()

656 except ValueError as exc:

657 raise BrokerResponseError(

658 "Invalid account response from Alpaca paper trading API.",

659 ) from exc

660

661 if not isinstance(payload, dict): 661 ↛ 662line 661 didn't jump to line 662 because the condition on line 661 was never true

662 raise BrokerResponseError("Invalid account response from Alpaca paper trading API.")

663

664 try:

665 cash = float(payload["cash"])

666 except (KeyError, TypeError, ValueError) as exc:

667 raise BrokerResponseError(

668 "Invalid account response from Alpaca paper trading API.",

669 ) from exc

670

671 if not math.isfinite(cash):

672 raise BrokerResponseError("Invalid account response from Alpaca paper trading API.")

673

674 broker_positions = self.get_positions()

675

676 return Portfolio(

677 cash=cash,

678 positions={

679 position.symbol: Position(

680 symbol=position.symbol,

681 quantity=position.quantity,

682 average_entry_price=position.average_entry_price,

683 )

684 for position in broker_positions

685 },

686 )