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)]
1054pub struct AlpacaClient {
1055 config: Arc<AlpacaConfig>,
1056 http: reqwest::Client,
1059 rate_limiter: Arc<RateLimitTracker>,
1060 orders_url: String,
1062 known_live: SharedKnownLiveOrders,
1065}
1066
1067impl std::fmt::Debug for AlpacaClient {
1068 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1069 f.debug_struct("AlpacaClient")
1070 .field("paper", &self.config.paper)
1071 .finish_non_exhaustive()
1072 }
1073}
1074
1075impl AlpacaClient {
1076 #[allow(clippy::expect_used)] fn build_http(config: &AlpacaConfig) -> reqwest::Client {
1084 use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
1085 let mut headers = HeaderMap::new();
1086 headers.insert(
1088 HeaderName::from_static("apca-api-key-id"),
1089 HeaderValue::from_str(&config.api_key)
1090 .expect("Alpaca API key contains invalid header characters"),
1091 );
1092 headers.insert(
1093 HeaderName::from_static("apca-api-secret-key"),
1094 HeaderValue::from_str(&config.secret_key)
1095 .expect("Alpaca secret key contains invalid header characters"),
1096 );
1097 reqwest::Client::builder()
1098 .default_headers(headers)
1099 .build()
1100 .expect("failed to build reqwest client for Alpaca")
1101 }
1102
1103 fn base_url(&self) -> &str {
1104 self.config.rest_base_url()
1105 }
1106
1107 fn remember_live<'a>(
1109 &self,
1110 orders: impl IntoIterator<Item = &'a Order<ExchangeId, InstrumentNameExchange, Open>>,
1111 ) {
1112 let mut known = self.known_live.lock();
1113 for order in orders {
1114 known.live(&order.key, order.quantity, order.kind, &order.state);
1115 }
1116 }
1117}
1118
1119fn parse_rate_limit_delay(headers: &reqwest::header::HeaderMap) -> Option<Duration> {
1130 headers
1131 .get("x-ratelimit-reset")
1132 .and_then(|v| v.to_str().ok())
1133 .and_then(|s| s.parse::<u64>().ok())
1134 .map(|reset_ts| {
1135 let now_secs = std::time::SystemTime::now()
1136 .duration_since(std::time::UNIX_EPOCH)
1137 .unwrap_or_default()
1138 .as_secs();
1139 Duration::from_secs(reset_ts.saturating_sub(now_secs).max(1))
1140 })
1141}
1142
1143fn observe_rate_limit_remaining(
1150 rate_limiter: &RateLimitTracker,
1151 headers: &reqwest::header::HeaderMap,
1152) {
1153 let exhausted = headers
1154 .get("x-ratelimit-remaining")
1155 .and_then(|v| v.to_str().ok())
1156 .and_then(|s| s.parse::<u32>().ok())
1157 == Some(0);
1158 if !exhausted {
1159 return;
1160 }
1161 match parse_rate_limit_delay(headers) {
1162 Some(reset) => rate_limiter
1165 .on_bucket_exhausted(reset.min(Duration::from_secs(DEFAULT_RATE_LIMIT_DELAY_SECS))),
1166 None => debug!(
1167 "Alpaca REST rate-limit window exhausted (X-Ratelimit-Remaining: 0) without X-Ratelimit-Reset"
1168 ),
1169 }
1170}
1171
1172async fn rest_with_retry<T>(
1191 rate_limiter: &RateLimitTracker,
1192 build_request: impl FnMut() -> reqwest::RequestBuilder,
1193) -> Result<T, UnindexedClientError>
1194where
1195 T: for<'de> Deserialize<'de>,
1196{
1197 match rest_request(rate_limiter, build_request).await? {
1198 Fetched::Found(value) => Ok(value),
1199 Fetched::NotFound(message) => Err(UnindexedClientError::Api(parse_api_error(
1200 reqwest::StatusCode::NOT_FOUND,
1201 &message,
1202 ))),
1203 }
1204}
1205
1206async fn rest_lookup_with_retry<T>(
1209 rate_limiter: &RateLimitTracker,
1210 build_request: impl FnMut() -> reqwest::RequestBuilder,
1211) -> Result<Option<T>, UnindexedClientError>
1212where
1213 T: for<'de> Deserialize<'de>,
1214{
1215 match rest_request(rate_limiter, build_request).await? {
1216 Fetched::Found(value) => Ok(Some(value)),
1217 Fetched::NotFound(_) => Ok(None),
1218 }
1219}
1220
1221#[derive(Debug)]
1223enum Fetched<T> {
1224 Found(T),
1226 NotFound(String),
1228}
1229
1230async fn rest_request<T>(
1232 rate_limiter: &RateLimitTracker,
1233 mut build_request: impl FnMut() -> reqwest::RequestBuilder,
1234) -> Result<Fetched<T>, UnindexedClientError>
1235where
1236 T: for<'de> Deserialize<'de>,
1237{
1238 for attempt in 0..MAX_RATE_LIMIT_ATTEMPTS {
1239 rate_limiter.wait_if_blocked().await;
1240 let response = build_request()
1241 .send()
1242 .await
1243 .map_err(|e| connectivity_err(format!("Alpaca REST request failed: {e}")))?;
1244
1245 let status = response.status();
1246
1247 if status == reqwest::StatusCode::TOO_MANY_REQUESTS {
1248 let reset_delay = parse_rate_limit_delay(response.headers());
1249
1250 if attempt + 1 < MAX_RATE_LIMIT_ATTEMPTS {
1251 warn!(
1252 attempt = attempt + 1,
1253 max_attempts = MAX_RATE_LIMIT_ATTEMPTS,
1254 "Alpaca REST rate-limited (429), retrying"
1255 );
1256 rate_limiter.on_rate_limited(reset_delay);
1257 continue;
1258 }
1259
1260 warn!(
1262 max_attempts = MAX_RATE_LIMIT_ATTEMPTS,
1263 "Alpaca REST rate-limit retries exhausted"
1264 );
1265 return Err(UnindexedClientError::Api(ApiError::RateLimit));
1266 }
1267
1268 observe_rate_limit_remaining(rate_limiter, response.headers());
1269
1270 if status == reqwest::StatusCode::NO_CONTENT {
1274 return Err(UnindexedClientError::Internal(
1275 "Alpaca REST returned 204 No Content — use rest_delete_with_retry for DELETE endpoints"
1276 .to_string(),
1277 ));
1278 }
1279
1280 let bytes = response
1281 .bytes()
1282 .await
1283 .map_err(|e| connectivity_err(format!("Alpaca REST read body failed: {e}")))?;
1284
1285 if status.is_success() {
1288 return serde_json::from_slice::<T>(&bytes)
1289 .map(Fetched::Found)
1290 .map_err(|e| {
1291 UnindexedClientError::Internal(format!(
1292 "Alpaca REST JSON parse error ({status}): {e} | body: {}",
1293 String::from_utf8_lossy(&bytes)
1294 .chars()
1295 .take(200)
1296 .collect::<String>()
1297 ))
1298 });
1299 }
1300
1301 let api_err = serde_json::from_slice::<AlpacaApiError>(&bytes)
1303 .map(|e| e.message)
1304 .unwrap_or_else(|_| String::from_utf8_lossy(&bytes).into_owned());
1305
1306 if status == reqwest::StatusCode::NOT_FOUND {
1307 return Ok(Fetched::NotFound(api_err));
1308 }
1309
1310 if status.is_client_error() {
1319 return Err(UnindexedClientError::Api(parse_api_error(status, &api_err)));
1320 }
1321 return Err(connectivity_err(format!(
1322 "Alpaca REST error {status}: {api_err}"
1323 )));
1324 }
1325 unreachable!("Alpaca REST retry loop exited without returning")
1326}
1327
1328async fn rest_delete_with_retry(
1333 rate_limiter: &RateLimitTracker,
1334 mut build_request: impl FnMut() -> reqwest::RequestBuilder,
1335) -> Result<(), UnindexedOrderError> {
1336 for attempt in 0..MAX_RATE_LIMIT_ATTEMPTS {
1337 rate_limiter.wait_if_blocked().await;
1338 let response = build_request().send().await.map_err(|e| {
1339 UnindexedOrderError::Connectivity(ConnectivityError::Socket(format!(
1340 "Alpaca cancel request failed: {e}"
1341 )))
1342 })?;
1343
1344 let status = response.status();
1345
1346 if status == reqwest::StatusCode::TOO_MANY_REQUESTS {
1347 let reset_delay = parse_rate_limit_delay(response.headers());
1348
1349 if attempt + 1 < MAX_RATE_LIMIT_ATTEMPTS {
1350 warn!(
1351 attempt = attempt + 1,
1352 max_attempts = MAX_RATE_LIMIT_ATTEMPTS,
1353 "Alpaca cancel rate-limited (429), retrying"
1354 );
1355 rate_limiter.on_rate_limited(reset_delay);
1356 continue;
1357 }
1358
1359 warn!(
1361 max_attempts = MAX_RATE_LIMIT_ATTEMPTS,
1362 "Alpaca cancel rate-limit retries exhausted"
1363 );
1364 return Err(UnindexedOrderError::Rejected(ApiError::RateLimit));
1365 }
1366
1367 observe_rate_limit_remaining(rate_limiter, response.headers());
1368
1369 if status == reqwest::StatusCode::NO_CONTENT || status.is_success() {
1371 return Ok(());
1372 }
1373
1374 let bytes = response
1375 .bytes()
1376 .await
1377 .inspect_err(
1378 |e| warn!(%e, %status, "Alpaca cancel_order: failed to read error response body"),
1379 )
1380 .unwrap_or_default();
1381 let msg = serde_json::from_slice::<AlpacaApiError>(&bytes)
1382 .map(|e| e.message)
1383 .unwrap_or_else(|_| String::from_utf8_lossy(&bytes).into_owned());
1384
1385 return Err(parse_order_error(status, &msg));
1386 }
1387 unreachable!("Alpaca cancel retry loop exited without returning")
1388}
1389
1390impl ExecutionClient for AlpacaClient {
1395 const EXCHANGE: ExchangeId = ExchangeId::AlpacaBroker;
1396
1397 const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant] = &[
1404 InstrumentKindDiscriminant::Spot,
1405 InstrumentKindDiscriminant::Option,
1406 ];
1407 type Config = AlpacaConfig;
1408 type AccountStream = BoxStream<'static, UnindexedAccountEvent>;
1409
1410 fn new(config: Self::Config) -> Self {
1415 let http = Self::build_http(&config);
1416 let orders_url = format!("{}/v2/orders", config.rest_base_url());
1417 Self {
1418 config: Arc::new(config),
1419 http,
1420 rate_limiter: Arc::new(RateLimitTracker::new()),
1421 orders_url,
1422 known_live: KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
1423 }
1424 }
1425
1426 async fn account_snapshot(
1473 &self,
1474 assets: &[AssetNameExchange],
1475 instruments: &[InstrumentNameExchange],
1476 ) -> Result<UnindexedAccountSnapshot, UnindexedClientError> {
1477 let base = self.base_url();
1478 let http = self.http.clone();
1479 let rl = &self.rate_limiter;
1480
1481 let wants_usd = assets.is_empty() || assets.iter().any(is_usd);
1482 let wants_non_usd = assets.is_empty() || assets.iter().any(|a| !is_usd(a));
1483
1484 let account_url = format!("{base}/v2/account");
1486 let positions_url = format!("{base}/v2/positions");
1487 let (account, positions): (Option<AlpacaAccount>, Vec<AlpacaPosition>) = tokio::try_join!(
1488 async {
1489 if wants_usd {
1490 rest_with_retry(rl, || http.get(&account_url))
1491 .await
1492 .map(Some)
1493 } else {
1494 Ok(None)
1495 }
1496 },
1497 rest_with_retry(rl, || http.get(&positions_url)),
1498 )?;
1499
1500 let mut balances = account
1501 .map(|account| convert_account_to_balances(&account, assets))
1502 .unwrap_or_default();
1503 if wants_non_usd {
1504 balances.extend(convert_positions_to_balances(&positions, assets));
1505 }
1506
1507 let converted_positions = convert_positions(&positions, Utc::now())?;
1510 let open_orders = fetch_raw_open_orders(&http, rl, base, instruments).await?;
1511
1512 let instrument_snapshots =
1514 build_instrument_snapshots(open_orders, converted_positions, instruments);
1515 {
1516 let mut known = self.known_live.lock();
1517 for order in instrument_snapshots
1518 .iter()
1519 .flat_map(|snapshot| &snapshot.orders)
1520 {
1521 if let OrderState::Active(ActiveOrderState::Open(open)) = &order.state {
1522 known.live(&order.key, order.quantity, order.kind, open);
1523 }
1524 }
1525 }
1526
1527 Ok(AccountSnapshot::new(
1528 ExchangeId::AlpacaBroker,
1529 balances,
1530 instrument_snapshots,
1531 ))
1532 }
1533
1534 async fn account_stream(
1660 &self,
1661 _assets: &[AssetNameExchange], instruments: &[InstrumentNameExchange],
1666 ) -> Result<Self::AccountStream, UnindexedClientError> {
1667 let initial_ws = connect_and_subscribe(&self.config).await?;
1670
1671 let (tx, rx) = mpsc::unbounded_channel::<UnindexedAccountEvent>();
1676 let dedup = new_dedup_cache();
1677 let config = self.config.clone();
1678 let http = self.http.clone();
1679 let rate_limiter = self.rate_limiter.clone();
1680 let known = self.known_live.clone();
1681 let instruments = instruments.to_vec();
1682
1683 let cm_handle = tokio::spawn(connection_manager(
1684 tx,
1685 dedup,
1686 config,
1687 http,
1688 rate_limiter,
1689 known,
1690 instruments,
1691 Some(initial_ws),
1692 ));
1693
1694 let rx_stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx);
1695 let guarded = GracefulShutdownStream::new(rx_stream, cm_handle);
1696 Ok(futures::StreamExt::boxed(guarded))
1697 }
1698
1699 async fn cancel_order(
1718 &self,
1719 request: OrderRequestCancel<ExchangeId, &InstrumentNameExchange>,
1720 ) -> UnindexedOrderResponseCancel {
1721 let key = crate::order::OrderKey {
1722 exchange: request.key.exchange,
1723 instrument: request.key.instrument.clone(),
1724 strategy: request.key.strategy.clone(),
1725 cid: request.key.cid.clone(),
1726 };
1727
1728 let order_id: SmolStr = match request.state.id.as_ref().and_then(VenueOrderId::assigned) {
1732 Some(id) => id.0.clone(),
1733 None => {
1734 warn!(
1735 instrument = %key.instrument,
1736 "Alpaca cancel_order: no exchange order ID available (clientOrderId-only cancel not supported)"
1737 );
1738 return crate::order::request::OrderResponseCancel {
1739 key,
1740 state: Err(UnindexedOrderError::Rejected(ApiError::OrderRejected(
1741 "exchange order ID required for cancel (fetch_open_orders to resolve)"
1742 .into(),
1743 ))),
1744 };
1745 }
1746 };
1747
1748 let base = self.base_url();
1749 let http = self.http.clone();
1750 let url = format!("{base}/v2/orders/{order_id}");
1751
1752 match rest_delete_with_retry(&self.rate_limiter, || http.delete(&url)).await {
1753 Ok(()) => {
1754 let exchange_order_id = OrderId(order_id);
1755 crate::order::request::OrderResponseCancel {
1758 key,
1759 state: Ok(Cancelled::new(exchange_order_id, Utc::now(), None)),
1760 }
1761 }
1762 Err(e) => crate::order::request::OrderResponseCancel { key, state: Err(e) },
1763 }
1764 }
1765
1766 async fn open_order(
1787 &self,
1788 request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
1789 ) -> Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState> {
1790 let side = request.state.side;
1791 let reduce_only = request.state.reduce_only;
1792 self.open_order_inner(request, map_position_intent(side, reduce_only))
1793 .await
1794 }
1795
1796 async fn fetch_balances(
1804 &self,
1805 assets: &[AssetNameExchange],
1806 ) -> Result<Vec<AssetBalance<AssetNameExchange>>, UnindexedClientError> {
1807 let base = self.base_url();
1808 let http = self.http.clone();
1809 let mut result = Vec::new();
1810
1811 let wants_usd = assets.is_empty() || assets.iter().any(is_usd);
1814 if wants_usd {
1815 let account_url = format!("{base}/v2/account");
1817 let account: AlpacaAccount =
1818 rest_with_retry(&self.rate_limiter, || http.get(&account_url)).await?;
1819 result.extend(convert_account_to_balances(&account, assets));
1820 }
1821
1822 let wants_non_usd = assets.is_empty() || assets.iter().any(|a| !is_usd(a));
1824 if wants_non_usd {
1825 let positions_url = format!("{base}/v2/positions");
1827 let positions: Vec<AlpacaPosition> =
1828 rest_with_retry(&self.rate_limiter, || http.get(&positions_url)).await?;
1829 result.extend(convert_positions_to_balances(&positions, assets));
1830 }
1831
1832 Ok(result)
1833 }
1834
1835 async fn fetch_open_orders(
1836 &self,
1837 instruments: &[InstrumentNameExchange],
1838 ) -> Result<Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, UnindexedClientError> {
1839 let base = self.base_url();
1840 let http = self.http.clone();
1841
1842 let open_orders =
1843 fetch_raw_open_orders(&http, &self.rate_limiter, base, instruments).await?;
1844
1845 let result: Vec<_> = open_orders
1846 .into_iter()
1847 .filter_map(|o| convert_open_order(&o))
1848 .collect();
1849 self.remember_live(&result);
1850 Ok(result)
1851 }
1852
1853 async fn fetch_trades(
1866 &self,
1867 start: DateTime<Utc>,
1868 end: DateTime<Utc>,
1869 instruments: &[InstrumentNameExchange],
1870 ) -> Result<TradesRead<AssetNameExchange, InstrumentNameExchange>, UnindexedClientError> {
1871 if start > end || end < DateTime::UNIX_EPOCH {
1873 return Ok(TradesRead::complete(Vec::new()));
1874 }
1875 let page = paginate_activities(
1876 &self.http,
1877 &self.rate_limiter,
1878 self.base_url(),
1879 start,
1880 Some(end),
1881 )
1882 .await?;
1883
1884 let resume = page.resume.filter(|resume| *resume <= end);
1888 if resume.is_some_and(|resume| resume <= floor_millis(start)) {
1890 error!(
1891 %start,
1892 fills_read = page.activities.len(),
1893 "Alpaca fetch_trades: a full read did not advance past the span's first millisecond"
1894 );
1895 return Err(UnindexedClientError::Truncated {
1896 fills_read: page.activities.len(),
1897 });
1898 }
1899
1900 let instrument_set: fnv::FnvHashSet<&str> =
1903 instruments.iter().map(|i| i.name().as_str()).collect();
1904 let trades = page
1905 .activities
1906 .iter()
1907 .filter(|a| instrument_set.is_empty() || instrument_set.contains(a.symbol.as_str()))
1908 .filter(|a| {
1911 parse_timestamp(&a.transaction_time)
1912 .is_none_or(|time| (start..=end).contains(&time))
1913 })
1914 .filter_map(convert_activity_to_trade)
1915 .collect();
1916
1917 Ok(TradesRead::new(trades, resume))
1918 }
1919}
1920
1921impl AlpacaClient {
1926 pub async fn open_order_with_intent(
1939 &self,
1940 request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
1941 intent: AlpacaPositionIntent,
1942 ) -> Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState> {
1943 self.open_order_inner(request, intent).await
1944 }
1945
1946 pub async fn open_bracket_order(
1979 &self,
1980 request: AlpacaBracketOrderRequest,
1981 ) -> AlpacaBracketOrderResult {
1982 let order_key = crate::order::OrderKey::new(
1983 ExchangeId::AlpacaBroker,
1984 request.instrument.clone(),
1985 request.strategy.clone(),
1986 request.cid.clone(),
1987 );
1988
1989 let tif_str = match request.time_in_force {
1991 TimeInForce::GoodUntilEndOfDay => "day",
1992 TimeInForce::GoodUntilCancelled { post_only } => {
1993 if post_only {
1994 return AlpacaBracketOrderResult {
1995 parent: Order {
1996 key: order_key,
1997 side: request.side,
1998 price: Some(request.entry_price),
1999 quantity: request.quantity,
2000 kind: OrderKind::Limit,
2001 time_in_force: request.time_in_force,
2002 state: OrderState::inactive(OrderError::Rejected(
2003 ApiError::OrderRejected(
2004 "Alpaca does not support post_only for bracket orders"
2005 .to_string(),
2006 ),
2007 )),
2008 },
2009 };
2010 }
2011 "gtc"
2012 }
2013 other => {
2014 return AlpacaBracketOrderResult {
2015 parent: Order {
2016 key: order_key,
2017 side: request.side,
2018 price: Some(request.entry_price),
2019 quantity: request.quantity,
2020 kind: OrderKind::Limit,
2021 time_in_force: request.time_in_force,
2022 state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
2023 format!(
2024 "Alpaca bracket orders only support Day or GTC time_in_force, got {:?}",
2025 other
2026 ),
2027 ))),
2028 },
2029 };
2030 }
2031 };
2032
2033 let price_ordering_ok = match request.side {
2036 Side::Buy => {
2037 request.stop_loss_price < request.entry_price
2038 && request.entry_price < request.take_profit_price
2039 }
2040 Side::Sell => {
2041 request.take_profit_price < request.entry_price
2042 && request.entry_price < request.stop_loss_price
2043 }
2044 };
2045 if !price_ordering_ok {
2046 return AlpacaBracketOrderResult {
2047 parent: Order {
2048 key: order_key,
2049 side: request.side,
2050 price: Some(request.entry_price),
2051 quantity: request.quantity,
2052 kind: OrderKind::Limit,
2053 time_in_force: request.time_in_force,
2054 state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
2055 format!(
2056 "Invalid bracket price ordering for {:?} side: entry={}, take_profit={}, stop_loss={}",
2057 request.side,
2058 request.entry_price,
2059 request.take_profit_price,
2060 request.stop_loss_price,
2061 ),
2062 ))),
2063 },
2064 };
2065 }
2066
2067 if let Some(sl_limit) = request.stop_loss_limit_price {
2071 let sl_limit_ok = match request.side {
2072 Side::Buy => sl_limit <= request.stop_loss_price,
2073 Side::Sell => sl_limit >= request.stop_loss_price,
2074 };
2075 if !sl_limit_ok {
2076 return AlpacaBracketOrderResult {
2077 parent: Order {
2078 key: order_key,
2079 side: request.side,
2080 price: Some(request.entry_price),
2081 quantity: request.quantity,
2082 kind: OrderKind::Limit,
2083 time_in_force: request.time_in_force,
2084 state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
2085 format!(
2086 "Invalid stop-loss limit price for {:?} bracket: \
2087 stop_loss_price={}, stop_loss_limit_price={}",
2088 request.side, request.stop_loss_price, sl_limit,
2089 ),
2090 ))),
2091 },
2092 };
2093 }
2094 }
2095
2096 let body = AlpacaOrderRequest {
2098 symbol: request.instrument.name().as_str(),
2099 qty: request.quantity.to_string(),
2100 side: map_side(request.side),
2101 order_type: "limit",
2102 time_in_force: tif_str,
2103 limit_price: Some(request.entry_price.to_string()),
2104 stop_price: None,
2105 trail_percent: None,
2106 trail_price: None,
2107 client_order_id: Some(request.cid.0.as_str()),
2108 position_intent: if is_options_or_equity_symbol(request.instrument.name().as_str()) {
2109 Some(map_position_intent(request.side, false))
2110 } else {
2111 None
2112 },
2113 order_class: Some("bracket"),
2114 take_profit: Some(TakeProfitParams {
2115 limit_price: request.take_profit_price.to_string(),
2116 }),
2117 stop_loss: Some(StopLossParams {
2118 stop_price: request.stop_loss_price.to_string(),
2119 limit_price: request.stop_loss_limit_price.map(|p| p.to_string()),
2120 }),
2121 };
2122
2123 let http = self.http.clone();
2124 let rl = &self.rate_limiter;
2125
2126 let result: Result<AlpacaOrderResponse, UnindexedClientError> =
2127 rest_with_retry(rl, || http.post(&self.orders_url).json(&body)).await;
2128
2129 match result {
2130 Ok(resp) => {
2131 let state = placed_order_state(&resp, &order_key.instrument, request.quantity);
2132 self.known_live.lock().placed(
2133 &order_key,
2134 request.quantity,
2135 OrderKind::Limit,
2136 &state,
2137 );
2138
2139 AlpacaBracketOrderResult {
2140 parent: Order {
2141 key: order_key,
2142 side: request.side,
2143 price: Some(request.entry_price),
2144 quantity: request.quantity,
2145 kind: OrderKind::Limit,
2146 time_in_force: request.time_in_force,
2147 state,
2148 },
2149 }
2150 }
2151 Err(e) => {
2152 let order_err = order_post_error(e);
2153 AlpacaBracketOrderResult {
2154 parent: Order {
2155 key: order_key,
2156 side: request.side,
2157 price: Some(request.entry_price),
2158 quantity: request.quantity,
2159 kind: OrderKind::Limit,
2160 time_in_force: request.time_in_force,
2161 state: OrderState::inactive(order_err),
2162 },
2163 }
2164 }
2165 }
2166 }
2167
2168 async fn open_order_inner(
2169 &self,
2170 request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
2171 intent: AlpacaPositionIntent,
2172 ) -> Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState> {
2173 let instrument = request.key.instrument.clone();
2174 let side = request.state.side;
2175 let price = request.state.price;
2176 let quantity = request.state.quantity;
2177 let kind = request.state.kind;
2178 let time_in_force = request.state.time_in_force;
2179 let cid = request.key.cid.clone();
2180
2181 let order_key = crate::order::OrderKey::new(
2182 ExchangeId::AlpacaBroker,
2183 instrument.clone(),
2184 request.key.strategy.clone(),
2185 cid.clone(),
2186 );
2187
2188 let tif_str = match map_time_in_force(time_in_force) {
2190 Ok(s) => s,
2191 Err(msg) => {
2192 return Order {
2193 key: order_key,
2194 side,
2195 price,
2196 quantity,
2197 kind,
2198 time_in_force,
2199 state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
2200 msg.to_string(),
2201 ))),
2202 };
2203 }
2204 };
2205
2206 let order_type_str = match map_order_kind(kind) {
2208 Some(s) => s,
2209 None => {
2210 return Order {
2211 key: order_key,
2212 side,
2213 price,
2214 quantity,
2215 kind,
2216 time_in_force,
2217 state: OrderState::inactive(OrderError::UnsupportedOrderType(format!(
2218 "Alpaca connector does not yet support OrderKind::{kind:?}"
2219 ))),
2220 };
2221 }
2222 };
2223
2224 if let OrderKind::TrailingStop {
2226 offset_type: TrailingOffsetType::BasisPoints,
2227 ..
2228 } = kind
2229 {
2230 return Order {
2231 key: order_key,
2232 side,
2233 price,
2234 quantity,
2235 kind,
2236 time_in_force,
2237 state: OrderState::inactive(OrderError::UnsupportedOrderType(
2238 "Alpaca does not support TrailingOffsetType::BasisPoints; \
2239 use Percentage or Absolute"
2240 .to_string(),
2241 )),
2242 };
2243 }
2244
2245 if matches!(kind, OrderKind::StopLimit { .. }) && price.is_none() {
2248 return Order {
2249 key: order_key,
2250 side,
2251 price,
2252 quantity,
2253 kind,
2254 time_in_force,
2255 state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
2256 "StopLimit order requires Order.price (the limit price) to be set".to_string(),
2257 ))),
2258 };
2259 }
2260
2261 let (stop_price, trail_percent, trail_price) = match kind {
2263 OrderKind::Stop { trigger_price } | OrderKind::StopLimit { trigger_price } => {
2264 (Some(trigger_price.to_string()), None, None)
2265 }
2266 OrderKind::TrailingStop {
2267 offset,
2268 offset_type,
2269 } => match offset_type {
2270 TrailingOffsetType::Percentage => (None, Some(offset.to_string()), None),
2271 TrailingOffsetType::Absolute => (None, None, Some(offset.to_string())),
2272 TrailingOffsetType::BasisPoints => unreachable!("validated above"),
2273 },
2274 OrderKind::Market
2276 | OrderKind::Limit
2277 | OrderKind::TakeProfit { .. }
2278 | OrderKind::TakeProfitLimit { .. }
2279 | OrderKind::TrailingStopLimit { .. } => (None, None, None),
2280 };
2281
2282 let body = AlpacaOrderRequest {
2283 symbol: instrument.name().as_str(),
2284 qty: quantity.to_string(),
2285 side: map_side(side),
2286 order_type: order_type_str,
2287 time_in_force: tif_str,
2288 limit_price: match kind {
2289 OrderKind::Limit | OrderKind::StopLimit { .. } => price.map(|p| p.to_string()),
2290 _ => None,
2291 },
2292 stop_price,
2293 trail_percent,
2294 trail_price,
2295 client_order_id: Some(cid.0.as_str()),
2296 position_intent: if is_options_or_equity_symbol(instrument.name().as_str()) {
2302 Some(intent)
2303 } else {
2304 None
2305 },
2306 order_class: None,
2308 take_profit: None,
2309 stop_loss: None,
2310 };
2311
2312 let http = self.http.clone();
2313 let rl = &self.rate_limiter;
2314
2315 let result: Result<AlpacaOrderResponse, UnindexedClientError> =
2316 rest_with_retry(rl, || http.post(&self.orders_url).json(&body)).await;
2317
2318 match result {
2319 Ok(resp) => {
2320 let state = placed_order_state(&resp, &order_key.instrument, quantity);
2321 self.known_live
2322 .lock()
2323 .placed(&order_key, quantity, kind, &state);
2324
2325 Order {
2326 key: order_key,
2327 side,
2328 price,
2329 quantity,
2330 kind,
2331 time_in_force,
2332 state,
2333 }
2334 }
2335 Err(e) => {
2336 let order_err = order_post_error(e);
2337 Order {
2338 key: order_key,
2339 side,
2340 price,
2341 quantity,
2342 kind,
2343 time_in_force,
2344 state: OrderState::inactive(order_err),
2345 }
2346 }
2347 }
2348 }
2349}
2350
2351const MAX_OPEN_ORDERS: usize = 500;
2359
2360async fn fetch_raw_open_orders(
2368 http: &reqwest::Client,
2369 rate_limiter: &RateLimitTracker,
2370 base: &str,
2371 instruments: &[InstrumentNameExchange],
2372) -> Result<Vec<AlpacaOrderResponse>, UnindexedClientError> {
2373 let orders: Vec<AlpacaOrderResponse> = if instruments.is_empty() {
2374 rest_with_retry(rate_limiter, || {
2375 http.get(format!("{base}/v2/orders"))
2376 .query(&[("status", "open"), ("limit", "500")])
2377 })
2378 .await?
2379 } else {
2380 let symbols = instruments.iter().map(|i| i.name().as_str()).join(",");
2381 rest_with_retry(rate_limiter, || {
2382 http.get(format!("{base}/v2/orders")).query(&[
2383 ("status", "open"),
2384 ("limit", "500"),
2385 ("symbols", &symbols),
2386 ])
2387 })
2388 .await?
2389 };
2390 if orders.len() == MAX_OPEN_ORDERS {
2391 warn!(
2392 limit = MAX_OPEN_ORDERS,
2393 "Alpaca fetch_raw_open_orders: received exactly {MAX_OPEN_ORDERS} results — \
2394 response is likely truncated"
2395 );
2396 return Err(UnindexedClientError::TruncatedSnapshot {
2397 limit: MAX_OPEN_ORDERS,
2398 });
2399 }
2400 Ok(orders)
2401}
2402
2403const MAX_ACTIVITY_PAGES: usize = 50;
2412
2413struct ActivityPage {
2415 activities: Vec<AlpacaActivity>,
2417 resume: Option<DateTime<Utc>>,
2420}
2421
2422fn floor_millis(time: DateTime<Utc>) -> DateTime<Utc> {
2424 DateTime::from_timestamp_millis(time.timestamp_millis()).unwrap_or(time)
2425}
2426
2427const PAGE_SIZE_STR: &str = "100"; const _: () = assert!(
2457 ALPACA_MAX_ACTIVITIES == 100,
2458 "PAGE_SIZE_STR must be updated to match ALPACA_MAX_ACTIVITIES",
2459);
2460
2461async fn paginate_activities(
2462 http: &reqwest::Client,
2463 rate_limiter: &RateLimitTracker,
2464 base: &str,
2465 start: DateTime<Utc>,
2466 end: Option<DateTime<Utc>>,
2467) -> Result<ActivityPage, UnindexedClientError> {
2468 let after =
2471 floor_millis(start.max(DateTime::UNIX_EPOCH)).to_rfc3339_opts(SecondsFormat::Millis, true);
2472 let until = end.filter(|end| *end < Utc::now()).map(|end| {
2473 (floor_millis(end) + TimeDelta::milliseconds(1))
2474 .to_rfc3339_opts(SecondsFormat::Millis, true)
2475 });
2476 let mut all = Vec::with_capacity(ALPACA_MAX_ACTIVITIES);
2477 let mut page_token: Option<String> = None;
2478 let mut pages = 0usize;
2479 let mut truncated = false;
2480
2481 loop {
2482 if pages >= MAX_ACTIVITY_PAGES {
2483 truncated = true;
2484 break;
2485 }
2486 pages += 1;
2487 let page_token_ref = page_token.as_deref();
2489 let activities: Vec<AlpacaActivity> = rest_with_retry(rate_limiter, || {
2490 let mut req = http.get(format!("{base}/v2/account/activities")).query(&[
2491 ("activity_type", "FILL"),
2492 ("after", after.as_str()),
2493 ("page_size", PAGE_SIZE_STR),
2494 ("direction", "asc"),
2495 ]);
2496 if let Some(until) = until.as_deref() {
2497 req = req.query(&[("until", until)]);
2498 }
2499 if let Some(token) = page_token_ref {
2500 req = req.query(&[("page_token", token)]);
2501 }
2502 req
2503 })
2504 .await?;
2505
2506 let page_len = activities.len();
2507 let page_token_candidate = activities.last().map(|a| a.id.clone());
2517 all.extend(activities);
2518
2519 if page_len < ALPACA_MAX_ACTIVITIES {
2520 break;
2521 }
2522 match page_token_candidate {
2523 Some(token) if !token.is_empty() => {
2524 debug!("Alpaca paginate_activities: fetching next page ({page_len} results)");
2525 page_token = Some(token);
2526 }
2527 _ => break,
2531 }
2532 }
2533
2534 let resume = truncated.then(|| {
2537 floor_millis(
2538 all.iter()
2539 .rev()
2540 .find_map(|activity| parse_timestamp(&activity.transaction_time))
2541 .unwrap_or(start),
2542 )
2543 });
2544 Ok(ActivityPage {
2545 activities: all,
2546 resume,
2547 })
2548}
2549
2550#[allow(clippy::cognitive_complexity)] async fn connection_manager(
2566 tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
2567 dedup: SharedDedupCache,
2568 config: Arc<AlpacaConfig>,
2569 http: reqwest::Client,
2570 rate_limiter: Arc<RateLimitTracker>,
2571 known: SharedKnownLiveOrders,
2572 instruments: Vec<InstrumentNameExchange>,
2573 initial_ws: Option<WebSocket>,
2574) {
2575 let mut backoff = ExponentialBackoff::new();
2576 let mut disconnect_time: Option<DateTime<Utc>> = None;
2577 let mut unchecked = UncheckedOrders::default();
2580 let mut current_ws = initial_ws;
2581
2582 'outer: loop {
2583 let mut ws = match current_ws.take() {
2585 Some(ws) => ws,
2586 None => match connect_and_subscribe(&config).await {
2587 Ok(ws) => ws,
2588 Err(e) => {
2589 error!(%e, "Alpaca WS connect/subscribe failed");
2590 if !backoff.wait().await {
2591 error!("Alpaca max reconnect attempts exhausted");
2592 emit_stream_terminated(
2594 &tx,
2595 ExchangeId::AlpacaBroker,
2596 StreamTerminationReason::ReconnectBudgetExhausted {
2597 attempts: backoff.attempts(),
2598 last_error: e.to_string(),
2599 },
2600 );
2601 break;
2602 }
2603 continue;
2604 }
2605 },
2606 };
2607 info!("Alpaca account_stream connected and subscribed");
2608 backoff.reset();
2619
2620 if let Some(dt) = disconnect_time.take() {
2625 recover_fills_or_report(
2626 &http,
2627 &rate_limiter,
2628 &instruments,
2629 config.rest_base_url(),
2630 dt,
2631 Duration::from_secs(FILL_RECOVERY_TIMEOUT_SECS),
2632 &tx,
2633 &dedup,
2634 &known,
2635 )
2636 .await;
2637 let held = {
2639 let known = known.lock();
2640 if instruments.is_empty() {
2641 known.instruments()
2642 } else {
2643 known.instruments_among(&instruments)
2644 }
2645 };
2646 unchecked.open(held);
2647 }
2648
2649 let mut last_message_time = Utc::now();
2662 let heartbeat = tokio::time::sleep(Duration::from_secs(HEARTBEAT_TIMEOUT_SECS));
2663 tokio::pin!(heartbeat);
2664
2665 let order_checks =
2670 run_order_checks(&http, &rate_limiter, &config, &known, &mut unchecked, &tx);
2671 tokio::pin!(order_checks);
2672
2673 macro_rules! reset_heartbeat {
2678 () => {
2679 heartbeat.as_mut().reset(
2680 tokio::time::Instant::now() + Duration::from_secs(HEARTBEAT_TIMEOUT_SECS),
2681 );
2682 last_message_time = Utc::now();
2683 };
2684 }
2685 enum DisconnectReason {
2688 ServerClose,
2689 Error(String),
2690 StreamEnded,
2691 HeartbeatTimeout,
2692 }
2693 let reason = loop {
2694 tokio::select! {
2695 msg = ws.next() => {
2696 match msg {
2697 Some(Ok(WsMessage::Ping(_))) => {
2698 reset_heartbeat!();
2701 }
2702 Some(Ok(WsMessage::Text(text))) => {
2703 process_ws_text(text.as_str(), &tx, &dedup, &known, &mut backoff);
2704 reset_heartbeat!();
2705 }
2706 Some(Ok(WsMessage::Binary(bytes))) => {
2707 match std::str::from_utf8(&bytes) {
2709 Ok(text) => {
2710 process_ws_text(text, &tx, &dedup, &known, &mut backoff);
2711 reset_heartbeat!();
2715 }
2716 Err(e) => warn!(%e, "Alpaca WS binary frame: not valid UTF-8"),
2717 }
2718 }
2719 Some(Ok(WsMessage::Close(frame))) => {
2720 warn!(frame = ?frame, "Alpaca WS closed by server");
2721 break DisconnectReason::ServerClose;
2722 }
2723 Some(Ok(_)) => {} Some(Err(e)) => {
2725 warn!(%e, "Alpaca WS error, reconnecting");
2726 break DisconnectReason::Error(e.to_string());
2727 }
2728 None => {
2729 warn!("Alpaca WS stream ended, reconnecting");
2730 break DisconnectReason::StreamEnded;
2731 }
2732 }
2733 }
2734 () = &mut order_checks => {}
2735 _ = &mut heartbeat => {
2736 warn!(
2737 timeout_secs = HEARTBEAT_TIMEOUT_SECS,
2738 "Alpaca heartbeat timeout, reconnecting"
2739 );
2740 break DisconnectReason::HeartbeatTimeout;
2743 }
2744 _ = tx.closed() => {
2745 debug!("Alpaca account_stream consumer dropped, terminating");
2746 let _ = tokio::time::timeout(
2747 Duration::from_secs(WS_CLOSE_TIMEOUT_SECS),
2748 ws.close(None),
2749 ).await;
2750 break 'outer;
2753 }
2754 }
2755 };
2756
2757 disconnect_time =
2762 Some(last_message_time - chrono::Duration::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS));
2763
2764 let _ =
2766 tokio::time::timeout(Duration::from_secs(WS_CLOSE_TIMEOUT_SECS), ws.close(None)).await;
2767
2768 if tx.is_closed() {
2769 break;
2772 }
2773 if !backoff.wait().await {
2774 error!("Alpaca max reconnect attempts exhausted, stream terminating");
2775 let last_error = match reason {
2778 DisconnectReason::HeartbeatTimeout => {
2779 format!("heartbeat timeout ({HEARTBEAT_TIMEOUT_SECS}s)")
2780 }
2781 DisconnectReason::ServerClose => "WebSocket closed by server".to_string(),
2782 DisconnectReason::Error(e) => e,
2783 DisconnectReason::StreamEnded => "WebSocket stream ended".to_string(),
2784 };
2785 emit_stream_terminated(
2786 &tx,
2787 ExchangeId::AlpacaBroker,
2788 StreamTerminationReason::ReconnectBudgetExhausted {
2789 attempts: backoff.attempts(),
2790 last_error,
2791 },
2792 );
2793 break;
2794 }
2795 }
2796}
2797
2798#[derive(Debug)]
2803enum HandshakeError {
2804 Transport(String),
2807 Auth(String),
2810}
2811
2812async fn connect_and_subscribe(config: &AlpacaConfig) -> Result<WebSocket, UnindexedClientError> {
2816 let url = config.ws_url();
2817 debug!(%url, "Alpaca: connecting to WebSocket");
2818
2819 let mut ws = rustrade_integration::protocol::websocket::connect(url)
2820 .await
2821 .map_err(|e| {
2822 UnindexedClientError::Connectivity(ConnectivityError::Socket(format!(
2823 "WS connect: {e}"
2824 )))
2825 })?;
2826
2827 let result = tokio::time::timeout(
2829 Duration::from_secs(WS_HANDSHAKE_TIMEOUT_SECS),
2830 ws_handshake(&mut ws, config),
2831 )
2832 .await;
2833
2834 match result {
2835 Ok(Ok(())) => Ok(ws),
2836 Ok(Err(HandshakeError::Transport(e))) => {
2837 let _ = ws.close(None).await;
2838 Err(UnindexedClientError::Connectivity(
2839 ConnectivityError::Socket(e),
2840 ))
2841 }
2842 Ok(Err(HandshakeError::Auth(e))) => {
2843 let _ = ws.close(None).await;
2844 Err(UnindexedClientError::Api(ApiError::Unauthenticated(e)))
2845 }
2846 Err(_) => {
2847 let _ = ws.close(None).await;
2848 Err(UnindexedClientError::Connectivity(
2849 ConnectivityError::Timeout,
2850 ))
2851 }
2852 }
2853}
2854
2855async fn ws_handshake(ws: &mut WebSocket, config: &AlpacaConfig) -> Result<(), HandshakeError> {
2863 let auth = serde_json::json!({
2865 "action": "auth",
2866 "key": config.api_key(),
2867 "secret": config.secret_key,
2870 })
2871 .to_string();
2872 ws.send(WsMessage::Text(auth.into()))
2873 .await
2874 .map_err(|e| HandshakeError::Transport(format!("WS auth send: {e}")))?;
2875
2876 loop {
2878 match ws.next().await {
2879 Some(Ok(WsMessage::Text(text))) => {
2880 if let Some(result) = check_auth_response(text.as_str()) {
2881 result?;
2882 break;
2883 }
2884 }
2885 Some(Ok(WsMessage::Binary(bytes))) => {
2886 if let Ok(text) = std::str::from_utf8(&bytes)
2887 && let Some(result) = check_auth_response(text)
2888 {
2889 result?;
2890 break;
2891 }
2892 }
2893 Some(Err(e)) => {
2894 return Err(HandshakeError::Transport(format!(
2895 "WS error during auth: {e}"
2896 )));
2897 }
2898 None => {
2899 return Err(HandshakeError::Transport(
2900 "WS closed before auth response".into(),
2901 ));
2902 }
2903 _ => {} }
2905 }
2906
2907 let sub = serde_json::json!({
2909 "action": "listen",
2910 "data": { "streams": ["trade_updates"] }
2911 })
2912 .to_string();
2913 ws.send(WsMessage::Text(sub.into()))
2914 .await
2915 .map_err(|e| HandshakeError::Transport(format!("WS subscribe send: {e}")))?;
2916
2917 loop {
2919 match ws.next().await {
2920 Some(Ok(WsMessage::Text(text))) => {
2921 if check_listen_ack(text.as_str()) {
2922 break;
2923 }
2924 if let Ok(msg) = serde_json::from_str::<AlpacaStreamMessage<'_>>(text.as_str()) {
2931 if msg.stream == "trade_updates" {
2932 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");
2933 } else {
2934 trace!(stream = %msg.stream, "WS message dropped during listen-ack handshake");
2935 }
2936 } else {
2937 trace!(
2938 bytes = text.len(),
2939 "WS non-stream message dropped during listen-ack handshake"
2940 );
2941 }
2942 }
2943 Some(Ok(WsMessage::Binary(bytes))) => {
2944 if let Ok(text) = std::str::from_utf8(&bytes)
2945 && check_listen_ack(text)
2946 {
2947 break;
2948 }
2949 trace!(
2950 bytes = bytes.len(),
2951 "WS binary message dropped during listen-ack handshake"
2952 );
2953 }
2954 Some(Err(e)) => {
2955 return Err(HandshakeError::Transport(format!(
2956 "WS error during subscribe: {e}"
2957 )));
2958 }
2959 None => {
2960 return Err(HandshakeError::Transport(
2961 "WS closed before subscribe ack".into(),
2962 ));
2963 }
2964 _ => {}
2965 }
2966 }
2967
2968 info!("Alpaca WS authenticated and subscribed to trade_updates");
2969 Ok(())
2970}
2971
2972fn check_auth_response(text: &str) -> Option<Result<(), HandshakeError>> {
2978 let msg = serde_json::from_str::<AlpacaStreamMessage<'_>>(text).ok()?;
2979 if msg.stream != "authorization" {
2980 return None;
2981 }
2982 #[derive(Deserialize)]
2983 struct AuthData<'a> {
2984 status: &'a str,
2985 }
2986 let data = serde_json::from_str::<AuthData<'_>>(msg.data.get()).ok()?;
2987 if data.status == "authorized" {
2988 Some(Ok(()))
2989 } else {
2990 Some(Err(HandshakeError::Auth(format!(
2991 "Alpaca WS auth failed: status={}",
2992 data.status
2993 ))))
2994 }
2995}
2996
2997fn check_listen_ack(text: &str) -> bool {
3001 let Ok(msg) = serde_json::from_str::<AlpacaStreamMessage<'_>>(text) else {
3002 return false;
3003 };
3004 if msg.stream != "listening" {
3005 return false;
3006 }
3007 #[derive(Deserialize)]
3008 struct ListenData<'a> {
3009 #[serde(borrow)]
3010 streams: Vec<&'a str>,
3011 }
3012 let Ok(data) = serde_json::from_str::<ListenData<'_>>(msg.data.get()) else {
3013 return false;
3014 };
3015 data.streams.contains(&"trade_updates")
3016}
3017
3018fn process_ws_text(
3028 text: &str,
3029 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
3030 dedup: &SharedDedupCache,
3031 known: &SharedKnownLiveOrders,
3032 backoff: &mut ExponentialBackoff,
3033) {
3034 let msg: AlpacaStreamMessage<'_> = match serde_json::from_str(text) {
3035 Ok(m) => m,
3036 Err(e) => {
3037 trace!(
3038 %e,
3039 raw = ?&text[..text.len().min(200)],
3040 "Alpaca WS: skipped non-JSON message"
3041 );
3042 return;
3043 }
3044 };
3045
3046 backoff.reset();
3049
3050 match msg.stream.as_str() {
3051 "trade_updates" => {
3052 let update: AlpacaTradeUpdate<'_> = match serde_json::from_str(msg.data.get()) {
3053 Ok(u) => u,
3054 Err(e) => {
3055 warn!(
3060 %e,
3061 raw = ?&msg.data.get()[..msg.data.get().len().min(200)],
3062 "Alpaca WS trade_updates: failed to deserialize event — event dropped"
3063 );
3064 return;
3065 }
3066 };
3067
3068 if is_fill_event(&update)
3072 && is_duplicate_fill(dedup, ws_execution_id(&update), early_dedup_key(&update))
3073 {
3074 trace!("Alpaca WS: skipping duplicate fill event (early check)");
3075 return;
3076 }
3077
3078 if is_trade_amendment(&update) {
3081 warn!(
3082 event = %update.event,
3083 raw = %msg.data.get(),
3084 "Alpaca WS: a fill reported earlier was amended — reporting it as TradeAmended"
3085 );
3086 }
3087
3088 for event in convert_trade_update(update).into_iter().flatten() {
3089 if let AccountEventKind::TradeAmended(amendment) = &event.kind
3093 && let TradeAmendmentKind::Corrected { replacement } = &amendment.kind
3094 {
3095 record_execution(dedup, &replacement.id.0);
3096 }
3097 let known = KnownLiveOrders::observes(&event.kind).then(|| {
3098 let mut known = known.lock();
3099 known.observe(&event);
3100 known
3101 });
3102 let _ = tx.send(event);
3105 drop(known);
3106 }
3107 }
3108 "authorization" | "listening" => {
3109 trace!(stream = %msg.stream, "Alpaca WS: auth/listen ack received during stream");
3112 }
3113 other => {
3114 trace!(%other, "Alpaca WS: ignoring unknown stream type");
3115 }
3116 }
3117}
3118
3119fn fill_dedup_key(order_id: &str, cumulative: Decimal) -> SmolStr {
3130 format_smolstr!("{}:{}", order_id, cumulative.normalize())
3131}
3132
3133fn is_trade_amendment(update: &AlpacaTradeUpdate<'_>) -> bool {
3135 matches!(update.event.as_str(), "trade_bust" | "trade_correct")
3136}
3137
3138#[inline]
3142fn is_fill_event(update: &AlpacaTradeUpdate<'_>) -> bool {
3143 matches!(update.event.as_str(), "fill" | "partial_fill")
3144}
3145
3146fn ws_execution_id<'a>(update: &AlpacaTradeUpdate<'a>) -> Option<&'a str> {
3148 update.execution_id.filter(|id| !id.is_empty())
3149}
3150
3151fn early_dedup_key(update: &AlpacaTradeUpdate<'_>) -> SmolStr {
3155 let filled_qty = update.order.filled_qty.unwrap_or("0");
3156 let qty = Decimal::from_str(filled_qty).unwrap_or(Decimal::ZERO);
3157 fill_dedup_key(&update.order.id, qty)
3158}
3159
3160fn activity_execution_id(activity_id: &str) -> &str {
3164 activity_execution_id_part(activity_id).unwrap_or(activity_id)
3165}
3166
3167fn activity_execution_id_part(activity_id: &str) -> Option<&str> {
3171 activity_id
3172 .split_once("::")
3173 .map(|(_, execution_id)| execution_id)
3174 .filter(|execution_id| !execution_id.is_empty())
3175}
3176
3177#[derive(Debug, PartialEq)]
3183struct UnreadFills {
3184 start: DateTime<Utc>,
3185 reason: FillRecoveryFailure,
3186}
3187
3188async fn recover_fills_or_report(
3192 http: &reqwest::Client,
3193 rate_limiter: &RateLimitTracker,
3194 instruments: &[InstrumentNameExchange],
3195 base: &str,
3196 disconnect: DateTime<Utc>,
3197 timeout: Duration,
3198 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
3199 dedup: &SharedDedupCache,
3200 known: &SharedKnownLiveOrders,
3201) {
3202 let recovery_start = Utc::now();
3204 let outcome = match tokio::time::timeout(
3205 timeout,
3206 recover_fills(
3207 http,
3208 rate_limiter,
3209 instruments,
3210 base,
3211 disconnect,
3212 tx,
3213 dedup,
3214 known,
3215 ),
3216 )
3217 .await
3218 {
3219 Ok(outcome) => outcome,
3220 Err(_) => {
3224 warn!(
3225 timeout_secs = timeout.as_secs(),
3226 instruments = ?instruments
3227 .iter()
3228 .map(|instrument| instrument.name().as_str())
3229 .collect::<Vec<_>>(),
3230 "Alpaca fill recovery timed out — fills since the disconnect may be missing for \
3231 any of the instruments (every instrument when the list is empty)"
3232 );
3233 Err(UnreadFills {
3234 start: disconnect,
3235 reason: FillRecoveryFailure::TimedOut {
3236 timeout_secs: timeout.as_secs(),
3237 },
3238 })
3239 }
3240 };
3241 if let Err(unread) = outcome
3242 && tx
3243 .send(fill_recovery_gave_up(instruments, unread, recovery_start))
3244 .is_err()
3245 {
3246 debug!("Alpaca fill recovery: consumer dropped before the give-up was sent");
3247 }
3248}
3249
3250fn fill_recovery_gave_up(
3259 instruments: &[InstrumentNameExchange],
3260 unread: UnreadFills,
3261 end: DateTime<Utc>,
3262) -> UnindexedAccountEvent {
3263 let UnreadFills { start, reason } = unread;
3264 let scope = if instruments.is_empty() {
3266 FillRecoveryScope::AllInstruments
3267 } else {
3268 FillRecoveryScope::Instruments(instruments.to_vec())
3269 };
3270 UnindexedAccountEvent::new(
3271 ExchangeId::AlpacaBroker,
3272 AccountEventKind::FillRecoveryGaveUp(FillRecoveryGap::new(
3273 scope,
3274 start,
3275 end.max(start),
3276 1,
3277 reason,
3278 )),
3279 )
3280}
3281
3282async fn recover_fills(
3291 http: &reqwest::Client,
3292 rate_limiter: &RateLimitTracker,
3293 instruments: &[InstrumentNameExchange],
3294 base: &str,
3295 after: DateTime<Utc>,
3296 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
3297 dedup: &SharedDedupCache,
3298 known: &SharedKnownLiveOrders,
3299) -> Result<(), UnreadFills> {
3300 info!(%after, instruments = instruments.len(), "Alpaca recovering fills after reconnect");
3303 let instrument_set: fnv::FnvHashSet<&str> = if instruments.is_empty() {
3304 fnv::FnvHashSet::default()
3305 } else {
3306 instruments.iter().map(|i| i.name().as_str()).collect()
3307 };
3308
3309 let page = match paginate_activities(http, rate_limiter, base, after, None).await {
3310 Ok(p) => p,
3311 Err(e) => {
3312 error!(%e, "Alpaca fill recovery: REST request failed");
3313 return Err(UnreadFills {
3314 start: after,
3315 reason: FillRecoveryFailure::Request(e.to_string()),
3316 });
3317 }
3318 };
3319
3320 let activities = page.activities;
3321
3322 let unread = page.resume.map(|start| {
3326 error!(
3327 max_pages = MAX_ACTIVITY_PAGES,
3328 fills_read = activities.len(),
3329 %start,
3330 "Alpaca fill recovery: max page limit reached, truncating — fills from this time on \
3331 are not delivered; read them with fetch_trades"
3332 );
3333 UnreadFills {
3334 start,
3335 reason: FillRecoveryFailure::Truncated {
3336 fills_read: activities.len(),
3337 },
3338 }
3339 });
3340
3341 let mut recovered = 0u32;
3342 let mut duplicates = 0u32;
3343
3344 let mut cumulative_qty: FnvHashMap<&str, Decimal> = FnvHashMap::default();
3361
3362 for activity in &activities {
3363 if !instrument_set.is_empty() && !instrument_set.contains(activity.symbol.as_str()) {
3364 continue;
3368 }
3369
3370 let exec_qty = Decimal::from_str(&activity.qty).unwrap_or(Decimal::ZERO);
3385 let cum = cumulative_qty
3386 .entry(activity.order_id.as_str())
3387 .or_default();
3388 *cum += exec_qty;
3389 let cumulative = *cum;
3390
3391 let trade = match convert_activity_to_trade(activity) {
3392 Some(t) => t,
3393 None => {
3394 warn!(id = %activity.id, symbol = %activity.symbol, "Alpaca: skipping activity with unparseable fields");
3395 continue; }
3397 };
3398
3399 let key = fill_dedup_key(
3409 &activity.order_id,
3410 trade.order_filled_quantity.unwrap_or(cumulative),
3411 );
3412 if is_duplicate_fill(dedup, activity_execution_id_part(&activity.id), key) {
3413 duplicates += 1;
3414 continue;
3415 }
3416
3417 let event =
3418 UnindexedAccountEvent::new(ExchangeId::AlpacaBroker, AccountEventKind::Trade(trade));
3419 let mut held = known.lock();
3420 held.observe(&event);
3421 let sent = tx.send(event);
3422 drop(held);
3423 if sent.is_err() {
3424 debug!("Alpaca fill recovery: consumer dropped during recovery");
3425 return Ok(());
3426 }
3427 recovered += 1;
3428 }
3429
3430 info!(recovered, duplicates, "Alpaca fill recovery complete");
3431 unread.map_or(Ok(()), Err)
3432}
3433
3434async fn fetch_order_lookup(
3443 http: reqwest::Client,
3444 rate_limiter: Arc<RateLimitTracker>,
3445 config: Arc<AlpacaConfig>,
3446 key: UnindexedOrderKey,
3447) -> Result<OrderLookup, UnindexedClientError> {
3448 let url = format!("{}/v2/orders:by_client_order_id", config.rest_base_url());
3449 let cid = key.cid.0.as_str();
3450 let found: Option<AlpacaOrderResponse> = rest_lookup_with_retry(&rate_limiter, || {
3451 http.get(&url).query(&[("client_order_id", cid)])
3452 })
3453 .await?;
3454 let Some(order) = found else {
3455 debug!(instrument = %key.instrument, cid = %key.cid, "Alpaca does not know this order");
3456 return Ok(OrderLookup::Unknown);
3457 };
3458 if !order
3459 .symbol
3460 .eq_ignore_ascii_case(key.instrument.name().as_str())
3461 {
3462 debug!(
3463 instrument = %key.instrument,
3464 cid = %key.cid,
3465 symbol = %order.symbol,
3466 "Alpaca knows this client order id on another symbol"
3467 );
3468 return Ok(OrderLookup::Unknown);
3469 }
3470 Ok(
3471 convert_ended_order(&order, &key).map_or(OrderLookup::NotEnded, |order| {
3472 OrderLookup::Ended(Box::new(order))
3473 }),
3474 )
3475}
3476
3477async fn listed_open_cids(
3480 http: reqwest::Client,
3481 rate_limiter: Arc<RateLimitTracker>,
3482 config: Arc<AlpacaConfig>,
3483 instruments: Vec<InstrumentNameExchange>,
3484) -> Result<FnvHashSet<ClientOrderId>, UnindexedClientError> {
3485 let orders =
3486 fetch_raw_open_orders(&http, &rate_limiter, config.rest_base_url(), &instruments).await?;
3487 Ok(orders
3488 .iter()
3489 .map(|order| alpaca_cid(order.client_order_id.as_deref(), &order.id))
3490 .collect())
3491}
3492
3493async fn recover_alpaca_ended_orders(
3497 http: &reqwest::Client,
3498 rate_limiter: &Arc<RateLimitTracker>,
3499 config: &Arc<AlpacaConfig>,
3500 known: &SharedKnownLiveOrders,
3501 unchecked: &mut UncheckedOrders,
3502 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
3503) {
3504 recover_ended_orders(
3505 ExchangeId::AlpacaBroker,
3506 known,
3507 unchecked,
3508 &NoPendingFills,
3509 tx,
3510 OpenListing::Batched,
3511 |instruments| {
3512 listed_open_cids(
3513 http.clone(),
3514 rate_limiter.clone(),
3515 config.clone(),
3516 instruments,
3517 )
3518 },
3519 |key| fetch_order_lookup(http.clone(), rate_limiter.clone(), config.clone(), key),
3520 )
3521 .await;
3522}
3523
3524async fn run_order_checks(
3529 http: &reqwest::Client,
3530 rate_limiter: &Arc<RateLimitTracker>,
3531 config: &Arc<AlpacaConfig>,
3532 known: &SharedKnownLiveOrders,
3533 unchecked: &mut UncheckedOrders,
3534 tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
3535) {
3536 loop {
3537 if tx.is_closed() {
3539 return std::future::pending().await;
3540 }
3541 match unchecked.next_due(&NoPendingFills) {
3542 Some(due) => tokio::time::sleep_until(due).await,
3543 None => return std::future::pending().await,
3544 }
3545 recover_alpaca_ended_orders(http, rate_limiter, config, known, unchecked, tx).await;
3546 }
3547}
3548
3549fn convert_ended_order(
3557 o: &AlpacaOrderResponse,
3558 key: &UnindexedOrderKey,
3559) -> Option<UnindexedInactiveOrder> {
3560 let instrument = &key.instrument;
3561 let Some(status) = o.status.as_deref() else {
3562 warn!(%instrument, cid = %key.cid, order_id = %o.id, "Alpaca order missing status");
3563 return None;
3564 };
3565 if order_status_is_live(status) {
3566 return None;
3567 }
3568 let Some(order) = convert_representable_open_order(o) else {
3569 warn!(%instrument, cid = %key.cid, order_id = %o.id, status, "Alpaca order cannot be represented, treating it as not ended");
3570 return None;
3571 };
3572 let filled_qty = Decimal::from_str(&o.filled_qty).ok();
3575 let Some(state) = ended_order_state(
3576 status,
3577 OrderId(SmolStr::new(&o.id)),
3578 order.state.time_exchange,
3579 order.quantity,
3580 filled_qty,
3581 o.filled_avg_price.as_deref(),
3582 ) else {
3583 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");
3584 return None;
3585 };
3586 let mut order = order.map_state(|_| state);
3587 order.key = key.clone();
3588 Some(order)
3589}
3590
3591fn order_status_is_live(status: &str) -> bool {
3597 matches!(
3598 status,
3599 "new"
3600 | "partially_filled"
3601 | "done_for_day"
3602 | "pending_cancel"
3603 | "pending_replace"
3604 | "accepted"
3605 | "pending_new"
3606 | "accepted_for_bidding"
3607 | "stopped"
3608 | "suspended"
3609 | "calculated"
3610 | "held"
3611 )
3612}
3613
3614fn ended_order_state<AssetKey, InstrumentKey>(
3624 status: &str,
3625 order_id: OrderId,
3626 time_exchange: DateTime<Utc>,
3627 quantity: Decimal,
3628 filled_qty: Option<Decimal>,
3629 filled_avg_price: Option<&str>,
3630) -> Option<InactiveOrderState<AssetKey, InstrumentKey>> {
3631 Some(match status {
3632 "filled" => InactiveOrderState::FullyFilled(Filled::new(
3634 order_id,
3635 time_exchange,
3636 filled_qty.unwrap_or(quantity),
3637 alpaca_avg_price(filled_avg_price),
3638 )),
3639 "canceled" | "replaced" => {
3640 InactiveOrderState::Cancelled(Cancelled::new(order_id, time_exchange, filled_qty))
3641 }
3642 "expired" => InactiveOrderState::Expired(Expired::new(order_id, time_exchange, filled_qty)),
3643 "rejected" => {
3644 InactiveOrderState::OpenFailed(OrderError::Rejected(ApiError::OrderRejected(format!(
3645 "Alpaca rejected order {order_id} after accepting it"
3646 ))))
3647 }
3648 _ => return None,
3649 })
3650}
3651
3652fn placed_order_state(
3662 resp: &AlpacaOrderResponse,
3663 instrument: &InstrumentNameExchange,
3664 quantity: Decimal,
3665) -> UnindexedOrderState {
3666 let order_id = OrderId(SmolStr::new(&resp.id));
3667 let time_exchange = order_state_time(resp);
3668 let filled_qty = alpaca_filled_qty(&resp.id, Some(&resp.filled_qty));
3669 let filled_avg_price = resp.filled_avg_price.as_deref();
3670 match resp.status.as_deref() {
3671 Some(status) => {
3672 if let Some(ended) = ended_order_state(
3673 status,
3674 order_id.clone(),
3675 time_exchange,
3676 quantity,
3677 filled_qty,
3678 filled_avg_price,
3679 ) {
3680 return OrderState::Inactive(ended);
3681 }
3682 if !order_status_is_live(status) {
3683 warn!(%instrument, %order_id, status, "Alpaca placed an order with a status this version does not know, reading it from what filled");
3684 }
3685 }
3686 None => {
3687 warn!(%instrument, %order_id, "Alpaca placed an order without a status, reading it from what filled");
3688 }
3689 }
3690 match filled_qty {
3691 Some(filled_qty) if filled_qty >= quantity => OrderState::fully_filled(Filled::new(
3692 order_id,
3693 time_exchange,
3694 filled_qty,
3695 alpaca_avg_price(filled_avg_price),
3696 )),
3697 _ => OrderState::active(Open::new(
3700 VenueOrderId::Assigned(order_id),
3701 time_exchange,
3702 filled_qty.unwrap_or(Decimal::ZERO),
3703 )),
3704 }
3705}
3706
3707fn alpaca_avg_price(filled_avg_price: Option<&str>) -> Option<Decimal> {
3710 filled_avg_price.and_then(|price| Decimal::from_str(price).ok())
3711}
3712
3713fn is_usd(asset: &AssetNameExchange) -> bool {
3720 asset.name().as_str().eq_ignore_ascii_case(USD)
3721}
3722
3723fn convert_account_to_balances(
3741 account: &AlpacaAccount,
3742 assets: &[AssetNameExchange],
3743) -> Vec<AssetBalance<AssetNameExchange>> {
3744 if !assets.is_empty() && !assets.iter().any(is_usd) {
3745 return Vec::new();
3746 }
3747
3748 vec![AssetBalance::new(
3749 AssetNameExchange::new(USD),
3750 Balance::new(
3751 account.cash,
3752 account.cash.min(account.non_marginable_buying_power),
3753 ),
3754 Utc::now(),
3755 )]
3756}
3757
3758fn convert_positions_to_balances(
3770 positions: &[AlpacaPosition],
3771 assets: &[AssetNameExchange],
3772) -> Vec<AssetBalance<AssetNameExchange>> {
3773 let now = Utc::now();
3774 positions
3775 .iter()
3776 .filter(|p| is_crypto_position(p))
3777 .filter_map(|p| {
3778 let base = crypto_position_asset(&p.symbol);
3779
3780 if !assets.is_empty()
3782 && !assets
3783 .iter()
3784 .any(|a| a.name().as_str().eq_ignore_ascii_case(base))
3785 {
3786 return None;
3787 }
3788
3789 let asset_name = AssetNameExchange::new(base);
3792 Some(AssetBalance::new(
3793 asset_name,
3794 Balance::new(p.qty, p.qty_available),
3795 now,
3796 ))
3797 })
3798 .collect()
3799}
3800
3801fn crypto_position_asset(symbol: &str) -> &str {
3809 let base = match symbol.split_once('/') {
3810 Some((base, _)) => Some(base),
3811 None => {
3812 let base_len = symbol.len().saturating_sub(USD.len());
3813 symbol
3814 .get(base_len..)
3815 .filter(|suffix| suffix.eq_ignore_ascii_case(USD))
3816 .and_then(|_| symbol.get(..base_len))
3817 }
3818 };
3819 match base {
3820 Some(base) if !base.is_empty() => base,
3821 _ => {
3822 warn!(
3823 %symbol,
3824 "Alpaca crypto position symbol is neither ASSETUSD nor BASE/QUOTE; using it whole"
3825 );
3826 symbol
3827 }
3828 }
3829}
3830
3831fn is_crypto_position(position: &AlpacaPosition) -> bool {
3834 position.asset_class.eq_ignore_ascii_case("crypto")
3835}
3836
3837fn convert_positions(
3859 positions: &[AlpacaPosition],
3860 now: DateTime<Utc>,
3861) -> Result<Vec<(&str, Position)>, UnindexedClientError> {
3862 positions
3863 .iter()
3864 .filter(|p| !is_crypto_position(p) && !p.qty.is_zero())
3865 .map(|p| {
3866 let size = p.qty.abs();
3867 let quantity = match p.side {
3868 AlpacaPositionSide::Long => size,
3869 AlpacaPositionSide::Short => -size,
3870 AlpacaPositionSide::Unknown => {
3871 return Err(UnindexedClientError::Internal(format!(
3872 "Alpaca position {} has a missing or unrecognised side",
3873 p.symbol
3874 )));
3875 }
3876 };
3877 let position = Position::new(
3878 quantity,
3879 p.avg_entry_price,
3880 p.unrealized_pl,
3881 None,
3882 None,
3883 None,
3884 now,
3885 );
3886 Ok((p.symbol.as_str(), position))
3887 })
3888 .collect()
3889}
3890
3891fn build_instrument_snapshots(
3912 orders: Vec<AlpacaOrderResponse>,
3913 positions: Vec<(&str, Position)>,
3914 instruments: &[InstrumentNameExchange],
3915) -> Vec<InstrumentAccountSnapshot<ExchangeId, AssetNameExchange, InstrumentNameExchange>> {
3916 struct Listing {
3918 orders: Vec<UnindexedOrderSnapshot>,
3919 complete: bool,
3920 position: Option<Position>,
3921 }
3922
3923 impl Listing {
3924 fn empty() -> Self {
3925 Self {
3926 orders: Vec::new(),
3927 complete: true,
3928 position: None,
3929 }
3930 }
3931 }
3932
3933 let mut by_symbol: IndexMap<SmolStr, Listing> = IndexMap::new();
3935
3936 for order in orders {
3939 let listing = by_symbol
3940 .entry(order.symbol.to_ascii_uppercase().into())
3941 .or_insert_with(Listing::empty);
3942 match convert_open_order(&order) {
3943 Some(converted) => listing.orders.push(converted.into()),
3944 None => listing.complete = false,
3945 }
3946 }
3947
3948 for (symbol, position) in positions {
3949 by_symbol
3950 .entry(symbol.to_ascii_uppercase().into())
3951 .or_insert_with(Listing::empty)
3952 .position = Some(position);
3953 }
3954
3955 let snapshot = |instrument: InstrumentNameExchange, listing: Listing| {
3956 let position = if is_options_or_equity_symbol(instrument.name().as_str()) {
3959 listing
3960 .position
3961 .map_or(PositionReport::Flat, PositionReport::from_position)
3962 } else {
3963 PositionReport::Unreported
3964 };
3965 InstrumentAccountSnapshot::new(instrument, listing.orders, listing.complete, position, None)
3966 };
3967
3968 if instruments.is_empty() {
3970 by_symbol
3971 .into_iter()
3972 .map(|(sym, listing)| snapshot(InstrumentNameExchange::new(sym), listing))
3973 .collect()
3974 } else {
3975 let mut claimed = std::collections::HashSet::new();
3977 instruments
3978 .iter()
3979 .map(|inst| {
3980 let symbol = inst.name().as_str().to_ascii_uppercase();
3981 if !claimed.insert(symbol.clone()) {
3984 return InstrumentAccountSnapshot::new(
3985 inst.clone(),
3986 Vec::new(),
3987 false,
3988 PositionReport::Unreported,
3989 None,
3990 );
3991 }
3992 let listing = by_symbol
3995 .swap_remove(symbol.as_str())
3996 .unwrap_or_else(Listing::empty);
3997 snapshot(inst.clone(), listing)
3998 })
3999 .collect()
4000 }
4001}
4002
4003fn convert_open_order(
4013 o: &AlpacaOrderResponse,
4014) -> Option<Order<ExchangeId, InstrumentNameExchange, Open>> {
4015 let converted = convert_representable_open_order(o);
4016 if converted.is_none() {
4017 warn!(
4018 order_id = %o.id,
4019 symbol = %o.symbol,
4020 qty = ?o.qty,
4021 side = %o.side,
4022 order_type = %o.order_type,
4023 "Alpaca open order cannot be represented - leaving it out"
4024 );
4025 }
4026 converted
4027}
4028
4029fn alpaca_cid(client_order_id: Option<&str>, order_id: &str) -> ClientOrderId {
4035 ClientOrderId::new(client_order_id.unwrap_or(order_id))
4036}
4037
4038fn convert_representable_open_order(
4040 o: &AlpacaOrderResponse,
4041) -> Option<Order<ExchangeId, InstrumentNameExchange, Open>> {
4042 let order_id = OrderId(SmolStr::new(&o.id));
4043 let cid = alpaca_cid(o.client_order_id.as_deref(), &o.id);
4044
4045 let instrument = InstrumentNameExchange::new(&o.symbol);
4046 let side = parse_side(&o.side)?;
4047 let quantity = Decimal::from_str(o.qty.as_deref().unwrap_or("0")).ok()?;
4048 if quantity.is_zero() {
4052 return None;
4053 }
4054 let price = o
4055 .limit_price
4056 .as_deref()
4057 .and_then(|s| Decimal::from_str(s).ok());
4058 let filled_qty = alpaca_filled_qty(&o.id, Some(&o.filled_qty)).unwrap_or(Decimal::ZERO);
4059 let kind = parse_order_kind(
4060 &o.order_type,
4061 o.stop_price.as_deref(),
4062 o.trail_percent.as_deref(),
4063 o.trail_price.as_deref(),
4064 )?;
4065 let time_in_force = parse_time_in_force(&o.time_in_force);
4066 let time_exchange = order_state_time(o);
4067
4068 Some(Order {
4069 key: OrderKey::new(
4070 ExchangeId::AlpacaBroker,
4071 instrument,
4072 StrategyId::unknown(),
4074 cid,
4075 ),
4076 side,
4077 price,
4078 quantity,
4079 kind,
4080 time_in_force,
4081 state: Open::new(VenueOrderId::Assigned(order_id), time_exchange, filled_qty),
4082 })
4083}
4084
4085fn convert_activity_to_trade(
4087 a: &AlpacaActivity,
4088) -> Option<Trade<AssetNameExchange, InstrumentNameExchange>> {
4089 let trade_id = TradeId::new(activity_execution_id(&a.id));
4090 let order_id = OrderId(SmolStr::new(&a.order_id));
4091 let instrument = InstrumentNameExchange::new(&a.symbol);
4092 let side = parse_side(&a.side)?;
4093 let price = Decimal::from_str(&a.price).ok()?;
4094 let quantity = Decimal::from_str(&a.qty).ok()?;
4095 let order_filled_quantity = a.cum_qty.as_deref().and_then(|s| Decimal::from_str(s).ok());
4098 let time_exchange = parse_timestamp(&a.transaction_time).unwrap_or_else(|| {
4099 warn!(id = %a.id, "Alpaca activity: unparseable transaction_time, using now");
4100 Utc::now()
4101 });
4102
4103 Some(Trade::new(
4108 trade_id,
4109 order_id,
4110 instrument,
4111 StrategyId::unknown(),
4112 time_exchange,
4113 side,
4114 price,
4115 quantity,
4116 order_filled_quantity,
4117 AssetFees::new(
4118 AssetNameExchange::from(USD),
4119 Decimal::ZERO,
4120 Some(Decimal::ZERO),
4121 ),
4122 ))
4123}
4124
4125fn ws_order_snapshot(
4135 order: &AlpacaOrderWs<'_>,
4136 instrument: InstrumentNameExchange,
4137 cid: ClientOrderId,
4138 order_id: OrderId,
4139 time_exchange: DateTime<Utc>,
4140) -> Option<UnindexedAccountEvent> {
4141 let side = parse_side(&order.side)?;
4142 let quantity = Decimal::from_str(order.qty.unwrap_or("0")).unwrap_or(Decimal::ZERO);
4143 if quantity.is_zero() {
4144 trace!(order_id = %order.id, "Alpaca WS: skipping notional order snapshot (qty=None)");
4145 return None;
4146 }
4147 let price = order.limit_price.and_then(|s| Decimal::from_str(s).ok());
4148 let filled_qty = order
4151 .filled_qty
4152 .and_then(|_| alpaca_filled_qty(&order.id, order.filled_qty))
4153 .unwrap_or(Decimal::ZERO);
4154 let kind = parse_order_kind(
4155 &order.order_type,
4156 order.stop_price,
4157 order.trail_percent,
4158 order.trail_price,
4159 )?;
4160 let time_in_force = parse_time_in_force(&order.time_in_force);
4161
4162 let order_snapshot = crate::order::Order {
4163 key: OrderKey::new(
4164 ExchangeId::AlpacaBroker,
4165 instrument,
4166 StrategyId::unknown(),
4167 cid,
4168 ),
4169 side,
4170 price,
4171 quantity,
4172 kind,
4173 time_in_force,
4174 state: OrderState::active(Open::new(
4175 VenueOrderId::Assigned(order_id),
4176 time_exchange,
4177 filled_qty,
4178 )),
4179 };
4180 Some(UnindexedAccountEvent::new(
4181 ExchangeId::AlpacaBroker,
4182 AccountEventKind::OrderSnapshot(rustrade_integration::collection::snapshot::Snapshot(
4183 order_snapshot,
4184 )),
4185 ))
4186}
4187
4188fn ws_fill_order_is_live(status: &str) -> bool {
4198 matches!(status, "partially_filled" | "filled")
4199}
4200
4201fn trade_amendment_event(
4213 update: &AlpacaTradeUpdate<'_>,
4214 instrument: InstrumentNameExchange,
4215 order_id: OrderId,
4216) -> UnindexedAccountEvent {
4217 let order = &update.order;
4218 let time_exchange = update
4219 .timestamp
4220 .and_then(parse_timestamp)
4221 .unwrap_or_else(Utc::now);
4222 let original = update
4223 .previous_execution_id
4224 .filter(|id| !id.is_empty())
4225 .map(TradeId::new);
4226 let quantity = update.qty.and_then(|s| Decimal::from_str(s).ok());
4227
4228 let kind = if update.event == "trade_bust" {
4229 TradeAmendmentKind::Busted {
4230 quantity: quantity.map(|quantity| quantity.abs()),
4231 }
4232 } else {
4233 let id = ws_execution_id(update).map(TradeId::new);
4234 let price = update.price.and_then(|s| Decimal::from_str(s).ok());
4235 match (id, price, quantity, parse_side(&order.side)) {
4236 (Some(id), Some(price), Some(quantity), Some(side)) if quantity > Decimal::ZERO => {
4237 TradeAmendmentKind::Corrected {
4238 replacement: Trade::new(
4239 id,
4240 order_id.clone(),
4241 instrument.clone(),
4242 StrategyId::unknown(),
4243 time_exchange,
4244 side,
4245 price,
4246 quantity,
4247 None,
4250 AssetFees::new(
4252 AssetNameExchange::from(USD),
4253 Decimal::ZERO,
4254 Some(Decimal::ZERO),
4255 ),
4256 ),
4257 }
4258 }
4259 (id, price, quantity, _) => TradeAmendmentKind::CorrectedUnresolved {
4260 id,
4261 price,
4262 quantity,
4263 },
4264 }
4265 };
4266
4267 UnindexedAccountEvent::new(
4268 ExchangeId::AlpacaBroker,
4269 AccountEventKind::TradeAmended(crate::trade::TradeAmendment::new(
4270 instrument,
4271 order_id,
4272 time_exchange,
4273 original,
4274 kind,
4275 )),
4276 )
4277}
4278
4279fn convert_trade_update(update: AlpacaTradeUpdate<'_>) -> [Option<UnindexedAccountEvent>; 2] {
4289 let event_str = update.event.as_str();
4292 if !matches!(
4293 event_str,
4294 "fill"
4295 | "partial_fill"
4296 | "new"
4297 | "accepted"
4298 | "pending_new"
4299 | "canceled"
4300 | "expired"
4301 | "replaced"
4302 | "done_for_day"
4303 | "rejected"
4304 | "trade_bust"
4305 | "trade_correct"
4306 ) {
4307 trace!(event = %event_str, "Alpaca WS: ignoring trade_updates event type");
4308 return [None, None];
4309 }
4310
4311 let order = &update.order;
4312 let instrument = InstrumentNameExchange::new(&*order.symbol);
4313 let order_id = OrderId(order.id.clone());
4314 let cid = alpaca_cid(order.client_order_id.as_deref(), &order.id);
4315
4316 match event_str {
4317 "fill" | "partial_fill" => {
4318 let (Some(price), Some(quantity), Some(side)) = (
4320 update.price.and_then(|s| Decimal::from_str(s).ok()),
4321 update.qty.and_then(|s| Decimal::from_str(s).ok()),
4322 parse_side(&order.side),
4323 ) else {
4324 return [None, None];
4325 };
4326 let time_exchange = update
4327 .timestamp
4328 .and_then(parse_timestamp)
4329 .unwrap_or_else(Utc::now);
4330
4331 let cum_qty = Decimal::from_str(order.filled_qty.unwrap_or("0"))
4336 .inspect_err(|e| {
4337 warn!(
4338 order_id = %order.id,
4339 filled_qty = ?order.filled_qty,
4340 %e,
4341 "Alpaca WS: failed to parse filled_qty — the fill reports a cumulative \
4342 of 0, and without an execution id a second such fill on the same order \
4343 would be deduplicated away"
4344 );
4345 })
4346 .unwrap_or(Decimal::ZERO);
4347 let trade_id = match ws_execution_id(&update) {
4351 Some(execution_id) => TradeId::new(execution_id),
4352 None => {
4353 warn!(
4354 order_id = %order.id,
4355 "Alpaca WS: fill without a usable execution_id — its TradeId is \
4356 \"{{order_id}}:{{filled_qty}}\", which fetch_trades will not match"
4357 );
4358 TradeId(fill_dedup_key(&order.id, cum_qty))
4359 }
4360 };
4361
4362 let trade = Trade::new(
4366 trade_id,
4367 order_id.clone(),
4368 instrument.clone(),
4369 StrategyId::unknown(),
4370 time_exchange,
4371 side,
4372 price,
4373 quantity,
4374 Some(cum_qty),
4376 AssetFees::new(
4377 AssetNameExchange::from(USD),
4378 Decimal::ZERO,
4379 Some(Decimal::ZERO),
4380 ),
4381 );
4382 let trade_event = UnindexedAccountEvent::new(
4383 ExchangeId::AlpacaBroker,
4384 AccountEventKind::Trade(trade),
4385 );
4386
4387 let snapshot_event = if ws_fill_order_is_live(&order.status) {
4391 ws_order_snapshot(order, instrument, cid, order_id, time_exchange)
4392 } else {
4393 trace!(
4394 order_id = %order.id,
4395 status = %order.status,
4396 "Alpaca WS: fill for an order the exchange no longer reports as working — \
4397 emitting the execution without an order snapshot"
4398 );
4399 None
4400 };
4401
4402 [Some(trade_event), snapshot_event]
4403 }
4404
4405 "new" | "accepted" | "pending_new" | "done_for_day" => {
4406 let time_exchange = update
4410 .timestamp
4411 .and_then(parse_timestamp)
4412 .unwrap_or_else(Utc::now);
4413
4414 [
4415 ws_order_snapshot(order, instrument, cid, order_id, time_exchange),
4416 None,
4417 ]
4418 }
4419
4420 "canceled" | "expired" | "replaced" => {
4421 let time_exchange = update
4429 .timestamp
4430 .and_then(parse_timestamp)
4431 .unwrap_or_else(Utc::now);
4432 let filled_qty = alpaca_filled_qty(&order.id, order.filled_qty);
4433 let cancelled = Cancelled::new(order_id, time_exchange, filled_qty);
4434 let response = crate::order::request::OrderResponseCancel {
4435 key: OrderKey::new(
4436 ExchangeId::AlpacaBroker,
4437 instrument,
4438 StrategyId::unknown(),
4439 cid,
4440 ),
4441 state: Ok(cancelled),
4442 };
4443 [
4444 Some(UnindexedAccountEvent::new(
4445 ExchangeId::AlpacaBroker,
4446 AccountEventKind::OrderCancelled(response),
4447 )),
4448 None,
4449 ]
4450 }
4451
4452 "trade_bust" | "trade_correct" => [
4453 Some(trade_amendment_event(&update, instrument, order_id)),
4454 None,
4455 ],
4456
4457 "rejected" => {
4458 let response = crate::order::request::OrderResponseCancel {
4459 key: OrderKey::new(
4460 ExchangeId::AlpacaBroker,
4461 instrument,
4462 StrategyId::unknown(),
4463 cid,
4464 ),
4465 state: Err(UnindexedOrderError::Rejected(ApiError::OrderRejected(
4466 format!("order rejected: status={}", order.status),
4467 ))),
4468 };
4469 [
4470 Some(UnindexedAccountEvent::new(
4471 ExchangeId::AlpacaBroker,
4472 AccountEventKind::OrderCancelled(response),
4473 )),
4474 None,
4475 ]
4476 }
4477
4478 _ => unreachable!("convert_trade_update: unrecognised event passed early-return guard"),
4481 }
4482}
4483
4484fn parse_side(s: &str) -> Option<Side> {
4489 match s {
4490 "buy" | "Buy" | "BUY" => Some(Side::Buy),
4491 "sell" | "Sell" | "SELL" => Some(Side::Sell),
4492 other => {
4493 trace!(%other, "Alpaca: unknown order side");
4494 None
4495 }
4496 }
4497}
4498
4499fn parse_order_kind(
4500 order_type: &str,
4501 stop_price: Option<&str>,
4502 trail_percent: Option<&str>,
4503 trail_price: Option<&str>,
4504) -> Option<OrderKind> {
4505 match order_type {
4506 "market" | "Market" => Some(OrderKind::Market),
4507 "limit" | "Limit" => Some(OrderKind::Limit),
4508 "stop" | "Stop" => {
4509 let trigger_price = stop_price.and_then(|s| Decimal::from_str(s).ok())?;
4510 Some(OrderKind::Stop { trigger_price })
4511 }
4512 "stop_limit" | "Stop_limit" => {
4513 let trigger_price = stop_price.and_then(|s| Decimal::from_str(s).ok())?;
4514 Some(OrderKind::StopLimit { trigger_price })
4515 }
4516 "trailing_stop" | "Trailing_stop" => {
4517 if let Some(pct) = trail_percent.and_then(|s| Decimal::from_str(s).ok()) {
4519 Some(OrderKind::TrailingStop {
4520 offset: pct,
4521 offset_type: TrailingOffsetType::Percentage,
4522 })
4523 } else if let Some(price) = trail_price.and_then(|s| Decimal::from_str(s).ok()) {
4524 Some(OrderKind::TrailingStop {
4525 offset: price,
4526 offset_type: TrailingOffsetType::Absolute,
4527 })
4528 } else {
4529 trace!("Alpaca: trailing_stop missing trail_percent and trail_price");
4530 None
4531 }
4532 }
4533 other => {
4534 trace!(%other, "Alpaca: unsupported order type, skipping");
4535 None
4536 }
4537 }
4538}
4539
4540fn parse_time_in_force(s: &str) -> TimeInForce {
4541 match s {
4542 "gtc" | "GTC" => TimeInForce::GoodUntilCancelled { post_only: false },
4543 "day" | "DAY" => TimeInForce::GoodUntilEndOfDay,
4544 "fok" | "FOK" => TimeInForce::FillOrKill,
4545 "ioc" | "IOC" => TimeInForce::ImmediateOrCancel,
4546 other => {
4547 warn!(%other, "Alpaca: unknown time_in_force, defaulting to GoodUntilEndOfDay");
4548 TimeInForce::GoodUntilEndOfDay
4549 }
4550 }
4551}
4552
4553fn order_state_time(order: &AlpacaOrderResponse) -> DateTime<Utc> {
4563 order
4564 .updated_at
4565 .as_deref()
4566 .and_then(parse_timestamp)
4567 .or_else(|| parse_timestamp(&order.created_at))
4568 .unwrap_or_else(Utc::now)
4569}
4570
4571fn alpaca_filled_qty(order_id: &str, filled_qty: Option<&str>) -> Option<Decimal> {
4574 let Some(raw) = filled_qty else {
4575 warn!(%order_id, "Alpaca did not report how much the order filled");
4576 return None;
4577 };
4578 match Decimal::from_str(raw) {
4579 Ok(filled_qty) => Some(filled_qty),
4580 Err(_) => {
4581 warn!(%order_id, filled_qty = raw, "Alpaca reported an unparseable filled_qty, treating it as unknown");
4582 None
4583 }
4584 }
4585}
4586
4587fn parse_timestamp(s: &str) -> Option<DateTime<Utc>> {
4588 DateTime::parse_from_rfc3339(s)
4589 .ok()
4590 .map(|dt| dt.with_timezone(&Utc))
4591}
4592
4593fn map_side(side: Side) -> &'static str {
4594 match side {
4595 Side::Buy => "buy",
4596 Side::Sell => "sell",
4597 }
4598}
4599
4600fn map_order_kind(kind: OrderKind) -> Option<&'static str> {
4601 match kind {
4602 OrderKind::Market => Some("market"),
4603 OrderKind::Limit => Some("limit"),
4604 OrderKind::Stop { .. } => Some("stop"),
4605 OrderKind::StopLimit { .. } => Some("stop_limit"),
4606 OrderKind::TrailingStop { .. } => Some("trailing_stop"),
4607 OrderKind::TakeProfit { .. }
4609 | OrderKind::TakeProfitLimit { .. }
4610 | OrderKind::TrailingStopLimit { .. } => None,
4611 }
4612}
4613
4614fn map_time_in_force(tif: TimeInForce) -> Result<&'static str, &'static str> {
4623 match tif {
4624 TimeInForce::GoodUntilCancelled { post_only } => {
4625 if post_only {
4626 return Err("Alpaca does not support post_only orders");
4627 }
4628 Ok("gtc")
4629 }
4630 TimeInForce::GoodUntilEndOfDay => Ok("day"),
4631 TimeInForce::FillOrKill => Ok("fok"),
4632 TimeInForce::ImmediateOrCancel => Ok("ioc"),
4633 TimeInForce::AtOpen => Ok("opg"),
4634 TimeInForce::AtClose => Ok("cls"),
4635 TimeInForce::GoodTillDate { .. } => {
4639 Err("Alpaca GoodTillDate is not yet wired through this client")
4640 }
4641 }
4642}
4643
4644fn is_options_or_equity_symbol(symbol: &str) -> bool {
4655 !symbol.contains('/')
4656}
4657
4658fn map_position_intent(side: Side, reduce_only: bool) -> AlpacaPositionIntent {
4667 match (reduce_only, side) {
4668 (false, Side::Buy) => AlpacaPositionIntent::BuyToOpen,
4669 (false, Side::Sell) => AlpacaPositionIntent::SellToOpen,
4670 (true, Side::Buy) => AlpacaPositionIntent::BuyToClose,
4671 (true, Side::Sell) => AlpacaPositionIntent::SellToClose,
4672 }
4673}
4674
4675fn parse_api_error(status: reqwest::StatusCode, message: &str) -> crate::error::UnindexedApiError {
4680 if status == reqwest::StatusCode::TOO_MANY_REQUESTS {
4682 return ApiError::RateLimit;
4683 }
4684
4685 let lower = message.to_ascii_lowercase();
4687 match status.as_u16() {
4688 422 if lower.contains("already") => ApiError::OrderAlreadyCancelled,
4692 422 if lower.contains("insufficient") => {
4699 ApiError::BalanceInsufficient(None, message.to_owned())
4700 }
4701 401 => ApiError::Unauthenticated(format!("unauthorized: {message}")),
4702 403 => ApiError::Unauthenticated(format!("forbidden: {message}")),
4703 404 => ApiError::OrderRejected(format!("order not found: {message}")),
4704 _ => ApiError::OrderRejected(message.to_owned()),
4705 }
4706}
4707
4708fn parse_order_error(status: reqwest::StatusCode, message: &str) -> UnindexedOrderError {
4710 UnindexedOrderError::Rejected(parse_api_error(status, message))
4711}
4712
4713fn connectivity_err(msg: impl Into<String>) -> UnindexedClientError {
4714 UnindexedClientError::Connectivity(ConnectivityError::Socket(msg.into()))
4715}
4716
4717fn order_post_error(error: UnindexedClientError) -> UnindexedOrderError {
4724 match error {
4725 UnindexedClientError::Connectivity(ce) => OrderError::Connectivity(ce),
4726 UnindexedClientError::Api(ae) => OrderError::Rejected(ae),
4727 UnindexedClientError::Internal(msg) => OrderError::Connectivity(ConnectivityError::Socket(
4728 format!("Alpaca order status unknown, its 2xx response could not be read: {msg}"),
4729 )),
4730 other @ (UnindexedClientError::TaskFailed(_)
4734 | UnindexedClientError::Truncated { .. }
4735 | UnindexedClientError::TruncatedSnapshot { .. }) => OrderError::Connectivity(
4736 ConnectivityError::Socket(format!("Alpaca order status unknown: {other}")),
4737 ),
4738 }
4739}
4740
4741impl OrderStatusClient for AlpacaClient {
4746 async fn fetch_ended_orders(
4754 &self,
4755 orders: &[UnindexedOrderKey],
4756 ) -> Result<Vec<UnindexedInactiveOrder>, UnindexedClientError> {
4757 fetch_ended_by_key(orders, |key| {
4758 fetch_order_lookup(
4759 self.http.clone(),
4760 self.rate_limiter.clone(),
4761 self.config.clone(),
4762 key,
4763 )
4764 })
4765 .await
4766 }
4767}
4768
4769impl BracketOrderClient for AlpacaClient {
4774 async fn open_bracket_order(
4775 &self,
4776 request: UnifiedBracketOrderRequest<ExchangeId, &InstrumentNameExchange>,
4777 ) -> UnifiedBracketOrderResult {
4778 let alpaca_request = AlpacaBracketOrderRequest {
4779 instrument: request.key.instrument.clone(),
4780 strategy: request.key.strategy.clone(),
4781 cid: request.key.cid.clone(),
4782 side: request.state.side,
4783 quantity: request.state.quantity,
4784 entry_price: request.state.entry_price,
4785 take_profit_price: request.state.take_profit_price,
4786 stop_loss_price: request.state.stop_loss_price,
4787 stop_loss_limit_price: request.state.stop_loss_limit_price,
4788 time_in_force: request.state.time_in_force,
4789 };
4790
4791 let result = self.open_bracket_order(alpaca_request).await;
4792
4793 UnifiedBracketOrderResult::parent_only(result.parent)
4794 }
4795}
4796
4797#[cfg(test)]
4802#[allow(clippy::unwrap_used, clippy::expect_used)] mod tests {
4804 use super::*;
4805 use rust_decimal_macros::dec;
4806
4807 #[test]
4808 fn test_alpaca_config_new_uses_paper_trading_by_default() {
4809 let cfg = AlpacaConfig::new("my_key".into(), "my_secret".into());
4810 assert!(cfg.paper);
4811 }
4812
4813 #[test]
4814 fn test_alpaca_config_debug_redacts_credentials() {
4815 let cfg = AlpacaConfig::new("my_key".into(), "my_secret".into());
4816 let debug = format!("{cfg:?}");
4817 assert!(!debug.contains("my_key"), "api_key should be redacted");
4818 assert!(
4819 !debug.contains("my_secret"),
4820 "secret_key should be redacted"
4821 );
4822 assert!(debug.contains("paper: true"));
4823 }
4824
4825 #[test]
4826 fn test_alpaca_config_deserialize_omitted_paper_defaults_to_paper() {
4827 let cfg: AlpacaConfig = serde_json::from_str(
4828 r#"{
4829 "api_key": "my_key",
4830 "secret_key": "my_secret"
4831 }"#,
4832 )
4833 .unwrap();
4834
4835 assert!(cfg.paper);
4836 assert_eq!(cfg.rest_base_url(), "https://paper-api.alpaca.markets");
4837 }
4838
4839 #[test]
4840 fn test_alpaca_config_deserialize_paper_true() {
4841 let cfg: AlpacaConfig = serde_json::from_str(
4842 r#"{
4843 "api_key": "my_key",
4844 "secret_key": "my_secret",
4845 "paper": true
4846 }"#,
4847 )
4848 .unwrap();
4849
4850 assert!(cfg.paper);
4851 assert_eq!(cfg.rest_base_url(), "https://paper-api.alpaca.markets");
4852 }
4853
4854 #[test]
4855 fn test_alpaca_config_deserialize_paper_false() {
4856 let cfg: AlpacaConfig = serde_json::from_str(
4857 r#"{
4858 "api_key": "my_key",
4859 "secret_key": "my_secret",
4860 "paper": false
4861 }"#,
4862 )
4863 .unwrap();
4864
4865 assert!(!cfg.paper);
4866 assert_eq!(cfg.rest_base_url(), "https://api.alpaca.markets");
4867 }
4868
4869 #[test]
4870 fn test_alpaca_config_urls() {
4871 let paper = AlpacaConfig::paper("k".into(), "s".into());
4872 assert!(paper.rest_base_url().contains("paper-api"));
4873 assert!(paper.ws_url().contains("paper-api"));
4874
4875 let live = AlpacaConfig::production("k".into(), "s".into());
4876 assert!(!live.rest_base_url().contains("paper-api"));
4877 assert!(!live.ws_url().contains("paper-api"));
4878 }
4879
4880 #[test]
4881 #[serial_test::serial]
4882 fn test_alpaca_config_from_env_defaults_to_paper_trading() {
4883 temp_env::with_vars(
4884 [
4885 ("ALPACA_API_KEY", Some("my_key")),
4886 ("ALPACA_SECRET_KEY", Some("my_secret")),
4887 ("ALPACA_PAPER", None),
4888 ],
4889 || {
4890 let cfg = AlpacaConfig::from_env().unwrap();
4891 assert!(cfg.paper);
4892 assert_eq!(cfg.rest_base_url(), "https://paper-api.alpaca.markets");
4893 },
4894 );
4895 }
4896
4897 #[test]
4898 #[serial_test::serial]
4899 fn test_alpaca_config_from_env_accepts_explicit_production() {
4900 temp_env::with_vars(
4901 [
4902 ("ALPACA_API_KEY", Some("my_key")),
4903 ("ALPACA_SECRET_KEY", Some("my_secret")),
4904 ("ALPACA_PAPER", Some("false")),
4905 ],
4906 || {
4907 let cfg = AlpacaConfig::from_env().unwrap();
4908 assert!(!cfg.paper);
4909 assert_eq!(cfg.rest_base_url(), "https://api.alpaca.markets");
4910 },
4911 );
4912 }
4913
4914 #[test]
4915 #[serial_test::serial]
4916 fn test_alpaca_config_from_env_rejects_invalid_paper() {
4917 temp_env::with_vars(
4918 [
4919 ("ALPACA_API_KEY", Some("my_key")),
4920 ("ALPACA_SECRET_KEY", Some("my_secret")),
4921 ("ALPACA_PAPER", Some("maybe")),
4922 ],
4923 || {
4924 let err = AlpacaConfig::from_env().unwrap_err();
4925 assert!(matches!(err, AlpacaConfigError::InvalidPaper(value) if value == "maybe"));
4926 },
4927 );
4928 }
4929
4930 #[test]
4931 #[serial_test::serial]
4932 fn test_alpaca_config_from_env_requires_credentials() {
4933 temp_env::with_vars(
4934 [
4935 ("ALPACA_API_KEY", None),
4936 ("ALPACA_SECRET_KEY", Some("my_secret")),
4937 ("ALPACA_PAPER", None),
4938 ],
4939 || {
4940 let err = AlpacaConfig::from_env().unwrap_err();
4941 assert!(matches!(err, AlpacaConfigError::MissingApiKey));
4942 },
4943 );
4944 }
4945
4946 #[test]
4947 fn test_parse_side() {
4948 assert_eq!(parse_side("buy"), Some(Side::Buy));
4949 assert_eq!(parse_side("sell"), Some(Side::Sell));
4950 assert_eq!(parse_side("Buy"), Some(Side::Buy));
4951 assert_eq!(parse_side("BUY"), Some(Side::Buy));
4952 assert_eq!(parse_side("unknown"), None);
4953 }
4954
4955 #[test]
4956 fn test_parse_order_kind() {
4957 assert_eq!(
4959 parse_order_kind("market", None, None, None),
4960 Some(OrderKind::Market)
4961 );
4962 assert_eq!(
4963 parse_order_kind("limit", None, None, None),
4964 Some(OrderKind::Limit)
4965 );
4966
4967 assert_eq!(
4969 parse_order_kind("stop", Some("150.00"), None, None),
4970 Some(OrderKind::Stop {
4971 trigger_price: Decimal::from_str("150.00").unwrap()
4972 })
4973 );
4974 assert_eq!(parse_order_kind("stop", None, None, None), None);
4975
4976 assert_eq!(
4978 parse_order_kind("stop_limit", Some("145.00"), None, None),
4979 Some(OrderKind::StopLimit {
4980 trigger_price: Decimal::from_str("145.00").unwrap()
4981 })
4982 );
4983
4984 assert_eq!(
4986 parse_order_kind("trailing_stop", None, Some("5.0"), None),
4987 Some(OrderKind::TrailingStop {
4988 offset: Decimal::from_str("5.0").unwrap(),
4989 offset_type: TrailingOffsetType::Percentage,
4990 })
4991 );
4992
4993 assert_eq!(
4995 parse_order_kind("trailing_stop", None, None, Some("2.50")),
4996 Some(OrderKind::TrailingStop {
4997 offset: Decimal::from_str("2.50").unwrap(),
4998 offset_type: TrailingOffsetType::Absolute,
4999 })
5000 );
5001
5002 assert_eq!(parse_order_kind("trailing_stop", None, None, None), None);
5004
5005 assert_eq!(parse_order_kind("unknown", None, None, None), None);
5007 }
5008
5009 #[test]
5010 fn test_map_order_kind() {
5011 assert_eq!(map_order_kind(OrderKind::Market), Some("market"));
5012 assert_eq!(map_order_kind(OrderKind::Limit), Some("limit"));
5013 assert_eq!(
5014 map_order_kind(OrderKind::Stop {
5015 trigger_price: Decimal::from_str("150.00").unwrap()
5016 }),
5017 Some("stop")
5018 );
5019 assert_eq!(
5020 map_order_kind(OrderKind::StopLimit {
5021 trigger_price: Decimal::from_str("145.00").unwrap()
5022 }),
5023 Some("stop_limit")
5024 );
5025 assert_eq!(
5026 map_order_kind(OrderKind::TrailingStop {
5027 offset: Decimal::from_str("5.0").unwrap(),
5028 offset_type: TrailingOffsetType::Percentage,
5029 }),
5030 Some("trailing_stop")
5031 );
5032 assert_eq!(
5033 map_order_kind(OrderKind::TrailingStop {
5034 offset: Decimal::from_str("2.50").unwrap(),
5035 offset_type: TrailingOffsetType::Absolute,
5036 }),
5037 Some("trailing_stop")
5038 );
5039 assert_eq!(
5041 map_order_kind(OrderKind::TrailingStopLimit {
5042 offset: Decimal::from_str("5.0").unwrap(),
5043 offset_type: TrailingOffsetType::Percentage,
5044 stop_price: Decimal::from_str("95.0").unwrap(),
5045 limit_offset: Decimal::from_str("1.0").unwrap(),
5046 }),
5047 None
5048 );
5049 assert_eq!(
5051 map_order_kind(OrderKind::TakeProfit {
5052 trigger_price: Decimal::from_str("160.00").unwrap()
5053 }),
5054 None
5055 );
5056 assert_eq!(
5057 map_order_kind(OrderKind::TakeProfitLimit {
5058 trigger_price: Decimal::from_str("160.00").unwrap()
5059 }),
5060 None
5061 );
5062 }
5063
5064 #[test]
5065 fn test_map_time_in_force_roundtrip() {
5066 assert_eq!(
5067 map_time_in_force(TimeInForce::GoodUntilCancelled { post_only: false }),
5068 Ok("gtc")
5069 );
5070 assert_eq!(map_time_in_force(TimeInForce::GoodUntilEndOfDay), Ok("day"));
5071 assert_eq!(map_time_in_force(TimeInForce::FillOrKill), Ok("fok"));
5072 assert_eq!(map_time_in_force(TimeInForce::ImmediateOrCancel), Ok("ioc"));
5073 }
5074
5075 #[test]
5076 fn test_map_time_in_force_rejects_post_only() {
5077 let result = map_time_in_force(TimeInForce::GoodUntilCancelled { post_only: true });
5078 assert!(result.is_err(), "post_only must be rejected");
5079 assert!(result.unwrap_err().contains("post_only"));
5080 }
5081
5082 #[test]
5087 fn test_bracket_order_serializes_with_stop_loss_stop_order() {
5088 let body = AlpacaOrderRequest {
5090 symbol: "AAPL",
5091 qty: "10".to_string(),
5092 side: "buy",
5093 order_type: "limit",
5094 time_in_force: "gtc",
5095 limit_price: Some("150.00".to_string()),
5096 stop_price: None,
5097 trail_percent: None,
5098 trail_price: None,
5099 client_order_id: Some("bracket-001"),
5100 position_intent: Some(AlpacaPositionIntent::BuyToOpen),
5101 order_class: Some("bracket"),
5102 take_profit: Some(TakeProfitParams {
5103 limit_price: "160.00".to_string(),
5104 }),
5105 stop_loss: Some(StopLossParams {
5106 stop_price: "145.00".to_string(),
5107 limit_price: None,
5108 }),
5109 };
5110
5111 let json = serde_json::to_value(&body).unwrap();
5112
5113 assert_eq!(json["symbol"], "AAPL");
5114 assert_eq!(json["qty"], "10");
5115 assert_eq!(json["side"], "buy");
5116 assert_eq!(json["type"], "limit");
5117 assert_eq!(json["time_in_force"], "gtc");
5118 assert_eq!(json["limit_price"], "150.00");
5119 assert_eq!(json["order_class"], "bracket");
5120
5121 assert_eq!(json["take_profit"]["limit_price"], "160.00");
5123
5124 assert_eq!(json["stop_loss"]["stop_price"], "145.00");
5126 assert!(
5127 json["stop_loss"].get("limit_price").is_none(),
5128 "stop_loss.limit_price should be omitted when None"
5129 );
5130 }
5131
5132 #[test]
5133 fn test_bracket_order_serializes_with_stop_loss_stop_limit_order() {
5134 let body = AlpacaOrderRequest {
5136 symbol: "SPY",
5137 qty: "5".to_string(),
5138 side: "sell",
5139 order_type: "limit",
5140 time_in_force: "day",
5141 limit_price: Some("450.00".to_string()),
5142 stop_price: None,
5143 trail_percent: None,
5144 trail_price: None,
5145 client_order_id: Some("bracket-002"),
5146 position_intent: Some(AlpacaPositionIntent::SellToClose),
5147 order_class: Some("bracket"),
5148 take_profit: Some(TakeProfitParams {
5149 limit_price: "440.00".to_string(),
5150 }),
5151 stop_loss: Some(StopLossParams {
5152 stop_price: "455.00".to_string(),
5153 limit_price: Some("456.00".to_string()),
5154 }),
5155 };
5156
5157 let json = serde_json::to_value(&body).unwrap();
5158
5159 assert_eq!(json["symbol"], "SPY");
5160 assert_eq!(json["side"], "sell");
5161 assert_eq!(json["time_in_force"], "day");
5162 assert_eq!(json["order_class"], "bracket");
5163
5164 assert_eq!(json["take_profit"]["limit_price"], "440.00");
5166
5167 assert_eq!(json["stop_loss"]["stop_price"], "455.00");
5169 assert_eq!(
5170 json["stop_loss"]["limit_price"], "456.00",
5171 "stop_loss.limit_price should be present for stop-limit orders"
5172 );
5173 }
5174
5175 #[tokio::test]
5180 async fn test_open_bracket_order_rejects_invalid_tif() {
5181 use rust_decimal_macros::dec;
5182 use rustrade_instrument::instrument::name::InstrumentNameExchange;
5183
5184 let config = AlpacaConfig::new("dummy_key".into(), "dummy_secret".into());
5186 let client = AlpacaClient::new(config);
5187
5188 let request = AlpacaBracketOrderRequest::new(
5189 InstrumentNameExchange::new("SPY"),
5190 crate::order::id::StrategyId::new("test"),
5191 crate::order::id::ClientOrderId::new("test-tif"),
5192 Side::Buy,
5193 dec!(1),
5194 dec!(100.00),
5195 dec!(120.00),
5196 dec!(90.00),
5197 TimeInForce::ImmediateOrCancel, );
5199
5200 let result = client.open_bracket_order(request).await;
5201
5202 assert!(
5203 result.parent.state.is_failed(),
5204 "Bracket order with IOC TIF should be rejected locally"
5205 );
5206 }
5207
5208 #[tokio::test]
5209 async fn test_open_bracket_order_rejects_invalid_price_ordering() {
5210 use rust_decimal_macros::dec;
5211 use rustrade_instrument::instrument::name::InstrumentNameExchange;
5212
5213 let config = AlpacaConfig::new("dummy_key".into(), "dummy_secret".into());
5214 let client = AlpacaClient::new(config);
5215
5216 let request = AlpacaBracketOrderRequest::new(
5218 InstrumentNameExchange::new("SPY"),
5219 crate::order::id::StrategyId::new("test"),
5220 crate::order::id::ClientOrderId::new("test-price"),
5221 Side::Buy,
5222 dec!(1),
5223 dec!(100.00),
5224 dec!(120.00),
5225 dec!(105.00), TimeInForce::GoodUntilCancelled { post_only: false },
5227 );
5228
5229 let result = client.open_bracket_order(request).await;
5230
5231 assert!(
5232 result.parent.state.is_failed(),
5233 "Bracket order with invalid price ordering should be rejected locally"
5234 );
5235 }
5236
5237 #[tokio::test]
5238 async fn test_open_bracket_order_rejects_invalid_sl_limit_price() {
5239 use rust_decimal_macros::dec;
5240 use rustrade_instrument::instrument::name::InstrumentNameExchange;
5241
5242 let config = AlpacaConfig::new("dummy_key".into(), "dummy_secret".into());
5243 let client = AlpacaClient::new(config);
5244
5245 let request = AlpacaBracketOrderRequest::new(
5247 InstrumentNameExchange::new("SPY"),
5248 crate::order::id::StrategyId::new("test"),
5249 crate::order::id::ClientOrderId::new("test-sl-limit"),
5250 Side::Buy,
5251 dec!(1),
5252 dec!(100.00),
5253 dec!(120.00),
5254 dec!(90.00),
5255 TimeInForce::GoodUntilCancelled { post_only: false },
5256 )
5257 .with_stop_loss_limit_price(dec!(95.00)); let result = client.open_bracket_order(request).await;
5260
5261 assert!(
5262 result.parent.state.is_failed(),
5263 "Bracket order with invalid SL limit price should be rejected locally"
5264 );
5265 }
5266
5267 #[test]
5268 fn test_non_bracket_order_omits_bracket_fields() {
5269 let body = AlpacaOrderRequest {
5271 symbol: "AAPL",
5272 qty: "1".to_string(),
5273 side: "buy",
5274 order_type: "limit",
5275 time_in_force: "gtc",
5276 limit_price: Some("150.00".to_string()),
5277 stop_price: None,
5278 trail_percent: None,
5279 trail_price: None,
5280 client_order_id: Some("regular-001"),
5281 position_intent: Some(AlpacaPositionIntent::BuyToOpen),
5282 order_class: None,
5283 take_profit: None,
5284 stop_loss: None,
5285 };
5286
5287 let json = serde_json::to_value(&body).unwrap();
5288
5289 assert_eq!(json["symbol"], "AAPL");
5290 assert!(
5291 json.get("order_class").is_none(),
5292 "order_class should be omitted for non-bracket orders"
5293 );
5294 assert!(
5295 json.get("take_profit").is_none(),
5296 "take_profit should be omitted for non-bracket orders"
5297 );
5298 assert!(
5299 json.get("stop_loss").is_none(),
5300 "stop_loss should be omitted for non-bracket orders"
5301 );
5302 }
5303
5304 #[test]
5305 fn test_parse_timestamp_valid() {
5306 let ts = parse_timestamp("2025-04-18T14:30:00Z");
5307 assert!(ts.is_some());
5308 let ts2 = parse_timestamp("2025-04-18T14:30:00.123456Z");
5309 assert!(ts2.is_some());
5310 assert_eq!(parse_timestamp("not-a-timestamp"), None);
5311 }
5312
5313 #[test]
5314 fn test_check_auth_response_authorized() {
5315 let msg =
5316 r#"{"stream":"authorization","data":{"status":"authorized","action":"authenticate"}}"#;
5317 assert!(matches!(check_auth_response(msg), Some(Ok(()))));
5318 }
5319
5320 #[test]
5321 fn test_check_auth_response_unauthorized() {
5322 let msg = r#"{"stream":"authorization","data":{"status":"unauthorized"}}"#;
5323 assert!(matches!(
5324 check_auth_response(msg),
5325 Some(Err(HandshakeError::Auth(_)))
5326 ));
5327 }
5328
5329 #[test]
5330 fn test_check_auth_response_non_auth_message() {
5331 let msg = r#"{"stream":"trade_updates","data":{}}"#;
5332 assert!(check_auth_response(msg).is_none());
5333 }
5334
5335 #[test]
5336 fn test_check_listen_ack() {
5337 let ack = r#"{"stream":"listening","data":{"streams":["trade_updates"]}}"#;
5338 assert!(check_listen_ack(ack));
5339
5340 let other = r#"{"stream":"authorization","data":{}}"#;
5341 assert!(!check_listen_ack(other));
5342 }
5343
5344 #[test]
5345 fn test_dedup_cache() {
5346 let cache = new_dedup_cache();
5347 let key = || SmolStr::new("order-1:1");
5348 assert!(
5349 !is_duplicate_fill(&cache, Some("exec-1"), key()),
5350 "first time should not be duplicate"
5351 );
5352 assert!(
5353 is_duplicate_fill(&cache, Some("exec-1"), key()),
5354 "second time should be duplicate"
5355 );
5356 }
5357
5358 #[test]
5361 fn dedup_tells_fills_apart_by_execution_id_not_cumulative() {
5362 let cache = new_dedup_cache();
5363 assert!(!is_duplicate_fill(
5364 &cache,
5365 Some("exec-a"),
5366 SmolStr::new("ord-1:5")
5367 ));
5368 assert!(
5370 !is_duplicate_fill(&cache, Some("exec-b"), SmolStr::new("ord-1:5")),
5371 "a different execution at a cumulative seen before is a new fill"
5372 );
5373 }
5374
5375 #[test]
5378 fn dedup_matches_a_fill_with_an_execution_id_to_the_same_fill_without_one() {
5379 let cache = new_dedup_cache();
5380
5381 assert!(!is_duplicate_fill(&cache, None, SmolStr::new("ord-1:2")));
5382 assert!(
5383 is_duplicate_fill(&cache, Some("exec-a"), SmolStr::new("ord-1:2")),
5384 "without an id first, then with one"
5385 );
5386
5387 assert!(!is_duplicate_fill(
5388 &cache,
5389 Some("exec-b"),
5390 SmolStr::new("ord-1:4")
5391 ));
5392 assert!(
5393 is_duplicate_fill(&cache, None, SmolStr::new("ord-1:4")),
5394 "with an id first, then without one"
5395 );
5396 }
5397
5398 #[tokio::test]
5399 async fn test_exponential_backoff_progression_and_exhaustion() {
5400 tokio::time::pause();
5401
5402 let mut b = ExponentialBackoff::new();
5403
5404 assert!(b.wait().await, "first wait should return true");
5406 assert_eq!(b.attempt, 1);
5407
5408 while b.wait().await {}
5410
5411 assert_eq!(b.attempt, MAX_RECONNECT_ATTEMPTS);
5413
5414 assert!(!b.wait().await, "exhausted backoff should return false");
5416
5417 b.reset();
5419 assert_eq!(b.attempt, 0);
5420
5421 assert!(b.wait().await, "wait should succeed after reset");
5423 assert_eq!(b.attempt, 1);
5424 }
5425
5426 #[test]
5427 fn test_convert_account_to_balances_empty_assets() {
5428 let account: AlpacaAccount = serde_json::from_value(serde_json::json!({
5431 "cash": "9000.50",
5432 "equity": "12000.00",
5433 "buying_power": "36000.00",
5434 "regt_buying_power": "18000.00",
5435 "options_buying_power": "8000.00",
5436 "non_marginable_buying_power": "8500.25",
5437 }))
5438 .unwrap();
5439 let balances = convert_account_to_balances(&account, &[]);
5440 assert_eq!(balances.len(), 1);
5441 assert_eq!(balances[0].asset, AssetNameExchange::new("USD"));
5442 assert_eq!(balances[0].balance.total, dec!(9000.50), "total is cash");
5443 assert_eq!(
5444 balances[0].balance.free,
5445 dec!(8500.25),
5446 "free is non-marginable buying power when below cash"
5447 );
5448 assert_eq!(balances[0].balance.margin, None);
5449 }
5450
5451 #[test]
5454 fn test_convert_account_to_balances_free_capped_at_cash_with_long_stock() {
5455 let account: AlpacaAccount = serde_json::from_value(serde_json::json!({
5456 "cash": "50000",
5457 "equity": "100000",
5458 "initial_margin": "25000",
5459 "non_marginable_buying_power": "75000",
5460 }))
5461 .unwrap();
5462 let balances = convert_account_to_balances(&account, &[]);
5463 assert_eq!(balances[0].balance.total, dec!(50000));
5464 assert_eq!(balances[0].balance.free, dec!(50000));
5465 }
5466
5467 #[test]
5470 fn test_convert_account_to_balances_open_short_keeps_free_below_cash() {
5471 let account: AlpacaAccount = serde_json::from_value(serde_json::json!({
5472 "cash": "10200",
5473 "equity": "10000",
5474 "initial_margin": "100",
5475 "non_marginable_buying_power": "9900",
5476 }))
5477 .unwrap();
5478 let balances = convert_account_to_balances(&account, &[]);
5479 assert_eq!(balances[0].balance.total, dec!(10200));
5480 assert_eq!(balances[0].balance.free, dec!(9900));
5481 }
5482
5483 #[test]
5484 fn test_convert_account_to_balances_negative_cash_on_margin() {
5485 let account: AlpacaAccount = serde_json::from_value(serde_json::json!({
5486 "cash": "-2500.75",
5487 "equity": "7500.00",
5488 "buying_power": "5000.00",
5489 "non_marginable_buying_power": "1250.00",
5490 }))
5491 .unwrap();
5492 let balances = convert_account_to_balances(&account, &[]);
5493 assert_eq!(balances.len(), 1);
5494 assert_eq!(balances[0].balance.total, dec!(-2500.75));
5495 assert_eq!(
5496 balances[0].balance.free,
5497 dec!(-2500.75),
5498 "no free cash while borrowing; free <= total still holds"
5499 );
5500 }
5501
5502 #[test]
5503 fn test_alpaca_account_malformed_cash_fails_to_decode() {
5504 let result = serde_json::from_value::<AlpacaAccount>(serde_json::json!({
5505 "cash": "not-a-number",
5506 "non_marginable_buying_power": "0",
5507 }));
5508 assert!(result.is_err(), "a malformed cash must not read as zero");
5509 }
5510
5511 #[test]
5512 fn test_convert_account_to_balances_usd_filter() {
5513 let account = AlpacaAccount {
5514 cash: dec!(12000.00),
5515 non_marginable_buying_power: dec!(10000.00),
5516 };
5517 let usd = vec![AssetNameExchange::new("USD")];
5518 let balances = convert_account_to_balances(&account, &usd);
5519 assert_eq!(balances.len(), 1);
5520 assert_eq!(balances[0].asset, AssetNameExchange::new("USD"));
5521
5522 let non_usd = vec![AssetNameExchange::new("BTC")];
5523 let balances = convert_account_to_balances(&account, &non_usd);
5524 assert!(balances.is_empty());
5525 }
5526
5527 #[test]
5530 fn a_usd_balance_is_named_usd_however_the_filter_spells_it() {
5531 let account = AlpacaAccount {
5532 cash: dec!(12000.00),
5533 non_marginable_buying_power: dec!(10000.00),
5534 };
5535 let balances = convert_account_to_balances(&account, &[AssetNameExchange::new("usd")]);
5536 assert_eq!(balances.len(), 1);
5537 assert_eq!(balances[0].asset, AssetNameExchange::new("USD"));
5538 }
5539
5540 #[test]
5541 fn test_is_options_or_equity_symbol() {
5542 assert!(!is_options_or_equity_symbol("BTC/USD"));
5544 assert!(!is_options_or_equity_symbol("ETH/USD"));
5545 assert!(!is_options_or_equity_symbol("SOL/USD"));
5546
5547 assert!(is_options_or_equity_symbol("AAPL"));
5549 assert!(is_options_or_equity_symbol("SPY"));
5550 assert!(is_options_or_equity_symbol("MSFT"));
5551
5552 assert!(is_options_or_equity_symbol("SPY250418C00450000"));
5554 assert!(is_options_or_equity_symbol("AAPL250418P00145000"));
5555 }
5556
5557 #[test]
5558 fn test_parse_order_error_already_cancelled() {
5559 assert!(matches!(
5562 parse_order_error(
5563 reqwest::StatusCode::UNPROCESSABLE_ENTITY,
5564 "order is already cancelled"
5565 ),
5566 UnindexedOrderError::Rejected(ApiError::OrderAlreadyCancelled)
5567 ));
5568 }
5569
5570 #[test]
5571 fn test_parse_order_error_already_wins_over_insufficient_on_422() {
5572 assert!(matches!(
5575 parse_order_error(
5576 reqwest::StatusCode::UNPROCESSABLE_ENTITY,
5577 "order already cancelled due to insufficient margin"
5578 ),
5579 UnindexedOrderError::Rejected(ApiError::OrderAlreadyCancelled)
5580 ));
5581 }
5582
5583 fn make_order_ws<'a>(
5584 id: &str,
5585 symbol: &str,
5586 side: &str,
5587 filled_qty: &'a str,
5588 ) -> AlpacaOrderWs<'a> {
5589 AlpacaOrderWs {
5590 id: SmolStr::new(id),
5591 client_order_id: None,
5592 symbol: SmolStr::new(symbol),
5593 qty: Some("2"),
5594 filled_qty: Some(filled_qty),
5595 side: SmolStr::new(side),
5596 order_type: SmolStr::new("limit"),
5597 time_in_force: SmolStr::new("day"),
5598 limit_price: Some("100.00"),
5599 stop_price: None,
5600 trail_percent: None,
5601 trail_price: None,
5602 status: SmolStr::new("partially_filled"),
5603 }
5604 }
5605
5606 fn sole_event(events: [Option<UnindexedAccountEvent>; 2]) -> Option<UnindexedAccountEvent> {
5611 let [first, second] = events;
5612 assert!(
5613 second.is_none(),
5614 "expected a single event, got a second: {second:?}"
5615 );
5616 first
5617 }
5618
5619 fn fill_events(
5621 events: [Option<UnindexedAccountEvent>; 2],
5622 ) -> (
5623 Trade<AssetNameExchange, InstrumentNameExchange>,
5624 crate::order::Order<
5625 ExchangeId,
5626 InstrumentNameExchange,
5627 OrderState<AssetNameExchange, InstrumentNameExchange>,
5628 >,
5629 ) {
5630 let [first, second] = events;
5631 let first = first.expect("a fill frame must produce an execution");
5632 let second = second.expect("a fill frame must produce an order snapshot");
5633 let AccountEventKind::Trade(trade) = first.kind else {
5634 panic!("first event must be the Trade, got {:?}", first.kind);
5635 };
5636 let AccountEventKind::OrderSnapshot(rustrade_integration::collection::snapshot::Snapshot(
5637 order,
5638 )) = second.kind
5639 else {
5640 panic!(
5641 "second event must be the OrderSnapshot, got {:?}",
5642 second.kind
5643 );
5644 };
5645 (trade, order)
5646 }
5647
5648 #[test]
5649 fn test_convert_trade_update_fill_produces_trade_with_execution_id() {
5650 let update = AlpacaTradeUpdate {
5651 event: SmolStr::new("fill"),
5652 execution_id: Some("exec-1"),
5653 previous_execution_id: None,
5654 order: make_order_ws("ord-1", "SPY", "buy", "1"),
5655 price: Some("150.00"),
5656 qty: Some("1"),
5657 timestamp: Some("2025-04-18T14:30:00Z"),
5658 };
5659 let (trade, order) = fill_events(convert_trade_update(update));
5660 assert_eq!(trade.id.0.as_str(), "exec-1");
5662 assert_eq!(trade.price, Decimal::from_str("150.00").unwrap());
5663 assert_eq!(trade.quantity, Decimal::from_str("1").unwrap());
5664
5665 let OrderState::Active(crate::order::state::ActiveOrderState::Open(open)) = order.state
5668 else {
5669 panic!("expected an Open snapshot, got {:?}", order.state);
5670 };
5671 assert_eq!(open.filled_quantity, Decimal::from_str("1").unwrap());
5672 assert_eq!(open.id.assigned().map(|id| id.0.as_str()), Some("ord-1"));
5673 }
5674
5675 #[test]
5676 fn test_convert_trade_update_partial_fill() {
5677 let update = AlpacaTradeUpdate {
5678 event: SmolStr::new("partial_fill"),
5679 execution_id: None,
5680 previous_execution_id: None,
5681 order: make_order_ws("ord-2", "AAPL", "sell", "0.5"),
5682 price: Some("200.00"),
5683 qty: Some("0.5"),
5684 timestamp: None,
5685 };
5686 let (trade, order) = fill_events(convert_trade_update(update));
5687 assert_eq!(trade.quantity, Decimal::from_str("0.5").unwrap());
5688
5689 let OrderState::Active(crate::order::state::ActiveOrderState::Open(open)) = order.state
5692 else {
5693 panic!("expected an Open snapshot, got {:?}", order.state);
5694 };
5695 assert_eq!(open.filled_quantity, Decimal::from_str("0.5").unwrap());
5696 assert_eq!(
5697 open.quantity_remaining(order.quantity),
5698 Decimal::from_str("1.5").unwrap()
5699 );
5700 }
5701
5702 #[test]
5706 fn a_full_fill_snapshot_reports_nothing_left_to_fill() {
5707 let update = AlpacaTradeUpdate {
5708 event: SmolStr::new("fill"),
5709 execution_id: None,
5710 previous_execution_id: None,
5711 order: AlpacaOrderWs {
5713 status: SmolStr::new("filled"),
5714 ..make_order_ws("ord-full", "SPY", "buy", "2")
5715 },
5716 price: Some("150.00"),
5717 qty: Some("1"),
5718 timestamp: Some("2025-04-18T14:30:00Z"),
5719 };
5720 let (_trade, order) = fill_events(convert_trade_update(update));
5721 let OrderState::Active(crate::order::state::ActiveOrderState::Open(open)) = order.state
5722 else {
5723 panic!("expected an Open snapshot, got {:?}", order.state);
5724 };
5725 assert_eq!(open.filled_quantity, Decimal::from_str("2").unwrap());
5726 assert!(
5727 open.quantity_remaining(order.quantity).is_zero(),
5728 "a completed order must report nothing left to fill so the engine retires it"
5729 );
5730 }
5731
5732 #[test]
5736 fn a_fill_for_an_order_no_longer_working_emits_the_execution_without_a_snapshot() {
5737 for status in [
5738 "canceled",
5739 "expired",
5740 "rejected",
5741 "done_for_day",
5742 "pending_cancel",
5743 ] {
5744 let update = AlpacaTradeUpdate {
5745 event: SmolStr::new("partial_fill"),
5746 execution_id: None,
5747 previous_execution_id: None,
5748 order: AlpacaOrderWs {
5749 status: SmolStr::new(status),
5750 ..make_order_ws("ord-late", "SPY", "buy", "1")
5751 },
5752 price: Some("150.00"),
5753 qty: Some("1"),
5754 timestamp: None,
5755 };
5756 let [trade, snapshot] = convert_trade_update(update);
5757 assert!(
5758 matches!(trade.map(|e| e.kind), Some(AccountEventKind::Trade(_))),
5759 "the execution must still be reported for status {status}"
5760 );
5761 assert!(
5762 snapshot.is_none(),
5763 "status {status} must not produce an order snapshot"
5764 );
5765 }
5766 }
5767
5768 #[test]
5771 fn a_notional_order_fill_emits_the_execution_without_a_snapshot() {
5772 let update = AlpacaTradeUpdate {
5773 event: SmolStr::new("partial_fill"),
5774 execution_id: None,
5775 previous_execution_id: None,
5776 order: AlpacaOrderWs {
5777 qty: None,
5778 ..make_order_ws("ord-notional-ws", "SPY", "buy", "1")
5779 },
5780 price: Some("150.00"),
5781 qty: Some("1"),
5782 timestamp: None,
5783 };
5784 let [trade, snapshot] = convert_trade_update(update);
5785 assert!(matches!(
5786 trade.map(|e| e.kind),
5787 Some(AccountEventKind::Trade(_))
5788 ));
5789 assert!(
5790 snapshot.is_none(),
5791 "a notional order has no quantity to snapshot against"
5792 );
5793 }
5794
5795 #[test]
5796 fn test_convert_trade_update_new_order_produces_snapshot() {
5797 let update = AlpacaTradeUpdate {
5798 event: SmolStr::new("new"),
5799 execution_id: None,
5800 previous_execution_id: None,
5801 order: AlpacaOrderWs {
5802 id: SmolStr::new("ord-new"),
5803 client_order_id: Some(SmolStr::new("cid-1")),
5804 symbol: SmolStr::new("AAPL"),
5805 qty: Some("10"),
5806 filled_qty: Some("0"),
5807 side: SmolStr::new("buy"),
5808 order_type: SmolStr::new("limit"),
5809 time_in_force: SmolStr::new("day"),
5810 limit_price: Some("150.00"),
5811 stop_price: None,
5812 trail_percent: None,
5813 trail_price: None,
5814 status: SmolStr::new("new"),
5815 },
5816 price: None,
5817 qty: None,
5818 timestamp: Some("2025-04-18T14:30:00Z"),
5819 };
5820 let event =
5821 sole_event(convert_trade_update(update)).expect("new event should produce an event");
5822 assert!(matches!(event.kind, AccountEventKind::OrderSnapshot(_)));
5823 }
5824
5825 #[test]
5826 fn test_convert_trade_update_canceled_produces_cancel() {
5827 let update = AlpacaTradeUpdate {
5828 event: SmolStr::new("canceled"),
5829 execution_id: None,
5830 previous_execution_id: None,
5831 order: make_order_ws("ord-3", "AAPL", "sell", "0"),
5832 price: None,
5833 qty: None,
5834 timestamp: Some("2025-04-18T14:30:00Z"),
5835 };
5836 let event =
5837 sole_event(convert_trade_update(update)).expect("canceled should produce an event");
5838 let AccountEventKind::OrderCancelled(response) = event.kind else {
5839 panic!("expected OrderCancelled, got {:?}", event.kind);
5840 };
5841 assert!(response.state.is_ok());
5842 }
5843
5844 #[test]
5846 fn a_canceled_update_carries_its_fill_or_none() {
5847 let cancelled = |filled_qty: Option<&'static str>| {
5848 let mut order = make_order_ws("ord-3", "AAPL", "sell", "0");
5849 order.filled_qty = filled_qty;
5850 let update = AlpacaTradeUpdate {
5851 event: SmolStr::new("canceled"),
5852 execution_id: None,
5853 previous_execution_id: None,
5854 order,
5855 price: None,
5856 qty: None,
5857 timestamp: Some("2025-04-18T14:30:00Z"),
5858 };
5859 let event =
5860 sole_event(convert_trade_update(update)).expect("canceled should produce an event");
5861 let AccountEventKind::OrderCancelled(response) = event.kind else {
5862 panic!("expected OrderCancelled, got {:?}", event.kind);
5863 };
5864 let Ok(cancelled) = response.state else {
5865 panic!("expected Ok, got {:?}", response.state);
5866 };
5867 cancelled.filled_quantity
5868 };
5869
5870 assert_eq!(cancelled(Some("1")), Some(Decimal::ONE));
5871 assert_eq!(cancelled(Some("0")), Some(Decimal::ZERO));
5872 assert_eq!(cancelled(None), None);
5873 assert_eq!(cancelled(Some("not a number")), None);
5874 }
5875
5876 #[test]
5879 fn an_ended_order_with_an_unparseable_fill_reports_it_unknown() {
5880 let key = OrderKey::new(
5881 ExchangeId::AlpacaBroker,
5882 InstrumentNameExchange::new("AAPL"),
5883 StrategyId::new("strategy"),
5884 ClientOrderId::new("c"),
5885 );
5886 let ended = |status: &str| {
5887 let mut order = make_order_response("o1", "AAPL");
5888 order.status = Some(status.to_string());
5889 order.filled_qty = "not a number".to_string();
5890 convert_ended_order(&order, &key).map(|order| order.state)
5891 };
5892
5893 assert!(matches!(
5894 ended("canceled"),
5895 Some(InactiveOrderState::Cancelled(Cancelled {
5896 filled_quantity: None,
5897 ..
5898 }))
5899 ));
5900 assert!(matches!(
5901 ended("expired"),
5902 Some(InactiveOrderState::Expired(Expired {
5903 filled_quantity: None,
5904 ..
5905 }))
5906 ));
5907 assert!(matches!(
5908 ended("filled"),
5909 Some(InactiveOrderState::FullyFilled(Filled { filled_quantity, .. }))
5910 if filled_quantity == Decimal::ONE
5911 ));
5912 }
5913
5914 #[test]
5915 fn test_convert_trade_update_rejected_produces_error() {
5916 let update = AlpacaTradeUpdate {
5917 event: SmolStr::new("rejected"),
5918 execution_id: None,
5919 previous_execution_id: None,
5920 order: make_order_ws("ord-4", "SPY", "buy", "0"),
5921 price: None,
5922 qty: None,
5923 timestamp: None,
5924 };
5925 let event =
5926 sole_event(convert_trade_update(update)).expect("rejected should produce an event");
5927 let AccountEventKind::OrderCancelled(response) = event.kind else {
5928 panic!("expected OrderCancelled, got {:?}", event.kind);
5929 };
5930 assert!(response.state.is_err());
5931 }
5932
5933 #[test]
5934 fn test_convert_open_order_notional_qty_none_is_skipped() {
5935 let order = AlpacaOrderResponse {
5938 id: "ord-notional".to_string(),
5939 client_order_id: None,
5940 symbol: "SPY".to_string(),
5941 qty: None,
5942 filled_qty: "0".to_string(),
5943 filled_avg_price: None,
5944 status: None,
5945 side: "buy".to_string(),
5946 order_type: "market".to_string(),
5947 time_in_force: "day".to_string(),
5948 limit_price: None,
5949 stop_price: None,
5950 trail_percent: None,
5951 trail_price: None,
5952 created_at: "2025-04-18T14:30:00Z".to_string(),
5953 updated_at: None,
5954 };
5955 assert!(convert_open_order(&order).is_none());
5956 }
5957
5958 #[test]
5959 fn test_convert_activity_to_trade_bad_price_returns_none() {
5960 let activity = AlpacaActivity {
5964 id: "act-1".to_string(),
5965 order_id: "ord-1".to_string(),
5966 symbol: "SPY250418C00450000".to_string(),
5967 side: "buy".to_string(),
5968 price: "not-a-number".to_string(),
5969 qty: "1".to_string(),
5970 transaction_time: "2025-04-18T14:30:00Z".to_string(),
5971 cum_qty: Some("1".to_string()),
5972 };
5973 assert!(convert_activity_to_trade(&activity).is_none());
5974 }
5975
5976 #[test]
5977 fn test_convert_positions_to_balances_crypto() {
5978 let positions = vec![
5979 AlpacaPosition {
5981 qty_available: dec!(0.4),
5982 ..alpaca_position("BTCUSD", "crypto", AlpacaPositionSide::Long, dec!(0.5))
5983 },
5984 alpaca_position("ETH/USD", "crypto", AlpacaPositionSide::Long, dec!(2.0)),
5986 alpaca_position("AAPL", "us_equity", AlpacaPositionSide::Long, dec!(10)),
5988 ];
5989
5990 let balances = convert_positions_to_balances(&positions, &[]);
5992 assert_eq!(balances.len(), 2, "only crypto positions returned");
5993 assert_eq!(balances[0].asset.name().as_str(), "BTC");
5994 assert_eq!(balances[1].asset.name().as_str(), "ETH");
5995 assert_eq!(balances[0].balance.total, dec!(0.5));
5997 assert_eq!(balances[0].balance.free, dec!(0.4));
5998
5999 let btc_only = vec![AssetNameExchange::new("btc")];
6001 let balances = convert_positions_to_balances(&positions, &btc_only);
6002 assert_eq!(balances.len(), 1);
6003 assert_eq!(balances[0].asset.name().as_str(), "BTC");
6004 }
6005
6006 #[test]
6007 fn test_crypto_position_asset() {
6008 assert_eq!(crypto_position_asset("BTCUSD"), "BTC");
6009 assert_eq!(
6010 crypto_position_asset("btcusd"),
6011 "btc",
6012 "suffix matched ignoring case"
6013 );
6014 assert_eq!(
6015 crypto_position_asset("USDTUSD"),
6016 "USDT",
6017 "only one USD is stripped"
6018 );
6019 assert_eq!(crypto_position_asset("BTC/USD"), "BTC");
6020 assert_eq!(crypto_position_asset("BTC/USDT"), "BTC");
6021 assert_eq!(crypto_position_asset("€USD"), "€");
6022 assert_eq!(crypto_position_asset("USD"), "USD");
6024 assert_eq!(crypto_position_asset("/USD"), "/USD");
6025 assert_eq!(crypto_position_asset("BTCUSDT"), "BTCUSDT");
6026 assert_eq!(
6027 crypto_position_asset("BTCێ"),
6028 "BTCێ",
6029 "no char boundary at the suffix"
6030 );
6031 assert_eq!(crypto_position_asset(""), "");
6032 }
6033
6034 fn alpaca_position(
6036 symbol: &str,
6037 asset_class: &str,
6038 side: AlpacaPositionSide,
6039 qty: Decimal,
6040 ) -> AlpacaPosition {
6041 AlpacaPosition {
6042 symbol: symbol.into(),
6043 asset_class: asset_class.into(),
6044 side,
6045 qty,
6046 qty_available: qty,
6047 avg_entry_price: Some(dec!(100)),
6048 unrealized_pl: Some(dec!(0)),
6049 }
6050 }
6051
6052 #[test]
6053 fn test_convert_positions_sign_comes_from_side_not_qty() {
6054 let now = Utc::now();
6055 let positions = vec![
6056 alpaca_position("AAPL", "us_equity", AlpacaPositionSide::Long, dec!(10)),
6057 alpaca_position("TSLA", "us_equity", AlpacaPositionSide::Short, dec!(-3)),
6058 alpaca_position("MSFT", "us_equity", AlpacaPositionSide::Short, dec!(4)),
6060 ];
6061
6062 let converted = convert_positions(&positions, now).unwrap();
6063 let quantities: Vec<_> = converted.iter().map(|(s, p)| (*s, p.quantity)).collect();
6064 assert_eq!(
6065 quantities,
6066 vec![("AAPL", dec!(10)), ("TSLA", dec!(-3)), ("MSFT", dec!(-4))]
6067 );
6068 assert!(converted.iter().all(|(_, p)| p.time_exchange == now));
6069 }
6070
6071 #[test]
6072 fn test_convert_positions_leaves_out_crypto_and_flat() {
6073 let positions = vec![
6074 alpaca_position("BTC/USD", "crypto", AlpacaPositionSide::Long, dec!(0.5)),
6075 alpaca_position("AAPL", "us_equity", AlpacaPositionSide::Long, dec!(0)),
6076 alpaca_position("SPY", "us_equity", AlpacaPositionSide::Long, dec!(1)),
6077 ];
6078
6079 let converted = convert_positions(&positions, Utc::now()).unwrap();
6080 let symbols: Vec<_> = converted.iter().map(|(s, _)| *s).collect();
6081 assert_eq!(symbols, vec!["SPY"]);
6082 }
6083
6084 #[test]
6085 fn test_alpaca_position_decodes_equity_option_and_fractional() {
6086 let positions: Vec<AlpacaPosition> = serde_json::from_value(serde_json::json!([
6087 {
6088 "asset_id": "00000000-0000-0000-0000-000000000001",
6089 "symbol": "XYZ",
6090 "exchange": "NASDAQ",
6091 "asset_class": "us_equity",
6092 "qty": "-7",
6093 "qty_available": "-7",
6094 "side": "short",
6095 "avg_entry_price": "50.25",
6096 "market_value": "-351.75",
6097 "unrealized_pl": "0",
6098 },
6099 {
6100 "symbol": "XYZ271217C00055000",
6101 "asset_class": "us_option",
6102 "qty": "2",
6103 "qty_available": "2",
6104 "side": "long",
6105 "avg_entry_price": "1.35",
6106 "unrealized_pl": "-20",
6107 },
6108 {
6109 "symbol": "ABC",
6110 "asset_class": "us_equity",
6111 "qty": "0.125",
6112 "qty_available": "0.125",
6113 "side": "long",
6114 "avg_entry_price": "80",
6115 "unrealized_pl": null,
6116 },
6117 ]))
6118 .unwrap();
6119
6120 let converted = convert_positions(&positions, Utc::now()).unwrap();
6121 let [
6122 (short_sym, short),
6123 (option_sym, option),
6124 (fraction_sym, fraction),
6125 ] = converted.as_slice()
6126 else {
6127 panic!("expected three positions, got {converted:?}");
6128 };
6129
6130 assert_eq!(*short_sym, "XYZ");
6131 assert_eq!(short.quantity, dec!(-7));
6132 assert_eq!(short.entry_price, Some(dec!(50.25)));
6133 assert!(short.is_short());
6134
6135 assert_eq!(*option_sym, "XYZ271217C00055000");
6136 assert_eq!(option.quantity, dec!(2));
6137 assert_eq!(option.entry_price, Some(dec!(1.35)), "per-share premium");
6138 assert_eq!(option.unrealized_pnl, Some(dec!(-20)));
6139
6140 assert_eq!(*fraction_sym, "ABC");
6141 assert_eq!(fraction.quantity, dec!(0.125));
6142 assert_eq!(fraction.unrealized_pnl, None);
6143
6144 for (_, p) in &converted {
6145 assert_eq!(p.margin_used, None);
6146 assert_eq!(p.liquidation_price, None);
6147 assert_eq!(p.leverage, None);
6148 }
6149 }
6150
6151 #[test]
6154 fn test_alpaca_position_unknown_side_fails_only_where_a_direction_is_needed() {
6155 let positions: Vec<AlpacaPosition> = serde_json::from_value(serde_json::json!([
6156 {
6157 "symbol": "BTC/USD",
6158 "asset_class": "crypto",
6159 "qty": "0.5",
6160 "qty_available": "0.5",
6161 },
6162 {
6163 "symbol": "XYZ",
6164 "asset_class": "us_equity",
6165 "qty": "1",
6166 "qty_available": "1",
6167 "side": "sideways",
6168 },
6169 ]))
6170 .unwrap();
6171 assert_eq!(positions[0].side, AlpacaPositionSide::Unknown);
6172 assert_eq!(positions[1].side, AlpacaPositionSide::Unknown);
6173
6174 let balances = convert_positions_to_balances(&positions, &[]);
6175 assert_eq!(balances.len(), 1);
6176 assert_eq!(balances[0].balance.total, dec!(0.5));
6177
6178 let result = convert_positions(&positions, Utc::now());
6179 assert!(
6180 matches!(&result, Err(UnindexedClientError::Internal(msg)) if msg.contains("XYZ")),
6181 "an unknown side must not guess a direction: {result:?}"
6182 );
6183 }
6184
6185 #[test]
6187 fn order_state_time_prefers_updated_at() {
6188 let mut resp = make_order_response("ord-1", "SPY");
6189 resp.created_at = "2025-04-18T14:30:00Z".to_string();
6190 resp.updated_at = Some("2025-04-18T15:45:00Z".to_string());
6191 assert_eq!(
6192 order_state_time(&resp),
6193 parse_timestamp("2025-04-18T15:45:00Z").unwrap(),
6194 "updated_at must win over created_at"
6195 );
6196
6197 resp.updated_at = None;
6199 assert_eq!(
6200 order_state_time(&resp),
6201 parse_timestamp("2025-04-18T14:30:00Z").unwrap(),
6202 "absent updated_at falls back to created_at"
6203 );
6204
6205 resp.updated_at = Some("not-a-timestamp".to_string());
6208 assert_eq!(
6209 order_state_time(&resp),
6210 parse_timestamp("2025-04-18T14:30:00Z").unwrap(),
6211 "unparseable updated_at falls back to created_at"
6212 );
6213 }
6214
6215 #[test]
6218 fn a_placement_response_without_a_status_reads_from_what_filled() {
6219 let instrument = InstrumentNameExchange::new("SPY");
6220 let mut resp = make_order_response("ord-1", "SPY");
6221 resp.filled_qty = "0.4".to_string();
6222 let OrderState::Active(ActiveOrderState::Open(open)) =
6223 placed_order_state(&resp, &instrument, Decimal::ONE)
6224 else {
6225 panic!("expected Open");
6226 };
6227 assert_eq!(open.filled_quantity, Decimal::new(4, 1));
6228
6229 resp.filled_qty = "1".to_string();
6230 resp.filled_avg_price = Some("100.5".to_string());
6231 let state = placed_order_state(&resp, &instrument, Decimal::ONE);
6232 let OrderState::Inactive(InactiveOrderState::FullyFilled(filled)) = state else {
6233 panic!("expected FullyFilled, got {state:?}");
6234 };
6235 assert_eq!(filled.filled_quantity, Decimal::ONE);
6236 assert_eq!(filled.avg_price, Some(Decimal::new(1005, 1)));
6237 }
6238
6239 fn make_order_response(id: &str, symbol: &str) -> AlpacaOrderResponse {
6240 AlpacaOrderResponse {
6241 id: id.to_string(),
6242 client_order_id: None,
6243 symbol: symbol.to_string(),
6244 qty: Some("1".to_string()),
6245 filled_qty: "0".to_string(),
6246 filled_avg_price: None,
6247 status: None,
6248 side: "buy".to_string(),
6249 order_type: "limit".to_string(),
6250 time_in_force: "day".to_string(),
6251 limit_price: Some("100.00".to_string()),
6252 stop_price: None,
6253 trail_percent: None,
6254 trail_price: None,
6255 created_at: "2025-04-18T14:30:00Z".to_string(),
6256 updated_at: None,
6257 }
6258 }
6259
6260 #[test]
6261 fn test_build_instrument_snapshots_empty_instruments_returns_only_with_orders() {
6262 let orders = vec![
6263 make_order_response("o1", "AAPL"),
6264 make_order_response("o2", "SPY"),
6265 ];
6266 let snapshots = build_instrument_snapshots(orders, Vec::new(), &[]);
6267 assert_eq!(snapshots.len(), 2);
6268 let symbols: Vec<&str> = snapshots
6269 .iter()
6270 .map(|s| s.instrument.name().as_str())
6271 .collect();
6272 assert!(symbols.contains(&"AAPL"));
6273 assert!(symbols.contains(&"SPY"));
6274 }
6275
6276 #[test]
6277 fn test_build_instrument_snapshots_requested_instrument_no_orders_gets_empty_snapshot() {
6278 let orders = vec![make_order_response("o1", "AAPL")];
6281 let instruments = vec![
6282 InstrumentNameExchange::new("AAPL"),
6283 InstrumentNameExchange::new("SPY"),
6284 ];
6285 let snapshots = build_instrument_snapshots(orders, Vec::new(), &instruments);
6286 assert_eq!(snapshots.len(), 2);
6287 let spy = snapshots
6288 .iter()
6289 .find(|s| s.instrument.name().as_str() == "SPY")
6290 .expect("SPY snapshot must be present even with no orders");
6291 assert!(spy.orders.is_empty());
6292 }
6293
6294 fn make_notional_order_response(id: &str, symbol: &str) -> AlpacaOrderResponse {
6296 AlpacaOrderResponse {
6297 qty: None,
6298 ..make_order_response(id, symbol)
6299 }
6300 }
6301
6302 fn snapshot_for<'a>(
6303 snapshots: &'a [InstrumentAccountSnapshot<
6304 ExchangeId,
6305 AssetNameExchange,
6306 InstrumentNameExchange,
6307 >],
6308 symbol: &str,
6309 ) -> &'a InstrumentAccountSnapshot<ExchangeId, AssetNameExchange, InstrumentNameExchange> {
6310 snapshots
6311 .iter()
6312 .find(|snapshot| snapshot.instrument.name().as_str() == symbol)
6313 .unwrap_or_else(|| panic!("no snapshot for {symbol}"))
6314 }
6315
6316 #[test]
6320 fn test_build_instrument_snapshots_declares_complete_lists_complete() {
6321 let instruments = [
6322 InstrumentNameExchange::new("AAPL"),
6323 InstrumentNameExchange::new("SPY"),
6324 ];
6325 for requested in [&instruments[..], &[]] {
6326 let orders = vec![make_order_response("o1", "AAPL")];
6327 let snapshots = build_instrument_snapshots(orders, Vec::new(), requested);
6328 assert!(!snapshots.is_empty());
6329 assert!(snapshots.iter().all(|snapshot| snapshot.orders_complete));
6330 }
6331 }
6332
6333 #[test]
6337 fn test_build_instrument_snapshots_notional_order_leaves_only_its_instrument_incomplete() {
6338 let orders = vec![
6339 make_order_response("o1", "AAPL"),
6340 make_order_response("o2", "SPY"),
6341 make_notional_order_response("o3", "SPY"),
6342 ];
6343 let instruments = [
6344 InstrumentNameExchange::new("AAPL"),
6345 InstrumentNameExchange::new("SPY"),
6346 ];
6347 let snapshots = build_instrument_snapshots(orders, Vec::new(), &instruments);
6348
6349 assert!(snapshot_for(&snapshots, "AAPL").orders_complete);
6350 let spy = snapshot_for(&snapshots, "SPY");
6351 assert!(!spy.orders_complete);
6352 assert_eq!(spy.orders.len(), 1);
6353 }
6354
6355 #[test]
6358 fn test_build_instrument_snapshots_unfiltered_lists_an_instrument_with_only_a_notional_order() {
6359 let orders = vec![make_notional_order_response("o1", "SPY")];
6360 let snapshots = build_instrument_snapshots(orders, Vec::new(), &[]);
6361
6362 let spy = snapshot_for(&snapshots, "SPY");
6363 assert!(!spy.orders_complete);
6364 assert!(spy.orders.is_empty());
6365 }
6366
6367 #[test]
6368 fn test_build_instrument_snapshots_non_requested_instrument_excluded() {
6369 let orders = vec![
6372 make_order_response("o1", "AAPL"),
6373 make_order_response("o2", "MSFT"), ];
6375 let instruments = vec![InstrumentNameExchange::new("AAPL")];
6376 let snapshots = build_instrument_snapshots(orders, Vec::new(), &instruments);
6377 assert_eq!(snapshots.len(), 1);
6378 assert_eq!(snapshots[0].instrument.name().as_str(), "AAPL");
6379 }
6380
6381 fn position(quantity: Decimal) -> Position {
6382 Position::new(
6383 quantity,
6384 Some(dec!(100)),
6385 None,
6386 None,
6387 None,
6388 None,
6389 Utc::now(),
6390 )
6391 }
6392
6393 #[test]
6394 fn test_build_instrument_snapshots_unfiltered_lists_positions_without_orders() {
6395 let orders = vec![
6396 make_order_response("o1", "AAPL"),
6397 make_order_response("o2", "MSFT"),
6398 make_order_response("o3", "BTC/USD"),
6399 ];
6400 let positions = vec![("AAPL", position(dec!(5))), ("TSLA", position(dec!(-2)))];
6401
6402 let snapshots = build_instrument_snapshots(orders, positions, &[]);
6403 let listed: Vec<_> = snapshots
6404 .iter()
6405 .map(|s| {
6406 (
6407 s.instrument.name().as_str(),
6408 s.orders.len(),
6409 s.position.quantity(),
6410 )
6411 })
6412 .collect();
6413 assert_eq!(
6415 listed,
6416 vec![
6417 ("AAPL", 1, Some(dec!(5))),
6418 ("MSFT", 1, Some(Decimal::ZERO)),
6419 ("BTC/USD", 1, None),
6420 ("TSLA", 0, Some(dec!(-2)))
6421 ]
6422 );
6423 assert!(snapshots.iter().all(|s| s.orders_complete));
6424 }
6425
6426 #[test]
6427 fn test_build_instrument_snapshots_filtered_attaches_only_requested_positions() {
6428 let positions = vec![("AAPL", position(dec!(5))), ("TSLA", position(dec!(-2)))];
6429 let instruments = vec![
6430 InstrumentNameExchange::new("tsla"),
6431 InstrumentNameExchange::new("MSFT"),
6432 InstrumentNameExchange::new("BTC/USD"),
6433 InstrumentNameExchange::new("TSLA"),
6434 ];
6435
6436 let snapshots = build_instrument_snapshots(Vec::new(), positions, &instruments);
6437 let listed: Vec<_> = snapshots
6438 .iter()
6439 .map(|s| (s.instrument.name().as_str(), s.position.quantity()))
6440 .collect();
6441 assert_eq!(
6445 listed,
6446 vec![
6447 ("tsla", Some(dec!(-2))),
6448 ("MSFT", Some(Decimal::ZERO)),
6449 ("BTC/USD", None),
6450 ("TSLA", None)
6451 ]
6452 );
6453 assert!(!snapshots[3].orders_complete);
6454 }
6455
6456 #[test]
6462 fn test_recover_fills_dedup_key_matches_ws_path() {
6463 let order_id = "ord-1";
6464
6465 let ws_keys: Vec<SmolStr> = ["1", "2"]
6468 .iter()
6469 .map(|filled_qty| {
6470 early_dedup_key(&AlpacaTradeUpdate {
6471 event: SmolStr::new("partial_fill"),
6472 execution_id: None,
6473 previous_execution_id: None,
6474 order: make_order_ws(order_id, "SPY", "buy", filled_qty),
6475 price: Some("150.00"),
6476 qty: Some("1"),
6477 timestamp: None,
6478 })
6479 })
6480 .collect();
6481
6482 let mut cumulative = Decimal::ZERO;
6485 let rest_keys: Vec<SmolStr> = ["1", "1"]
6486 .iter()
6487 .map(|exec_qty| {
6488 cumulative += Decimal::from_str(exec_qty).unwrap();
6489 fill_dedup_key(order_id, cumulative)
6490 })
6491 .collect();
6492
6493 assert_eq!(
6494 ws_keys, rest_keys,
6495 "REST recovery dedup keys must match WS path keys for cross-source dedup to work"
6496 );
6497 assert_eq!(ws_keys[0].as_str(), "ord-1:1");
6498 assert_eq!(ws_keys[1].as_str(), "ord-1:2");
6499 }
6500
6501 #[test]
6506 fn a_ws_fill_without_an_execution_id_is_identified_by_its_dedup_key() {
6507 let update = AlpacaTradeUpdate {
6508 event: SmolStr::new("fill"),
6509 execution_id: None,
6510 previous_execution_id: None,
6511 order: make_order_ws("ord-abc", "SPY", "buy", "5"),
6512 price: Some("150.00"),
6513 qty: Some("5"),
6514 timestamp: None,
6515 };
6516
6517 let early_key = early_dedup_key(&update);
6519
6520 let [event, _snapshot] = convert_trade_update(update);
6522 let Some(UnindexedAccountEvent {
6523 kind: AccountEventKind::Trade(trade),
6524 ..
6525 }) = event
6526 else {
6527 panic!("fill should produce an execution");
6528 };
6529
6530 assert_eq!(
6531 trade.id.0.as_str(),
6532 early_key.as_str(),
6533 "without an execution_id the TradeId is the dedup key"
6534 );
6535 assert_eq!(early_key.as_str(), "ord-abc:5");
6536 }
6537
6538 #[test]
6543 fn early_dedup_key_normalizes_decimal_strings() {
6544 let update1 = AlpacaTradeUpdate {
6546 event: SmolStr::new("fill"),
6547 execution_id: None,
6548 previous_execution_id: None,
6549 order: AlpacaOrderWs {
6550 id: SmolStr::new("ord-x"),
6551 client_order_id: Some(SmolStr::new("cid")),
6552 symbol: SmolStr::new("AAPL"),
6553 qty: Some("10"),
6554 filled_qty: Some("1.00"),
6555 side: SmolStr::new("buy"),
6556 order_type: SmolStr::new("market"),
6557 time_in_force: SmolStr::new("day"),
6558 limit_price: None,
6559 stop_price: None,
6560 trail_percent: None,
6561 trail_price: None,
6562 status: SmolStr::new("filled"),
6563 },
6564 price: Some("100.00"),
6565 qty: Some("10"),
6566 timestamp: None,
6567 };
6568 assert_eq!(early_dedup_key(&update1).as_str(), "ord-x:1");
6569
6570 let update2 = AlpacaTradeUpdate {
6572 event: SmolStr::new("fill"),
6573 execution_id: None,
6574 previous_execution_id: None,
6575 order: AlpacaOrderWs {
6576 id: SmolStr::new("ord-x"),
6577 client_order_id: Some(SmolStr::new("cid")),
6578 symbol: SmolStr::new("AAPL"),
6579 qty: Some("10"),
6580 filled_qty: Some("1"),
6581 side: SmolStr::new("buy"),
6582 order_type: SmolStr::new("market"),
6583 time_in_force: SmolStr::new("day"),
6584 limit_price: None,
6585 stop_price: None,
6586 trail_percent: None,
6587 trail_price: None,
6588 status: SmolStr::new("filled"),
6589 },
6590 price: Some("100.00"),
6591 qty: Some("10"),
6592 timestamp: None,
6593 };
6594 assert_eq!(early_dedup_key(&update2).as_str(), "ord-x:1");
6595
6596 let update3 = AlpacaTradeUpdate {
6598 event: SmolStr::new("fill"),
6599 execution_id: None,
6600 previous_execution_id: None,
6601 order: AlpacaOrderWs {
6602 id: SmolStr::new("ord-x"),
6603 client_order_id: Some(SmolStr::new("cid")),
6604 symbol: SmolStr::new("AAPL"),
6605 qty: Some("10"),
6606 filled_qty: Some("1.0"),
6607 side: SmolStr::new("buy"),
6608 order_type: SmolStr::new("market"),
6609 time_in_force: SmolStr::new("day"),
6610 limit_price: None,
6611 stop_price: None,
6612 trail_percent: None,
6613 trail_price: None,
6614 status: SmolStr::new("filled"),
6615 },
6616 price: Some("100.00"),
6617 qty: Some("10"),
6618 timestamp: None,
6619 };
6620 assert_eq!(early_dedup_key(&update3).as_str(), "ord-x:1");
6621 }
6622
6623 #[test]
6628 fn process_ws_text_rejected_event_without_filled_qty_is_not_dropped() {
6629 let (tx, mut rx) = mpsc::unbounded_channel();
6630 let dedup = new_dedup_cache();
6631 let mut backoff = ExponentialBackoff::new();
6632
6633 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"}}}"#;
6635
6636 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
6637 process_ws_text(json, &tx, &dedup, &known, &mut backoff);
6638
6639 let event = rx.try_recv()
6641 .expect("rejected event without filled_qty must produce an AccountEvent, not be silently dropped");
6642 assert!(
6643 matches!(event.kind, AccountEventKind::OrderCancelled(_)),
6644 "rejected event must map to OrderCancelled, got: {:?}",
6645 event.kind
6646 );
6647 }
6648
6649 #[test]
6652 fn process_ws_text_identifies_a_fill_by_its_execution_id() {
6653 let frame = |execution_id: &str| {
6654 format!(
6655 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"}}}}"#
6656 )
6657 };
6658 let cases = [
6659 (
6660 r#""execution_id":"524b1902-817e-446c-825b-a9fcfebbc17e","#,
6661 "524b1902-817e-446c-825b-a9fcfebbc17e",
6662 ),
6663 (r#""execution_id":null,"#, "ord-1:2"),
6664 (r#""execution_id":"","#, "ord-1:2"),
6665 ("", "ord-1:2"),
6666 ];
6667 for (field, expected) in cases {
6668 let (tx, mut rx) = mpsc::unbounded_channel();
6669 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
6670 process_ws_text(
6671 &frame(field),
6672 &tx,
6673 &new_dedup_cache(),
6674 &known,
6675 &mut ExponentialBackoff::new(),
6676 );
6677 let Ok(UnindexedAccountEvent {
6678 kind: AccountEventKind::Trade(trade),
6679 ..
6680 }) = rx.try_recv()
6681 else {
6682 panic!("the fill frame {field:?} produces a Trade");
6683 };
6684 assert_eq!(trade.id.0.as_str(), expected, "frame {field:?}");
6685 }
6686 }
6687
6688 fn amendment_frame(event: &str, fields: &str) -> String {
6691 format!(
6692 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"}}}}"#
6693 )
6694 }
6695
6696 fn stream_events(frames: &[String]) -> Vec<UnindexedAccountEvent> {
6698 let (tx, mut rx) = mpsc::unbounded_channel();
6699 let dedup = new_dedup_cache();
6700 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
6701 let mut backoff = ExponentialBackoff::new();
6702 for frame in frames {
6703 process_ws_text(frame, &tx, &dedup, &known, &mut backoff);
6704 }
6705 std::iter::from_fn(|| rx.try_recv().ok()).collect()
6706 }
6707
6708 fn amendment_of(
6710 frame: String,
6711 ) -> crate::trade::TradeAmendment<AssetNameExchange, InstrumentNameExchange> {
6712 let events = stream_events(&[frame]);
6713 let [
6714 UnindexedAccountEvent {
6715 kind: AccountEventKind::TradeAmended(amendment),
6716 ..
6717 },
6718 ] = <[_; 1]>::try_from(events).unwrap_or_else(|events| {
6719 panic!("one TradeAmended, got {events:?}");
6720 })
6721 else {
6722 panic!("expected TradeAmended");
6723 };
6724 amendment
6725 }
6726
6727 #[test]
6730 fn process_ws_text_reports_a_trade_bust() {
6731 let amendment = amendment_of(amendment_frame(
6732 "trade_bust",
6733 r#""execution_id":"exec-x","previous_execution_id":"exec-a","qty":"-2","#,
6734 ));
6735 assert_eq!(amendment.original, Some(TradeId::new("exec-a")));
6736 assert_eq!(amendment.order_id, OrderId::new("ord-1"));
6737 assert_eq!(amendment.instrument, InstrumentNameExchange::new("SPY"));
6738 assert_eq!(
6739 Some(amendment.time_exchange),
6740 parse_timestamp("2025-04-18T14:31:00Z")
6741 );
6742 assert_eq!(
6743 amendment.kind,
6744 TradeAmendmentKind::Busted {
6745 quantity: Some(Decimal::TWO)
6746 }
6747 );
6748 }
6749
6750 #[test]
6752 fn process_ws_text_reports_a_bust_that_names_no_fill() {
6753 let amendment = amendment_of(amendment_frame("trade_bust", ""));
6754 assert_eq!(amendment.original, None);
6755 assert_eq!(
6756 amendment.kind,
6757 TradeAmendmentKind::Busted { quantity: None }
6758 );
6759 }
6760
6761 #[test]
6764 fn process_ws_text_reports_a_trade_correct_with_its_replacement() {
6765 let amendment = amendment_of(amendment_frame(
6766 "trade_correct",
6767 r#""execution_id":"exec-c","previous_execution_id":"exec-a","price":"101.50","qty":"2","#,
6768 ));
6769 assert_eq!(amendment.original, Some(TradeId::new("exec-a")));
6770 let TradeAmendmentKind::Corrected { replacement } = amendment.kind else {
6771 panic!("a full replacement: {:?}", amendment.kind);
6772 };
6773 assert_eq!(replacement.id, TradeId::new("exec-c"));
6774 assert_eq!(replacement.order_id, OrderId::new("ord-1"));
6775 assert_eq!(replacement.side, Side::Buy);
6776 assert_eq!(replacement.price, Decimal::new(10150, 2));
6777 assert_eq!(replacement.quantity, Decimal::TWO);
6778 assert_eq!(replacement.order_filled_quantity, None);
6779 assert_eq!(replacement.time_exchange, amendment.time_exchange);
6780 }
6781
6782 #[test]
6785 fn process_ws_text_reports_a_correction_it_cannot_resolve() {
6786 let price = Some(Decimal::new(10150, 2));
6787 let cases = [
6788 (
6789 r#""execution_id":"exec-c","qty":"2","#,
6790 (Some("exec-c"), None, Some(Decimal::TWO)),
6791 ),
6792 (
6793 r#""price":"101.50","qty":"2","#,
6794 (None, price, Some(Decimal::TWO)),
6795 ),
6796 (
6797 r#""execution_id":"exec-c","price":"101.50","qty":"0","#,
6798 (Some("exec-c"), price, Some(Decimal::ZERO)),
6799 ),
6800 (
6801 r#""execution_id":"exec-c","price":"101.50","#,
6802 (Some("exec-c"), price, None),
6803 ),
6804 ];
6805 for (fields, (id, price, quantity)) in cases {
6806 let amendment = amendment_of(amendment_frame("trade_correct", fields));
6807 assert_eq!(
6808 amendment.kind,
6809 TradeAmendmentKind::CorrectedUnresolved {
6810 id: id.map(TradeId::new),
6811 price,
6812 quantity,
6813 },
6814 "{fields}"
6815 );
6816 }
6817 }
6818
6819 #[test]
6822 fn a_fill_after_a_bust_is_delivered_at_a_cumulative_seen_before() {
6823 let fill = |execution_id: &str| {
6824 format!(
6825 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"}}}}"#
6826 )
6827 };
6828 let events = stream_events(&[
6829 fill("exec-a"),
6830 amendment_frame(
6831 "trade_bust",
6832 r#""execution_id":"exec-x","previous_execution_id":"exec-a","qty":"-2","#,
6833 ),
6834 fill("exec-b"),
6835 ]);
6836
6837 let kinds: Vec<_> = events
6838 .iter()
6839 .filter_map(|event| match &event.kind {
6840 AccountEventKind::Trade(trade) => Some(format!("trade {}", trade.id)),
6841 AccountEventKind::TradeAmended(amendment) => {
6842 Some(format!("amended {:?}", amendment.original))
6843 }
6844 _ => None,
6845 })
6846 .collect();
6847 assert_eq!(
6848 kinds,
6849 vec![
6850 "trade exec-a",
6851 "amended Some(TradeId(\"exec-a\"))",
6852 "trade exec-b"
6853 ]
6854 );
6855 }
6856
6857 #[test]
6860 fn decimal_zero_normalize_is_zero_str() {
6861 assert_eq!(Decimal::ZERO.normalize().to_string(), "0");
6862 }
6863
6864 #[test]
6868 fn decimal_normalize_strips_trailing_zeros() {
6869 let from_rest = Decimal::from_str("1.00").unwrap().normalize();
6871 let from_ws = Decimal::from_str("1").unwrap().normalize();
6872 assert_eq!(from_rest.to_string(), from_ws.to_string());
6873 assert_eq!(from_rest.to_string(), "1");
6874
6875 assert_eq!(
6877 Decimal::from_str("100.000")
6878 .unwrap()
6879 .normalize()
6880 .to_string(),
6881 "100"
6882 );
6883 assert_eq!(
6884 Decimal::from_str("0.10").unwrap().normalize().to_string(),
6885 "0.1"
6886 );
6887 assert_eq!(
6888 Decimal::from_str("0.100").unwrap().normalize().to_string(),
6889 "0.1"
6890 );
6891 }
6892
6893 #[test]
6895 fn map_position_intent_open_buy_maps_to_buy_to_open() {
6896 assert_eq!(
6897 map_position_intent(Side::Buy, false),
6898 AlpacaPositionIntent::BuyToOpen
6899 );
6900 }
6901
6902 #[test]
6903 fn map_position_intent_open_sell_maps_to_sell_to_open() {
6904 assert_eq!(
6905 map_position_intent(Side::Sell, false),
6906 AlpacaPositionIntent::SellToOpen
6907 );
6908 }
6909
6910 #[test]
6911 fn map_position_intent_reduce_buy_maps_to_buy_to_close() {
6912 assert_eq!(
6913 map_position_intent(Side::Buy, true),
6914 AlpacaPositionIntent::BuyToClose
6915 );
6916 }
6917
6918 #[test]
6919 fn map_position_intent_reduce_sell_maps_to_sell_to_close() {
6920 assert_eq!(
6921 map_position_intent(Side::Sell, true),
6922 AlpacaPositionIntent::SellToClose
6923 );
6924 }
6925
6926 #[test]
6928 fn parse_order_error_401_maps_to_unauthenticated() {
6929 assert!(matches!(
6931 parse_order_error(reqwest::StatusCode::UNAUTHORIZED, "bad credentials"),
6932 UnindexedOrderError::Rejected(ApiError::Unauthenticated(_))
6933 ));
6934 }
6935
6936 #[test]
6937 fn parse_order_error_403_maps_to_unauthenticated() {
6938 assert!(matches!(
6941 parse_order_error(reqwest::StatusCode::FORBIDDEN, "account suspended"),
6942 UnindexedOrderError::Rejected(ApiError::Unauthenticated(_))
6943 ));
6944 }
6945
6946 #[test]
6947 fn parse_order_error_404_maps_to_order_rejected_with_not_found_prefix() {
6948 let err = parse_order_error(reqwest::StatusCode::NOT_FOUND, "order not found");
6949 let UnindexedOrderError::Rejected(ApiError::OrderRejected(msg)) = err else {
6950 panic!("expected OrderRejected, got {err:?}");
6951 };
6952 assert!(
6953 msg.contains("order not found"),
6954 "message should contain 'order not found': {msg}"
6955 );
6956 }
6957
6958 #[test]
6959 fn parse_order_error_422_insufficient_only_maps_to_balance_insufficient() {
6960 assert!(matches!(
6962 parse_order_error(
6963 reqwest::StatusCode::UNPROCESSABLE_ENTITY,
6964 "insufficient funds for this order"
6965 ),
6966 UnindexedOrderError::Rejected(ApiError::BalanceInsufficient(None, _))
6967 ));
6968 }
6969
6970 #[test]
6971 fn parse_order_error_429_maps_to_rate_limit() {
6972 assert!(matches!(
6973 parse_order_error(reqwest::StatusCode::TOO_MANY_REQUESTS, "rate limited"),
6974 UnindexedOrderError::Rejected(ApiError::RateLimit)
6975 ));
6976 }
6977
6978 #[test]
6982 fn parse_time_in_force_unknown_value_falls_back_to_good_until_end_of_day() {
6983 assert_eq!(
6984 parse_time_in_force("opg"),
6985 TimeInForce::GoodUntilEndOfDay,
6986 "unknown TIF must fall back to GoodUntilEndOfDay (with a warn! in production)"
6987 );
6988 }
6989
6990 #[test]
6993 fn build_instrument_snapshots_output_order_matches_instruments_slice() {
6994 let orders = vec![
6996 make_order_response("o1", "SPY"),
6997 make_order_response("o2", "AAPL"),
6998 make_order_response("o3", "MSFT"),
6999 ];
7000 let instruments = vec![
7002 InstrumentNameExchange::new("MSFT"),
7003 InstrumentNameExchange::new("AAPL"),
7004 ];
7005 let snapshots = build_instrument_snapshots(orders, Vec::new(), &instruments);
7006 assert_eq!(snapshots.len(), 2);
7007 assert_eq!(
7008 snapshots[0].instrument.name().as_str(),
7009 "MSFT",
7010 "first snapshot must be MSFT (first in instruments slice)"
7011 );
7012 assert_eq!(
7013 snapshots[1].instrument.name().as_str(),
7014 "AAPL",
7015 "second snapshot must be AAPL (second in instruments slice)"
7016 );
7017 }
7018
7019 mod http_tests {
7026 use super::super::*;
7027 use wiremock::matchers::{method, path};
7028 use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate};
7029
7030 struct Sequential {
7037 call: std::sync::atomic::AtomicU32,
7038 pages: Vec<serde_json::Value>,
7039 }
7040
7041 impl Sequential {
7042 fn new(pages: Vec<serde_json::Value>) -> Self {
7043 Self {
7044 call: std::sync::atomic::AtomicU32::new(0),
7045 pages,
7046 }
7047 }
7048 }
7049
7050 impl Respond for Sequential {
7051 fn respond(&self, _: &Request) -> ResponseTemplate {
7052 let i = self.call.fetch_add(1, std::sync::atomic::Ordering::Relaxed) as usize;
7053 let body = self.pages.get(i).unwrap_or_else(|| {
7054 panic!(
7055 "Sequential: request #{i} has no configured response \
7056 (only {} page(s) supplied)",
7057 self.pages.len()
7058 )
7059 });
7060 ResponseTemplate::new(200).set_body_json(body)
7061 }
7062 }
7063
7064 fn make_activities_json(count: usize, id_prefix: &str) -> serde_json::Value {
7066 serde_json::Value::Array(
7067 (0..count)
7068 .map(|i| {
7069 serde_json::json!({
7070 "id": format!("{id_prefix}-{i:05}"),
7071 "order_id": "ord-1",
7072 "symbol": "SPY",
7073 "side": "buy",
7074 "price": "100.00",
7075 "qty": "1",
7076 "transaction_time": "2025-04-18T14:30:00Z"
7077 })
7078 })
7079 .collect(),
7080 )
7081 }
7082
7083 fn make_orders_json(count: usize) -> serde_json::Value {
7085 serde_json::Value::Array(
7086 (0..count)
7087 .map(|i| {
7088 serde_json::json!({
7089 "id": format!("order-{i:05}"),
7090 "client_order_id": null,
7091 "symbol": "SPY",
7092 "qty": "1",
7093 "filled_qty": "0",
7094 "side": "buy",
7095 "type": "limit",
7096 "time_in_force": "day",
7097 "limit_price": "100.00",
7098 "created_at": "2025-04-18T14:30:00Z"
7099 })
7100 })
7101 .collect(),
7102 )
7103 }
7104
7105 #[tokio::test]
7110 async fn paginate_activities_single_page_below_max_returns_all_not_truncated() {
7111 let server = MockServer::start().await;
7112 Mock::given(method("GET"))
7113 .and(path("/v2/account/activities"))
7114 .respond_with(
7115 ResponseTemplate::new(200).set_body_json(make_activities_json(5, "act")),
7116 )
7117 .mount(&server)
7118 .await;
7119
7120 let http = reqwest::Client::new();
7121 let rl = RateLimitTracker::new();
7122 let result = paginate_activities(&http, &rl, &server.uri(), recovery_after(), None)
7123 .await
7124 .unwrap();
7125
7126 assert_eq!(result.activities.len(), 5);
7127 assert_eq!(result.resume, None);
7128 assert_eq!(server.received_requests().await.unwrap().len(), 1);
7129 }
7130
7131 #[tokio::test]
7134 async fn paginate_activities_exactly_page_size_items_fetches_second_page() {
7135 let server = MockServer::start().await;
7136 Mock::given(method("GET"))
7137 .and(path("/v2/account/activities"))
7138 .respond_with(Sequential::new(vec![
7139 make_activities_json(ALPACA_MAX_ACTIVITIES, "act"),
7140 serde_json::json!([]), ]))
7142 .mount(&server)
7143 .await;
7144
7145 let http = reqwest::Client::new();
7146 let rl = RateLimitTracker::new();
7147 let result = paginate_activities(&http, &rl, &server.uri(), recovery_after(), None)
7148 .await
7149 .unwrap();
7150
7151 assert_eq!(result.activities.len(), ALPACA_MAX_ACTIVITIES);
7152 assert_eq!(result.resume, None);
7153 assert_eq!(
7154 server.received_requests().await.unwrap().len(),
7155 2,
7156 "exactly 2 requests: first full page + second empty page"
7157 );
7158 }
7159
7160 #[tokio::test]
7163 async fn paginate_activities_two_pages_returns_combined_activities_not_truncated() {
7164 let server = MockServer::start().await;
7165 Mock::given(method("GET"))
7166 .and(path("/v2/account/activities"))
7167 .respond_with(Sequential::new(vec![
7168 make_activities_json(ALPACA_MAX_ACTIVITIES, "p1"),
7169 make_activities_json(37, "p2"),
7170 ]))
7171 .mount(&server)
7172 .await;
7173
7174 let http = reqwest::Client::new();
7175 let rl = RateLimitTracker::new();
7176 let result = paginate_activities(&http, &rl, &server.uri(), recovery_after(), None)
7177 .await
7178 .unwrap();
7179
7180 assert_eq!(result.activities.len(), ALPACA_MAX_ACTIVITIES + 37);
7181 assert_eq!(result.resume, None);
7182 assert_eq!(server.received_requests().await.unwrap().len(), 2);
7183 }
7184
7185 #[tokio::test]
7189 async fn paginate_activities_at_max_pages_sets_truncated_true() {
7190 let server = MockServer::start().await;
7191
7192 Mock::given(method("GET"))
7194 .and(path("/v2/account/activities"))
7195 .respond_with(
7196 ResponseTemplate::new(200)
7197 .set_body_json(make_activities_json(ALPACA_MAX_ACTIVITIES, "act")),
7198 )
7199 .mount(&server)
7200 .await;
7201
7202 let http = reqwest::Client::new();
7203 let rl = RateLimitTracker::new();
7204 let result = paginate_activities(&http, &rl, &server.uri(), recovery_after(), None)
7205 .await
7206 .unwrap();
7207
7208 assert!(
7209 result.resume.is_some(),
7210 "must say where to read on after MAX_ACTIVITY_PAGES pages"
7211 );
7212 assert_eq!(
7213 result.activities.len(),
7214 MAX_ACTIVITY_PAGES * ALPACA_MAX_ACTIVITIES,
7215 "must accumulate exactly MAX_ACTIVITY_PAGES * page_size activities"
7216 );
7217 assert_eq!(
7220 server.received_requests().await.unwrap().len(),
7221 MAX_ACTIVITY_PAGES,
7222 "loop must issue exactly MAX_ACTIVITY_PAGES requests then stop"
7223 );
7224 }
7225
7226 #[tokio::test]
7230 async fn fetch_raw_open_orders_499_results_returns_ok() {
7231 let server = MockServer::start().await;
7232 Mock::given(method("GET"))
7233 .and(path("/v2/orders"))
7234 .respond_with(
7235 ResponseTemplate::new(200).set_body_json(make_orders_json(MAX_OPEN_ORDERS - 1)),
7236 )
7237 .mount(&server)
7238 .await;
7239
7240 let http = reqwest::Client::new();
7241 let rl = RateLimitTracker::new();
7242 let result = fetch_raw_open_orders(&http, &rl, &server.uri(), &[]).await;
7243
7244 assert!(
7245 result.is_ok(),
7246 "499 orders must not trigger truncation: {result:?}"
7247 );
7248 assert_eq!(result.unwrap().len(), MAX_OPEN_ORDERS - 1);
7249 }
7250
7251 #[tokio::test]
7255 async fn fetch_raw_open_orders_500_results_returns_truncated_snapshot_error() {
7256 let server = MockServer::start().await;
7257 Mock::given(method("GET"))
7258 .and(path("/v2/orders"))
7259 .respond_with(
7260 ResponseTemplate::new(200).set_body_json(make_orders_json(MAX_OPEN_ORDERS)),
7261 )
7262 .mount(&server)
7263 .await;
7264
7265 let http = reqwest::Client::new();
7266 let rl = RateLimitTracker::new();
7267 let result = fetch_raw_open_orders(&http, &rl, &server.uri(), &[]).await;
7268
7269 assert!(
7270 matches!(
7271 result,
7272 Err(UnindexedClientError::TruncatedSnapshot { limit }) if limit == MAX_OPEN_ORDERS
7273 ),
7274 "500 orders must return TruncatedSnapshot, got: {result:?}"
7275 );
7276 }
7277
7278 async fn mount_account_and_positions(server: &MockServer) {
7283 Mock::given(method("GET"))
7284 .and(path("/v2/account"))
7285 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
7286 "status": "ACTIVE",
7287 "currency": "USD",
7288 "cash": "-1500.50",
7289 "equity": "4000.00",
7290 "buying_power": "5000.00",
7291 "options_buying_power": "2000.00",
7292 "non_marginable_buying_power": "1000.00",
7293 })))
7294 .mount(server)
7295 .await;
7296 Mock::given(method("GET"))
7297 .and(path("/v2/positions"))
7298 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([
7299 {
7300 "symbol": "XYZ",
7301 "asset_class": "us_equity",
7302 "side": "short",
7303 "qty": "-10",
7304 "qty_available": "-10",
7305 "avg_entry_price": "20.00",
7306 "unrealized_pl": "15.50",
7307 },
7308 {
7309 "symbol": "XYZ271217C00025000",
7310 "asset_class": "us_option",
7311 "side": "long",
7312 "qty": "3",
7313 "qty_available": "3",
7314 "avg_entry_price": "0.85",
7315 "unrealized_pl": "-45",
7316 },
7317 {
7318 "symbol": "ABC",
7319 "asset_class": "us_equity",
7320 "side": "long",
7321 "qty": "2.5",
7322 "qty_available": "2.5",
7323 "avg_entry_price": "40.00",
7324 "unrealized_pl": "1.25",
7325 },
7326 {
7327 "symbol": "BTCUSD",
7328 "asset_class": "crypto",
7329 "side": "long",
7330 "qty": "0.75",
7331 "qty_available": "0.5",
7332 "avg_entry_price": "50000",
7333 "unrealized_pl": "0",
7334 },
7335 ])))
7336 .mount(server)
7337 .await;
7338 Mock::given(method("GET"))
7339 .and(path("/v2/orders"))
7340 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([])))
7341 .mount(server)
7342 .await;
7343 }
7344
7345 fn client_for(server: &MockServer) -> AlpacaClient {
7346 AlpacaClient::new(AlpacaConfig::with_base_url(
7347 "test-key".into(),
7348 "test-secret".into(),
7349 server.uri(),
7350 ))
7351 }
7352
7353 #[tokio::test]
7354 async fn account_snapshot_reports_cash_positions_and_crypto_balances() {
7355 use crate::client::ExecutionClient;
7356 use rust_decimal_macros::dec;
7357
7358 let server = MockServer::start().await;
7359 mount_account_and_positions(&server).await;
7360
7361 let snapshot = client_for(&server)
7362 .account_snapshot(&[], &[])
7363 .await
7364 .unwrap();
7365
7366 let balances: Vec<_> = snapshot
7368 .balances
7369 .iter()
7370 .map(|b| (b.asset.name().as_str(), b.balance.total, b.balance.free))
7371 .collect();
7372 assert_eq!(
7373 balances,
7374 vec![
7375 ("USD", dec!(-1500.50), dec!(-1500.50)),
7376 ("BTC", dec!(0.75), dec!(0.5)),
7377 ],
7378 "crypto stays a balance; equities and options do not appear as balances"
7379 );
7380
7381 let positions: Vec<_> = snapshot
7382 .instruments
7383 .iter()
7384 .map(|i| {
7385 let p = i.position.open().unwrap();
7386 (
7387 i.instrument.name().as_str(),
7388 p.quantity,
7389 p.entry_price,
7390 p.unrealized_pnl,
7391 )
7392 })
7393 .collect();
7394 assert_eq!(
7395 positions,
7396 vec![
7397 ("XYZ", dec!(-10), Some(dec!(20.00)), Some(dec!(15.50))),
7398 (
7399 "XYZ271217C00025000",
7400 dec!(3),
7401 Some(dec!(0.85)),
7402 Some(dec!(-45))
7403 ),
7404 ("ABC", dec!(2.5), Some(dec!(40.00)), Some(dec!(1.25))),
7405 ],
7406 "short equity, option and fractional positions; no crypto position"
7407 );
7408 }
7409
7410 #[tokio::test]
7413 async fn account_snapshot_usd_only_still_reports_positions() {
7414 use crate::client::ExecutionClient;
7415 use rust_decimal_macros::dec;
7416
7417 let server = MockServer::start().await;
7418 mount_account_and_positions(&server).await;
7419
7420 let usd = [AssetNameExchange::new("USD")];
7421 let instruments = [InstrumentNameExchange::new("XYZ")];
7422 let snapshot = client_for(&server)
7423 .account_snapshot(&usd, &instruments)
7424 .await
7425 .unwrap();
7426
7427 assert_eq!(snapshot.balances.len(), 1);
7428 assert_eq!(snapshot.balances[0].asset, AssetNameExchange::new("USD"));
7429 assert_eq!(snapshot.balances[0].balance.total, dec!(-1500.50));
7430
7431 assert_eq!(snapshot.instruments.len(), 1);
7432 assert_eq!(snapshot.instruments[0].position.quantity(), Some(dec!(-10)));
7433 }
7434
7435 #[tokio::test]
7437 async fn account_snapshot_non_usd_only_skips_account_request() {
7438 use crate::client::ExecutionClient;
7439
7440 let server = MockServer::start().await;
7441 mount_account_and_positions(&server).await;
7442
7443 let btc = [AssetNameExchange::new("btc")];
7444 let snapshot = client_for(&server)
7445 .account_snapshot(&btc, &[])
7446 .await
7447 .unwrap();
7448
7449 assert_eq!(snapshot.balances.len(), 1);
7450 assert_eq!(snapshot.balances[0].asset, AssetNameExchange::new("BTC"));
7451 let requests = server.received_requests().await.unwrap();
7452 assert!(
7453 requests.iter().all(|r| r.url.path() != "/v2/account"),
7454 "no USD requested, so /v2/account must not be fetched"
7455 );
7456 }
7457
7458 #[tokio::test]
7466 async fn open_order_reduce_only_sell_sends_sell_to_close_intent() {
7467 use crate::client::ExecutionClient;
7468 use crate::order::request::{OrderRequestOpen, RequestOpen};
7469 use crate::order::{
7470 OrderKey, OrderKind, TimeInForce,
7471 id::{ClientOrderId, StrategyId},
7472 };
7473 use rust_decimal::Decimal;
7474 use rustrade_instrument::Side;
7475 use rustrade_instrument::exchange::ExchangeId;
7476 use rustrade_instrument::instrument::name::InstrumentNameExchange;
7477 use wiremock::matchers::{method, path};
7478
7479 let server = MockServer::start().await;
7480
7481 let captured_body = std::sync::Arc::new(parking_lot::Mutex::new(None));
7484 let captured_clone = captured_body.clone();
7485
7486 Mock::given(method("POST"))
7487 .and(path("/v2/orders"))
7488 .respond_with(move |req: &Request| {
7489 let body: serde_json::Value = serde_json::from_slice(&req.body).unwrap();
7491 *captured_clone.lock() = Some(body);
7492
7493 ResponseTemplate::new(200).set_body_json(serde_json::json!({
7494 "id": "test-order-id",
7495 "client_order_id": "test-cid",
7496 "symbol": "AAPL",
7497 "qty": "10",
7498 "filled_qty": "0",
7499 "side": "sell",
7500 "type": "market",
7501 "time_in_force": "ioc",
7502 "limit_price": null,
7503 "created_at": "2025-04-18T14:30:00Z"
7504 }))
7505 })
7506 .mount(&server)
7507 .await;
7508
7509 let config =
7511 AlpacaConfig::with_base_url("test-key".into(), "test-secret".into(), server.uri());
7512 let client = AlpacaClient::new(config);
7513
7514 let request = OrderRequestOpen {
7516 key: OrderKey {
7517 exchange: ExchangeId::AlpacaBroker,
7518 instrument: InstrumentNameExchange::new("AAPL"),
7519 strategy: StrategyId::new("test-strategy"),
7520 cid: ClientOrderId::new("test-cid"),
7521 },
7522 state: RequestOpen {
7523 side: Side::Sell,
7524 price: None,
7525 quantity: Decimal::new(10, 0),
7526 kind: OrderKind::Market,
7527 time_in_force: TimeInForce::ImmediateOrCancel,
7528 position_id: None,
7529 reduce_only: true, market: None,
7531 },
7532 };
7533
7534 let order = client
7536 .open_order(OrderRequestOpen {
7537 key: OrderKey {
7538 exchange: request.key.exchange,
7539 instrument: &request.key.instrument,
7540 strategy: request.key.strategy.clone(),
7541 cid: request.key.cid.clone(),
7542 },
7543 state: request.state.clone(),
7544 })
7545 .await;
7546
7547 assert!(
7549 order.state.is_accepted(),
7550 "order should be accepted: {:?}",
7551 order.state
7552 );
7553
7554 let body = captured_body
7556 .lock()
7557 .take()
7558 .expect("request body should be captured");
7559 assert_eq!(
7560 body.get("position_intent").and_then(|v| v.as_str()),
7561 Some("sell_to_close"),
7562 "reduce_only=true + Side::Sell should produce position_intent=sell_to_close, got: {body}"
7563 );
7564 }
7565
7566 #[tokio::test]
7568 async fn open_order_not_reduce_only_buy_sends_buy_to_open_intent() {
7569 use crate::client::ExecutionClient;
7570 use crate::order::request::{OrderRequestOpen, RequestOpen};
7571 use crate::order::{
7572 OrderKey, OrderKind, TimeInForce,
7573 id::{ClientOrderId, StrategyId},
7574 };
7575 use rust_decimal::Decimal;
7576 use rustrade_instrument::Side;
7577 use rustrade_instrument::exchange::ExchangeId;
7578 use rustrade_instrument::instrument::name::InstrumentNameExchange;
7579
7580 let server = MockServer::start().await;
7581
7582 let captured_body = std::sync::Arc::new(parking_lot::Mutex::new(None));
7583 let captured_clone = captured_body.clone();
7584
7585 Mock::given(method("POST"))
7586 .and(path("/v2/orders"))
7587 .respond_with(move |req: &Request| {
7588 let body: serde_json::Value = serde_json::from_slice(&req.body).unwrap();
7589 *captured_clone.lock() = Some(body);
7590
7591 ResponseTemplate::new(200).set_body_json(serde_json::json!({
7592 "id": "test-order-id",
7593 "client_order_id": "test-cid",
7594 "symbol": "AAPL",
7595 "qty": "10",
7596 "filled_qty": "0",
7597 "side": "buy",
7598 "type": "market",
7599 "time_in_force": "ioc",
7600 "limit_price": null,
7601 "created_at": "2025-04-18T14:30:00Z"
7602 }))
7603 })
7604 .mount(&server)
7605 .await;
7606
7607 let config =
7608 AlpacaConfig::with_base_url("test-key".into(), "test-secret".into(), server.uri());
7609 let client = AlpacaClient::new(config);
7610
7611 let instrument = InstrumentNameExchange::new("AAPL");
7612 let request = OrderRequestOpen {
7613 key: OrderKey {
7614 exchange: ExchangeId::AlpacaBroker,
7615 instrument: &instrument,
7616 strategy: StrategyId::new("test-strategy"),
7617 cid: ClientOrderId::new("test-cid"),
7618 },
7619 state: RequestOpen {
7620 side: Side::Buy,
7621 price: None,
7622 quantity: Decimal::new(10, 0),
7623 kind: OrderKind::Market,
7624 time_in_force: TimeInForce::ImmediateOrCancel,
7625 position_id: None,
7626 reduce_only: false, market: None,
7628 },
7629 };
7630
7631 let order = client.open_order(request).await;
7632
7633 assert!(
7634 order.state.is_accepted(),
7635 "order should be accepted: {:?}",
7636 order.state
7637 );
7638
7639 let body = captured_body
7640 .lock()
7641 .take()
7642 .expect("request body should be captured");
7643 assert_eq!(
7644 body.get("position_intent").and_then(|v| v.as_str()),
7645 Some("buy_to_open"),
7646 "reduce_only=false + Side::Buy should produce position_intent=buy_to_open, got: {body}"
7647 );
7648 }
7649
7650 #[tokio::test]
7653 async fn rest_with_retry_reports_an_undecodable_2xx_body_as_internal() {
7654 let server = MockServer::start().await;
7655 Mock::given(method("GET"))
7656 .and(path("/v2/account"))
7657 .respond_with(ResponseTemplate::new(200).set_body_string("not json"))
7658 .mount(&server)
7659 .await;
7660
7661 let http = reqwest::Client::new();
7662 let rl = RateLimitTracker::new();
7663 let url = format!("{}/v2/account", server.uri());
7664 let result: Result<serde_json::Value, _> =
7665 rest_with_retry(&rl, || http.get(&url)).await;
7666 assert!(
7667 matches!(result, Err(UnindexedClientError::Internal(_))),
7668 "{result:?}"
7669 );
7670 }
7671
7672 async fn holds_back(rl: &RateLimitTracker) -> bool {
7674 tokio::time::timeout(Duration::from_millis(50), rl.wait_if_blocked())
7675 .await
7676 .is_err()
7677 }
7678
7679 fn with_rate_limit(
7681 template: ResponseTemplate,
7682 remaining: u32,
7683 reset: bool,
7684 ) -> ResponseTemplate {
7685 let template = template.insert_header("x-ratelimit-remaining", remaining.to_string());
7686 if !reset {
7687 return template;
7688 }
7689 let now_secs = std::time::SystemTime::now()
7690 .duration_since(std::time::UNIX_EPOCH)
7691 .unwrap()
7692 .as_secs();
7693 template.insert_header("x-ratelimit-reset", (now_secs + 30).to_string())
7694 }
7695
7696 #[tokio::test]
7699 async fn rest_with_retry_pauses_once_no_requests_remain() {
7700 for (remaining, reset, pauses) in [(0, true, true), (1, true, false), (0, false, false)]
7701 {
7702 let server = MockServer::start().await;
7703 Mock::given(method("GET"))
7704 .and(path("/v2/account"))
7705 .respond_with(with_rate_limit(
7706 ResponseTemplate::new(200).set_body_json(serde_json::json!({})),
7707 remaining,
7708 reset,
7709 ))
7710 .mount(&server)
7711 .await;
7712
7713 let http = reqwest::Client::new();
7714 let rl = RateLimitTracker::new();
7715 let url = format!("{}/v2/account", server.uri());
7716 let _: serde_json::Value = rest_with_retry(&rl, || http.get(&url)).await.unwrap();
7717 assert_eq!(
7718 holds_back(&rl).await,
7719 pauses,
7720 "remaining {remaining}, reset {reset}"
7721 );
7722 }
7723 }
7724
7725 #[tokio::test]
7728 async fn a_pause_lasts_a_minute_at_most() {
7729 tokio::time::pause();
7730 let now_secs = std::time::SystemTime::now()
7731 .duration_since(std::time::UNIX_EPOCH)
7732 .unwrap()
7733 .as_secs();
7734 let mut headers = reqwest::header::HeaderMap::new();
7735 headers.insert("x-ratelimit-remaining", "0".parse().unwrap());
7736 headers.insert(
7737 "x-ratelimit-reset",
7738 (now_secs + 3_600).to_string().parse().unwrap(),
7739 );
7740 let rl = RateLimitTracker::new();
7741
7742 observe_rate_limit_remaining(&rl, &headers);
7743 assert!(holds_back(&rl).await);
7744 tokio::time::advance(Duration::from_secs(DEFAULT_RATE_LIMIT_DELAY_SECS)).await;
7745 assert!(!holds_back(&rl).await);
7746 }
7747
7748 #[tokio::test]
7750 async fn rest_delete_with_retry_pauses_once_no_requests_remain() {
7751 let server = MockServer::start().await;
7752 Mock::given(method("DELETE"))
7753 .and(path("/v2/orders/abc"))
7754 .respond_with(with_rate_limit(ResponseTemplate::new(204), 0, true))
7755 .mount(&server)
7756 .await;
7757
7758 let http = reqwest::Client::new();
7759 let rl = RateLimitTracker::new();
7760 let url = format!("{}/v2/orders/abc", server.uri());
7761 rest_delete_with_retry(&rl, || http.delete(&url))
7762 .await
7763 .unwrap();
7764 assert!(holds_back(&rl).await);
7765 }
7766
7767 #[tokio::test]
7770 async fn open_order_with_an_undecodable_2xx_response_is_status_unknown() {
7771 use crate::client::ExecutionClient;
7772 use crate::error::OrderError;
7773 use crate::order::request::{OrderRequestOpen, RequestOpen};
7774 use crate::order::state::{InactiveOrderState, OrderState};
7775 use crate::order::{
7776 OrderKey, OrderKind, TimeInForce,
7777 id::{ClientOrderId, StrategyId},
7778 };
7779 use rustrade_instrument::Side;
7780 use rustrade_instrument::exchange::ExchangeId;
7781 use rustrade_instrument::instrument::name::InstrumentNameExchange;
7782
7783 let server = MockServer::start().await;
7784 Mock::given(method("POST"))
7785 .and(path("/v2/orders"))
7786 .respond_with(
7787 ResponseTemplate::new(200).set_body_json(serde_json::json!({ "id": 1 })),
7788 )
7789 .mount(&server)
7790 .await;
7791
7792 let instrument = InstrumentNameExchange::new("AAPL");
7793 let order = client_for(&server)
7794 .open_order(OrderRequestOpen {
7795 key: OrderKey {
7796 exchange: ExchangeId::AlpacaBroker,
7797 instrument: &instrument,
7798 strategy: StrategyId::new("test-strategy"),
7799 cid: ClientOrderId::new("test-cid"),
7800 },
7801 state: RequestOpen {
7802 side: Side::Buy,
7803 price: None,
7804 quantity: Decimal::new(10, 0),
7805 kind: OrderKind::Market,
7806 time_in_force: TimeInForce::ImmediateOrCancel,
7807 position_id: None,
7808 reduce_only: false,
7809 market: None,
7810 },
7811 })
7812 .await;
7813
7814 assert!(
7815 matches!(
7816 order.state,
7817 OrderState::Inactive(InactiveOrderState::OpenFailed(OrderError::Connectivity(
7818 _
7819 )))
7820 ),
7821 "{:?}",
7822 order.state
7823 );
7824 }
7825
7826 #[tokio::test]
7829 async fn open_bracket_order_with_an_undecodable_2xx_response_is_status_unknown() {
7830 use crate::error::OrderError;
7831 use crate::order::state::{InactiveOrderState, OrderState};
7832 use rustrade_instrument::instrument::name::InstrumentNameExchange;
7833
7834 let server = MockServer::start().await;
7835 Mock::given(method("POST"))
7836 .and(path("/v2/orders"))
7837 .respond_with(
7838 ResponseTemplate::new(200).set_body_json(serde_json::json!({ "id": 1 })),
7839 )
7840 .mount(&server)
7841 .await;
7842
7843 let result = client_for(&server)
7844 .open_bracket_order(AlpacaBracketOrderRequest::new(
7845 InstrumentNameExchange::new("SPY"),
7846 crate::order::id::StrategyId::new("test"),
7847 crate::order::id::ClientOrderId::new("test-bracket"),
7848 Side::Buy,
7849 Decimal::ONE,
7850 Decimal::new(100, 0),
7851 Decimal::new(120, 0),
7852 Decimal::new(90, 0),
7853 TimeInForce::GoodUntilCancelled { post_only: false },
7854 ))
7855 .await;
7856
7857 assert!(
7858 matches!(
7859 result.parent.state,
7860 OrderState::Inactive(InactiveOrderState::OpenFailed(OrderError::Connectivity(
7861 _
7862 )))
7863 ),
7864 "{:?}",
7865 result.parent.state
7866 );
7867 }
7868
7869 fn recovery_after() -> DateTime<Utc> {
7878 "2025-01-01T00:00:00Z".parse().expect("valid time")
7879 }
7880
7881 fn activity_json(
7884 id: &str,
7885 order_id: &str,
7886 qty: &str,
7887 cum_qty: Option<&str>,
7888 ) -> serde_json::Value {
7889 let mut v = serde_json::json!({
7890 "id": id,
7891 "order_id": order_id,
7892 "symbol": "SPY",
7893 "side": "buy",
7894 "price": "100.00",
7895 "qty": qty,
7896 "transaction_time": "2025-04-18T14:30:00Z"
7897 });
7898 if let Some(c) = cum_qty {
7899 v["cum_qty"] = serde_json::Value::String(c.to_string());
7900 }
7901 v
7902 }
7903
7904 async fn drive_recover_fills_with(
7907 activities: Vec<serde_json::Value>,
7908 dedup: SharedDedupCache,
7909 known: &SharedKnownLiveOrders,
7910 ) -> Vec<UnindexedAccountEvent> {
7911 let server = MockServer::start().await;
7912 Mock::given(method("GET"))
7913 .and(path("/v2/account/activities"))
7914 .respond_with(
7915 ResponseTemplate::new(200).set_body_json(serde_json::Value::Array(activities)),
7916 )
7917 .mount(&server)
7918 .await;
7919
7920 let http = reqwest::Client::new();
7921 let rl = RateLimitTracker::new();
7922 let (tx, mut rx) = mpsc::unbounded_channel();
7923 let outcome = recover_fills(
7924 &http,
7925 &rl,
7926 &[],
7927 &server.uri(),
7928 recovery_after(),
7929 &tx,
7930 &dedup,
7931 known,
7932 )
7933 .await;
7934 assert_eq!(outcome, Ok(()), "a full read leaves nothing unrecovered");
7935 drop(tx);
7936
7937 let mut out = Vec::new();
7938 while let Ok(event) = rx.try_recv() {
7939 out.push(event);
7940 }
7941 out
7942 }
7943
7944 async fn drive_recover_fills(
7945 activities: Vec<serde_json::Value>,
7946 ) -> Vec<UnindexedAccountEvent> {
7947 drive_recover_fills_with(
7948 activities,
7949 new_dedup_cache(),
7950 &KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
7951 )
7952 .await
7953 }
7954
7955 async fn recover_fills_from(
7958 server: &MockServer,
7959 ) -> (Result<(), UnreadFills>, Vec<UnindexedAccountEvent>) {
7960 let (tx, mut rx) = mpsc::unbounded_channel();
7961 let outcome = recover_fills(
7962 &reqwest::Client::new(),
7963 &RateLimitTracker::new(),
7964 &[],
7965 &server.uri(),
7966 recovery_after(),
7967 &tx,
7968 &new_dedup_cache(),
7969 &KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
7970 )
7971 .await;
7972 drop(tx);
7973 (outcome, std::iter::from_fn(|| rx.try_recv().ok()).collect())
7974 }
7975
7976 #[tokio::test]
7979 async fn a_failed_recovery_read_leaves_every_fill_since_the_disconnect_unread() {
7980 let server = MockServer::start().await;
7981 Mock::given(method("GET"))
7982 .and(path("/v2/account/activities"))
7983 .respond_with(ResponseTemplate::new(500).set_body_string("unavailable"))
7984 .mount(&server)
7985 .await;
7986
7987 let (outcome, events) = recover_fills_from(&server).await;
7988
7989 let Err(UnreadFills {
7990 start,
7991 reason: FillRecoveryFailure::Request(_),
7992 }) = outcome
7993 else {
7994 panic!("expected the request's failure, got {outcome:?}");
7995 };
7996 assert_eq!(start, recovery_after(), "from the disconnect");
7997 assert!(events.is_empty(), "nothing forwarded: {events:?}");
7998 }
7999
8000 #[tokio::test]
8005 async fn a_truncated_recovery_read_delivers_what_it_read_and_leaves_the_rest_unread() {
8006 let time = |s: &str| s.parse::<DateTime<Utc>>().expect("valid time");
8007 let cases = [
8009 (
8010 Some("2025-04-18T14:30:01.234567Z"),
8011 time("2025-04-18T14:30:01.234Z"),
8012 ),
8013 (Some("not a time"), time("2025-04-18T14:30:00Z")),
8014 (None, recovery_after()),
8015 ];
8016 for (last_time, expected_start) in cases {
8017 let mut page = make_activities_json(ALPACA_MAX_ACTIVITIES, "act");
8018 let Some(activities) = page.as_array_mut() else {
8019 panic!("an array of activities");
8020 };
8021 match last_time {
8022 Some(last_time) => {
8023 if let Some(last) = activities.last_mut() {
8024 last["transaction_time"] =
8025 serde_json::Value::String(last_time.to_string());
8026 }
8027 }
8028 None => {
8029 for activity in activities.iter_mut() {
8030 activity["transaction_time"] =
8031 serde_json::Value::String("not a time".to_string());
8032 }
8033 }
8034 }
8035 let server = MockServer::start().await;
8036 Mock::given(method("GET"))
8037 .and(path("/v2/account/activities"))
8038 .respond_with(ResponseTemplate::new(200).set_body_json(page))
8039 .mount(&server)
8040 .await;
8041
8042 let (outcome, events) = recover_fills_from(&server).await;
8043
8044 let fills_read = MAX_ACTIVITY_PAGES * ALPACA_MAX_ACTIVITIES;
8045 assert_eq!(
8046 outcome,
8047 Err(UnreadFills {
8048 start: expected_start,
8049 reason: FillRecoveryFailure::Truncated { fills_read },
8050 }),
8051 "last time {last_time:?}"
8052 );
8053 assert_eq!(events.len(), fills_read, "every fill read is forwarded");
8054 }
8055 }
8056
8057 #[test]
8060 fn a_fill_recovery_give_up_covers_every_instrument_the_stream_does() {
8061 let start = recovery_after();
8062 let end = start + chrono::Duration::minutes(5);
8063 let spy = InstrumentNameExchange::new("SPY");
8064 let qqq = InstrumentNameExchange::new("QQQ");
8065 let cases = [
8066 (vec![], FillRecoveryScope::AllInstruments),
8067 (
8068 vec![spy.clone(), qqq.clone()],
8069 FillRecoveryScope::Instruments(vec![spy, qqq]),
8070 ),
8071 ];
8072 for (instruments, scope) in cases {
8073 let reason = FillRecoveryFailure::TimedOut { timeout_secs: 30 };
8074 let unread = UnreadFills {
8075 start,
8076 reason: reason.clone(),
8077 };
8078 assert_eq!(
8079 fill_recovery_gave_up(&instruments, unread, end),
8080 UnindexedAccountEvent::new(
8081 ExchangeId::AlpacaBroker,
8082 AccountEventKind::FillRecoveryGaveUp(FillRecoveryGap::new(
8083 scope, start, end, 1, reason,
8084 )),
8085 )
8086 );
8087 }
8088 }
8089
8090 #[test]
8094 fn a_read_that_reached_past_the_recovery_start_still_reports() {
8095 let end = recovery_after();
8096 let start = end + chrono::Duration::seconds(1);
8097 let unread = UnreadFills {
8098 start,
8099 reason: FillRecoveryFailure::Truncated { fills_read: 5_000 },
8100 };
8101 let event = fill_recovery_gave_up(&[], unread, end);
8102 let AccountEventKind::FillRecoveryGaveUp(gave_up) = event.kind else {
8103 panic!("expected FillRecoveryGaveUp, got {event:?}");
8104 };
8105 assert_eq!((gave_up.start, gave_up.end), (start, start));
8106 }
8107
8108 async fn recover_fills_or_report_from(
8111 server: &MockServer,
8112 instruments: &[InstrumentNameExchange],
8113 timeout: Duration,
8114 ) -> Vec<UnindexedAccountEvent> {
8115 let (tx, mut rx) = mpsc::unbounded_channel();
8116 recover_fills_or_report(
8117 &reqwest::Client::new(),
8118 &RateLimitTracker::new(),
8119 instruments,
8120 &server.uri(),
8121 recovery_after(),
8122 timeout,
8123 &tx,
8124 &new_dedup_cache(),
8125 &KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
8126 )
8127 .await;
8128 drop(tx);
8129 std::iter::from_fn(|| rx.try_recv().ok()).collect()
8130 }
8131
8132 #[tokio::test]
8135 async fn a_timed_out_recovery_reports_every_fill_since_the_disconnect() {
8136 let server = MockServer::start().await;
8137 Mock::given(method("GET"))
8138 .and(path("/v2/account/activities"))
8139 .respond_with(
8140 ResponseTemplate::new(200)
8141 .set_body_json(serde_json::json!([]))
8142 .set_delay(Duration::from_secs(3_600)),
8143 )
8144 .mount(&server)
8145 .await;
8146 let spy = InstrumentNameExchange::new("SPY");
8147
8148 let events = recover_fills_or_report_from(
8149 &server,
8150 std::slice::from_ref(&spy),
8151 Duration::from_millis(50),
8152 )
8153 .await;
8154
8155 let [event] = events.as_slice() else {
8156 panic!("only the give-up is sent: {events:?}");
8157 };
8158 let AccountEventKind::FillRecoveryGaveUp(gave_up) = &event.kind else {
8159 panic!("expected FillRecoveryGaveUp, got {event:?}");
8160 };
8161 assert_eq!(gave_up.scope, FillRecoveryScope::Instruments(vec![spy]));
8162 assert_eq!(gave_up.start, recovery_after(), "from the disconnect");
8163 assert!(gave_up.end >= gave_up.start);
8164 assert_eq!(gave_up.attempts, 1);
8165 assert!(
8166 matches!(gave_up.reason, FillRecoveryFailure::TimedOut { .. }),
8167 "{:?}",
8168 gave_up.reason
8169 );
8170 }
8171
8172 #[tokio::test]
8175 async fn a_truncated_recovery_reports_after_the_fills_it_read() {
8176 let server = MockServer::start().await;
8177 Mock::given(method("GET"))
8178 .and(path("/v2/account/activities"))
8179 .respond_with(
8180 ResponseTemplate::new(200)
8181 .set_body_json(make_activities_json(ALPACA_MAX_ACTIVITIES, "act")),
8182 )
8183 .mount(&server)
8184 .await;
8185
8186 let events = recover_fills_or_report_from(&server, &[], Duration::from_secs(30)).await;
8187
8188 let Some((last, fills)) = events.split_last() else {
8189 panic!("events are sent");
8190 };
8191 assert_eq!(fills.len(), MAX_ACTIVITY_PAGES * ALPACA_MAX_ACTIVITIES);
8192 assert!(
8193 fills
8194 .iter()
8195 .all(|event| matches!(event.kind, AccountEventKind::Trade(_))),
8196 "every fill comes first"
8197 );
8198 let AccountEventKind::FillRecoveryGaveUp(gave_up) = &last.kind else {
8199 panic!("the give-up comes last, got {last:?}");
8200 };
8201 assert_eq!(gave_up.scope, FillRecoveryScope::AllInstruments);
8202 assert!(matches!(
8203 gave_up.reason,
8204 FillRecoveryFailure::Truncated { .. }
8205 ));
8206 }
8207
8208 #[test]
8211 fn an_activity_id_names_its_execution_after_the_separator() {
8212 assert_eq!(
8213 activity_execution_id("20261005041849348::524b1902-817e-446c-825b-a9fcfebbc17e"),
8214 "524b1902-817e-446c-825b-a9fcfebbc17e"
8215 );
8216 assert_eq!(activity_execution_id("act-1"), "act-1");
8217 assert_eq!(
8218 activity_execution_id("20261005041849348::"),
8219 "20261005041849348::"
8220 );
8221 }
8222
8223 #[tokio::test]
8227 async fn a_fill_has_one_trade_id_on_the_stream_in_recovery_and_from_fetch_trades() {
8228 use crate::client::ExecutionClient;
8229
8230 let execution_id = "524b1902-817e-446c-825b-a9fcfebbc17e";
8231 let activity = activity_json(
8232 &format!("20250418103000000::{execution_id}"),
8233 "ord-1",
8234 "2",
8235 Some("2"),
8236 );
8237
8238 let [streamed, _snapshot] = convert_trade_update(AlpacaTradeUpdate {
8239 event: SmolStr::new("fill"),
8240 execution_id: Some(execution_id),
8241 previous_execution_id: None,
8242 order: super::make_order_ws("ord-1", "SPY", "buy", "2"),
8243 price: Some("100.00"),
8244 qty: Some("2"),
8245 timestamp: Some("2025-04-18T14:30:00Z"),
8246 });
8247 let Some(streamed) = streamed else {
8248 panic!("a fill converts to a Trade");
8249 };
8250
8251 let recovered = drive_recover_fills(vec![activity.clone()]).await;
8252 let [recovered] = recovered.as_slice() else {
8253 panic!("one fill recovered: {recovered:?}");
8254 };
8255
8256 let server = MockServer::start().await;
8257 Mock::given(method("GET"))
8258 .and(path("/v2/account/activities"))
8259 .respond_with(
8260 ResponseTemplate::new(200)
8261 .set_body_json(serde_json::Value::Array(vec![activity])),
8262 )
8263 .mount(&server)
8264 .await;
8265 let Ok(fetched) = client_for(&server)
8266 .fetch_trades(recovery_after(), Utc::now(), &[])
8267 .await
8268 else {
8269 panic!("fetch_trades reads the activity");
8270 };
8271 let [fetched] = fetched.trades.as_slice() else {
8272 panic!("one fill fetched: {fetched:?}");
8273 };
8274
8275 assert_eq!(trade_of(&streamed).id.0.as_str(), execution_id);
8276 assert_eq!(trade_of(recovered).id.0.as_str(), execution_id);
8277 assert_eq!(fetched.id.0.as_str(), execution_id);
8278 }
8279
8280 fn time(s: &str) -> DateTime<Utc> {
8285 s.parse().expect("valid time")
8286 }
8287
8288 struct ActivitiesVenue {
8293 activities: Vec<serde_json::Value>,
8294 }
8295
8296 impl Respond for ActivitiesVenue {
8297 fn respond(&self, request: &Request) -> ResponseTemplate {
8298 let query: std::collections::HashMap<String, String> =
8299 request.url.query_pairs().into_owned().collect();
8300 let bound = |name: &str| query.get(name).map(|s| time(s));
8301 let (after, until) = (bound("after"), bound("until"));
8302 let size: usize = query
8303 .get("page_size")
8304 .and_then(|s| s.parse().ok())
8305 .unwrap_or(ALPACA_MAX_ACTIVITIES);
8306 let in_bounds = self.activities.iter().filter(|activity| {
8307 let at =
8308 floor_millis(time(activity["transaction_time"].as_str().expect("a time")));
8309 after.is_none_or(|after| at >= after) && until.is_none_or(|until| at < until)
8310 });
8311 let page: Vec<_> = match query.get("page_token") {
8312 Some(token) => in_bounds
8313 .skip_while(|activity| activity["id"] != token.as_str())
8314 .skip(1)
8315 .take(size)
8316 .cloned()
8317 .collect(),
8318 None => in_bounds.take(size).cloned().collect(),
8319 };
8320 ResponseTemplate::new(200).set_body_json(page)
8321 }
8322 }
8323
8324 fn fill_at(id: &str, symbol: &str, transaction_time: DateTime<Utc>) -> serde_json::Value {
8326 serde_json::json!({
8327 "id": id,
8328 "order_id": format!("ord-{id}"),
8329 "symbol": symbol,
8330 "side": "buy",
8331 "price": "100.00",
8332 "qty": "1",
8333 "cum_qty": "1",
8334 "transaction_time": transaction_time.to_rfc3339_opts(SecondsFormat::Micros, true),
8335 })
8336 }
8337
8338 async fn serve_activities(activities: Vec<serde_json::Value>) -> MockServer {
8339 let server = MockServer::start().await;
8340 Mock::given(method("GET"))
8341 .and(path("/v2/account/activities"))
8342 .respond_with(ActivitiesVenue { activities })
8343 .mount(&server)
8344 .await;
8345 server
8346 }
8347
8348 async fn read_span(
8351 client: &AlpacaClient,
8352 start: DateTime<Utc>,
8353 end: DateTime<Utc>,
8354 instruments: &[InstrumentNameExchange],
8355 ) -> (std::collections::HashSet<String>, usize) {
8356 use crate::client::ExecutionClient;
8357
8358 let mut ids = std::collections::HashSet::new();
8359 let mut from = start;
8360 for calls in 1..=10 {
8361 let read = match client.fetch_trades(from, end, instruments).await {
8362 Ok(read) => read,
8363 Err(error) => panic!("call {calls} failed: {error:?}"),
8364 };
8365 ids.extend(read.trades.iter().map(|trade| trade.id.0.to_string()));
8366 match read.resume {
8367 Some(resume) => from = resume,
8368 None => return (ids, calls),
8369 }
8370 }
8371 panic!("the reads did not reach the span's end");
8372 }
8373
8374 #[tokio::test]
8380 async fn fetch_trades_reads_a_busy_span_completely_across_calls() {
8381 let base = time("2025-04-18T14:30:00Z");
8382 let mut activities = Vec::new();
8383 for ms in 0..4_000_i64 {
8384 let symbol = if ms % 2 == 0 { "SPY" } else { "QQQ" };
8385 for (k, micros) in [("a", 900), ("b", 500), ("c", 100)] {
8386 let at = base + TimeDelta::milliseconds(ms) + TimeDelta::microseconds(micros);
8387 activities.push(fill_at(&format!("{ms:017}::{ms}-{k}"), symbol, at));
8388 }
8389 }
8390 let server = serve_activities(activities).await;
8391 let client = client_for(&server);
8392 let end = time("2025-04-18T15:00:00Z");
8393
8394 let (ids, calls) = read_span(&client, recovery_after(), end, &[]).await;
8395 assert_eq!(ids.len(), 12_000, "every fill is read once or more");
8396 assert_eq!(calls, 3, "5,000 activities per call");
8397
8398 let spy = [InstrumentNameExchange::new("SPY")];
8399 let (ids, _) = read_span(&client, recovery_after(), end, &spy).await;
8400 assert_eq!(ids.len(), 6_000, "every SPY fill, and only those");
8401
8402 let Some(requests) = server.received_requests().await else {
8403 panic!("requests are recorded");
8404 };
8405 assert!(
8406 requests
8407 .iter()
8408 .all(|request| request.url.query_pairs().any(
8409 |(name, value)| name == "until" && value == "2025-04-18T15:00:00.001Z"
8410 )),
8411 "every request is bounded by the span's end"
8412 );
8413 }
8414
8415 #[tokio::test]
8418 async fn fetch_trades_says_where_to_read_on_after_a_call_without_matches() {
8419 use crate::client::ExecutionClient;
8420
8421 let server = MockServer::start().await;
8422 Mock::given(method("GET"))
8423 .and(path("/v2/account/activities"))
8424 .respond_with(
8425 ResponseTemplate::new(200)
8426 .set_body_json(make_activities_json(ALPACA_MAX_ACTIVITIES, "act")),
8427 )
8428 .mount(&server)
8429 .await;
8430
8431 let qqq = [InstrumentNameExchange::new("QQQ")];
8432 let Ok(read) = client_for(&server)
8433 .fetch_trades(recovery_after(), Utc::now(), &qqq)
8434 .await
8435 else {
8436 panic!("the read succeeds");
8437 };
8438 assert!(read.trades.is_empty(), "every fill read is SPY's");
8439 assert_eq!(read.resume, Some(time("2025-04-18T14:30:00Z")));
8440 }
8441
8442 #[tokio::test]
8445 async fn fetch_trades_that_cannot_advance_is_truncated() {
8446 use crate::client::ExecutionClient;
8447
8448 let server = MockServer::start().await;
8449 Mock::given(method("GET"))
8450 .and(path("/v2/account/activities"))
8451 .respond_with(
8452 ResponseTemplate::new(200)
8453 .set_body_json(make_activities_json(ALPACA_MAX_ACTIVITIES, "act")),
8454 )
8455 .mount(&server)
8456 .await;
8457
8458 let result = client_for(&server)
8459 .fetch_trades(time("2025-04-18T14:30:00.000500Z"), Utc::now(), &[])
8460 .await;
8461 assert!(
8462 matches!(
8463 result,
8464 Err(UnindexedClientError::Truncated { fills_read: 5_000 })
8465 ),
8466 "got {result:?}"
8467 );
8468 }
8469
8470 #[tokio::test]
8473 async fn fetch_trades_returns_only_the_span() {
8474 let (start, end) = (
8475 time("2025-04-18T14:30:00.123456Z"),
8476 time("2025-04-18T14:30:05.678901Z"),
8477 );
8478 let micro = TimeDelta::microseconds(1);
8479 let server = serve_activities(vec![
8480 fill_at("1::before", "SPY", start - micro),
8481 fill_at("2::start", "SPY", start),
8482 fill_at("3::end", "SPY", end),
8483 fill_at("4::after", "SPY", end + micro),
8484 ])
8485 .await;
8486
8487 let (ids, calls) = read_span(&client_for(&server), start, end, &[]).await;
8488 assert_eq!(calls, 1);
8489 let mut ids: Vec<_> = ids.into_iter().collect();
8490 ids.sort();
8491 assert_eq!(ids, ["end", "start"]);
8492 }
8493
8494 #[tokio::test]
8497 async fn fetch_trades_that_read_past_the_end_is_complete() {
8498 use crate::client::ExecutionClient;
8499
8500 let server = MockServer::start().await;
8501 Mock::given(method("GET"))
8503 .and(path("/v2/account/activities"))
8504 .respond_with(
8505 ResponseTemplate::new(200)
8506 .set_body_json(make_activities_json(ALPACA_MAX_ACTIVITIES, "act")),
8507 )
8508 .mount(&server)
8509 .await;
8510
8511 let Ok(read) = client_for(&server)
8512 .fetch_trades(recovery_after(), time("2025-04-18T14:29:00Z"), &[])
8513 .await
8514 else {
8515 panic!("the read succeeds");
8516 };
8517 assert_eq!(read, TradesRead::complete(Vec::new()));
8518 }
8519
8520 #[tokio::test]
8524 async fn fetch_trades_accepts_the_extreme_times() {
8525 use crate::client::ExecutionClient;
8526
8527 let server = serve_activities(vec![fill_at(
8528 "1::fill",
8529 "SPY",
8530 time("2025-04-18T14:30:00Z"),
8531 )])
8532 .await;
8533 let Ok(read) = client_for(&server)
8534 .fetch_trades(DateTime::<Utc>::MIN_UTC, DateTime::<Utc>::MAX_UTC, &[])
8535 .await
8536 else {
8537 panic!("the read succeeds");
8538 };
8539 assert_eq!(read.trades.len(), 1);
8540 assert_eq!(read.resume, None);
8541
8542 let Some(requests) = server.received_requests().await else {
8543 panic!("requests are recorded");
8544 };
8545 let [request] = requests.as_slice() else {
8546 panic!("one request: {requests:?}");
8547 };
8548 let query: std::collections::HashMap<_, _> =
8549 request.url.query_pairs().into_owned().collect();
8550 assert_eq!(
8551 query.get("after").map(String::as_str),
8552 Some("1970-01-01T00:00:00.000Z")
8553 );
8554 assert!(!query.contains_key("until"), "{query:?}");
8555
8556 let before_epoch = DateTime::UNIX_EPOCH - TimeDelta::days(1);
8558 let Ok(read) = client_for(&server)
8559 .fetch_trades(DateTime::<Utc>::MIN_UTC, before_epoch, &[])
8560 .await
8561 else {
8562 panic!("the read succeeds");
8563 };
8564 assert_eq!(read, TradesRead::complete(Vec::new()));
8565 assert_eq!(server.received_requests().await.map(|r| r.len()), Some(1));
8566 }
8567
8568 #[tokio::test]
8570 async fn fetch_trades_reads_nothing_for_an_empty_span() {
8571 use crate::client::ExecutionClient;
8572
8573 let server = MockServer::start().await;
8574 let start = recovery_after();
8575 let Ok(read) = client_for(&server)
8576 .fetch_trades(start, start - TimeDelta::seconds(1), &[])
8577 .await
8578 else {
8579 panic!("an empty span reads");
8580 };
8581 assert_eq!(read, TradesRead::complete(Vec::new()));
8582 assert_eq!(server.received_requests().await.map(|r| r.len()), Some(0));
8583 }
8584
8585 #[tokio::test]
8588 async fn recovery_reads_from_the_start_of_the_disconnects_millisecond() {
8589 let server = MockServer::start().await;
8590 Mock::given(method("GET"))
8591 .and(path("/v2/account/activities"))
8592 .and(wiremock::matchers::query_param(
8593 "after",
8594 "2025-01-01T00:00:00.123Z",
8595 ))
8596 .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([])))
8597 .expect(1)
8598 .mount(&server)
8599 .await;
8600
8601 let (tx, _rx) = mpsc::unbounded_channel();
8602 let outcome = recover_fills(
8603 &reqwest::Client::new(),
8604 &RateLimitTracker::new(),
8605 &[],
8606 &server.uri(),
8607 time("2025-01-01T00:00:00.123456Z"),
8608 &tx,
8609 &new_dedup_cache(),
8610 &KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
8611 )
8612 .await;
8613 assert_eq!(outcome, Ok(()));
8614 server.verify().await;
8615 }
8616
8617 fn dedup_keys(dedup: &SharedDedupCache) -> Vec<String> {
8619 let mut keys: Vec<_> = dedup
8620 .lock()
8621 .iter()
8622 .filter_map(|(key, _)| match key {
8623 FillKey::Cumulative(key) => Some(key.to_string()),
8624 FillKey::Execution(_) | FillKey::CumulativeWithoutId(_) => None,
8625 })
8626 .collect();
8627 keys.reverse();
8629 keys
8630 }
8631
8632 fn trade_of(
8633 event: &UnindexedAccountEvent,
8634 ) -> &Trade<AssetNameExchange, InstrumentNameExchange> {
8635 match &event.kind {
8636 AccountEventKind::Trade(t) => t,
8637 other => panic!("expected a Trade event, got {other:?}"),
8638 }
8639 }
8640
8641 #[tokio::test]
8644 async fn a_recovered_fill_reports_the_cumulative_alpaca_sent() {
8645 let events = drive_recover_fills(vec![
8646 activity_json("act-1", "ord-1", "2", Some("2")),
8647 activity_json("act-2", "ord-1", "3", Some("5")),
8648 ])
8649 .await;
8650
8651 let reported: Vec<_> = events
8652 .iter()
8653 .map(|e| trade_of(e).order_filled_quantity)
8654 .collect();
8655 assert_eq!(
8656 reported,
8657 vec![
8658 Some(Decimal::from_str("2").unwrap()),
8659 Some(Decimal::from_str("5").unwrap())
8660 ]
8661 );
8662 }
8663
8664 #[tokio::test]
8671 async fn the_dedup_key_uses_the_venue_cumulative_not_an_intra_batch_count() {
8672 let dedup = new_dedup_cache();
8674 drive_recover_fills_with(
8675 vec![activity_json("act-1", "ord-1", "2", Some("5"))],
8676 dedup.clone(),
8677 &KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
8678 )
8679 .await;
8680
8681 assert_eq!(
8682 dedup_keys(&dedup),
8683 vec!["ord-1:5"],
8684 "counting executions within the batch would key this as ord-1:2"
8685 );
8686 }
8687
8688 async fn recover_after_ws_fill(
8692 execution_id: Option<&str>,
8693 activity_id: &str,
8694 ) -> Vec<UnindexedAccountEvent> {
8695 let dedup = new_dedup_cache();
8696 let update = AlpacaTradeUpdate {
8697 event: SmolStr::new("partial_fill"),
8698 execution_id,
8699 previous_execution_id: None,
8700 order: super::make_order_ws("ord-1", "SPY", "buy", "5"),
8701 price: Some("100.00"),
8702 qty: Some("2"),
8703 timestamp: None,
8704 };
8705 assert!(
8706 !is_duplicate_fill(&dedup, ws_execution_id(&update), early_dedup_key(&update)),
8707 "precondition: first sighting"
8708 );
8709
8710 drive_recover_fills_with(
8711 vec![activity_json(activity_id, "ord-1", "2", Some("5"))],
8712 dedup,
8713 &KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
8714 )
8715 .await
8716 }
8717
8718 #[tokio::test]
8721 async fn a_fill_already_delivered_over_websocket_is_not_recovered_twice() {
8722 for (ws_execution_id, activity_id) in [
8723 (Some("exec-a"), "20250418103000000::exec-a"),
8724 (None, "20250418103000000::exec-a"),
8725 (Some("exec-a"), "act-1"),
8726 ] {
8727 let events = recover_after_ws_fill(ws_execution_id, activity_id).await;
8728 assert!(
8729 events.is_empty(),
8730 "a fill already delivered over WebSocket ({ws_execution_id:?}) must not be \
8731 re-delivered by recovery ({activity_id}), got {events:?}"
8732 );
8733 }
8734 }
8735
8736 async fn recover_after_correction(fields: &str) -> Vec<UnindexedAccountEvent> {
8739 let (tx, _rx) = mpsc::unbounded_channel();
8740 let dedup = new_dedup_cache();
8741 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
8742 process_ws_text(
8743 &super::amendment_frame("trade_correct", fields),
8744 &tx,
8745 &dedup,
8746 &known,
8747 &mut ExponentialBackoff::new(),
8748 );
8749
8750 drive_recover_fills_with(
8751 vec![activity_json(
8752 "20250418103100000::exec-c",
8753 "ord-1",
8754 "2",
8755 Some("2"),
8756 )],
8757 dedup,
8758 &known,
8759 )
8760 .await
8761 }
8762
8763 #[tokio::test]
8766 async fn a_corrected_execution_is_not_recovered_again() {
8767 let events = recover_after_correction(
8768 r#""execution_id":"exec-c","previous_execution_id":"exec-a","price":"101.50","qty":"2","#,
8769 )
8770 .await;
8771 assert!(
8772 events.is_empty(),
8773 "the corrected execution was delivered as the replacement, got {events:?}"
8774 );
8775 }
8776
8777 #[tokio::test]
8779 async fn an_unresolved_correction_does_not_keep_its_execution_from_recovery() {
8780 let events = recover_after_correction(
8781 r#""execution_id":"exec-c","previous_execution_id":"exec-a","qty":"2","#,
8782 )
8783 .await;
8784 let ids: Vec<_> = events.iter().map(|e| trade_of(e).id.0.as_str()).collect();
8785 assert_eq!(ids, vec!["exec-c"]);
8786 }
8787
8788 #[tokio::test]
8791 async fn a_different_execution_at_a_cumulative_already_seen_is_recovered() {
8792 let events = recover_after_ws_fill(Some("exec-a"), "20250418103000000::exec-b").await;
8793 let ids: Vec<_> = events.iter().map(|e| trade_of(e).id.0.as_str()).collect();
8794 assert_eq!(ids, vec!["exec-b"]);
8795 }
8796
8797 #[tokio::test]
8800 async fn without_a_reported_cumulative_the_previous_behaviour_is_preserved() {
8801 let dedup = new_dedup_cache();
8802 let events = drive_recover_fills_with(
8803 vec![
8804 activity_json("act-1", "ord-1", "2", None),
8805 activity_json("act-2", "ord-1", "3", None),
8806 ],
8807 dedup.clone(),
8808 &KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
8809 )
8810 .await;
8811
8812 assert_eq!(
8813 dedup_keys(&dedup),
8814 vec!["ord-1:2", "ord-1:5"],
8815 "intra-batch accumulation"
8816 );
8817 let ids: Vec<_> = events.iter().map(|e| trade_of(e).id.0.as_str()).collect();
8818 assert_eq!(ids, vec!["act-1", "act-2"], "each keeps its activity's id");
8819
8820 assert!(
8821 events
8822 .iter()
8823 .all(|e| trade_of(e).order_filled_quantity.is_none()),
8824 "a venue that reported no cumulative must not have one invented for it"
8825 );
8826 }
8827
8828 fn order_json(cid: &str, symbol: &str, status: &str, filled: &str) -> serde_json::Value {
8833 serde_json::json!({
8834 "id": format!("id-{cid}"),
8835 "client_order_id": cid,
8836 "symbol": symbol,
8837 "qty": "2",
8838 "filled_qty": filled,
8839 "filled_avg_price": if filled == "0" { None } else { Some("101.5") },
8840 "status": status,
8841 "side": "buy",
8842 "type": "limit",
8843 "time_in_force": "gtc",
8844 "limit_price": "100",
8845 "created_at": "2026-10-01T14:30:00Z",
8846 "updated_at": "2026-10-01T15:00:00Z"
8847 })
8848 }
8849
8850 async fn mount_lookup(server: &MockServer, cid: &str, order: ResponseTemplate, calls: u64) {
8852 use wiremock::matchers::query_param;
8853 Mock::given(method("GET"))
8854 .and(path("/v2/orders:by_client_order_id"))
8855 .and(query_param("client_order_id", cid))
8856 .respond_with(order)
8857 .expect(calls)
8858 .mount(server)
8859 .await;
8860 }
8861
8862 fn ok(body: serde_json::Value) -> ResponseTemplate {
8863 ResponseTemplate::new(200).set_body_json(body)
8864 }
8865
8866 fn alpaca_key(instrument: &str, cid: &str) -> UnindexedOrderKey {
8867 OrderKey::new(
8868 ExchangeId::AlpacaBroker,
8869 InstrumentNameExchange::new(instrument),
8870 StrategyId::new("strategy"),
8871 ClientOrderId::new(cid),
8872 )
8873 }
8874
8875 fn hold(known: &SharedKnownLiveOrders, key: &UnindexedOrderKey) {
8876 let open = Open::new(
8877 VenueOrderId::Assigned(OrderId::new(format!("id-{}", key.cid))),
8878 Utc::now(),
8879 Decimal::ZERO,
8880 );
8881 known
8882 .lock()
8883 .live(key, Decimal::TWO, OrderKind::Limit, &open);
8884 }
8885
8886 #[tokio::test]
8887 async fn fetch_ended_orders_reads_each_end_alpaca_reports() {
8888 use rust_decimal_macros::dec;
8889
8890 let server = MockServer::start().await;
8891 for (cid, status, filled) in [
8892 ("filled", "filled", "2"),
8893 ("canceled", "canceled", "1"),
8894 ("replaced", "replaced", "0"),
8895 ("expired", "expired", "1"),
8896 ("rejected", "rejected", "0"),
8897 ("done", "done_for_day", "1"),
8898 ("live", "new", "0"),
8899 ("unknown-status", "frozen", "0"),
8900 ] {
8901 mount_lookup(&server, cid, ok(order_json(cid, "SPY", status, filled)), 1).await;
8902 }
8903 mount_lookup(
8904 &server,
8905 "elsewhere",
8906 ok(order_json("elsewhere", "QQQ", "canceled", "0")),
8907 1,
8908 )
8909 .await;
8910 mount_lookup(
8911 &server,
8912 "gone",
8913 ResponseTemplate::new(404).set_body_json(serde_json::json!({
8914 "code": 40410000,
8915 "message": "order not found for gone"
8916 })),
8917 1,
8918 )
8919 .await;
8920
8921 let keys: Vec<_> = [
8922 "filled",
8923 "canceled",
8924 "replaced",
8925 "expired",
8926 "rejected",
8927 "done",
8928 "live",
8929 "unknown-status",
8930 "elsewhere",
8931 "gone",
8932 ]
8933 .into_iter()
8934 .map(|cid| alpaca_key("spy", cid))
8935 .collect();
8936 let ended = client_for(&server).fetch_ended_orders(&keys).await.unwrap();
8937
8938 let mut found: Vec<_> = ended
8939 .iter()
8940 .map(|order| (order.key.cid.0.as_str(), &order.state))
8941 .collect();
8942 found.sort_by_key(|(cid, _)| *cid);
8943 let [
8944 (canceled, InactiveOrderState::Cancelled(cancelled)),
8945 (expired, InactiveOrderState::Expired(expiry)),
8946 (filled, InactiveOrderState::FullyFilled(fill)),
8947 (rejected, InactiveOrderState::OpenFailed(failed)),
8948 (replaced, InactiveOrderState::Cancelled(replacement)),
8949 ] = found.as_slice()
8950 else {
8951 panic!("filled, canceled, replaced, expired and rejected: {found:?}");
8952 };
8953 assert_eq!(
8954 [*canceled, *expired, *filled, *rejected, *replaced],
8955 ["canceled", "expired", "filled", "rejected", "replaced"]
8956 );
8957 assert_eq!(
8958 (fill.filled_quantity, fill.avg_price),
8959 (dec!(2), Some(dec!(101.5)))
8960 );
8961 assert_eq!(
8962 cancelled.filled_quantity,
8963 Some(dec!(1)),
8964 "what filled before"
8965 );
8966 assert_eq!(expiry.filled_quantity, Some(dec!(1)));
8967 assert_eq!(replacement.filled_quantity, Some(Decimal::ZERO));
8968 assert_eq!(
8969 cancelled.time_exchange,
8970 parse_timestamp("2026-10-01T15:00:00Z").unwrap(),
8971 "stamped when the order last changed"
8972 );
8973 assert!(matches!(
8974 failed,
8975 OrderError::Rejected(ApiError::OrderRejected(message))
8976 if message.contains("id-rejected")
8977 ));
8978 assert!(
8979 ended
8980 .iter()
8981 .all(|order| order.key.strategy == StrategyId::new("strategy")
8982 && order.key.instrument == InstrumentNameExchange::new("spy")),
8983 "each carries the key it was asked for"
8984 );
8985 }
8986
8987 #[tokio::test]
8988 async fn fetch_ended_orders_fails_when_a_lookup_fails() {
8989 let server = MockServer::start().await;
8990 mount_lookup(&server, "a", ok(order_json("a", "SPY", "canceled", "0")), 1).await;
8991 mount_lookup(&server, "b", ResponseTemplate::new(500), 1).await;
8992
8993 let result = client_for(&server)
8994 .fetch_ended_orders(&[alpaca_key("SPY", "a"), alpaca_key("SPY", "b")])
8995 .await;
8996
8997 assert!(
8998 matches!(result, Err(UnindexedClientError::Connectivity(_))),
8999 "{result:?}"
9000 );
9001 }
9002
9003 #[tokio::test]
9006 async fn a_reconnect_check_lists_once_and_reports_the_orders_that_ended() {
9007 use wiremock::matchers::query_param;
9008
9009 let server = MockServer::start().await;
9010 Mock::given(method("GET"))
9011 .and(path("/v2/orders"))
9012 .and(query_param("status", "open"))
9013 .and(query_param("symbols", "QQQ,SPY"))
9014 .respond_with(ok(serde_json::json!([order_json(
9015 "spy-listed",
9016 "SPY",
9017 "new",
9018 "0"
9019 )])))
9020 .expect(1)
9021 .mount(&server)
9022 .await;
9023 mount_lookup(&server, "spy-listed", ResponseTemplate::new(500), 0).await;
9024 mount_lookup(
9025 &server,
9026 "spy-cancelled",
9027 ok(order_json("spy-cancelled", "SPY", "canceled", "1")),
9028 1,
9029 )
9030 .await;
9031 mount_lookup(
9032 &server,
9033 "spy-done",
9034 ok(order_json("spy-done", "SPY", "done_for_day", "1")),
9035 1,
9036 )
9037 .await;
9038 mount_lookup(
9039 &server,
9040 "qqq-filled",
9041 ok(order_json("qqq-filled", "QQQ", "filled", "2")),
9042 1,
9043 )
9044 .await;
9045
9046 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
9047 let [listed, cancelled, done, filled] = [
9048 ("SPY", "spy-listed"),
9049 ("SPY", "spy-cancelled"),
9050 ("SPY", "spy-done"),
9051 ("QQQ", "qqq-filled"),
9052 ]
9053 .map(|(instrument, cid)| alpaca_key(instrument, cid));
9054 for key in [&listed, &cancelled, &done, &filled] {
9055 hold(&known, key);
9056 }
9057 let mut unchecked = UncheckedOrders::default();
9058 unchecked.open(known.lock().instruments());
9059 let config = Arc::new(AlpacaConfig::with_base_url(
9060 "test-key".into(),
9061 "test-secret".into(),
9062 server.uri(),
9063 ));
9064 let (tx, mut rx) = mpsc::unbounded_channel();
9065
9066 recover_alpaca_ended_orders(
9067 &reqwest::Client::new(),
9068 &Arc::new(RateLimitTracker::new()),
9069 &config,
9070 &known,
9071 &mut unchecked,
9072 &tx,
9073 )
9074 .await;
9075
9076 let mut reported: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok())
9077 .map(|event| {
9078 let AccountEventKind::OrderSnapshot(
9079 rustrade_integration::collection::snapshot::Snapshot(order),
9080 ) = event.kind
9081 else {
9082 panic!("an order snapshot: {event:?}");
9083 };
9084 assert!(matches!(order.state, OrderState::Inactive(_)), "{order:?}");
9085 order.key.cid
9086 })
9087 .collect();
9088 reported.sort();
9089 assert_eq!(reported, [filled.cid.clone(), cancelled.cid.clone()]);
9090 let known = known.lock();
9091 assert!(known.contains(&listed.cid) && known.contains(&done.cid));
9092 assert!(!known.contains(&cancelled.cid) && !known.contains(&filled.cid));
9093 assert!(unchecked.is_empty(), "both instruments checked");
9094 }
9095
9096 #[tokio::test]
9098 async fn a_failed_listing_is_retried_for_every_instrument_in_it() {
9099 let server = MockServer::start().await;
9100 Mock::given(method("GET"))
9101 .and(path("/v2/orders"))
9102 .respond_with(ResponseTemplate::new(503))
9103 .expect(1)
9104 .mount(&server)
9105 .await;
9106
9107 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
9108 hold(&known, &alpaca_key("SPY", "a"));
9109 hold(&known, &alpaca_key("QQQ", "b"));
9110 let mut unchecked = UncheckedOrders::default();
9111 unchecked.open(known.lock().instruments());
9112 let config = Arc::new(AlpacaConfig::with_base_url(
9113 "test-key".into(),
9114 "test-secret".into(),
9115 server.uri(),
9116 ));
9117 let (tx, mut rx) = mpsc::unbounded_channel();
9118
9119 recover_alpaca_ended_orders(
9120 &reqwest::Client::new(),
9121 &Arc::new(RateLimitTracker::new()),
9122 &config,
9123 &known,
9124 &mut unchecked,
9125 &tx,
9126 )
9127 .await;
9128
9129 assert!(rx.try_recv().is_err(), "nothing reported");
9130 let now = tokio::time::Instant::now();
9131 assert!(
9132 unchecked.ready(&NoPendingFills, now).is_empty(),
9133 "both wait"
9134 );
9135 for instrument in ["QQQ", "SPY"] {
9136 assert_eq!(
9137 unchecked.failed(&InstrumentNameExchange::new(instrument), now),
9138 Some(crate::client::order_recovery::GapFailure::Retry(
9139 Duration::from_secs(crate::client::order_recovery::GAP_RETRY_BASE_SECS * 2)
9140 )),
9141 "{instrument} has failed once already"
9142 );
9143 }
9144 assert_eq!(known.lock().instruments().len(), 2, "both still held");
9145 }
9146
9147 #[tokio::test]
9148 async fn placing_listing_and_cancelling_keep_the_held_orders_in_step() {
9149 use crate::order::request::{OrderRequestCancel, RequestCancel, RequestOpen};
9150
9151 let server = MockServer::start().await;
9152 Mock::given(method("POST"))
9153 .and(path("/v2/orders"))
9154 .respond_with(ok(order_json("resting", "SPY", "new", "0")))
9155 .up_to_n_times(1)
9156 .mount(&server)
9157 .await;
9158 Mock::given(method("POST"))
9159 .and(path("/v2/orders"))
9160 .respond_with(ok(order_json("filled", "SPY", "filled", "2")))
9161 .mount(&server)
9162 .await;
9163 Mock::given(method("GET"))
9164 .and(path("/v2/orders"))
9165 .respond_with(ok(serde_json::json!([order_json(
9166 "listed", "QQQ", "new", "0"
9167 )])))
9168 .mount(&server)
9169 .await;
9170 Mock::given(method("DELETE"))
9171 .and(path("/v2/orders/id-resting"))
9172 .respond_with(ResponseTemplate::new(204))
9173 .mount(&server)
9174 .await;
9175 let client = client_for(&server);
9176 let spy = InstrumentNameExchange::new("SPY");
9177
9178 for cid in ["resting", "filled"] {
9179 let request = OrderRequestOpen {
9180 key: OrderKey::new(
9181 ExchangeId::AlpacaBroker,
9182 &spy,
9183 StrategyId::new("strategy"),
9184 ClientOrderId::new(cid),
9185 ),
9186 state: RequestOpen {
9187 side: Side::Buy,
9188 price: Some(Decimal::ONE_HUNDRED),
9189 quantity: Decimal::TWO,
9190 kind: OrderKind::Limit,
9191 time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
9192 position_id: None,
9193 reduce_only: false,
9194 market: None,
9195 },
9196 };
9197 client.open_order(request).await;
9198 }
9199 client.fetch_open_orders(&[]).await.unwrap();
9200 {
9201 let known = client.known_live.lock();
9202 assert!(known.contains(&ClientOrderId::new("resting")));
9203 assert!(
9204 !known.contains(&ClientOrderId::new("filled")),
9205 "filled in its placement response"
9206 );
9207 assert!(known.contains(&ClientOrderId::new("listed")));
9208 }
9209
9210 let cancel = client
9211 .cancel_order(OrderRequestCancel {
9212 key: OrderKey::new(
9213 ExchangeId::AlpacaBroker,
9214 &spy,
9215 StrategyId::new("strategy"),
9216 ClientOrderId::new("resting"),
9217 ),
9218 state: RequestCancel {
9219 id: Some(VenueOrderId::Assigned(OrderId::new("id-resting"))),
9220 },
9221 })
9222 .await;
9223 let Ok(cancelled) = &cancel.state else {
9224 panic!("expected Ok, got {cancel:?}");
9225 };
9226 assert_eq!(
9227 cancelled.filled_quantity, None,
9228 "the DELETE response has no body, so what filled is unknown"
9229 );
9230 assert!(
9231 client
9232 .known_live
9233 .lock()
9234 .contains(&ClientOrderId::new("resting")),
9235 "a cancel Alpaca has only accepted has not ended the order"
9236 );
9237 }
9238
9239 #[tokio::test]
9246 async fn an_order_that_ended_in_its_placement_response_is_reported_as_it_ended() {
9247 use crate::order::request::RequestOpen;
9248
9249 let Some(time) = parse_timestamp("2026-10-01T15:00:00Z") else {
9250 panic!("order_json's updated_at parses");
9251 };
9252 let id = |cid: &str| OrderId::new(format!("id-{cid}"));
9253 let open = |cid: &str, filled| {
9254 OrderState::active(Open::new(VenueOrderId::Assigned(id(cid)), time, filled))
9255 };
9256 let cases: [(&str, &str, &str, UnindexedOrderState, bool); 13] = [
9257 (
9258 "ioc-unfilled",
9259 "expired",
9260 "0",
9261 OrderState::expired(Expired::new(
9262 id("ioc-unfilled"),
9263 time,
9264 Some(Decimal::ZERO),
9265 )),
9266 false,
9267 ),
9268 (
9269 "ioc-partial",
9270 "expired",
9271 "1",
9272 OrderState::expired(Expired::new(id("ioc-partial"), time, Some(Decimal::ONE))),
9273 false,
9274 ),
9275 (
9276 "fill-unknown",
9277 "expired",
9278 "not-a-number",
9279 OrderState::expired(Expired::new(id("fill-unknown"), time, None)),
9280 false,
9281 ),
9282 (
9283 "cancelled",
9284 "canceled",
9285 "1",
9286 OrderState::inactive(Cancelled::new(id("cancelled"), time, Some(Decimal::ONE))),
9287 false,
9288 ),
9289 (
9290 "rejected",
9291 "rejected",
9292 "0",
9293 OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
9294 "Alpaca rejected order id-rejected after accepting it".to_string(),
9295 ))),
9296 false,
9297 ),
9298 (
9299 "filled",
9300 "filled",
9301 "2",
9302 OrderState::fully_filled(Filled::new(
9303 id("filled"),
9304 time,
9305 Decimal::TWO,
9306 Some(Decimal::new(1015, 1)),
9307 )),
9308 false,
9309 ),
9310 (
9311 "replaced",
9312 "replaced",
9313 "0",
9314 OrderState::inactive(Cancelled::new(id("replaced"), time, Some(Decimal::ZERO))),
9315 false,
9316 ),
9317 (
9318 "live-covered",
9319 "partially_filled",
9320 "2",
9321 OrderState::fully_filled(Filled::new(
9322 id("live-covered"),
9323 time,
9324 Decimal::TWO,
9325 Some(Decimal::new(1015, 1)),
9326 )),
9327 false,
9328 ),
9329 ("new", "new", "0", open("new", Decimal::ZERO), true),
9330 (
9331 "live-fill-unknown",
9332 "new",
9333 "not-a-number",
9334 open("live-fill-unknown", Decimal::ZERO),
9335 true,
9336 ),
9337 (
9338 "partial",
9339 "partially_filled",
9340 "1",
9341 open("partial", Decimal::ONE),
9342 true,
9343 ),
9344 (
9345 "for-the-day",
9346 "done_for_day",
9347 "1",
9348 open("for-the-day", Decimal::ONE),
9349 true,
9350 ),
9351 (
9352 "unknown",
9353 "a_status_not_yet_documented",
9354 "0",
9355 open("unknown", Decimal::ZERO),
9356 true,
9357 ),
9358 ];
9359
9360 for (cid, status, filled, expected, held) in cases {
9361 for bracket in [false, true] {
9362 let server = MockServer::start().await;
9363 Mock::given(method("POST"))
9364 .and(path("/v2/orders"))
9365 .respond_with(ok(order_json(cid, "SPY", status, filled)))
9366 .expect(1)
9367 .mount(&server)
9368 .await;
9369 let client = client_for(&server);
9370 let spy = InstrumentNameExchange::new("SPY");
9371 let time_in_force = TimeInForce::GoodUntilCancelled { post_only: false };
9372
9373 let state = if bracket {
9374 client
9375 .open_bracket_order(AlpacaBracketOrderRequest::new(
9376 spy,
9377 StrategyId::new("strategy"),
9378 ClientOrderId::new(cid),
9379 Side::Buy,
9380 Decimal::TWO,
9381 Decimal::ONE_HUNDRED,
9382 Decimal::new(120, 0),
9383 Decimal::new(90, 0),
9384 time_in_force,
9385 ))
9386 .await
9387 .parent
9388 .state
9389 } else {
9390 let request = OrderRequestOpen {
9391 key: OrderKey::new(
9392 ExchangeId::AlpacaBroker,
9393 &spy,
9394 StrategyId::new("strategy"),
9395 ClientOrderId::new(cid),
9396 ),
9397 state: RequestOpen {
9398 side: Side::Buy,
9399 price: Some(Decimal::ONE_HUNDRED),
9400 quantity: Decimal::TWO,
9401 kind: OrderKind::Limit,
9402 time_in_force,
9403 position_id: None,
9404 reduce_only: false,
9405 market: None,
9406 },
9407 };
9408 client.open_order(request).await.state
9409 };
9410
9411 assert_eq!(state, expected, "{status} {filled}, bracket: {bracket}");
9412 assert_eq!(
9413 client.known_live.lock().contains(&ClientOrderId::new(cid)),
9414 held,
9415 "{status} {filled}, bracket: {bracket}"
9416 );
9417 }
9418 }
9419 }
9420
9421 #[test]
9424 fn the_stream_keeps_the_held_orders_in_step() {
9425 let (tx, mut rx) = mpsc::unbounded_channel();
9426 let dedup = new_dedup_cache();
9427 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
9428 let mut backoff = ExponentialBackoff::new();
9429 let frame = |event: &str, filled: &str| {
9430 format!(
9431 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}"}}}}}}"#
9432 )
9433 };
9434 let cid = ClientOrderId::new("a");
9435
9436 process_ws_text(&frame("new", "0"), &tx, &dedup, &known, &mut backoff);
9437 assert!(known.lock().contains(&cid), "acknowledged");
9438
9439 process_ws_text(
9440 &frame("done_for_day", "1"),
9441 &tx,
9442 &dedup,
9443 &known,
9444 &mut backoff,
9445 );
9446 assert!(known.lock().contains(&cid), "done for the day is not ended");
9447
9448 process_ws_text(&frame("canceled", "1"), &tx, &dedup, &known, &mut backoff);
9449 assert!(!known.lock().contains(&cid), "cancelled");
9450
9451 let events: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
9452 let [_, done, cancelled] = events.as_slice() else {
9453 panic!("three events: {events:?}");
9454 };
9455 let AccountEventKind::OrderSnapshot(
9456 rustrade_integration::collection::snapshot::Snapshot(order),
9457 ) = &done.kind
9458 else {
9459 panic!("done_for_day reports an order snapshot: {done:?}");
9460 };
9461 let OrderState::Active(ActiveOrderState::Open(open)) = &order.state else {
9462 panic!("open: {order:?}");
9463 };
9464 assert_eq!(open.filled_quantity, Decimal::ONE, "with what it filled");
9465 assert!(matches!(
9466 cancelled.kind,
9467 AccountEventKind::OrderCancelled(_)
9468 ));
9469 }
9470
9471 #[tokio::test]
9474 async fn a_recovered_fill_that_completes_an_order_stops_it_being_held() {
9475 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
9476 let key = alpaca_key("SPY", "a");
9477 known.lock().live(
9478 &key,
9479 Decimal::TWO,
9480 OrderKind::Limit,
9481 &Open::new(
9482 VenueOrderId::Assigned(OrderId::new("ord-1")),
9483 Utc::now(),
9484 Decimal::ZERO,
9485 ),
9486 );
9487
9488 drive_recover_fills_with(
9489 vec![activity_json("act-1", "ord-1", "1", Some("1"))],
9490 new_dedup_cache(),
9491 &known,
9492 )
9493 .await;
9494 assert!(known.lock().contains(&key.cid), "half filled");
9495
9496 drive_recover_fills_with(
9497 vec![activity_json("act-2", "ord-1", "1", Some("2"))],
9498 new_dedup_cache(),
9499 &known,
9500 )
9501 .await;
9502 assert!(!known.lock().contains(&key.cid), "filled");
9503 }
9504
9505 #[tokio::test]
9508 async fn the_check_loop_runs_at_once_then_waits_out_a_failure() {
9509 let server = MockServer::start().await;
9510 Mock::given(method("GET"))
9511 .and(path("/v2/orders"))
9512 .respond_with(ResponseTemplate::new(503))
9513 .expect(1)
9514 .mount(&server)
9515 .await;
9516 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
9517 hold(&known, &alpaca_key("SPY", "a"));
9518 let mut unchecked = UncheckedOrders::default();
9519 unchecked.open(known.lock().instruments());
9520 let config = Arc::new(AlpacaConfig::with_base_url(
9521 "test-key".into(),
9522 "test-secret".into(),
9523 server.uri(),
9524 ));
9525 let (tx, _rx) = mpsc::unbounded_channel();
9526
9527 let ran = tokio::time::timeout(
9528 Duration::from_secs(1),
9529 run_order_checks(
9530 &reqwest::Client::new(),
9531 &Arc::new(RateLimitTracker::new()),
9532 &config,
9533 &known,
9534 &mut unchecked,
9535 &tx,
9536 ),
9537 )
9538 .await;
9539
9540 assert!(ran.is_err(), "the loop never completes");
9541 assert!(
9542 unchecked.contains(&InstrumentNameExchange::new("SPY")),
9543 "SPY waits for its retry"
9544 );
9545 assert_eq!(
9546 server.received_requests().await.unwrap().len(),
9547 1,
9548 "the failed check is not run again at once"
9549 );
9550 }
9551
9552 #[tokio::test]
9554 async fn the_check_loop_waits_once_the_consumer_has_gone() {
9555 let server = MockServer::start().await;
9556 Mock::given(method("GET"))
9557 .respond_with(ResponseTemplate::new(503))
9558 .expect(0)
9559 .mount(&server)
9560 .await;
9561 let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
9562 hold(&known, &alpaca_key("SPY", "a"));
9563 let mut unchecked = UncheckedOrders::default();
9564 unchecked.open(known.lock().instruments());
9565 let config = Arc::new(AlpacaConfig::with_base_url(
9566 "test-key".into(),
9567 "test-secret".into(),
9568 server.uri(),
9569 ));
9570 let (tx, rx) = mpsc::unbounded_channel();
9571 drop(rx);
9572
9573 let ran = tokio::time::timeout(
9574 Duration::from_millis(200),
9575 run_order_checks(
9576 &reqwest::Client::new(),
9577 &Arc::new(RateLimitTracker::new()),
9578 &config,
9579 &known,
9580 &mut unchecked,
9581 &tx,
9582 ),
9583 )
9584 .await;
9585
9586 assert!(ran.is_err(), "the loop never completes");
9587 assert!(
9588 server.received_requests().await.unwrap().is_empty(),
9589 "nothing asked once the consumer has gone"
9590 );
9591 }
9592 }
9593}