1use crate::{
46 AccountEventKind, AccountSnapshot, FillRecoveryFailure, FillRecoveryGap, FillRecoveryScope,
47 InstrumentAccountSnapshot, UnindexedAccountEvent, UnindexedAccountSnapshot,
48 balance::{AssetBalance, Balance},
49 client::{
50 BracketOrderClient, ExecutionClient, OrderStatusClient,
51 order_recovery::{
52 KnownLiveOrders, NoPendingFills, OpenListing, OrderLookup, SharedKnownLiveOrders,
53 UncheckedOrders, fetch_ended_by_key, recover_ended_orders,
54 },
55 },
56 emit_stream_terminated,
57 error::{
58 ApiError, ConnectivityError, OrderError, StreamTerminationReason, UnindexedClientError,
59 UnindexedOrderError,
60 },
61 order::{
62 Order, OrderKey, OrderKind, TimeInForce, TrailingOffsetType, UnindexedInactiveOrder,
63 UnindexedOrderKey, UnindexedOrderSnapshot,
64 bracket::{
65 BracketOrderRequest as UnifiedBracketOrderRequest,
66 BracketOrderResult as UnifiedBracketOrderResult,
67 },
68 id::{ClientOrderId, OrderId, StrategyId, VenueOrderId},
69 request::{OrderRequestCancel, OrderRequestOpen, UnindexedOrderResponseCancel},
70 state::{
71 ActiveOrderState, Cancelled, Expired, Filled, InactiveOrderState, Open, OrderState,
72 UnindexedOrderState,
73 },
74 },
75 parse_env_bool,
76 position::{Position, PositionReport},
77 trade::{AssetFees, Trade, TradeAmendmentKind, TradeId, TradesRead},
78};
79use chrono::{DateTime, SecondsFormat, TimeDelta, Utc};
80use fnv::{FnvHashMap, FnvHashSet};
81use futures::{SinkExt as _, StreamExt as _, stream::BoxStream};
82use indexmap::IndexMap;
83use itertools::Itertools as _;
84use lru::LruCache;
85use rust_decimal::Decimal;
86use rustrade_instrument::{
87 Side,
88 asset::name::AssetNameExchange,
89 exchange::ExchangeId,
90 instrument::{kind::InstrumentKindDiscriminant, name::InstrumentNameExchange},
91};
92use rustrade_integration::protocol::websocket::{WebSocket, WsMessage};
93use serde::{Deserialize, Serialize};
94use smol_str::{SmolStr, format_smolstr};
95use std::{num::NonZeroUsize, pin::Pin, str::FromStr, sync::Arc, time::Duration};
96use tokio::sync::mpsc;
97use tracing::{debug, error, info, trace, warn};
98
99const INITIAL_BACKOFF_MS: u64 = 1_000;
104const MAX_BACKOFF_MS: u64 = 30_000;
105const MAX_RECONNECT_ATTEMPTS: u32 = 10;
106const HEARTBEAT_TIMEOUT_SECS: u64 = 35;
108const FILL_RECOVERY_TIMEOUT_SECS: u64 = 30;
110const SIGNAL_RECOVERY_LOOKBACK_MS: i64 = 1_500;
113const ALPACA_MAX_ACTIVITIES: usize = 100;
115const DEFAULT_RATE_LIMIT_DELAY_SECS: u64 = 60;
117const MAX_RATE_LIMIT_ATTEMPTS: u32 = 4;
120const DEDUP_CACHE_SIZE: usize = 4_000;
124
125const USD: &str = "USD";
128const WS_HANDSHAKE_TIMEOUT_SECS: u64 = 15;
130const WS_CLOSE_TIMEOUT_SECS: u64 = 5;
133
134struct GracefulShutdownStream<S> {
151 inner: S,
152 _handle: tokio::task::JoinHandle<()>,
158}
159
160impl<S> GracefulShutdownStream<S> {
161 fn new(inner: S, handle: tokio::task::JoinHandle<()>) -> Self {
162 Self {
163 inner,
164 _handle: handle,
165 }
166 }
167}
168
169impl<S: futures::Stream + Unpin> futures::Stream for GracefulShutdownStream<S> {
170 type Item = S::Item;
171 fn poll_next(
172 mut self: Pin<&mut Self>,
173 cx: &mut std::task::Context<'_>,
174 ) -> std::task::Poll<Option<Self::Item>> {
175 Pin::new(&mut self.inner).poll_next(cx)
176 }
177 fn size_hint(&self) -> (usize, Option<usize>) {
178 self.inner.size_hint()
179 }
180}
181
182impl<S> Drop for GracefulShutdownStream<S> {
183 fn drop(&mut self) {
184 }
188}
189
190struct RateLimitTracker {
196 blocked_until: parking_lot::Mutex<Option<tokio::time::Instant>>,
197}
198
199impl RateLimitTracker {
200 fn new() -> Self {
201 Self {
202 blocked_until: parking_lot::Mutex::new(None),
203 }
204 }
205
206 async fn wait_if_blocked(&self) {
208 loop {
209 let now = tokio::time::Instant::now();
213 let deadline = {
221 let mut guard = self.blocked_until.lock();
222 let d = *guard;
223 if matches!(d, Some(t) if t <= now) {
224 *guard = None;
227 }
228 d
229 };
230 match deadline {
231 None => return,
232 Some(until) => {
233 if until <= now {
234 return;
236 }
237 #[allow(clippy::cast_possible_truncation)]
239 let delay_ms = (until - now).as_millis() as u64;
240 debug!(delay_ms, "Alpaca REST rate-limited, waiting before request");
241 tokio::time::sleep_until(until).await;
242 }
243 }
244 }
245 }
246
247 fn on_rate_limited(&self, retry_after: Option<Duration>) {
249 let delay = retry_after.unwrap_or(Duration::from_secs(DEFAULT_RATE_LIMIT_DELAY_SECS));
250 let was_blocked = self.extend(delay);
251 if was_blocked {
252 debug!(
253 delay_secs = delay.as_secs(),
254 "Alpaca rate-limit cooldown extended"
255 );
256 } else {
257 warn!(
258 delay_secs = delay.as_secs(),
259 "Alpaca entering rate-limit degradation mode"
260 );
261 }
262 }
263
264 fn on_bucket_exhausted(&self, reset: Duration) {
267 let was_blocked = self.extend(reset);
268 if !was_blocked {
270 info!(
271 delay_ms = u64::try_from(reset.as_millis()).unwrap_or(u64::MAX),
272 "Alpaca rate-limit window exhausted (X-Ratelimit-Remaining: 0), pausing requests until it resets"
273 );
274 }
275 }
276
277 fn extend(&self, delay: Duration) -> bool {
280 let now = tokio::time::Instant::now();
281 let new_deadline = now + delay;
282 let mut guard = self.blocked_until.lock();
283 let was_blocked = guard.is_some_and(|until| until > now);
284 *guard = Some(guard.map_or(new_deadline, |existing| existing.max(new_deadline)));
285 was_blocked
286 }
287}
288
289struct ExponentialBackoff {
294 attempt: u32,
295 max_attempts: u32,
296 initial_ms: u64,
297 max_ms: u64,
298}
299
300impl ExponentialBackoff {
301 fn new() -> Self {
302 Self {
303 attempt: 0,
304 max_attempts: MAX_RECONNECT_ATTEMPTS,
305 initial_ms: INITIAL_BACKOFF_MS,
306 max_ms: MAX_BACKOFF_MS,
307 }
308 }
309
310 fn reset(&mut self) {
311 self.attempt = 0;
312 }
313
314 fn attempts(&self) -> u32 {
320 self.attempt
321 }
322
323 async fn wait(&mut self) -> bool {
325 if self.attempt >= self.max_attempts {
326 return false;
327 }
328 let delay_ms = self
329 .initial_ms
330 .saturating_mul(2u64.saturating_pow(self.attempt))
331 .min(self.max_ms);
332 self.attempt += 1;
333 debug!(
334 attempt = self.attempt,
335 max = self.max_attempts,
336 delay_ms,
337 "Alpaca reconnect backoff"
338 );
339 tokio::time::sleep(Duration::from_millis(delay_ms)).await;
340 true
341 }
342}
343
344type SharedDedupCache = Arc<parking_lot::Mutex<LruCache<FillKey, ()>>>;
352
353#[derive(Debug, Clone, PartialEq, Eq, Hash)]
358enum FillKey {
359 Execution(SmolStr),
362 Cumulative(SmolStr),
364 CumulativeWithoutId(SmolStr),
366}
367
368fn new_dedup_cache() -> SharedDedupCache {
369 #[allow(clippy::unwrap_used)]
372 Arc::new(parking_lot::Mutex::new(LruCache::new(
373 NonZeroUsize::new(DEDUP_CACHE_SIZE).unwrap(),
374 )))
375}
376
377fn is_duplicate_fill(
397 cache: &SharedDedupCache,
398 execution_id: Option<&str>,
399 cumulative: SmolStr,
400) -> bool {
401 let execution = execution_id.map(|id| FillKey::Execution(SmolStr::new(id)));
403 let mut guard = cache.lock();
404 match execution {
406 Some(execution) => {
407 if guard.peek(&execution).is_some()
408 || guard
409 .peek(&FillKey::CumulativeWithoutId(cumulative.clone()))
410 .is_some()
411 {
412 return true;
413 }
414 guard.put(execution, ());
415 guard.put(FillKey::Cumulative(cumulative), ());
416 }
417 None => {
418 if guard
419 .peek(&FillKey::Cumulative(cumulative.clone()))
420 .is_some()
421 {
422 return true;
423 }
424 guard.put(FillKey::Cumulative(cumulative.clone()), ());
425 guard.put(FillKey::CumulativeWithoutId(cumulative), ());
426 }
427 }
428 false
429}
430
431fn record_execution(cache: &SharedDedupCache, execution_id: &str) {
434 let execution = FillKey::Execution(SmolStr::new(execution_id));
435 cache.lock().put(execution, ());
436}
437
438#[derive(Clone, Deserialize)]
445#[serde(deny_unknown_fields)]
446pub struct AlpacaConfig {
447 api_key: String,
449 secret_key: String,
450
451 #[serde(default = "default_paper")]
453 pub paper: bool,
454 #[cfg(test)]
456 pub base_url_override: Option<String>,
457}
458
459fn default_paper() -> bool {
465 true
466}
467
468impl std::fmt::Debug for AlpacaConfig {
470 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
471 f.debug_struct("AlpacaConfig")
472 .field("api_key", &"***")
473 .field("secret_key", &"***")
474 .field("paper", &self.paper)
475 .finish()
476 }
477}
478
479impl AlpacaConfig {
480 pub fn new(api_key: String, secret_key: String) -> Self {
483 Self::paper(api_key, secret_key)
484 }
485
486 pub fn paper(api_key: String, secret_key: String) -> Self {
488 Self {
489 api_key,
490 secret_key,
491 paper: true,
492 #[cfg(test)]
493 base_url_override: None,
494 }
495 }
496
497 pub fn production(api_key: String, secret_key: String) -> Self {
502 Self {
503 api_key,
504 secret_key,
505 paper: false,
506 #[cfg(test)]
507 base_url_override: None,
508 }
509 }
510
511 pub fn from_env() -> Result<Self, AlpacaConfigError> {
529 let api_key = match std::env::var("ALPACA_API_KEY") {
530 Ok(value) => value,
531 Err(std::env::VarError::NotPresent) => return Err(AlpacaConfigError::MissingApiKey),
532 Err(std::env::VarError::NotUnicode(_)) => {
533 return Err(AlpacaConfigError::InvalidApiKey);
534 }
535 };
536 let secret_key = match std::env::var("ALPACA_SECRET_KEY") {
537 Ok(value) => value,
538 Err(std::env::VarError::NotPresent) => return Err(AlpacaConfigError::MissingSecretKey),
539 Err(std::env::VarError::NotUnicode(_)) => {
540 return Err(AlpacaConfigError::InvalidSecretKey);
541 }
542 };
543
544 let paper = match std::env::var("ALPACA_PAPER") {
545 Ok(value) => parse_env_bool(&value).ok_or(AlpacaConfigError::InvalidPaper(value))?,
546 Err(std::env::VarError::NotPresent) => true,
547 Err(std::env::VarError::NotUnicode(value)) => {
550 return Err(AlpacaConfigError::InvalidPaper(
551 value.to_string_lossy().into_owned(),
552 ));
553 }
554 };
555
556 if paper {
557 Ok(Self::paper(api_key, secret_key))
558 } else {
559 Ok(Self::production(api_key, secret_key))
560 }
561 }
562
563 #[cfg(test)]
565 pub fn with_base_url(api_key: String, secret_key: String, base_url: String) -> Self {
566 Self {
567 api_key,
568 secret_key,
569 paper: true,
570 base_url_override: Some(base_url),
571 }
572 }
573
574 pub fn api_key(&self) -> &str {
575 &self.api_key
576 }
577
578 pub fn rest_base_url(&self) -> &str {
582 #[cfg(test)]
583 if let Some(ref url) = self.base_url_override {
584 return url.as_str();
585 }
586 if self.paper {
587 "https://paper-api.alpaca.markets"
588 } else {
589 "https://api.alpaca.markets"
590 }
591 }
592
593 pub fn ws_url(&self) -> &'static str {
595 if self.paper {
596 "wss://paper-api.alpaca.markets/stream"
597 } else {
598 "wss://api.alpaca.markets/stream"
599 }
600 }
601}
602
603#[derive(Debug, PartialEq, thiserror::Error)]
604pub enum AlpacaConfigError {
605 #[error("ALPACA_API_KEY environment variable not set")]
606 MissingApiKey,
607
608 #[error("ALPACA_API_KEY environment variable is not valid UTF-8")]
611 InvalidApiKey,
612
613 #[error("ALPACA_SECRET_KEY environment variable not set")]
614 MissingSecretKey,
615
616 #[error("ALPACA_SECRET_KEY environment variable is not valid UTF-8")]
617 InvalidSecretKey,
618
619 #[error("ALPACA_PAPER must be true or false, got {0}")]
620 InvalidPaper(String),
621}
622
623#[derive(Debug, Deserialize)]
632struct AlpacaAccount {
633 #[serde(with = "rust_decimal::serde::str")]
636 cash: Decimal,
637 #[serde(with = "rust_decimal::serde::str")]
641 non_marginable_buying_power: Decimal,
642}
643
644#[derive(Debug, Deserialize)]
649struct AlpacaPosition {
650 symbol: String,
653 asset_class: String,
655 #[serde(default)]
659 side: AlpacaPositionSide,
660 #[serde(with = "rust_decimal::serde::str")]
662 qty: Decimal,
663 #[serde(with = "rust_decimal::serde::str")]
665 qty_available: Decimal,
666 #[serde(default, with = "rust_decimal::serde::str_option")]
668 avg_entry_price: Option<Decimal>,
669 #[serde(default, with = "rust_decimal::serde::str_option")]
671 unrealized_pl: Option<Decimal>,
672}
673
674#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)]
676#[serde(rename_all = "lowercase")]
677enum AlpacaPositionSide {
678 Long,
679 Short,
680 #[default]
684 #[serde(other)]
685 Unknown,
686}
687
688#[derive(Debug, Deserialize)]
689struct AlpacaOrderResponse {
690 id: String,
691 client_order_id: Option<String>,
692 symbol: String,
693 qty: Option<String>,
694 filled_qty: String,
695 #[serde(default)]
697 filled_avg_price: Option<String>,
698 #[serde(default)]
702 status: Option<String>,
703 side: String,
704 #[serde(rename = "type")]
705 order_type: String,
706 time_in_force: String,
707 limit_price: Option<String>,
708 stop_price: Option<String>,
709 trail_percent: Option<String>,
710 trail_price: Option<String>,
711 created_at: String,
712 updated_at: Option<String>,
715}
716
717#[derive(Debug, Deserialize)]
718struct AlpacaActivity {
719 id: String,
720 order_id: String,
721 symbol: String,
722 side: String,
723 price: String,
724 qty: String,
725 transaction_time: String,
726 cum_qty: Option<String>,
734}
735
736#[derive(Debug, Deserialize)]
737struct AlpacaApiError {
738 message: String,
739}
740
741#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
753#[serde(rename_all = "snake_case")]
754pub enum AlpacaPositionIntent {
755 BuyToOpen,
756 BuyToClose,
757 SellToOpen,
758 SellToClose,
759}
760
761#[derive(Debug, Serialize)]
769struct TakeProfitParams {
770 limit_price: String,
771}
772
773#[derive(Debug, Serialize)]
778struct StopLossParams {
779 stop_price: String,
780 #[serde(skip_serializing_if = "Option::is_none")]
781 limit_price: Option<String>,
782}
783
784#[non_exhaustive]
819#[derive(Debug, Clone)]
820pub struct AlpacaBracketOrderRequest {
821 pub instrument: InstrumentNameExchange,
823 pub strategy: StrategyId,
825 pub cid: ClientOrderId,
827 pub side: Side,
829 pub quantity: Decimal,
831 pub entry_price: Decimal,
833 pub take_profit_price: Decimal,
835 pub stop_loss_price: Decimal,
837 pub stop_loss_limit_price: Option<Decimal>,
839 pub time_in_force: TimeInForce,
841}
842
843impl AlpacaBracketOrderRequest {
844 #[allow(clippy::too_many_arguments)] pub fn new(
849 instrument: InstrumentNameExchange,
850 strategy: StrategyId,
851 cid: ClientOrderId,
852 side: Side,
853 quantity: Decimal,
854 entry_price: Decimal,
855 take_profit_price: Decimal,
856 stop_loss_price: Decimal,
857 time_in_force: TimeInForce,
858 ) -> Self {
859 Self {
860 instrument,
861 strategy,
862 cid,
863 side,
864 quantity,
865 entry_price,
866 take_profit_price,
867 stop_loss_price,
868 stop_loss_limit_price: None,
869 time_in_force,
870 }
871 }
872
873 #[must_use]
875 pub fn with_stop_loss_limit_price(mut self, price: Decimal) -> Self {
876 self.stop_loss_limit_price = Some(price);
877 self
878 }
879}
880
881#[non_exhaustive]
892#[derive(Debug, Clone)]
893pub struct AlpacaBracketOrderResult {
894 pub parent: Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState>,
896}
897
898#[derive(Debug, Serialize)]
903struct AlpacaOrderRequest<'a> {
904 symbol: &'a str,
905 qty: String,
906 side: &'static str,
907 #[serde(rename = "type")]
908 order_type: &'static str,
909 time_in_force: &'static str,
910 #[serde(skip_serializing_if = "Option::is_none")]
911 limit_price: Option<String>,
912 #[serde(skip_serializing_if = "Option::is_none")]
913 stop_price: Option<String>,
914 #[serde(skip_serializing_if = "Option::is_none")]
915 trail_percent: Option<String>,
916 #[serde(skip_serializing_if = "Option::is_none")]
917 trail_price: Option<String>,
918 #[serde(skip_serializing_if = "Option::is_none")]
919 client_order_id: Option<&'a str>,
920 #[serde(skip_serializing_if = "Option::is_none")]
924 position_intent: Option<AlpacaPositionIntent>,
925 #[serde(skip_serializing_if = "Option::is_none")]
928 order_class: Option<&'static str>,
929 #[serde(skip_serializing_if = "Option::is_none")]
930 take_profit: Option<TakeProfitParams>,
931 #[serde(skip_serializing_if = "Option::is_none")]
932 stop_loss: Option<StopLossParams>,
933}
934
935#[derive(Debug, Deserialize)]
944struct AlpacaStreamMessage<'a> {
945 stream: SmolStr,
948 #[serde(borrow)]
949 data: &'a serde_json::value::RawValue,
950}
951
952#[derive(Debug, Deserialize)]
960struct AlpacaTradeUpdate<'a> {
961 event: SmolStr,
963 #[serde(borrow)]
964 order: AlpacaOrderWs<'a>,
965 #[serde(borrow)]
967 price: Option<&'a str>,
968 #[serde(borrow)]
970 qty: Option<&'a str>,
971 #[serde(borrow)]
972 timestamp: Option<&'a str>,
973 #[serde(borrow)]
977 execution_id: Option<&'a str>,
978 #[serde(borrow)]
982 previous_execution_id: Option<&'a str>,
983}
984
985#[derive(Debug, Deserialize)]
987struct AlpacaOrderWs<'a> {
988 id: SmolStr,
991 client_order_id: Option<SmolStr>,
992 symbol: SmolStr,
995 #[serde(borrow)]
996 qty: Option<&'a str>,
997 #[serde(borrow)]
1004 filled_qty: Option<&'a str>,
1005 side: SmolStr,
1008 #[serde(rename = "type")]
1009 order_type: SmolStr,
1010 time_in_force: SmolStr,
1011 #[serde(borrow)]
1012 limit_price: Option<&'a str>,
1013 #[serde(borrow)]
1014 stop_price: Option<&'a str>,
1015 #[serde(borrow)]
1016 trail_percent: Option<&'a str>,
1017 #[serde(borrow)]
1018 trail_price: Option<&'a str>,
1019 status: SmolStr,
1020}
1021
1022#[derive(Clone)]
1043pub struct AlpacaClient {
1044 config: Arc<AlpacaConfig>,
1045 http: reqwest::Client,
1048 rate_limiter: Arc<RateLimitTracker>,
1049 orders_url: String,
1051 known_live: SharedKnownLiveOrders,
1054}
1055
1056impl std::fmt::Debug for AlpacaClient {
1057 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1058 f.debug_struct("AlpacaClient")
1059 .field("paper", &self.config.paper)
1060 .finish_non_exhaustive()
1061 }
1062}
1063
1064impl AlpacaClient {
1065 #[allow(clippy::expect_used)] fn build_http(config: &AlpacaConfig) -> reqwest::Client {
1073 use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
1074 let mut headers = HeaderMap::new();
1075 headers.insert(
1077 HeaderName::from_static("apca-api-key-id"),
1078 HeaderValue::from_str(&config.api_key)
1079 .expect("Alpaca API key contains invalid header characters"),
1080 );
1081 headers.insert(
1082 HeaderName::from_static("apca-api-secret-key"),
1083 HeaderValue::from_str(&config.secret_key)
1084 .expect("Alpaca secret key contains invalid header characters"),
1085 );
1086 reqwest::Client::builder()
1087 .default_headers(headers)
1088 .build()
1089 .expect("failed to build reqwest client for Alpaca")
1090 }
1091
1092 fn base_url(&self) -> &str {
1093 self.config.rest_base_url()
1094 }
1095
1096 fn remember_live<'a>(
1098 &self,
1099 orders: impl IntoIterator<Item = &'a Order<ExchangeId, InstrumentNameExchange, Open>>,
1100 ) {
1101 let mut known = self.known_live.lock();
1102 for order in orders {
1103 known.live(&order.key, order.quantity, order.kind, &order.state);
1104 }
1105 }
1106}
1107
1108fn parse_rate_limit_delay(headers: &reqwest::header::HeaderMap) -> Option<Duration> {
1119 headers
1120 .get("x-ratelimit-reset")
1121 .and_then(|v| v.to_str().ok())
1122 .and_then(|s| s.parse::<u64>().ok())
1123 .map(|reset_ts| {
1124 let now_secs = std::time::SystemTime::now()
1125 .duration_since(std::time::UNIX_EPOCH)
1126 .unwrap_or_default()
1127 .as_secs();
1128 Duration::from_secs(reset_ts.saturating_sub(now_secs).max(1))
1129 })
1130}
1131
1132fn observe_rate_limit_remaining(
1139 rate_limiter: &RateLimitTracker,
1140 headers: &reqwest::header::HeaderMap,
1141) {
1142 let exhausted = headers
1143 .get("x-ratelimit-remaining")
1144 .and_then(|v| v.to_str().ok())
1145 .and_then(|s| s.parse::<u32>().ok())
1146 == Some(0);
1147 if !exhausted {
1148 return;
1149 }
1150 match parse_rate_limit_delay(headers) {
1151 Some(reset) => rate_limiter
1154 .on_bucket_exhausted(reset.min(Duration::from_secs(DEFAULT_RATE_LIMIT_DELAY_SECS))),
1155 None => debug!(
1156 "Alpaca REST rate-limit window exhausted (X-Ratelimit-Remaining: 0) without X-Ratelimit-Reset"
1157 ),
1158 }
1159}
1160
1161async fn rest_with_retry<T>(
1180 rate_limiter: &RateLimitTracker,
1181 build_request: impl FnMut() -> reqwest::RequestBuilder,
1182) -> Result<T, UnindexedClientError>
1183where
1184 T: for<'de> Deserialize<'de>,
1185{
1186 match rest_request(rate_limiter, build_request).await? {
1187 Fetched::Found(value) => Ok(value),
1188 Fetched::NotFound(message) => Err(UnindexedClientError::Api(parse_api_error(
1189 reqwest::StatusCode::NOT_FOUND,
1190 &message,
1191 ))),
1192 }
1193}
1194
1195async fn rest_lookup_with_retry<T>(
1198 rate_limiter: &RateLimitTracker,
1199 build_request: impl FnMut() -> reqwest::RequestBuilder,
1200) -> Result<Option<T>, UnindexedClientError>
1201where
1202 T: for<'de> Deserialize<'de>,
1203{
1204 match rest_request(rate_limiter, build_request).await? {
1205 Fetched::Found(value) => Ok(Some(value)),
1206 Fetched::NotFound(_) => Ok(None),
1207 }
1208}
1209
1210#[derive(Debug)]
1212enum Fetched<T> {
1213 Found(T),
1215 NotFound(String),
1217}
1218
1219async fn rest_request<T>(
1221 rate_limiter: &RateLimitTracker,
1222 mut build_request: impl FnMut() -> reqwest::RequestBuilder,
1223) -> Result<Fetched<T>, UnindexedClientError>
1224where
1225 T: for<'de> Deserialize<'de>,
1226{
1227 for attempt in 0..MAX_RATE_LIMIT_ATTEMPTS {
1228 rate_limiter.wait_if_blocked().await;
1229 let response = build_request()
1230 .send()
1231 .await
1232 .map_err(|e| connectivity_err(format!("Alpaca REST request failed: {e}")))?;
1233
1234 let status = response.status();
1235
1236 if status == reqwest::StatusCode::TOO_MANY_REQUESTS {
1237 let reset_delay = parse_rate_limit_delay(response.headers());
1238
1239 if attempt + 1 < MAX_RATE_LIMIT_ATTEMPTS {
1240 warn!(
1241 attempt = attempt + 1,
1242 max_attempts = MAX_RATE_LIMIT_ATTEMPTS,
1243 "Alpaca REST rate-limited (429), retrying"
1244 );
1245 rate_limiter.on_rate_limited(reset_delay);
1246 continue;
1247 }
1248
1249 warn!(
1251 max_attempts = MAX_RATE_LIMIT_ATTEMPTS,
1252 "Alpaca REST rate-limit retries exhausted"
1253 );
1254 return Err(UnindexedClientError::Api(ApiError::RateLimit));
1255 }
1256
1257 observe_rate_limit_remaining(rate_limiter, response.headers());
1258
1259 if status == reqwest::StatusCode::NO_CONTENT {
1263 return Err(UnindexedClientError::Internal(
1264 "Alpaca REST returned 204 No Content — use rest_delete_with_retry for DELETE endpoints"
1265 .to_string(),
1266 ));
1267 }
1268
1269 let bytes = response
1270 .bytes()
1271 .await
1272 .map_err(|e| connectivity_err(format!("Alpaca REST read body failed: {e}")))?;
1273
1274 if status.is_success() {
1277 return serde_json::from_slice::<T>(&bytes)
1278 .map(Fetched::Found)
1279 .map_err(|e| {
1280 UnindexedClientError::Internal(format!(
1281 "Alpaca REST JSON parse error ({status}): {e} | body: {}",
1282 String::from_utf8_lossy(&bytes)
1283 .chars()
1284 .take(200)
1285 .collect::<String>()
1286 ))
1287 });
1288 }
1289
1290 let api_err = serde_json::from_slice::<AlpacaApiError>(&bytes)
1292 .map(|e| e.message)
1293 .unwrap_or_else(|_| String::from_utf8_lossy(&bytes).into_owned());
1294
1295 if status == reqwest::StatusCode::NOT_FOUND {
1296 return Ok(Fetched::NotFound(api_err));
1297 }
1298
1299 if status.is_client_error() {
1308 return Err(UnindexedClientError::Api(parse_api_error(status, &api_err)));
1309 }
1310 return Err(connectivity_err(format!(
1311 "Alpaca REST error {status}: {api_err}"
1312 )));
1313 }
1314 unreachable!("Alpaca REST retry loop exited without returning")
1315}
1316
1317async fn rest_delete_with_retry(
1322 rate_limiter: &RateLimitTracker,
1323 mut build_request: impl FnMut() -> reqwest::RequestBuilder,
1324) -> Result<(), UnindexedOrderError> {
1325 for attempt in 0..MAX_RATE_LIMIT_ATTEMPTS {
1326 rate_limiter.wait_if_blocked().await;
1327 let response = build_request().send().await.map_err(|e| {
1328 UnindexedOrderError::Connectivity(ConnectivityError::Socket(format!(
1329 "Alpaca cancel request failed: {e}"
1330 )))
1331 })?;
1332
1333 let status = response.status();
1334
1335 if status == reqwest::StatusCode::TOO_MANY_REQUESTS {
1336 let reset_delay = parse_rate_limit_delay(response.headers());
1337
1338 if attempt + 1 < MAX_RATE_LIMIT_ATTEMPTS {
1339 warn!(
1340 attempt = attempt + 1,
1341 max_attempts = MAX_RATE_LIMIT_ATTEMPTS,
1342 "Alpaca cancel rate-limited (429), retrying"
1343 );
1344 rate_limiter.on_rate_limited(reset_delay);
1345 continue;
1346 }
1347
1348 warn!(
1350 max_attempts = MAX_RATE_LIMIT_ATTEMPTS,
1351 "Alpaca cancel rate-limit retries exhausted"
1352 );
1353 return Err(UnindexedOrderError::Rejected(ApiError::RateLimit));
1354 }
1355
1356 observe_rate_limit_remaining(rate_limiter, response.headers());
1357
1358 if status == reqwest::StatusCode::NO_CONTENT || status.is_success() {
1360 return Ok(());
1361 }
1362
1363 let bytes = response
1364 .bytes()
1365 .await
1366 .inspect_err(
1367 |e| warn!(%e, %status, "Alpaca cancel_order: failed to read error response body"),
1368 )
1369 .unwrap_or_default();
1370 let msg = serde_json::from_slice::<AlpacaApiError>(&bytes)
1371 .map(|e| e.message)
1372 .unwrap_or_else(|_| String::from_utf8_lossy(&bytes).into_owned());
1373
1374 return Err(parse_order_error(status, &msg));
1375 }
1376 unreachable!("Alpaca cancel retry loop exited without returning")
1377}
1378
1379impl ExecutionClient for AlpacaClient {
1384 const EXCHANGE: ExchangeId = ExchangeId::AlpacaBroker;
1385
1386 const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant] = &[
1393 InstrumentKindDiscriminant::Spot,
1394 InstrumentKindDiscriminant::Option,
1395 ];
1396 type Config = AlpacaConfig;
1397 type AccountStream = BoxStream<'static, UnindexedAccountEvent>;
1398
1399 fn new(config: Self::Config) -> Self {
1404 let http = Self::build_http(&config);
1405 let orders_url = format!("{}/v2/orders", config.rest_base_url());
1406 Self {
1407 config: Arc::new(config),
1408 http,
1409 rate_limiter: Arc::new(RateLimitTracker::new()),
1410 orders_url,
1411 known_live: KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
1412 }
1413 }
1414
1415 async fn account_snapshot(
1462 &self,
1463 assets: &[AssetNameExchange],
1464 instruments: &[InstrumentNameExchange],
1465 ) -> Result<UnindexedAccountSnapshot, UnindexedClientError> {
1466 let base = self.base_url();
1467 let http = self.http.clone();
1468 let rl = &self.rate_limiter;
1469
1470 let wants_usd = assets.is_empty() || assets.iter().any(is_usd);
1471 let wants_non_usd = assets.is_empty() || assets.iter().any(|a| !is_usd(a));
1472
1473 let account_url = format!("{base}/v2/account");
1475 let positions_url = format!("{base}/v2/positions");
1476 let (account, positions): (Option<AlpacaAccount>, Vec<AlpacaPosition>) = tokio::try_join!(
1477 async {
1478 if wants_usd {
1479 rest_with_retry(rl, || http.get(&account_url))
1480 .await
1481 .map(Some)
1482 } else {
1483 Ok(None)
1484 }
1485 },
1486 rest_with_retry(rl, || http.get(&positions_url)),
1487 )?;
1488
1489 let mut balances = account
1490 .map(|account| convert_account_to_balances(&account, assets))
1491 .unwrap_or_default();
1492 if wants_non_usd {
1493 balances.extend(convert_positions_to_balances(&positions, assets));
1494 }
1495
1496 let converted_positions = convert_positions(&positions, Utc::now())?;
1499 let open_orders = fetch_raw_open_orders(&http, rl, base, instruments).await?;
1500
1501 let instrument_snapshots =
1503 build_instrument_snapshots(open_orders, converted_positions, instruments);
1504 {
1505 let mut known = self.known_live.lock();
1506 for order in instrument_snapshots
1507 .iter()
1508 .flat_map(|snapshot| &snapshot.orders)
1509 {
1510 if let OrderState::Active(ActiveOrderState::Open(open)) = &order.state {
1511 known.live(&order.key, order.quantity, order.kind, open);
1512 }
1513 }
1514 }
1515
1516 Ok(AccountSnapshot::new(
1517 ExchangeId::AlpacaBroker,
1518 balances,
1519 instrument_snapshots,
1520 ))
1521 }
1522
1523 async fn account_stream(
1649 &self,
1650 _assets: &[AssetNameExchange], instruments: &[InstrumentNameExchange],
1655 ) -> Result<Self::AccountStream, UnindexedClientError> {
1656 let initial_ws = connect_and_subscribe(&self.config).await?;
1659
1660 let (tx, rx) = mpsc::unbounded_channel::<UnindexedAccountEvent>();
1665 let dedup = new_dedup_cache();
1666 let config = self.config.clone();
1667 let http = self.http.clone();
1668 let rate_limiter = self.rate_limiter.clone();
1669 let known = self.known_live.clone();
1670 let instruments = instruments.to_vec();
1671
1672 let cm_handle = tokio::spawn(connection_manager(
1673 tx,
1674 dedup,
1675 config,
1676 http,
1677 rate_limiter,
1678 known,
1679 instruments,
1680 Some(initial_ws),
1681 ));
1682
1683 let rx_stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx);
1684 let guarded = GracefulShutdownStream::new(rx_stream, cm_handle);
1685 Ok(futures::StreamExt::boxed(guarded))
1686 }
1687
1688 async fn cancel_order(
1707 &self,
1708 request: OrderRequestCancel<ExchangeId, &InstrumentNameExchange>,
1709 ) -> UnindexedOrderResponseCancel {
1710 let key = crate::order::OrderKey {
1711 exchange: request.key.exchange,
1712 instrument: request.key.instrument.clone(),
1713 strategy: request.key.strategy.clone(),
1714 cid: request.key.cid.clone(),
1715 };
1716
1717 let order_id: SmolStr = match request.state.id.as_ref().and_then(VenueOrderId::assigned) {
1721 Some(id) => id.0.clone(),
1722 None => {
1723 warn!(
1724 instrument = %key.instrument,
1725 "Alpaca cancel_order: no exchange order ID available (clientOrderId-only cancel not supported)"
1726 );
1727 return crate::order::request::OrderResponseCancel {
1728 key,
1729 state: Err(UnindexedOrderError::Rejected(ApiError::OrderRejected(
1730 "exchange order ID required for cancel (fetch_open_orders to resolve)"
1731 .into(),
1732 ))),
1733 };
1734 }
1735 };
1736
1737 let base = self.base_url();
1738 let http = self.http.clone();
1739 let url = format!("{base}/v2/orders/{order_id}");
1740
1741 match rest_delete_with_retry(&self.rate_limiter, || http.delete(&url)).await {
1742 Ok(()) => {
1743 let exchange_order_id = OrderId(order_id);
1744 crate::order::request::OrderResponseCancel {
1747 key,
1748 state: Ok(Cancelled::new(exchange_order_id, Utc::now(), None)),
1749 }
1750 }
1751 Err(e) => crate::order::request::OrderResponseCancel { key, state: Err(e) },
1752 }
1753 }
1754
1755 async fn open_order(
1776 &self,
1777 request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
1778 ) -> Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState> {
1779 let side = request.state.side;
1780 let reduce_only = request.state.reduce_only;
1781 self.open_order_inner(request, map_position_intent(side, reduce_only))
1782 .await
1783 }
1784
1785 async fn fetch_balances(
1793 &self,
1794 assets: &[AssetNameExchange],
1795 ) -> Result<Vec<AssetBalance<AssetNameExchange>>, UnindexedClientError> {
1796 let base = self.base_url();
1797 let http = self.http.clone();
1798 let mut result = Vec::new();
1799
1800 let wants_usd = assets.is_empty() || assets.iter().any(is_usd);
1803 if wants_usd {
1804 let account_url = format!("{base}/v2/account");
1806 let account: AlpacaAccount =
1807 rest_with_retry(&self.rate_limiter, || http.get(&account_url)).await?;
1808 result.extend(convert_account_to_balances(&account, assets));
1809 }
1810
1811 let wants_non_usd = assets.is_empty() || assets.iter().any(|a| !is_usd(a));
1813 if wants_non_usd {
1814 let positions_url = format!("{base}/v2/positions");
1816 let positions: Vec<AlpacaPosition> =
1817 rest_with_retry(&self.rate_limiter, || http.get(&positions_url)).await?;
1818 result.extend(convert_positions_to_balances(&positions, assets));
1819 }
1820
1821 Ok(result)
1822 }
1823
1824 async fn fetch_open_orders(
1825 &self,
1826 instruments: &[InstrumentNameExchange],
1827 ) -> Result<Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, UnindexedClientError> {
1828 let base = self.base_url();
1829 let http = self.http.clone();
1830
1831 let open_orders =
1832 fetch_raw_open_orders(&http, &self.rate_limiter, base, instruments).await?;
1833
1834 let result: Vec<_> = open_orders
1835 .into_iter()
1836 .filter_map(|o| convert_open_order(&o))
1837 .collect();
1838 self.remember_live(&result);
1839 Ok(result)
1840 }
1841
1842 async fn fetch_trades(
1855 &self,
1856 start: DateTime<Utc>,
1857 end: DateTime<Utc>,
1858 instruments: &[InstrumentNameExchange],
1859 ) -> Result<TradesRead<AssetNameExchange, InstrumentNameExchange>, UnindexedClientError> {
1860 if start > end || end < DateTime::UNIX_EPOCH {
1862 return Ok(TradesRead::complete(Vec::new()));
1863 }
1864 let page = paginate_activities(
1865 &self.http,
1866 &self.rate_limiter,
1867 self.base_url(),
1868 start,
1869 Some(end),
1870 )
1871 .await?;
1872
1873 let resume = page.resume.filter(|resume| *resume <= end);
1877 if resume.is_some_and(|resume| resume <= floor_millis(start)) {
1879 error!(
1880 %start,
1881 fills_read = page.activities.len(),
1882 "Alpaca fetch_trades: a full read did not advance past the span's first millisecond"
1883 );
1884 return Err(UnindexedClientError::Truncated {
1885 fills_read: page.activities.len(),
1886 });
1887 }
1888
1889 let instrument_set: fnv::FnvHashSet<&str> =
1892 instruments.iter().map(|i| i.name().as_str()).collect();
1893 let trades = page
1894 .activities
1895 .iter()
1896 .filter(|a| instrument_set.is_empty() || instrument_set.contains(a.symbol.as_str()))
1897 .filter(|a| {
1900 parse_timestamp(&a.transaction_time)
1901 .is_none_or(|time| (start..=end).contains(&time))
1902 })
1903 .filter_map(convert_activity_to_trade)
1904 .collect();
1905
1906 Ok(TradesRead::new(trades, resume))
1907 }
1908}
1909
1910impl AlpacaClient {
1915 pub async fn open_order_with_intent(
1928 &self,
1929 request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
1930 intent: AlpacaPositionIntent,
1931 ) -> Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState> {
1932 self.open_order_inner(request, intent).await
1933 }
1934
1935 pub async fn open_bracket_order(
1968 &self,
1969 request: AlpacaBracketOrderRequest,
1970 ) -> AlpacaBracketOrderResult {
1971 let order_key = crate::order::OrderKey::new(
1972 ExchangeId::AlpacaBroker,
1973 request.instrument.clone(),
1974 request.strategy.clone(),
1975 request.cid.clone(),
1976 );
1977
1978 let tif_str = match request.time_in_force {
1980 TimeInForce::GoodUntilEndOfDay => "day",
1981 TimeInForce::GoodUntilCancelled { post_only } => {
1982 if post_only {
1983 return AlpacaBracketOrderResult {
1984 parent: Order {
1985 key: order_key,
1986 side: request.side,
1987 price: Some(request.entry_price),
1988 quantity: request.quantity,
1989 kind: OrderKind::Limit,
1990 time_in_force: request.time_in_force,
1991 state: OrderState::inactive(OrderError::Rejected(
1992 ApiError::OrderRejected(
1993 "Alpaca does not support post_only for bracket orders"
1994 .to_string(),
1995 ),
1996 )),
1997 },
1998 };
1999 }
2000 "gtc"
2001 }
2002 other => {
2003 return AlpacaBracketOrderResult {
2004 parent: Order {
2005 key: order_key,
2006 side: request.side,
2007 price: Some(request.entry_price),
2008 quantity: request.quantity,
2009 kind: OrderKind::Limit,
2010 time_in_force: request.time_in_force,
2011 state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
2012 format!(
2013 "Alpaca bracket orders only support Day or GTC time_in_force, got {:?}",
2014 other
2015 ),
2016 ))),
2017 },
2018 };
2019 }
2020 };
2021
2022 let price_ordering_ok = match request.side {
2025 Side::Buy => {
2026 request.stop_loss_price < request.entry_price
2027 && request.entry_price < request.take_profit_price
2028 }
2029 Side::Sell => {
2030 request.take_profit_price < request.entry_price
2031 && request.entry_price < request.stop_loss_price
2032 }
2033 };
2034 if !price_ordering_ok {
2035 return AlpacaBracketOrderResult {
2036 parent: Order {
2037 key: order_key,
2038 side: request.side,
2039 price: Some(request.entry_price),
2040 quantity: request.quantity,
2041 kind: OrderKind::Limit,
2042 time_in_force: request.time_in_force,
2043 state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
2044 format!(
2045 "Invalid bracket price ordering for {:?} side: entry={}, take_profit={}, stop_loss={}",
2046 request.side,
2047 request.entry_price,
2048 request.take_profit_price,
2049 request.stop_loss_price,
2050 ),
2051 ))),
2052 },
2053 };
2054 }
2055
2056 if let Some(sl_limit) = request.stop_loss_limit_price {
2060 let sl_limit_ok = match request.side {
2061 Side::Buy => sl_limit <= request.stop_loss_price,
2062 Side::Sell => sl_limit >= request.stop_loss_price,
2063 };
2064 if !sl_limit_ok {
2065 return AlpacaBracketOrderResult {
2066 parent: Order {
2067 key: order_key,
2068 side: request.side,
2069 price: Some(request.entry_price),
2070 quantity: request.quantity,
2071 kind: OrderKind::Limit,
2072 time_in_force: request.time_in_force,
2073 state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
2074 format!(
2075 "Invalid stop-loss limit price for {:?} bracket: \
2076 stop_loss_price={}, stop_loss_limit_price={}",
2077 request.side, request.stop_loss_price, sl_limit,
2078 ),
2079 ))),
2080 },
2081 };
2082 }
2083 }
2084
2085 let body = AlpacaOrderRequest {
2087 symbol: request.instrument.name().as_str(),
2088 qty: request.quantity.to_string(),
2089 side: map_side(request.side),
2090 order_type: "limit",
2091 time_in_force: tif_str,
2092 limit_price: Some(request.entry_price.to_string()),
2093 stop_price: None,
2094 trail_percent: None,
2095 trail_price: None,
2096 client_order_id: Some(request.cid.0.as_str()),
2097 position_intent: if is_options_or_equity_symbol(request.instrument.name().as_str()) {
2098 Some(map_position_intent(request.side, false))
2099 } else {
2100 None
2101 },
2102 order_class: Some("bracket"),
2103 take_profit: Some(TakeProfitParams {
2104 limit_price: request.take_profit_price.to_string(),
2105 }),
2106 stop_loss: Some(StopLossParams {
2107 stop_price: request.stop_loss_price.to_string(),
2108 limit_price: request.stop_loss_limit_price.map(|p| p.to_string()),
2109 }),
2110 };
2111
2112 let http = self.http.clone();
2113 let rl = &self.rate_limiter;
2114
2115 let result: Result<AlpacaOrderResponse, UnindexedClientError> =
2116 rest_with_retry(rl, || http.post(&self.orders_url).json(&body)).await;
2117
2118 match result {
2119 Ok(resp) => {
2120 let state = placed_order_state(&resp, &order_key.instrument, request.quantity);
2121 self.known_live.lock().placed(
2122 &order_key,
2123 request.quantity,
2124 OrderKind::Limit,
2125 &state,
2126 );
2127
2128 AlpacaBracketOrderResult {
2129 parent: Order {
2130 key: order_key,
2131 side: request.side,
2132 price: Some(request.entry_price),
2133 quantity: request.quantity,
2134 kind: OrderKind::Limit,
2135 time_in_force: request.time_in_force,
2136 state,
2137 },
2138 }
2139 }
2140 Err(e) => {
2141 let order_err = order_post_error(e);
2142 AlpacaBracketOrderResult {
2143 parent: Order {
2144 key: order_key,
2145 side: request.side,
2146 price: Some(request.entry_price),
2147 quantity: request.quantity,
2148 kind: OrderKind::Limit,
2149 time_in_force: request.time_in_force,
2150 state: OrderState::inactive(order_err),
2151 },
2152 }
2153 }
2154 }
2155 }
2156
2157 async fn open_order_inner(
2158 &self,
2159 request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
2160 intent: AlpacaPositionIntent,
2161 ) -> Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState> {
2162 let instrument = request.key.instrument.clone();
2163 let side = request.state.side;
2164 let price = request.state.price;
2165 let quantity = request.state.quantity;
2166 let kind = request.state.kind;
2167 let time_in_force = request.state.time_in_force;
2168 let cid = request.key.cid.clone();
2169
2170 let order_key = crate::order::OrderKey::new(
2171 ExchangeId::AlpacaBroker,
2172 instrument.clone(),
2173 request.key.strategy.clone(),
2174 cid.clone(),
2175 );
2176
2177 let tif_str = match map_time_in_force(time_in_force) {
2179 Ok(s) => s,
2180 Err(msg) => {
2181 return Order {
2182 key: order_key,
2183 side,
2184 price,
2185 quantity,
2186 kind,
2187 time_in_force,
2188 state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
2189 msg.to_string(),
2190 ))),
2191 };
2192 }
2193 };
2194
2195 let order_type_str = match map_order_kind(kind) {
2197 Some(s) => s,
2198 None => {
2199 return Order {
2200 key: order_key,
2201 side,
2202 price,
2203 quantity,
2204 kind,
2205 time_in_force,
2206 state: OrderState::inactive(OrderError::UnsupportedOrderType(format!(
2207 "Alpaca connector does not yet support OrderKind::{kind:?}"
2208 ))),
2209 };
2210 }
2211 };
2212
2213 if let OrderKind::TrailingStop {
2215 offset_type: TrailingOffsetType::BasisPoints,
2216 ..
2217 } = kind
2218 {
2219 return Order {
2220 key: order_key,
2221 side,
2222 price,
2223 quantity,
2224 kind,
2225 time_in_force,
2226 state: OrderState::inactive(OrderError::UnsupportedOrderType(
2227 "Alpaca does not support TrailingOffsetType::BasisPoints; \
2228 use Percentage or Absolute"
2229 .to_string(),
2230 )),
2231 };
2232 }
2233
2234 if matches!(kind, OrderKind::StopLimit { .. }) && price.is_none() {
2237 return Order {
2238 key: order_key,
2239 side,
2240 price,
2241 quantity,
2242 kind,
2243 time_in_force,
2244 state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
2245 "StopLimit order requires Order.price (the limit price) to be set".to_string(),
2246 ))),
2247 };
2248 }
2249
2250 let (stop_price, trail_percent, trail_price) = match kind {
2252 OrderKind::Stop { trigger_price } | OrderKind::StopLimit { trigger_price } => {
2253 (Some(trigger_price.to_string()), None, None)
2254 }
2255 OrderKind::TrailingStop {
2256 offset,
2257 offset_type,
2258 } => match offset_type {
2259 TrailingOffsetType::Percentage => (None, Some(offset.to_string()), None),
2260 TrailingOffsetType::Absolute => (None, None, Some(offset.to_string())),
2261 TrailingOffsetType::BasisPoints => unreachable!("validated above"),
2262 },
2263 OrderKind::Market
2265 | OrderKind::Limit
2266 | OrderKind::TakeProfit { .. }
2267 | OrderKind::TakeProfitLimit { .. }
2268 | OrderKind::TrailingStopLimit { .. } => (None, None, None),
2269 };
2270
2271 let body = AlpacaOrderRequest {
2272 symbol: instrument.name().as_str(),
2273 qty: quantity.to_string(),
2274 side: map_side(side),
2275 order_type: order_type_str,
2276 time_in_force: tif_str,
2277 limit_price: match kind {
2278 OrderKind::Limit | OrderKind::StopLimit { .. } => price.map(|p| p.to_string()),
2279 _ => None,
2280 },
2281 stop_price,
2282 trail_percent,
2283 trail_price,
2284 client_order_id: Some(cid.0.as_str()),
2285 position_intent: if is_options_or_equity_symbol(instrument.name().as_str()) {
2291 Some(intent)
2292 } else {
2293 None
2294 },
2295 order_class: None,
2297 take_profit: None,
2298 stop_loss: None,
2299 };
2300
2301 let http = self.http.clone();
2302 let rl = &self.rate_limiter;
2303
2304 let result: Result<AlpacaOrderResponse, UnindexedClientError> =
2305 rest_with_retry(rl, || http.post(&self.orders_url).json(&body)).await;
2306
2307 match result {
2308 Ok(resp) => {
2309 let state = placed_order_state(&resp, &order_key.instrument, quantity);
2310 self.known_live
2311 .lock()
2312 .placed(&order_key, quantity, kind, &state);
2313
2314 Order {
2315 key: order_key,
2316 side,
2317 price,
2318 quantity,
2319 kind,
2320 time_in_force,
2321 state,
2322 }
2323 }
2324 Err(e) => {
2325 let order_err = order_post_error(e);
2326 Order {
2327 key: order_key,
2328 side,
2329 price,
2330 quantity,
2331 kind,
2332 time_in_force,
2333 state: OrderState::inactive(order_err),
2334 }
2335 }
2336 }
2337 }
2338}
2339
2340const MAX_OPEN_ORDERS: usize = 500;
2348
2349async fn fetch_raw_open_orders(
2357 http: &reqwest::Client,
2358 rate_limiter: &RateLimitTracker,
2359 base: &str,
2360 instruments: &[InstrumentNameExchange],
2361) -> Result<Vec<AlpacaOrderResponse>, UnindexedClientError> {
2362 let orders: Vec<AlpacaOrderResponse> = if instruments.is_empty() {
2363 rest_with_retry(rate_limiter, || {
2364 http.get(format!("{base}/v2/orders"))
2365 .query(&[("status", "open"), ("limit", "500")])
2366 })
2367 .await?
2368 } else {
2369 let symbols = instruments.iter().map(|i| i.name().as_str()).join(",");
2370 rest_with_retry(rate_limiter, || {
2371 http.get(format!("{base}/v2/orders")).query(&[
2372 ("status", "open"),
2373 ("limit", "500"),
2374 ("symbols", &symbols),
2375 ])
2376 })
2377 .await?
2378 };
2379 if orders.len() == MAX_OPEN_ORDERS {
2380 warn!(
2381 limit = MAX_OPEN_ORDERS,
2382 "Alpaca fetch_raw_open_orders: received exactly {MAX_OPEN_ORDERS} results — \
2383 response is likely truncated"
2384 );
2385 return Err(UnindexedClientError::TruncatedSnapshot {
2386 limit: MAX_OPEN_ORDERS,
2387 });
2388 }
2389 Ok(orders)
2390}
2391
2392const MAX_ACTIVITY_PAGES: usize = 50;
2401
2402struct ActivityPage {
2404 activities: Vec<AlpacaActivity>,
2406 resume: Option<DateTime<Utc>>,
2409}
2410
2411fn floor_millis(time: DateTime<Utc>) -> DateTime<Utc> {
2413 DateTime::from_timestamp_millis(time.timestamp_millis()).unwrap_or(time)
2414}
2415
2416const PAGE_SIZE_STR: &str = "100"; const _: () = assert!(
2446 ALPACA_MAX_ACTIVITIES == 100,
2447 "PAGE_SIZE_STR must be updated to match ALPACA_MAX_ACTIVITIES",
2448);
2449
2450async fn paginate_activities(
2451 http: &reqwest::Client,
2452 rate_limiter: &RateLimitTracker,
2453 base: &str,
2454 start: DateTime<Utc>,
2455 end: Option<DateTime<Utc>>,
2456) -> Result<ActivityPage, UnindexedClientError> {
2457 let after =
2460 floor_millis(start.max(DateTime::UNIX_EPOCH)).to_rfc3339_opts(SecondsFormat::Millis, true);
2461 let until = end.filter(|end| *end < Utc::now()).map(|end| {
2462 (floor_millis(end) + TimeDelta::milliseconds(1))
2463 .to_rfc3339_opts(SecondsFormat::Millis, true)
2464 });
2465 let mut all = Vec::with_capacity(ALPACA_MAX_ACTIVITIES);
2466 let mut page_token: Option<String> = None;
2467 let mut pages = 0usize;
2468 let mut truncated = false;
2469
2470 loop {
2471 if pages >= MAX_ACTIVITY_PAGES {
2472 truncated = true;
2473 break;
2474 }
2475 pages += 1;
2476 let page_token_ref = page_token.as_deref();
2478 let activities: Vec<AlpacaActivity> = rest_with_retry(rate_limiter, || {
2479 let mut req = http.get(format!("{base}/v2/account/activities")).query(&[
2480 ("activity_type", "FILL"),
2481 ("after", after.as_str()),
2482 ("page_size", PAGE_SIZE_STR),
2483 ("direction", "asc"),
2484 ]);
2485 if let Some(until) = until.as_deref() {
2486 req = req.query(&[("until", until)]);
2487 }
2488 if let Some(token) = page_token_ref {
2489 req = req.query(&[("page_token", token)]);
2490 }
2491 req
2492 })
2493 .await?;
2494
2495 let page_len = activities.len();
2496 let page_token_candidate = activities.last().map(|a| a.id.clone());
2506 all.extend(activities);
2507
2508 if page_len < ALPACA_MAX_ACTIVITIES {
2509 break;
2510 }
2511 match page_token_candidate {
2512 Some(token) if !token.is_empty() => {
2513 debug!("Alpaca paginate_activities: fetching next page ({page_len} results)");
2514 page_token = Some(token);
2515 }
2516 _ => break,
2520 }
2521 }
2522
2523 let resume = truncated.then(|| {
2526 floor_millis(
2527 all.iter()
2528 .rev()
2529 .find_map(|activity| parse_timestamp(&activity.transaction_time))
2530 .unwrap_or(start),
2531 )
2532 });
2533 Ok(ActivityPage {
2534 activities: all,
2535 resume,
2536 })
2537}
2538
2539#[allow(clippy::cognitive_complexity)] async fn connection_manager(
2555 tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
2556 dedup: SharedDedupCache,
2557 config: Arc<AlpacaConfig>,
2558 http: reqwest::Client,
2559 rate_limiter: Arc<RateLimitTracker>,
2560 known: SharedKnownLiveOrders,
2561 instruments: Vec<InstrumentNameExchange>,
2562 initial_ws: Option<WebSocket>,
2563) {
2564 let mut backoff = ExponentialBackoff::new();
2565 let mut disconnect_time: Option<DateTime<Utc>> = None;
2566 let mut unchecked = UncheckedOrders::default();
2569 let mut current_ws = initial_ws;
2570
2571 'outer: loop {
2572 let mut ws = match current_ws.take() {
2574 Some(ws) => ws,
2575 None => match connect_and_subscribe(&config).await {
2576 Ok(ws) => ws,
2577 Err(e) => {
2578 error!(%e, "Alpaca WS connect/subscribe failed");
2579 if !backoff.wait().await {
2580 error!("Alpaca max reconnect attempts exhausted");
2581 emit_stream_terminated(
2583 &tx,
2584 ExchangeId::AlpacaBroker,
2585 StreamTerminationReason::ReconnectBudgetExhausted {
2586 attempts: backoff.attempts(),
2587 last_error: e.to_string(),
2588 },
2589 );
2590 break;
2591 }
2592 continue;
2593 }
2594 },
2595 };
2596 info!("Alpaca account_stream connected and subscribed");
2597 backoff.reset();
2608
2609 if let Some(dt) = disconnect_time.take() {
2614 recover_fills_or_report(
2615 &http,
2616 &rate_limiter,
2617 &instruments,
2618 config.rest_base_url(),
2619 dt,
2620 Duration::from_secs(FILL_RECOVERY_TIMEOUT_SECS),
2621 &tx,
2622 &dedup,
2623 &known,
2624 )
2625 .await;
2626 let held = {
2628 let known = known.lock();
2629 if instruments.is_empty() {
2630 known.instruments()
2631 } else {
2632 known.instruments_among(&instruments)
2633 }
2634 };
2635 unchecked.open(held);
2636 }
2637
2638 let mut last_message_time = Utc::now();
2651 let heartbeat = tokio::time::sleep(Duration::from_secs(HEARTBEAT_TIMEOUT_SECS));
2652 tokio::pin!(heartbeat);
2653
2654 let order_checks =
2659 run_order_checks(&http, &rate_limiter, &config, &known, &mut unchecked, &tx);
2660 tokio::pin!(order_checks);
2661
2662 macro_rules! reset_heartbeat {
2667 () => {
2668 heartbeat.as_mut().reset(
2669 tokio::time::Instant::now() + Duration::from_secs(HEARTBEAT_TIMEOUT_SECS),
2670 );
2671 last_message_time = Utc::now();
2672 };
2673 }
2674 enum DisconnectReason {
2677 ServerClose,
2678 Error(String),
2679 StreamEnded,
2680 HeartbeatTimeout,
2681 }
2682 let reason = loop {
2683 tokio::select! {
2684 msg = ws.next() => {
2685 match msg {
2686 Some(Ok(WsMessage::Ping(_))) => {
2687 reset_heartbeat!();
2690 }
2691 Some(Ok(WsMessage::Text(text))) => {
2692 process_ws_text(text.as_str(), &tx, &dedup, &known, &mut backoff);
2693 reset_heartbeat!();
2694 }
2695 Some(Ok(WsMessage::Binary(bytes))) => {
2696 match std::str::from_utf8(&bytes) {
2698 Ok(text) => {
2699 process_ws_text(text, &tx, &dedup, &known, &mut backoff);
2700 reset_heartbeat!();
2704 }
2705 Err(e) => warn!(%e, "Alpaca WS binary frame: not valid UTF-8"),
2706 }
2707 }
2708 Some(Ok(WsMessage::Close(frame))) => {
2709 warn!(frame = ?frame, "Alpaca WS closed by server");
2710 break DisconnectReason::ServerClose;
2711 }
2712 Some(Ok(_)) => {} Some(Err(e)) => {
2714 warn!(%e, "Alpaca WS error, reconnecting");
2715 break DisconnectReason::Error(e.to_string());
2716 }
2717 None => {
2718 warn!("Alpaca WS stream ended, reconnecting");
2719 break DisconnectReason::StreamEnded;
2720 }
2721 }
2722 }
2723 () = &mut order_checks => {}
2724 _ = &mut heartbeat => {
2725 warn!(
2726 timeout_secs = HEARTBEAT_TIMEOUT_SECS,
2727 "Alpaca heartbeat timeout, reconnecting"
2728 );
2729 break DisconnectReason::HeartbeatTimeout;
2732 }
2733 _ = tx.closed() => {
2734 debug!("Alpaca account_stream consumer dropped, terminating");
2735 let _ = tokio::time::timeout(
2736 Duration::from_secs(WS_CLOSE_TIMEOUT_SECS),
2737 ws.close(None),
2738 ).await;
2739 break 'outer;
2742 }
2743 }
2744 };
2745
2746 disconnect_time =
2751 Some(last_message_time - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS));
2752
2753 let _ =
2755 tokio::time::timeout(Duration::from_secs(WS_CLOSE_TIMEOUT_SECS), ws.close(None)).await;
2756
2757 if tx.is_closed() {
2758 break;
2761 }
2762 if !backoff.wait().await {
2763 error!("Alpaca max reconnect attempts exhausted, stream terminating");
2764 let last_error = match reason {
2767 DisconnectReason::HeartbeatTimeout => {
2768 format!("heartbeat timeout ({HEARTBEAT_TIMEOUT_SECS}s)")
2769 }
2770 DisconnectReason::ServerClose => "WebSocket closed by server".to_string(),
2771 DisconnectReason::Error(e) => e,
2772 DisconnectReason::StreamEnded => "WebSocket stream ended".to_string(),
2773 };
2774 emit_stream_terminated(
2775 &tx,
2776 ExchangeId::AlpacaBroker,
2777 StreamTerminationReason::ReconnectBudgetExhausted {
2778 attempts: backoff.attempts(),
2779 last_error,
2780 },
2781 );
2782 break;
2783 }
2784 }
2785}
2786
2787#[derive(Debug)]
2792enum HandshakeError {
2793 Transport(String),
2796 Auth(String),
2799}
2800
2801async fn connect_and_subscribe(config: &AlpacaConfig) -> Result<WebSocket, UnindexedClientError> {
2805 let url = config.ws_url();
2806 debug!(%url, "Alpaca: connecting to WebSocket");
2807
2808 let mut ws = rustrade_integration::protocol::websocket::connect(url)
2809 .await
2810 .map_err(|e| {
2811 UnindexedClientError::Connectivity(ConnectivityError::Socket(format!(
2812 "WS connect: {e}"
2813 )))
2814 })?;
2815
2816 let result = tokio::time::timeout(
2818 Duration::from_secs(WS_HANDSHAKE_TIMEOUT_SECS),
2819 ws_handshake(&mut ws, config),
2820 )
2821 .await;
2822
2823 match result {
2824 Ok(Ok(())) => Ok(ws),
2825 Ok(Err(HandshakeError::Transport(e))) => {
2826 let _ = ws.close(None).await;
2827 Err(UnindexedClientError::Connectivity(
2828 ConnectivityError::Socket(e),
2829 ))
2830 }
2831 Ok(Err(HandshakeError::Auth(e))) => {
2832 let _ = ws.close(None).await;
2833 Err(UnindexedClientError::Api(ApiError::Unauthenticated(e)))
2834 }
2835 Err(_) => {
2836 let _ = ws.close(None).await;
2837 Err(UnindexedClientError::Connectivity(
2838 ConnectivityError::Timeout,
2839 ))
2840 }
2841 }
2842}
2843
2844async fn ws_handshake(ws: &mut WebSocket, config: &AlpacaConfig) -> Result<(), HandshakeError> {
2852 let auth = serde_json::json!({
2854 "action": "auth",
2855 "key": config.api_key(),
2856 "secret": config.secret_key,
2859 })
2860 .to_string();
2861 ws.send(WsMessage::Text(auth.into()))
2862 .await
2863 .map_err(|e| HandshakeError::Transport(format!("WS auth send: {e}")))?;
2864
2865 loop {
2867 match ws.next().await {
2868 Some(Ok(WsMessage::Text(text))) => {
2869 if let Some(result) = check_auth_response(text.as_str()) {
2870 result?;
2871 break;
2872 }
2873 }
2874 Some(Ok(WsMessage::Binary(bytes))) => {
2875 if let Ok(text) = std::str::from_utf8(&bytes)
2876 && let Some(result) = check_auth_response(text)
2877 {
2878 result?;
2879 break;
2880 }
2881 }
2882 Some(Err(e)) => {
2883 return Err(HandshakeError::Transport(format!(
2884 "WS error during auth: {e}"
2885 )));
2886 }
2887 None => {
2888 return Err(HandshakeError::Transport(
2889 "WS closed before auth response".into(),
2890 ));
2891 }
2892 _ => {} }
2894 }
2895
2896 let sub = serde_json::json!({
2898 "action": "listen",
2899 "data": { "streams": ["trade_updates"] }
2900 })
2901 .to_string();
2902 ws.send(WsMessage::Text(sub.into()))
2903 .await
2904 .map_err(|e| HandshakeError::Transport(format!("WS subscribe send: {e}")))?;
2905
2906 loop {
2908 match ws.next().await {
2909 Some(Ok(WsMessage::Text(text))) => {
2910 if check_listen_ack(text.as_str()) {
2911 break;
2912 }
2913 if let Ok(msg) = serde_json::from_str::<AlpacaStreamMessage<'_>>(text.as_str()) {
2920 if msg.stream == "trade_updates" {
2921 warn!(stream = %msg.stream, "WS trade_updates event dropped during listen-ack handshake — on a reconnect, fills and how orders held as live ended are recovered via REST; other lifecycle events are lost");
2922 } else {
2923 trace!(stream = %msg.stream, "WS message dropped during listen-ack handshake");
2924 }
2925 } else {
2926 trace!(
2927 bytes = text.len(),
2928 "WS non-stream message dropped during listen-ack handshake"
2929 );
2930 }
2931 }
2932 Some(Ok(WsMessage::Binary(bytes))) => {
2933 if let Ok(text) = std::str::from_utf8(&bytes)
2934 && check_listen_ack(text)
2935 {
2936 break;
2937 }
2938 trace!(
2939 bytes = bytes.len(),
2940 "WS binary message dropped during listen-ack handshake"
2941 );
2942 }
2943 Some(Err(e)) => {
2944 return Err(HandshakeError::Transport(format!(
2945 "WS error during subscribe: {e}"
2946 )));
2947 }
2948 None => {
2949 return Err(HandshakeError::Transport(
2950 "WS closed before subscribe ack".into(),
2951 ));
2952 }
2953 _ => {}
2954 }
2955 }
2956
2957 info!("Alpaca WS authenticated and subscribed to trade_updates");
2958 Ok(())
2959}
2960
2961fn check_auth_response(text: &str) -> Option<Result<(), HandshakeError>> {
2967 let msg = serde_json::from_str::<AlpacaStreamMessage<'_>>(text).ok()?;
2968 if msg.stream != "authorization" {
2969 return None;
2970 }
2971 #[derive(Deserialize)]
2972 struct AuthData<'a> {
2973 status: &'a str,
2974 }
2975 let data = serde_json::from_str::<AuthData<'_>>(msg.data.get()).ok()?;
2976 if data.status == "authorized" {
2977 Some(Ok(()))
2978 } else {
2979 Some(Err(HandshakeError::Auth(format!(
2980 "Alpaca WS auth failed: status={}",
2981 data.status
2982 ))))
2983 }
2984}
2985
2986fn check_listen_ack(text: &str) -> bool {
2990 let Ok(msg) = serde_json::from_str::<AlpacaStreamMessage<'_>>(text) else {
2991 return false;
2992 };
2993 if msg.stream != "listening" {
2994 return false;
2995 }
2996 #[derive(Deserialize)]
2997 struct ListenData<'a> {
2998 #[serde(borrow)]
2999 streams: Vec<&'a str>,
3000 }
3001 let Ok(data) = serde_json::from_str::<ListenData<'_>>(msg.data.get()) else {
3002 return false;
3003 };
3004 data.streams.contains(&"trade_updates")
3005}
3006
3007fn process_ws_text(
3017 text: &str,
3018 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
3019 dedup: &SharedDedupCache,
3020 known: &SharedKnownLiveOrders,
3021 backoff: &mut ExponentialBackoff,
3022) {
3023 let msg: AlpacaStreamMessage<'_> = match serde_json::from_str(text) {
3024 Ok(m) => m,
3025 Err(e) => {
3026 trace!(
3027 %e,
3028 raw = ?&text[..text.len().min(200)],
3029 "Alpaca WS: skipped non-JSON message"
3030 );
3031 return;
3032 }
3033 };
3034
3035 backoff.reset();
3038
3039 match msg.stream.as_str() {
3040 "trade_updates" => {
3041 let update: AlpacaTradeUpdate<'_> = match serde_json::from_str(msg.data.get()) {
3042 Ok(u) => u,
3043 Err(e) => {
3044 warn!(
3049 %e,
3050 raw = ?&msg.data.get()[..msg.data.get().len().min(200)],
3051 "Alpaca WS trade_updates: failed to deserialize event — event dropped"
3052 );
3053 return;
3054 }
3055 };
3056
3057 if is_fill_event(&update)
3061 && is_duplicate_fill(dedup, ws_execution_id(&update), early_dedup_key(&update))
3062 {
3063 trace!("Alpaca WS: skipping duplicate fill event (early check)");
3064 return;
3065 }
3066
3067 if is_trade_amendment(&update) {
3070 warn!(
3071 event = %update.event,
3072 raw = %msg.data.get(),
3073 "Alpaca WS: a fill reported earlier was amended — reporting it as TradeAmended"
3074 );
3075 }
3076
3077 for event in convert_trade_update(update).into_iter().flatten() {
3078 if let AccountEventKind::TradeAmended(amendment) = &event.kind
3082 && let TradeAmendmentKind::Corrected { replacement } = &amendment.kind
3083 {
3084 record_execution(dedup, &replacement.id.0);
3085 }
3086 let known = KnownLiveOrders::observes(&event.kind).then(|| {
3087 let mut known = known.lock();
3088 known.observe(&event);
3089 known
3090 });
3091 let _ = tx.send(event);
3094 drop(known);
3095 }
3096 }
3097 "authorization" | "listening" => {
3098 trace!(stream = %msg.stream, "Alpaca WS: auth/listen ack received during stream");
3101 }
3102 other => {
3103 trace!(%other, "Alpaca WS: ignoring unknown stream type");
3104 }
3105 }
3106}
3107
3108fn fill_dedup_key(order_id: &str, cumulative: Decimal) -> SmolStr {
3119 format_smolstr!("{}:{}", order_id, cumulative.normalize())
3120}
3121
3122fn is_trade_amendment(update: &AlpacaTradeUpdate<'_>) -> bool {
3124 matches!(update.event.as_str(), "trade_bust" | "trade_correct")
3125}
3126
3127#[inline]
3131fn is_fill_event(update: &AlpacaTradeUpdate<'_>) -> bool {
3132 matches!(update.event.as_str(), "fill" | "partial_fill")
3133}
3134
3135fn ws_execution_id<'a>(update: &AlpacaTradeUpdate<'a>) -> Option<&'a str> {
3137 update.execution_id.filter(|id| !id.is_empty())
3138}
3139
3140fn early_dedup_key(update: &AlpacaTradeUpdate<'_>) -> SmolStr {
3144 let filled_qty = update.order.filled_qty.unwrap_or("0");
3145 let qty = Decimal::from_str(filled_qty).unwrap_or(Decimal::ZERO);
3146 fill_dedup_key(&update.order.id, qty)
3147}
3148
3149fn activity_execution_id(activity_id: &str) -> &str {
3153 activity_execution_id_part(activity_id).unwrap_or(activity_id)
3154}
3155
3156fn activity_execution_id_part(activity_id: &str) -> Option<&str> {
3160 activity_id
3161 .split_once("::")
3162 .map(|(_, execution_id)| execution_id)
3163 .filter(|execution_id| !execution_id.is_empty())
3164}
3165
3166#[derive(Debug, PartialEq)]
3172struct UnreadFills {
3173 start: DateTime<Utc>,
3174 reason: FillRecoveryFailure,
3175}
3176
3177async fn recover_fills_or_report(
3181 http: &reqwest::Client,
3182 rate_limiter: &RateLimitTracker,
3183 instruments: &[InstrumentNameExchange],
3184 base: &str,
3185 disconnect: DateTime<Utc>,
3186 timeout: Duration,
3187 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
3188 dedup: &SharedDedupCache,
3189 known: &SharedKnownLiveOrders,
3190) {
3191 let recovery_start = Utc::now();
3193 let outcome = match tokio::time::timeout(
3194 timeout,
3195 recover_fills(
3196 http,
3197 rate_limiter,
3198 instruments,
3199 base,
3200 disconnect,
3201 tx,
3202 dedup,
3203 known,
3204 ),
3205 )
3206 .await
3207 {
3208 Ok(outcome) => outcome,
3209 Err(_) => {
3213 warn!(
3214 timeout_secs = timeout.as_secs(),
3215 instruments = ?instruments
3216 .iter()
3217 .map(|instrument| instrument.name().as_str())
3218 .collect::<Vec<_>>(),
3219 "Alpaca fill recovery timed out — fills since the disconnect may be missing for \
3220 any of the instruments (every instrument when the list is empty)"
3221 );
3222 Err(UnreadFills {
3223 start: disconnect,
3224 reason: FillRecoveryFailure::TimedOut {
3225 timeout_secs: timeout.as_secs(),
3226 },
3227 })
3228 }
3229 };
3230 if let Err(unread) = outcome
3231 && tx
3232 .send(fill_recovery_gave_up(instruments, unread, recovery_start))
3233 .is_err()
3234 {
3235 debug!("Alpaca fill recovery: consumer dropped before the give-up was sent");
3236 }
3237}
3238
3239fn fill_recovery_gave_up(
3248 instruments: &[InstrumentNameExchange],
3249 unread: UnreadFills,
3250 end: DateTime<Utc>,
3251) -> UnindexedAccountEvent {
3252 let UnreadFills { start, reason } = unread;
3253 let scope = if instruments.is_empty() {
3255 FillRecoveryScope::AllInstruments
3256 } else {
3257 FillRecoveryScope::Instruments(instruments.to_vec())
3258 };
3259 UnindexedAccountEvent::new(
3260 ExchangeId::AlpacaBroker,
3261 AccountEventKind::FillRecoveryGaveUp(FillRecoveryGap::new(
3262 scope,
3263 start,
3264 end.max(start),
3265 1,
3266 reason,
3267 )),
3268 )
3269}
3270
3271async fn recover_fills(
3280 http: &reqwest::Client,
3281 rate_limiter: &RateLimitTracker,
3282 instruments: &[InstrumentNameExchange],
3283 base: &str,
3284 after: DateTime<Utc>,
3285 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
3286 dedup: &SharedDedupCache,
3287 known: &SharedKnownLiveOrders,
3288) -> Result<(), UnreadFills> {
3289 info!(%after, instruments = instruments.len(), "Alpaca recovering fills after reconnect");
3292 let instrument_set: fnv::FnvHashSet<&str> = if instruments.is_empty() {
3293 fnv::FnvHashSet::default()
3294 } else {
3295 instruments.iter().map(|i| i.name().as_str()).collect()
3296 };
3297
3298 let page = match paginate_activities(http, rate_limiter, base, after, None).await {
3299 Ok(p) => p,
3300 Err(e) => {
3301 error!(%e, "Alpaca fill recovery: REST request failed");
3302 return Err(UnreadFills {
3303 start: after,
3304 reason: FillRecoveryFailure::Request(e.to_string()),
3305 });
3306 }
3307 };
3308
3309 let activities = page.activities;
3310
3311 let unread = page.resume.map(|start| {
3315 error!(
3316 max_pages = MAX_ACTIVITY_PAGES,
3317 fills_read = activities.len(),
3318 %start,
3319 "Alpaca fill recovery: max page limit reached, truncating — fills from this time on \
3320 are not delivered; read them with fetch_trades"
3321 );
3322 UnreadFills {
3323 start,
3324 reason: FillRecoveryFailure::Truncated {
3325 fills_read: activities.len(),
3326 },
3327 }
3328 });
3329
3330 let mut recovered = 0u32;
3331 let mut duplicates = 0u32;
3332
3333 let mut cumulative_qty: FnvHashMap<&str, Decimal> = FnvHashMap::default();
3350
3351 for activity in &activities {
3352 if !instrument_set.is_empty() && !instrument_set.contains(activity.symbol.as_str()) {
3353 continue;
3357 }
3358
3359 let exec_qty = Decimal::from_str(&activity.qty).unwrap_or(Decimal::ZERO);
3374 let cum = cumulative_qty
3375 .entry(activity.order_id.as_str())
3376 .or_default();
3377 *cum += exec_qty;
3378 let cumulative = *cum;
3379
3380 let trade = match convert_activity_to_trade(activity) {
3381 Some(t) => t,
3382 None => {
3383 warn!(id = %activity.id, symbol = %activity.symbol, "Alpaca: skipping activity with unparseable fields");
3384 continue; }
3386 };
3387
3388 let key = fill_dedup_key(
3398 &activity.order_id,
3399 trade.order_filled_quantity.unwrap_or(cumulative),
3400 );
3401 if is_duplicate_fill(dedup, activity_execution_id_part(&activity.id), key) {
3402 duplicates += 1;
3403 continue;
3404 }
3405
3406 let event =
3407 UnindexedAccountEvent::new(ExchangeId::AlpacaBroker, AccountEventKind::Trade(trade));
3408 let mut held = known.lock();
3409 held.observe(&event);
3410 let sent = tx.send(event);
3411 drop(held);
3412 if sent.is_err() {
3413 debug!("Alpaca fill recovery: consumer dropped during recovery");
3414 return Ok(());
3415 }
3416 recovered += 1;
3417 }
3418
3419 info!(recovered, duplicates, "Alpaca fill recovery complete");
3420 unread.map_or(Ok(()), Err)
3421}
3422
3423async fn fetch_order_lookup(
3432 http: reqwest::Client,
3433 rate_limiter: Arc<RateLimitTracker>,
3434 config: Arc<AlpacaConfig>,
3435 key: UnindexedOrderKey,
3436) -> Result<OrderLookup, UnindexedClientError> {
3437 let url = format!("{}/v2/orders:by_client_order_id", config.rest_base_url());
3438 let cid = key.cid.0.as_str();
3439 let found: Option<AlpacaOrderResponse> = rest_lookup_with_retry(&rate_limiter, || {
3440 http.get(&url).query(&[("client_order_id", cid)])
3441 })
3442 .await?;
3443 let Some(order) = found else {
3444 debug!(instrument = %key.instrument, cid = %key.cid, "Alpaca does not know this order");
3445 return Ok(OrderLookup::Unknown);
3446 };
3447 if !order
3448 .symbol
3449 .eq_ignore_ascii_case(key.instrument.name().as_str())
3450 {
3451 debug!(
3452 instrument = %key.instrument,
3453 cid = %key.cid,
3454 symbol = %order.symbol,
3455 "Alpaca knows this client order id on another symbol"
3456 );
3457 return Ok(OrderLookup::Unknown);
3458 }
3459 Ok(
3460 convert_ended_order(&order, &key).map_or(OrderLookup::NotEnded, |order| {
3461 OrderLookup::Ended(Box::new(order))
3462 }),
3463 )
3464}
3465
3466async fn listed_open_cids(
3469 http: reqwest::Client,
3470 rate_limiter: Arc<RateLimitTracker>,
3471 config: Arc<AlpacaConfig>,
3472 instruments: Vec<InstrumentNameExchange>,
3473) -> Result<FnvHashSet<ClientOrderId>, UnindexedClientError> {
3474 let orders =
3475 fetch_raw_open_orders(&http, &rate_limiter, config.rest_base_url(), &instruments).await?;
3476 Ok(orders
3477 .iter()
3478 .map(|order| alpaca_cid(order.client_order_id.as_deref(), &order.id))
3479 .collect())
3480}
3481
3482async fn recover_alpaca_ended_orders(
3486 http: &reqwest::Client,
3487 rate_limiter: &Arc<RateLimitTracker>,
3488 config: &Arc<AlpacaConfig>,
3489 known: &SharedKnownLiveOrders,
3490 unchecked: &mut UncheckedOrders,
3491 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
3492) {
3493 recover_ended_orders(
3494 ExchangeId::AlpacaBroker,
3495 known,
3496 unchecked,
3497 &NoPendingFills,
3498 tx,
3499 OpenListing::Batched,
3500 |instruments| {
3501 listed_open_cids(
3502 http.clone(),
3503 rate_limiter.clone(),
3504 config.clone(),
3505 instruments,
3506 )
3507 },
3508 |key| fetch_order_lookup(http.clone(), rate_limiter.clone(), config.clone(), key),
3509 )
3510 .await;
3511}
3512
3513async fn run_order_checks(
3518 http: &reqwest::Client,
3519 rate_limiter: &Arc<RateLimitTracker>,
3520 config: &Arc<AlpacaConfig>,
3521 known: &SharedKnownLiveOrders,
3522 unchecked: &mut UncheckedOrders,
3523 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
3524) {
3525 loop {
3526 if tx.is_closed() {
3528 return std::future::pending().await;
3529 }
3530 match unchecked.next_due(&NoPendingFills) {
3531 Some(due) => tokio::time::sleep_until(due).await,
3532 None => return std::future::pending().await,
3533 }
3534 recover_alpaca_ended_orders(http, rate_limiter, config, known, unchecked, tx).await;
3535 }
3536}
3537
3538fn convert_ended_order(
3546 o: &AlpacaOrderResponse,
3547 key: &UnindexedOrderKey,
3548) -> Option<UnindexedInactiveOrder> {
3549 let instrument = &key.instrument;
3550 let Some(status) = o.status.as_deref() else {
3551 warn!(%instrument, cid = %key.cid, order_id = %o.id, "Alpaca order missing status");
3552 return None;
3553 };
3554 if order_status_is_live(status) {
3555 return None;
3556 }
3557 let Some(order) = convert_representable_open_order(o) else {
3558 warn!(%instrument, cid = %key.cid, order_id = %o.id, status, "Alpaca order cannot be represented, treating it as not ended");
3559 return None;
3560 };
3561 let filled_qty = Decimal::from_str(&o.filled_qty).ok();
3564 let Some(state) = ended_order_state(
3565 status,
3566 OrderId(SmolStr::new(&o.id)),
3567 order.state.time_exchange,
3568 order.quantity,
3569 filled_qty,
3570 o.filled_avg_price.as_deref(),
3571 ) else {
3572 warn!(%instrument, cid = %key.cid, order_id = %o.id, status, "Alpaca order has a status this version does not know, treating it as not ended");
3573 return None;
3574 };
3575 let mut order = order.map_state(|_| state);
3576 order.key = key.clone();
3577 Some(order)
3578}
3579
3580fn order_status_is_live(status: &str) -> bool {
3586 matches!(
3587 status,
3588 "new"
3589 | "partially_filled"
3590 | "done_for_day"
3591 | "pending_cancel"
3592 | "pending_replace"
3593 | "accepted"
3594 | "pending_new"
3595 | "accepted_for_bidding"
3596 | "stopped"
3597 | "suspended"
3598 | "calculated"
3599 | "held"
3600 )
3601}
3602
3603fn ended_order_state<AssetKey, InstrumentKey>(
3613 status: &str,
3614 order_id: OrderId,
3615 time_exchange: DateTime<Utc>,
3616 quantity: Decimal,
3617 filled_qty: Option<Decimal>,
3618 filled_avg_price: Option<&str>,
3619) -> Option<InactiveOrderState<AssetKey, InstrumentKey>> {
3620 Some(match status {
3621 "filled" => InactiveOrderState::FullyFilled(Filled::new(
3623 order_id,
3624 time_exchange,
3625 filled_qty.unwrap_or(quantity),
3626 alpaca_avg_price(filled_avg_price),
3627 )),
3628 "canceled" | "replaced" => {
3629 InactiveOrderState::Cancelled(Cancelled::new(order_id, time_exchange, filled_qty))
3630 }
3631 "expired" => InactiveOrderState::Expired(Expired::new(order_id, time_exchange, filled_qty)),
3632 "rejected" => {
3633 InactiveOrderState::OpenFailed(OrderError::Rejected(ApiError::OrderRejected(format!(
3634 "Alpaca rejected order {order_id} after accepting it"
3635 ))))
3636 }
3637 _ => return None,
3638 })
3639}
3640
3641fn placed_order_state(
3651 resp: &AlpacaOrderResponse,
3652 instrument: &InstrumentNameExchange,
3653 quantity: Decimal,
3654) -> UnindexedOrderState {
3655 let order_id = OrderId(SmolStr::new(&resp.id));
3656 let time_exchange = order_state_time(resp);
3657 let filled_qty = alpaca_filled_qty(&resp.id, Some(&resp.filled_qty));
3658 let filled_avg_price = resp.filled_avg_price.as_deref();
3659 match resp.status.as_deref() {
3660 Some(status) => {
3661 if let Some(ended) = ended_order_state(
3662 status,
3663 order_id.clone(),
3664 time_exchange,
3665 quantity,
3666 filled_qty,
3667 filled_avg_price,
3668 ) {
3669 return OrderState::Inactive(ended);
3670 }
3671 if !order_status_is_live(status) {
3672 warn!(%instrument, %order_id, status, "Alpaca placed an order with a status this version does not know, reading it from what filled");
3673 }
3674 }
3675 None => {
3676 warn!(%instrument, %order_id, "Alpaca placed an order without a status, reading it from what filled");
3677 }
3678 }
3679 match filled_qty {
3680 Some(filled_qty) if filled_qty >= quantity => OrderState::fully_filled(Filled::new(
3681 order_id,
3682 time_exchange,
3683 filled_qty,
3684 alpaca_avg_price(filled_avg_price),
3685 )),
3686 _ => OrderState::active(Open::new(
3689 VenueOrderId::Assigned(order_id),
3690 time_exchange,
3691 filled_qty.unwrap_or(Decimal::ZERO),
3692 )),
3693 }
3694}
3695
3696fn alpaca_avg_price(filled_avg_price: Option<&str>) -> Option<Decimal> {
3699 filled_avg_price.and_then(|price| Decimal::from_str(price).ok())
3700}
3701
3702fn is_usd(asset: &AssetNameExchange) -> bool {
3709 asset.name().as_str().eq_ignore_ascii_case(USD)
3710}
3711
3712fn convert_account_to_balances(
3730 account: &AlpacaAccount,
3731 assets: &[AssetNameExchange],
3732) -> Vec<AssetBalance<AssetNameExchange>> {
3733 if !assets.is_empty() && !assets.iter().any(is_usd) {
3734 return Vec::new();
3735 }
3736
3737 vec![AssetBalance::new(
3738 AssetNameExchange::new(USD),
3739 Balance::new(
3740 account.cash,
3741 account.cash.min(account.non_marginable_buying_power),
3742 ),
3743 Utc::now(),
3744 )]
3745}
3746
3747fn convert_positions_to_balances(
3759 positions: &[AlpacaPosition],
3760 assets: &[AssetNameExchange],
3761) -> Vec<AssetBalance<AssetNameExchange>> {
3762 let now = Utc::now();
3763 positions
3764 .iter()
3765 .filter(|p| is_crypto_position(p))
3766 .filter_map(|p| {
3767 let base = crypto_position_asset(&p.symbol);
3768
3769 if !assets.is_empty()
3771 && !assets
3772 .iter()
3773 .any(|a| a.name().as_str().eq_ignore_ascii_case(base))
3774 {
3775 return None;
3776 }
3777
3778 let asset_name = AssetNameExchange::new(base);
3781 Some(AssetBalance::new(
3782 asset_name,
3783 Balance::new(p.qty, p.qty_available),
3784 now,
3785 ))
3786 })
3787 .collect()
3788}
3789
3790fn crypto_position_asset(symbol: &str) -> &str {
3798 let base = match symbol.split_once('/') {
3799 Some((base, _)) => Some(base),
3800 None => {
3801 let base_len = symbol.len().saturating_sub(USD.len());
3802 symbol
3803 .get(base_len..)
3804 .filter(|suffix| suffix.eq_ignore_ascii_case(USD))
3805 .and_then(|_| symbol.get(..base_len))
3806 }
3807 };
3808 match base {
3809 Some(base) if !base.is_empty() => base,
3810 _ => {
3811 warn!(
3812 %symbol,
3813 "Alpaca crypto position symbol is neither ASSETUSD nor BASE/QUOTE; using it whole"
3814 );
3815 symbol
3816 }
3817 }
3818}
3819
3820fn is_crypto_position(position: &AlpacaPosition) -> bool {
3823 position.asset_class.eq_ignore_ascii_case("crypto")
3824}
3825
3826fn convert_positions(
3848 positions: &[AlpacaPosition],
3849 now: DateTime<Utc>,
3850) -> Result<Vec<(&str, Position)>, UnindexedClientError> {
3851 positions
3852 .iter()
3853 .filter(|p| !is_crypto_position(p) && !p.qty.is_zero())
3854 .map(|p| {
3855 let size = p.qty.abs();
3856 let quantity = match p.side {
3857 AlpacaPositionSide::Long => size,
3858 AlpacaPositionSide::Short => -size,
3859 AlpacaPositionSide::Unknown => {
3860 return Err(UnindexedClientError::Internal(format!(
3861 "Alpaca position {} has a missing or unrecognised side",
3862 p.symbol
3863 )));
3864 }
3865 };
3866 let position = Position::new(
3867 quantity,
3868 p.avg_entry_price,
3869 p.unrealized_pl,
3870 None,
3871 None,
3872 None,
3873 now,
3874 );
3875 Ok((p.symbol.as_str(), position))
3876 })
3877 .collect()
3878}
3879
3880fn build_instrument_snapshots(
3901 orders: Vec<AlpacaOrderResponse>,
3902 positions: Vec<(&str, Position)>,
3903 instruments: &[InstrumentNameExchange],
3904) -> Vec<InstrumentAccountSnapshot<ExchangeId, AssetNameExchange, InstrumentNameExchange>> {
3905 struct Listing {
3907 orders: Vec<UnindexedOrderSnapshot>,
3908 complete: bool,
3909 position: Option<Position>,
3910 }
3911
3912 impl Listing {
3913 fn empty() -> Self {
3914 Self {
3915 orders: Vec::new(),
3916 complete: true,
3917 position: None,
3918 }
3919 }
3920 }
3921
3922 let mut by_symbol: IndexMap<SmolStr, Listing> = IndexMap::new();
3924
3925 for order in orders {
3928 let listing = by_symbol
3929 .entry(order.symbol.to_ascii_uppercase().into())
3930 .or_insert_with(Listing::empty);
3931 match convert_open_order(&order) {
3932 Some(converted) => listing.orders.push(converted.into()),
3933 None => listing.complete = false,
3934 }
3935 }
3936
3937 for (symbol, position) in positions {
3938 by_symbol
3939 .entry(symbol.to_ascii_uppercase().into())
3940 .or_insert_with(Listing::empty)
3941 .position = Some(position);
3942 }
3943
3944 let snapshot = |instrument: InstrumentNameExchange, listing: Listing| {
3945 let position = if is_options_or_equity_symbol(instrument.name().as_str()) {
3948 listing
3949 .position
3950 .map_or(PositionReport::Flat, PositionReport::from_position)
3951 } else {
3952 PositionReport::Unreported
3953 };
3954 InstrumentAccountSnapshot::new(instrument, listing.orders, listing.complete, position, None)
3955 };
3956
3957 if instruments.is_empty() {
3959 by_symbol
3960 .into_iter()
3961 .map(|(sym, listing)| snapshot(InstrumentNameExchange::new(sym), listing))
3962 .collect()
3963 } else {
3964 let mut claimed = std::collections::HashSet::new();
3966 instruments
3967 .iter()
3968 .map(|inst| {
3969 let symbol = inst.name().as_str().to_ascii_uppercase();
3970 if !claimed.insert(symbol.clone()) {
3973 return InstrumentAccountSnapshot::new(
3974 inst.clone(),
3975 Vec::new(),
3976 false,
3977 PositionReport::Unreported,
3978 None,
3979 );
3980 }
3981 let listing = by_symbol
3984 .swap_remove(symbol.as_str())
3985 .unwrap_or_else(Listing::empty);
3986 snapshot(inst.clone(), listing)
3987 })
3988 .collect()
3989 }
3990}
3991
3992fn convert_open_order(
4002 o: &AlpacaOrderResponse,
4003) -> Option<Order<ExchangeId, InstrumentNameExchange, Open>> {
4004 let converted = convert_representable_open_order(o);
4005 if converted.is_none() {
4006 warn!(
4007 order_id = %o.id,
4008 symbol = %o.symbol,
4009 qty = ?o.qty,
4010 side = %o.side,
4011 order_type = %o.order_type,
4012 "Alpaca open order cannot be represented - leaving it out"
4013 );
4014 }
4015 converted
4016}
4017
4018fn alpaca_cid(client_order_id: Option<&str>, order_id: &str) -> ClientOrderId {
4024 ClientOrderId::new(client_order_id.unwrap_or(order_id))
4025}
4026
4027fn convert_representable_open_order(
4029 o: &AlpacaOrderResponse,
4030) -> Option<Order<ExchangeId, InstrumentNameExchange, Open>> {
4031 let order_id = OrderId(SmolStr::new(&o.id));
4032 let cid = alpaca_cid(o.client_order_id.as_deref(), &o.id);
4033
4034 let instrument = InstrumentNameExchange::new(&o.symbol);
4035 let side = parse_side(&o.side)?;
4036 let quantity = Decimal::from_str(o.qty.as_deref().unwrap_or("0")).ok()?;
4037 if quantity.is_zero() {
4041 return None;
4042 }
4043 let price = o
4044 .limit_price
4045 .as_deref()
4046 .and_then(|s| Decimal::from_str(s).ok());
4047 let filled_qty = alpaca_filled_qty(&o.id, Some(&o.filled_qty)).unwrap_or(Decimal::ZERO);
4048 let kind = parse_order_kind(
4049 &o.order_type,
4050 o.stop_price.as_deref(),
4051 o.trail_percent.as_deref(),
4052 o.trail_price.as_deref(),
4053 )?;
4054 let time_in_force = parse_time_in_force(&o.time_in_force);
4055 let time_exchange = order_state_time(o);
4056
4057 Some(Order {
4058 key: OrderKey::new(
4059 ExchangeId::AlpacaBroker,
4060 instrument,
4061 StrategyId::unknown(),
4063 cid,
4064 ),
4065 side,
4066 price,
4067 quantity,
4068 kind,
4069 time_in_force,
4070 state: Open::new(VenueOrderId::Assigned(order_id), time_exchange, filled_qty),
4071 })
4072}
4073
4074fn convert_activity_to_trade(
4076 a: &AlpacaActivity,
4077) -> Option<Trade<AssetNameExchange, InstrumentNameExchange>> {
4078 let trade_id = TradeId::new(activity_execution_id(&a.id));
4079 let order_id = OrderId(SmolStr::new(&a.order_id));
4080 let instrument = InstrumentNameExchange::new(&a.symbol);
4081 let side = parse_side(&a.side)?;
4082 let price = Decimal::from_str(&a.price).ok()?;
4083 let quantity = Decimal::from_str(&a.qty).ok()?;
4084 let order_filled_quantity = a.cum_qty.as_deref().and_then(|s| Decimal::from_str(s).ok());
4087 let time_exchange = parse_timestamp(&a.transaction_time).unwrap_or_else(|| {
4088 warn!(id = %a.id, "Alpaca activity: unparseable transaction_time, using now");
4089 Utc::now()
4090 });
4091
4092 Some(Trade::new(
4097 trade_id,
4098 order_id,
4099 instrument,
4100 StrategyId::unknown(),
4101 time_exchange,
4102 side,
4103 price,
4104 quantity,
4105 order_filled_quantity,
4106 AssetFees::new(
4107 AssetNameExchange::from(USD),
4108 Decimal::ZERO,
4109 Some(Decimal::ZERO),
4110 ),
4111 ))
4112}
4113
4114fn ws_order_snapshot(
4124 order: &AlpacaOrderWs<'_>,
4125 instrument: InstrumentNameExchange,
4126 cid: ClientOrderId,
4127 order_id: OrderId,
4128 time_exchange: DateTime<Utc>,
4129) -> Option<UnindexedAccountEvent> {
4130 let side = parse_side(&order.side)?;
4131 let quantity = Decimal::from_str(order.qty.unwrap_or("0")).unwrap_or(Decimal::ZERO);
4132 if quantity.is_zero() {
4133 trace!(order_id = %order.id, "Alpaca WS: skipping notional order snapshot (qty=None)");
4134 return None;
4135 }
4136 let price = order.limit_price.and_then(|s| Decimal::from_str(s).ok());
4137 let filled_qty = order
4140 .filled_qty
4141 .and_then(|_| alpaca_filled_qty(&order.id, order.filled_qty))
4142 .unwrap_or(Decimal::ZERO);
4143 let kind = parse_order_kind(
4144 &order.order_type,
4145 order.stop_price,
4146 order.trail_percent,
4147 order.trail_price,
4148 )?;
4149 let time_in_force = parse_time_in_force(&order.time_in_force);
4150
4151 let order_snapshot = crate::order::Order {
4152 key: OrderKey::new(
4153 ExchangeId::AlpacaBroker,
4154 instrument,
4155 StrategyId::unknown(),
4156 cid,
4157 ),
4158 side,
4159 price,
4160 quantity,
4161 kind,
4162 time_in_force,
4163 state: OrderState::active(Open::new(
4164 VenueOrderId::Assigned(order_id),
4165 time_exchange,
4166 filled_qty,
4167 )),
4168 };
4169 Some(UnindexedAccountEvent::new(
4170 ExchangeId::AlpacaBroker,
4171 AccountEventKind::OrderSnapshot(rustrade_integration::collection::snapshot::Snapshot(
4172 order_snapshot,
4173 )),
4174 ))
4175}
4176
4177fn ws_fill_order_is_live(status: &str) -> bool {
4187 matches!(status, "partially_filled" | "filled")
4188}
4189
4190fn trade_amendment_event(
4202 update: &AlpacaTradeUpdate<'_>,
4203 instrument: InstrumentNameExchange,
4204 order_id: OrderId,
4205) -> UnindexedAccountEvent {
4206 let order = &update.order;
4207 let time_exchange = update
4208 .timestamp
4209 .and_then(parse_timestamp)
4210 .unwrap_or_else(Utc::now);
4211 let original = update
4212 .previous_execution_id
4213 .filter(|id| !id.is_empty())
4214 .map(TradeId::new);
4215 let quantity = update.qty.and_then(|s| Decimal::from_str(s).ok());
4216
4217 let kind = if update.event == "trade_bust" {
4218 TradeAmendmentKind::Busted {
4219 quantity: quantity.map(|quantity| quantity.abs()),
4220 }
4221 } else {
4222 let id = ws_execution_id(update).map(TradeId::new);
4223 let price = update.price.and_then(|s| Decimal::from_str(s).ok());
4224 match (id, price, quantity, parse_side(&order.side)) {
4225 (Some(id), Some(price), Some(quantity), Some(side)) if quantity > Decimal::ZERO => {
4226 TradeAmendmentKind::Corrected {
4227 replacement: Trade::new(
4228 id,
4229 order_id.clone(),
4230 instrument.clone(),
4231 StrategyId::unknown(),
4232 time_exchange,
4233 side,
4234 price,
4235 quantity,
4236 None,
4239 AssetFees::new(
4241 AssetNameExchange::from(USD),
4242 Decimal::ZERO,
4243 Some(Decimal::ZERO),
4244 ),
4245 ),
4246 }
4247 }
4248 (id, price, quantity, _) => TradeAmendmentKind::CorrectedUnresolved {
4249 id,
4250 price,
4251 quantity,
4252 },
4253 }
4254 };
4255
4256 UnindexedAccountEvent::new(
4257 ExchangeId::AlpacaBroker,
4258 AccountEventKind::TradeAmended(crate::trade::TradeAmendment::new(
4259 instrument,
4260 order_id,
4261 time_exchange,
4262 original,
4263 kind,
4264 )),
4265 )
4266}
4267
4268fn convert_trade_update(update: AlpacaTradeUpdate<'_>) -> [Option<UnindexedAccountEvent>; 2] {
4278 let event_str = update.event.as_str();
4281 if !matches!(
4282 event_str,
4283 "fill"
4284 | "partial_fill"
4285 | "new"
4286 | "accepted"
4287 | "pending_new"
4288 | "canceled"
4289 | "expired"
4290 | "replaced"
4291 | "done_for_day"
4292 | "rejected"
4293 | "trade_bust"
4294 | "trade_correct"
4295 ) {
4296 trace!(event = %event_str, "Alpaca WS: ignoring trade_updates event type");
4297 return [None, None];
4298 }
4299
4300 let order = &update.order;
4301 let instrument = InstrumentNameExchange::new(&*order.symbol);
4302 let order_id = OrderId(order.id.clone());
4303 let cid = alpaca_cid(order.client_order_id.as_deref(), &order.id);
4304
4305 match event_str {
4306 "fill" | "partial_fill" => {
4307 let (Some(price), Some(quantity), Some(side)) = (
4309 update.price.and_then(|s| Decimal::from_str(s).ok()),
4310 update.qty.and_then(|s| Decimal::from_str(s).ok()),
4311 parse_side(&order.side),
4312 ) else {
4313 return [None, None];
4314 };
4315 let time_exchange = update
4316 .timestamp
4317 .and_then(parse_timestamp)
4318 .unwrap_or_else(Utc::now);
4319
4320 let cum_qty = Decimal::from_str(order.filled_qty.unwrap_or("0"))
4325 .inspect_err(|e| {
4326 warn!(
4327 order_id = %order.id,
4328 filled_qty = ?order.filled_qty,
4329 %e,
4330 "Alpaca WS: failed to parse filled_qty — the fill reports a cumulative \
4331 of 0, and without an execution id a second such fill on the same order \
4332 would be deduplicated away"
4333 );
4334 })
4335 .unwrap_or(Decimal::ZERO);
4336 let trade_id = match ws_execution_id(&update) {
4340 Some(execution_id) => TradeId::new(execution_id),
4341 None => {
4342 warn!(
4343 order_id = %order.id,
4344 "Alpaca WS: fill without a usable execution_id — its TradeId is \
4345 \"{{order_id}}:{{filled_qty}}\", which fetch_trades will not match"
4346 );
4347 TradeId(fill_dedup_key(&order.id, cum_qty))
4348 }
4349 };
4350
4351 let trade = Trade::new(
4355 trade_id,
4356 order_id.clone(),
4357 instrument.clone(),
4358 StrategyId::unknown(),
4359 time_exchange,
4360 side,
4361 price,
4362 quantity,
4363 Some(cum_qty),
4365 AssetFees::new(
4366 AssetNameExchange::from(USD),
4367 Decimal::ZERO,
4368 Some(Decimal::ZERO),
4369 ),
4370 );
4371 let trade_event = UnindexedAccountEvent::new(
4372 ExchangeId::AlpacaBroker,
4373 AccountEventKind::Trade(trade),
4374 );
4375
4376 let snapshot_event = if ws_fill_order_is_live(&order.status) {
4380 ws_order_snapshot(order, instrument, cid, order_id, time_exchange)
4381 } else {
4382 trace!(
4383 order_id = %order.id,
4384 status = %order.status,
4385 "Alpaca WS: fill for an order the exchange no longer reports as working — \
4386 emitting the execution without an order snapshot"
4387 );
4388 None
4389 };
4390
4391 [Some(trade_event), snapshot_event]
4392 }
4393
4394 "new" | "accepted" | "pending_new" | "done_for_day" => {
4395 let time_exchange = update
4399 .timestamp
4400 .and_then(parse_timestamp)
4401 .unwrap_or_else(Utc::now);
4402
4403 [
4404 ws_order_snapshot(order, instrument, cid, order_id, time_exchange),
4405 None,
4406 ]
4407 }
4408
4409 "canceled" | "expired" | "replaced" => {
4410 let time_exchange = update
4418 .timestamp
4419 .and_then(parse_timestamp)
4420 .unwrap_or_else(Utc::now);
4421 let filled_qty = alpaca_filled_qty(&order.id, order.filled_qty);
4422 let cancelled = Cancelled::new(order_id, time_exchange, filled_qty);
4423 let response = crate::order::request::OrderResponseCancel {
4424 key: OrderKey::new(
4425 ExchangeId::AlpacaBroker,
4426 instrument,
4427 StrategyId::unknown(),
4428 cid,
4429 ),
4430 state: Ok(cancelled),
4431 };
4432 [
4433 Some(UnindexedAccountEvent::new(
4434 ExchangeId::AlpacaBroker,
4435 AccountEventKind::OrderCancelled(response),
4436 )),
4437 None,
4438 ]
4439 }
4440
4441 "trade_bust" | "trade_correct" => [
4442 Some(trade_amendment_event(&update, instrument, order_id)),
4443 None,
4444 ],
4445
4446 "rejected" => {
4447 let response = crate::order::request::OrderResponseCancel {
4448 key: OrderKey::new(
4449 ExchangeId::AlpacaBroker,
4450 instrument,
4451 StrategyId::unknown(),
4452 cid,
4453 ),
4454 state: Err(UnindexedOrderError::Rejected(ApiError::OrderRejected(
4455 format!("order rejected: status={}", order.status),
4456 ))),
4457 };
4458 [
4459 Some(UnindexedAccountEvent::new(
4460 ExchangeId::AlpacaBroker,
4461 AccountEventKind::OrderCancelled(response),
4462 )),
4463 None,
4464 ]
4465 }
4466
4467 _ => unreachable!("convert_trade_update: unrecognised event passed early-return guard"),
4470 }
4471}
4472
4473fn parse_side(s: &str) -> Option<Side> {
4478 match s {
4479 "buy" | "Buy" | "BUY" => Some(Side::Buy),
4480 "sell" | "Sell" | "SELL" => Some(Side::Sell),
4481 other => {
4482 trace!(%other, "Alpaca: unknown order side");
4483 None
4484 }
4485 }
4486}
4487
4488fn parse_order_kind(
4489 order_type: &str,
4490 stop_price: Option<&str>,
4491 trail_percent: Option<&str>,
4492 trail_price: Option<&str>,
4493) -> Option<OrderKind> {
4494 match order_type {
4495 "market" | "Market" => Some(OrderKind::Market),
4496 "limit" | "Limit" => Some(OrderKind::Limit),
4497 "stop" | "Stop" => {
4498 let trigger_price = stop_price.and_then(|s| Decimal::from_str(s).ok())?;
4499 Some(OrderKind::Stop { trigger_price })
4500 }
4501 "stop_limit" | "Stop_limit" => {
4502 let trigger_price = stop_price.and_then(|s| Decimal::from_str(s).ok())?;
4503 Some(OrderKind::StopLimit { trigger_price })
4504 }
4505 "trailing_stop" | "Trailing_stop" => {
4506 if let Some(pct) = trail_percent.and_then(|s| Decimal::from_str(s).ok()) {
4508 Some(OrderKind::TrailingStop {
4509 offset: pct,
4510 offset_type: TrailingOffsetType::Percentage,
4511 })
4512 } else if let Some(price) = trail_price.and_then(|s| Decimal::from_str(s).ok()) {
4513 Some(OrderKind::TrailingStop {
4514 offset: price,
4515 offset_type: TrailingOffsetType::Absolute,
4516 })
4517 } else {
4518 trace!("Alpaca: trailing_stop missing trail_percent and trail_price");
4519 None
4520 }
4521 }
4522 other => {
4523 trace!(%other, "Alpaca: unsupported order type, skipping");
4524 None
4525 }
4526 }
4527}
4528
4529fn parse_time_in_force(s: &str) -> TimeInForce {
4530 match s {
4531 "gtc" | "GTC" => TimeInForce::GoodUntilCancelled { post_only: false },
4532 "day" | "DAY" => TimeInForce::GoodUntilEndOfDay,
4533 "fok" | "FOK" => TimeInForce::FillOrKill,
4534 "ioc" | "IOC" => TimeInForce::ImmediateOrCancel,
4535 other => {
4536 warn!(%other, "Alpaca: unknown time_in_force, defaulting to GoodUntilEndOfDay");
4537 TimeInForce::GoodUntilEndOfDay
4538 }
4539 }
4540}
4541
4542fn order_state_time(order: &AlpacaOrderResponse) -> DateTime<Utc> {
4552 order
4553 .updated_at
4554 .as_deref()
4555 .and_then(parse_timestamp)
4556 .or_else(|| parse_timestamp(&order.created_at))
4557 .unwrap_or_else(Utc::now)
4558}
4559
4560fn alpaca_filled_qty(order_id: &str, filled_qty: Option<&str>) -> Option<Decimal> {
4563 let Some(raw) = filled_qty else {
4564 warn!(%order_id, "Alpaca did not report how much the order filled");
4565 return None;
4566 };
4567 match Decimal::from_str(raw) {
4568 Ok(filled_qty) => Some(filled_qty),
4569 Err(_) => {
4570 warn!(%order_id, filled_qty = raw, "Alpaca reported an unparseable filled_qty, treating it as unknown");
4571 None
4572 }
4573 }
4574}
4575
4576fn parse_timestamp(s: &str) -> Option<DateTime<Utc>> {
4577 DateTime::parse_from_rfc3339(s)
4578 .ok()
4579 .map(|dt| dt.with_timezone(&Utc))
4580}
4581
4582fn map_side(side: Side) -> &'static str {
4583 match side {
4584 Side::Buy => "buy",
4585 Side::Sell => "sell",
4586 }
4587}
4588
4589fn map_order_kind(kind: OrderKind) -> Option<&'static str> {
4590 match kind {
4591 OrderKind::Market => Some("market"),
4592 OrderKind::Limit => Some("limit"),
4593 OrderKind::Stop { .. } => Some("stop"),
4594 OrderKind::StopLimit { .. } => Some("stop_limit"),
4595 OrderKind::TrailingStop { .. } => Some("trailing_stop"),
4596 OrderKind::TakeProfit { .. }
4598 | OrderKind::TakeProfitLimit { .. }
4599 | OrderKind::TrailingStopLimit { .. } => None,
4600 }
4601}
4602
4603fn map_time_in_force(tif: TimeInForce) -> Result<&'static str, &'static str> {
4612 match tif {
4613 TimeInForce::GoodUntilCancelled { post_only } => {
4614 if post_only {
4615 return Err("Alpaca does not support post_only orders");
4616 }
4617 Ok("gtc")
4618 }
4619 TimeInForce::GoodUntilEndOfDay => Ok("day"),
4620 TimeInForce::FillOrKill => Ok("fok"),
4621 TimeInForce::ImmediateOrCancel => Ok("ioc"),
4622 TimeInForce::AtOpen => Ok("opg"),
4623 TimeInForce::AtClose => Ok("cls"),
4624 TimeInForce::GoodTillDate { .. } => {
4628 Err("Alpaca GoodTillDate is not yet wired through this client")
4629 }
4630 }
4631}
4632
4633fn is_options_or_equity_symbol(symbol: &str) -> bool {
4644 !symbol.contains('/')
4645}
4646
4647fn map_position_intent(side: Side, reduce_only: bool) -> AlpacaPositionIntent {
4656 match (reduce_only, side) {
4657 (false, Side::Buy) => AlpacaPositionIntent::BuyToOpen,
4658 (false, Side::Sell) => AlpacaPositionIntent::SellToOpen,
4659 (true, Side::Buy) => AlpacaPositionIntent::BuyToClose,
4660 (true, Side::Sell) => AlpacaPositionIntent::SellToClose,
4661 }
4662}
4663
4664fn parse_api_error(status: reqwest::StatusCode, message: &str) -> crate::error::UnindexedApiError {
4669 if status == reqwest::StatusCode::TOO_MANY_REQUESTS {
4671 return ApiError::RateLimit;
4672 }
4673
4674 let lower = message.to_ascii_lowercase();
4676 match status.as_u16() {
4677 422 if lower.contains("already") => ApiError::OrderAlreadyCancelled,
4681 422 if lower.contains("insufficient") => {
4688 ApiError::BalanceInsufficient(None, message.to_owned())
4689 }
4690 401 => ApiError::Unauthenticated(format!("unauthorized: {message}")),
4691 403 => ApiError::Unauthenticated(format!("forbidden: {message}")),
4692 404 => ApiError::OrderRejected(format!("order not found: {message}")),
4693 _ => ApiError::OrderRejected(message.to_owned()),
4694 }
4695}
4696
4697fn parse_order_error(status: reqwest::StatusCode, message: &str) -> UnindexedOrderError {
4699 UnindexedOrderError::Rejected(parse_api_error(status, message))
4700}
4701
4702fn connectivity_err(msg: impl Into<String>) -> UnindexedClientError {
4703 UnindexedClientError::Connectivity(ConnectivityError::Socket(msg.into()))
4704}
4705
4706fn order_post_error(error: UnindexedClientError) -> UnindexedOrderError {
4713 match error {
4714 UnindexedClientError::Connectivity(ce) => OrderError::Connectivity(ce),
4715 UnindexedClientError::Api(ae) => OrderError::Rejected(ae),
4716 UnindexedClientError::Internal(msg) => OrderError::Connectivity(ConnectivityError::Socket(
4717 format!("Alpaca order status unknown, its 2xx response could not be read: {msg}"),
4718 )),
4719 other @ (UnindexedClientError::TaskFailed(_)
4723 | UnindexedClientError::Truncated { .. }
4724 | UnindexedClientError::TruncatedSnapshot { .. }) => OrderError::Connectivity(
4725 ConnectivityError::Socket(format!("Alpaca order status unknown: {other}")),
4726 ),
4727 }
4728}
4729
4730impl OrderStatusClient for AlpacaClient {
4735 async fn fetch_ended_orders(
4743 &self,
4744 orders: &[UnindexedOrderKey],
4745 ) -> Result<Vec<UnindexedInactiveOrder>, UnindexedClientError> {
4746 fetch_ended_by_key(orders, |key| {
4747 fetch_order_lookup(
4748 self.http.clone(),
4749 self.rate_limiter.clone(),
4750 self.config.clone(),
4751 key,
4752 )
4753 })
4754 .await
4755 }
4756}
4757
4758impl BracketOrderClient for AlpacaClient {
4763 async fn open_bracket_order(
4764 &self,
4765 request: UnifiedBracketOrderRequest<ExchangeId, &InstrumentNameExchange>,
4766 ) -> UnifiedBracketOrderResult {
4767 let alpaca_request = AlpacaBracketOrderRequest {
4768 instrument: request.key.instrument.clone(),
4769 strategy: request.key.strategy.clone(),
4770 cid: request.key.cid.clone(),
4771 side: request.state.side,
4772 quantity: request.state.quantity,
4773 entry_price: request.state.entry_price,
4774 take_profit_price: request.state.take_profit_price,
4775 stop_loss_price: request.state.stop_loss_price,
4776 stop_loss_limit_price: request.state.stop_loss_limit_price,
4777 time_in_force: request.state.time_in_force,
4778 };
4779
4780 let result = self.open_bracket_order(alpaca_request).await;
4781
4782 UnifiedBracketOrderResult::parent_only(result.parent)
4783 }
4784}
4785
4786#[cfg(test)]
4791#[allow(clippy::unwrap_used, clippy::expect_used)] mod tests {
4793 use super::*;
4794 use rust_decimal_macros::dec;
4795
4796 #[test]
4797 fn test_alpaca_config_new_uses_paper_trading_by_default() {
4798 let cfg = AlpacaConfig::new("my_key".into(), "my_secret".into());
4799 assert!(cfg.paper);
4800 }
4801
4802 #[test]
4803 fn test_alpaca_config_debug_redacts_credentials() {
4804 let cfg = AlpacaConfig::new("my_key".into(), "my_secret".into());
4805 let debug = format!("{cfg:?}");
4806 assert!(!debug.contains("my_key"), "api_key should be redacted");
4807 assert!(
4808 !debug.contains("my_secret"),
4809 "secret_key should be redacted"
4810 );
4811 assert!(debug.contains("paper: true"));
4812 }
4813
4814 #[test]
4815 fn test_alpaca_config_deserialize_omitted_paper_defaults_to_paper() {
4816 let cfg: AlpacaConfig = serde_json::from_str(
4817 r#"{
4818 "api_key": "my_key",
4819 "secret_key": "my_secret"
4820 }"#,
4821 )
4822 .unwrap();
4823
4824 assert!(cfg.paper);
4825 assert_eq!(cfg.rest_base_url(), "https://paper-api.alpaca.markets");
4826 }
4827
4828 #[test]
4829 fn test_alpaca_config_deserialize_paper_true() {
4830 let cfg: AlpacaConfig = serde_json::from_str(
4831 r#"{
4832 "api_key": "my_key",
4833 "secret_key": "my_secret",
4834 "paper": true
4835 }"#,
4836 )
4837 .unwrap();
4838
4839 assert!(cfg.paper);
4840 assert_eq!(cfg.rest_base_url(), "https://paper-api.alpaca.markets");
4841 }
4842
4843 #[test]
4844 fn test_alpaca_config_deserialize_paper_false() {
4845 let cfg: AlpacaConfig = serde_json::from_str(
4846 r#"{
4847 "api_key": "my_key",
4848 "secret_key": "my_secret",
4849 "paper": false
4850 }"#,
4851 )
4852 .unwrap();
4853
4854 assert!(!cfg.paper);
4855 assert_eq!(cfg.rest_base_url(), "https://api.alpaca.markets");
4856 }
4857
4858 #[test]
4859 fn test_alpaca_config_urls() {
4860 let paper = AlpacaConfig::paper("k".into(), "s".into());
4861 assert!(paper.rest_base_url().contains("paper-api"));
4862 assert!(paper.ws_url().contains("paper-api"));
4863
4864 let live = AlpacaConfig::production("k".into(), "s".into());
4865 assert!(!live.rest_base_url().contains("paper-api"));
4866 assert!(!live.ws_url().contains("paper-api"));
4867 }
4868
4869 #[test]
4870 #[serial_test::serial]
4871 fn test_alpaca_config_from_env_defaults_to_paper_trading() {
4872 temp_env::with_vars(
4873 [
4874 ("ALPACA_API_KEY", Some("my_key")),
4875 ("ALPACA_SECRET_KEY", Some("my_secret")),
4876 ("ALPACA_PAPER", None),
4877 ],
4878 || {
4879 let cfg = AlpacaConfig::from_env().unwrap();
4880 assert!(cfg.paper);
4881 assert_eq!(cfg.rest_base_url(), "https://paper-api.alpaca.markets");
4882 },
4883 );
4884 }
4885
4886 #[test]
4887 #[serial_test::serial]
4888 fn test_alpaca_config_from_env_accepts_explicit_production() {
4889 temp_env::with_vars(
4890 [
4891 ("ALPACA_API_KEY", Some("my_key")),
4892 ("ALPACA_SECRET_KEY", Some("my_secret")),
4893 ("ALPACA_PAPER", Some("false")),
4894 ],
4895 || {
4896 let cfg = AlpacaConfig::from_env().unwrap();
4897 assert!(!cfg.paper);
4898 assert_eq!(cfg.rest_base_url(), "https://api.alpaca.markets");
4899 },
4900 );
4901 }
4902
4903 #[test]
4904 #[serial_test::serial]
4905 fn test_alpaca_config_from_env_rejects_invalid_paper() {
4906 temp_env::with_vars(
4907 [
4908 ("ALPACA_API_KEY", Some("my_key")),
4909 ("ALPACA_SECRET_KEY", Some("my_secret")),
4910 ("ALPACA_PAPER", Some("maybe")),
4911 ],
4912 || {
4913 let err = AlpacaConfig::from_env().unwrap_err();
4914 assert!(matches!(err, AlpacaConfigError::InvalidPaper(value) if value == "maybe"));
4915 },
4916 );
4917 }
4918
4919 #[test]
4920 #[serial_test::serial]
4921 fn test_alpaca_config_from_env_requires_credentials() {
4922 temp_env::with_vars(
4923 [
4924 ("ALPACA_API_KEY", None),
4925 ("ALPACA_SECRET_KEY", Some("my_secret")),
4926 ("ALPACA_PAPER", None),
4927 ],
4928 || {
4929 let err = AlpacaConfig::from_env().unwrap_err();
4930 assert!(matches!(err, AlpacaConfigError::MissingApiKey));
4931 },
4932 );
4933 }
4934
4935 #[test]
4936 fn test_parse_side() {
4937 assert_eq!(parse_side("buy"), Some(Side::Buy));
4938 assert_eq!(parse_side("sell"), Some(Side::Sell));
4939 assert_eq!(parse_side("Buy"), Some(Side::Buy));
4940 assert_eq!(parse_side("BUY"), Some(Side::Buy));
4941 assert_eq!(parse_side("unknown"), None);
4942 }
4943
4944 #[test]
4945 fn test_parse_order_kind() {
4946 assert_eq!(
4948 parse_order_kind("market", None, None, None),
4949 Some(OrderKind::Market)
4950 );
4951 assert_eq!(
4952 parse_order_kind("limit", None, None, None),
4953 Some(OrderKind::Limit)
4954 );
4955
4956 assert_eq!(
4958 parse_order_kind("stop", Some("150.00"), None, None),
4959 Some(OrderKind::Stop {
4960 trigger_price: Decimal::from_str("150.00").unwrap()
4961 })
4962 );
4963 assert_eq!(parse_order_kind("stop", None, None, None), None);
4964
4965 assert_eq!(
4967 parse_order_kind("stop_limit", Some("145.00"), None, None),
4968 Some(OrderKind::StopLimit {
4969 trigger_price: Decimal::from_str("145.00").unwrap()
4970 })
4971 );
4972
4973 assert_eq!(
4975 parse_order_kind("trailing_stop", None, Some("5.0"), None),
4976 Some(OrderKind::TrailingStop {
4977 offset: Decimal::from_str("5.0").unwrap(),
4978 offset_type: TrailingOffsetType::Percentage,
4979 })
4980 );
4981
4982 assert_eq!(
4984 parse_order_kind("trailing_stop", None, None, Some("2.50")),
4985 Some(OrderKind::TrailingStop {
4986 offset: Decimal::from_str("2.50").unwrap(),
4987 offset_type: TrailingOffsetType::Absolute,
4988 })
4989 );
4990
4991 assert_eq!(parse_order_kind("trailing_stop", None, None, None), None);
4993
4994 assert_eq!(parse_order_kind("unknown", None, None, None), None);
4996 }
4997
4998 #[test]
4999 fn test_map_order_kind() {
5000 assert_eq!(map_order_kind(OrderKind::Market), Some("market"));
5001 assert_eq!(map_order_kind(OrderKind::Limit), Some("limit"));
5002 assert_eq!(
5003 map_order_kind(OrderKind::Stop {
5004 trigger_price: Decimal::from_str("150.00").unwrap()
5005 }),
5006 Some("stop")
5007 );
5008 assert_eq!(
5009 map_order_kind(OrderKind::StopLimit {
5010 trigger_price: Decimal::from_str("145.00").unwrap()
5011 }),
5012 Some("stop_limit")
5013 );
5014 assert_eq!(
5015 map_order_kind(OrderKind::TrailingStop {
5016 offset: Decimal::from_str("5.0").unwrap(),
5017 offset_type: TrailingOffsetType::Percentage,
5018 }),
5019 Some("trailing_stop")
5020 );
5021 assert_eq!(
5022 map_order_kind(OrderKind::TrailingStop {
5023 offset: Decimal::from_str("2.50").unwrap(),
5024 offset_type: TrailingOffsetType::Absolute,
5025 }),
5026 Some("trailing_stop")
5027 );
5028 assert_eq!(
5030 map_order_kind(OrderKind::TrailingStopLimit {
5031 offset: Decimal::from_str("5.0").unwrap(),
5032 offset_type: TrailingOffsetType::Percentage,
5033 stop_price: Decimal::from_str("95.0").unwrap(),
5034 limit_offset: Decimal::from_str("1.0").unwrap(),
5035 }),
5036 None
5037 );
5038 assert_eq!(
5040 map_order_kind(OrderKind::TakeProfit {
5041 trigger_price: Decimal::from_str("160.00").unwrap()
5042 }),
5043 None
5044 );
5045 assert_eq!(
5046 map_order_kind(OrderKind::TakeProfitLimit {
5047 trigger_price: Decimal::from_str("160.00").unwrap()
5048 }),
5049 None
5050 );
5051 }
5052
5053 #[test]
5054 fn test_map_time_in_force_roundtrip() {
5055 assert_eq!(
5056 map_time_in_force(TimeInForce::GoodUntilCancelled { post_only: false }),
5057 Ok("gtc")
5058 );
5059 assert_eq!(map_time_in_force(TimeInForce::GoodUntilEndOfDay), Ok("day"));
5060 assert_eq!(map_time_in_force(TimeInForce::FillOrKill), Ok("fok"));
5061 assert_eq!(map_time_in_force(TimeInForce::ImmediateOrCancel), Ok("ioc"));
5062 }
5063
5064 #[test]
5065 fn test_map_time_in_force_rejects_post_only() {
5066 let result = map_time_in_force(TimeInForce::GoodUntilCancelled { post_only: true });
5067 assert!(result.is_err(), "post_only must be rejected");
5068 assert!(result.unwrap_err().contains("post_only"));
5069 }
5070
5071 #[test]
5076 fn test_bracket_order_serializes_with_stop_loss_stop_order() {
5077 let body = AlpacaOrderRequest {
5079 symbol: "AAPL",
5080 qty: "10".to_string(),
5081 side: "buy",
5082 order_type: "limit",
5083 time_in_force: "gtc",
5084 limit_price: Some("150.00".to_string()),
5085 stop_price: None,
5086 trail_percent: None,
5087 trail_price: None,
5088 client_order_id: Some("bracket-001"),
5089 position_intent: Some(AlpacaPositionIntent::BuyToOpen),
5090 order_class: Some("bracket"),
5091 take_profit: Some(TakeProfitParams {
5092 limit_price: "160.00".to_string(),
5093 }),
5094 stop_loss: Some(StopLossParams {
5095 stop_price: "145.00".to_string(),
5096 limit_price: None,
5097 }),
5098 };
5099
5100 let json = serde_json::to_value(&body).unwrap();
5101
5102 assert_eq!(json["symbol"], "AAPL");
5103 assert_eq!(json["qty"], "10");
5104 assert_eq!(json["side"], "buy");
5105 assert_eq!(json["type"], "limit");
5106 assert_eq!(json["time_in_force"], "gtc");
5107 assert_eq!(json["limit_price"], "150.00");
5108 assert_eq!(json["order_class"], "bracket");
5109
5110 assert_eq!(json["take_profit"]["limit_price"], "160.00");
5112
5113 assert_eq!(json["stop_loss"]["stop_price"], "145.00");
5115 assert!(
5116 json["stop_loss"].get("limit_price").is_none(),
5117 "stop_loss.limit_price should be omitted when None"
5118 );
5119 }
5120
5121 #[test]
5122 fn test_bracket_order_serializes_with_stop_loss_stop_limit_order() {
5123 let body = AlpacaOrderRequest {
5125 symbol: "SPY",
5126 qty: "5".to_string(),
5127 side: "sell",
5128 order_type: "limit",
5129 time_in_force: "day",
5130 limit_price: Some("450.00".to_string()),
5131 stop_price: None,
5132 trail_percent: None,
5133 trail_price: None,
5134 client_order_id: Some("bracket-002"),
5135 position_intent: Some(AlpacaPositionIntent::SellToClose),
5136 order_class: Some("bracket"),
5137 take_profit: Some(TakeProfitParams {
5138 limit_price: "440.00".to_string(),
5139 }),
5140 stop_loss: Some(StopLossParams {
5141 stop_price: "455.00".to_string(),
5142 limit_price: Some("456.00".to_string()),
5143 }),
5144 };
5145
5146 let json = serde_json::to_value(&body).unwrap();
5147
5148 assert_eq!(json["symbol"], "SPY");
5149 assert_eq!(json["side"], "sell");
5150 assert_eq!(json["time_in_force"], "day");
5151 assert_eq!(json["order_class"], "bracket");
5152
5153 assert_eq!(json["take_profit"]["limit_price"], "440.00");
5155
5156 assert_eq!(json["stop_loss"]["stop_price"], "455.00");
5158 assert_eq!(
5159 json["stop_loss"]["limit_price"], "456.00",
5160 "stop_loss.limit_price should be present for stop-limit orders"
5161 );
5162 }
5163
5164 #[tokio::test]
5169 async fn test_open_bracket_order_rejects_invalid_tif() {
5170 use rust_decimal_macros::dec;
5171 use rustrade_instrument::instrument::name::InstrumentNameExchange;
5172
5173 let config = AlpacaConfig::new("dummy_key".into(), "dummy_secret".into());
5175 let client = AlpacaClient::new(config);
5176
5177 let request = AlpacaBracketOrderRequest::new(
5178 InstrumentNameExchange::new("SPY"),
5179 crate::order::id::StrategyId::new("test"),
5180 crate::order::id::ClientOrderId::new("test-tif"),
5181 Side::Buy,
5182 dec!(1),
5183 dec!(100.00),
5184 dec!(120.00),
5185 dec!(90.00),
5186 TimeInForce::ImmediateOrCancel, );
5188
5189 let result = client.open_bracket_order(request).await;
5190
5191 assert!(
5192 result.parent.state.is_failed(),
5193 "Bracket order with IOC TIF should be rejected locally"
5194 );
5195 }
5196
5197 #[tokio::test]
5198 async fn test_open_bracket_order_rejects_invalid_price_ordering() {
5199 use rust_decimal_macros::dec;
5200 use rustrade_instrument::instrument::name::InstrumentNameExchange;
5201
5202 let config = AlpacaConfig::new("dummy_key".into(), "dummy_secret".into());
5203 let client = AlpacaClient::new(config);
5204
5205 let request = AlpacaBracketOrderRequest::new(
5207 InstrumentNameExchange::new("SPY"),
5208 crate::order::id::StrategyId::new("test"),
5209 crate::order::id::ClientOrderId::new("test-price"),
5210 Side::Buy,
5211 dec!(1),
5212 dec!(100.00),
5213 dec!(120.00),
5214 dec!(105.00), TimeInForce::GoodUntilCancelled { post_only: false },
5216 );
5217
5218 let result = client.open_bracket_order(request).await;
5219
5220 assert!(
5221 result.parent.state.is_failed(),
5222 "Bracket order with invalid price ordering should be rejected locally"
5223 );
5224 }
5225
5226 #[tokio::test]
5227 async fn test_open_bracket_order_rejects_invalid_sl_limit_price() {
5228 use rust_decimal_macros::dec;
5229 use rustrade_instrument::instrument::name::InstrumentNameExchange;
5230
5231 let config = AlpacaConfig::new("dummy_key".into(), "dummy_secret".into());
5232 let client = AlpacaClient::new(config);
5233
5234 let request = AlpacaBracketOrderRequest::new(
5236 InstrumentNameExchange::new("SPY"),
5237 crate::order::id::StrategyId::new("test"),
5238 crate::order::id::ClientOrderId::new("test-sl-limit"),
5239 Side::Buy,
5240 dec!(1),
5241 dec!(100.00),
5242 dec!(120.00),
5243 dec!(90.00),
5244 TimeInForce::GoodUntilCancelled { post_only: false },
5245 )
5246 .with_stop_loss_limit_price(dec!(95.00)); let result = client.open_bracket_order(request).await;
5249
5250 assert!(
5251 result.parent.state.is_failed(),
5252 "Bracket order with invalid SL limit price should be rejected locally"
5253 );
5254 }
5255
5256 #[test]
5257 fn test_non_bracket_order_omits_bracket_fields() {
5258 let body = AlpacaOrderRequest {
5260 symbol: "AAPL",
5261 qty: "1".to_string(),
5262 side: "buy",
5263 order_type: "limit",
5264 time_in_force: "gtc",
5265 limit_price: Some("150.00".to_string()),
5266 stop_price: None,
5267 trail_percent: None,
5268 trail_price: None,
5269 client_order_id: Some("regular-001"),
5270 position_intent: Some(AlpacaPositionIntent::BuyToOpen),
5271 order_class: None,
5272 take_profit: None,
5273 stop_loss: None,
5274 };
5275
5276 let json = serde_json::to_value(&body).unwrap();
5277
5278 assert_eq!(json["symbol"], "AAPL");
5279 assert!(
5280 json.get("order_class").is_none(),
5281 "order_class should be omitted for non-bracket orders"
5282 );
5283 assert!(
5284 json.get("take_profit").is_none(),
5285 "take_profit should be omitted for non-bracket orders"
5286 );
5287 assert!(
5288 json.get("stop_loss").is_none(),
5289 "stop_loss should be omitted for non-bracket orders"
5290 );
5291 }
5292
5293 #[test]
5294 fn test_parse_timestamp_valid() {
5295 let ts = parse_timestamp("2025-04-18T14:30:00Z");
5296 assert!(ts.is_some());
5297 let ts2 = parse_timestamp("2025-04-18T14:30:00.123456Z");
5298 assert!(ts2.is_some());
5299 assert_eq!(parse_timestamp("not-a-timestamp"), None);
5300 }
5301
5302 #[test]
5303 fn test_check_auth_response_authorized() {
5304 let msg =
5305 r#"{"stream":"authorization","data":{"status":"authorized","action":"authenticate"}}"#;
5306 assert!(matches!(check_auth_response(msg), Some(Ok(()))));
5307 }
5308
5309 #[test]
5310 fn test_check_auth_response_unauthorized() {
5311 let msg = r#"{"stream":"authorization","data":{"status":"unauthorized"}}"#;
5312 assert!(matches!(
5313 check_auth_response(msg),
5314 Some(Err(HandshakeError::Auth(_)))
5315 ));
5316 }
5317
5318 #[test]
5319 fn test_check_auth_response_non_auth_message() {
5320 let msg = r#"{"stream":"trade_updates","data":{}}"#;
5321 assert!(check_auth_response(msg).is_none());
5322 }
5323
5324 #[test]
5325 fn test_check_listen_ack() {
5326 let ack = r#"{"stream":"listening","data":{"streams":["trade_updates"]}}"#;
5327 assert!(check_listen_ack(ack));
5328
5329 let other = r#"{"stream":"authorization","data":{}}"#;
5330 assert!(!check_listen_ack(other));
5331 }
5332
5333 #[test]
5334 fn test_dedup_cache() {
5335 let cache = new_dedup_cache();
5336 let key = || SmolStr::new("order-1:1");
5337 assert!(
5338 !is_duplicate_fill(&cache, Some("exec-1"), key()),
5339 "first time should not be duplicate"
5340 );
5341 assert!(
5342 is_duplicate_fill(&cache, Some("exec-1"), key()),
5343 "second time should be duplicate"
5344 );
5345 }
5346
5347 #[test]
5350 fn dedup_tells_fills_apart_by_execution_id_not_cumulative() {
5351 let cache = new_dedup_cache();
5352 assert!(!is_duplicate_fill(
5353 &cache,
5354 Some("exec-a"),
5355 SmolStr::new("ord-1:5")
5356 ));
5357 assert!(
5359 !is_duplicate_fill(&cache, Some("exec-b"), SmolStr::new("ord-1:5")),
5360 "a different execution at a cumulative seen before is a new fill"
5361 );
5362 }
5363
5364 #[test]
5367 fn dedup_matches_a_fill_with_an_execution_id_to_the_same_fill_without_one() {
5368 let cache = new_dedup_cache();
5369
5370 assert!(!is_duplicate_fill(&cache, None, SmolStr::new("ord-1:2")));
5371 assert!(
5372 is_duplicate_fill(&cache, Some("exec-a"), SmolStr::new("ord-1:2")),
5373 "without an id first, then with one"
5374 );
5375
5376 assert!(!is_duplicate_fill(
5377 &cache,
5378 Some("exec-b"),
5379 SmolStr::new("ord-1:4")
5380 ));
5381 assert!(
5382 is_duplicate_fill(&cache, None, SmolStr::new("ord-1:4")),
5383 "with an id first, then without one"
5384 );
5385 }
5386
5387 #[tokio::test]
5388 async fn test_exponential_backoff_progression_and_exhaustion() {
5389 tokio::time::pause();
5390
5391 let mut b = ExponentialBackoff::new();
5392
5393 assert!(b.wait().await, "first wait should return true");
5395 assert_eq!(b.attempt, 1);
5396
5397 while b.wait().await {}
5399
5400 assert_eq!(b.attempt, MAX_RECONNECT_ATTEMPTS);
5402
5403 assert!(!b.wait().await, "exhausted backoff should return false");
5405
5406 b.reset();
5408 assert_eq!(b.attempt, 0);
5409
5410 assert!(b.wait().await, "wait should succeed after reset");
5412 assert_eq!(b.attempt, 1);
5413 }
5414
5415 #[test]
5416 fn test_convert_account_to_balances_empty_assets() {
5417 let account: AlpacaAccount = serde_json::from_value(serde_json::json!({
5420 "cash": "9000.50",
5421 "equity": "12000.00",
5422 "buying_power": "36000.00",
5423 "regt_buying_power": "18000.00",
5424 "options_buying_power": "8000.00",
5425 "non_marginable_buying_power": "8500.25",
5426 }))
5427 .unwrap();
5428 let balances = convert_account_to_balances(&account, &[]);
5429 assert_eq!(balances.len(), 1);
5430 assert_eq!(balances[0].asset, AssetNameExchange::new("USD"));
5431 assert_eq!(balances[0].balance.total, dec!(9000.50), "total is cash");
5432 assert_eq!(
5433 balances[0].balance.free,
5434 dec!(8500.25),
5435 "free is non-marginable buying power when below cash"
5436 );
5437 assert_eq!(balances[0].balance.margin, None);
5438 }
5439
5440 #[test]
5443 fn test_convert_account_to_balances_free_capped_at_cash_with_long_stock() {
5444 let account: AlpacaAccount = serde_json::from_value(serde_json::json!({
5445 "cash": "50000",
5446 "equity": "100000",
5447 "initial_margin": "25000",
5448 "non_marginable_buying_power": "75000",
5449 }))
5450 .unwrap();
5451 let balances = convert_account_to_balances(&account, &[]);
5452 assert_eq!(balances[0].balance.total, dec!(50000));
5453 assert_eq!(balances[0].balance.free, dec!(50000));
5454 }
5455
5456 #[test]
5459 fn test_convert_account_to_balances_open_short_keeps_free_below_cash() {
5460 let account: AlpacaAccount = serde_json::from_value(serde_json::json!({
5461 "cash": "10200",
5462 "equity": "10000",
5463 "initial_margin": "100",
5464 "non_marginable_buying_power": "9900",
5465 }))
5466 .unwrap();
5467 let balances = convert_account_to_balances(&account, &[]);
5468 assert_eq!(balances[0].balance.total, dec!(10200));
5469 assert_eq!(balances[0].balance.free, dec!(9900));
5470 }
5471
5472 #[test]
5473 fn test_convert_account_to_balances_negative_cash_on_margin() {
5474 let account: AlpacaAccount = serde_json::from_value(serde_json::json!({
5475 "cash": "-2500.75",
5476 "equity": "7500.00",
5477 "buying_power": "5000.00",
5478 "non_marginable_buying_power": "1250.00",
5479 }))
5480 .unwrap();
5481 let balances = convert_account_to_balances(&account, &[]);
5482 assert_eq!(balances.len(), 1);
5483 assert_eq!(balances[0].balance.total, dec!(-2500.75));
5484 assert_eq!(
5485 balances[0].balance.free,
5486 dec!(-2500.75),
5487 "no free cash while borrowing; free <= total still holds"
5488 );
5489 }
5490
5491 #[test]
5492 fn test_alpaca_account_malformed_cash_fails_to_decode() {
5493 let result = serde_json::from_value::<AlpacaAccount>(serde_json::json!({
5494 "cash": "not-a-number",
5495 "non_marginable_buying_power": "0",
5496 }));
5497 assert!(result.is_err(), "a malformed cash must not read as zero");
5498 }
5499
5500 #[test]
5501 fn test_convert_account_to_balances_usd_filter() {
5502 let account = AlpacaAccount {
5503 cash: dec!(12000.00),
5504 non_marginable_buying_power: dec!(10000.00),
5505 };
5506 let usd = vec![AssetNameExchange::new("USD")];
5507 let balances = convert_account_to_balances(&account, &usd);
5508 assert_eq!(balances.len(), 1);
5509 assert_eq!(balances[0].asset, AssetNameExchange::new("USD"));
5510
5511 let non_usd = vec![AssetNameExchange::new("BTC")];
5512 let balances = convert_account_to_balances(&account, &non_usd);
5513 assert!(balances.is_empty());
5514 }
5515
5516 #[test]
5519 fn a_usd_balance_is_named_usd_however_the_filter_spells_it() {
5520 let account = AlpacaAccount {
5521 cash: dec!(12000.00),
5522 non_marginable_buying_power: dec!(10000.00),
5523 };
5524 let balances = convert_account_to_balances(&account, &[AssetNameExchange::new("usd")]);
5525 assert_eq!(balances.len(), 1);
5526 assert_eq!(balances[0].asset, AssetNameExchange::new("USD"));
5527 }
5528
5529 #[test]
5530 fn test_is_options_or_equity_symbol() {
5531 assert!(!is_options_or_equity_symbol("BTC/USD"));
5533 assert!(!is_options_or_equity_symbol("ETH/USD"));
5534 assert!(!is_options_or_equity_symbol("SOL/USD"));
5535
5536 assert!(is_options_or_equity_symbol("AAPL"));
5538 assert!(is_options_or_equity_symbol("SPY"));
5539 assert!(is_options_or_equity_symbol("MSFT"));
5540
5541 assert!(is_options_or_equity_symbol("SPY250418C00450000"));
5543 assert!(is_options_or_equity_symbol("AAPL250418P00145000"));
5544 }
5545
5546 #[test]
5547 fn test_parse_order_error_already_cancelled() {
5548 assert!(matches!(
5551 parse_order_error(
5552 reqwest::StatusCode::UNPROCESSABLE_ENTITY,
5553 "order is already cancelled"
5554 ),
5555 UnindexedOrderError::Rejected(ApiError::OrderAlreadyCancelled)
5556 ));
5557 }
5558
5559 #[test]
5560 fn test_parse_order_error_already_wins_over_insufficient_on_422() {
5561 assert!(matches!(
5564 parse_order_error(
5565 reqwest::StatusCode::UNPROCESSABLE_ENTITY,
5566 "order already cancelled due to insufficient margin"
5567 ),
5568 UnindexedOrderError::Rejected(ApiError::OrderAlreadyCancelled)
5569 ));
5570 }
5571
5572 fn make_order_ws<'a>(
5573 id: &str,
5574 symbol: &str,
5575 side: &str,
5576 filled_qty: &'a str,
5577 ) -> AlpacaOrderWs<'a> {
5578 AlpacaOrderWs {
5579 id: SmolStr::new(id),
5580 client_order_id: None,
5581 symbol: SmolStr::new(symbol),
5582 qty: Some("2"),
5583 filled_qty: Some(filled_qty),
5584 side: SmolStr::new(side),
5585 order_type: SmolStr::new("limit"),
5586 time_in_force: SmolStr::new("day"),
5587 limit_price: Some("100.00"),
5588 stop_price: None,
5589 trail_percent: None,
5590 trail_price: None,
5591 status: SmolStr::new("partially_filled"),
5592 }
5593 }
5594
5595 fn sole_event(events: [Option<UnindexedAccountEvent>; 2]) -> Option<UnindexedAccountEvent> {
5600 let [first, second] = events;
5601 assert!(
5602 second.is_none(),
5603 "expected a single event, got a second: {second:?}"
5604 );
5605 first
5606 }
5607
5608 fn fill_events(
5610 events: [Option<UnindexedAccountEvent>; 2],
5611 ) -> (
5612 Trade<AssetNameExchange, InstrumentNameExchange>,
5613 crate::order::Order<
5614 ExchangeId,
5615 InstrumentNameExchange,
5616 OrderState<AssetNameExchange, InstrumentNameExchange>,
5617 >,
5618 ) {
5619 let [first, second] = events;
5620 let first = first.expect("a fill frame must produce an execution");
5621 let second = second.expect("a fill frame must produce an order snapshot");
5622 let AccountEventKind::Trade(trade) = first.kind else {
5623 panic!("first event must be the Trade, got {:?}", first.kind);
5624 };
5625 let AccountEventKind::OrderSnapshot(rustrade_integration::collection::snapshot::Snapshot(
5626 order,
5627 )) = second.kind
5628 else {
5629 panic!(
5630 "second event must be the OrderSnapshot, got {:?}",
5631 second.kind
5632 );
5633 };
5634 (trade, order)
5635 }
5636
5637 #[test]
5638 fn test_convert_trade_update_fill_produces_trade_with_execution_id() {
5639 let update = AlpacaTradeUpdate {
5640 event: SmolStr::new("fill"),
5641 execution_id: Some("exec-1"),
5642 previous_execution_id: None,
5643 order: make_order_ws("ord-1", "SPY", "buy", "1"),
5644 price: Some("150.00"),
5645 qty: Some("1"),
5646 timestamp: Some("2025-04-18T14:30:00Z"),
5647 };
5648 let (trade, order) = fill_events(convert_trade_update(update));
5649 assert_eq!(trade.id.0.as_str(), "exec-1");
5651 assert_eq!(trade.price, Decimal::from_str("150.00").unwrap());
5652 assert_eq!(trade.quantity, Decimal::from_str("1").unwrap());
5653
5654 let OrderState::Active(crate::order::state::ActiveOrderState::Open(open)) = order.state
5657 else {
5658 panic!("expected an Open snapshot, got {:?}", order.state);
5659 };
5660 assert_eq!(open.filled_quantity, Decimal::from_str("1").unwrap());
5661 assert_eq!(open.id.assigned().map(|id| id.0.as_str()), Some("ord-1"));
5662 }
5663
5664 #[test]
5665 fn test_convert_trade_update_partial_fill() {
5666 let update = AlpacaTradeUpdate {
5667 event: SmolStr::new("partial_fill"),
5668 execution_id: None,
5669 previous_execution_id: None,
5670 order: make_order_ws("ord-2", "AAPL", "sell", "0.5"),
5671 price: Some("200.00"),
5672 qty: Some("0.5"),
5673 timestamp: None,
5674 };
5675 let (trade, order) = fill_events(convert_trade_update(update));
5676 assert_eq!(trade.quantity, Decimal::from_str("0.5").unwrap());
5677
5678 let OrderState::Active(crate::order::state::ActiveOrderState::Open(open)) = order.state
5681 else {
5682 panic!("expected an Open snapshot, got {:?}", order.state);
5683 };
5684 assert_eq!(open.filled_quantity, Decimal::from_str("0.5").unwrap());
5685 assert_eq!(
5686 open.quantity_remaining(order.quantity),
5687 Decimal::from_str("1.5").unwrap()
5688 );
5689 }
5690
5691 #[test]
5695 fn a_full_fill_snapshot_reports_nothing_left_to_fill() {
5696 let update = AlpacaTradeUpdate {
5697 event: SmolStr::new("fill"),
5698 execution_id: None,
5699 previous_execution_id: None,
5700 order: AlpacaOrderWs {
5702 status: SmolStr::new("filled"),
5703 ..make_order_ws("ord-full", "SPY", "buy", "2")
5704 },
5705 price: Some("150.00"),
5706 qty: Some("1"),
5707 timestamp: Some("2025-04-18T14:30:00Z"),
5708 };
5709 let (_trade, order) = fill_events(convert_trade_update(update));
5710 let OrderState::Active(crate::order::state::ActiveOrderState::Open(open)) = order.state
5711 else {
5712 panic!("expected an Open snapshot, got {:?}", order.state);
5713 };
5714 assert_eq!(open.filled_quantity, Decimal::from_str("2").unwrap());
5715 assert!(
5716 open.quantity_remaining(order.quantity).is_zero(),
5717 "a completed order must report nothing left to fill so the engine retires it"
5718 );
5719 }
5720
5721 #[test]
5725 fn a_fill_for_an_order_no_longer_working_emits_the_execution_without_a_snapshot() {
5726 for status in [
5727 "canceled",
5728 "expired",
5729 "rejected",
5730 "done_for_day",
5731 "pending_cancel",
5732 ] {
5733 let update = AlpacaTradeUpdate {
5734 event: SmolStr::new("partial_fill"),
5735 execution_id: None,
5736 previous_execution_id: None,
5737 order: AlpacaOrderWs {
5738 status: SmolStr::new(status),
5739 ..make_order_ws("ord-late", "SPY", "buy", "1")
5740 },
5741 price: Some("150.00"),
5742 qty: Some("1"),
5743 timestamp: None,
5744 };
5745 let [trade, snapshot] = convert_trade_update(update);
5746 assert!(
5747 matches!(trade.map(|e| e.kind), Some(AccountEventKind::Trade(_))),
5748 "the execution must still be reported for status {status}"
5749 );
5750 assert!(
5751 snapshot.is_none(),
5752 "status {status} must not produce an order snapshot"
5753 );
5754 }
5755 }
5756
5757 #[test]
5760 fn a_notional_order_fill_emits_the_execution_without_a_snapshot() {
5761 let update = AlpacaTradeUpdate {
5762 event: SmolStr::new("partial_fill"),
5763 execution_id: None,
5764 previous_execution_id: None,
5765 order: AlpacaOrderWs {
5766 qty: None,
5767 ..make_order_ws("ord-notional-ws", "SPY", "buy", "1")
5768 },
5769 price: Some("150.00"),
5770 qty: Some("1"),
5771 timestamp: None,
5772 };
5773 let [trade, snapshot] = convert_trade_update(update);
5774 assert!(matches!(
5775 trade.map(|e| e.kind),
5776 Some(AccountEventKind::Trade(_))
5777 ));
5778 assert!(
5779 snapshot.is_none(),
5780 "a notional order has no quantity to snapshot against"
5781 );
5782 }
5783
5784 #[test]
5785 fn test_convert_trade_update_new_order_produces_snapshot() {
5786 let update = AlpacaTradeUpdate {
5787 event: SmolStr::new("new"),
5788 execution_id: None,
5789 previous_execution_id: None,
5790 order: AlpacaOrderWs {
5791 id: SmolStr::new("ord-new"),
5792 client_order_id: Some(SmolStr::new("cid-1")),
5793 symbol: SmolStr::new("AAPL"),
5794 qty: Some("10"),
5795 filled_qty: Some("0"),
5796 side: SmolStr::new("buy"),
5797 order_type: SmolStr::new("limit"),
5798 time_in_force: SmolStr::new("day"),
5799 limit_price: Some("150.00"),
5800 stop_price: None,
5801 trail_percent: None,
5802 trail_price: None,
5803 status: SmolStr::new("new"),
5804 },
5805 price: None,
5806 qty: None,
5807 timestamp: Some("2025-04-18T14:30:00Z"),
5808 };
5809 let event =
5810 sole_event(convert_trade_update(update)).expect("new event should produce an event");
5811 assert!(matches!(event.kind, AccountEventKind::OrderSnapshot(_)));
5812 }
5813
5814 #[test]
5815 fn test_convert_trade_update_canceled_produces_cancel() {
5816 let update = AlpacaTradeUpdate {
5817 event: SmolStr::new("canceled"),
5818 execution_id: None,
5819 previous_execution_id: None,
5820 order: make_order_ws("ord-3", "AAPL", "sell", "0"),
5821 price: None,
5822 qty: None,
5823 timestamp: Some("2025-04-18T14:30:00Z"),
5824 };
5825 let event =
5826 sole_event(convert_trade_update(update)).expect("canceled should produce an event");
5827 let AccountEventKind::OrderCancelled(response) = event.kind else {
5828 panic!("expected OrderCancelled, got {:?}", event.kind);
5829 };
5830 assert!(response.state.is_ok());
5831 }
5832
5833 #[test]
5835 fn a_canceled_update_carries_its_fill_or_none() {
5836 let cancelled = |filled_qty: Option<&'static str>| {
5837 let mut order = make_order_ws("ord-3", "AAPL", "sell", "0");
5838 order.filled_qty = filled_qty;
5839 let update = AlpacaTradeUpdate {
5840 event: SmolStr::new("canceled"),
5841 execution_id: None,
5842 previous_execution_id: None,
5843 order,
5844 price: None,
5845 qty: None,
5846 timestamp: Some("2025-04-18T14:30:00Z"),
5847 };
5848 let event =
5849 sole_event(convert_trade_update(update)).expect("canceled should produce an event");
5850 let AccountEventKind::OrderCancelled(response) = event.kind else {
5851 panic!("expected OrderCancelled, got {:?}", event.kind);
5852 };
5853 let Ok(cancelled) = response.state else {
5854 panic!("expected Ok, got {:?}", response.state);
5855 };
5856 cancelled.filled_quantity
5857 };
5858
5859 assert_eq!(cancelled(Some("1")), Some(Decimal::ONE));
5860 assert_eq!(cancelled(Some("0")), Some(Decimal::ZERO));
5861 assert_eq!(cancelled(None), None);
5862 assert_eq!(cancelled(Some("not a number")), None);
5863 }
5864
5865 #[test]
5868 fn an_ended_order_with_an_unparseable_fill_reports_it_unknown() {
5869 let key = OrderKey::new(
5870 ExchangeId::AlpacaBroker,
5871 InstrumentNameExchange::new("AAPL"),
5872 StrategyId::new("strategy"),
5873 ClientOrderId::new("c"),
5874 );
5875 let ended = |status: &str| {
5876 let mut order = make_order_response("o1", "AAPL");
5877 order.status = Some(status.to_string());
5878 order.filled_qty = "not a number".to_string();
5879 convert_ended_order(&order, &key).map(|order| order.state)
5880 };
5881
5882 assert!(matches!(
5883 ended("canceled"),
5884 Some(InactiveOrderState::Cancelled(Cancelled {
5885 filled_quantity: None,
5886 ..
5887 }))
5888 ));
5889 assert!(matches!(
5890 ended("expired"),
5891 Some(InactiveOrderState::Expired(Expired {
5892 filled_quantity: None,
5893 ..
5894 }))
5895 ));
5896 assert!(matches!(
5897 ended("filled"),
5898 Some(InactiveOrderState::FullyFilled(Filled { filled_quantity, .. }))
5899 if filled_quantity == Decimal::ONE
5900 ));
5901 }
5902
5903 #[test]
5904 fn test_convert_trade_update_rejected_produces_error() {
5905 let update = AlpacaTradeUpdate {
5906 event: SmolStr::new("rejected"),
5907 execution_id: None,
5908 previous_execution_id: None,
5909 order: make_order_ws("ord-4", "SPY", "buy", "0"),
5910 price: None,
5911 qty: None,
5912 timestamp: None,
5913 };
5914 let event =
5915 sole_event(convert_trade_update(update)).expect("rejected should produce an event");
5916 let AccountEventKind::OrderCancelled(response) = event.kind else {
5917 panic!("expected OrderCancelled, got {:?}", event.kind);
5918 };
5919 assert!(response.state.is_err());
5920 }
5921
5922 #[test]
5923 fn test_convert_open_order_notional_qty_none_is_skipped() {
5924 let order = AlpacaOrderResponse {
5927 id: "ord-notional".to_string(),
5928 client_order_id: None,
5929 symbol: "SPY".to_string(),
5930 qty: None,
5931 filled_qty: "0".to_string(),
5932 filled_avg_price: None,
5933 status: None,
5934 side: "buy".to_string(),
5935 order_type: "market".to_string(),
5936 time_in_force: "day".to_string(),
5937 limit_price: None,
5938 stop_price: None,
5939 trail_percent: None,
5940 trail_price: None,
5941 created_at: "2025-04-18T14:30:00Z".to_string(),
5942 updated_at: None,
5943 };
5944 assert!(convert_open_order(&order).is_none());
5945 }
5946
5947 #[test]
5948 fn test_convert_activity_to_trade_bad_price_returns_none() {
5949 let activity = AlpacaActivity {
5953 id: "act-1".to_string(),
5954 order_id: "ord-1".to_string(),
5955 symbol: "SPY250418C00450000".to_string(),
5956 side: "buy".to_string(),
5957 price: "not-a-number".to_string(),
5958 qty: "1".to_string(),
5959 transaction_time: "2025-04-18T14:30:00Z".to_string(),
5960 cum_qty: Some("1".to_string()),
5961 };
5962 assert!(convert_activity_to_trade(&activity).is_none());
5963 }
5964
5965 #[test]
5966 fn test_convert_positions_to_balances_crypto() {
5967 let positions = vec![
5968 AlpacaPosition {
5970 qty_available: dec!(0.4),
5971 ..alpaca_position("BTCUSD", "crypto", AlpacaPositionSide::Long, dec!(0.5))
5972 },
5973 alpaca_position("ETH/USD", "crypto", AlpacaPositionSide::Long, dec!(2.0)),
5975 alpaca_position("AAPL", "us_equity", AlpacaPositionSide::Long, dec!(10)),
5977 ];
5978
5979 let balances = convert_positions_to_balances(&positions, &[]);
5981 assert_eq!(balances.len(), 2, "only crypto positions returned");
5982 assert_eq!(balances[0].asset.name().as_str(), "BTC");
5983 assert_eq!(balances[1].asset.name().as_str(), "ETH");
5984 assert_eq!(balances[0].balance.total, dec!(0.5));
5986 assert_eq!(balances[0].balance.free, dec!(0.4));
5987
5988 let btc_only = vec![AssetNameExchange::new("btc")];
5990 let balances = convert_positions_to_balances(&positions, &btc_only);
5991 assert_eq!(balances.len(), 1);
5992 assert_eq!(balances[0].asset.name().as_str(), "BTC");
5993 }
5994
5995 #[test]
5996 fn test_crypto_position_asset() {
5997 assert_eq!(crypto_position_asset("BTCUSD"), "BTC");
5998 assert_eq!(
5999 crypto_position_asset("btcusd"),
6000 "btc",
6001 "suffix matched ignoring case"
6002 );
6003 assert_eq!(
6004 crypto_position_asset("USDTUSD"),
6005 "USDT",
6006 "only one USD is stripped"
6007 );
6008 assert_eq!(crypto_position_asset("BTC/USD"), "BTC");
6009 assert_eq!(crypto_position_asset("BTC/USDT"), "BTC");
6010 assert_eq!(crypto_position_asset("€USD"), "€");
6011 assert_eq!(crypto_position_asset("USD"), "USD");
6013 assert_eq!(crypto_position_asset("/USD"), "/USD");
6014 assert_eq!(crypto_position_asset("BTCUSDT"), "BTCUSDT");
6015 assert_eq!(
6016 crypto_position_asset("BTCێ"),
6017 "BTCێ",
6018 "no char boundary at the suffix"
6019 );
6020 assert_eq!(crypto_position_asset(""), "");
6021 }
6022
6023 fn alpaca_position(
6025 symbol: &str,
6026 asset_class: &str,
6027 side: AlpacaPositionSide,
6028 qty: Decimal,
6029 ) -> AlpacaPosition {
6030 AlpacaPosition {
6031 symbol: symbol.into(),
6032 asset_class: asset_class.into(),
6033 side,
6034 qty,
6035 qty_available: qty,
6036 avg_entry_price: Some(dec!(100)),
6037 unrealized_pl: Some(dec!(0)),
6038 }
6039 }
6040
6041 #[test]
6042 fn test_convert_positions_sign_comes_from_side_not_qty() {
6043 let now = Utc::now();
6044 let positions = vec![
6045 alpaca_position("AAPL", "us_equity", AlpacaPositionSide::Long, dec!(10)),
6046 alpaca_position("TSLA", "us_equity", AlpacaPositionSide::Short, dec!(-3)),
6047 alpaca_position("MSFT", "us_equity", AlpacaPositionSide::Short, dec!(4)),
6049 ];
6050
6051 let converted = convert_positions(&positions, now).unwrap();
6052 let quantities: Vec<_> = converted.iter().map(|(s, p)| (*s, p.quantity)).collect();
6053 assert_eq!(
6054 quantities,
6055 vec![("AAPL", dec!(10)), ("TSLA", dec!(-3)), ("MSFT", dec!(-4))]
6056 );
6057 assert!(converted.iter().all(|(_, p)| p.time_exchange == now));
6058 }
6059
6060 #[test]
6061 fn test_convert_positions_leaves_out_crypto_and_flat() {
6062 let positions = vec![
6063 alpaca_position("BTC/USD", "crypto", AlpacaPositionSide::Long, dec!(0.5)),
6064 alpaca_position("AAPL", "us_equity", AlpacaPositionSide::Long, dec!(0)),
6065 alpaca_position("SPY", "us_equity", AlpacaPositionSide::Long, dec!(1)),
6066 ];
6067
6068 let converted = convert_positions(&positions, Utc::now()).unwrap();
6069 let symbols: Vec<_> = converted.iter().map(|(s, _)| *s).collect();
6070 assert_eq!(symbols, vec!["SPY"]);
6071 }
6072
6073 #[test]
6074 fn test_alpaca_position_decodes_equity_option_and_fractional() {
6075 let positions: Vec<AlpacaPosition> = serde_json::from_value(serde_json::json!([
6076 {
6077 "asset_id": "00000000-0000-0000-0000-000000000001",
6078 "symbol": "XYZ",
6079 "exchange": "NASDAQ",
6080 "asset_class": "us_equity",
6081 "qty": "-7",
6082 "qty_available": "-7",
6083 "side": "short",
6084 "avg_entry_price": "50.25",
6085 "market_value": "-351.75",
6086 "unrealized_pl": "0",
6087 },
6088 {
6089 "symbol": "XYZ271217C00055000",
6090 "asset_class": "us_option",
6091 "qty": "2",
6092 "qty_available": "2",
6093 "side": "long",
6094 "avg_entry_price": "1.35",
6095 "unrealized_pl": "-20",
6096 },
6097 {
6098 "symbol": "ABC",
6099 "asset_class": "us_equity",
6100 "qty": "0.125",
6101 "qty_available": "0.125",
6102 "side": "long",
6103 "avg_entry_price": "80",
6104 "unrealized_pl": null,
6105 },
6106 ]))
6107 .unwrap();
6108
6109 let converted = convert_positions(&positions, Utc::now()).unwrap();
6110 let [
6111 (short_sym, short),
6112 (option_sym, option),
6113 (fraction_sym, fraction),
6114 ] = converted.as_slice()
6115 else {
6116 panic!("expected three positions, got {converted:?}");
6117 };
6118
6119 assert_eq!(*short_sym, "XYZ");
6120 assert_eq!(short.quantity, dec!(-7));
6121 assert_eq!(short.entry_price, Some(dec!(50.25)));
6122 assert!(short.is_short());
6123
6124 assert_eq!(*option_sym, "XYZ271217C00055000");
6125 assert_eq!(option.quantity, dec!(2));
6126 assert_eq!(option.entry_price, Some(dec!(1.35)), "per-share premium");
6127 assert_eq!(option.unrealized_pnl, Some(dec!(-20)));
6128
6129 assert_eq!(*fraction_sym, "ABC");
6130 assert_eq!(fraction.quantity, dec!(0.125));
6131 assert_eq!(fraction.unrealized_pnl, None);
6132
6133 for (_, p) in &converted {
6134 assert_eq!(p.margin_used, None);
6135 assert_eq!(p.liquidation_price, None);
6136 assert_eq!(p.leverage, None);
6137 }
6138 }
6139
6140 #[test]
6143 fn test_alpaca_position_unknown_side_fails_only_where_a_direction_is_needed() {
6144 let positions: Vec<AlpacaPosition> = serde_json::from_value(serde_json::json!([
6145 {
6146 "symbol": "BTC/USD",
6147 "asset_class": "crypto",
6148 "qty": "0.5",
6149 "qty_available": "0.5",
6150 },
6151 {
6152 "symbol": "XYZ",
6153 "asset_class": "us_equity",
6154 "qty": "1",
6155 "qty_available": "1",
6156 "side": "sideways",
6157 },
6158 ]))
6159 .unwrap();
6160 assert_eq!(positions[0].side, AlpacaPositionSide::Unknown);
6161 assert_eq!(positions[1].side, AlpacaPositionSide::Unknown);
6162
6163 let balances = convert_positions_to_balances(&positions, &[]);
6164 assert_eq!(balances.len(), 1);
6165 assert_eq!(balances[0].balance.total, dec!(0.5));
6166
6167 let result = convert_positions(&positions, Utc::now());
6168 assert!(
6169 matches!(&result, Err(UnindexedClientError::Internal(msg)) if msg.contains("XYZ")),
6170 "an unknown side must not guess a direction: {result:?}"
6171 );
6172 }
6173
6174 #[test]
6176 fn order_state_time_prefers_updated_at() {
6177 let mut resp = make_order_response("ord-1", "SPY");
6178 resp.created_at = "2025-04-18T14:30:00Z".to_string();
6179 resp.updated_at = Some("2025-04-18T15:45:00Z".to_string());
6180 assert_eq!(
6181 order_state_time(&resp),
6182 parse_timestamp("2025-04-18T15:45:00Z").unwrap(),
6183 "updated_at must win over created_at"
6184 );
6185
6186 resp.updated_at = None;
6188 assert_eq!(
6189 order_state_time(&resp),
6190 parse_timestamp("2025-04-18T14:30:00Z").unwrap(),
6191 "absent updated_at falls back to created_at"
6192 );
6193
6194 resp.updated_at = Some("not-a-timestamp".to_string());
6197 assert_eq!(
6198 order_state_time(&resp),
6199 parse_timestamp("2025-04-18T14:30:00Z").unwrap(),
6200 "unparseable updated_at falls back to created_at"
6201 );
6202 }
6203
6204 #[test]
6207 fn a_placement_response_without_a_status_reads_from_what_filled() {
6208 let instrument = InstrumentNameExchange::new("SPY");
6209 let mut resp = make_order_response("ord-1", "SPY");
6210 resp.filled_qty = "0.4".to_string();
6211 let OrderState::Active(ActiveOrderState::Open(open)) =
6212 placed_order_state(&resp, &instrument, Decimal::ONE)
6213 else {
6214 panic!("expected Open");
6215 };
6216 assert_eq!(open.filled_quantity, Decimal::new(4, 1));
6217
6218 resp.filled_qty = "1".to_string();
6219 resp.filled_avg_price = Some("100.5".to_string());
6220 let state = placed_order_state(&resp, &instrument, Decimal::ONE);
6221 let OrderState::Inactive(InactiveOrderState::FullyFilled(filled)) = state else {
6222 panic!("expected FullyFilled, got {state:?}");
6223 };
6224 assert_eq!(filled.filled_quantity, Decimal::ONE);
6225 assert_eq!(filled.avg_price, Some(Decimal::new(1005, 1)));
6226 }
6227
6228 fn make_order_response(id: &str, symbol: &str) -> AlpacaOrderResponse {
6229 AlpacaOrderResponse {
6230 id: id.to_string(),
6231 client_order_id: None,
6232 symbol: symbol.to_string(),
6233 qty: Some("1".to_string()),
6234 filled_qty: "0".to_string(),
6235 filled_avg_price: None,
6236 status: None,
6237 side: "buy".to_string(),
6238 order_type: "limit".to_string(),
6239 time_in_force: "day".to_string(),
6240 limit_price: Some("100.00".to_string()),
6241 stop_price: None,
6242 trail_percent: None,
6243 trail_price: None,
6244 created_at: "2025-04-18T14:30:00Z".to_string(),
6245 updated_at: None,
6246 }
6247 }
6248
6249 #[test]
6250 fn test_build_instrument_snapshots_empty_instruments_returns_only_with_orders() {
6251 let orders = vec![
6252 make_order_response("o1", "AAPL"),
6253 make_order_response("o2", "SPY"),
6254 ];
6255 let snapshots = build_instrument_snapshots(orders, Vec::new(), &[]);
6256 assert_eq!(snapshots.len(), 2);
6257 let symbols: Vec<&str> = snapshots
6258 .iter()
6259 .map(|s| s.instrument.name().as_str())
6260 .collect();
6261 assert!(symbols.contains(&"AAPL"));
6262 assert!(symbols.contains(&"SPY"));
6263 }
6264
6265 #[test]
6266 fn test_build_instrument_snapshots_requested_instrument_no_orders_gets_empty_snapshot() {
6267 let orders = vec![make_order_response("o1", "AAPL")];
6270 let instruments = vec![
6271 InstrumentNameExchange::new("AAPL"),
6272 InstrumentNameExchange::new("SPY"),
6273 ];
6274 let snapshots = build_instrument_snapshots(orders, Vec::new(), &instruments);
6275 assert_eq!(snapshots.len(), 2);
6276 let spy = snapshots
6277 .iter()
6278 .find(|s| s.instrument.name().as_str() == "SPY")
6279 .expect("SPY snapshot must be present even with no orders");
6280 assert!(spy.orders.is_empty());
6281 }
6282
6283 fn make_notional_order_response(id: &str, symbol: &str) -> AlpacaOrderResponse {
6285 AlpacaOrderResponse {
6286 qty: None,
6287 ..make_order_response(id, symbol)
6288 }
6289 }
6290
6291 fn snapshot_for<'a>(
6292 snapshots: &'a [InstrumentAccountSnapshot<
6293 ExchangeId,
6294 AssetNameExchange,
6295 InstrumentNameExchange,
6296 >],
6297 symbol: &str,
6298 ) -> &'a InstrumentAccountSnapshot<ExchangeId, AssetNameExchange, InstrumentNameExchange> {
6299 snapshots
6300 .iter()
6301 .find(|snapshot| snapshot.instrument.name().as_str() == symbol)
6302 .unwrap_or_else(|| panic!("no snapshot for {symbol}"))
6303 }
6304
6305 #[test]
6309 fn test_build_instrument_snapshots_declares_complete_lists_complete() {
6310 let instruments = [
6311 InstrumentNameExchange::new("AAPL"),
6312 InstrumentNameExchange::new("SPY"),
6313 ];
6314 for requested in [&instruments[..], &[]] {
6315 let orders = vec![make_order_response("o1", "AAPL")];
6316 let snapshots = build_instrument_snapshots(orders, Vec::new(), requested);
6317 assert!(!snapshots.is_empty());
6318 assert!(snapshots.iter().all(|snapshot| snapshot.orders_complete));
6319 }
6320 }
6321
6322 #[test]
6326 fn test_build_instrument_snapshots_notional_order_leaves_only_its_instrument_incomplete() {
6327 let orders = vec![
6328 make_order_response("o1", "AAPL"),
6329 make_order_response("o2", "SPY"),
6330 make_notional_order_response("o3", "SPY"),
6331 ];
6332 let instruments = [
6333 InstrumentNameExchange::new("AAPL"),
6334 InstrumentNameExchange::new("SPY"),
6335 ];
6336 let snapshots = build_instrument_snapshots(orders, Vec::new(), &instruments);
6337
6338 assert!(snapshot_for(&snapshots, "AAPL").orders_complete);
6339 let spy = snapshot_for(&snapshots, "SPY");
6340 assert!(!spy.orders_complete);
6341 assert_eq!(spy.orders.len(), 1);
6342 }
6343
6344 #[test]
6347 fn test_build_instrument_snapshots_unfiltered_lists_an_instrument_with_only_a_notional_order() {
6348 let orders = vec![make_notional_order_response("o1", "SPY")];
6349 let snapshots = build_instrument_snapshots(orders, Vec::new(), &[]);
6350
6351 let spy = snapshot_for(&snapshots, "SPY");
6352 assert!(!spy.orders_complete);
6353 assert!(spy.orders.is_empty());
6354 }
6355
6356 #[test]
6357 fn test_build_instrument_snapshots_non_requested_instrument_excluded() {
6358 let orders = vec![
6361 make_order_response("o1", "AAPL"),
6362 make_order_response("o2", "MSFT"), ];
6364 let instruments = vec![InstrumentNameExchange::new("AAPL")];
6365 let snapshots = build_instrument_snapshots(orders, Vec::new(), &instruments);
6366 assert_eq!(snapshots.len(), 1);
6367 assert_eq!(snapshots[0].instrument.name().as_str(), "AAPL");
6368 }
6369
6370 fn position(quantity: Decimal) -> Position {
6371 Position::new(
6372 quantity,
6373 Some(dec!(100)),
6374 None,
6375 None,
6376 None,
6377 None,
6378 Utc::now(),
6379 )
6380 }
6381
6382 #[test]
6383 fn test_build_instrument_snapshots_unfiltered_lists_positions_without_orders() {
6384 let orders = vec![
6385 make_order_response("o1", "AAPL"),
6386 make_order_response("o2", "MSFT"),
6387 make_order_response("o3", "BTC/USD"),
6388 ];
6389 let positions = vec![("AAPL", position(dec!(5))), ("TSLA", position(dec!(-2)))];
6390
6391 let snapshots = build_instrument_snapshots(orders, positions, &[]);
6392 let listed: Vec<_> = snapshots
6393 .iter()
6394 .map(|s| {
6395 (
6396 s.instrument.name().as_str(),
6397 s.orders.len(),
6398 s.position.quantity(),
6399 )
6400 })
6401 .collect();
6402 assert_eq!(
6404 listed,
6405 vec![
6406 ("AAPL", 1, Some(dec!(5))),
6407 ("MSFT", 1, Some(Decimal::ZERO)),
6408 ("BTC/USD", 1, None),
6409 ("TSLA", 0, Some(dec!(-2)))
6410 ]
6411 );
6412 assert!(snapshots.iter().all(|s| s.orders_complete));
6413 }
6414
6415 #[test]
6416 fn test_build_instrument_snapshots_filtered_attaches_only_requested_positions() {
6417 let positions = vec![("AAPL", position(dec!(5))), ("TSLA", position(dec!(-2)))];
6418 let instruments = vec![
6419 InstrumentNameExchange::new("tsla"),
6420 InstrumentNameExchange::new("MSFT"),
6421 InstrumentNameExchange::new("BTC/USD"),
6422 InstrumentNameExchange::new("TSLA"),
6423 ];
6424
6425 let snapshots = build_instrument_snapshots(Vec::new(), positions, &instruments);
6426 let listed: Vec<_> = snapshots
6427 .iter()
6428 .map(|s| (s.instrument.name().as_str(), s.position.quantity()))
6429 .collect();
6430 assert_eq!(
6434 listed,
6435 vec![
6436 ("tsla", Some(dec!(-2))),
6437 ("MSFT", Some(Decimal::ZERO)),
6438 ("BTC/USD", None),
6439 ("TSLA", None)
6440 ]
6441 );
6442 assert!(!snapshots[3].orders_complete);
6443 }
6444
6445 #[test]
6451 fn test_recover_fills_dedup_key_matches_ws_path() {
6452 let order_id = "ord-1";
6453
6454 let ws_keys: Vec<SmolStr> = ["1", "2"]
6457 .iter()
6458 .map(|filled_qty| {
6459 early_dedup_key(&AlpacaTradeUpdate {
6460 event: SmolStr::new("partial_fill"),
6461 execution_id: None,
6462 previous_execution_id: None,
6463 order: make_order_ws(order_id, "SPY", "buy", filled_qty),
6464 price: Some("150.00"),
6465 qty: Some("1"),
6466 timestamp: None,
6467 })
6468 })
6469 .collect();
6470
6471 let mut cumulative = Decimal::ZERO;
6474 let rest_keys: Vec<SmolStr> = ["1", "1"]
6475 .iter()
6476 .map(|exec_qty| {
6477 cumulative += Decimal::from_str(exec_qty).unwrap();
6478 fill_dedup_key(order_id, cumulative)
6479 })
6480 .collect();
6481
6482 assert_eq!(
6483 ws_keys, rest_keys,
6484 "REST recovery dedup keys must match WS path keys for cross-source dedup to work"
6485 );
6486 assert_eq!(ws_keys[0].as_str(), "ord-1:1");
6487 assert_eq!(ws_keys[1].as_str(), "ord-1:2");
6488 }
6489
6490 #[test]
6495 fn a_ws_fill_without_an_execution_id_is_identified_by_its_dedup_key() {
6496 let update = AlpacaTradeUpdate {
6497 event: SmolStr::new("fill"),
6498 execution_id: None,
6499 previous_execution_id: None,
6500 order: make_order_ws("ord-abc", "SPY", "buy", "5"),
6501 price: Some("150.00"),
6502 qty: Some("5"),
6503 timestamp: None,
6504 };
6505
6506 let early_key = early_dedup_key(&update);
6508
6509 let [event, _snapshot] = convert_trade_update(update);
6511 let Some(UnindexedAccountEvent {
6512 kind: AccountEventKind::Trade(trade),
6513 ..
6514 }) = event
6515 else {
6516 panic!("fill should produce an execution");
6517 };
6518
6519 assert_eq!(
6520 trade.id.0.as_str(),
6521 early_key.as_str(),
6522 "without an execution_id the TradeId is the dedup key"
6523 );
6524 assert_eq!(early_key.as_str(), "ord-abc:5");
6525 }
6526
6527 #[test]
6532 fn early_dedup_key_normalizes_decimal_strings() {
6533 let update1 = AlpacaTradeUpdate {
6535 event: SmolStr::new("fill"),
6536 execution_id: None,
6537 previous_execution_id: None,
6538 order: AlpacaOrderWs {
6539 id: SmolStr::new("ord-x"),
6540 client_order_id: Some(SmolStr::new("cid")),
6541 symbol: SmolStr::new("AAPL"),
6542 qty: Some("10"),
6543 filled_qty: Some("1.00"),
6544 side: SmolStr::new("buy"),
6545 order_type: SmolStr::new("market"),
6546 time_in_force: SmolStr::new("day"),
6547 limit_price: None,
6548 stop_price: None,
6549 trail_percent: None,
6550 trail_price: None,
6551 status: SmolStr::new("filled"),
6552 },
6553 price: Some("100.00"),
6554 qty: Some("10"),
6555 timestamp: None,
6556 };
6557 assert_eq!(early_dedup_key(&update1).as_str(), "ord-x:1");
6558
6559 let update2 = AlpacaTradeUpdate {
6561 event: SmolStr::new("fill"),
6562 execution_id: None,
6563 previous_execution_id: None,
6564 order: AlpacaOrderWs {
6565 id: SmolStr::new("ord-x"),
6566 client_order_id: Some(SmolStr::new("cid")),
6567 symbol: SmolStr::new("AAPL"),
6568 qty: Some("10"),
6569 filled_qty: Some("1"),
6570 side: SmolStr::new("buy"),
6571 order_type: SmolStr::new("market"),
6572 time_in_force: SmolStr::new("day"),
6573 limit_price: None,
6574 stop_price: None,
6575 trail_percent: None,
6576 trail_price: None,
6577 status: SmolStr::new("filled"),
6578 },
6579 price: Some("100.00"),
6580 qty: Some("10"),
6581 timestamp: None,
6582 };
6583 assert_eq!(early_dedup_key(&update2).as_str(), "ord-x:1");
6584
6585 let update3 = AlpacaTradeUpdate {
6587 event: SmolStr::new("fill"),
6588 execution_id: None,
6589 previous_execution_id: None,
6590 order: AlpacaOrderWs {
6591 id: SmolStr::new("ord-x"),
6592 client_order_id: Some(SmolStr::new("cid")),
6593 symbol: SmolStr::new("AAPL"),
6594 qty: Some("10"),
6595 filled_qty: Some("1.0"),
6596 side: SmolStr::new("buy"),
6597 order_type: SmolStr::new("market"),
6598 time_in_force: SmolStr::new("day"),
6599 limit_price: None,
6600 stop_price: None,
6601 trail_percent: None,
6602 trail_price: None,
6603 status: SmolStr::new("filled"),
6604 },
6605 price: Some("100.00"),
6606 qty: Some("10"),
6607 timestamp: None,
6608 };
6609 assert_eq!(early_dedup_key(&update3).as_str(), "ord-x:1");
6610 }
6611
6612 #[test]
6617 fn process_ws_text_rejected_event_without_filled_qty_is_not_dropped() {
6618 let (tx, mut rx) = mpsc::unbounded_channel();
6619 let dedup = new_dedup_cache();
6620 let mut backoff = ExponentialBackoff::new();
6621
6622 let json = r#"{"stream":"trade_updates","data":{"event":"rejected","order":{"id":"test-rej-id","client_order_id":"test-cid","symbol":"AAPL","qty":"10","side":"buy","type":"limit","time_in_force":"day","limit_price":"100.00","status":"rejected"}}}"#;
6624
6625 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
6626 process_ws_text(json, &tx, &dedup, &known, &mut backoff);
6627
6628 let event = rx.try_recv()
6630 .expect("rejected event without filled_qty must produce an AccountEvent, not be silently dropped");
6631 assert!(
6632 matches!(event.kind, AccountEventKind::OrderCancelled(_)),
6633 "rejected event must map to OrderCancelled, got: {:?}",
6634 event.kind
6635 );
6636 }
6637
6638 #[test]
6641 fn process_ws_text_identifies_a_fill_by_its_execution_id() {
6642 let frame = |execution_id: &str| {
6643 format!(
6644 r#"{{"stream":"trade_updates","data":{{"event":"fill",{execution_id}"order":{{"id":"ord-1","client_order_id":"cid-1","symbol":"SPY","qty":"2","filled_qty":"2","side":"buy","type":"market","time_in_force":"day","status":"filled"}},"price":"100.00","qty":"2","timestamp":"2025-04-18T14:30:00Z"}}}}"#
6645 )
6646 };
6647 let cases = [
6648 (
6649 r#""execution_id":"524b1902-817e-446c-825b-a9fcfebbc17e","#,
6650 "524b1902-817e-446c-825b-a9fcfebbc17e",
6651 ),
6652 (r#""execution_id":null,"#, "ord-1:2"),
6653 (r#""execution_id":"","#, "ord-1:2"),
6654 ("", "ord-1:2"),
6655 ];
6656 for (field, expected) in cases {
6657 let (tx, mut rx) = mpsc::unbounded_channel();
6658 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
6659 process_ws_text(
6660 &frame(field),
6661 &tx,
6662 &new_dedup_cache(),
6663 &known,
6664 &mut ExponentialBackoff::new(),
6665 );
6666 let Ok(UnindexedAccountEvent {
6667 kind: AccountEventKind::Trade(trade),
6668 ..
6669 }) = rx.try_recv()
6670 else {
6671 panic!("the fill frame {field:?} produces a Trade");
6672 };
6673 assert_eq!(trade.id.0.as_str(), expected, "frame {field:?}");
6674 }
6675 }
6676
6677 fn amendment_frame(event: &str, fields: &str) -> String {
6680 format!(
6681 r#"{{"stream":"trade_updates","data":{{"event":"{event}",{fields}"order":{{"id":"ord-1","client_order_id":"cid-1","symbol":"SPY","qty":"2","filled_qty":"0","side":"buy","type":"market","time_in_force":"day","status":"filled"}},"timestamp":"2025-04-18T14:31:00Z"}}}}"#
6682 )
6683 }
6684
6685 fn stream_events(frames: &[String]) -> Vec<UnindexedAccountEvent> {
6687 let (tx, mut rx) = mpsc::unbounded_channel();
6688 let dedup = new_dedup_cache();
6689 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
6690 let mut backoff = ExponentialBackoff::new();
6691 for frame in frames {
6692 process_ws_text(frame, &tx, &dedup, &known, &mut backoff);
6693 }
6694 std::iter::from_fn(|| rx.try_recv().ok()).collect()
6695 }
6696
6697 fn amendment_of(
6699 frame: String,
6700 ) -> crate::trade::TradeAmendment<AssetNameExchange, InstrumentNameExchange> {
6701 let events = stream_events(&[frame]);
6702 let [
6703 UnindexedAccountEvent {
6704 kind: AccountEventKind::TradeAmended(amendment),
6705 ..
6706 },
6707 ] = <[_; 1]>::try_from(events).unwrap_or_else(|events| {
6708 panic!("one TradeAmended, got {events:?}");
6709 })
6710 else {
6711 panic!("expected TradeAmended");
6712 };
6713 amendment
6714 }
6715
6716 #[test]
6719 fn process_ws_text_reports_a_trade_bust() {
6720 let amendment = amendment_of(amendment_frame(
6721 "trade_bust",
6722 r#""execution_id":"exec-x","previous_execution_id":"exec-a","qty":"-2","#,
6723 ));
6724 assert_eq!(amendment.original, Some(TradeId::new("exec-a")));
6725 assert_eq!(amendment.order_id, OrderId::new("ord-1"));
6726 assert_eq!(amendment.instrument, InstrumentNameExchange::new("SPY"));
6727 assert_eq!(
6728 Some(amendment.time_exchange),
6729 parse_timestamp("2025-04-18T14:31:00Z")
6730 );
6731 assert_eq!(
6732 amendment.kind,
6733 TradeAmendmentKind::Busted {
6734 quantity: Some(Decimal::TWO)
6735 }
6736 );
6737 }
6738
6739 #[test]
6741 fn process_ws_text_reports_a_bust_that_names_no_fill() {
6742 let amendment = amendment_of(amendment_frame("trade_bust", ""));
6743 assert_eq!(amendment.original, None);
6744 assert_eq!(
6745 amendment.kind,
6746 TradeAmendmentKind::Busted { quantity: None }
6747 );
6748 }
6749
6750 #[test]
6753 fn process_ws_text_reports_a_trade_correct_with_its_replacement() {
6754 let amendment = amendment_of(amendment_frame(
6755 "trade_correct",
6756 r#""execution_id":"exec-c","previous_execution_id":"exec-a","price":"101.50","qty":"2","#,
6757 ));
6758 assert_eq!(amendment.original, Some(TradeId::new("exec-a")));
6759 let TradeAmendmentKind::Corrected { replacement } = amendment.kind else {
6760 panic!("a full replacement: {:?}", amendment.kind);
6761 };
6762 assert_eq!(replacement.id, TradeId::new("exec-c"));
6763 assert_eq!(replacement.order_id, OrderId::new("ord-1"));
6764 assert_eq!(replacement.side, Side::Buy);
6765 assert_eq!(replacement.price, Decimal::new(10150, 2));
6766 assert_eq!(replacement.quantity, Decimal::TWO);
6767 assert_eq!(replacement.order_filled_quantity, None);
6768 assert_eq!(replacement.time_exchange, amendment.time_exchange);
6769 }
6770
6771 #[test]
6774 fn process_ws_text_reports_a_correction_it_cannot_resolve() {
6775 let price = Some(Decimal::new(10150, 2));
6776 let cases = [
6777 (
6778 r#""execution_id":"exec-c","qty":"2","#,
6779 (Some("exec-c"), None, Some(Decimal::TWO)),
6780 ),
6781 (
6782 r#""price":"101.50","qty":"2","#,
6783 (None, price, Some(Decimal::TWO)),
6784 ),
6785 (
6786 r#""execution_id":"exec-c","price":"101.50","qty":"0","#,
6787 (Some("exec-c"), price, Some(Decimal::ZERO)),
6788 ),
6789 (
6790 r#""execution_id":"exec-c","price":"101.50","#,
6791 (Some("exec-c"), price, None),
6792 ),
6793 ];
6794 for (fields, (id, price, quantity)) in cases {
6795 let amendment = amendment_of(amendment_frame("trade_correct", fields));
6796 assert_eq!(
6797 amendment.kind,
6798 TradeAmendmentKind::CorrectedUnresolved {
6799 id: id.map(TradeId::new),
6800 price,
6801 quantity,
6802 },
6803 "{fields}"
6804 );
6805 }
6806 }
6807
6808 #[test]
6811 fn a_fill_after_a_bust_is_delivered_at_a_cumulative_seen_before() {
6812 let fill = |execution_id: &str| {
6813 format!(
6814 r#"{{"stream":"trade_updates","data":{{"event":"partial_fill","execution_id":"{execution_id}","order":{{"id":"ord-1","client_order_id":"cid-1","symbol":"SPY","qty":"4","filled_qty":"2","side":"buy","type":"market","time_in_force":"day","status":"partially_filled"}},"price":"100.00","qty":"2","timestamp":"2025-04-18T14:30:00Z"}}}}"#
6815 )
6816 };
6817 let events = stream_events(&[
6818 fill("exec-a"),
6819 amendment_frame(
6820 "trade_bust",
6821 r#""execution_id":"exec-x","previous_execution_id":"exec-a","qty":"-2","#,
6822 ),
6823 fill("exec-b"),
6824 ]);
6825
6826 let kinds: Vec<_> = events
6827 .iter()
6828 .filter_map(|event| match &event.kind {
6829 AccountEventKind::Trade(trade) => Some(format!("trade {}", trade.id)),
6830 AccountEventKind::TradeAmended(amendment) => {
6831 Some(format!("amended {:?}", amendment.original))
6832 }
6833 _ => None,
6834 })
6835 .collect();
6836 assert_eq!(
6837 kinds,
6838 vec![
6839 "trade exec-a",
6840 "amended Some(TradeId(\"exec-a\"))",
6841 "trade exec-b"
6842 ]
6843 );
6844 }
6845
6846 #[test]
6849 fn decimal_zero_normalize_is_zero_str() {
6850 assert_eq!(Decimal::ZERO.normalize().to_string(), "0");
6851 }
6852
6853 #[test]
6857 fn decimal_normalize_strips_trailing_zeros() {
6858 let from_rest = Decimal::from_str("1.00").unwrap().normalize();
6860 let from_ws = Decimal::from_str("1").unwrap().normalize();
6861 assert_eq!(from_rest.to_string(), from_ws.to_string());
6862 assert_eq!(from_rest.to_string(), "1");
6863
6864 assert_eq!(
6866 Decimal::from_str("100.000")
6867 .unwrap()
6868 .normalize()
6869 .to_string(),
6870 "100"
6871 );
6872 assert_eq!(
6873 Decimal::from_str("0.10").unwrap().normalize().to_string(),
6874 "0.1"
6875 );
6876 assert_eq!(
6877 Decimal::from_str("0.100").unwrap().normalize().to_string(),
6878 "0.1"
6879 );
6880 }
6881
6882 #[test]
6884 fn map_position_intent_open_buy_maps_to_buy_to_open() {
6885 assert_eq!(
6886 map_position_intent(Side::Buy, false),
6887 AlpacaPositionIntent::BuyToOpen
6888 );
6889 }
6890
6891 #[test]
6892 fn map_position_intent_open_sell_maps_to_sell_to_open() {
6893 assert_eq!(
6894 map_position_intent(Side::Sell, false),
6895 AlpacaPositionIntent::SellToOpen
6896 );
6897 }
6898
6899 #[test]
6900 fn map_position_intent_reduce_buy_maps_to_buy_to_close() {
6901 assert_eq!(
6902 map_position_intent(Side::Buy, true),
6903 AlpacaPositionIntent::BuyToClose
6904 );
6905 }
6906
6907 #[test]
6908 fn map_position_intent_reduce_sell_maps_to_sell_to_close() {
6909 assert_eq!(
6910 map_position_intent(Side::Sell, true),
6911 AlpacaPositionIntent::SellToClose
6912 );
6913 }
6914
6915 #[test]
6917 fn parse_order_error_401_maps_to_unauthenticated() {
6918 assert!(matches!(
6920 parse_order_error(reqwest::StatusCode::UNAUTHORIZED, "bad credentials"),
6921 UnindexedOrderError::Rejected(ApiError::Unauthenticated(_))
6922 ));
6923 }
6924
6925 #[test]
6926 fn parse_order_error_403_maps_to_unauthenticated() {
6927 assert!(matches!(
6930 parse_order_error(reqwest::StatusCode::FORBIDDEN, "account suspended"),
6931 UnindexedOrderError::Rejected(ApiError::Unauthenticated(_))
6932 ));
6933 }
6934
6935 #[test]
6936 fn parse_order_error_404_maps_to_order_rejected_with_not_found_prefix() {
6937 let err = parse_order_error(reqwest::StatusCode::NOT_FOUND, "order not found");
6938 let UnindexedOrderError::Rejected(ApiError::OrderRejected(msg)) = err else {
6939 panic!("expected OrderRejected, got {err:?}");
6940 };
6941 assert!(
6942 msg.contains("order not found"),
6943 "message should contain 'order not found': {msg}"
6944 );
6945 }
6946
6947 #[test]
6948 fn parse_order_error_422_insufficient_only_maps_to_balance_insufficient() {
6949 assert!(matches!(
6951 parse_order_error(
6952 reqwest::StatusCode::UNPROCESSABLE_ENTITY,
6953 "insufficient funds for this order"
6954 ),
6955 UnindexedOrderError::Rejected(ApiError::BalanceInsufficient(None, _))
6956 ));
6957 }
6958
6959 #[test]
6960 fn parse_order_error_429_maps_to_rate_limit() {
6961 assert!(matches!(
6962 parse_order_error(reqwest::StatusCode::TOO_MANY_REQUESTS, "rate limited"),
6963 UnindexedOrderError::Rejected(ApiError::RateLimit)
6964 ));
6965 }
6966
6967 #[test]
6971 fn parse_time_in_force_unknown_value_falls_back_to_good_until_end_of_day() {
6972 assert_eq!(
6973 parse_time_in_force("opg"),
6974 TimeInForce::GoodUntilEndOfDay,
6975 "unknown TIF must fall back to GoodUntilEndOfDay (with a warn! in production)"
6976 );
6977 }
6978
6979 #[test]
6982 fn build_instrument_snapshots_output_order_matches_instruments_slice() {
6983 let orders = vec![
6985 make_order_response("o1", "SPY"),
6986 make_order_response("o2", "AAPL"),
6987 make_order_response("o3", "MSFT"),
6988 ];
6989 let instruments = vec![
6991 InstrumentNameExchange::new("MSFT"),
6992 InstrumentNameExchange::new("AAPL"),
6993 ];
6994 let snapshots = build_instrument_snapshots(orders, Vec::new(), &instruments);
6995 assert_eq!(snapshots.len(), 2);
6996 assert_eq!(
6997 snapshots[0].instrument.name().as_str(),
6998 "MSFT",
6999 "first snapshot must be MSFT (first in instruments slice)"
7000 );
7001 assert_eq!(
7002 snapshots[1].instrument.name().as_str(),
7003 "AAPL",
7004 "second snapshot must be AAPL (second in instruments slice)"
7005 );
7006 }
7007
7008 mod http_tests {
7015 use super::super::*;
7016 use wiremock::matchers::{method, path};
7017 use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate};
7018
7019 struct Sequential {
7026 call: std::sync::atomic::AtomicU32,
7027 pages: Vec<serde_json::Value>,
7028 }
7029
7030 impl Sequential {
7031 fn new(pages: Vec<serde_json::Value>) -> Self {
7032 Self {
7033 call: std::sync::atomic::AtomicU32::new(0),
7034 pages,
7035 }
7036 }
7037 }
7038
7039 impl Respond for Sequential {
7040 fn respond(&self, _: &Request) -> ResponseTemplate {
7041 let i = self.call.fetch_add(1, std::sync::atomic::Ordering::Relaxed) as usize;
7042 let body = self.pages.get(i).unwrap_or_else(|| {
7043 panic!(
7044 "Sequential: request #{i} has no configured response \
7045 (only {} page(s) supplied)",
7046 self.pages.len()
7047 )
7048 });
7049 ResponseTemplate::new(200).set_body_json(body)
7050 }
7051 }
7052
7053 fn make_activities_json(count: usize, id_prefix: &str) -> serde_json::Value {
7055 serde_json::Value::Array(
7056 (0..count)
7057 .map(|i| {
7058 serde_json::json!({
7059 "id": format!("{id_prefix}-{i:05}"),
7060 "order_id": "ord-1",
7061 "symbol": "SPY",
7062 "side": "buy",
7063 "price": "100.00",
7064 "qty": "1",
7065 "transaction_time": "2025-04-18T14:30:00Z"
7066 })
7067 })
7068 .collect(),
7069 )
7070 }
7071
7072 fn make_orders_json(count: usize) -> serde_json::Value {
7074 serde_json::Value::Array(
7075 (0..count)
7076 .map(|i| {
7077 serde_json::json!({
7078 "id": format!("order-{i:05}"),
7079 "client_order_id": null,
7080 "symbol": "SPY",
7081 "qty": "1",
7082 "filled_qty": "0",
7083 "side": "buy",
7084 "type": "limit",
7085 "time_in_force": "day",
7086 "limit_price": "100.00",
7087 "created_at": "2025-04-18T14:30:00Z"
7088 })
7089 })
7090 .collect(),
7091 )
7092 }
7093
7094 #[tokio::test]
7099 async fn paginate_activities_single_page_below_max_returns_all_not_truncated() {
7100 let server = MockServer::start().await;
7101 Mock::given(method("GET"))
7102 .and(path("/v2/account/activities"))
7103 .respond_with(
7104 ResponseTemplate::new(200).set_body_json(make_activities_json(5, "act")),
7105 )
7106 .mount(&server)
7107 .await;
7108
7109 let http = reqwest::Client::new();
7110 let rl = RateLimitTracker::new();
7111 let result = paginate_activities(&http, &rl, &server.uri(), recovery_after(), None)
7112 .await
7113 .unwrap();
7114
7115 assert_eq!(result.activities.len(), 5);
7116 assert_eq!(result.resume, None);
7117 assert_eq!(server.received_requests().await.unwrap().len(), 1);
7118 }
7119
7120 #[tokio::test]
7123 async fn paginate_activities_exactly_page_size_items_fetches_second_page() {
7124 let server = MockServer::start().await;
7125 Mock::given(method("GET"))
7126 .and(path("/v2/account/activities"))
7127 .respond_with(Sequential::new(vec![
7128 make_activities_json(ALPACA_MAX_ACTIVITIES, "act"),
7129 serde_json::json!([]), ]))
7131 .mount(&server)
7132 .await;
7133
7134 let http = reqwest::Client::new();
7135 let rl = RateLimitTracker::new();
7136 let result = paginate_activities(&http, &rl, &server.uri(), recovery_after(), None)
7137 .await
7138 .unwrap();
7139
7140 assert_eq!(result.activities.len(), ALPACA_MAX_ACTIVITIES);
7141 assert_eq!(result.resume, None);
7142 assert_eq!(
7143 server.received_requests().await.unwrap().len(),
7144 2,
7145 "exactly 2 requests: first full page + second empty page"
7146 );
7147 }
7148
7149 #[tokio::test]
7152 async fn paginate_activities_two_pages_returns_combined_activities_not_truncated() {
7153 let server = MockServer::start().await;
7154 Mock::given(method("GET"))
7155 .and(path("/v2/account/activities"))
7156 .respond_with(Sequential::new(vec![
7157 make_activities_json(ALPACA_MAX_ACTIVITIES, "p1"),
7158 make_activities_json(37, "p2"),
7159 ]))
7160 .mount(&server)
7161 .await;
7162
7163 let http = reqwest::Client::new();
7164 let rl = RateLimitTracker::new();
7165 let result = paginate_activities(&http, &rl, &server.uri(), recovery_after(), None)
7166 .await
7167 .unwrap();
7168
7169 assert_eq!(result.activities.len(), ALPACA_MAX_ACTIVITIES + 37);
7170 assert_eq!(result.resume, None);
7171 assert_eq!(server.received_requests().await.unwrap().len(), 2);
7172 }
7173
7174 #[tokio::test]
7178 async fn paginate_activities_at_max_pages_sets_truncated_true() {
7179 let server = MockServer::start().await;
7180
7181 Mock::given(method("GET"))
7183 .and(path("/v2/account/activities"))
7184 .respond_with(
7185 ResponseTemplate::new(200)
7186 .set_body_json(make_activities_json(ALPACA_MAX_ACTIVITIES, "act")),
7187 )
7188 .mount(&server)
7189 .await;
7190
7191 let http = reqwest::Client::new();
7192 let rl = RateLimitTracker::new();
7193 let result = paginate_activities(&http, &rl, &server.uri(), recovery_after(), None)
7194 .await
7195 .unwrap();
7196
7197 assert!(
7198 result.resume.is_some(),
7199 "must say where to read on after MAX_ACTIVITY_PAGES pages"
7200 );
7201 assert_eq!(
7202 result.activities.len(),
7203 MAX_ACTIVITY_PAGES * ALPACA_MAX_ACTIVITIES,
7204 "must accumulate exactly MAX_ACTIVITY_PAGES * page_size activities"
7205 );
7206 assert_eq!(
7209 server.received_requests().await.unwrap().len(),
7210 MAX_ACTIVITY_PAGES,
7211 "loop must issue exactly MAX_ACTIVITY_PAGES requests then stop"
7212 );
7213 }
7214
7215 #[tokio::test]
7219 async fn fetch_raw_open_orders_499_results_returns_ok() {
7220 let server = MockServer::start().await;
7221 Mock::given(method("GET"))
7222 .and(path("/v2/orders"))
7223 .respond_with(
7224 ResponseTemplate::new(200).set_body_json(make_orders_json(MAX_OPEN_ORDERS - 1)),
7225 )
7226 .mount(&server)
7227 .await;
7228
7229 let http = reqwest::Client::new();
7230 let rl = RateLimitTracker::new();
7231 let result = fetch_raw_open_orders(&http, &rl, &server.uri(), &[]).await;
7232
7233 assert!(
7234 result.is_ok(),
7235 "499 orders must not trigger truncation: {result:?}"
7236 );
7237 assert_eq!(result.unwrap().len(), MAX_OPEN_ORDERS - 1);
7238 }
7239
7240 #[tokio::test]
7244 async fn fetch_raw_open_orders_500_results_returns_truncated_snapshot_error() {
7245 let server = MockServer::start().await;
7246 Mock::given(method("GET"))
7247 .and(path("/v2/orders"))
7248 .respond_with(
7249 ResponseTemplate::new(200).set_body_json(make_orders_json(MAX_OPEN_ORDERS)),
7250 )
7251 .mount(&server)
7252 .await;
7253
7254 let http = reqwest::Client::new();
7255 let rl = RateLimitTracker::new();
7256 let result = fetch_raw_open_orders(&http, &rl, &server.uri(), &[]).await;
7257
7258 assert!(
7259 matches!(
7260 result,
7261 Err(UnindexedClientError::TruncatedSnapshot { limit }) if limit == MAX_OPEN_ORDERS
7262 ),
7263 "500 orders must return TruncatedSnapshot, got: {result:?}"
7264 );
7265 }
7266
7267 async fn mount_account_and_positions(server: &MockServer) {
7272 Mock::given(method("GET"))
7273 .and(path("/v2/account"))
7274 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
7275 "status": "ACTIVE",
7276 "currency": "USD",
7277 "cash": "-1500.50",
7278 "equity": "4000.00",
7279 "buying_power": "5000.00",
7280 "options_buying_power": "2000.00",
7281 "non_marginable_buying_power": "1000.00",
7282 })))
7283 .mount(server)
7284 .await;
7285 Mock::given(method("GET"))
7286 .and(path("/v2/positions"))
7287 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([
7288 {
7289 "symbol": "XYZ",
7290 "asset_class": "us_equity",
7291 "side": "short",
7292 "qty": "-10",
7293 "qty_available": "-10",
7294 "avg_entry_price": "20.00",
7295 "unrealized_pl": "15.50",
7296 },
7297 {
7298 "symbol": "XYZ271217C00025000",
7299 "asset_class": "us_option",
7300 "side": "long",
7301 "qty": "3",
7302 "qty_available": "3",
7303 "avg_entry_price": "0.85",
7304 "unrealized_pl": "-45",
7305 },
7306 {
7307 "symbol": "ABC",
7308 "asset_class": "us_equity",
7309 "side": "long",
7310 "qty": "2.5",
7311 "qty_available": "2.5",
7312 "avg_entry_price": "40.00",
7313 "unrealized_pl": "1.25",
7314 },
7315 {
7316 "symbol": "BTCUSD",
7317 "asset_class": "crypto",
7318 "side": "long",
7319 "qty": "0.75",
7320 "qty_available": "0.5",
7321 "avg_entry_price": "50000",
7322 "unrealized_pl": "0",
7323 },
7324 ])))
7325 .mount(server)
7326 .await;
7327 Mock::given(method("GET"))
7328 .and(path("/v2/orders"))
7329 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([])))
7330 .mount(server)
7331 .await;
7332 }
7333
7334 fn client_for(server: &MockServer) -> AlpacaClient {
7335 AlpacaClient::new(AlpacaConfig::with_base_url(
7336 "test-key".into(),
7337 "test-secret".into(),
7338 server.uri(),
7339 ))
7340 }
7341
7342 #[tokio::test]
7343 async fn account_snapshot_reports_cash_positions_and_crypto_balances() {
7344 use crate::client::ExecutionClient;
7345 use rust_decimal_macros::dec;
7346
7347 let server = MockServer::start().await;
7348 mount_account_and_positions(&server).await;
7349
7350 let snapshot = client_for(&server)
7351 .account_snapshot(&[], &[])
7352 .await
7353 .unwrap();
7354
7355 let balances: Vec<_> = snapshot
7357 .balances
7358 .iter()
7359 .map(|b| (b.asset.name().as_str(), b.balance.total, b.balance.free))
7360 .collect();
7361 assert_eq!(
7362 balances,
7363 vec![
7364 ("USD", dec!(-1500.50), dec!(-1500.50)),
7365 ("BTC", dec!(0.75), dec!(0.5)),
7366 ],
7367 "crypto stays a balance; equities and options do not appear as balances"
7368 );
7369
7370 let positions: Vec<_> = snapshot
7371 .instruments
7372 .iter()
7373 .map(|i| {
7374 let p = i.position.open().unwrap();
7375 (
7376 i.instrument.name().as_str(),
7377 p.quantity,
7378 p.entry_price,
7379 p.unrealized_pnl,
7380 )
7381 })
7382 .collect();
7383 assert_eq!(
7384 positions,
7385 vec![
7386 ("XYZ", dec!(-10), Some(dec!(20.00)), Some(dec!(15.50))),
7387 (
7388 "XYZ271217C00025000",
7389 dec!(3),
7390 Some(dec!(0.85)),
7391 Some(dec!(-45))
7392 ),
7393 ("ABC", dec!(2.5), Some(dec!(40.00)), Some(dec!(1.25))),
7394 ],
7395 "short equity, option and fractional positions; no crypto position"
7396 );
7397 }
7398
7399 #[tokio::test]
7402 async fn account_snapshot_usd_only_still_reports_positions() {
7403 use crate::client::ExecutionClient;
7404 use rust_decimal_macros::dec;
7405
7406 let server = MockServer::start().await;
7407 mount_account_and_positions(&server).await;
7408
7409 let usd = [AssetNameExchange::new("USD")];
7410 let instruments = [InstrumentNameExchange::new("XYZ")];
7411 let snapshot = client_for(&server)
7412 .account_snapshot(&usd, &instruments)
7413 .await
7414 .unwrap();
7415
7416 assert_eq!(snapshot.balances.len(), 1);
7417 assert_eq!(snapshot.balances[0].asset, AssetNameExchange::new("USD"));
7418 assert_eq!(snapshot.balances[0].balance.total, dec!(-1500.50));
7419
7420 assert_eq!(snapshot.instruments.len(), 1);
7421 assert_eq!(snapshot.instruments[0].position.quantity(), Some(dec!(-10)));
7422 }
7423
7424 #[tokio::test]
7426 async fn account_snapshot_non_usd_only_skips_account_request() {
7427 use crate::client::ExecutionClient;
7428
7429 let server = MockServer::start().await;
7430 mount_account_and_positions(&server).await;
7431
7432 let btc = [AssetNameExchange::new("btc")];
7433 let snapshot = client_for(&server)
7434 .account_snapshot(&btc, &[])
7435 .await
7436 .unwrap();
7437
7438 assert_eq!(snapshot.balances.len(), 1);
7439 assert_eq!(snapshot.balances[0].asset, AssetNameExchange::new("BTC"));
7440 let requests = server.received_requests().await.unwrap();
7441 assert!(
7442 requests.iter().all(|r| r.url.path() != "/v2/account"),
7443 "no USD requested, so /v2/account must not be fetched"
7444 );
7445 }
7446
7447 #[tokio::test]
7455 async fn open_order_reduce_only_sell_sends_sell_to_close_intent() {
7456 use crate::client::ExecutionClient;
7457 use crate::order::request::{OrderRequestOpen, RequestOpen};
7458 use crate::order::{
7459 OrderKey, OrderKind, TimeInForce,
7460 id::{ClientOrderId, StrategyId},
7461 };
7462 use rust_decimal::Decimal;
7463 use rustrade_instrument::Side;
7464 use rustrade_instrument::exchange::ExchangeId;
7465 use rustrade_instrument::instrument::name::InstrumentNameExchange;
7466 use wiremock::matchers::{method, path};
7467
7468 let server = MockServer::start().await;
7469
7470 let captured_body = std::sync::Arc::new(parking_lot::Mutex::new(None));
7473 let captured_clone = captured_body.clone();
7474
7475 Mock::given(method("POST"))
7476 .and(path("/v2/orders"))
7477 .respond_with(move |req: &Request| {
7478 let body: serde_json::Value = serde_json::from_slice(&req.body).unwrap();
7480 *captured_clone.lock() = Some(body);
7481
7482 ResponseTemplate::new(200).set_body_json(serde_json::json!({
7483 "id": "test-order-id",
7484 "client_order_id": "test-cid",
7485 "symbol": "AAPL",
7486 "qty": "10",
7487 "filled_qty": "0",
7488 "side": "sell",
7489 "type": "market",
7490 "time_in_force": "ioc",
7491 "limit_price": null,
7492 "created_at": "2025-04-18T14:30:00Z"
7493 }))
7494 })
7495 .mount(&server)
7496 .await;
7497
7498 let config =
7500 AlpacaConfig::with_base_url("test-key".into(), "test-secret".into(), server.uri());
7501 let client = AlpacaClient::new(config);
7502
7503 let request = OrderRequestOpen {
7505 key: OrderKey {
7506 exchange: ExchangeId::AlpacaBroker,
7507 instrument: InstrumentNameExchange::new("AAPL"),
7508 strategy: StrategyId::new("test-strategy"),
7509 cid: ClientOrderId::new("test-cid"),
7510 },
7511 state: RequestOpen {
7512 side: Side::Sell,
7513 price: None,
7514 quantity: Decimal::new(10, 0),
7515 kind: OrderKind::Market,
7516 time_in_force: TimeInForce::ImmediateOrCancel,
7517 position_id: None,
7518 reduce_only: true, market: None,
7520 },
7521 };
7522
7523 let order = client
7525 .open_order(OrderRequestOpen {
7526 key: OrderKey {
7527 exchange: request.key.exchange,
7528 instrument: &request.key.instrument,
7529 strategy: request.key.strategy.clone(),
7530 cid: request.key.cid.clone(),
7531 },
7532 state: request.state.clone(),
7533 })
7534 .await;
7535
7536 assert!(
7538 order.state.is_accepted(),
7539 "order should be accepted: {:?}",
7540 order.state
7541 );
7542
7543 let body = captured_body
7545 .lock()
7546 .take()
7547 .expect("request body should be captured");
7548 assert_eq!(
7549 body.get("position_intent").and_then(|v| v.as_str()),
7550 Some("sell_to_close"),
7551 "reduce_only=true + Side::Sell should produce position_intent=sell_to_close, got: {body}"
7552 );
7553 }
7554
7555 #[tokio::test]
7557 async fn open_order_not_reduce_only_buy_sends_buy_to_open_intent() {
7558 use crate::client::ExecutionClient;
7559 use crate::order::request::{OrderRequestOpen, RequestOpen};
7560 use crate::order::{
7561 OrderKey, OrderKind, TimeInForce,
7562 id::{ClientOrderId, StrategyId},
7563 };
7564 use rust_decimal::Decimal;
7565 use rustrade_instrument::Side;
7566 use rustrade_instrument::exchange::ExchangeId;
7567 use rustrade_instrument::instrument::name::InstrumentNameExchange;
7568
7569 let server = MockServer::start().await;
7570
7571 let captured_body = std::sync::Arc::new(parking_lot::Mutex::new(None));
7572 let captured_clone = captured_body.clone();
7573
7574 Mock::given(method("POST"))
7575 .and(path("/v2/orders"))
7576 .respond_with(move |req: &Request| {
7577 let body: serde_json::Value = serde_json::from_slice(&req.body).unwrap();
7578 *captured_clone.lock() = Some(body);
7579
7580 ResponseTemplate::new(200).set_body_json(serde_json::json!({
7581 "id": "test-order-id",
7582 "client_order_id": "test-cid",
7583 "symbol": "AAPL",
7584 "qty": "10",
7585 "filled_qty": "0",
7586 "side": "buy",
7587 "type": "market",
7588 "time_in_force": "ioc",
7589 "limit_price": null,
7590 "created_at": "2025-04-18T14:30:00Z"
7591 }))
7592 })
7593 .mount(&server)
7594 .await;
7595
7596 let config =
7597 AlpacaConfig::with_base_url("test-key".into(), "test-secret".into(), server.uri());
7598 let client = AlpacaClient::new(config);
7599
7600 let instrument = InstrumentNameExchange::new("AAPL");
7601 let request = OrderRequestOpen {
7602 key: OrderKey {
7603 exchange: ExchangeId::AlpacaBroker,
7604 instrument: &instrument,
7605 strategy: StrategyId::new("test-strategy"),
7606 cid: ClientOrderId::new("test-cid"),
7607 },
7608 state: RequestOpen {
7609 side: Side::Buy,
7610 price: None,
7611 quantity: Decimal::new(10, 0),
7612 kind: OrderKind::Market,
7613 time_in_force: TimeInForce::ImmediateOrCancel,
7614 position_id: None,
7615 reduce_only: false, market: None,
7617 },
7618 };
7619
7620 let order = client.open_order(request).await;
7621
7622 assert!(
7623 order.state.is_accepted(),
7624 "order should be accepted: {:?}",
7625 order.state
7626 );
7627
7628 let body = captured_body
7629 .lock()
7630 .take()
7631 .expect("request body should be captured");
7632 assert_eq!(
7633 body.get("position_intent").and_then(|v| v.as_str()),
7634 Some("buy_to_open"),
7635 "reduce_only=false + Side::Buy should produce position_intent=buy_to_open, got: {body}"
7636 );
7637 }
7638
7639 #[tokio::test]
7642 async fn rest_with_retry_reports_an_undecodable_2xx_body_as_internal() {
7643 let server = MockServer::start().await;
7644 Mock::given(method("GET"))
7645 .and(path("/v2/account"))
7646 .respond_with(ResponseTemplate::new(200).set_body_string("not json"))
7647 .mount(&server)
7648 .await;
7649
7650 let http = reqwest::Client::new();
7651 let rl = RateLimitTracker::new();
7652 let url = format!("{}/v2/account", server.uri());
7653 let result: Result<serde_json::Value, _> =
7654 rest_with_retry(&rl, || http.get(&url)).await;
7655 assert!(
7656 matches!(result, Err(UnindexedClientError::Internal(_))),
7657 "{result:?}"
7658 );
7659 }
7660
7661 async fn holds_back(rl: &RateLimitTracker) -> bool {
7663 tokio::time::timeout(Duration::from_millis(50), rl.wait_if_blocked())
7664 .await
7665 .is_err()
7666 }
7667
7668 fn with_rate_limit(
7670 template: ResponseTemplate,
7671 remaining: u32,
7672 reset: bool,
7673 ) -> ResponseTemplate {
7674 let template = template.insert_header("x-ratelimit-remaining", remaining.to_string());
7675 if !reset {
7676 return template;
7677 }
7678 let now_secs = std::time::SystemTime::now()
7679 .duration_since(std::time::UNIX_EPOCH)
7680 .unwrap()
7681 .as_secs();
7682 template.insert_header("x-ratelimit-reset", (now_secs + 30).to_string())
7683 }
7684
7685 #[tokio::test]
7688 async fn rest_with_retry_pauses_once_no_requests_remain() {
7689 for (remaining, reset, pauses) in [(0, true, true), (1, true, false), (0, false, false)]
7690 {
7691 let server = MockServer::start().await;
7692 Mock::given(method("GET"))
7693 .and(path("/v2/account"))
7694 .respond_with(with_rate_limit(
7695 ResponseTemplate::new(200).set_body_json(serde_json::json!({})),
7696 remaining,
7697 reset,
7698 ))
7699 .mount(&server)
7700 .await;
7701
7702 let http = reqwest::Client::new();
7703 let rl = RateLimitTracker::new();
7704 let url = format!("{}/v2/account", server.uri());
7705 let _: serde_json::Value = rest_with_retry(&rl, || http.get(&url)).await.unwrap();
7706 assert_eq!(
7707 holds_back(&rl).await,
7708 pauses,
7709 "remaining {remaining}, reset {reset}"
7710 );
7711 }
7712 }
7713
7714 #[tokio::test]
7717 async fn a_pause_lasts_a_minute_at_most() {
7718 tokio::time::pause();
7719 let now_secs = std::time::SystemTime::now()
7720 .duration_since(std::time::UNIX_EPOCH)
7721 .unwrap()
7722 .as_secs();
7723 let mut headers = reqwest::header::HeaderMap::new();
7724 headers.insert("x-ratelimit-remaining", "0".parse().unwrap());
7725 headers.insert(
7726 "x-ratelimit-reset",
7727 (now_secs + 3_600).to_string().parse().unwrap(),
7728 );
7729 let rl = RateLimitTracker::new();
7730
7731 observe_rate_limit_remaining(&rl, &headers);
7732 assert!(holds_back(&rl).await);
7733 tokio::time::advance(Duration::from_secs(DEFAULT_RATE_LIMIT_DELAY_SECS)).await;
7734 assert!(!holds_back(&rl).await);
7735 }
7736
7737 #[tokio::test]
7739 async fn rest_delete_with_retry_pauses_once_no_requests_remain() {
7740 let server = MockServer::start().await;
7741 Mock::given(method("DELETE"))
7742 .and(path("/v2/orders/abc"))
7743 .respond_with(with_rate_limit(ResponseTemplate::new(204), 0, true))
7744 .mount(&server)
7745 .await;
7746
7747 let http = reqwest::Client::new();
7748 let rl = RateLimitTracker::new();
7749 let url = format!("{}/v2/orders/abc", server.uri());
7750 rest_delete_with_retry(&rl, || http.delete(&url))
7751 .await
7752 .unwrap();
7753 assert!(holds_back(&rl).await);
7754 }
7755
7756 #[tokio::test]
7759 async fn open_order_with_an_undecodable_2xx_response_is_status_unknown() {
7760 use crate::client::ExecutionClient;
7761 use crate::error::OrderError;
7762 use crate::order::request::{OrderRequestOpen, RequestOpen};
7763 use crate::order::state::{InactiveOrderState, OrderState};
7764 use crate::order::{
7765 OrderKey, OrderKind, TimeInForce,
7766 id::{ClientOrderId, StrategyId},
7767 };
7768 use rustrade_instrument::Side;
7769 use rustrade_instrument::exchange::ExchangeId;
7770 use rustrade_instrument::instrument::name::InstrumentNameExchange;
7771
7772 let server = MockServer::start().await;
7773 Mock::given(method("POST"))
7774 .and(path("/v2/orders"))
7775 .respond_with(
7776 ResponseTemplate::new(200).set_body_json(serde_json::json!({ "id": 1 })),
7777 )
7778 .mount(&server)
7779 .await;
7780
7781 let instrument = InstrumentNameExchange::new("AAPL");
7782 let order = client_for(&server)
7783 .open_order(OrderRequestOpen {
7784 key: OrderKey {
7785 exchange: ExchangeId::AlpacaBroker,
7786 instrument: &instrument,
7787 strategy: StrategyId::new("test-strategy"),
7788 cid: ClientOrderId::new("test-cid"),
7789 },
7790 state: RequestOpen {
7791 side: Side::Buy,
7792 price: None,
7793 quantity: Decimal::new(10, 0),
7794 kind: OrderKind::Market,
7795 time_in_force: TimeInForce::ImmediateOrCancel,
7796 position_id: None,
7797 reduce_only: false,
7798 market: None,
7799 },
7800 })
7801 .await;
7802
7803 assert!(
7804 matches!(
7805 order.state,
7806 OrderState::Inactive(InactiveOrderState::OpenFailed(OrderError::Connectivity(
7807 _
7808 )))
7809 ),
7810 "{:?}",
7811 order.state
7812 );
7813 }
7814
7815 #[tokio::test]
7818 async fn open_bracket_order_with_an_undecodable_2xx_response_is_status_unknown() {
7819 use crate::error::OrderError;
7820 use crate::order::state::{InactiveOrderState, OrderState};
7821 use rustrade_instrument::instrument::name::InstrumentNameExchange;
7822
7823 let server = MockServer::start().await;
7824 Mock::given(method("POST"))
7825 .and(path("/v2/orders"))
7826 .respond_with(
7827 ResponseTemplate::new(200).set_body_json(serde_json::json!({ "id": 1 })),
7828 )
7829 .mount(&server)
7830 .await;
7831
7832 let result = client_for(&server)
7833 .open_bracket_order(AlpacaBracketOrderRequest::new(
7834 InstrumentNameExchange::new("SPY"),
7835 crate::order::id::StrategyId::new("test"),
7836 crate::order::id::ClientOrderId::new("test-bracket"),
7837 Side::Buy,
7838 Decimal::ONE,
7839 Decimal::new(100, 0),
7840 Decimal::new(120, 0),
7841 Decimal::new(90, 0),
7842 TimeInForce::GoodUntilCancelled { post_only: false },
7843 ))
7844 .await;
7845
7846 assert!(
7847 matches!(
7848 result.parent.state,
7849 OrderState::Inactive(InactiveOrderState::OpenFailed(OrderError::Connectivity(
7850 _
7851 )))
7852 ),
7853 "{:?}",
7854 result.parent.state
7855 );
7856 }
7857
7858 fn recovery_after() -> DateTime<Utc> {
7867 "2025-01-01T00:00:00Z".parse().expect("valid time")
7868 }
7869
7870 fn activity_json(
7873 id: &str,
7874 order_id: &str,
7875 qty: &str,
7876 cum_qty: Option<&str>,
7877 ) -> serde_json::Value {
7878 let mut v = serde_json::json!({
7879 "id": id,
7880 "order_id": order_id,
7881 "symbol": "SPY",
7882 "side": "buy",
7883 "price": "100.00",
7884 "qty": qty,
7885 "transaction_time": "2025-04-18T14:30:00Z"
7886 });
7887 if let Some(c) = cum_qty {
7888 v["cum_qty"] = serde_json::Value::String(c.to_string());
7889 }
7890 v
7891 }
7892
7893 async fn drive_recover_fills_with(
7896 activities: Vec<serde_json::Value>,
7897 dedup: SharedDedupCache,
7898 known: &SharedKnownLiveOrders,
7899 ) -> Vec<UnindexedAccountEvent> {
7900 let server = MockServer::start().await;
7901 Mock::given(method("GET"))
7902 .and(path("/v2/account/activities"))
7903 .respond_with(
7904 ResponseTemplate::new(200).set_body_json(serde_json::Value::Array(activities)),
7905 )
7906 .mount(&server)
7907 .await;
7908
7909 let http = reqwest::Client::new();
7910 let rl = RateLimitTracker::new();
7911 let (tx, mut rx) = mpsc::unbounded_channel();
7912 let outcome = recover_fills(
7913 &http,
7914 &rl,
7915 &[],
7916 &server.uri(),
7917 recovery_after(),
7918 &tx,
7919 &dedup,
7920 known,
7921 )
7922 .await;
7923 assert_eq!(outcome, Ok(()), "a full read leaves nothing unrecovered");
7924 drop(tx);
7925
7926 let mut out = Vec::new();
7927 while let Ok(event) = rx.try_recv() {
7928 out.push(event);
7929 }
7930 out
7931 }
7932
7933 async fn drive_recover_fills(
7934 activities: Vec<serde_json::Value>,
7935 ) -> Vec<UnindexedAccountEvent> {
7936 drive_recover_fills_with(
7937 activities,
7938 new_dedup_cache(),
7939 &KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
7940 )
7941 .await
7942 }
7943
7944 async fn recover_fills_from(
7947 server: &MockServer,
7948 ) -> (Result<(), UnreadFills>, Vec<UnindexedAccountEvent>) {
7949 let (tx, mut rx) = mpsc::unbounded_channel();
7950 let outcome = recover_fills(
7951 &reqwest::Client::new(),
7952 &RateLimitTracker::new(),
7953 &[],
7954 &server.uri(),
7955 recovery_after(),
7956 &tx,
7957 &new_dedup_cache(),
7958 &KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
7959 )
7960 .await;
7961 drop(tx);
7962 (outcome, std::iter::from_fn(|| rx.try_recv().ok()).collect())
7963 }
7964
7965 #[tokio::test]
7968 async fn a_failed_recovery_read_leaves_every_fill_since_the_disconnect_unread() {
7969 let server = MockServer::start().await;
7970 Mock::given(method("GET"))
7971 .and(path("/v2/account/activities"))
7972 .respond_with(ResponseTemplate::new(500).set_body_string("unavailable"))
7973 .mount(&server)
7974 .await;
7975
7976 let (outcome, events) = recover_fills_from(&server).await;
7977
7978 let Err(UnreadFills {
7979 start,
7980 reason: FillRecoveryFailure::Request(_),
7981 }) = outcome
7982 else {
7983 panic!("expected the request's failure, got {outcome:?}");
7984 };
7985 assert_eq!(start, recovery_after(), "from the disconnect");
7986 assert!(events.is_empty(), "nothing forwarded: {events:?}");
7987 }
7988
7989 #[tokio::test]
7994 async fn a_truncated_recovery_read_delivers_what_it_read_and_leaves_the_rest_unread() {
7995 let time = |s: &str| s.parse::<DateTime<Utc>>().expect("valid time");
7996 let cases = [
7998 (
7999 Some("2025-04-18T14:30:01.234567Z"),
8000 time("2025-04-18T14:30:01.234Z"),
8001 ),
8002 (Some("not a time"), time("2025-04-18T14:30:00Z")),
8003 (None, recovery_after()),
8004 ];
8005 for (last_time, expected_start) in cases {
8006 let mut page = make_activities_json(ALPACA_MAX_ACTIVITIES, "act");
8007 let Some(activities) = page.as_array_mut() else {
8008 panic!("an array of activities");
8009 };
8010 match last_time {
8011 Some(last_time) => {
8012 if let Some(last) = activities.last_mut() {
8013 last["transaction_time"] =
8014 serde_json::Value::String(last_time.to_string());
8015 }
8016 }
8017 None => {
8018 for activity in activities.iter_mut() {
8019 activity["transaction_time"] =
8020 serde_json::Value::String("not a time".to_string());
8021 }
8022 }
8023 }
8024 let server = MockServer::start().await;
8025 Mock::given(method("GET"))
8026 .and(path("/v2/account/activities"))
8027 .respond_with(ResponseTemplate::new(200).set_body_json(page))
8028 .mount(&server)
8029 .await;
8030
8031 let (outcome, events) = recover_fills_from(&server).await;
8032
8033 let fills_read = MAX_ACTIVITY_PAGES * ALPACA_MAX_ACTIVITIES;
8034 assert_eq!(
8035 outcome,
8036 Err(UnreadFills {
8037 start: expected_start,
8038 reason: FillRecoveryFailure::Truncated { fills_read },
8039 }),
8040 "last time {last_time:?}"
8041 );
8042 assert_eq!(events.len(), fills_read, "every fill read is forwarded");
8043 }
8044 }
8045
8046 #[test]
8049 fn a_fill_recovery_give_up_covers_every_instrument_the_stream_does() {
8050 let start = recovery_after();
8051 let end = start + chrono::Duration::minutes(5);
8052 let spy = InstrumentNameExchange::new("SPY");
8053 let qqq = InstrumentNameExchange::new("QQQ");
8054 let cases = [
8055 (vec![], FillRecoveryScope::AllInstruments),
8056 (
8057 vec![spy.clone(), qqq.clone()],
8058 FillRecoveryScope::Instruments(vec![spy, qqq]),
8059 ),
8060 ];
8061 for (instruments, scope) in cases {
8062 let reason = FillRecoveryFailure::TimedOut { timeout_secs: 30 };
8063 let unread = UnreadFills {
8064 start,
8065 reason: reason.clone(),
8066 };
8067 assert_eq!(
8068 fill_recovery_gave_up(&instruments, unread, end),
8069 UnindexedAccountEvent::new(
8070 ExchangeId::AlpacaBroker,
8071 AccountEventKind::FillRecoveryGaveUp(FillRecoveryGap::new(
8072 scope, start, end, 1, reason,
8073 )),
8074 )
8075 );
8076 }
8077 }
8078
8079 #[test]
8083 fn a_read_that_reached_past_the_recovery_start_still_reports() {
8084 let end = recovery_after();
8085 let start = end + chrono::Duration::seconds(1);
8086 let unread = UnreadFills {
8087 start,
8088 reason: FillRecoveryFailure::Truncated { fills_read: 5_000 },
8089 };
8090 let event = fill_recovery_gave_up(&[], unread, end);
8091 let AccountEventKind::FillRecoveryGaveUp(gave_up) = event.kind else {
8092 panic!("expected FillRecoveryGaveUp, got {event:?}");
8093 };
8094 assert_eq!((gave_up.start, gave_up.end), (start, start));
8095 }
8096
8097 async fn recover_fills_or_report_from(
8100 server: &MockServer,
8101 instruments: &[InstrumentNameExchange],
8102 timeout: Duration,
8103 ) -> Vec<UnindexedAccountEvent> {
8104 let (tx, mut rx) = mpsc::unbounded_channel();
8105 recover_fills_or_report(
8106 &reqwest::Client::new(),
8107 &RateLimitTracker::new(),
8108 instruments,
8109 &server.uri(),
8110 recovery_after(),
8111 timeout,
8112 &tx,
8113 &new_dedup_cache(),
8114 &KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
8115 )
8116 .await;
8117 drop(tx);
8118 std::iter::from_fn(|| rx.try_recv().ok()).collect()
8119 }
8120
8121 #[tokio::test]
8124 async fn a_timed_out_recovery_reports_every_fill_since_the_disconnect() {
8125 let server = MockServer::start().await;
8126 Mock::given(method("GET"))
8127 .and(path("/v2/account/activities"))
8128 .respond_with(
8129 ResponseTemplate::new(200)
8130 .set_body_json(serde_json::json!([]))
8131 .set_delay(Duration::from_secs(3_600)),
8132 )
8133 .mount(&server)
8134 .await;
8135 let spy = InstrumentNameExchange::new("SPY");
8136
8137 let events = recover_fills_or_report_from(
8138 &server,
8139 std::slice::from_ref(&spy),
8140 Duration::from_millis(50),
8141 )
8142 .await;
8143
8144 let [event] = events.as_slice() else {
8145 panic!("only the give-up is sent: {events:?}");
8146 };
8147 let AccountEventKind::FillRecoveryGaveUp(gave_up) = &event.kind else {
8148 panic!("expected FillRecoveryGaveUp, got {event:?}");
8149 };
8150 assert_eq!(gave_up.scope, FillRecoveryScope::Instruments(vec![spy]));
8151 assert_eq!(gave_up.start, recovery_after(), "from the disconnect");
8152 assert!(gave_up.end >= gave_up.start);
8153 assert_eq!(gave_up.attempts, 1);
8154 assert!(
8155 matches!(gave_up.reason, FillRecoveryFailure::TimedOut { .. }),
8156 "{:?}",
8157 gave_up.reason
8158 );
8159 }
8160
8161 #[tokio::test]
8164 async fn a_truncated_recovery_reports_after_the_fills_it_read() {
8165 let server = MockServer::start().await;
8166 Mock::given(method("GET"))
8167 .and(path("/v2/account/activities"))
8168 .respond_with(
8169 ResponseTemplate::new(200)
8170 .set_body_json(make_activities_json(ALPACA_MAX_ACTIVITIES, "act")),
8171 )
8172 .mount(&server)
8173 .await;
8174
8175 let events = recover_fills_or_report_from(&server, &[], Duration::from_secs(30)).await;
8176
8177 let Some((last, fills)) = events.split_last() else {
8178 panic!("events are sent");
8179 };
8180 assert_eq!(fills.len(), MAX_ACTIVITY_PAGES * ALPACA_MAX_ACTIVITIES);
8181 assert!(
8182 fills
8183 .iter()
8184 .all(|event| matches!(event.kind, AccountEventKind::Trade(_))),
8185 "every fill comes first"
8186 );
8187 let AccountEventKind::FillRecoveryGaveUp(gave_up) = &last.kind else {
8188 panic!("the give-up comes last, got {last:?}");
8189 };
8190 assert_eq!(gave_up.scope, FillRecoveryScope::AllInstruments);
8191 assert!(matches!(
8192 gave_up.reason,
8193 FillRecoveryFailure::Truncated { .. }
8194 ));
8195 }
8196
8197 #[test]
8200 fn an_activity_id_names_its_execution_after_the_separator() {
8201 assert_eq!(
8202 activity_execution_id("20261005041849348::524b1902-817e-446c-825b-a9fcfebbc17e"),
8203 "524b1902-817e-446c-825b-a9fcfebbc17e"
8204 );
8205 assert_eq!(activity_execution_id("act-1"), "act-1");
8206 assert_eq!(
8207 activity_execution_id("20261005041849348::"),
8208 "20261005041849348::"
8209 );
8210 }
8211
8212 #[tokio::test]
8216 async fn a_fill_has_one_trade_id_on_the_stream_in_recovery_and_from_fetch_trades() {
8217 use crate::client::ExecutionClient;
8218
8219 let execution_id = "524b1902-817e-446c-825b-a9fcfebbc17e";
8220 let activity = activity_json(
8221 &format!("20250418103000000::{execution_id}"),
8222 "ord-1",
8223 "2",
8224 Some("2"),
8225 );
8226
8227 let [streamed, _snapshot] = convert_trade_update(AlpacaTradeUpdate {
8228 event: SmolStr::new("fill"),
8229 execution_id: Some(execution_id),
8230 previous_execution_id: None,
8231 order: super::make_order_ws("ord-1", "SPY", "buy", "2"),
8232 price: Some("100.00"),
8233 qty: Some("2"),
8234 timestamp: Some("2025-04-18T14:30:00Z"),
8235 });
8236 let Some(streamed) = streamed else {
8237 panic!("a fill converts to a Trade");
8238 };
8239
8240 let recovered = drive_recover_fills(vec![activity.clone()]).await;
8241 let [recovered] = recovered.as_slice() else {
8242 panic!("one fill recovered: {recovered:?}");
8243 };
8244
8245 let server = MockServer::start().await;
8246 Mock::given(method("GET"))
8247 .and(path("/v2/account/activities"))
8248 .respond_with(
8249 ResponseTemplate::new(200)
8250 .set_body_json(serde_json::Value::Array(vec![activity])),
8251 )
8252 .mount(&server)
8253 .await;
8254 let Ok(fetched) = client_for(&server)
8255 .fetch_trades(recovery_after(), Utc::now(), &[])
8256 .await
8257 else {
8258 panic!("fetch_trades reads the activity");
8259 };
8260 let [fetched] = fetched.trades.as_slice() else {
8261 panic!("one fill fetched: {fetched:?}");
8262 };
8263
8264 assert_eq!(trade_of(&streamed).id.0.as_str(), execution_id);
8265 assert_eq!(trade_of(recovered).id.0.as_str(), execution_id);
8266 assert_eq!(fetched.id.0.as_str(), execution_id);
8267 }
8268
8269 fn time(s: &str) -> DateTime<Utc> {
8274 s.parse().expect("valid time")
8275 }
8276
8277 struct ActivitiesVenue {
8282 activities: Vec<serde_json::Value>,
8283 }
8284
8285 impl Respond for ActivitiesVenue {
8286 fn respond(&self, request: &Request) -> ResponseTemplate {
8287 let query: std::collections::HashMap<String, String> =
8288 request.url.query_pairs().into_owned().collect();
8289 let bound = |name: &str| query.get(name).map(|s| time(s));
8290 let (after, until) = (bound("after"), bound("until"));
8291 let size: usize = query
8292 .get("page_size")
8293 .and_then(|s| s.parse().ok())
8294 .unwrap_or(ALPACA_MAX_ACTIVITIES);
8295 let in_bounds = self.activities.iter().filter(|activity| {
8296 let at =
8297 floor_millis(time(activity["transaction_time"].as_str().expect("a time")));
8298 after.is_none_or(|after| at >= after) && until.is_none_or(|until| at < until)
8299 });
8300 let page: Vec<_> = match query.get("page_token") {
8301 Some(token) => in_bounds
8302 .skip_while(|activity| activity["id"] != token.as_str())
8303 .skip(1)
8304 .take(size)
8305 .cloned()
8306 .collect(),
8307 None => in_bounds.take(size).cloned().collect(),
8308 };
8309 ResponseTemplate::new(200).set_body_json(page)
8310 }
8311 }
8312
8313 fn fill_at(id: &str, symbol: &str, transaction_time: DateTime<Utc>) -> serde_json::Value {
8315 serde_json::json!({
8316 "id": id,
8317 "order_id": format!("ord-{id}"),
8318 "symbol": symbol,
8319 "side": "buy",
8320 "price": "100.00",
8321 "qty": "1",
8322 "cum_qty": "1",
8323 "transaction_time": transaction_time.to_rfc3339_opts(SecondsFormat::Micros, true),
8324 })
8325 }
8326
8327 async fn serve_activities(activities: Vec<serde_json::Value>) -> MockServer {
8328 let server = MockServer::start().await;
8329 Mock::given(method("GET"))
8330 .and(path("/v2/account/activities"))
8331 .respond_with(ActivitiesVenue { activities })
8332 .mount(&server)
8333 .await;
8334 server
8335 }
8336
8337 async fn read_span(
8340 client: &AlpacaClient,
8341 start: DateTime<Utc>,
8342 end: DateTime<Utc>,
8343 instruments: &[InstrumentNameExchange],
8344 ) -> (std::collections::HashSet<String>, usize) {
8345 use crate::client::ExecutionClient;
8346
8347 let mut ids = std::collections::HashSet::new();
8348 let mut from = start;
8349 for calls in 1..=10 {
8350 let read = match client.fetch_trades(from, end, instruments).await {
8351 Ok(read) => read,
8352 Err(error) => panic!("call {calls} failed: {error:?}"),
8353 };
8354 ids.extend(read.trades.iter().map(|trade| trade.id.0.to_string()));
8355 match read.resume {
8356 Some(resume) => from = resume,
8357 None => return (ids, calls),
8358 }
8359 }
8360 panic!("the reads did not reach the span's end");
8361 }
8362
8363 #[tokio::test]
8369 async fn fetch_trades_reads_a_busy_span_completely_across_calls() {
8370 let base = time("2025-04-18T14:30:00Z");
8371 let mut activities = Vec::new();
8372 for ms in 0..4_000_i64 {
8373 let symbol = if ms % 2 == 0 { "SPY" } else { "QQQ" };
8374 for (k, micros) in [("a", 900), ("b", 500), ("c", 100)] {
8375 let at = base + TimeDelta::milliseconds(ms) + TimeDelta::microseconds(micros);
8376 activities.push(fill_at(&format!("{ms:017}::{ms}-{k}"), symbol, at));
8377 }
8378 }
8379 let server = serve_activities(activities).await;
8380 let client = client_for(&server);
8381 let end = time("2025-04-18T15:00:00Z");
8382
8383 let (ids, calls) = read_span(&client, recovery_after(), end, &[]).await;
8384 assert_eq!(ids.len(), 12_000, "every fill is read once or more");
8385 assert_eq!(calls, 3, "5,000 activities per call");
8386
8387 let spy = [InstrumentNameExchange::new("SPY")];
8388 let (ids, _) = read_span(&client, recovery_after(), end, &spy).await;
8389 assert_eq!(ids.len(), 6_000, "every SPY fill, and only those");
8390
8391 let Some(requests) = server.received_requests().await else {
8392 panic!("requests are recorded");
8393 };
8394 assert!(
8395 requests
8396 .iter()
8397 .all(|request| request.url.query_pairs().any(
8398 |(name, value)| name == "until" && value == "2025-04-18T15:00:00.001Z"
8399 )),
8400 "every request is bounded by the span's end"
8401 );
8402 }
8403
8404 #[tokio::test]
8407 async fn fetch_trades_says_where_to_read_on_after_a_call_without_matches() {
8408 use crate::client::ExecutionClient;
8409
8410 let server = MockServer::start().await;
8411 Mock::given(method("GET"))
8412 .and(path("/v2/account/activities"))
8413 .respond_with(
8414 ResponseTemplate::new(200)
8415 .set_body_json(make_activities_json(ALPACA_MAX_ACTIVITIES, "act")),
8416 )
8417 .mount(&server)
8418 .await;
8419
8420 let qqq = [InstrumentNameExchange::new("QQQ")];
8421 let Ok(read) = client_for(&server)
8422 .fetch_trades(recovery_after(), Utc::now(), &qqq)
8423 .await
8424 else {
8425 panic!("the read succeeds");
8426 };
8427 assert!(read.trades.is_empty(), "every fill read is SPY's");
8428 assert_eq!(read.resume, Some(time("2025-04-18T14:30:00Z")));
8429 }
8430
8431 #[tokio::test]
8434 async fn fetch_trades_that_cannot_advance_is_truncated() {
8435 use crate::client::ExecutionClient;
8436
8437 let server = MockServer::start().await;
8438 Mock::given(method("GET"))
8439 .and(path("/v2/account/activities"))
8440 .respond_with(
8441 ResponseTemplate::new(200)
8442 .set_body_json(make_activities_json(ALPACA_MAX_ACTIVITIES, "act")),
8443 )
8444 .mount(&server)
8445 .await;
8446
8447 let result = client_for(&server)
8448 .fetch_trades(time("2025-04-18T14:30:00.000500Z"), Utc::now(), &[])
8449 .await;
8450 assert!(
8451 matches!(
8452 result,
8453 Err(UnindexedClientError::Truncated { fills_read: 5_000 })
8454 ),
8455 "got {result:?}"
8456 );
8457 }
8458
8459 #[tokio::test]
8462 async fn fetch_trades_returns_only_the_span() {
8463 let (start, end) = (
8464 time("2025-04-18T14:30:00.123456Z"),
8465 time("2025-04-18T14:30:05.678901Z"),
8466 );
8467 let micro = TimeDelta::microseconds(1);
8468 let server = serve_activities(vec![
8469 fill_at("1::before", "SPY", start - micro),
8470 fill_at("2::start", "SPY", start),
8471 fill_at("3::end", "SPY", end),
8472 fill_at("4::after", "SPY", end + micro),
8473 ])
8474 .await;
8475
8476 let (ids, calls) = read_span(&client_for(&server), start, end, &[]).await;
8477 assert_eq!(calls, 1);
8478 let mut ids: Vec<_> = ids.into_iter().collect();
8479 ids.sort();
8480 assert_eq!(ids, ["end", "start"]);
8481 }
8482
8483 #[tokio::test]
8486 async fn fetch_trades_that_read_past_the_end_is_complete() {
8487 use crate::client::ExecutionClient;
8488
8489 let server = MockServer::start().await;
8490 Mock::given(method("GET"))
8492 .and(path("/v2/account/activities"))
8493 .respond_with(
8494 ResponseTemplate::new(200)
8495 .set_body_json(make_activities_json(ALPACA_MAX_ACTIVITIES, "act")),
8496 )
8497 .mount(&server)
8498 .await;
8499
8500 let Ok(read) = client_for(&server)
8501 .fetch_trades(recovery_after(), time("2025-04-18T14:29:00Z"), &[])
8502 .await
8503 else {
8504 panic!("the read succeeds");
8505 };
8506 assert_eq!(read, TradesRead::complete(Vec::new()));
8507 }
8508
8509 #[tokio::test]
8513 async fn fetch_trades_accepts_the_extreme_times() {
8514 use crate::client::ExecutionClient;
8515
8516 let server = serve_activities(vec![fill_at(
8517 "1::fill",
8518 "SPY",
8519 time("2025-04-18T14:30:00Z"),
8520 )])
8521 .await;
8522 let Ok(read) = client_for(&server)
8523 .fetch_trades(DateTime::<Utc>::MIN_UTC, DateTime::<Utc>::MAX_UTC, &[])
8524 .await
8525 else {
8526 panic!("the read succeeds");
8527 };
8528 assert_eq!(read.trades.len(), 1);
8529 assert_eq!(read.resume, None);
8530
8531 let Some(requests) = server.received_requests().await else {
8532 panic!("requests are recorded");
8533 };
8534 let [request] = requests.as_slice() else {
8535 panic!("one request: {requests:?}");
8536 };
8537 let query: std::collections::HashMap<_, _> =
8538 request.url.query_pairs().into_owned().collect();
8539 assert_eq!(
8540 query.get("after").map(String::as_str),
8541 Some("1970-01-01T00:00:00.000Z")
8542 );
8543 assert!(!query.contains_key("until"), "{query:?}");
8544
8545 let before_epoch = DateTime::UNIX_EPOCH - TimeDelta::days(1);
8547 let Ok(read) = client_for(&server)
8548 .fetch_trades(DateTime::<Utc>::MIN_UTC, before_epoch, &[])
8549 .await
8550 else {
8551 panic!("the read succeeds");
8552 };
8553 assert_eq!(read, TradesRead::complete(Vec::new()));
8554 assert_eq!(server.received_requests().await.map(|r| r.len()), Some(1));
8555 }
8556
8557 #[tokio::test]
8559 async fn fetch_trades_reads_nothing_for_an_empty_span() {
8560 use crate::client::ExecutionClient;
8561
8562 let server = MockServer::start().await;
8563 let start = recovery_after();
8564 let Ok(read) = client_for(&server)
8565 .fetch_trades(start, start - TimeDelta::seconds(1), &[])
8566 .await
8567 else {
8568 panic!("an empty span reads");
8569 };
8570 assert_eq!(read, TradesRead::complete(Vec::new()));
8571 assert_eq!(server.received_requests().await.map(|r| r.len()), Some(0));
8572 }
8573
8574 #[tokio::test]
8577 async fn recovery_reads_from_the_start_of_the_disconnects_millisecond() {
8578 let server = MockServer::start().await;
8579 Mock::given(method("GET"))
8580 .and(path("/v2/account/activities"))
8581 .and(wiremock::matchers::query_param(
8582 "after",
8583 "2025-01-01T00:00:00.123Z",
8584 ))
8585 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([])))
8586 .expect(1)
8587 .mount(&server)
8588 .await;
8589
8590 let (tx, _rx) = mpsc::unbounded_channel();
8591 let outcome = recover_fills(
8592 &reqwest::Client::new(),
8593 &RateLimitTracker::new(),
8594 &[],
8595 &server.uri(),
8596 time("2025-01-01T00:00:00.123456Z"),
8597 &tx,
8598 &new_dedup_cache(),
8599 &KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
8600 )
8601 .await;
8602 assert_eq!(outcome, Ok(()));
8603 server.verify().await;
8604 }
8605
8606 fn dedup_keys(dedup: &SharedDedupCache) -> Vec<String> {
8608 let mut keys: Vec<_> = dedup
8609 .lock()
8610 .iter()
8611 .filter_map(|(key, _)| match key {
8612 FillKey::Cumulative(key) => Some(key.to_string()),
8613 FillKey::Execution(_) | FillKey::CumulativeWithoutId(_) => None,
8614 })
8615 .collect();
8616 keys.reverse();
8618 keys
8619 }
8620
8621 fn trade_of(
8622 event: &UnindexedAccountEvent,
8623 ) -> &Trade<AssetNameExchange, InstrumentNameExchange> {
8624 match &event.kind {
8625 AccountEventKind::Trade(t) => t,
8626 other => panic!("expected a Trade event, got {other:?}"),
8627 }
8628 }
8629
8630 #[tokio::test]
8633 async fn a_recovered_fill_reports_the_cumulative_alpaca_sent() {
8634 let events = drive_recover_fills(vec![
8635 activity_json("act-1", "ord-1", "2", Some("2")),
8636 activity_json("act-2", "ord-1", "3", Some("5")),
8637 ])
8638 .await;
8639
8640 let reported: Vec<_> = events
8641 .iter()
8642 .map(|e| trade_of(e).order_filled_quantity)
8643 .collect();
8644 assert_eq!(
8645 reported,
8646 vec![
8647 Some(Decimal::from_str("2").unwrap()),
8648 Some(Decimal::from_str("5").unwrap())
8649 ]
8650 );
8651 }
8652
8653 #[tokio::test]
8660 async fn the_dedup_key_uses_the_venue_cumulative_not_an_intra_batch_count() {
8661 let dedup = new_dedup_cache();
8663 drive_recover_fills_with(
8664 vec![activity_json("act-1", "ord-1", "2", Some("5"))],
8665 dedup.clone(),
8666 &KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
8667 )
8668 .await;
8669
8670 assert_eq!(
8671 dedup_keys(&dedup),
8672 vec!["ord-1:5"],
8673 "counting executions within the batch would key this as ord-1:2"
8674 );
8675 }
8676
8677 async fn recover_after_ws_fill(
8681 execution_id: Option<&str>,
8682 activity_id: &str,
8683 ) -> Vec<UnindexedAccountEvent> {
8684 let dedup = new_dedup_cache();
8685 let update = AlpacaTradeUpdate {
8686 event: SmolStr::new("partial_fill"),
8687 execution_id,
8688 previous_execution_id: None,
8689 order: super::make_order_ws("ord-1", "SPY", "buy", "5"),
8690 price: Some("100.00"),
8691 qty: Some("2"),
8692 timestamp: None,
8693 };
8694 assert!(
8695 !is_duplicate_fill(&dedup, ws_execution_id(&update), early_dedup_key(&update)),
8696 "precondition: first sighting"
8697 );
8698
8699 drive_recover_fills_with(
8700 vec![activity_json(activity_id, "ord-1", "2", Some("5"))],
8701 dedup,
8702 &KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
8703 )
8704 .await
8705 }
8706
8707 #[tokio::test]
8710 async fn a_fill_already_delivered_over_websocket_is_not_recovered_twice() {
8711 for (ws_execution_id, activity_id) in [
8712 (Some("exec-a"), "20250418103000000::exec-a"),
8713 (None, "20250418103000000::exec-a"),
8714 (Some("exec-a"), "act-1"),
8715 ] {
8716 let events = recover_after_ws_fill(ws_execution_id, activity_id).await;
8717 assert!(
8718 events.is_empty(),
8719 "a fill already delivered over WebSocket ({ws_execution_id:?}) must not be \
8720 re-delivered by recovery ({activity_id}), got {events:?}"
8721 );
8722 }
8723 }
8724
8725 async fn recover_after_correction(fields: &str) -> Vec<UnindexedAccountEvent> {
8728 let (tx, _rx) = mpsc::unbounded_channel();
8729 let dedup = new_dedup_cache();
8730 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
8731 process_ws_text(
8732 &super::amendment_frame("trade_correct", fields),
8733 &tx,
8734 &dedup,
8735 &known,
8736 &mut ExponentialBackoff::new(),
8737 );
8738
8739 drive_recover_fills_with(
8740 vec![activity_json(
8741 "20250418103100000::exec-c",
8742 "ord-1",
8743 "2",
8744 Some("2"),
8745 )],
8746 dedup,
8747 &known,
8748 )
8749 .await
8750 }
8751
8752 #[tokio::test]
8755 async fn a_corrected_execution_is_not_recovered_again() {
8756 let events = recover_after_correction(
8757 r#""execution_id":"exec-c","previous_execution_id":"exec-a","price":"101.50","qty":"2","#,
8758 )
8759 .await;
8760 assert!(
8761 events.is_empty(),
8762 "the corrected execution was delivered as the replacement, got {events:?}"
8763 );
8764 }
8765
8766 #[tokio::test]
8768 async fn an_unresolved_correction_does_not_keep_its_execution_from_recovery() {
8769 let events = recover_after_correction(
8770 r#""execution_id":"exec-c","previous_execution_id":"exec-a","qty":"2","#,
8771 )
8772 .await;
8773 let ids: Vec<_> = events.iter().map(|e| trade_of(e).id.0.as_str()).collect();
8774 assert_eq!(ids, vec!["exec-c"]);
8775 }
8776
8777 #[tokio::test]
8780 async fn a_different_execution_at_a_cumulative_already_seen_is_recovered() {
8781 let events = recover_after_ws_fill(Some("exec-a"), "20250418103000000::exec-b").await;
8782 let ids: Vec<_> = events.iter().map(|e| trade_of(e).id.0.as_str()).collect();
8783 assert_eq!(ids, vec!["exec-b"]);
8784 }
8785
8786 #[tokio::test]
8789 async fn without_a_reported_cumulative_the_previous_behaviour_is_preserved() {
8790 let dedup = new_dedup_cache();
8791 let events = drive_recover_fills_with(
8792 vec![
8793 activity_json("act-1", "ord-1", "2", None),
8794 activity_json("act-2", "ord-1", "3", None),
8795 ],
8796 dedup.clone(),
8797 &KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
8798 )
8799 .await;
8800
8801 assert_eq!(
8802 dedup_keys(&dedup),
8803 vec!["ord-1:2", "ord-1:5"],
8804 "intra-batch accumulation"
8805 );
8806 let ids: Vec<_> = events.iter().map(|e| trade_of(e).id.0.as_str()).collect();
8807 assert_eq!(ids, vec!["act-1", "act-2"], "each keeps its activity's id");
8808
8809 assert!(
8810 events
8811 .iter()
8812 .all(|e| trade_of(e).order_filled_quantity.is_none()),
8813 "a venue that reported no cumulative must not have one invented for it"
8814 );
8815 }
8816
8817 fn order_json(cid: &str, symbol: &str, status: &str, filled: &str) -> serde_json::Value {
8822 serde_json::json!({
8823 "id": format!("id-{cid}"),
8824 "client_order_id": cid,
8825 "symbol": symbol,
8826 "qty": "2",
8827 "filled_qty": filled,
8828 "filled_avg_price": if filled == "0" { None } else { Some("101.5") },
8829 "status": status,
8830 "side": "buy",
8831 "type": "limit",
8832 "time_in_force": "gtc",
8833 "limit_price": "100",
8834 "created_at": "2026-10-01T14:30:00Z",
8835 "updated_at": "2026-10-01T15:00:00Z"
8836 })
8837 }
8838
8839 async fn mount_lookup(server: &MockServer, cid: &str, order: ResponseTemplate, calls: u64) {
8841 use wiremock::matchers::query_param;
8842 Mock::given(method("GET"))
8843 .and(path("/v2/orders:by_client_order_id"))
8844 .and(query_param("client_order_id", cid))
8845 .respond_with(order)
8846 .expect(calls)
8847 .mount(server)
8848 .await;
8849 }
8850
8851 fn ok(body: serde_json::Value) -> ResponseTemplate {
8852 ResponseTemplate::new(200).set_body_json(body)
8853 }
8854
8855 fn alpaca_key(instrument: &str, cid: &str) -> UnindexedOrderKey {
8856 OrderKey::new(
8857 ExchangeId::AlpacaBroker,
8858 InstrumentNameExchange::new(instrument),
8859 StrategyId::new("strategy"),
8860 ClientOrderId::new(cid),
8861 )
8862 }
8863
8864 fn hold(known: &SharedKnownLiveOrders, key: &UnindexedOrderKey) {
8865 let open = Open::new(
8866 VenueOrderId::Assigned(OrderId::new(format!("id-{}", key.cid))),
8867 Utc::now(),
8868 Decimal::ZERO,
8869 );
8870 known
8871 .lock()
8872 .live(key, Decimal::TWO, OrderKind::Limit, &open);
8873 }
8874
8875 #[tokio::test]
8876 async fn fetch_ended_orders_reads_each_end_alpaca_reports() {
8877 use rust_decimal_macros::dec;
8878
8879 let server = MockServer::start().await;
8880 for (cid, status, filled) in [
8881 ("filled", "filled", "2"),
8882 ("canceled", "canceled", "1"),
8883 ("replaced", "replaced", "0"),
8884 ("expired", "expired", "1"),
8885 ("rejected", "rejected", "0"),
8886 ("done", "done_for_day", "1"),
8887 ("live", "new", "0"),
8888 ("unknown-status", "frozen", "0"),
8889 ] {
8890 mount_lookup(&server, cid, ok(order_json(cid, "SPY", status, filled)), 1).await;
8891 }
8892 mount_lookup(
8893 &server,
8894 "elsewhere",
8895 ok(order_json("elsewhere", "QQQ", "canceled", "0")),
8896 1,
8897 )
8898 .await;
8899 mount_lookup(
8900 &server,
8901 "gone",
8902 ResponseTemplate::new(404).set_body_json(serde_json::json!({
8903 "code": 40410000,
8904 "message": "order not found for gone"
8905 })),
8906 1,
8907 )
8908 .await;
8909
8910 let keys: Vec<_> = [
8911 "filled",
8912 "canceled",
8913 "replaced",
8914 "expired",
8915 "rejected",
8916 "done",
8917 "live",
8918 "unknown-status",
8919 "elsewhere",
8920 "gone",
8921 ]
8922 .into_iter()
8923 .map(|cid| alpaca_key("spy", cid))
8924 .collect();
8925 let ended = client_for(&server).fetch_ended_orders(&keys).await.unwrap();
8926
8927 let mut found: Vec<_> = ended
8928 .iter()
8929 .map(|order| (order.key.cid.0.as_str(), &order.state))
8930 .collect();
8931 found.sort_by_key(|(cid, _)| *cid);
8932 let [
8933 (canceled, InactiveOrderState::Cancelled(cancelled)),
8934 (expired, InactiveOrderState::Expired(expiry)),
8935 (filled, InactiveOrderState::FullyFilled(fill)),
8936 (rejected, InactiveOrderState::OpenFailed(failed)),
8937 (replaced, InactiveOrderState::Cancelled(replacement)),
8938 ] = found.as_slice()
8939 else {
8940 panic!("filled, canceled, replaced, expired and rejected: {found:?}");
8941 };
8942 assert_eq!(
8943 [*canceled, *expired, *filled, *rejected, *replaced],
8944 ["canceled", "expired", "filled", "rejected", "replaced"]
8945 );
8946 assert_eq!(
8947 (fill.filled_quantity, fill.avg_price),
8948 (dec!(2), Some(dec!(101.5)))
8949 );
8950 assert_eq!(
8951 cancelled.filled_quantity,
8952 Some(dec!(1)),
8953 "what filled before"
8954 );
8955 assert_eq!(expiry.filled_quantity, Some(dec!(1)));
8956 assert_eq!(replacement.filled_quantity, Some(Decimal::ZERO));
8957 assert_eq!(
8958 cancelled.time_exchange,
8959 parse_timestamp("2026-10-01T15:00:00Z").unwrap(),
8960 "stamped when the order last changed"
8961 );
8962 assert!(matches!(
8963 failed,
8964 OrderError::Rejected(ApiError::OrderRejected(message))
8965 if message.contains("id-rejected")
8966 ));
8967 assert!(
8968 ended
8969 .iter()
8970 .all(|order| order.key.strategy == StrategyId::new("strategy")
8971 && order.key.instrument == InstrumentNameExchange::new("spy")),
8972 "each carries the key it was asked for"
8973 );
8974 }
8975
8976 #[tokio::test]
8977 async fn fetch_ended_orders_fails_when_a_lookup_fails() {
8978 let server = MockServer::start().await;
8979 mount_lookup(&server, "a", ok(order_json("a", "SPY", "canceled", "0")), 1).await;
8980 mount_lookup(&server, "b", ResponseTemplate::new(500), 1).await;
8981
8982 let result = client_for(&server)
8983 .fetch_ended_orders(&[alpaca_key("SPY", "a"), alpaca_key("SPY", "b")])
8984 .await;
8985
8986 assert!(
8987 matches!(result, Err(UnindexedClientError::Connectivity(_))),
8988 "{result:?}"
8989 );
8990 }
8991
8992 #[tokio::test]
8995 async fn a_reconnect_check_lists_once_and_reports_the_orders_that_ended() {
8996 use wiremock::matchers::query_param;
8997
8998 let server = MockServer::start().await;
8999 Mock::given(method("GET"))
9000 .and(path("/v2/orders"))
9001 .and(query_param("status", "open"))
9002 .and(query_param("symbols", "QQQ,SPY"))
9003 .respond_with(ok(serde_json::json!([order_json(
9004 "spy-listed",
9005 "SPY",
9006 "new",
9007 "0"
9008 )])))
9009 .expect(1)
9010 .mount(&server)
9011 .await;
9012 mount_lookup(&server, "spy-listed", ResponseTemplate::new(500), 0).await;
9013 mount_lookup(
9014 &server,
9015 "spy-cancelled",
9016 ok(order_json("spy-cancelled", "SPY", "canceled", "1")),
9017 1,
9018 )
9019 .await;
9020 mount_lookup(
9021 &server,
9022 "spy-done",
9023 ok(order_json("spy-done", "SPY", "done_for_day", "1")),
9024 1,
9025 )
9026 .await;
9027 mount_lookup(
9028 &server,
9029 "qqq-filled",
9030 ok(order_json("qqq-filled", "QQQ", "filled", "2")),
9031 1,
9032 )
9033 .await;
9034
9035 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
9036 let [listed, cancelled, done, filled] = [
9037 ("SPY", "spy-listed"),
9038 ("SPY", "spy-cancelled"),
9039 ("SPY", "spy-done"),
9040 ("QQQ", "qqq-filled"),
9041 ]
9042 .map(|(instrument, cid)| alpaca_key(instrument, cid));
9043 for key in [&listed, &cancelled, &done, &filled] {
9044 hold(&known, key);
9045 }
9046 let mut unchecked = UncheckedOrders::default();
9047 unchecked.open(known.lock().instruments());
9048 let config = Arc::new(AlpacaConfig::with_base_url(
9049 "test-key".into(),
9050 "test-secret".into(),
9051 server.uri(),
9052 ));
9053 let (tx, mut rx) = mpsc::unbounded_channel();
9054
9055 recover_alpaca_ended_orders(
9056 &reqwest::Client::new(),
9057 &Arc::new(RateLimitTracker::new()),
9058 &config,
9059 &known,
9060 &mut unchecked,
9061 &tx,
9062 )
9063 .await;
9064
9065 let mut reported: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok())
9066 .map(|event| {
9067 let AccountEventKind::OrderSnapshot(
9068 rustrade_integration::collection::snapshot::Snapshot(order),
9069 ) = event.kind
9070 else {
9071 panic!("an order snapshot: {event:?}");
9072 };
9073 assert!(matches!(order.state, OrderState::Inactive(_)), "{order:?}");
9074 order.key.cid
9075 })
9076 .collect();
9077 reported.sort();
9078 assert_eq!(reported, [filled.cid.clone(), cancelled.cid.clone()]);
9079 let known = known.lock();
9080 assert!(known.contains(&listed.cid) && known.contains(&done.cid));
9081 assert!(!known.contains(&cancelled.cid) && !known.contains(&filled.cid));
9082 assert!(unchecked.is_empty(), "both instruments checked");
9083 }
9084
9085 #[tokio::test]
9087 async fn a_failed_listing_is_retried_for_every_instrument_in_it() {
9088 let server = MockServer::start().await;
9089 Mock::given(method("GET"))
9090 .and(path("/v2/orders"))
9091 .respond_with(ResponseTemplate::new(503))
9092 .expect(1)
9093 .mount(&server)
9094 .await;
9095
9096 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
9097 hold(&known, &alpaca_key("SPY", "a"));
9098 hold(&known, &alpaca_key("QQQ", "b"));
9099 let mut unchecked = UncheckedOrders::default();
9100 unchecked.open(known.lock().instruments());
9101 let config = Arc::new(AlpacaConfig::with_base_url(
9102 "test-key".into(),
9103 "test-secret".into(),
9104 server.uri(),
9105 ));
9106 let (tx, mut rx) = mpsc::unbounded_channel();
9107
9108 recover_alpaca_ended_orders(
9109 &reqwest::Client::new(),
9110 &Arc::new(RateLimitTracker::new()),
9111 &config,
9112 &known,
9113 &mut unchecked,
9114 &tx,
9115 )
9116 .await;
9117
9118 assert!(rx.try_recv().is_err(), "nothing reported");
9119 let now = tokio::time::Instant::now();
9120 assert!(
9121 unchecked.ready(&NoPendingFills, now).is_empty(),
9122 "both wait"
9123 );
9124 for instrument in ["QQQ", "SPY"] {
9125 assert_eq!(
9126 unchecked.failed(&InstrumentNameExchange::new(instrument), now),
9127 Some(crate::client::order_recovery::GapFailure::Retry(
9128 Duration::from_secs(crate::client::order_recovery::GAP_RETRY_BASE_SECS * 2)
9129 )),
9130 "{instrument} has failed once already"
9131 );
9132 }
9133 assert_eq!(known.lock().instruments().len(), 2, "both still held");
9134 }
9135
9136 #[tokio::test]
9137 async fn placing_listing_and_cancelling_keep_the_held_orders_in_step() {
9138 use crate::order::request::{OrderRequestCancel, RequestCancel, RequestOpen};
9139
9140 let server = MockServer::start().await;
9141 Mock::given(method("POST"))
9142 .and(path("/v2/orders"))
9143 .respond_with(ok(order_json("resting", "SPY", "new", "0")))
9144 .up_to_n_times(1)
9145 .mount(&server)
9146 .await;
9147 Mock::given(method("POST"))
9148 .and(path("/v2/orders"))
9149 .respond_with(ok(order_json("filled", "SPY", "filled", "2")))
9150 .mount(&server)
9151 .await;
9152 Mock::given(method("GET"))
9153 .and(path("/v2/orders"))
9154 .respond_with(ok(serde_json::json!([order_json(
9155 "listed", "QQQ", "new", "0"
9156 )])))
9157 .mount(&server)
9158 .await;
9159 Mock::given(method("DELETE"))
9160 .and(path("/v2/orders/id-resting"))
9161 .respond_with(ResponseTemplate::new(204))
9162 .mount(&server)
9163 .await;
9164 let client = client_for(&server);
9165 let spy = InstrumentNameExchange::new("SPY");
9166
9167 for cid in ["resting", "filled"] {
9168 let request = OrderRequestOpen {
9169 key: OrderKey::new(
9170 ExchangeId::AlpacaBroker,
9171 &spy,
9172 StrategyId::new("strategy"),
9173 ClientOrderId::new(cid),
9174 ),
9175 state: RequestOpen {
9176 side: Side::Buy,
9177 price: Some(Decimal::ONE_HUNDRED),
9178 quantity: Decimal::TWO,
9179 kind: OrderKind::Limit,
9180 time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
9181 position_id: None,
9182 reduce_only: false,
9183 market: None,
9184 },
9185 };
9186 client.open_order(request).await;
9187 }
9188 client.fetch_open_orders(&[]).await.unwrap();
9189 {
9190 let known = client.known_live.lock();
9191 assert!(known.contains(&ClientOrderId::new("resting")));
9192 assert!(
9193 !known.contains(&ClientOrderId::new("filled")),
9194 "filled in its placement response"
9195 );
9196 assert!(known.contains(&ClientOrderId::new("listed")));
9197 }
9198
9199 let cancel = client
9200 .cancel_order(OrderRequestCancel {
9201 key: OrderKey::new(
9202 ExchangeId::AlpacaBroker,
9203 &spy,
9204 StrategyId::new("strategy"),
9205 ClientOrderId::new("resting"),
9206 ),
9207 state: RequestCancel {
9208 id: Some(VenueOrderId::Assigned(OrderId::new("id-resting"))),
9209 },
9210 })
9211 .await;
9212 let Ok(cancelled) = &cancel.state else {
9213 panic!("expected Ok, got {cancel:?}");
9214 };
9215 assert_eq!(
9216 cancelled.filled_quantity, None,
9217 "the DELETE response has no body, so what filled is unknown"
9218 );
9219 assert!(
9220 client
9221 .known_live
9222 .lock()
9223 .contains(&ClientOrderId::new("resting")),
9224 "a cancel Alpaca has only accepted has not ended the order"
9225 );
9226 }
9227
9228 #[tokio::test]
9235 async fn an_order_that_ended_in_its_placement_response_is_reported_as_it_ended() {
9236 use crate::order::request::RequestOpen;
9237
9238 let Some(time) = parse_timestamp("2026-10-01T15:00:00Z") else {
9239 panic!("order_json's updated_at parses");
9240 };
9241 let id = |cid: &str| OrderId::new(format!("id-{cid}"));
9242 let open = |cid: &str, filled| {
9243 OrderState::active(Open::new(VenueOrderId::Assigned(id(cid)), time, filled))
9244 };
9245 let cases: [(&str, &str, &str, UnindexedOrderState, bool); 13] = [
9246 (
9247 "ioc-unfilled",
9248 "expired",
9249 "0",
9250 OrderState::expired(Expired::new(
9251 id("ioc-unfilled"),
9252 time,
9253 Some(Decimal::ZERO),
9254 )),
9255 false,
9256 ),
9257 (
9258 "ioc-partial",
9259 "expired",
9260 "1",
9261 OrderState::expired(Expired::new(id("ioc-partial"), time, Some(Decimal::ONE))),
9262 false,
9263 ),
9264 (
9265 "fill-unknown",
9266 "expired",
9267 "not-a-number",
9268 OrderState::expired(Expired::new(id("fill-unknown"), time, None)),
9269 false,
9270 ),
9271 (
9272 "cancelled",
9273 "canceled",
9274 "1",
9275 OrderState::inactive(Cancelled::new(id("cancelled"), time, Some(Decimal::ONE))),
9276 false,
9277 ),
9278 (
9279 "rejected",
9280 "rejected",
9281 "0",
9282 OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
9283 "Alpaca rejected order id-rejected after accepting it".to_string(),
9284 ))),
9285 false,
9286 ),
9287 (
9288 "filled",
9289 "filled",
9290 "2",
9291 OrderState::fully_filled(Filled::new(
9292 id("filled"),
9293 time,
9294 Decimal::TWO,
9295 Some(Decimal::new(1015, 1)),
9296 )),
9297 false,
9298 ),
9299 (
9300 "replaced",
9301 "replaced",
9302 "0",
9303 OrderState::inactive(Cancelled::new(id("replaced"), time, Some(Decimal::ZERO))),
9304 false,
9305 ),
9306 (
9307 "live-covered",
9308 "partially_filled",
9309 "2",
9310 OrderState::fully_filled(Filled::new(
9311 id("live-covered"),
9312 time,
9313 Decimal::TWO,
9314 Some(Decimal::new(1015, 1)),
9315 )),
9316 false,
9317 ),
9318 ("new", "new", "0", open("new", Decimal::ZERO), true),
9319 (
9320 "live-fill-unknown",
9321 "new",
9322 "not-a-number",
9323 open("live-fill-unknown", Decimal::ZERO),
9324 true,
9325 ),
9326 (
9327 "partial",
9328 "partially_filled",
9329 "1",
9330 open("partial", Decimal::ONE),
9331 true,
9332 ),
9333 (
9334 "for-the-day",
9335 "done_for_day",
9336 "1",
9337 open("for-the-day", Decimal::ONE),
9338 true,
9339 ),
9340 (
9341 "unknown",
9342 "a_status_not_yet_documented",
9343 "0",
9344 open("unknown", Decimal::ZERO),
9345 true,
9346 ),
9347 ];
9348
9349 for (cid, status, filled, expected, held) in cases {
9350 for bracket in [false, true] {
9351 let server = MockServer::start().await;
9352 Mock::given(method("POST"))
9353 .and(path("/v2/orders"))
9354 .respond_with(ok(order_json(cid, "SPY", status, filled)))
9355 .expect(1)
9356 .mount(&server)
9357 .await;
9358 let client = client_for(&server);
9359 let spy = InstrumentNameExchange::new("SPY");
9360 let time_in_force = TimeInForce::GoodUntilCancelled { post_only: false };
9361
9362 let state = if bracket {
9363 client
9364 .open_bracket_order(AlpacaBracketOrderRequest::new(
9365 spy,
9366 StrategyId::new("strategy"),
9367 ClientOrderId::new(cid),
9368 Side::Buy,
9369 Decimal::TWO,
9370 Decimal::ONE_HUNDRED,
9371 Decimal::new(120, 0),
9372 Decimal::new(90, 0),
9373 time_in_force,
9374 ))
9375 .await
9376 .parent
9377 .state
9378 } else {
9379 let request = OrderRequestOpen {
9380 key: OrderKey::new(
9381 ExchangeId::AlpacaBroker,
9382 &spy,
9383 StrategyId::new("strategy"),
9384 ClientOrderId::new(cid),
9385 ),
9386 state: RequestOpen {
9387 side: Side::Buy,
9388 price: Some(Decimal::ONE_HUNDRED),
9389 quantity: Decimal::TWO,
9390 kind: OrderKind::Limit,
9391 time_in_force,
9392 position_id: None,
9393 reduce_only: false,
9394 market: None,
9395 },
9396 };
9397 client.open_order(request).await.state
9398 };
9399
9400 assert_eq!(state, expected, "{status} {filled}, bracket: {bracket}");
9401 assert_eq!(
9402 client.known_live.lock().contains(&ClientOrderId::new(cid)),
9403 held,
9404 "{status} {filled}, bracket: {bracket}"
9405 );
9406 }
9407 }
9408 }
9409
9410 #[test]
9413 fn the_stream_keeps_the_held_orders_in_step() {
9414 let (tx, mut rx) = mpsc::unbounded_channel();
9415 let dedup = new_dedup_cache();
9416 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
9417 let mut backoff = ExponentialBackoff::new();
9418 let frame = |event: &str, filled: &str| {
9419 format!(
9420 r#"{{"stream":"trade_updates","data":{{"event":"{event}","order":{{"id":"id-a","client_order_id":"a","symbol":"SPY","qty":"2","filled_qty":"{filled}","side":"buy","type":"limit","time_in_force":"gtc","limit_price":"100","status":"{event}"}}}}}}"#
9421 )
9422 };
9423 let cid = ClientOrderId::new("a");
9424
9425 process_ws_text(&frame("new", "0"), &tx, &dedup, &known, &mut backoff);
9426 assert!(known.lock().contains(&cid), "acknowledged");
9427
9428 process_ws_text(
9429 &frame("done_for_day", "1"),
9430 &tx,
9431 &dedup,
9432 &known,
9433 &mut backoff,
9434 );
9435 assert!(known.lock().contains(&cid), "done for the day is not ended");
9436
9437 process_ws_text(&frame("canceled", "1"), &tx, &dedup, &known, &mut backoff);
9438 assert!(!known.lock().contains(&cid), "cancelled");
9439
9440 let events: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
9441 let [_, done, cancelled] = events.as_slice() else {
9442 panic!("three events: {events:?}");
9443 };
9444 let AccountEventKind::OrderSnapshot(
9445 rustrade_integration::collection::snapshot::Snapshot(order),
9446 ) = &done.kind
9447 else {
9448 panic!("done_for_day reports an order snapshot: {done:?}");
9449 };
9450 let OrderState::Active(ActiveOrderState::Open(open)) = &order.state else {
9451 panic!("open: {order:?}");
9452 };
9453 assert_eq!(open.filled_quantity, Decimal::ONE, "with what it filled");
9454 assert!(matches!(
9455 cancelled.kind,
9456 AccountEventKind::OrderCancelled(_)
9457 ));
9458 }
9459
9460 #[tokio::test]
9463 async fn a_recovered_fill_that_completes_an_order_stops_it_being_held() {
9464 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
9465 let key = alpaca_key("SPY", "a");
9466 known.lock().live(
9467 &key,
9468 Decimal::TWO,
9469 OrderKind::Limit,
9470 &Open::new(
9471 VenueOrderId::Assigned(OrderId::new("ord-1")),
9472 Utc::now(),
9473 Decimal::ZERO,
9474 ),
9475 );
9476
9477 drive_recover_fills_with(
9478 vec![activity_json("act-1", "ord-1", "1", Some("1"))],
9479 new_dedup_cache(),
9480 &known,
9481 )
9482 .await;
9483 assert!(known.lock().contains(&key.cid), "half filled");
9484
9485 drive_recover_fills_with(
9486 vec![activity_json("act-2", "ord-1", "1", Some("2"))],
9487 new_dedup_cache(),
9488 &known,
9489 )
9490 .await;
9491 assert!(!known.lock().contains(&key.cid), "filled");
9492 }
9493
9494 #[tokio::test]
9497 async fn the_check_loop_runs_at_once_then_waits_out_a_failure() {
9498 let server = MockServer::start().await;
9499 Mock::given(method("GET"))
9500 .and(path("/v2/orders"))
9501 .respond_with(ResponseTemplate::new(503))
9502 .expect(1)
9503 .mount(&server)
9504 .await;
9505 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
9506 hold(&known, &alpaca_key("SPY", "a"));
9507 let mut unchecked = UncheckedOrders::default();
9508 unchecked.open(known.lock().instruments());
9509 let config = Arc::new(AlpacaConfig::with_base_url(
9510 "test-key".into(),
9511 "test-secret".into(),
9512 server.uri(),
9513 ));
9514 let (tx, _rx) = mpsc::unbounded_channel();
9515
9516 let ran = tokio::time::timeout(
9517 Duration::from_secs(1),
9518 run_order_checks(
9519 &reqwest::Client::new(),
9520 &Arc::new(RateLimitTracker::new()),
9521 &config,
9522 &known,
9523 &mut unchecked,
9524 &tx,
9525 ),
9526 )
9527 .await;
9528
9529 assert!(ran.is_err(), "the loop never completes");
9530 assert!(
9531 unchecked.contains(&InstrumentNameExchange::new("SPY")),
9532 "SPY waits for its retry"
9533 );
9534 assert_eq!(
9535 server.received_requests().await.unwrap().len(),
9536 1,
9537 "the failed check is not run again at once"
9538 );
9539 }
9540
9541 #[tokio::test]
9543 async fn the_check_loop_waits_once_the_consumer_has_gone() {
9544 let server = MockServer::start().await;
9545 Mock::given(method("GET"))
9546 .respond_with(ResponseTemplate::new(503))
9547 .expect(0)
9548 .mount(&server)
9549 .await;
9550 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
9551 hold(&known, &alpaca_key("SPY", "a"));
9552 let mut unchecked = UncheckedOrders::default();
9553 unchecked.open(known.lock().instruments());
9554 let config = Arc::new(AlpacaConfig::with_base_url(
9555 "test-key".into(),
9556 "test-secret".into(),
9557 server.uri(),
9558 ));
9559 let (tx, rx) = mpsc::unbounded_channel();
9560 drop(rx);
9561
9562 let ran = tokio::time::timeout(
9563 Duration::from_millis(200),
9564 run_order_checks(
9565 &reqwest::Client::new(),
9566 &Arc::new(RateLimitTracker::new()),
9567 &config,
9568 &known,
9569 &mut unchecked,
9570 &tx,
9571 ),
9572 )
9573 .await;
9574
9575 assert!(ran.is_err(), "the loop never completes");
9576 assert!(
9577 server.received_requests().await.unwrap().is_empty(),
9578 "nothing asked once the consumer has gone"
9579 );
9580 }
9581 }
9582}