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