use crate::{
AccountEventKind, AccountSnapshot, FillRecoveryFailure, FillRecoveryGap, FillRecoveryScope,
InstrumentAccountSnapshot, UnindexedAccountEvent, UnindexedAccountSnapshot,
balance::{AssetBalance, Balance},
cash_flow::{CashFlow, CashFlowId, CashFlowKind, CashFlowsRead},
client::{
BracketOrderClient, ExecutionClient, OrderStatusClient, ShortabilityClient,
order_recovery::{
KnownLiveOrders, NoPendingFills, OpenListing, OrderLookup, SharedKnownLiveOrders,
UncheckedOrders, fetch_ended_by_key, recover_ended_orders,
},
reconnect_budget::{BackoffLimits, ReconnectBudget},
venue_clock::{HostMoment, VenueClock},
},
emit_stream_terminated,
error::{
ApiError, BorrowReject, BorrowRejectReason, ConnectivityError, OrderError,
StreamTerminationReason, UnindexedApiError, UnindexedClientError, UnindexedOrderError,
},
order::{
Order, OrderKey, OrderKind, TimeInForce, TrailingOffsetType, UnindexedInactiveOrder,
UnindexedOrderKey, UnindexedOrderSnapshot,
bracket::{
BracketOrderRequest as UnifiedBracketOrderRequest,
BracketOrderResult as UnifiedBracketOrderResult,
},
id::{ClientOrderId, OrderId, StrategyId, VenueOrderId},
request::{OrderRequestCancel, OrderRequestOpen, ReduceOnly, UnindexedOrderResponseCancel},
state::{
ActiveOrderState, Cancelled, Expired, Filled, InactiveOrderState, Open, OrderState,
UnindexedOrderState,
},
},
parse_env_bool,
position::{Position, PositionReport},
shortability::Shortability,
trade::{AssetFees, TimeSource, Trade, TradeAmendmentKind, TradeId, TradeOrigin, TradesRead},
};
use chrono::{DateTime, NaiveDate, NaiveTime, SecondsFormat, TimeDelta, Utc};
use fnv::{FnvHashMap, FnvHashSet};
use futures::{SinkExt as _, StreamExt as _, stream::BoxStream};
use indexmap::IndexMap;
use itertools::Itertools as _;
use lru::LruCache;
use rust_decimal::Decimal;
use rustrade_instrument::{
Side,
asset::name::AssetNameExchange,
exchange::ExchangeId,
instrument::{kind::InstrumentKindDiscriminant, name::InstrumentNameExchange},
};
use rustrade_integration::protocol::websocket::{WebSocket, WsMessage};
use serde::{Deserialize, Serialize, de::DeserializeOwned};
use smol_str::{SmolStr, format_smolstr};
use std::{num::NonZeroUsize, pin::Pin, str::FromStr, sync::Arc, time::Duration};
use tokio::sync::mpsc;
use tracing::{debug, error, info, trace, warn};
const HEARTBEAT_TIMEOUT_SECS: u64 = 35;
const FILL_RECOVERY_TIMEOUT_SECS: u64 = 30;
const SIGNAL_RECOVERY_LOOKBACK_MS: i64 = 1_500;
const UNSAMPLED_CLOCK_SKEW: TimeDelta = TimeDelta::seconds(30);
const ALPACA_MAX_ACTIVITIES: usize = 100;
const DEFAULT_RATE_LIMIT_DELAY_SECS: u64 = 60;
const MAX_RATE_LIMIT_ATTEMPTS: u32 = 4;
const DEDUP_CACHE_SIZE: usize = 4_000;
const USD: &str = "USD";
const WS_HANDSHAKE_TIMEOUT_SECS: u64 = 15;
const WS_CLOSE_TIMEOUT_SECS: u64 = 5;
struct GracefulShutdownStream<S> {
inner: S,
_handle: tokio::task::JoinHandle<()>,
}
impl<S> GracefulShutdownStream<S> {
fn new(inner: S, handle: tokio::task::JoinHandle<()>) -> Self {
Self {
inner,
_handle: handle,
}
}
}
impl<S: futures::Stream + Unpin> futures::Stream for GracefulShutdownStream<S> {
type Item = S::Item;
fn poll_next(
mut self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Option<Self::Item>> {
Pin::new(&mut self.inner).poll_next(cx)
}
fn size_hint(&self) -> (usize, Option<usize>) {
self.inner.size_hint()
}
}
impl<S> Drop for GracefulShutdownStream<S> {
fn drop(&mut self) {
}
}
struct RateLimitTracker {
blocked_until: parking_lot::Mutex<Option<tokio::time::Instant>>,
}
impl RateLimitTracker {
fn new() -> Self {
Self {
blocked_until: parking_lot::Mutex::new(None),
}
}
async fn wait_if_blocked(&self) {
loop {
let now = tokio::time::Instant::now();
let deadline = {
let mut guard = self.blocked_until.lock();
let d = *guard;
if matches!(d, Some(t) if t <= now) {
*guard = None;
}
d
};
match deadline {
None => return,
Some(until) => {
if until <= now {
return;
}
#[allow(clippy::cast_possible_truncation)]
let delay_ms = (until - now).as_millis() as u64;
debug!(delay_ms, "Alpaca REST rate-limited, waiting before request");
tokio::time::sleep_until(until).await;
}
}
}
}
fn on_rate_limited(&self, retry_after: Option<Duration>) {
let delay = retry_after.unwrap_or(Duration::from_secs(DEFAULT_RATE_LIMIT_DELAY_SECS));
let was_blocked = self.extend(delay);
if was_blocked {
debug!(
delay_secs = delay.as_secs(),
"Alpaca rate-limit cooldown extended"
);
} else {
warn!(
delay_secs = delay.as_secs(),
"Alpaca entering rate-limit degradation mode"
);
}
}
fn on_bucket_exhausted(&self, reset: Duration) {
let was_blocked = self.extend(reset);
if !was_blocked {
info!(
delay_ms = u64::try_from(reset.as_millis()).unwrap_or(u64::MAX),
"Alpaca rate-limit window exhausted (X-Ratelimit-Remaining: 0), pausing requests until it resets"
);
}
}
fn extend(&self, delay: Duration) -> bool {
let now = tokio::time::Instant::now();
let new_deadline = now + delay;
let mut guard = self.blocked_until.lock();
let was_blocked = guard.is_some_and(|until| until > now);
*guard = Some(guard.map_or(new_deadline, |existing| existing.max(new_deadline)));
was_blocked
}
}
type SharedDedupCache = Arc<parking_lot::Mutex<LruCache<FillKey, ()>>>;
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
enum FillKey {
Execution(SmolStr),
Cumulative(SmolStr),
CumulativeWithoutId(SmolStr),
}
fn new_dedup_cache() -> SharedDedupCache {
#[allow(clippy::unwrap_used)]
Arc::new(parking_lot::Mutex::new(LruCache::new(
NonZeroUsize::new(DEDUP_CACHE_SIZE).unwrap(),
)))
}
fn is_duplicate_fill(
cache: &SharedDedupCache,
execution_id: Option<&str>,
cumulative: SmolStr,
) -> bool {
let execution = execution_id.map(|id| FillKey::Execution(SmolStr::new(id)));
let mut guard = cache.lock();
match execution {
Some(execution) => {
if guard.peek(&execution).is_some()
|| guard
.peek(&FillKey::CumulativeWithoutId(cumulative.clone()))
.is_some()
{
return true;
}
guard.put(execution, ());
guard.put(FillKey::Cumulative(cumulative), ());
}
None => {
if guard
.peek(&FillKey::Cumulative(cumulative.clone()))
.is_some()
{
return true;
}
guard.put(FillKey::Cumulative(cumulative.clone()), ());
guard.put(FillKey::CumulativeWithoutId(cumulative), ());
}
}
false
}
fn record_execution(cache: &SharedDedupCache, execution_id: &str) {
let execution = FillKey::Execution(SmolStr::new(execution_id));
cache.lock().put(execution, ());
}
#[derive(Clone, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AlpacaConfig {
api_key: String,
secret_key: String,
#[serde(default = "default_paper")]
pub paper: bool,
#[cfg(test)]
#[serde(skip)]
base_url_override: Option<String>,
#[cfg(test)]
#[serde(skip)]
ws_url_override: Option<String>,
}
fn default_paper() -> bool {
true
}
impl std::fmt::Debug for AlpacaConfig {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("AlpacaConfig")
.field("api_key", &"***")
.field("secret_key", &"***")
.field("paper", &self.paper)
.finish()
}
}
impl AlpacaConfig {
pub fn new(api_key: String, secret_key: String) -> Self {
Self::paper(api_key, secret_key)
}
pub fn paper(api_key: String, secret_key: String) -> Self {
Self {
api_key,
secret_key,
paper: true,
#[cfg(test)]
base_url_override: None,
#[cfg(test)]
ws_url_override: None,
}
}
pub fn production(api_key: String, secret_key: String) -> Self {
Self {
api_key,
secret_key,
paper: false,
#[cfg(test)]
base_url_override: None,
#[cfg(test)]
ws_url_override: None,
}
}
pub fn from_env() -> Result<Self, AlpacaConfigError> {
let api_key = match std::env::var("ALPACA_API_KEY") {
Ok(value) => value,
Err(std::env::VarError::NotPresent) => return Err(AlpacaConfigError::MissingApiKey),
Err(std::env::VarError::NotUnicode(_)) => {
return Err(AlpacaConfigError::InvalidApiKey);
}
};
let secret_key = match std::env::var("ALPACA_SECRET_KEY") {
Ok(value) => value,
Err(std::env::VarError::NotPresent) => return Err(AlpacaConfigError::MissingSecretKey),
Err(std::env::VarError::NotUnicode(_)) => {
return Err(AlpacaConfigError::InvalidSecretKey);
}
};
let paper = match std::env::var("ALPACA_PAPER") {
Ok(value) => parse_env_bool(&value).ok_or(AlpacaConfigError::InvalidPaper(value))?,
Err(std::env::VarError::NotPresent) => true,
Err(std::env::VarError::NotUnicode(value)) => {
return Err(AlpacaConfigError::InvalidPaper(
value.to_string_lossy().into_owned(),
));
}
};
if paper {
Ok(Self::paper(api_key, secret_key))
} else {
Ok(Self::production(api_key, secret_key))
}
}
#[cfg(test)]
pub fn with_base_url(api_key: String, secret_key: String, base_url: String) -> Self {
Self {
api_key,
secret_key,
paper: true,
base_url_override: Some(base_url),
ws_url_override: None,
}
}
pub fn api_key(&self) -> &str {
&self.api_key
}
pub fn rest_base_url(&self) -> &str {
#[cfg(test)]
if let Some(ref url) = self.base_url_override {
return url.as_str();
}
if self.paper {
"https://paper-api.alpaca.markets"
} else {
"https://api.alpaca.markets"
}
}
pub fn ws_url(&self) -> &'static str {
if self.paper {
"wss://paper-api.alpaca.markets/stream"
} else {
"wss://api.alpaca.markets/stream"
}
}
fn stream_url(&self) -> &str {
#[cfg(test)]
if let Some(ref url) = self.ws_url_override {
return url.as_str();
}
self.ws_url()
}
}
#[derive(Debug, PartialEq, thiserror::Error)]
pub enum AlpacaConfigError {
#[error("ALPACA_API_KEY environment variable not set")]
MissingApiKey,
#[error("ALPACA_API_KEY environment variable is not valid UTF-8")]
InvalidApiKey,
#[error("ALPACA_SECRET_KEY environment variable not set")]
MissingSecretKey,
#[error("ALPACA_SECRET_KEY environment variable is not valid UTF-8")]
InvalidSecretKey,
#[error("ALPACA_PAPER must be true or false, got {0}")]
InvalidPaper(String),
}
#[derive(Debug, Deserialize)]
struct AlpacaAccount {
#[serde(with = "rust_decimal::serde::str")]
cash: Decimal,
#[serde(with = "rust_decimal::serde::str")]
non_marginable_buying_power: Decimal,
}
#[derive(Debug, Deserialize)]
struct AlpacaPosition {
symbol: String,
asset_class: String,
#[serde(default)]
side: AlpacaPositionSide,
#[serde(with = "rust_decimal::serde::str")]
qty: Decimal,
#[serde(with = "rust_decimal::serde::str")]
qty_available: Decimal,
#[serde(default, with = "rust_decimal::serde::str_option")]
avg_entry_price: Option<Decimal>,
#[serde(default, with = "rust_decimal::serde::str_option")]
unrealized_pl: Option<Decimal>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Deserialize)]
#[serde(rename_all = "lowercase")]
enum AlpacaPositionSide {
Long,
Short,
#[default]
#[serde(other)]
Unknown,
}
#[derive(Debug, Deserialize)]
struct AlpacaOrderResponse {
id: String,
client_order_id: Option<String>,
symbol: String,
qty: Option<String>,
filled_qty: String,
#[serde(default)]
filled_avg_price: Option<String>,
#[serde(default)]
status: Option<String>,
side: String,
#[serde(rename = "type")]
order_type: String,
time_in_force: String,
limit_price: Option<String>,
stop_price: Option<String>,
trail_percent: Option<String>,
trail_price: Option<String>,
created_at: String,
updated_at: Option<String>,
}
#[derive(Debug, Deserialize)]
struct AlpacaActivity {
id: String,
order_id: String,
symbol: String,
side: String,
price: String,
qty: String,
transaction_time: String,
cum_qty: Option<String>,
}
#[derive(Debug, Deserialize)]
struct AlpacaNonTradeActivity {
id: String,
activity_type: String,
#[serde(default)]
activity_sub_type: Option<String>,
#[serde(default)]
created_at: Option<String>,
#[serde(default)]
date: Option<String>,
#[serde(default)]
net_amount: Option<String>,
#[serde(default)]
qty: Option<String>,
#[serde(default)]
price: Option<String>,
#[serde(default)]
currency: Option<String>,
#[serde(default)]
symbol: Option<String>,
#[serde(default)]
status: Option<String>,
}
#[derive(Debug, Deserialize)]
struct AlpacaApiError {
code: Option<i64>,
message: String,
}
impl AlpacaApiError {
fn from_body(bytes: &[u8]) -> Self {
serde_json::from_slice(bytes).unwrap_or_else(|_| Self {
code: None,
message: String::from_utf8_lossy(bytes).into_owned(),
})
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum AlpacaPositionIntent {
BuyToOpen,
BuyToClose,
SellToOpen,
SellToClose,
}
#[derive(Debug, Serialize)]
struct TakeProfitParams {
limit_price: String,
}
#[derive(Debug, Serialize)]
struct StopLossParams {
stop_price: String,
#[serde(skip_serializing_if = "Option::is_none")]
limit_price: Option<String>,
}
#[non_exhaustive]
#[derive(Debug, Clone)]
pub struct AlpacaBracketOrderRequest {
pub instrument: InstrumentNameExchange,
pub strategy: StrategyId,
pub cid: ClientOrderId,
pub side: Side,
pub quantity: Decimal,
pub entry_price: Decimal,
pub take_profit_price: Decimal,
pub stop_loss_price: Decimal,
pub stop_loss_limit_price: Option<Decimal>,
pub time_in_force: TimeInForce,
}
impl AlpacaBracketOrderRequest {
#[allow(clippy::too_many_arguments)] pub fn new(
instrument: InstrumentNameExchange,
strategy: StrategyId,
cid: ClientOrderId,
side: Side,
quantity: Decimal,
entry_price: Decimal,
take_profit_price: Decimal,
stop_loss_price: Decimal,
time_in_force: TimeInForce,
) -> Self {
Self {
instrument,
strategy,
cid,
side,
quantity,
entry_price,
take_profit_price,
stop_loss_price,
stop_loss_limit_price: None,
time_in_force,
}
}
#[must_use]
pub fn with_stop_loss_limit_price(mut self, price: Decimal) -> Self {
self.stop_loss_limit_price = Some(price);
self
}
}
#[non_exhaustive]
#[derive(Debug, Clone)]
pub struct AlpacaBracketOrderResult {
pub parent: Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState>,
}
#[derive(Debug, Serialize)]
struct AlpacaOrderRequest<'a> {
symbol: &'a str,
qty: String,
side: &'static str,
#[serde(rename = "type")]
order_type: &'static str,
time_in_force: &'static str,
#[serde(skip_serializing_if = "Option::is_none")]
limit_price: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
stop_price: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
trail_percent: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
trail_price: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
client_order_id: Option<&'a str>,
#[serde(skip_serializing_if = "Option::is_none")]
position_intent: Option<AlpacaPositionIntent>,
#[serde(skip_serializing_if = "Option::is_none")]
order_class: Option<&'static str>,
#[serde(skip_serializing_if = "Option::is_none")]
take_profit: Option<TakeProfitParams>,
#[serde(skip_serializing_if = "Option::is_none")]
stop_loss: Option<StopLossParams>,
}
#[derive(Debug, Deserialize)]
struct AlpacaStreamMessage<'a> {
stream: SmolStr,
#[serde(borrow)]
data: &'a serde_json::value::RawValue,
}
#[derive(Debug, Deserialize)]
struct AlpacaTradeUpdate<'a> {
event: SmolStr,
#[serde(borrow)]
order: AlpacaOrderWs<'a>,
#[serde(borrow)]
price: Option<&'a str>,
#[serde(borrow)]
qty: Option<&'a str>,
#[serde(borrow)]
timestamp: Option<&'a str>,
#[serde(borrow)]
execution_id: Option<&'a str>,
#[serde(borrow)]
previous_execution_id: Option<&'a str>,
}
#[derive(Debug, Deserialize)]
struct AlpacaOrderWs<'a> {
id: SmolStr,
client_order_id: Option<SmolStr>,
symbol: SmolStr,
#[serde(borrow)]
qty: Option<&'a str>,
#[serde(borrow)]
filled_qty: Option<&'a str>,
side: SmolStr,
#[serde(rename = "type")]
order_type: SmolStr,
time_in_force: SmolStr,
#[serde(borrow)]
limit_price: Option<&'a str>,
#[serde(borrow)]
stop_price: Option<&'a str>,
#[serde(borrow)]
trail_percent: Option<&'a str>,
#[serde(borrow)]
trail_price: Option<&'a str>,
status: SmolStr,
#[serde(borrow, default)]
updated_at: Option<&'a str>,
}
#[derive(Clone)]
pub struct AlpacaClient {
config: Arc<AlpacaConfig>,
http: reqwest::Client,
rate_limiter: Arc<RateLimitTracker>,
orders_url: String,
known_live: SharedKnownLiveOrders,
}
impl std::fmt::Debug for AlpacaClient {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("AlpacaClient")
.field("paper", &self.config.paper)
.finish_non_exhaustive()
}
}
impl AlpacaClient {
#[allow(clippy::expect_used)] fn build_http(config: &AlpacaConfig) -> reqwest::Client {
use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
let mut headers = HeaderMap::new();
headers.insert(
HeaderName::from_static("apca-api-key-id"),
HeaderValue::from_str(&config.api_key)
.expect("Alpaca API key contains invalid header characters"),
);
headers.insert(
HeaderName::from_static("apca-api-secret-key"),
HeaderValue::from_str(&config.secret_key)
.expect("Alpaca secret key contains invalid header characters"),
);
reqwest::Client::builder()
.default_headers(headers)
.build()
.expect("failed to build reqwest client for Alpaca")
}
fn base_url(&self) -> &str {
self.config.rest_base_url()
}
fn remember_live<'a>(
&self,
orders: impl IntoIterator<Item = &'a Order<ExchangeId, InstrumentNameExchange, Open>>,
) {
let mut known = self.known_live.lock();
for order in orders {
known.live(&order.key, order.quantity, order.kind, &order.state);
}
}
}
fn parse_rate_limit_delay(headers: &reqwest::header::HeaderMap) -> Option<Duration> {
let header = |name: &str| {
headers
.get(name)
.and_then(|value| value.to_str().ok())
.and_then(|value| value.trim().parse::<u64>().ok())
};
let seconds = match header("retry-after") {
Some(seconds) => seconds,
None => {
let reset = header("x-ratelimit-reset")?;
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
reset.saturating_sub(now).max(1)
}
};
Some(Duration::from_secs(
seconds.min(DEFAULT_RATE_LIMIT_DELAY_SECS),
))
}
fn observe_rate_limit_remaining(
rate_limiter: &RateLimitTracker,
headers: &reqwest::header::HeaderMap,
) {
let exhausted = headers
.get("x-ratelimit-remaining")
.and_then(|v| v.to_str().ok())
.and_then(|s| s.parse::<u32>().ok())
== Some(0);
if !exhausted {
return;
}
match parse_rate_limit_delay(headers) {
Some(reset) => rate_limiter.on_bucket_exhausted(reset),
None => debug!(
"Alpaca REST rate-limit window exhausted (X-Ratelimit-Remaining: 0) without X-Ratelimit-Reset"
),
}
}
async fn rest_with_retry<T>(
rate_limiter: &RateLimitTracker,
build_request: impl FnMut() -> reqwest::RequestBuilder,
) -> Result<T, UnindexedClientError>
where
T: for<'de> Deserialize<'de>,
{
match rest_request(rate_limiter, build_request).await? {
Fetched::Found(value) => Ok(value),
Fetched::NotFound(message) => Err(UnindexedClientError::Api(parse_api_error(
reqwest::StatusCode::NOT_FOUND,
None,
&message,
))),
}
}
async fn rest_lookup_with_retry<T>(
rate_limiter: &RateLimitTracker,
build_request: impl FnMut() -> reqwest::RequestBuilder,
) -> Result<Option<T>, UnindexedClientError>
where
T: for<'de> Deserialize<'de>,
{
match rest_request(rate_limiter, build_request).await? {
Fetched::Found(value) => Ok(Some(value)),
Fetched::NotFound(_) => Ok(None),
}
}
#[derive(Debug)]
enum Fetched<T> {
Found(T),
NotFound(String),
}
async fn rest_request<T>(
rate_limiter: &RateLimitTracker,
mut build_request: impl FnMut() -> reqwest::RequestBuilder,
) -> Result<Fetched<T>, UnindexedClientError>
where
T: for<'de> Deserialize<'de>,
{
for attempt in 0..MAX_RATE_LIMIT_ATTEMPTS {
rate_limiter.wait_if_blocked().await;
let response = build_request()
.send()
.await
.map_err(|e| connectivity_err(format!("Alpaca REST request failed: {e}")))?;
let status = response.status();
if status == reqwest::StatusCode::TOO_MANY_REQUESTS {
let reset_delay = parse_rate_limit_delay(response.headers());
rate_limiter.on_rate_limited(reset_delay);
if attempt + 1 < MAX_RATE_LIMIT_ATTEMPTS {
warn!(
attempt = attempt + 1,
max_attempts = MAX_RATE_LIMIT_ATTEMPTS,
"Alpaca REST rate-limited (429), retrying"
);
continue;
}
warn!(
max_attempts = MAX_RATE_LIMIT_ATTEMPTS,
"Alpaca REST rate-limit retries exhausted"
);
return Err(UnindexedClientError::Api(ApiError::rate_limited(
reset_delay,
)));
}
observe_rate_limit_remaining(rate_limiter, response.headers());
if status == reqwest::StatusCode::NO_CONTENT {
return Err(UnindexedClientError::Internal(
"Alpaca REST returned 204 No Content — use rest_delete_with_retry for DELETE endpoints"
.to_string(),
));
}
let bytes = response
.bytes()
.await
.map_err(|e| connectivity_err(format!("Alpaca REST read body failed: {e}")))?;
if status.is_success() {
return serde_json::from_slice::<T>(&bytes)
.map(Fetched::Found)
.map_err(|e| {
UnindexedClientError::Internal(format!(
"Alpaca REST JSON parse error ({status}): {e} | body: {}",
String::from_utf8_lossy(&bytes)
.chars()
.take(200)
.collect::<String>()
))
});
}
let api_err = AlpacaApiError::from_body(&bytes);
if status == reqwest::StatusCode::NOT_FOUND {
return Ok(Fetched::NotFound(api_err.message));
}
if status.is_client_error() {
return Err(UnindexedClientError::Api(parse_api_error(
status,
api_err.code,
&api_err.message,
)));
}
return Err(connectivity_err(format!(
"Alpaca REST error {status}: {}",
api_err.message
)));
}
unreachable!("Alpaca REST retry loop exited without returning")
}
async fn rest_delete_with_retry(
rate_limiter: &RateLimitTracker,
mut build_request: impl FnMut() -> reqwest::RequestBuilder,
) -> Result<(), UnindexedOrderError> {
for attempt in 0..MAX_RATE_LIMIT_ATTEMPTS {
rate_limiter.wait_if_blocked().await;
let response = build_request().send().await.map_err(|e| {
UnindexedOrderError::Connectivity(ConnectivityError::Socket(format!(
"Alpaca cancel request failed: {e}"
)))
})?;
let status = response.status();
if status == reqwest::StatusCode::TOO_MANY_REQUESTS {
let reset_delay = parse_rate_limit_delay(response.headers());
rate_limiter.on_rate_limited(reset_delay);
if attempt + 1 < MAX_RATE_LIMIT_ATTEMPTS {
warn!(
attempt = attempt + 1,
max_attempts = MAX_RATE_LIMIT_ATTEMPTS,
"Alpaca cancel rate-limited (429), retrying"
);
continue;
}
warn!(
max_attempts = MAX_RATE_LIMIT_ATTEMPTS,
"Alpaca cancel rate-limit retries exhausted"
);
return Err(UnindexedOrderError::Rejected(ApiError::rate_limited(
reset_delay,
)));
}
observe_rate_limit_remaining(rate_limiter, response.headers());
if status == reqwest::StatusCode::NO_CONTENT || status.is_success() {
return Ok(());
}
let bytes = response
.bytes()
.await
.inspect_err(
|e| warn!(%e, %status, "Alpaca cancel_order: failed to read error response body"),
)
.unwrap_or_default();
let api_err = AlpacaApiError::from_body(&bytes);
return Err(parse_order_error(status, api_err.code, &api_err.message));
}
unreachable!("Alpaca cancel retry loop exited without returning")
}
impl ExecutionClient for AlpacaClient {
const EXCHANGE: ExchangeId = ExchangeId::AlpacaBroker;
const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant] = &[
InstrumentKindDiscriminant::Spot,
InstrumentKindDiscriminant::Option,
];
type Config = AlpacaConfig;
type AccountStream = BoxStream<'static, UnindexedAccountEvent>;
fn new(config: Self::Config) -> Self {
let http = Self::build_http(&config);
let orders_url = format!("{}/v2/orders", config.rest_base_url());
Self {
config: Arc::new(config),
http,
rate_limiter: Arc::new(RateLimitTracker::new()),
orders_url,
known_live: KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
}
}
async fn account_snapshot(
&self,
assets: &[AssetNameExchange],
instruments: &[InstrumentNameExchange],
) -> Result<UnindexedAccountSnapshot, UnindexedClientError> {
let base = self.base_url();
let http = self.http.clone();
let rl = &self.rate_limiter;
let wants_usd = assets.is_empty() || assets.iter().any(is_usd);
let wants_non_usd = assets.is_empty() || assets.iter().any(|a| !is_usd(a));
let account_url = format!("{base}/v2/account");
let positions_url = format!("{base}/v2/positions");
let sent = Utc::now();
let (account, positions): (Option<AlpacaAccount>, Vec<AlpacaPosition>) = tokio::try_join!(
async {
if wants_usd {
rest_with_retry(rl, || http.get(&account_url))
.await
.map(Some)
} else {
Ok(None)
}
},
rest_with_retry(rl, || http.get(&positions_url)),
)?;
let mut balances = account
.map(|account| convert_account_to_balances(&account, assets, sent))
.unwrap_or_default();
if wants_non_usd {
balances.extend(convert_positions_to_balances(&positions, assets, sent));
}
let converted_positions = convert_positions(&positions, Utc::now())?;
let open_orders = fetch_raw_open_orders(&http, rl, base, instruments).await?;
let instrument_snapshots =
build_instrument_snapshots(open_orders, converted_positions, instruments);
{
let mut known = self.known_live.lock();
for order in instrument_snapshots
.iter()
.flat_map(|snapshot| &snapshot.orders)
{
if let OrderState::Active(ActiveOrderState::Open(open)) = &order.state {
known.live(&order.key, order.quantity, order.kind, open);
}
}
}
Ok(AccountSnapshot::new(
ExchangeId::AlpacaBroker,
balances,
instrument_snapshots,
))
}
async fn account_stream(
&self,
_assets: &[AssetNameExchange], instruments: &[InstrumentNameExchange],
) -> Result<Self::AccountStream, UnindexedClientError> {
let initial_ws = connect_and_subscribe(&self.config).await?;
let (tx, rx) = mpsc::unbounded_channel::<UnindexedAccountEvent>();
let dedup = new_dedup_cache();
let config = self.config.clone();
let http = self.http.clone();
let rate_limiter = self.rate_limiter.clone();
let known = self.known_live.clone();
let instruments = instruments.to_vec();
let cm_handle = tokio::spawn(connection_manager(
tx,
dedup,
config,
http,
rate_limiter,
known,
instruments,
Some(initial_ws),
));
let rx_stream = tokio_stream::wrappers::UnboundedReceiverStream::new(rx);
let guarded = GracefulShutdownStream::new(rx_stream, cm_handle);
Ok(futures::StreamExt::boxed(guarded))
}
async fn cancel_order(
&self,
request: OrderRequestCancel<ExchangeId, &InstrumentNameExchange>,
) -> UnindexedOrderResponseCancel {
let key = crate::order::OrderKey {
exchange: request.key.exchange,
instrument: request.key.instrument.clone(),
strategy: request.key.strategy.clone(),
cid: request.key.cid.clone(),
};
let order_id: SmolStr = match request.state.id.as_ref().and_then(VenueOrderId::assigned) {
Some(id) => id.0.clone(),
None => {
warn!(
instrument = %key.instrument,
"Alpaca cancel_order: no exchange order ID available (clientOrderId-only cancel not supported)"
);
return crate::order::request::OrderResponseCancel {
key,
state: Err(UnindexedOrderError::Rejected(ApiError::OrderRejected(
"exchange order ID required for cancel (fetch_open_orders to resolve)"
.into(),
))),
};
}
};
let base = self.base_url();
let http = self.http.clone();
let url = format!("{base}/v2/orders/{order_id}");
match rest_delete_with_retry(&self.rate_limiter, || http.delete(&url)).await {
Ok(()) => {
let exchange_order_id = OrderId(order_id);
crate::order::request::OrderResponseCancel {
key,
state: Ok(Cancelled::new(exchange_order_id, Utc::now(), None)),
}
}
Err(e) => crate::order::request::OrderResponseCancel { key, state: Err(e) },
}
}
async fn open_order(
&self,
request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
) -> Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState> {
let side = request.state.side;
let reduce_only = request.state.reduce_only;
self.open_order_inner(request, map_position_intent(side, reduce_only))
.await
}
async fn fetch_balances(
&self,
assets: &[AssetNameExchange],
) -> Result<Vec<AssetBalance<AssetNameExchange>>, UnindexedClientError> {
let base = self.base_url();
let http = self.http.clone();
let mut result = Vec::new();
let wants_usd = assets.is_empty() || assets.iter().any(is_usd);
if wants_usd {
let account_url = format!("{base}/v2/account");
let sent = Utc::now();
let account: AlpacaAccount =
rest_with_retry(&self.rate_limiter, || http.get(&account_url)).await?;
result.extend(convert_account_to_balances(&account, assets, sent));
}
let wants_non_usd = assets.is_empty() || assets.iter().any(|a| !is_usd(a));
if wants_non_usd {
let positions_url = format!("{base}/v2/positions");
let sent = Utc::now();
let positions: Vec<AlpacaPosition> =
rest_with_retry(&self.rate_limiter, || http.get(&positions_url)).await?;
result.extend(convert_positions_to_balances(&positions, assets, sent));
}
Ok(result)
}
async fn fetch_open_orders(
&self,
instruments: &[InstrumentNameExchange],
) -> Result<Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, UnindexedClientError> {
let base = self.base_url();
let http = self.http.clone();
let open_orders =
fetch_raw_open_orders(&http, &self.rate_limiter, base, instruments).await?;
let result: Vec<_> = open_orders
.into_iter()
.filter_map(|o| convert_open_order(&o))
.collect();
self.remember_live(&result);
Ok(result)
}
async fn fetch_trades(
&self,
start: DateTime<Utc>,
end: DateTime<Utc>,
instruments: &[InstrumentNameExchange],
) -> Result<TradesRead<AssetNameExchange, InstrumentNameExchange>, UnindexedClientError> {
if start > end || end < DateTime::UNIX_EPOCH {
return Ok(TradesRead::complete(Vec::new()));
}
let page = paginate_activities::<AlpacaActivity>(
&self.http,
&self.rate_limiter,
self.base_url(),
FILL_ACTIVITIES,
start,
Some(end),
)
.await?;
let Ok(resume) = span_resume(page.resume, start, end) else {
error!(
%start,
fills_read = page.activities.len(),
"Alpaca fetch_trades: a full read did not advance past the span's first millisecond"
);
return Err(UnindexedClientError::Truncated {
fills_read: page.activities.len(),
});
};
let instrument_set: fnv::FnvHashSet<&str> =
instruments.iter().map(|i| i.name().as_str()).collect();
let trades = page
.activities
.iter()
.filter(|a| instrument_set.is_empty() || instrument_set.contains(a.symbol.as_str()))
.filter(|a| {
parse_timestamp(&a.transaction_time)
.is_none_or(|time| (start..=end).contains(&time))
})
.filter_map(convert_activity_to_trade)
.collect();
Ok(TradesRead::new(trades, resume))
}
}
impl AlpacaClient {
pub async fn fetch_cash_flows(
&self,
start: DateTime<Utc>,
end: DateTime<Utc>,
) -> Result<CashFlowsRead<AssetNameExchange, InstrumentNameExchange>, UnindexedClientError>
{
if start > end || end < DateTime::UNIX_EPOCH {
return Ok(CashFlowsRead::complete(Vec::new()));
}
let page = paginate_activities::<AlpacaNonTradeActivity>(
&self.http,
&self.rate_limiter,
self.base_url(),
CASH_FLOW_ACTIVITIES,
start,
Some(end),
)
.await?;
let Ok(resume) = span_resume(page.resume, start, end) else {
error!(
%start,
activities_read = page.activities.len(),
"Alpaca fetch_cash_flows: a full read did not advance past the span's first \
millisecond"
);
return Err(UnindexedClientError::Internal(format!(
"Alpaca cash-flow read of {} activities did not advance past {start}",
page.activities.len()
)));
};
let flows = page
.activities
.iter()
.filter_map(convert_cash_flow)
.filter(|flow| (start..=end).contains(&flow.time_exchange))
.collect();
Ok(CashFlowsRead::new(flows, resume))
}
pub async fn fetch_option_events(
&self,
start: DateTime<Utc>,
end: DateTime<Utc>,
) -> Result<TradesRead<AssetNameExchange, InstrumentNameExchange>, UnindexedClientError> {
if start > end || end < DateTime::UNIX_EPOCH {
return Ok(TradesRead::complete(Vec::new()));
}
let read_start = start.date_naive().and_time(NaiveTime::MIN).and_utc();
let read_end = end.date_naive().and_time(LAST_MILLISECOND).and_utc();
let page = paginate_activities::<AlpacaNonTradeActivity>(
&self.http,
&self.rate_limiter,
self.base_url(),
OPTION_EVENT_ACTIVITIES,
read_start,
Some(read_end),
)
.await?;
let Ok(resume) = span_resume(page.resume, start, end) else {
error!(
%start,
activities_read = page.activities.len(),
"Alpaca fetch_option_events: a full read did not advance past the span's start"
);
return Err(UnindexedClientError::Internal(format!(
"Alpaca option-event read of {} activities did not advance past {start}",
page.activities.len()
)));
};
Ok(TradesRead::new(
convert_option_events(&page.activities, start, end),
resume,
))
}
pub async fn open_order_with_intent(
&self,
request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
intent: AlpacaPositionIntent,
) -> Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState> {
self.open_order_inner(request, intent).await
}
pub async fn open_bracket_order(
&self,
request: AlpacaBracketOrderRequest,
) -> AlpacaBracketOrderResult {
let order_key = crate::order::OrderKey::new(
ExchangeId::AlpacaBroker,
request.instrument.clone(),
request.strategy.clone(),
request.cid.clone(),
);
let tif_str = match request.time_in_force {
TimeInForce::GoodUntilEndOfDay => "day",
TimeInForce::GoodUntilCancelled { post_only } => {
if post_only {
return AlpacaBracketOrderResult {
parent: Order {
key: order_key,
side: request.side,
price: Some(request.entry_price),
quantity: request.quantity,
kind: OrderKind::Limit,
time_in_force: request.time_in_force,
state: OrderState::inactive(OrderError::Rejected(
ApiError::OrderRejected(
"Alpaca does not support post_only for bracket orders"
.to_string(),
),
)),
},
};
}
"gtc"
}
other => {
return AlpacaBracketOrderResult {
parent: Order {
key: order_key,
side: request.side,
price: Some(request.entry_price),
quantity: request.quantity,
kind: OrderKind::Limit,
time_in_force: request.time_in_force,
state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
format!(
"Alpaca bracket orders only support Day or GTC time_in_force, got {:?}",
other
),
))),
},
};
}
};
let price_ordering_ok = match request.side {
Side::Buy => {
request.stop_loss_price < request.entry_price
&& request.entry_price < request.take_profit_price
}
Side::Sell => {
request.take_profit_price < request.entry_price
&& request.entry_price < request.stop_loss_price
}
};
if !price_ordering_ok {
return AlpacaBracketOrderResult {
parent: Order {
key: order_key,
side: request.side,
price: Some(request.entry_price),
quantity: request.quantity,
kind: OrderKind::Limit,
time_in_force: request.time_in_force,
state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
format!(
"Invalid bracket price ordering for {:?} side: entry={}, take_profit={}, stop_loss={}",
request.side,
request.entry_price,
request.take_profit_price,
request.stop_loss_price,
),
))),
},
};
}
if let Some(sl_limit) = request.stop_loss_limit_price {
let sl_limit_ok = match request.side {
Side::Buy => sl_limit <= request.stop_loss_price,
Side::Sell => sl_limit >= request.stop_loss_price,
};
if !sl_limit_ok {
return AlpacaBracketOrderResult {
parent: Order {
key: order_key,
side: request.side,
price: Some(request.entry_price),
quantity: request.quantity,
kind: OrderKind::Limit,
time_in_force: request.time_in_force,
state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
format!(
"Invalid stop-loss limit price for {:?} bracket: \
stop_loss_price={}, stop_loss_limit_price={}",
request.side, request.stop_loss_price, sl_limit,
),
))),
},
};
}
}
let body = AlpacaOrderRequest {
symbol: request.instrument.name().as_str(),
qty: request.quantity.to_string(),
side: map_side(request.side),
order_type: "limit",
time_in_force: tif_str,
limit_price: Some(request.entry_price.to_string()),
stop_price: None,
trail_percent: None,
trail_price: None,
client_order_id: Some(request.cid.0.as_str()),
position_intent: if is_options_or_equity_symbol(request.instrument.name().as_str()) {
Some(map_position_intent(request.side, ReduceOnly::No))
} else {
None
},
order_class: Some("bracket"),
take_profit: Some(TakeProfitParams {
limit_price: request.take_profit_price.to_string(),
}),
stop_loss: Some(StopLossParams {
stop_price: request.stop_loss_price.to_string(),
limit_price: request.stop_loss_limit_price.map(|p| p.to_string()),
}),
};
let http = self.http.clone();
let rl = &self.rate_limiter;
let result: Result<AlpacaOrderResponse, UnindexedClientError> =
rest_with_retry(rl, || http.post(&self.orders_url).json(&body)).await;
match result {
Ok(resp) => {
let state = placed_order_state(&resp, &order_key.instrument, request.quantity);
self.known_live.lock().placed(
&order_key,
request.quantity,
OrderKind::Limit,
&state,
);
AlpacaBracketOrderResult {
parent: Order {
key: order_key,
side: request.side,
price: Some(request.entry_price),
quantity: request.quantity,
kind: OrderKind::Limit,
time_in_force: request.time_in_force,
state,
},
}
}
Err(e) => {
let order_err = order_post_error(e);
AlpacaBracketOrderResult {
parent: Order {
key: order_key,
side: request.side,
price: Some(request.entry_price),
quantity: request.quantity,
kind: OrderKind::Limit,
time_in_force: request.time_in_force,
state: OrderState::inactive(order_err),
},
}
}
}
}
async fn open_order_inner(
&self,
request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
intent: AlpacaPositionIntent,
) -> Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState> {
let instrument = request.key.instrument.clone();
let side = request.state.side;
let price = request.state.price;
let quantity = request.state.quantity;
let kind = request.state.kind;
let time_in_force = request.state.time_in_force;
let cid = request.key.cid.clone();
let order_key = crate::order::OrderKey::new(
ExchangeId::AlpacaBroker,
instrument.clone(),
request.key.strategy.clone(),
cid.clone(),
);
if let Some(effect) = request.state.margin_effect {
return Order {
key: order_key,
side,
price,
quantity,
kind,
time_in_force,
state: OrderState::inactive(OrderError::UnsupportedMarginEffect(effect)),
};
}
let tif_str = match map_time_in_force(time_in_force) {
Ok(s) => s,
Err(msg) => {
return Order {
key: order_key,
side,
price,
quantity,
kind,
time_in_force,
state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
msg.to_string(),
))),
};
}
};
let order_type_str = match map_order_kind(kind) {
Some(s) => s,
None => {
return Order {
key: order_key,
side,
price,
quantity,
kind,
time_in_force,
state: OrderState::inactive(OrderError::UnsupportedOrderType(format!(
"Alpaca connector does not yet support OrderKind::{kind:?}"
))),
};
}
};
if let OrderKind::TrailingStop {
offset_type: TrailingOffsetType::BasisPoints,
..
} = kind
{
return Order {
key: order_key,
side,
price,
quantity,
kind,
time_in_force,
state: OrderState::inactive(OrderError::UnsupportedOrderType(
"Alpaca does not support TrailingOffsetType::BasisPoints; \
use Percentage or Absolute"
.to_string(),
)),
};
}
if matches!(kind, OrderKind::StopLimit { .. }) && price.is_none() {
return Order {
key: order_key,
side,
price,
quantity,
kind,
time_in_force,
state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
"StopLimit order requires Order.price (the limit price) to be set".to_string(),
))),
};
}
let (stop_price, trail_percent, trail_price) = match kind {
OrderKind::Stop { trigger_price } | OrderKind::StopLimit { trigger_price } => {
(Some(trigger_price.to_string()), None, None)
}
OrderKind::TrailingStop {
offset,
offset_type,
..
} => match offset_type {
TrailingOffsetType::Percentage => (None, Some(offset.to_string()), None),
TrailingOffsetType::Absolute => (None, None, Some(offset.to_string())),
TrailingOffsetType::BasisPoints => unreachable!("validated above"),
},
OrderKind::Market
| OrderKind::Limit
| OrderKind::TakeProfit { .. }
| OrderKind::TakeProfitLimit { .. }
| OrderKind::TrailingStopLimit { .. }
| OrderKind::TrailingStopFixedLimit { .. }
| OrderKind::ConditionalUnresolved { .. } => (None, None, None),
};
let body = AlpacaOrderRequest {
symbol: instrument.name().as_str(),
qty: quantity.to_string(),
side: map_side(side),
order_type: order_type_str,
time_in_force: tif_str,
limit_price: match kind {
OrderKind::Limit | OrderKind::StopLimit { .. } => price.map(|p| p.to_string()),
_ => None,
},
stop_price,
trail_percent,
trail_price,
client_order_id: Some(cid.0.as_str()),
position_intent: if is_options_or_equity_symbol(instrument.name().as_str()) {
Some(intent)
} else {
None
},
order_class: None,
take_profit: None,
stop_loss: None,
};
let http = self.http.clone();
let rl = &self.rate_limiter;
let result: Result<AlpacaOrderResponse, UnindexedClientError> =
rest_with_retry(rl, || http.post(&self.orders_url).json(&body)).await;
match result {
Ok(resp) => {
let state = placed_order_state(&resp, &order_key.instrument, quantity);
self.known_live
.lock()
.placed(&order_key, quantity, kind, &state);
Order {
key: order_key,
side,
price,
quantity,
kind,
time_in_force,
state,
}
}
Err(e) => {
let order_err = order_post_error(e);
Order {
key: order_key,
side,
price,
quantity,
kind,
time_in_force,
state: OrderState::inactive(order_err),
}
}
}
}
}
const MAX_OPEN_ORDERS: usize = 500;
async fn fetch_raw_open_orders(
http: &reqwest::Client,
rate_limiter: &RateLimitTracker,
base: &str,
instruments: &[InstrumentNameExchange],
) -> Result<Vec<AlpacaOrderResponse>, UnindexedClientError> {
let orders: Vec<AlpacaOrderResponse> = if instruments.is_empty() {
rest_with_retry(rate_limiter, || {
http.get(format!("{base}/v2/orders"))
.query(&[("status", "open"), ("limit", "500")])
})
.await?
} else {
let symbols = instruments.iter().map(|i| i.name().as_str()).join(",");
rest_with_retry(rate_limiter, || {
http.get(format!("{base}/v2/orders")).query(&[
("status", "open"),
("limit", "500"),
("symbols", &symbols),
])
})
.await?
};
if orders.len() == MAX_OPEN_ORDERS {
warn!(
limit = MAX_OPEN_ORDERS,
"Alpaca fetch_raw_open_orders: received exactly {MAX_OPEN_ORDERS} results — \
response is likely truncated"
);
return Err(UnindexedClientError::TruncatedSnapshot {
limit: MAX_OPEN_ORDERS,
});
}
Ok(orders)
}
const MAX_ACTIVITY_PAGES: usize = 50;
trait PagedActivity: DeserializeOwned {
fn id(&self) -> &str;
fn time(&self) -> Option<DateTime<Utc>>;
}
impl PagedActivity for AlpacaActivity {
fn id(&self) -> &str {
&self.id
}
fn time(&self) -> Option<DateTime<Utc>> {
parse_timestamp(&self.transaction_time)
}
}
impl PagedActivity for AlpacaNonTradeActivity {
fn id(&self) -> &str {
&self.id
}
fn time(&self) -> Option<DateTime<Utc>> {
self.created_at.as_deref().and_then(parse_timestamp)
}
}
const FILL_ACTIVITIES: &str = "FILL";
const CASH_FLOW_ACTIVITIES: &str = "INT,FEE,PTC,PTR";
const OPTION_EVENT_ACTIVITIES: &str = "OPASN,OPEXC,OPEXP,OPTRD";
const LAST_MILLISECOND: NaiveTime = match NaiveTime::from_hms_milli_opt(23, 59, 59, 999) {
Some(time) => time,
None => panic!("23:59:59.999 is a time of day"),
};
fn span_resume(
resume: Option<DateTime<Utc>>,
start: DateTime<Utc>,
end: DateTime<Utc>,
) -> Result<Option<DateTime<Utc>>, ()> {
let resume = resume.filter(|resume| *resume <= end);
if resume.is_some_and(|resume| resume <= floor_millis(start)) {
return Err(());
}
Ok(resume)
}
struct ActivityPage<A> {
activities: Vec<A>,
resume: Option<DateTime<Utc>>,
}
fn floor_millis(time: DateTime<Utc>) -> DateTime<Utc> {
DateTime::from_timestamp_millis(time.timestamp_millis()).unwrap_or(time)
}
const PAGE_SIZE_STR: &str = "100"; const _: () = assert!(
ALPACA_MAX_ACTIVITIES == 100,
"PAGE_SIZE_STR must be updated to match ALPACA_MAX_ACTIVITIES",
);
async fn paginate_activities<A: PagedActivity>(
http: &reqwest::Client,
rate_limiter: &RateLimitTracker,
base: &str,
types: &str,
start: DateTime<Utc>,
end: Option<DateTime<Utc>>,
) -> Result<ActivityPage<A>, UnindexedClientError> {
let after =
floor_millis(start.max(DateTime::UNIX_EPOCH)).to_rfc3339_opts(SecondsFormat::Millis, true);
let until = end.filter(|end| *end < Utc::now()).map(|end| {
(floor_millis(end) + TimeDelta::milliseconds(1))
.to_rfc3339_opts(SecondsFormat::Millis, true)
});
let mut all = Vec::with_capacity(ALPACA_MAX_ACTIVITIES);
let mut page_token: Option<String> = None;
let mut pages = 0usize;
let mut truncated = false;
loop {
if pages >= MAX_ACTIVITY_PAGES {
truncated = true;
break;
}
pages += 1;
let page_token_ref = page_token.as_deref();
let activities: Vec<A> = rest_with_retry(rate_limiter, || {
let mut req = http.get(format!("{base}/v2/account/activities")).query(&[
("activity_types", types),
("after", after.as_str()),
("page_size", PAGE_SIZE_STR),
("direction", "asc"),
]);
if let Some(until) = until.as_deref() {
req = req.query(&[("until", until)]);
}
if let Some(token) = page_token_ref {
req = req.query(&[("page_token", token)]);
}
req
})
.await?;
let page_len = activities.len();
let page_token_candidate = activities.last().map(|a| a.id().to_owned());
all.extend(activities);
if page_len < ALPACA_MAX_ACTIVITIES {
break;
}
match page_token_candidate {
Some(token) if !token.is_empty() => {
debug!("Alpaca paginate_activities: fetching next page ({page_len} results)");
page_token = Some(token);
}
_ => break,
}
}
let resume = truncated.then(|| {
floor_millis(
all.iter()
.rev()
.find_map(PagedActivity::time)
.unwrap_or(start),
)
});
Ok(ActivityPage {
activities: all,
resume,
})
}
#[allow(clippy::cognitive_complexity)] async fn connection_manager(
tx: mpsc::UnboundedSender<UnindexedAccountEvent>,
dedup: SharedDedupCache,
config: Arc<AlpacaConfig>,
http: reqwest::Client,
rate_limiter: Arc<RateLimitTracker>,
known: SharedKnownLiveOrders,
instruments: Vec<InstrumentNameExchange>,
initial_ws: Option<WebSocket>,
) {
let mut budget = ReconnectBudget::new("Alpaca", BackoffLimits::ACCOUNT_STREAM);
let mut disconnected: Option<HostMoment> = None;
let mut clock = VenueClock::new();
let mut unchecked = UncheckedOrders::default();
let mut current_ws = initial_ws;
'outer: loop {
let mut ws = match current_ws.take() {
Some(ws) => ws,
None => match connect_and_subscribe(&config).await {
Ok(ws) => ws,
Err(e) => {
error!(%e, "Alpaca WS connect/subscribe failed");
if !budget.wait().await {
error!("Alpaca max reconnect attempts exhausted");
emit_stream_terminated(
&tx,
ExchangeId::AlpacaBroker,
StreamTerminationReason::ReconnectBudgetExhausted {
attempts: budget.attempts(),
last_error: e.to_string(),
},
);
break;
}
continue;
}
},
};
info!("Alpaca account_stream connected and subscribed");
if let Some(last) = disconnected.take() {
let end = clock.span_end(last, HostMoment::now());
recover_fills_or_report(
&http,
&rate_limiter,
&instruments,
config.rest_base_url(),
fill_recovery_start(&clock, last),
end,
Duration::from_secs(FILL_RECOVERY_TIMEOUT_SECS),
&tx,
&dedup,
&known,
)
.await;
let held = {
let known = known.lock();
if instruments.is_empty() {
known.instruments()
} else {
known.instruments_among(&instruments)
}
};
unchecked.open(held);
}
let mut last_message = HostMoment::now();
let heartbeat = tokio::time::sleep(Duration::from_secs(HEARTBEAT_TIMEOUT_SECS));
tokio::pin!(heartbeat);
let stable_from = tokio::time::Instant::now() + Duration::from_secs(HEARTBEAT_TIMEOUT_SECS);
let order_checks =
run_order_checks(&http, &rate_limiter, &config, &known, &mut unchecked, &tx);
tokio::pin!(order_checks);
macro_rules! reset_heartbeat {
() => {
let now = tokio::time::Instant::now();
heartbeat
.as_mut()
.reset(now + Duration::from_secs(HEARTBEAT_TIMEOUT_SECS));
last_message = HostMoment::new(Utc::now(), now);
if now >= stable_from {
budget.reset();
}
};
}
enum DisconnectReason {
ServerClose,
Error(String),
StreamEnded,
HeartbeatTimeout,
}
let reason = loop {
tokio::select! {
msg = ws.next() => {
match msg {
Some(Ok(WsMessage::Ping(_))) => {
reset_heartbeat!();
}
Some(Ok(WsMessage::Text(text))) => {
process_ws_text(text.as_str(), &tx, &dedup, &known, &mut clock);
reset_heartbeat!();
}
Some(Ok(WsMessage::Binary(bytes))) => {
match std::str::from_utf8(&bytes) {
Ok(text) => {
process_ws_text(text, &tx, &dedup, &known, &mut clock);
reset_heartbeat!();
}
Err(e) => warn!(%e, "Alpaca WS binary frame: not valid UTF-8"),
}
}
Some(Ok(WsMessage::Close(frame))) => {
warn!(frame = ?frame, "Alpaca WS closed by server");
break DisconnectReason::ServerClose;
}
Some(Ok(_)) => {} Some(Err(e)) => {
warn!(%e, "Alpaca WS error, reconnecting");
break DisconnectReason::Error(e.to_string());
}
None => {
warn!("Alpaca WS stream ended, reconnecting");
break DisconnectReason::StreamEnded;
}
}
}
() = &mut order_checks => {}
_ = &mut heartbeat => {
warn!(
timeout_secs = HEARTBEAT_TIMEOUT_SECS,
"Alpaca heartbeat timeout, reconnecting"
);
break DisconnectReason::HeartbeatTimeout;
}
_ = tx.closed() => {
debug!("Alpaca account_stream consumer dropped, terminating");
let _ = tokio::time::timeout(
Duration::from_secs(WS_CLOSE_TIMEOUT_SECS),
ws.close(None),
).await;
break 'outer;
}
}
};
disconnected = Some(last_message);
let _ =
tokio::time::timeout(Duration::from_secs(WS_CLOSE_TIMEOUT_SECS), ws.close(None)).await;
if tx.is_closed() {
break;
}
if !budget.wait().await {
error!("Alpaca max reconnect attempts exhausted, stream terminating");
let last_error = match reason {
DisconnectReason::HeartbeatTimeout => {
format!("heartbeat timeout ({HEARTBEAT_TIMEOUT_SECS}s)")
}
DisconnectReason::ServerClose => "WebSocket closed by server".to_string(),
DisconnectReason::Error(e) => e,
DisconnectReason::StreamEnded => "WebSocket stream ended".to_string(),
};
emit_stream_terminated(
&tx,
ExchangeId::AlpacaBroker,
StreamTerminationReason::ReconnectBudgetExhausted {
attempts: budget.attempts(),
last_error,
},
);
break;
}
}
}
#[derive(Debug)]
enum HandshakeError {
Transport(String),
Auth(String),
}
async fn connect_and_subscribe(config: &AlpacaConfig) -> Result<WebSocket, UnindexedClientError> {
let url = config.stream_url();
debug!(%url, "Alpaca: connecting to WebSocket");
let mut ws = rustrade_integration::protocol::websocket::connect(url)
.await
.map_err(|e| {
UnindexedClientError::Connectivity(ConnectivityError::Socket(format!(
"WS connect: {e}"
)))
})?;
let result = tokio::time::timeout(
Duration::from_secs(WS_HANDSHAKE_TIMEOUT_SECS),
ws_handshake(&mut ws, config),
)
.await;
match result {
Ok(Ok(())) => Ok(ws),
Ok(Err(HandshakeError::Transport(e))) => {
let _ = ws.close(None).await;
Err(UnindexedClientError::Connectivity(
ConnectivityError::Socket(e),
))
}
Ok(Err(HandshakeError::Auth(e))) => {
let _ = ws.close(None).await;
Err(UnindexedClientError::Api(ApiError::Unauthenticated(e)))
}
Err(_) => {
let _ = ws.close(None).await;
Err(UnindexedClientError::Connectivity(
ConnectivityError::Timeout,
))
}
}
}
async fn ws_handshake(ws: &mut WebSocket, config: &AlpacaConfig) -> Result<(), HandshakeError> {
let auth = serde_json::json!({
"action": "auth",
"key": config.api_key(),
"secret": config.secret_key,
})
.to_string();
ws.send(WsMessage::Text(auth.into()))
.await
.map_err(|e| HandshakeError::Transport(format!("WS auth send: {e}")))?;
loop {
match ws.next().await {
Some(Ok(WsMessage::Text(text))) => {
if let Some(result) = check_auth_response(text.as_str()) {
result?;
break;
}
}
Some(Ok(WsMessage::Binary(bytes))) => {
if let Ok(text) = std::str::from_utf8(&bytes)
&& let Some(result) = check_auth_response(text)
{
result?;
break;
}
}
Some(Err(e)) => {
return Err(HandshakeError::Transport(format!(
"WS error during auth: {e}"
)));
}
None => {
return Err(HandshakeError::Transport(
"WS closed before auth response".into(),
));
}
_ => {} }
}
let sub = serde_json::json!({
"action": "listen",
"data": { "streams": ["trade_updates"] }
})
.to_string();
ws.send(WsMessage::Text(sub.into()))
.await
.map_err(|e| HandshakeError::Transport(format!("WS subscribe send: {e}")))?;
loop {
match ws.next().await {
Some(Ok(WsMessage::Text(text))) => {
if check_listen_ack(text.as_str()) {
break;
}
if let Ok(msg) = serde_json::from_str::<AlpacaStreamMessage<'_>>(text.as_str()) {
if msg.stream == "trade_updates" {
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");
} else {
trace!(stream = %msg.stream, "WS message dropped during listen-ack handshake");
}
} else {
trace!(
bytes = text.len(),
"WS non-stream message dropped during listen-ack handshake"
);
}
}
Some(Ok(WsMessage::Binary(bytes))) => {
if let Ok(text) = std::str::from_utf8(&bytes)
&& check_listen_ack(text)
{
break;
}
trace!(
bytes = bytes.len(),
"WS binary message dropped during listen-ack handshake"
);
}
Some(Err(e)) => {
return Err(HandshakeError::Transport(format!(
"WS error during subscribe: {e}"
)));
}
None => {
return Err(HandshakeError::Transport(
"WS closed before subscribe ack".into(),
));
}
_ => {}
}
}
info!("Alpaca WS authenticated and subscribed to trade_updates");
Ok(())
}
fn check_auth_response(text: &str) -> Option<Result<(), HandshakeError>> {
let msg = serde_json::from_str::<AlpacaStreamMessage<'_>>(text).ok()?;
if msg.stream != "authorization" {
return None;
}
#[derive(Deserialize)]
struct AuthData<'a> {
status: &'a str,
}
let data = serde_json::from_str::<AuthData<'_>>(msg.data.get()).ok()?;
if data.status == "authorized" {
Some(Ok(()))
} else {
Some(Err(HandshakeError::Auth(format!(
"Alpaca WS auth failed: status={}",
data.status
))))
}
}
fn check_listen_ack(text: &str) -> bool {
let Ok(msg) = serde_json::from_str::<AlpacaStreamMessage<'_>>(text) else {
return false;
};
if msg.stream != "listening" {
return false;
}
#[derive(Deserialize)]
struct ListenData<'a> {
#[serde(borrow)]
streams: Vec<&'a str>,
}
let Ok(data) = serde_json::from_str::<ListenData<'_>>(msg.data.get()) else {
return false;
};
data.streams.contains(&"trade_updates")
}
fn process_ws_text(
text: &str,
tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
dedup: &SharedDedupCache,
known: &SharedKnownLiveOrders,
clock: &mut VenueClock,
) {
let msg: AlpacaStreamMessage<'_> = match serde_json::from_str(text) {
Ok(m) => m,
Err(e) => {
trace!(
%e,
raw = ?&text[..text.len().min(200)],
"Alpaca WS: skipped non-JSON message"
);
return;
}
};
match msg.stream.as_str() {
"trade_updates" => {
let update: AlpacaTradeUpdate<'_> = match serde_json::from_str(msg.data.get()) {
Ok(u) => u,
Err(e) => {
warn!(
%e,
raw = ?&msg.data.get()[..msg.data.get().len().min(200)],
"Alpaca WS trade_updates: failed to deserialize event — event dropped"
);
return;
}
};
let time = trade_update_time(&update);
if time.1 == TimeSource::Venue {
clock.observe(time.0, tokio::time::Instant::now());
}
if is_fill_event(&update)
&& is_duplicate_fill(dedup, ws_execution_id(&update), early_dedup_key(&update))
{
trace!("Alpaca WS: skipping duplicate fill event (early check)");
return;
}
if is_trade_amendment(&update) {
warn!(
event = %update.event,
raw = %msg.data.get(),
"Alpaca WS: a fill reported earlier was amended — reporting it as TradeAmended"
);
}
for event in convert_trade_update_at(update, time).into_iter().flatten() {
if let AccountEventKind::TradeAmended(amendment) = &event.kind
&& let TradeAmendmentKind::Corrected { replacement } = &amendment.kind
{
record_execution(dedup, &replacement.id.0);
}
let known = KnownLiveOrders::observes(&event.kind).then(|| {
let mut known = known.lock();
known.observe(&event);
known
});
let _ = tx.send(event);
drop(known);
}
}
"authorization" | "listening" => {
trace!(stream = %msg.stream, "Alpaca WS: auth/listen ack received during stream");
}
other => {
trace!(%other, "Alpaca WS: ignoring unknown stream type");
}
}
}
fn fill_dedup_key(order_id: &str, cumulative: Decimal) -> SmolStr {
format_smolstr!("{}:{}", order_id, cumulative.normalize())
}
fn is_trade_amendment(update: &AlpacaTradeUpdate<'_>) -> bool {
matches!(update.event.as_str(), "trade_bust" | "trade_correct")
}
#[inline]
fn is_fill_event(update: &AlpacaTradeUpdate<'_>) -> bool {
matches!(update.event.as_str(), "fill" | "partial_fill")
}
fn ws_execution_id<'a>(update: &AlpacaTradeUpdate<'a>) -> Option<&'a str> {
update.execution_id.filter(|id| !id.is_empty())
}
fn early_dedup_key(update: &AlpacaTradeUpdate<'_>) -> SmolStr {
let filled_qty = update.order.filled_qty.unwrap_or("0");
let qty = Decimal::from_str(filled_qty).unwrap_or(Decimal::ZERO);
fill_dedup_key(&update.order.id, qty)
}
fn activity_execution_id(activity_id: &str) -> &str {
activity_execution_id_part(activity_id).unwrap_or(activity_id)
}
fn activity_execution_id_part(activity_id: &str) -> Option<&str> {
activity_id
.split_once("::")
.map(|(_, execution_id)| execution_id)
.filter(|execution_id| !execution_id.is_empty())
}
#[derive(Debug, PartialEq)]
struct UnreadFills {
start: DateTime<Utc>,
reason: FillRecoveryFailure,
}
fn fill_recovery_start(clock: &VenueClock, last: HostMoment) -> DateTime<Utc> {
let earlier = |time: DateTime<Utc>, by: TimeDelta| {
time.checked_sub_signed(by)
.unwrap_or(DateTime::<Utc>::MIN_UTC)
};
let at = clock
.span_start(last)
.unwrap_or_else(|| earlier(last.time, UNSAMPLED_CLOCK_SKEW));
earlier(at, TimeDelta::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS))
}
async fn recover_fills_or_report(
http: &reqwest::Client,
rate_limiter: &RateLimitTracker,
instruments: &[InstrumentNameExchange],
base: &str,
start: DateTime<Utc>,
end: DateTime<Utc>,
timeout: Duration,
tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
dedup: &SharedDedupCache,
known: &SharedKnownLiveOrders,
) {
let outcome = match tokio::time::timeout(
timeout,
recover_fills(
http,
rate_limiter,
instruments,
base,
start,
tx,
dedup,
known,
),
)
.await
{
Ok(outcome) => outcome,
Err(_) => {
warn!(
timeout_secs = timeout.as_secs(),
instruments = ?instruments
.iter()
.map(|instrument| instrument.name().as_str())
.collect::<Vec<_>>(),
"Alpaca fill recovery timed out — fills since the disconnect may be missing for \
any of the instruments (every instrument when the list is empty)"
);
Err(UnreadFills {
start,
reason: FillRecoveryFailure::TimedOut {
timeout_secs: timeout.as_secs(),
},
})
}
};
if let Err(unread) = outcome
&& tx
.send(fill_recovery_gave_up(instruments, unread, end))
.is_err()
{
debug!("Alpaca fill recovery: consumer dropped before the give-up was sent");
}
}
fn fill_recovery_gave_up(
instruments: &[InstrumentNameExchange],
unread: UnreadFills,
end: DateTime<Utc>,
) -> UnindexedAccountEvent {
let UnreadFills { start, reason } = unread;
let scope = if instruments.is_empty() {
FillRecoveryScope::AllInstruments
} else {
FillRecoveryScope::Instruments(instruments.to_vec())
};
UnindexedAccountEvent::new(
ExchangeId::AlpacaBroker,
AccountEventKind::FillRecoveryGaveUp(FillRecoveryGap::new(
scope,
start,
end.max(start),
1,
reason,
)),
)
}
async fn recover_fills(
http: &reqwest::Client,
rate_limiter: &RateLimitTracker,
instruments: &[InstrumentNameExchange],
base: &str,
after: DateTime<Utc>,
tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
dedup: &SharedDedupCache,
known: &SharedKnownLiveOrders,
) -> Result<(), UnreadFills> {
info!(%after, instruments = instruments.len(), "Alpaca recovering fills after reconnect");
let instrument_set: fnv::FnvHashSet<&str> = if instruments.is_empty() {
fnv::FnvHashSet::default()
} else {
instruments.iter().map(|i| i.name().as_str()).collect()
};
let page = match paginate_activities::<AlpacaActivity>(
http,
rate_limiter,
base,
FILL_ACTIVITIES,
after,
None,
)
.await
{
Ok(p) => p,
Err(e) => {
error!(%e, "Alpaca fill recovery: REST request failed");
return Err(UnreadFills {
start: after,
reason: FillRecoveryFailure::Request(e.to_string()),
});
}
};
let activities = page.activities;
let unread = page.resume.map(|start| {
error!(
max_pages = MAX_ACTIVITY_PAGES,
fills_read = activities.len(),
%start,
"Alpaca fill recovery: max page limit reached, truncating — fills from this time on \
are not delivered; read them with fetch_trades"
);
UnreadFills {
start,
reason: FillRecoveryFailure::Truncated {
fills_read: activities.len(),
},
}
});
let mut recovered = 0u32;
let mut duplicates = 0u32;
let mut cumulative_qty: FnvHashMap<&str, Decimal> = FnvHashMap::default();
for activity in &activities {
if !instrument_set.is_empty() && !instrument_set.contains(activity.symbol.as_str()) {
continue;
}
let exec_qty = Decimal::from_str(&activity.qty).unwrap_or(Decimal::ZERO);
let cum = cumulative_qty
.entry(activity.order_id.as_str())
.or_default();
*cum += exec_qty;
let cumulative = *cum;
let trade = match convert_activity_to_trade(activity) {
Some(t) => t,
None => {
warn!(id = %activity.id, symbol = %activity.symbol, "Alpaca: skipping activity with unparseable fields");
continue; }
};
let key = fill_dedup_key(
&activity.order_id,
trade.order_filled_quantity.unwrap_or(cumulative),
);
if is_duplicate_fill(dedup, activity_execution_id_part(&activity.id), key) {
duplicates += 1;
continue;
}
let event =
UnindexedAccountEvent::new(ExchangeId::AlpacaBroker, AccountEventKind::Trade(trade));
let mut held = known.lock();
held.observe(&event);
let sent = tx.send(event);
drop(held);
if sent.is_err() {
debug!("Alpaca fill recovery: consumer dropped during recovery");
return Ok(());
}
recovered += 1;
}
info!(recovered, duplicates, "Alpaca fill recovery complete");
unread.map_or(Ok(()), Err)
}
async fn fetch_order_lookup(
http: reqwest::Client,
rate_limiter: Arc<RateLimitTracker>,
config: Arc<AlpacaConfig>,
key: UnindexedOrderKey,
) -> Result<OrderLookup, UnindexedClientError> {
let url = format!("{}/v2/orders:by_client_order_id", config.rest_base_url());
let cid = key.cid.0.as_str();
let found: Option<AlpacaOrderResponse> = rest_lookup_with_retry(&rate_limiter, || {
http.get(&url).query(&[("client_order_id", cid)])
})
.await?;
let Some(order) = found else {
debug!(instrument = %key.instrument, cid = %key.cid, "Alpaca does not know this order");
return Ok(OrderLookup::Unknown);
};
if !order
.symbol
.eq_ignore_ascii_case(key.instrument.name().as_str())
{
debug!(
instrument = %key.instrument,
cid = %key.cid,
symbol = %order.symbol,
"Alpaca knows this client order id on another symbol"
);
return Ok(OrderLookup::Unknown);
}
Ok(
convert_ended_order(&order, &key).map_or(OrderLookup::NotEnded, |order| {
OrderLookup::Ended(Box::new(order))
}),
)
}
async fn listed_open_cids(
http: reqwest::Client,
rate_limiter: Arc<RateLimitTracker>,
config: Arc<AlpacaConfig>,
instruments: Vec<InstrumentNameExchange>,
) -> Result<FnvHashSet<ClientOrderId>, UnindexedClientError> {
let orders =
fetch_raw_open_orders(&http, &rate_limiter, config.rest_base_url(), &instruments).await?;
Ok(orders
.iter()
.map(|order| alpaca_cid(order.client_order_id.as_deref(), &order.id))
.collect())
}
async fn recover_alpaca_ended_orders(
http: &reqwest::Client,
rate_limiter: &Arc<RateLimitTracker>,
config: &Arc<AlpacaConfig>,
known: &SharedKnownLiveOrders,
unchecked: &mut UncheckedOrders,
tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
) {
recover_ended_orders(
ExchangeId::AlpacaBroker,
known,
unchecked,
&NoPendingFills,
tx,
OpenListing::Batched,
|instruments| {
listed_open_cids(
http.clone(),
rate_limiter.clone(),
config.clone(),
instruments,
)
},
|key| fetch_order_lookup(http.clone(), rate_limiter.clone(), config.clone(), key),
)
.await;
}
async fn run_order_checks(
http: &reqwest::Client,
rate_limiter: &Arc<RateLimitTracker>,
config: &Arc<AlpacaConfig>,
known: &SharedKnownLiveOrders,
unchecked: &mut UncheckedOrders,
tx: &mpsc::UnboundedSender<UnindexedAccountEvent>,
) {
loop {
if tx.is_closed() {
return std::future::pending().await;
}
match unchecked.next_due(&NoPendingFills) {
Some(due) => tokio::time::sleep_until(due).await,
None => return std::future::pending().await,
}
recover_alpaca_ended_orders(http, rate_limiter, config, known, unchecked, tx).await;
}
}
fn convert_ended_order(
o: &AlpacaOrderResponse,
key: &UnindexedOrderKey,
) -> Option<UnindexedInactiveOrder> {
let instrument = &key.instrument;
let Some(status) = o.status.as_deref() else {
warn!(%instrument, cid = %key.cid, order_id = %o.id, "Alpaca order missing status");
return None;
};
if order_status_is_live(status) {
return None;
}
let Some(order) = convert_representable_open_order(o) else {
warn!(%instrument, cid = %key.cid, order_id = %o.id, status, "Alpaca order cannot be represented, treating it as not ended");
return None;
};
let filled_qty = Decimal::from_str(&o.filled_qty).ok();
let Some(state) = ended_order_state(
status,
OrderId(SmolStr::new(&o.id)),
order.state.time_exchange,
order.quantity,
filled_qty,
o.filled_avg_price.as_deref(),
) else {
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");
return None;
};
let mut order = order.map_state(|_| state);
order.key = key.clone();
Some(order)
}
fn order_status_is_live(status: &str) -> bool {
matches!(
status,
"new"
| "partially_filled"
| "done_for_day"
| "pending_cancel"
| "pending_replace"
| "accepted"
| "pending_new"
| "accepted_for_bidding"
| "stopped"
| "suspended"
| "calculated"
| "held"
)
}
fn ended_order_state<AssetKey, InstrumentKey>(
status: &str,
order_id: OrderId,
time_exchange: DateTime<Utc>,
quantity: Decimal,
filled_qty: Option<Decimal>,
filled_avg_price: Option<&str>,
) -> Option<InactiveOrderState<AssetKey, InstrumentKey>> {
Some(match status {
"filled" => InactiveOrderState::FullyFilled(Filled::new(
order_id,
time_exchange,
filled_qty.unwrap_or(quantity),
alpaca_avg_price(filled_avg_price),
)),
"canceled" | "replaced" => {
InactiveOrderState::Cancelled(Cancelled::new(order_id, time_exchange, filled_qty))
}
"expired" => InactiveOrderState::Expired(Expired::new(order_id, time_exchange, filled_qty)),
"rejected" => {
InactiveOrderState::OpenFailed(OrderError::Rejected(ApiError::OrderRejected(format!(
"Alpaca rejected order {order_id} after accepting it"
))))
}
_ => return None,
})
}
fn placed_order_state(
resp: &AlpacaOrderResponse,
instrument: &InstrumentNameExchange,
quantity: Decimal,
) -> UnindexedOrderState {
let order_id = OrderId(SmolStr::new(&resp.id));
let time_exchange = order_state_time(resp);
let filled_qty = alpaca_filled_qty(&resp.id, Some(&resp.filled_qty));
let filled_avg_price = resp.filled_avg_price.as_deref();
match resp.status.as_deref() {
Some(status) => {
if let Some(ended) = ended_order_state(
status,
order_id.clone(),
time_exchange,
quantity,
filled_qty,
filled_avg_price,
) {
return OrderState::Inactive(ended);
}
if !order_status_is_live(status) {
warn!(%instrument, %order_id, status, "Alpaca placed an order with a status this version does not know, reading it from what filled");
}
}
None => {
warn!(%instrument, %order_id, "Alpaca placed an order without a status, reading it from what filled");
}
}
match filled_qty {
Some(filled_qty) if filled_qty >= quantity => OrderState::fully_filled(Filled::new(
order_id,
time_exchange,
filled_qty,
alpaca_avg_price(filled_avg_price),
)),
_ => OrderState::active(Open::new(
VenueOrderId::Assigned(order_id),
time_exchange,
filled_qty.unwrap_or(Decimal::ZERO),
)),
}
}
fn alpaca_avg_price(filled_avg_price: Option<&str>) -> Option<Decimal> {
filled_avg_price.and_then(|price| Decimal::from_str(price).ok())
}
fn is_usd(asset: &AssetNameExchange) -> bool {
asset.name().as_str().eq_ignore_ascii_case(USD)
}
fn convert_account_to_balances(
account: &AlpacaAccount,
assets: &[AssetNameExchange],
time: DateTime<Utc>,
) -> Vec<AssetBalance<AssetNameExchange>> {
if !assets.is_empty() && !assets.iter().any(is_usd) {
return Vec::new();
}
vec![AssetBalance::new(
AssetNameExchange::new(USD),
Balance::new(
account.cash,
account.cash.min(account.non_marginable_buying_power),
),
time,
)]
}
fn convert_positions_to_balances(
positions: &[AlpacaPosition],
assets: &[AssetNameExchange],
time: DateTime<Utc>,
) -> Vec<AssetBalance<AssetNameExchange>> {
positions
.iter()
.filter(|p| is_crypto_position(p))
.filter_map(|p| {
let base = crypto_position_asset(&p.symbol);
if !assets.is_empty()
&& !assets
.iter()
.any(|a| a.name().as_str().eq_ignore_ascii_case(base))
{
return None;
}
let asset_name = AssetNameExchange::new(base);
Some(AssetBalance::new(
asset_name,
Balance::new(p.qty, p.qty_available),
time,
))
})
.collect()
}
fn crypto_position_asset(symbol: &str) -> &str {
let base = match symbol.split_once('/') {
Some((base, _)) => Some(base),
None => {
let base_len = symbol.len().saturating_sub(USD.len());
symbol
.get(base_len..)
.filter(|suffix| suffix.eq_ignore_ascii_case(USD))
.and_then(|_| symbol.get(..base_len))
}
};
match base {
Some(base) if !base.is_empty() => base,
_ => {
warn!(
%symbol,
"Alpaca crypto position symbol is neither ASSETUSD nor BASE/QUOTE; using it whole"
);
symbol
}
}
}
fn is_crypto_position(position: &AlpacaPosition) -> bool {
position.asset_class.eq_ignore_ascii_case("crypto")
}
fn convert_positions(
positions: &[AlpacaPosition],
now: DateTime<Utc>,
) -> Result<Vec<(&str, Position)>, UnindexedClientError> {
positions
.iter()
.filter(|p| !is_crypto_position(p) && !p.qty.is_zero())
.map(|p| {
let size = p.qty.abs();
let quantity = match p.side {
AlpacaPositionSide::Long => size,
AlpacaPositionSide::Short => -size,
AlpacaPositionSide::Unknown => {
return Err(UnindexedClientError::Internal(format!(
"Alpaca position {} has a missing or unrecognised side",
p.symbol
)));
}
};
let position = Position::new(
quantity,
p.avg_entry_price,
p.unrealized_pl,
None,
None,
None,
None,
now,
);
Ok((p.symbol.as_str(), position))
})
.collect()
}
fn build_instrument_snapshots(
orders: Vec<AlpacaOrderResponse>,
positions: Vec<(&str, Position)>,
instruments: &[InstrumentNameExchange],
) -> Vec<InstrumentAccountSnapshot<ExchangeId, AssetNameExchange, InstrumentNameExchange>> {
struct Listing {
orders: Vec<UnindexedOrderSnapshot>,
complete: bool,
position: Option<Position>,
}
impl Listing {
fn empty() -> Self {
Self {
orders: Vec::new(),
complete: true,
position: None,
}
}
}
let mut by_symbol: IndexMap<SmolStr, Listing> = IndexMap::new();
for order in orders {
let listing = by_symbol
.entry(order.symbol.to_ascii_uppercase().into())
.or_insert_with(Listing::empty);
match convert_open_order(&order) {
Some(converted) => listing.orders.push(converted.into()),
None => listing.complete = false,
}
}
for (symbol, position) in positions {
by_symbol
.entry(symbol.to_ascii_uppercase().into())
.or_insert_with(Listing::empty)
.position = Some(position);
}
let snapshot = |instrument: InstrumentNameExchange, listing: Listing| {
let position = if is_options_or_equity_symbol(instrument.name().as_str()) {
listing
.position
.map_or(PositionReport::Flat, PositionReport::from_position)
} else {
PositionReport::Unreported
};
InstrumentAccountSnapshot::new(instrument, listing.orders, listing.complete, position, None)
};
if instruments.is_empty() {
by_symbol
.into_iter()
.map(|(sym, listing)| snapshot(InstrumentNameExchange::new(sym), listing))
.collect()
} else {
let mut claimed = std::collections::HashSet::new();
instruments
.iter()
.map(|inst| {
let symbol = inst.name().as_str().to_ascii_uppercase();
if !claimed.insert(symbol.clone()) {
return InstrumentAccountSnapshot::new(
inst.clone(),
Vec::new(),
false,
PositionReport::Unreported,
None,
);
}
let listing = by_symbol
.swap_remove(symbol.as_str())
.unwrap_or_else(Listing::empty);
snapshot(inst.clone(), listing)
})
.collect()
}
}
fn convert_open_order(
o: &AlpacaOrderResponse,
) -> Option<Order<ExchangeId, InstrumentNameExchange, Open>> {
let converted = convert_representable_open_order(o);
if converted.is_none() {
warn!(
order_id = %o.id,
symbol = %o.symbol,
qty = ?o.qty,
side = %o.side,
order_type = %o.order_type,
"Alpaca open order cannot be represented - leaving it out"
);
}
converted
}
fn alpaca_cid(client_order_id: Option<&str>, order_id: &str) -> ClientOrderId {
ClientOrderId::new(client_order_id.unwrap_or(order_id))
}
fn convert_representable_open_order(
o: &AlpacaOrderResponse,
) -> Option<Order<ExchangeId, InstrumentNameExchange, Open>> {
let order_id = OrderId(SmolStr::new(&o.id));
let cid = alpaca_cid(o.client_order_id.as_deref(), &o.id);
let instrument = InstrumentNameExchange::new(&o.symbol);
let side = parse_side(&o.side)?;
let quantity = Decimal::from_str(o.qty.as_deref().unwrap_or("0")).ok()?;
if quantity.is_zero() {
return None;
}
let price = o
.limit_price
.as_deref()
.and_then(|s| Decimal::from_str(s).ok());
let filled_qty = alpaca_filled_qty(&o.id, Some(&o.filled_qty)).unwrap_or(Decimal::ZERO);
let kind = parse_order_kind(
&o.order_type,
o.stop_price.as_deref(),
o.trail_percent.as_deref(),
o.trail_price.as_deref(),
)?;
let time_in_force = parse_time_in_force(&o.time_in_force);
let time_exchange = order_state_time(o);
Some(Order {
key: OrderKey::new(
ExchangeId::AlpacaBroker,
instrument,
StrategyId::unknown(),
cid,
),
side,
price,
quantity,
kind,
time_in_force,
state: Open::new(VenueOrderId::Assigned(order_id), time_exchange, filled_qty),
})
}
fn convert_activity_to_trade(
a: &AlpacaActivity,
) -> Option<Trade<AssetNameExchange, InstrumentNameExchange>> {
let trade_id = TradeId::new(activity_execution_id(&a.id));
let order_id = OrderId(SmolStr::new(&a.order_id));
let instrument = InstrumentNameExchange::new(&a.symbol);
let side = parse_side(&a.side)?;
let price = Decimal::from_str(&a.price).ok()?;
let quantity = Decimal::from_str(&a.qty).ok()?;
let order_filled_quantity = a.cum_qty.as_deref().and_then(|s| Decimal::from_str(s).ok());
let (time_exchange, time_source) = match parse_timestamp(&a.transaction_time) {
Some(time) => (time, TimeSource::Venue),
None => {
warn!(id = %a.id, "Alpaca activity: unparseable transaction_time, using now");
(Utc::now(), TimeSource::Host)
}
};
Some(
Trade::new(
trade_id,
order_id,
instrument,
StrategyId::unknown(),
time_exchange,
side,
price,
quantity,
order_filled_quantity,
AssetFees::new(
AssetNameExchange::from(USD),
Decimal::ZERO,
Some(Decimal::ZERO),
),
)
.with_time_source(time_source),
)
}
fn ws_order_snapshot(
order: &AlpacaOrderWs<'_>,
instrument: InstrumentNameExchange,
cid: ClientOrderId,
order_id: OrderId,
time_exchange: DateTime<Utc>,
) -> Option<UnindexedAccountEvent> {
let side = parse_side(&order.side)?;
let quantity = Decimal::from_str(order.qty.unwrap_or("0")).unwrap_or(Decimal::ZERO);
if quantity.is_zero() {
trace!(order_id = %order.id, "Alpaca WS: skipping notional order snapshot (qty=None)");
return None;
}
let price = order.limit_price.and_then(|s| Decimal::from_str(s).ok());
let filled_qty = order
.filled_qty
.and_then(|_| alpaca_filled_qty(&order.id, order.filled_qty))
.unwrap_or(Decimal::ZERO);
let kind = parse_order_kind(
&order.order_type,
order.stop_price,
order.trail_percent,
order.trail_price,
)?;
let time_in_force = parse_time_in_force(&order.time_in_force);
let order_snapshot = crate::order::Order {
key: OrderKey::new(
ExchangeId::AlpacaBroker,
instrument,
StrategyId::unknown(),
cid,
),
side,
price,
quantity,
kind,
time_in_force,
state: OrderState::active(Open::new(
VenueOrderId::Assigned(order_id),
time_exchange,
filled_qty,
)),
};
Some(UnindexedAccountEvent::new(
ExchangeId::AlpacaBroker,
AccountEventKind::OrderSnapshot(rustrade_integration::collection::snapshot::Snapshot(
order_snapshot,
)),
))
}
fn ws_fill_order_is_live(status: &str) -> bool {
matches!(status, "partially_filled" | "filled")
}
fn trade_amendment_event(
update: &AlpacaTradeUpdate<'_>,
instrument: InstrumentNameExchange,
order_id: OrderId,
(time_exchange, time_source): (DateTime<Utc>, TimeSource),
) -> UnindexedAccountEvent {
let order = &update.order;
let original = update
.previous_execution_id
.filter(|id| !id.is_empty())
.map(TradeId::new);
let quantity = update.qty.and_then(|s| Decimal::from_str(s).ok());
let kind = if update.event == "trade_bust" {
TradeAmendmentKind::Busted {
quantity: quantity.map(|quantity| quantity.abs()),
}
} else {
let id = ws_execution_id(update).map(TradeId::new);
let price = update.price.and_then(|s| Decimal::from_str(s).ok());
match (id, price, quantity, parse_side(&order.side)) {
(Some(id), Some(price), Some(quantity), Some(side)) if quantity > Decimal::ZERO => {
TradeAmendmentKind::Corrected {
replacement: with_reported_cid(
Trade::new(
id,
order_id.clone(),
instrument.clone(),
StrategyId::unknown(),
time_exchange,
side,
price,
quantity,
None,
AssetFees::new(
AssetNameExchange::from(USD),
Decimal::ZERO,
Some(Decimal::ZERO),
),
)
.with_time_source(time_source),
order.client_order_id.as_ref(),
),
}
}
(id, price, quantity, _) => TradeAmendmentKind::CorrectedUnresolved {
id,
price,
quantity,
},
}
};
UnindexedAccountEvent::new(
ExchangeId::AlpacaBroker,
AccountEventKind::TradeAmended(crate::trade::TradeAmendment::new(
instrument,
order_id,
time_exchange,
original,
kind,
)),
)
}
fn trade_update_time(update: &AlpacaTradeUpdate<'_>) -> (DateTime<Utc>, TimeSource) {
if let Some(time) = update.timestamp.and_then(parse_timestamp) {
return (time, TimeSource::Venue);
}
let order = &update.order;
if let Some(time) = order.updated_at.and_then(parse_timestamp) {
warn!(order_id = %order.id, event = %update.event, timestamp = ?update.timestamp, "Alpaca WS: event missing/unparseable timestamp, using its order's updated_at");
return (time, TimeSource::Venue);
}
warn!(order_id = %order.id, event = %update.event, timestamp = ?update.timestamp, updated_at = ?order.updated_at, "Alpaca WS: event missing/unparseable timestamp and order updated_at, using now");
(Utc::now(), TimeSource::Host)
}
#[cfg(test)]
fn convert_trade_update(update: AlpacaTradeUpdate<'_>) -> [Option<UnindexedAccountEvent>; 2] {
let time = trade_update_time(&update);
convert_trade_update_at(update, time)
}
fn convert_trade_update_at(
update: AlpacaTradeUpdate<'_>,
(time_exchange, time_source): (DateTime<Utc>, TimeSource),
) -> [Option<UnindexedAccountEvent>; 2] {
let event_str = update.event.as_str();
if !matches!(
event_str,
"fill"
| "partial_fill"
| "new"
| "accepted"
| "pending_new"
| "canceled"
| "expired"
| "replaced"
| "done_for_day"
| "rejected"
| "trade_bust"
| "trade_correct"
) {
trace!(event = %event_str, "Alpaca WS: ignoring trade_updates event type");
return [None, None];
}
let order = &update.order;
let instrument = InstrumentNameExchange::new(&*order.symbol);
let order_id = OrderId(order.id.clone());
let cid = alpaca_cid(order.client_order_id.as_deref(), &order.id);
match event_str {
"fill" | "partial_fill" => {
let (Some(price), Some(quantity), Some(side)) = (
update.price.and_then(|s| Decimal::from_str(s).ok()),
update.qty.and_then(|s| Decimal::from_str(s).ok()),
parse_side(&order.side),
) else {
return [None, None];
};
let cum_qty = Decimal::from_str(order.filled_qty.unwrap_or("0"))
.inspect_err(|e| {
warn!(
order_id = %order.id,
filled_qty = ?order.filled_qty,
%e,
"Alpaca WS: failed to parse filled_qty — the fill reports a cumulative \
of 0, and without an execution id a second such fill on the same order \
would be deduplicated away"
);
})
.unwrap_or(Decimal::ZERO);
let trade_id = match ws_execution_id(&update) {
Some(execution_id) => TradeId::new(execution_id),
None => {
warn!(
order_id = %order.id,
"Alpaca WS: fill without a usable execution_id — its TradeId is \
\"{{order_id}}:{{filled_qty}}\", which fetch_trades will not match"
);
TradeId(fill_dedup_key(&order.id, cum_qty))
}
};
let trade = Trade::new(
trade_id,
order_id.clone(),
instrument.clone(),
StrategyId::unknown(),
time_exchange,
side,
price,
quantity,
Some(cum_qty),
AssetFees::new(
AssetNameExchange::from(USD),
Decimal::ZERO,
Some(Decimal::ZERO),
),
)
.with_time_source(time_source);
let trade = with_reported_cid(trade, order.client_order_id.as_ref());
let trade_event = UnindexedAccountEvent::new(
ExchangeId::AlpacaBroker,
AccountEventKind::Trade(trade),
);
let snapshot_event = if ws_fill_order_is_live(&order.status) {
ws_order_snapshot(order, instrument, cid, order_id, time_exchange)
} else {
trace!(
order_id = %order.id,
status = %order.status,
"Alpaca WS: fill for an order the exchange no longer reports as working — \
emitting the execution without an order snapshot"
);
None
};
[Some(trade_event), snapshot_event]
}
"new" | "accepted" | "pending_new" | "done_for_day" => {
[
ws_order_snapshot(order, instrument, cid, order_id, time_exchange),
None,
]
}
"canceled" | "expired" | "replaced" => {
let filled_qty = alpaca_filled_qty(&order.id, order.filled_qty);
let cancelled = Cancelled::new(order_id, time_exchange, filled_qty);
let response = crate::order::request::OrderResponseCancel {
key: OrderKey::new(
ExchangeId::AlpacaBroker,
instrument,
StrategyId::unknown(),
cid,
),
state: Ok(cancelled),
};
[
Some(UnindexedAccountEvent::new(
ExchangeId::AlpacaBroker,
AccountEventKind::OrderCancelled(response),
)),
None,
]
}
"trade_bust" | "trade_correct" => [
Some(trade_amendment_event(
&update,
instrument,
order_id,
(time_exchange, time_source),
)),
None,
],
"rejected" => {
let response = crate::order::request::OrderResponseCancel {
key: OrderKey::new(
ExchangeId::AlpacaBroker,
instrument,
StrategyId::unknown(),
cid,
),
state: Err(UnindexedOrderError::Rejected(ApiError::OrderRejected(
format!("order rejected: status={}", order.status),
))),
};
[
Some(UnindexedAccountEvent::new(
ExchangeId::AlpacaBroker,
AccountEventKind::OrderCancelled(response),
)),
None,
]
}
_ => unreachable!("convert_trade_update: unrecognised event passed early-return guard"),
}
}
fn parse_side(s: &str) -> Option<Side> {
match s {
"buy" | "Buy" | "BUY" => Some(Side::Buy),
"sell" | "Sell" | "SELL" => Some(Side::Sell),
other => {
trace!(%other, "Alpaca: unknown order side");
None
}
}
}
fn parse_order_kind(
order_type: &str,
stop_price: Option<&str>,
trail_percent: Option<&str>,
trail_price: Option<&str>,
) -> Option<OrderKind> {
match order_type {
"market" | "Market" => Some(OrderKind::Market),
"limit" | "Limit" => Some(OrderKind::Limit),
"stop" | "Stop" => {
let trigger_price = stop_price.and_then(|s| Decimal::from_str(s).ok())?;
Some(OrderKind::Stop { trigger_price })
}
"stop_limit" | "Stop_limit" => {
let trigger_price = stop_price.and_then(|s| Decimal::from_str(s).ok())?;
Some(OrderKind::StopLimit { trigger_price })
}
"trailing_stop" | "Trailing_stop" => {
if let Some(pct) = trail_percent.and_then(|s| Decimal::from_str(s).ok()) {
Some(OrderKind::TrailingStop {
offset: pct,
offset_type: TrailingOffsetType::Percentage,
activation_price: None,
})
} else if let Some(price) = trail_price.and_then(|s| Decimal::from_str(s).ok()) {
Some(OrderKind::TrailingStop {
offset: price,
offset_type: TrailingOffsetType::Absolute,
activation_price: None,
})
} else {
trace!("Alpaca: trailing_stop missing trail_percent and trail_price");
None
}
}
other => {
trace!(%other, "Alpaca: unsupported order type, skipping");
None
}
}
}
fn parse_time_in_force(s: &str) -> TimeInForce {
match s {
"gtc" | "GTC" => TimeInForce::GoodUntilCancelled { post_only: false },
"day" | "DAY" => TimeInForce::GoodUntilEndOfDay,
"fok" | "FOK" => TimeInForce::FillOrKill,
"ioc" | "IOC" => TimeInForce::ImmediateOrCancel,
other => {
warn!(%other, "Alpaca: unknown time_in_force, defaulting to GoodUntilEndOfDay");
TimeInForce::GoodUntilEndOfDay
}
}
}
fn order_state_time(order: &AlpacaOrderResponse) -> DateTime<Utc> {
order
.updated_at
.as_deref()
.and_then(parse_timestamp)
.or_else(|| parse_timestamp(&order.created_at))
.unwrap_or_else(Utc::now)
}
fn alpaca_filled_qty(order_id: &str, filled_qty: Option<&str>) -> Option<Decimal> {
let Some(raw) = filled_qty else {
warn!(%order_id, "Alpaca did not report how much the order filled");
return None;
};
match Decimal::from_str(raw) {
Ok(filled_qty) => Some(filled_qty),
Err(_) => {
warn!(%order_id, filled_qty = raw, "Alpaca reported an unparseable filled_qty, treating it as unknown");
None
}
}
}
fn convert_cash_flow(
activity: &AlpacaNonTradeActivity,
) -> Option<CashFlow<AssetNameExchange, InstrumentNameExchange>> {
let id = activity.id.as_str();
if activity.status.as_deref() == Some("canceled") {
info!(id, activity_type = %activity.activity_type, "Alpaca activity canceled — left out");
return None;
}
let Some(amount) = activity
.net_amount
.as_deref()
.and_then(|amount| Decimal::from_str(amount).ok())
else {
warn!(id, net_amount = ?activity.net_amount, "Alpaca activity missing/unparseable net_amount — left out");
return None;
};
let Some(time_exchange) = activity.time() else {
warn!(
id,
created_at = ?activity.created_at,
date = ?activity.date,
"Alpaca activity missing/unparseable time — left out"
);
return None;
};
let sub_type = activity
.activity_sub_type
.as_deref()
.filter(|sub_type| !sub_type.is_empty());
let kind = match (activity.activity_type.as_str(), sub_type) {
("INT", Some("MGN")) => CashFlowKind::MarginInterest {
rate: None,
principal: None,
timing: None,
},
("INT", None) if amount.is_sign_negative() => CashFlowKind::MarginInterest {
rate: None,
principal: None,
timing: None,
},
("PTR", _) => CashFlowKind::Rebate,
(activity_type, Some(sub_type)) => {
CashFlowKind::Other(format!("{activity_type}.{sub_type}"))
}
(activity_type, None) => CashFlowKind::Other(activity_type.to_owned()),
};
Some(CashFlow::new(
kind,
AssetNameExchange::new(
activity
.currency
.as_deref()
.filter(|currency| !currency.is_empty())
.unwrap_or("USD"),
),
amount,
activity
.symbol
.as_deref()
.filter(|symbol| !symbol.is_empty())
.map(InstrumentNameExchange::new),
time_exchange,
Some(CashFlowId::new(id)),
))
}
struct OptionLeg<'a> {
id: &'a str,
symbol: &'a str,
quantity: Decimal,
price: Option<Decimal>,
date: Option<NaiveDate>,
time: DateTime<Utc>,
created: bool,
origin: Option<TradeOrigin>,
contract: Option<(&'a str, bool, Decimal)>,
}
impl OptionLeg<'_> {
fn is_within(&self, start: DateTime<Utc>, end: DateTime<Utc>) -> bool {
if self.created {
(start..=end).contains(&self.time)
} else {
(start.date_naive()..=end.date_naive()).contains(&self.time.date_naive())
}
}
}
fn option_leg(activity: &AlpacaNonTradeActivity) -> Option<OptionLeg<'_>> {
let id = activity.id.as_str();
let activity_type = activity.activity_type.as_str();
if activity.status.as_deref() == Some("canceled") {
info!(
id,
activity_type, "Alpaca option activity canceled — left out"
);
return None;
}
let origin = match activity_type {
"OPEXP" => Some(TradeOrigin::Expiry),
"OPEXC" => Some(TradeOrigin::Exercise),
"OPASN" => Some(TradeOrigin::Assignment),
"OPTRD" => None,
_ => {
warn!(
id,
activity_type, "Alpaca activity is not an option event — left out"
);
return None;
}
};
let date = activity
.date
.as_deref()
.and_then(|date| NaiveDate::parse_from_str(date, "%Y-%m-%d").ok());
let created_at = activity.created_at.as_deref().and_then(parse_timestamp);
let time = created_at.or_else(|| date.map(|date| date.and_time(NaiveTime::MIN).and_utc()));
let symbol = activity
.symbol
.as_deref()
.filter(|symbol| !symbol.is_empty());
let quantity = activity
.qty
.as_deref()
.and_then(|qty| Decimal::from_str(qty).ok())
.filter(|qty| !qty.is_zero());
let price = activity
.price
.as_deref()
.and_then(|price| Decimal::from_str(price).ok());
let (Some(symbol), Some(quantity), Some(time)) = (symbol, quantity, time) else {
warn!(
id,
activity_type,
symbol = ?activity.symbol,
qty = ?activity.qty,
created_at = ?activity.created_at,
date = ?activity.date,
"Alpaca option activity missing/unparseable symbol, quantity or time — left out"
);
return None;
};
if origin.is_none() && price.is_none() {
warn!(id, price = ?activity.price, "Alpaca OPTRD missing/unparseable price — left out");
return None;
}
let contract = matches!(
origin,
Some(TradeOrigin::Exercise | TradeOrigin::Assignment)
)
.then(|| occ_contract(symbol))
.flatten();
Some(OptionLeg {
id,
symbol,
quantity,
price,
date,
time,
created: created_at.is_some(),
origin,
contract,
})
}
fn occ_contract(symbol: &str) -> Option<(&str, bool, Decimal)> {
let split = symbol.len().checked_sub(15)?;
let (root, tail) = (
symbol.get(..split)?.trim_end(),
symbol.get(split..)?.as_bytes(),
);
let is_call = match tail[6] {
b'C' => true,
b'P' => false,
_ => return None,
};
if root.is_empty() || !tail[..6].iter().chain(&tail[7..]).all(u8::is_ascii_digit) {
return None;
}
let strike = std::str::from_utf8(&tail[7..]).ok()?.parse::<i64>().ok()?;
Some((root, is_call, Decimal::new(strike, 3)))
}
fn is_share_leg_of(shares: &OptionLeg<'_>, contract: &OptionLeg<'_>) -> bool {
let Some((root, is_call, strike)) = contract.contract else {
return false;
};
let buys = match contract.origin {
Some(TradeOrigin::Exercise) => is_call,
Some(TradeOrigin::Assignment) => !is_call,
_ => return false,
};
shares.date.is_some()
&& shares.date == contract.date
&& shares.symbol == root
&& shares.price == Some(strike)
&& shares.quantity.is_sign_positive() == buys
}
fn convert_option_events(
activities: &[AlpacaNonTradeActivity],
start: DateTime<Utc>,
end: DateTime<Utc>,
) -> Vec<Trade<AssetNameExchange, InstrumentNameExchange>> {
let legs: Vec<_> = activities.iter().filter_map(option_leg).collect();
let mut paired = vec![false; legs.len()];
legs.iter()
.filter_map(|leg| {
let origin = leg.origin.unwrap_or_else(|| {
let pairs = |index: &usize| !paired[*index] && is_share_leg_of(leg, &legs[*index]);
let delivers_its_shares = |index: &usize| {
legs[*index].quantity.abs() * Decimal::ONE_HUNDRED == leg.quantity.abs()
};
let partner = (0..legs.len())
.filter(pairs)
.find(delivers_its_shares)
.or_else(|| (0..legs.len()).find(pairs));
match partner {
Some(index) => {
paired[index] = true;
legs[index].origin.unwrap_or(TradeOrigin::External)
}
None => {
warn!(
id = leg.id,
symbol = leg.symbol,
date = ?leg.date,
"Alpaca OPTRD pairs with no exercise or assignment read — reported as \
External"
);
TradeOrigin::External
}
}
});
if !leg.is_within(start, end) {
return None;
}
let side = if leg.quantity.is_sign_negative() {
Side::Sell
} else {
Side::Buy
};
let trade = Trade::new(
TradeId::new(leg.id),
OrderId::new(leg.id),
InstrumentNameExchange::new(leg.symbol),
StrategyId::unknown(),
leg.time,
side,
leg.price.unwrap_or(Decimal::ZERO),
leg.quantity.abs(),
None,
AssetFees::new(
AssetNameExchange::from(USD),
Decimal::ZERO,
Some(Decimal::ZERO),
),
)
.with_origin(origin);
Some(trade)
})
.collect()
}
fn with_reported_cid<AssetKey, InstrumentKey>(
trade: Trade<AssetKey, InstrumentKey>,
client_order_id: Option<&SmolStr>,
) -> Trade<AssetKey, InstrumentKey> {
match client_order_id {
Some(cid) if !cid.is_empty() => trade.with_client_order_id(ClientOrderId(cid.clone())),
_ => trade,
}
}
fn parse_timestamp(s: &str) -> Option<DateTime<Utc>> {
DateTime::parse_from_rfc3339(s)
.ok()
.map(|dt| dt.with_timezone(&Utc))
}
fn map_side(side: Side) -> &'static str {
match side {
Side::Buy => "buy",
Side::Sell => "sell",
}
}
fn map_order_kind(kind: OrderKind) -> Option<&'static str> {
match kind {
OrderKind::Market => Some("market"),
OrderKind::Limit => Some("limit"),
OrderKind::Stop { .. } => Some("stop"),
OrderKind::StopLimit { .. } => Some("stop_limit"),
OrderKind::TrailingStop {
activation_price: None,
..
} => Some("trailing_stop"),
OrderKind::TakeProfit { .. }
| OrderKind::TakeProfitLimit { .. }
| OrderKind::TrailingStop {
activation_price: Some(_),
..
}
| OrderKind::TrailingStopLimit { .. }
| OrderKind::TrailingStopFixedLimit { .. }
| OrderKind::ConditionalUnresolved { .. } => None,
}
}
fn map_time_in_force(tif: TimeInForce) -> Result<&'static str, &'static str> {
match tif {
TimeInForce::GoodUntilCancelled { post_only } => {
if post_only {
return Err("Alpaca does not support post_only orders");
}
Ok("gtc")
}
TimeInForce::GoodUntilEndOfDay => Ok("day"),
TimeInForce::FillOrKill => Ok("fok"),
TimeInForce::ImmediateOrCancel => Ok("ioc"),
TimeInForce::AtOpen => Ok("opg"),
TimeInForce::AtClose => Ok("cls"),
TimeInForce::GoodTillDate { .. } => {
Err("Alpaca GoodTillDate is not yet wired through this client")
}
}
}
fn is_options_or_equity_symbol(symbol: &str) -> bool {
!symbol.contains('/')
}
fn map_position_intent(side: Side, reduce_only: ReduceOnly) -> AlpacaPositionIntent {
match (reduce_only.is_reduce_only(), side) {
(false, Side::Buy) => AlpacaPositionIntent::BuyToOpen,
(false, Side::Sell) => AlpacaPositionIntent::SellToOpen,
(true, Side::Buy) => AlpacaPositionIntent::BuyToClose,
(true, Side::Sell) => AlpacaPositionIntent::SellToClose,
}
}
fn parse_api_error(
status: reqwest::StatusCode,
code: Option<i64>,
message: &str,
) -> crate::error::UnindexedApiError {
if status == reqwest::StatusCode::TOO_MANY_REQUESTS {
return ApiError::rate_limited(None);
}
let lower = message.to_ascii_lowercase();
match status.as_u16() {
422 if lower.contains("already") => ApiError::OrderAlreadyCancelled,
422 => parse_business_refusal(code, message, &lower),
401 => ApiError::Unauthenticated(format!("unauthorized: {message}")),
403 => match code {
None => ApiError::Unauthenticated(format!("forbidden: {message}")),
Some(_) => parse_business_refusal(code, message, &lower),
},
404 => ApiError::OrderRejected(format!("order not found: {message}")),
_ => ApiError::OrderRejected(with_alpaca_code(code, message)),
}
}
fn parse_business_refusal(code: Option<i64>, message: &str, lower: &str) -> UnindexedApiError {
let borrow = if lower.contains("not allowed to short") {
Some(BorrowRejectReason::BorrowDisabled)
} else if lower.contains("cannot be sold short") && !lower.contains("fractional") {
Some(BorrowRejectReason::NotShortable)
} else {
None
};
if let Some(reason) = borrow {
ApiError::BorrowRejected(BorrowReject::new(
reason,
code.map(|code| code.to_string()),
message.to_owned(),
))
} else if lower.contains("insufficient") {
ApiError::BalanceInsufficient(None, with_alpaca_code(code, message))
} else {
ApiError::OrderRejected(with_alpaca_code(code, message))
}
}
fn with_alpaca_code(code: Option<i64>, message: &str) -> String {
match code {
Some(code) => format!("{code} {message}"),
None => message.to_owned(),
}
}
fn parse_order_error(
status: reqwest::StatusCode,
code: Option<i64>,
message: &str,
) -> UnindexedOrderError {
UnindexedOrderError::Rejected(parse_api_error(status, code, message))
}
fn connectivity_err(msg: impl Into<String>) -> UnindexedClientError {
UnindexedClientError::Connectivity(ConnectivityError::Socket(msg.into()))
}
fn order_post_error(error: UnindexedClientError) -> UnindexedOrderError {
match error {
UnindexedClientError::Connectivity(ce) => OrderError::Connectivity(ce),
UnindexedClientError::Api(ae) => OrderError::Rejected(ae),
UnindexedClientError::Internal(msg) => OrderError::Connectivity(ConnectivityError::Socket(
format!("Alpaca order status unknown, its 2xx response could not be read: {msg}"),
)),
other @ (UnindexedClientError::TaskFailed(_)
| UnindexedClientError::Truncated { .. }
| UnindexedClientError::TruncatedSnapshot { .. }) => OrderError::Connectivity(
ConnectivityError::Socket(format!("Alpaca order status unknown: {other}")),
),
}
}
impl OrderStatusClient for AlpacaClient {
async fn fetch_ended_orders(
&self,
orders: &[UnindexedOrderKey],
) -> Result<Vec<UnindexedInactiveOrder>, UnindexedClientError> {
fetch_ended_by_key(orders, |key| {
fetch_order_lookup(
self.http.clone(),
self.rate_limiter.clone(),
self.config.clone(),
key,
)
})
.await
}
}
impl ShortabilityClient for AlpacaClient {
async fn fetch_shortability(
&self,
instrument: &InstrumentNameExchange,
) -> Result<Shortability, UnindexedClientError> {
let url = asset_url(self.base_url(), instrument)?;
let found: Option<AlpacaAsset> =
rest_lookup_with_retry(&self.rate_limiter, || self.http.get(url.clone())).await?;
let Some(asset) = found else {
return Err(instrument_invalid(
instrument,
"Alpaca knows no asset under this symbol",
));
};
shortability_from_asset(instrument, &asset)
}
}
fn asset_url(
base_url: &str,
instrument: &InstrumentNameExchange,
) -> Result<reqwest::Url, UnindexedClientError> {
let mut url = reqwest::Url::parse(base_url).map_err(|error| {
UnindexedClientError::Internal(format!("Alpaca base URL {base_url:?}: {error}"))
})?;
url.path_segments_mut()
.map_err(|()| {
UnindexedClientError::Internal(format!("Alpaca base URL {base_url:?} has no path"))
})?
.pop_if_empty()
.extend(["v2", "assets", instrument.name().as_str()]);
Ok(url)
}
fn instrument_invalid(instrument: &InstrumentNameExchange, reason: &str) -> UnindexedClientError {
UnindexedClientError::Api(ApiError::InstrumentInvalid(
instrument.clone(),
reason.to_owned(),
))
}
fn shortability_from_asset(
instrument: &InstrumentNameExchange,
asset: &AlpacaAsset,
) -> Result<Shortability, UnindexedClientError> {
if asset.class == AlpacaAssetClass::UsOption {
return Err(instrument_invalid(
instrument,
"shortability does not apply to an option: writing one borrows nothing",
));
}
let easy_to_borrow = match asset.borrow_status {
Some(AlpacaBorrowStatus::EasyToBorrow) => Some(true),
Some(AlpacaBorrowStatus::HardToBorrow) => Some(false),
Some(AlpacaBorrowStatus::Other) | None => asset.easy_to_borrow,
};
let mut shortability = Shortability::new();
if let Some(shortable) = asset.shortable {
shortability = shortability.with_shortable(shortable);
}
if let Some(easy_to_borrow) = easy_to_borrow {
shortability = shortability.with_easy_to_borrow(easy_to_borrow);
if easy_to_borrow {
shortability = shortability.with_fee_rate(Decimal::ZERO);
}
}
Ok(shortability)
}
#[derive(Debug, Deserialize)]
struct AlpacaAsset {
#[serde(default)]
class: AlpacaAssetClass,
#[serde(default)]
shortable: Option<bool>,
#[serde(default)]
borrow_status: Option<AlpacaBorrowStatus>,
#[serde(default)]
easy_to_borrow: Option<bool>,
}
#[derive(Debug, Default, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "snake_case")]
enum AlpacaAssetClass {
UsOption,
#[default]
#[serde(other)]
Other,
}
#[derive(Debug, PartialEq, Eq, Deserialize)]
#[serde(rename_all = "snake_case")]
enum AlpacaBorrowStatus {
EasyToBorrow,
HardToBorrow,
#[serde(other)]
Other,
}
impl BracketOrderClient for AlpacaClient {
async fn open_bracket_order(
&self,
request: UnifiedBracketOrderRequest<ExchangeId, &InstrumentNameExchange>,
) -> UnifiedBracketOrderResult {
let alpaca_request = AlpacaBracketOrderRequest {
instrument: request.key.instrument.clone(),
strategy: request.key.strategy.clone(),
cid: request.key.cid.clone(),
side: request.state.side,
quantity: request.state.quantity,
entry_price: request.state.entry_price,
take_profit_price: request.state.take_profit_price,
stop_loss_price: request.state.stop_loss_price,
stop_loss_limit_price: request.state.stop_loss_limit_price,
time_in_force: request.state.time_in_force,
};
let result = self.open_bracket_order(alpaca_request).await;
UnifiedBracketOrderResult::parent_only(result.parent)
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)] mod tests {
use super::*;
use rust_decimal_macros::dec;
#[test]
fn test_alpaca_config_new_uses_paper_trading_by_default() {
let cfg = AlpacaConfig::new("my_key".into(), "my_secret".into());
assert!(cfg.paper);
}
#[test]
fn test_alpaca_config_debug_redacts_credentials() {
let cfg = AlpacaConfig::new("my_key".into(), "my_secret".into());
let debug = format!("{cfg:?}");
assert!(!debug.contains("my_key"), "api_key should be redacted");
assert!(
!debug.contains("my_secret"),
"secret_key should be redacted"
);
assert!(debug.contains("paper: true"));
}
#[test]
fn test_alpaca_config_deserialize_omitted_paper_defaults_to_paper() {
let cfg: AlpacaConfig = serde_json::from_str(
r#"{
"api_key": "my_key",
"secret_key": "my_secret"
}"#,
)
.unwrap();
assert!(cfg.paper);
assert_eq!(cfg.rest_base_url(), "https://paper-api.alpaca.markets");
}
#[test]
fn test_alpaca_config_deserialize_paper_true() {
let cfg: AlpacaConfig = serde_json::from_str(
r#"{
"api_key": "my_key",
"secret_key": "my_secret",
"paper": true
}"#,
)
.unwrap();
assert!(cfg.paper);
assert_eq!(cfg.rest_base_url(), "https://paper-api.alpaca.markets");
}
#[test]
fn test_alpaca_config_deserialize_paper_false() {
let cfg: AlpacaConfig = serde_json::from_str(
r#"{
"api_key": "my_key",
"secret_key": "my_secret",
"paper": false
}"#,
)
.unwrap();
assert!(!cfg.paper);
assert_eq!(cfg.rest_base_url(), "https://api.alpaca.markets");
}
#[test]
fn test_alpaca_config_deserialize_rejects_test_url_overrides() {
for key in ["base_url_override", "ws_url_override"] {
let json = format!(
r#"{{"api_key": "my_key", "secret_key": "my_secret", "{key}": "http://127.0.0.1:1"}}"#
);
let Err(err) = serde_json::from_str::<AlpacaConfig>(&json) else {
panic!("a config with `{key}` must not deserialize");
};
assert!(
err.to_string().contains(&format!("unknown field `{key}`")),
"unexpected error for `{key}`: {err}"
);
}
}
#[test]
fn test_alpaca_config_urls() {
let paper = AlpacaConfig::paper("k".into(), "s".into());
assert!(paper.rest_base_url().contains("paper-api"));
assert!(paper.ws_url().contains("paper-api"));
let live = AlpacaConfig::production("k".into(), "s".into());
assert!(!live.rest_base_url().contains("paper-api"));
assert!(!live.ws_url().contains("paper-api"));
}
#[test]
#[serial_test::serial]
fn test_alpaca_config_from_env_defaults_to_paper_trading() {
temp_env::with_vars(
[
("ALPACA_API_KEY", Some("my_key")),
("ALPACA_SECRET_KEY", Some("my_secret")),
("ALPACA_PAPER", None),
],
|| {
let cfg = AlpacaConfig::from_env().unwrap();
assert!(cfg.paper);
assert_eq!(cfg.rest_base_url(), "https://paper-api.alpaca.markets");
},
);
}
#[test]
#[serial_test::serial]
fn test_alpaca_config_from_env_accepts_explicit_production() {
temp_env::with_vars(
[
("ALPACA_API_KEY", Some("my_key")),
("ALPACA_SECRET_KEY", Some("my_secret")),
("ALPACA_PAPER", Some("false")),
],
|| {
let cfg = AlpacaConfig::from_env().unwrap();
assert!(!cfg.paper);
assert_eq!(cfg.rest_base_url(), "https://api.alpaca.markets");
},
);
}
#[test]
#[serial_test::serial]
fn test_alpaca_config_from_env_rejects_invalid_paper() {
temp_env::with_vars(
[
("ALPACA_API_KEY", Some("my_key")),
("ALPACA_SECRET_KEY", Some("my_secret")),
("ALPACA_PAPER", Some("maybe")),
],
|| {
let err = AlpacaConfig::from_env().unwrap_err();
assert!(matches!(err, AlpacaConfigError::InvalidPaper(value) if value == "maybe"));
},
);
}
#[test]
#[serial_test::serial]
fn test_alpaca_config_from_env_requires_credentials() {
temp_env::with_vars(
[
("ALPACA_API_KEY", None),
("ALPACA_SECRET_KEY", Some("my_secret")),
("ALPACA_PAPER", None),
],
|| {
let err = AlpacaConfig::from_env().unwrap_err();
assert!(matches!(err, AlpacaConfigError::MissingApiKey));
},
);
}
#[test]
fn test_parse_side() {
assert_eq!(parse_side("buy"), Some(Side::Buy));
assert_eq!(parse_side("sell"), Some(Side::Sell));
assert_eq!(parse_side("Buy"), Some(Side::Buy));
assert_eq!(parse_side("BUY"), Some(Side::Buy));
assert_eq!(parse_side("unknown"), None);
}
#[test]
fn test_parse_order_kind() {
assert_eq!(
parse_order_kind("market", None, None, None),
Some(OrderKind::Market)
);
assert_eq!(
parse_order_kind("limit", None, None, None),
Some(OrderKind::Limit)
);
assert_eq!(
parse_order_kind("stop", Some("150.00"), None, None),
Some(OrderKind::Stop {
trigger_price: Decimal::from_str("150.00").unwrap()
})
);
assert_eq!(parse_order_kind("stop", None, None, None), None);
assert_eq!(
parse_order_kind("stop_limit", Some("145.00"), None, None),
Some(OrderKind::StopLimit {
trigger_price: Decimal::from_str("145.00").unwrap()
})
);
assert_eq!(
parse_order_kind("trailing_stop", None, Some("5.0"), None),
Some(OrderKind::TrailingStop {
offset: Decimal::from_str("5.0").unwrap(),
offset_type: TrailingOffsetType::Percentage,
activation_price: None,
})
);
assert_eq!(
parse_order_kind("trailing_stop", None, None, Some("2.50")),
Some(OrderKind::TrailingStop {
offset: Decimal::from_str("2.50").unwrap(),
offset_type: TrailingOffsetType::Absolute,
activation_price: None,
})
);
assert_eq!(parse_order_kind("trailing_stop", None, None, None), None);
assert_eq!(parse_order_kind("unknown", None, None, None), None);
}
#[test]
fn test_map_order_kind() {
assert_eq!(map_order_kind(OrderKind::Market), Some("market"));
assert_eq!(map_order_kind(OrderKind::Limit), Some("limit"));
assert_eq!(
map_order_kind(OrderKind::Stop {
trigger_price: Decimal::from_str("150.00").unwrap()
}),
Some("stop")
);
assert_eq!(
map_order_kind(OrderKind::StopLimit {
trigger_price: Decimal::from_str("145.00").unwrap()
}),
Some("stop_limit")
);
assert_eq!(
map_order_kind(OrderKind::TrailingStop {
offset: Decimal::from_str("5.0").unwrap(),
offset_type: TrailingOffsetType::Percentage,
activation_price: None,
}),
Some("trailing_stop")
);
assert_eq!(
map_order_kind(OrderKind::TrailingStop {
offset: Decimal::from_str("2.50").unwrap(),
offset_type: TrailingOffsetType::Absolute,
activation_price: None,
}),
Some("trailing_stop")
);
assert_eq!(
map_order_kind(OrderKind::TrailingStop {
offset: Decimal::from_str("5.0").unwrap(),
offset_type: TrailingOffsetType::Percentage,
activation_price: Some(Decimal::from(100)),
}),
None
);
assert_eq!(
map_order_kind(OrderKind::TrailingStopFixedLimit {
offset: Decimal::from_str("5.0").unwrap(),
offset_type: TrailingOffsetType::Percentage,
activation_price: None,
}),
None
);
assert_eq!(
map_order_kind(OrderKind::TrailingStopLimit {
offset: Decimal::from_str("5.0").unwrap(),
offset_type: TrailingOffsetType::Percentage,
stop_price: Decimal::from_str("95.0").unwrap(),
limit_offset: Decimal::from_str("1.0").unwrap(),
}),
None
);
assert_eq!(
map_order_kind(OrderKind::ConditionalUnresolved {
family: crate::order::ConditionalFamily::Stop,
trigger_price: None,
}),
None
);
assert_eq!(
map_order_kind(OrderKind::TakeProfit {
trigger_price: Decimal::from_str("160.00").unwrap()
}),
None
);
assert_eq!(
map_order_kind(OrderKind::TakeProfitLimit {
trigger_price: Decimal::from_str("160.00").unwrap()
}),
None
);
}
#[test]
fn test_map_time_in_force_roundtrip() {
assert_eq!(
map_time_in_force(TimeInForce::GoodUntilCancelled { post_only: false }),
Ok("gtc")
);
assert_eq!(map_time_in_force(TimeInForce::GoodUntilEndOfDay), Ok("day"));
assert_eq!(map_time_in_force(TimeInForce::FillOrKill), Ok("fok"));
assert_eq!(map_time_in_force(TimeInForce::ImmediateOrCancel), Ok("ioc"));
}
#[test]
fn test_map_time_in_force_rejects_post_only() {
let result = map_time_in_force(TimeInForce::GoodUntilCancelled { post_only: true });
assert!(result.is_err(), "post_only must be rejected");
assert!(result.unwrap_err().contains("post_only"));
}
#[test]
fn test_bracket_order_serializes_with_stop_loss_stop_order() {
let body = AlpacaOrderRequest {
symbol: "AAPL",
qty: "10".to_string(),
side: "buy",
order_type: "limit",
time_in_force: "gtc",
limit_price: Some("150.00".to_string()),
stop_price: None,
trail_percent: None,
trail_price: None,
client_order_id: Some("bracket-001"),
position_intent: Some(AlpacaPositionIntent::BuyToOpen),
order_class: Some("bracket"),
take_profit: Some(TakeProfitParams {
limit_price: "160.00".to_string(),
}),
stop_loss: Some(StopLossParams {
stop_price: "145.00".to_string(),
limit_price: None,
}),
};
let json = serde_json::to_value(&body).unwrap();
assert_eq!(json["symbol"], "AAPL");
assert_eq!(json["qty"], "10");
assert_eq!(json["side"], "buy");
assert_eq!(json["type"], "limit");
assert_eq!(json["time_in_force"], "gtc");
assert_eq!(json["limit_price"], "150.00");
assert_eq!(json["order_class"], "bracket");
assert_eq!(json["take_profit"]["limit_price"], "160.00");
assert_eq!(json["stop_loss"]["stop_price"], "145.00");
assert!(
json["stop_loss"].get("limit_price").is_none(),
"stop_loss.limit_price should be omitted when None"
);
}
#[test]
fn test_bracket_order_serializes_with_stop_loss_stop_limit_order() {
let body = AlpacaOrderRequest {
symbol: "SPY",
qty: "5".to_string(),
side: "sell",
order_type: "limit",
time_in_force: "day",
limit_price: Some("450.00".to_string()),
stop_price: None,
trail_percent: None,
trail_price: None,
client_order_id: Some("bracket-002"),
position_intent: Some(AlpacaPositionIntent::SellToClose),
order_class: Some("bracket"),
take_profit: Some(TakeProfitParams {
limit_price: "440.00".to_string(),
}),
stop_loss: Some(StopLossParams {
stop_price: "455.00".to_string(),
limit_price: Some("456.00".to_string()),
}),
};
let json = serde_json::to_value(&body).unwrap();
assert_eq!(json["symbol"], "SPY");
assert_eq!(json["side"], "sell");
assert_eq!(json["time_in_force"], "day");
assert_eq!(json["order_class"], "bracket");
assert_eq!(json["take_profit"]["limit_price"], "440.00");
assert_eq!(json["stop_loss"]["stop_price"], "455.00");
assert_eq!(
json["stop_loss"]["limit_price"], "456.00",
"stop_loss.limit_price should be present for stop-limit orders"
);
}
#[tokio::test]
async fn test_open_bracket_order_rejects_invalid_tif() {
use rust_decimal_macros::dec;
use rustrade_instrument::instrument::name::InstrumentNameExchange;
let config = AlpacaConfig::new("dummy_key".into(), "dummy_secret".into());
let client = AlpacaClient::new(config);
let request = AlpacaBracketOrderRequest::new(
InstrumentNameExchange::new("SPY"),
crate::order::id::StrategyId::new("test"),
crate::order::id::ClientOrderId::new("test-tif"),
Side::Buy,
dec!(1),
dec!(100.00),
dec!(120.00),
dec!(90.00),
TimeInForce::ImmediateOrCancel, );
let result = client.open_bracket_order(request).await;
assert!(
result.parent.state.is_failed(),
"Bracket order with IOC TIF should be rejected locally"
);
}
#[tokio::test]
async fn test_open_bracket_order_rejects_invalid_price_ordering() {
use rust_decimal_macros::dec;
use rustrade_instrument::instrument::name::InstrumentNameExchange;
let config = AlpacaConfig::new("dummy_key".into(), "dummy_secret".into());
let client = AlpacaClient::new(config);
let request = AlpacaBracketOrderRequest::new(
InstrumentNameExchange::new("SPY"),
crate::order::id::StrategyId::new("test"),
crate::order::id::ClientOrderId::new("test-price"),
Side::Buy,
dec!(1),
dec!(100.00),
dec!(120.00),
dec!(105.00), TimeInForce::GoodUntilCancelled { post_only: false },
);
let result = client.open_bracket_order(request).await;
assert!(
result.parent.state.is_failed(),
"Bracket order with invalid price ordering should be rejected locally"
);
}
#[tokio::test]
async fn test_open_bracket_order_rejects_invalid_sl_limit_price() {
use rust_decimal_macros::dec;
use rustrade_instrument::instrument::name::InstrumentNameExchange;
let config = AlpacaConfig::new("dummy_key".into(), "dummy_secret".into());
let client = AlpacaClient::new(config);
let request = AlpacaBracketOrderRequest::new(
InstrumentNameExchange::new("SPY"),
crate::order::id::StrategyId::new("test"),
crate::order::id::ClientOrderId::new("test-sl-limit"),
Side::Buy,
dec!(1),
dec!(100.00),
dec!(120.00),
dec!(90.00),
TimeInForce::GoodUntilCancelled { post_only: false },
)
.with_stop_loss_limit_price(dec!(95.00));
let result = client.open_bracket_order(request).await;
assert!(
result.parent.state.is_failed(),
"Bracket order with invalid SL limit price should be rejected locally"
);
}
#[test]
fn test_non_bracket_order_omits_bracket_fields() {
let body = AlpacaOrderRequest {
symbol: "AAPL",
qty: "1".to_string(),
side: "buy",
order_type: "limit",
time_in_force: "gtc",
limit_price: Some("150.00".to_string()),
stop_price: None,
trail_percent: None,
trail_price: None,
client_order_id: Some("regular-001"),
position_intent: Some(AlpacaPositionIntent::BuyToOpen),
order_class: None,
take_profit: None,
stop_loss: None,
};
let json = serde_json::to_value(&body).unwrap();
assert_eq!(json["symbol"], "AAPL");
assert!(
json.get("order_class").is_none(),
"order_class should be omitted for non-bracket orders"
);
assert!(
json.get("take_profit").is_none(),
"take_profit should be omitted for non-bracket orders"
);
assert!(
json.get("stop_loss").is_none(),
"stop_loss should be omitted for non-bracket orders"
);
}
#[test]
fn test_parse_timestamp_valid() {
let ts = parse_timestamp("2025-04-18T14:30:00Z");
assert!(ts.is_some());
let ts2 = parse_timestamp("2025-04-18T14:30:00.123456Z");
assert!(ts2.is_some());
assert_eq!(parse_timestamp("not-a-timestamp"), None);
}
#[test]
fn test_check_auth_response_authorized() {
let msg =
r#"{"stream":"authorization","data":{"status":"authorized","action":"authenticate"}}"#;
assert!(matches!(check_auth_response(msg), Some(Ok(()))));
}
#[test]
fn test_check_auth_response_unauthorized() {
let msg = r#"{"stream":"authorization","data":{"status":"unauthorized"}}"#;
assert!(matches!(
check_auth_response(msg),
Some(Err(HandshakeError::Auth(_)))
));
}
#[test]
fn test_check_auth_response_non_auth_message() {
let msg = r#"{"stream":"trade_updates","data":{}}"#;
assert!(check_auth_response(msg).is_none());
}
#[test]
fn test_check_listen_ack() {
let ack = r#"{"stream":"listening","data":{"streams":["trade_updates"]}}"#;
assert!(check_listen_ack(ack));
let other = r#"{"stream":"authorization","data":{}}"#;
assert!(!check_listen_ack(other));
}
#[test]
fn test_dedup_cache() {
let cache = new_dedup_cache();
let key = || SmolStr::new("order-1:1");
assert!(
!is_duplicate_fill(&cache, Some("exec-1"), key()),
"first time should not be duplicate"
);
assert!(
is_duplicate_fill(&cache, Some("exec-1"), key()),
"second time should be duplicate"
);
}
#[test]
fn dedup_tells_fills_apart_by_execution_id_not_cumulative() {
let cache = new_dedup_cache();
assert!(!is_duplicate_fill(
&cache,
Some("exec-a"),
SmolStr::new("ord-1:5")
));
assert!(
!is_duplicate_fill(&cache, Some("exec-b"), SmolStr::new("ord-1:5")),
"a different execution at a cumulative seen before is a new fill"
);
}
#[test]
fn dedup_matches_a_fill_with_an_execution_id_to_the_same_fill_without_one() {
let cache = new_dedup_cache();
assert!(!is_duplicate_fill(&cache, None, SmolStr::new("ord-1:2")));
assert!(
is_duplicate_fill(&cache, Some("exec-a"), SmolStr::new("ord-1:2")),
"without an id first, then with one"
);
assert!(!is_duplicate_fill(
&cache,
Some("exec-b"),
SmolStr::new("ord-1:4")
));
assert!(
is_duplicate_fill(&cache, None, SmolStr::new("ord-1:4")),
"with an id first, then without one"
);
}
#[test]
fn the_heartbeat_window_is_as_documented() {
assert_eq!(HEARTBEAT_TIMEOUT_SECS, 35);
}
#[test]
fn test_convert_account_to_balances_empty_assets() {
let account: AlpacaAccount = serde_json::from_value(serde_json::json!({
"cash": "9000.50",
"equity": "12000.00",
"buying_power": "36000.00",
"regt_buying_power": "18000.00",
"options_buying_power": "8000.00",
"non_marginable_buying_power": "8500.25",
}))
.unwrap();
let balances = convert_account_to_balances(&account, &[], Utc::now());
assert_eq!(balances.len(), 1);
assert_eq!(balances[0].asset, AssetNameExchange::new("USD"));
assert_eq!(balances[0].balance.total, dec!(9000.50), "total is cash");
assert_eq!(
balances[0].balance.free,
dec!(8500.25),
"free is non-marginable buying power when below cash"
);
assert_eq!(balances[0].balance.margin, None);
}
#[test]
fn test_convert_account_to_balances_free_capped_at_cash_with_long_stock() {
let account: AlpacaAccount = serde_json::from_value(serde_json::json!({
"cash": "50000",
"equity": "100000",
"initial_margin": "25000",
"non_marginable_buying_power": "75000",
}))
.unwrap();
let balances = convert_account_to_balances(&account, &[], Utc::now());
assert_eq!(balances[0].balance.total, dec!(50000));
assert_eq!(balances[0].balance.free, dec!(50000));
}
#[test]
fn test_convert_account_to_balances_open_short_keeps_free_below_cash() {
let account: AlpacaAccount = serde_json::from_value(serde_json::json!({
"cash": "10200",
"equity": "10000",
"initial_margin": "100",
"non_marginable_buying_power": "9900",
}))
.unwrap();
let balances = convert_account_to_balances(&account, &[], Utc::now());
assert_eq!(balances[0].balance.total, dec!(10200));
assert_eq!(balances[0].balance.free, dec!(9900));
}
#[test]
fn test_convert_account_to_balances_negative_cash_on_margin() {
let account: AlpacaAccount = serde_json::from_value(serde_json::json!({
"cash": "-2500.75",
"equity": "7500.00",
"buying_power": "5000.00",
"non_marginable_buying_power": "1250.00",
}))
.unwrap();
let balances = convert_account_to_balances(&account, &[], Utc::now());
assert_eq!(balances.len(), 1);
assert_eq!(balances[0].balance.total, dec!(-2500.75));
assert_eq!(
balances[0].balance.free,
dec!(-2500.75),
"no free cash while borrowing; free <= total still holds"
);
}
#[test]
fn test_alpaca_account_malformed_cash_fails_to_decode() {
let result = serde_json::from_value::<AlpacaAccount>(serde_json::json!({
"cash": "not-a-number",
"non_marginable_buying_power": "0",
}));
assert!(result.is_err(), "a malformed cash must not read as zero");
}
#[test]
fn test_convert_account_to_balances_usd_filter() {
let account = AlpacaAccount {
cash: dec!(12000.00),
non_marginable_buying_power: dec!(10000.00),
};
let usd = vec![AssetNameExchange::new("USD")];
let balances = convert_account_to_balances(&account, &usd, Utc::now());
assert_eq!(balances.len(), 1);
assert_eq!(balances[0].asset, AssetNameExchange::new("USD"));
let non_usd = vec![AssetNameExchange::new("BTC")];
let balances = convert_account_to_balances(&account, &non_usd, Utc::now());
assert!(balances.is_empty());
}
#[test]
fn a_usd_balance_is_named_usd_however_the_filter_spells_it() {
let account = AlpacaAccount {
cash: dec!(12000.00),
non_marginable_buying_power: dec!(10000.00),
};
let balances =
convert_account_to_balances(&account, &[AssetNameExchange::new("usd")], Utc::now());
assert_eq!(balances.len(), 1);
assert_eq!(balances[0].asset, AssetNameExchange::new("USD"));
}
#[test]
fn test_is_options_or_equity_symbol() {
assert!(!is_options_or_equity_symbol("BTC/USD"));
assert!(!is_options_or_equity_symbol("ETH/USD"));
assert!(!is_options_or_equity_symbol("SOL/USD"));
assert!(is_options_or_equity_symbol("AAPL"));
assert!(is_options_or_equity_symbol("SPY"));
assert!(is_options_or_equity_symbol("MSFT"));
assert!(is_options_or_equity_symbol("SPY250418C00450000"));
assert!(is_options_or_equity_symbol("AAPL250418P00145000"));
}
#[test]
fn test_parse_order_error_already_cancelled() {
assert!(matches!(
parse_order_error(
reqwest::StatusCode::UNPROCESSABLE_ENTITY,
None,
"order is already cancelled"
),
UnindexedOrderError::Rejected(ApiError::OrderAlreadyCancelled)
));
}
#[test]
fn test_parse_order_error_already_wins_over_insufficient_on_422() {
assert!(matches!(
parse_order_error(
reqwest::StatusCode::UNPROCESSABLE_ENTITY,
None,
"order already cancelled due to insufficient margin"
),
UnindexedOrderError::Rejected(ApiError::OrderAlreadyCancelled)
));
}
fn make_order_ws<'a>(
id: &str,
symbol: &str,
side: &str,
filled_qty: &'a str,
) -> AlpacaOrderWs<'a> {
AlpacaOrderWs {
id: SmolStr::new(id),
client_order_id: None,
symbol: SmolStr::new(symbol),
qty: Some("2"),
filled_qty: Some(filled_qty),
side: SmolStr::new(side),
order_type: SmolStr::new("limit"),
time_in_force: SmolStr::new("day"),
limit_price: Some("100.00"),
stop_price: None,
trail_percent: None,
trail_price: None,
status: SmolStr::new("partially_filled"),
updated_at: None,
}
}
fn sole_event(events: [Option<UnindexedAccountEvent>; 2]) -> Option<UnindexedAccountEvent> {
let [first, second] = events;
assert!(
second.is_none(),
"expected a single event, got a second: {second:?}"
);
first
}
fn fill_events(
events: [Option<UnindexedAccountEvent>; 2],
) -> (
Trade<AssetNameExchange, InstrumentNameExchange>,
crate::order::Order<
ExchangeId,
InstrumentNameExchange,
OrderState<AssetNameExchange, InstrumentNameExchange>,
>,
) {
let [first, second] = events;
let first = first.expect("a fill frame must produce an execution");
let second = second.expect("a fill frame must produce an order snapshot");
let AccountEventKind::Trade(trade) = first.kind else {
panic!("first event must be the Trade, got {:?}", first.kind);
};
let AccountEventKind::OrderSnapshot(rustrade_integration::collection::snapshot::Snapshot(
order,
)) = second.kind
else {
panic!(
"second event must be the OrderSnapshot, got {:?}",
second.kind
);
};
(trade, order)
}
#[test]
fn test_convert_trade_update_fill_carries_the_client_order_id() {
let mut order = make_order_ws("ord-1", "SPY", "buy", "1");
order.client_order_id = Some(SmolStr::new("outside-1"));
let update = AlpacaTradeUpdate {
event: SmolStr::new("fill"),
execution_id: Some("exec-1"),
previous_execution_id: None,
order,
price: Some("150.00"),
qty: Some("1"),
timestamp: Some("2025-04-18T14:30:00Z"),
};
let (trade, _order) = fill_events(convert_trade_update(update));
assert_eq!(trade.client_order_id, Some(ClientOrderId::new("outside-1")));
}
#[test]
fn a_fills_time_is_its_timestamp_then_its_orders_update_time_then_this_hosts() {
let fill = |timestamp, updated_at| {
let mut order = make_order_ws("ord-1", "SPY", "buy", "1");
order.updated_at = updated_at;
let update = AlpacaTradeUpdate {
event: SmolStr::new("fill"),
execution_id: Some("exec-1"),
previous_execution_id: None,
order,
price: Some("150.00"),
qty: Some("1"),
timestamp,
};
let (trade, order) = fill_events(convert_trade_update(update));
let OrderState::Active(crate::order::state::ActiveOrderState::Open(open)) = order.state
else {
panic!("expected an Open snapshot, got {:?}", order.state);
};
assert_eq!(open.time_exchange, trade.time_exchange);
(trade.time_exchange, trade.time_source)
};
let at = |time| parse_timestamp(time).unwrap();
assert_eq!(
fill(Some("2025-04-18T14:30:00Z"), Some("2025-04-18T14:30:01Z")),
(at("2025-04-18T14:30:00Z"), TimeSource::Venue)
);
for timestamp in [None, Some("not a time")] {
assert_eq!(
fill(timestamp, Some("2025-04-18T14:30:01Z")),
(at("2025-04-18T14:30:01Z"), TimeSource::Venue),
"{timestamp:?}"
);
}
let before = Utc::now();
let (time, source) = fill(None, None);
assert_eq!(source, TimeSource::Host);
assert!((before..=Utc::now()).contains(&time));
}
#[test]
fn test_convert_trade_update_fill_produces_trade_with_execution_id() {
let update = AlpacaTradeUpdate {
event: SmolStr::new("fill"),
execution_id: Some("exec-1"),
previous_execution_id: None,
order: make_order_ws("ord-1", "SPY", "buy", "1"),
price: Some("150.00"),
qty: Some("1"),
timestamp: Some("2025-04-18T14:30:00Z"),
};
let (trade, order) = fill_events(convert_trade_update(update));
assert_eq!(trade.id.0.as_str(), "exec-1");
assert_eq!(
trade.client_order_id, None,
"the update names no client order id"
);
assert_eq!(trade.price, Decimal::from_str("150.00").unwrap());
assert_eq!(trade.quantity, Decimal::from_str("1").unwrap());
let OrderState::Active(crate::order::state::ActiveOrderState::Open(open)) = order.state
else {
panic!("expected an Open snapshot, got {:?}", order.state);
};
assert_eq!(open.filled_quantity, Decimal::from_str("1").unwrap());
assert_eq!(open.id.assigned().map(|id| id.0.as_str()), Some("ord-1"));
}
#[test]
fn test_convert_trade_update_partial_fill() {
let update = AlpacaTradeUpdate {
event: SmolStr::new("partial_fill"),
execution_id: None,
previous_execution_id: None,
order: make_order_ws("ord-2", "AAPL", "sell", "0.5"),
price: Some("200.00"),
qty: Some("0.5"),
timestamp: None,
};
let (trade, order) = fill_events(convert_trade_update(update));
assert_eq!(trade.quantity, Decimal::from_str("0.5").unwrap());
let OrderState::Active(crate::order::state::ActiveOrderState::Open(open)) = order.state
else {
panic!("expected an Open snapshot, got {:?}", order.state);
};
assert_eq!(open.filled_quantity, Decimal::from_str("0.5").unwrap());
assert_eq!(
open.quantity_remaining(order.quantity),
Decimal::from_str("1.5").unwrap()
);
}
#[test]
fn a_full_fill_snapshot_reports_nothing_left_to_fill() {
let update = AlpacaTradeUpdate {
event: SmolStr::new("fill"),
execution_id: None,
previous_execution_id: None,
order: AlpacaOrderWs {
status: SmolStr::new("filled"),
updated_at: None,
..make_order_ws("ord-full", "SPY", "buy", "2")
},
price: Some("150.00"),
qty: Some("1"),
timestamp: Some("2025-04-18T14:30:00Z"),
};
let (_trade, order) = fill_events(convert_trade_update(update));
let OrderState::Active(crate::order::state::ActiveOrderState::Open(open)) = order.state
else {
panic!("expected an Open snapshot, got {:?}", order.state);
};
assert_eq!(open.filled_quantity, Decimal::from_str("2").unwrap());
assert!(
open.quantity_remaining(order.quantity).is_zero(),
"a completed order must report nothing left to fill so the engine retires it"
);
}
#[test]
fn a_fill_for_an_order_no_longer_working_emits_the_execution_without_a_snapshot() {
for status in [
"canceled",
"expired",
"rejected",
"done_for_day",
"pending_cancel",
] {
let update = AlpacaTradeUpdate {
event: SmolStr::new("partial_fill"),
execution_id: None,
previous_execution_id: None,
order: AlpacaOrderWs {
status: SmolStr::new(status),
updated_at: None,
..make_order_ws("ord-late", "SPY", "buy", "1")
},
price: Some("150.00"),
qty: Some("1"),
timestamp: None,
};
let [trade, snapshot] = convert_trade_update(update);
assert!(
matches!(trade.map(|e| e.kind), Some(AccountEventKind::Trade(_))),
"the execution must still be reported for status {status}"
);
assert!(
snapshot.is_none(),
"status {status} must not produce an order snapshot"
);
}
}
#[test]
fn a_notional_order_fill_emits_the_execution_without_a_snapshot() {
let update = AlpacaTradeUpdate {
event: SmolStr::new("partial_fill"),
execution_id: None,
previous_execution_id: None,
order: AlpacaOrderWs {
qty: None,
..make_order_ws("ord-notional-ws", "SPY", "buy", "1")
},
price: Some("150.00"),
qty: Some("1"),
timestamp: None,
};
let [trade, snapshot] = convert_trade_update(update);
assert!(matches!(
trade.map(|e| e.kind),
Some(AccountEventKind::Trade(_))
));
assert!(
snapshot.is_none(),
"a notional order has no quantity to snapshot against"
);
}
#[test]
fn test_convert_trade_update_new_order_produces_snapshot() {
let update = AlpacaTradeUpdate {
event: SmolStr::new("new"),
execution_id: None,
previous_execution_id: None,
order: AlpacaOrderWs {
id: SmolStr::new("ord-new"),
client_order_id: Some(SmolStr::new("cid-1")),
symbol: SmolStr::new("AAPL"),
qty: Some("10"),
filled_qty: Some("0"),
side: SmolStr::new("buy"),
order_type: SmolStr::new("limit"),
time_in_force: SmolStr::new("day"),
limit_price: Some("150.00"),
stop_price: None,
trail_percent: None,
trail_price: None,
status: SmolStr::new("new"),
updated_at: None,
},
price: None,
qty: None,
timestamp: Some("2025-04-18T14:30:00Z"),
};
let event =
sole_event(convert_trade_update(update)).expect("new event should produce an event");
assert!(matches!(event.kind, AccountEventKind::OrderSnapshot(_)));
}
#[test]
fn test_convert_trade_update_canceled_produces_cancel() {
let update = AlpacaTradeUpdate {
event: SmolStr::new("canceled"),
execution_id: None,
previous_execution_id: None,
order: make_order_ws("ord-3", "AAPL", "sell", "0"),
price: None,
qty: None,
timestamp: Some("2025-04-18T14:30:00Z"),
};
let event =
sole_event(convert_trade_update(update)).expect("canceled should produce an event");
let AccountEventKind::OrderCancelled(response) = event.kind else {
panic!("expected OrderCancelled, got {:?}", event.kind);
};
assert!(response.state.is_ok());
}
#[test]
fn a_canceled_update_carries_its_fill_or_none() {
let cancelled = |filled_qty: Option<&'static str>| {
let mut order = make_order_ws("ord-3", "AAPL", "sell", "0");
order.filled_qty = filled_qty;
let update = AlpacaTradeUpdate {
event: SmolStr::new("canceled"),
execution_id: None,
previous_execution_id: None,
order,
price: None,
qty: None,
timestamp: Some("2025-04-18T14:30:00Z"),
};
let event =
sole_event(convert_trade_update(update)).expect("canceled should produce an event");
let AccountEventKind::OrderCancelled(response) = event.kind else {
panic!("expected OrderCancelled, got {:?}", event.kind);
};
let Ok(cancelled) = response.state else {
panic!("expected Ok, got {:?}", response.state);
};
cancelled.filled_quantity
};
assert_eq!(cancelled(Some("1")), Some(Decimal::ONE));
assert_eq!(cancelled(Some("0")), Some(Decimal::ZERO));
assert_eq!(cancelled(None), None);
assert_eq!(cancelled(Some("not a number")), None);
}
#[test]
fn an_ended_order_with_an_unparseable_fill_reports_it_unknown() {
let key = OrderKey::new(
ExchangeId::AlpacaBroker,
InstrumentNameExchange::new("AAPL"),
StrategyId::new("strategy"),
ClientOrderId::new("c"),
);
let ended = |status: &str| {
let mut order = make_order_response("o1", "AAPL");
order.status = Some(status.to_string());
order.filled_qty = "not a number".to_string();
convert_ended_order(&order, &key).map(|order| order.state)
};
assert!(matches!(
ended("canceled"),
Some(InactiveOrderState::Cancelled(Cancelled {
filled_quantity: None,
..
}))
));
assert!(matches!(
ended("expired"),
Some(InactiveOrderState::Expired(Expired {
filled_quantity: None,
..
}))
));
assert!(matches!(
ended("filled"),
Some(InactiveOrderState::FullyFilled(Filled { filled_quantity, .. }))
if filled_quantity == Decimal::ONE
));
}
#[test]
fn test_convert_trade_update_rejected_produces_error() {
let update = AlpacaTradeUpdate {
event: SmolStr::new("rejected"),
execution_id: None,
previous_execution_id: None,
order: make_order_ws("ord-4", "SPY", "buy", "0"),
price: None,
qty: None,
timestamp: None,
};
let event =
sole_event(convert_trade_update(update)).expect("rejected should produce an event");
let AccountEventKind::OrderCancelled(response) = event.kind else {
panic!("expected OrderCancelled, got {:?}", event.kind);
};
assert!(response.state.is_err());
}
#[test]
fn test_convert_open_order_notional_qty_none_is_skipped() {
let order = AlpacaOrderResponse {
id: "ord-notional".to_string(),
client_order_id: None,
symbol: "SPY".to_string(),
qty: None,
filled_qty: "0".to_string(),
filled_avg_price: None,
status: None,
side: "buy".to_string(),
order_type: "market".to_string(),
time_in_force: "day".to_string(),
limit_price: None,
stop_price: None,
trail_percent: None,
trail_price: None,
created_at: "2025-04-18T14:30:00Z".to_string(),
updated_at: None,
};
assert!(convert_open_order(&order).is_none());
}
#[test]
fn an_activity_without_a_transaction_time_is_at_this_hosts_time() {
let activity = |transaction_time: &str| AlpacaActivity {
id: "act-1".to_string(),
order_id: "ord-1".to_string(),
symbol: "SPY".to_string(),
side: "buy".to_string(),
price: "150.00".to_string(),
qty: "1".to_string(),
transaction_time: transaction_time.to_string(),
cum_qty: Some("1".to_string()),
};
let trade = convert_activity_to_trade(&activity("2025-04-18T14:30:00Z")).unwrap();
assert_eq!(
(trade.time_exchange, trade.time_source),
(
parse_timestamp("2025-04-18T14:30:00Z").unwrap(),
TimeSource::Venue
)
);
let before = Utc::now();
let trade = convert_activity_to_trade(&activity("not a time")).unwrap();
assert_eq!(trade.time_source, TimeSource::Host);
assert!((before..=Utc::now()).contains(&trade.time_exchange));
}
#[test]
fn test_convert_activity_to_trade_bad_price_returns_none() {
let activity = AlpacaActivity {
id: "act-1".to_string(),
order_id: "ord-1".to_string(),
symbol: "SPY250418C00450000".to_string(),
side: "buy".to_string(),
price: "not-a-number".to_string(),
qty: "1".to_string(),
transaction_time: "2025-04-18T14:30:00Z".to_string(),
cum_qty: Some("1".to_string()),
};
assert!(convert_activity_to_trade(&activity).is_none());
}
#[test]
fn test_convert_positions_to_balances_crypto() {
let positions = vec![
AlpacaPosition {
qty_available: dec!(0.4),
..alpaca_position("BTCUSD", "crypto", AlpacaPositionSide::Long, dec!(0.5))
},
alpaca_position("ETH/USD", "crypto", AlpacaPositionSide::Long, dec!(2.0)),
alpaca_position("AAPL", "us_equity", AlpacaPositionSide::Long, dec!(10)),
];
let balances = convert_positions_to_balances(&positions, &[], Utc::now());
assert_eq!(balances.len(), 2, "only crypto positions returned");
assert_eq!(balances[0].asset.name().as_str(), "BTC");
assert_eq!(balances[1].asset.name().as_str(), "ETH");
assert_eq!(balances[0].balance.total, dec!(0.5));
assert_eq!(balances[0].balance.free, dec!(0.4));
let btc_only = vec![AssetNameExchange::new("btc")];
let balances = convert_positions_to_balances(&positions, &btc_only, Utc::now());
assert_eq!(balances.len(), 1);
assert_eq!(balances[0].asset.name().as_str(), "BTC");
}
#[test]
fn test_crypto_position_asset() {
assert_eq!(crypto_position_asset("BTCUSD"), "BTC");
assert_eq!(
crypto_position_asset("btcusd"),
"btc",
"suffix matched ignoring case"
);
assert_eq!(
crypto_position_asset("USDTUSD"),
"USDT",
"only one USD is stripped"
);
assert_eq!(crypto_position_asset("BTC/USD"), "BTC");
assert_eq!(crypto_position_asset("BTC/USDT"), "BTC");
assert_eq!(crypto_position_asset("€USD"), "€");
assert_eq!(crypto_position_asset("USD"), "USD");
assert_eq!(crypto_position_asset("/USD"), "/USD");
assert_eq!(crypto_position_asset("BTCUSDT"), "BTCUSDT");
assert_eq!(
crypto_position_asset("BTCێ"),
"BTCێ",
"no char boundary at the suffix"
);
assert_eq!(crypto_position_asset(""), "");
}
fn alpaca_position(
symbol: &str,
asset_class: &str,
side: AlpacaPositionSide,
qty: Decimal,
) -> AlpacaPosition {
AlpacaPosition {
symbol: symbol.into(),
asset_class: asset_class.into(),
side,
qty,
qty_available: qty,
avg_entry_price: Some(dec!(100)),
unrealized_pl: Some(dec!(0)),
}
}
#[test]
fn test_convert_positions_sign_comes_from_side_not_qty() {
let now = Utc::now();
let positions = vec![
alpaca_position("AAPL", "us_equity", AlpacaPositionSide::Long, dec!(10)),
alpaca_position("TSLA", "us_equity", AlpacaPositionSide::Short, dec!(-3)),
alpaca_position("MSFT", "us_equity", AlpacaPositionSide::Short, dec!(4)),
];
let converted = convert_positions(&positions, now).unwrap();
let quantities: Vec<_> = converted.iter().map(|(s, p)| (*s, p.quantity)).collect();
assert_eq!(
quantities,
vec![("AAPL", dec!(10)), ("TSLA", dec!(-3)), ("MSFT", dec!(-4))]
);
assert!(converted.iter().all(|(_, p)| p.time_exchange == now));
}
#[test]
fn test_convert_positions_leaves_out_crypto_and_flat() {
let positions = vec![
alpaca_position("BTC/USD", "crypto", AlpacaPositionSide::Long, dec!(0.5)),
alpaca_position("AAPL", "us_equity", AlpacaPositionSide::Long, dec!(0)),
alpaca_position("SPY", "us_equity", AlpacaPositionSide::Long, dec!(1)),
];
let converted = convert_positions(&positions, Utc::now()).unwrap();
let symbols: Vec<_> = converted.iter().map(|(s, _)| *s).collect();
assert_eq!(symbols, vec!["SPY"]);
}
#[test]
fn test_alpaca_position_decodes_equity_option_and_fractional() {
let positions: Vec<AlpacaPosition> = serde_json::from_value(serde_json::json!([
{
"asset_id": "00000000-0000-0000-0000-000000000001",
"symbol": "XYZ",
"exchange": "NASDAQ",
"asset_class": "us_equity",
"qty": "-7",
"qty_available": "-7",
"side": "short",
"avg_entry_price": "50.25",
"market_value": "-351.75",
"unrealized_pl": "0",
},
{
"symbol": "XYZ271217C00055000",
"asset_class": "us_option",
"qty": "2",
"qty_available": "2",
"side": "long",
"avg_entry_price": "1.35",
"unrealized_pl": "-20",
},
{
"symbol": "ABC",
"asset_class": "us_equity",
"qty": "0.125",
"qty_available": "0.125",
"side": "long",
"avg_entry_price": "80",
"unrealized_pl": null,
},
]))
.unwrap();
let converted = convert_positions(&positions, Utc::now()).unwrap();
let [
(short_sym, short),
(option_sym, option),
(fraction_sym, fraction),
] = converted.as_slice()
else {
panic!("expected three positions, got {converted:?}");
};
assert_eq!(*short_sym, "XYZ");
assert_eq!(short.quantity, dec!(-7));
assert_eq!(short.entry_price, Some(dec!(50.25)));
assert!(short.is_short());
assert_eq!(*option_sym, "XYZ271217C00055000");
assert_eq!(option.quantity, dec!(2));
assert_eq!(option.entry_price, Some(dec!(1.35)), "per-share premium");
assert_eq!(option.unrealized_pnl, Some(dec!(-20)));
assert_eq!(*fraction_sym, "ABC");
assert_eq!(fraction.quantity, dec!(0.125));
assert_eq!(fraction.unrealized_pnl, None);
for (_, p) in &converted {
assert_eq!(p.margin_used, None);
assert_eq!(p.liquidation_price, None);
assert_eq!(p.leverage, None);
}
}
#[test]
fn test_alpaca_position_unknown_side_fails_only_where_a_direction_is_needed() {
let positions: Vec<AlpacaPosition> = serde_json::from_value(serde_json::json!([
{
"symbol": "BTC/USD",
"asset_class": "crypto",
"qty": "0.5",
"qty_available": "0.5",
},
{
"symbol": "XYZ",
"asset_class": "us_equity",
"qty": "1",
"qty_available": "1",
"side": "sideways",
},
]))
.unwrap();
assert_eq!(positions[0].side, AlpacaPositionSide::Unknown);
assert_eq!(positions[1].side, AlpacaPositionSide::Unknown);
let balances = convert_positions_to_balances(&positions, &[], Utc::now());
assert_eq!(balances.len(), 1);
assert_eq!(balances[0].balance.total, dec!(0.5));
let result = convert_positions(&positions, Utc::now());
assert!(
matches!(&result, Err(UnindexedClientError::Internal(msg)) if msg.contains("XYZ")),
"an unknown side must not guess a direction: {result:?}"
);
}
#[test]
fn order_state_time_prefers_updated_at() {
let mut resp = make_order_response("ord-1", "SPY");
resp.created_at = "2025-04-18T14:30:00Z".to_string();
resp.updated_at = Some("2025-04-18T15:45:00Z".to_string());
assert_eq!(
order_state_time(&resp),
parse_timestamp("2025-04-18T15:45:00Z").unwrap(),
"updated_at must win over created_at"
);
resp.updated_at = None;
assert_eq!(
order_state_time(&resp),
parse_timestamp("2025-04-18T14:30:00Z").unwrap(),
"absent updated_at falls back to created_at"
);
resp.updated_at = Some("not-a-timestamp".to_string());
assert_eq!(
order_state_time(&resp),
parse_timestamp("2025-04-18T14:30:00Z").unwrap(),
"unparseable updated_at falls back to created_at"
);
}
#[test]
fn a_placement_response_without_a_status_reads_from_what_filled() {
let instrument = InstrumentNameExchange::new("SPY");
let mut resp = make_order_response("ord-1", "SPY");
resp.filled_qty = "0.4".to_string();
let OrderState::Active(ActiveOrderState::Open(open)) =
placed_order_state(&resp, &instrument, Decimal::ONE)
else {
panic!("expected Open");
};
assert_eq!(open.filled_quantity, Decimal::new(4, 1));
resp.filled_qty = "1".to_string();
resp.filled_avg_price = Some("100.5".to_string());
let state = placed_order_state(&resp, &instrument, Decimal::ONE);
let OrderState::Inactive(InactiveOrderState::FullyFilled(filled)) = state else {
panic!("expected FullyFilled, got {state:?}");
};
assert_eq!(filled.filled_quantity, Decimal::ONE);
assert_eq!(filled.avg_price, Some(Decimal::new(1005, 1)));
}
fn make_order_response(id: &str, symbol: &str) -> AlpacaOrderResponse {
AlpacaOrderResponse {
id: id.to_string(),
client_order_id: None,
symbol: symbol.to_string(),
qty: Some("1".to_string()),
filled_qty: "0".to_string(),
filled_avg_price: None,
status: None,
side: "buy".to_string(),
order_type: "limit".to_string(),
time_in_force: "day".to_string(),
limit_price: Some("100.00".to_string()),
stop_price: None,
trail_percent: None,
trail_price: None,
created_at: "2025-04-18T14:30:00Z".to_string(),
updated_at: None,
}
}
#[test]
fn test_build_instrument_snapshots_empty_instruments_returns_only_with_orders() {
let orders = vec![
make_order_response("o1", "AAPL"),
make_order_response("o2", "SPY"),
];
let snapshots = build_instrument_snapshots(orders, Vec::new(), &[]);
assert_eq!(snapshots.len(), 2);
let symbols: Vec<&str> = snapshots
.iter()
.map(|s| s.instrument.name().as_str())
.collect();
assert!(symbols.contains(&"AAPL"));
assert!(symbols.contains(&"SPY"));
}
#[test]
fn test_build_instrument_snapshots_requested_instrument_no_orders_gets_empty_snapshot() {
let orders = vec![make_order_response("o1", "AAPL")];
let instruments = vec![
InstrumentNameExchange::new("AAPL"),
InstrumentNameExchange::new("SPY"),
];
let snapshots = build_instrument_snapshots(orders, Vec::new(), &instruments);
assert_eq!(snapshots.len(), 2);
let spy = snapshots
.iter()
.find(|s| s.instrument.name().as_str() == "SPY")
.expect("SPY snapshot must be present even with no orders");
assert!(spy.orders.is_empty());
}
fn make_notional_order_response(id: &str, symbol: &str) -> AlpacaOrderResponse {
AlpacaOrderResponse {
qty: None,
..make_order_response(id, symbol)
}
}
fn snapshot_for<'a>(
snapshots: &'a [InstrumentAccountSnapshot<
ExchangeId,
AssetNameExchange,
InstrumentNameExchange,
>],
symbol: &str,
) -> &'a InstrumentAccountSnapshot<ExchangeId, AssetNameExchange, InstrumentNameExchange> {
snapshots
.iter()
.find(|snapshot| snapshot.instrument.name().as_str() == symbol)
.unwrap_or_else(|| panic!("no snapshot for {symbol}"))
}
#[test]
fn test_build_instrument_snapshots_declares_complete_lists_complete() {
let instruments = [
InstrumentNameExchange::new("AAPL"),
InstrumentNameExchange::new("SPY"),
];
for requested in [&instruments[..], &[]] {
let orders = vec![make_order_response("o1", "AAPL")];
let snapshots = build_instrument_snapshots(orders, Vec::new(), requested);
assert!(!snapshots.is_empty());
assert!(snapshots.iter().all(|snapshot| snapshot.orders_complete));
}
}
#[test]
fn test_build_instrument_snapshots_notional_order_leaves_only_its_instrument_incomplete() {
let orders = vec![
make_order_response("o1", "AAPL"),
make_order_response("o2", "SPY"),
make_notional_order_response("o3", "SPY"),
];
let instruments = [
InstrumentNameExchange::new("AAPL"),
InstrumentNameExchange::new("SPY"),
];
let snapshots = build_instrument_snapshots(orders, Vec::new(), &instruments);
assert!(snapshot_for(&snapshots, "AAPL").orders_complete);
let spy = snapshot_for(&snapshots, "SPY");
assert!(!spy.orders_complete);
assert_eq!(spy.orders.len(), 1);
}
#[test]
fn test_build_instrument_snapshots_unfiltered_lists_an_instrument_with_only_a_notional_order() {
let orders = vec![make_notional_order_response("o1", "SPY")];
let snapshots = build_instrument_snapshots(orders, Vec::new(), &[]);
let spy = snapshot_for(&snapshots, "SPY");
assert!(!spy.orders_complete);
assert!(spy.orders.is_empty());
}
#[test]
fn test_build_instrument_snapshots_non_requested_instrument_excluded() {
let orders = vec![
make_order_response("o1", "AAPL"),
make_order_response("o2", "MSFT"), ];
let instruments = vec![InstrumentNameExchange::new("AAPL")];
let snapshots = build_instrument_snapshots(orders, Vec::new(), &instruments);
assert_eq!(snapshots.len(), 1);
assert_eq!(snapshots[0].instrument.name().as_str(), "AAPL");
}
fn position(quantity: Decimal) -> Position {
Position::new(
quantity,
Some(dec!(100)),
None,
None,
None,
None,
None,
Utc::now(),
)
}
#[test]
fn test_build_instrument_snapshots_unfiltered_lists_positions_without_orders() {
let orders = vec![
make_order_response("o1", "AAPL"),
make_order_response("o2", "MSFT"),
make_order_response("o3", "BTC/USD"),
];
let positions = vec![("AAPL", position(dec!(5))), ("TSLA", position(dec!(-2)))];
let snapshots = build_instrument_snapshots(orders, positions, &[]);
let listed: Vec<_> = snapshots
.iter()
.map(|s| {
(
s.instrument.name().as_str(),
s.orders.len(),
s.position.quantity(),
)
})
.collect();
assert_eq!(
listed,
vec![
("AAPL", 1, Some(dec!(5))),
("MSFT", 1, Some(Decimal::ZERO)),
("BTC/USD", 1, None),
("TSLA", 0, Some(dec!(-2)))
]
);
assert!(snapshots.iter().all(|s| s.orders_complete));
}
#[test]
fn test_build_instrument_snapshots_filtered_attaches_only_requested_positions() {
let positions = vec![("AAPL", position(dec!(5))), ("TSLA", position(dec!(-2)))];
let instruments = vec![
InstrumentNameExchange::new("tsla"),
InstrumentNameExchange::new("MSFT"),
InstrumentNameExchange::new("BTC/USD"),
InstrumentNameExchange::new("TSLA"),
];
let snapshots = build_instrument_snapshots(Vec::new(), positions, &instruments);
let listed: Vec<_> = snapshots
.iter()
.map(|s| (s.instrument.name().as_str(), s.position.quantity()))
.collect();
assert_eq!(
listed,
vec![
("tsla", Some(dec!(-2))),
("MSFT", Some(Decimal::ZERO)),
("BTC/USD", None),
("TSLA", None)
]
);
assert!(!snapshots[3].orders_complete);
}
fn epoch(secs: i64) -> DateTime<Utc> {
DateTime::<Utc>::from_timestamp(1_800_000_000 + secs, 0).unwrap()
}
#[test]
fn fill_recovery_from_a_host_clock_ahead_starts_at_alpacas_time() {
let base = tokio::time::Instant::now();
let mut clock = VenueClock::new();
clock.observe(epoch(0), base);
let last = HostMoment::new(epoch(60), base);
assert_eq!(
fill_recovery_start(&clock, last),
epoch(0) - TimeDelta::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS)
);
}
#[test]
fn fill_recovery_with_no_sample_allows_for_the_unsampled_skew() {
let last = HostMoment::new(epoch(60), tokio::time::Instant::now());
assert_eq!(
fill_recovery_start(&VenueClock::new(), last),
epoch(30) - TimeDelta::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS)
);
assert_eq!(UNSAMPLED_CLOCK_SKEW, TimeDelta::seconds(30));
}
#[test]
fn a_trade_update_samples_the_streams_clock() {
let (tx, _rx) = mpsc::unbounded_channel();
let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
let frame = |stamps: &str| {
format!(
r#"{{"stream":"trade_updates","data":{{"event":"new",{stamps}"order":{{"id":"id-a","client_order_id":"a","symbol":"SPY","qty":"2","filled_qty":"0","side":"buy","type":"limit","time_in_force":"gtc","limit_price":"100","status":"new"{order_stamp}}}}}}}"#,
order_stamp = if stamps.is_empty() {
r#","updated_at":"2027-01-15T08:00:05Z""#
} else {
""
},
)
};
let stamped = |frame: &str| {
let mut clock = VenueClock::new();
let before = tokio::time::Instant::now();
process_ws_text(frame, &tx, &new_dedup_cache(), &known, &mut clock);
clock.earliest_at(before)
};
let utc = |s: &str| DateTime::parse_from_rfc3339(s).unwrap().with_timezone(&Utc);
let by_timestamp = stamped(&frame(r#""timestamp":"2027-01-15T08:00:00Z","#)).unwrap();
assert!(
by_timestamp <= utc("2027-01-15T08:00:00Z"),
"{by_timestamp}"
);
assert!(by_timestamp > utc("2027-01-15T07:59:59Z"), "{by_timestamp}");
let by_updated_at = stamped(&frame("")).unwrap();
assert!(
by_updated_at <= utc("2027-01-15T08:00:05Z"),
"{by_updated_at}"
);
assert!(
by_updated_at > utc("2027-01-15T08:00:04Z"),
"{by_updated_at}"
);
}
#[test]
fn test_recover_fills_dedup_key_matches_ws_path() {
let order_id = "ord-1";
let ws_keys: Vec<SmolStr> = ["1", "2"]
.iter()
.map(|filled_qty| {
early_dedup_key(&AlpacaTradeUpdate {
event: SmolStr::new("partial_fill"),
execution_id: None,
previous_execution_id: None,
order: make_order_ws(order_id, "SPY", "buy", filled_qty),
price: Some("150.00"),
qty: Some("1"),
timestamp: None,
})
})
.collect();
let mut cumulative = Decimal::ZERO;
let rest_keys: Vec<SmolStr> = ["1", "1"]
.iter()
.map(|exec_qty| {
cumulative += Decimal::from_str(exec_qty).unwrap();
fill_dedup_key(order_id, cumulative)
})
.collect();
assert_eq!(
ws_keys, rest_keys,
"REST recovery dedup keys must match WS path keys for cross-source dedup to work"
);
assert_eq!(ws_keys[0].as_str(), "ord-1:1");
assert_eq!(ws_keys[1].as_str(), "ord-1:2");
}
#[test]
fn a_ws_fill_without_an_execution_id_is_identified_by_its_dedup_key() {
let update = AlpacaTradeUpdate {
event: SmolStr::new("fill"),
execution_id: None,
previous_execution_id: None,
order: make_order_ws("ord-abc", "SPY", "buy", "5"),
price: Some("150.00"),
qty: Some("5"),
timestamp: None,
};
let early_key = early_dedup_key(&update);
let [event, _snapshot] = convert_trade_update(update);
let Some(UnindexedAccountEvent {
kind: AccountEventKind::Trade(trade),
..
}) = event
else {
panic!("fill should produce an execution");
};
assert_eq!(
trade.id.0.as_str(),
early_key.as_str(),
"without an execution_id the TradeId is the dedup key"
);
assert_eq!(early_key.as_str(), "ord-abc:5");
}
#[test]
fn early_dedup_key_normalizes_decimal_strings() {
let update1 = AlpacaTradeUpdate {
event: SmolStr::new("fill"),
execution_id: None,
previous_execution_id: None,
order: AlpacaOrderWs {
id: SmolStr::new("ord-x"),
client_order_id: Some(SmolStr::new("cid")),
symbol: SmolStr::new("AAPL"),
qty: Some("10"),
filled_qty: Some("1.00"),
side: SmolStr::new("buy"),
order_type: SmolStr::new("market"),
time_in_force: SmolStr::new("day"),
limit_price: None,
stop_price: None,
trail_percent: None,
trail_price: None,
status: SmolStr::new("filled"),
updated_at: None,
},
price: Some("100.00"),
qty: Some("10"),
timestamp: None,
};
assert_eq!(early_dedup_key(&update1).as_str(), "ord-x:1");
let update2 = AlpacaTradeUpdate {
event: SmolStr::new("fill"),
execution_id: None,
previous_execution_id: None,
order: AlpacaOrderWs {
id: SmolStr::new("ord-x"),
client_order_id: Some(SmolStr::new("cid")),
symbol: SmolStr::new("AAPL"),
qty: Some("10"),
filled_qty: Some("1"),
side: SmolStr::new("buy"),
order_type: SmolStr::new("market"),
time_in_force: SmolStr::new("day"),
limit_price: None,
stop_price: None,
trail_percent: None,
trail_price: None,
status: SmolStr::new("filled"),
updated_at: None,
},
price: Some("100.00"),
qty: Some("10"),
timestamp: None,
};
assert_eq!(early_dedup_key(&update2).as_str(), "ord-x:1");
let update3 = AlpacaTradeUpdate {
event: SmolStr::new("fill"),
execution_id: None,
previous_execution_id: None,
order: AlpacaOrderWs {
id: SmolStr::new("ord-x"),
client_order_id: Some(SmolStr::new("cid")),
symbol: SmolStr::new("AAPL"),
qty: Some("10"),
filled_qty: Some("1.0"),
side: SmolStr::new("buy"),
order_type: SmolStr::new("market"),
time_in_force: SmolStr::new("day"),
limit_price: None,
stop_price: None,
trail_percent: None,
trail_price: None,
status: SmolStr::new("filled"),
updated_at: None,
},
price: Some("100.00"),
qty: Some("10"),
timestamp: None,
};
assert_eq!(early_dedup_key(&update3).as_str(), "ord-x:1");
}
#[test]
fn process_ws_text_rejected_event_without_filled_qty_is_not_dropped() {
let (tx, mut rx) = mpsc::unbounded_channel();
let dedup = new_dedup_cache();
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"}}}"#;
let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
process_ws_text(json, &tx, &dedup, &known, &mut VenueClock::new());
let event = rx.try_recv()
.expect("rejected event without filled_qty must produce an AccountEvent, not be silently dropped");
assert!(
matches!(event.kind, AccountEventKind::OrderCancelled(_)),
"rejected event must map to OrderCancelled, got: {:?}",
event.kind
);
}
#[test]
fn process_ws_text_identifies_a_fill_by_its_execution_id() {
let frame = |execution_id: &str| {
format!(
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"}}}}"#
)
};
let cases = [
(
r#""execution_id":"524b1902-817e-446c-825b-a9fcfebbc17e","#,
"524b1902-817e-446c-825b-a9fcfebbc17e",
),
(r#""execution_id":null,"#, "ord-1:2"),
(r#""execution_id":"","#, "ord-1:2"),
("", "ord-1:2"),
];
for (field, expected) in cases {
let (tx, mut rx) = mpsc::unbounded_channel();
let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
process_ws_text(
&frame(field),
&tx,
&new_dedup_cache(),
&known,
&mut VenueClock::new(),
);
let Ok(UnindexedAccountEvent {
kind: AccountEventKind::Trade(trade),
..
}) = rx.try_recv()
else {
panic!("the fill frame {field:?} produces a Trade");
};
assert_eq!(trade.id.0.as_str(), expected, "frame {field:?}");
}
}
fn amendment_frame(event: &str, fields: &str) -> String {
format!(
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"}}}}"#
)
}
fn stream_events(frames: &[String]) -> Vec<UnindexedAccountEvent> {
let (tx, mut rx) = mpsc::unbounded_channel();
let dedup = new_dedup_cache();
let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
for frame in frames {
process_ws_text(frame, &tx, &dedup, &known, &mut VenueClock::new());
}
std::iter::from_fn(|| rx.try_recv().ok()).collect()
}
fn amendment_of(
frame: String,
) -> crate::trade::TradeAmendment<AssetNameExchange, InstrumentNameExchange> {
let events = stream_events(&[frame]);
let [
UnindexedAccountEvent {
kind: AccountEventKind::TradeAmended(amendment),
..
},
] = <[_; 1]>::try_from(events).unwrap_or_else(|events| {
panic!("one TradeAmended, got {events:?}");
})
else {
panic!("expected TradeAmended");
};
amendment
}
#[test]
fn process_ws_text_reports_a_trade_bust() {
let amendment = amendment_of(amendment_frame(
"trade_bust",
r#""execution_id":"exec-x","previous_execution_id":"exec-a","qty":"-2","#,
));
assert_eq!(amendment.original, Some(TradeId::new("exec-a")));
assert_eq!(amendment.order_id, OrderId::new("ord-1"));
assert_eq!(amendment.instrument, InstrumentNameExchange::new("SPY"));
assert_eq!(
Some(amendment.time_exchange),
parse_timestamp("2025-04-18T14:31:00Z")
);
assert_eq!(
amendment.kind,
TradeAmendmentKind::Busted {
quantity: Some(Decimal::TWO)
}
);
}
#[test]
fn process_ws_text_reports_a_bust_that_names_no_fill() {
let amendment = amendment_of(amendment_frame("trade_bust", ""));
assert_eq!(amendment.original, None);
assert_eq!(
amendment.kind,
TradeAmendmentKind::Busted { quantity: None }
);
}
#[test]
fn process_ws_text_reports_a_trade_correct_with_its_replacement() {
let amendment = amendment_of(amendment_frame(
"trade_correct",
r#""execution_id":"exec-c","previous_execution_id":"exec-a","price":"101.50","qty":"2","#,
));
assert_eq!(amendment.original, Some(TradeId::new("exec-a")));
let TradeAmendmentKind::Corrected { replacement } = amendment.kind else {
panic!("a full replacement: {:?}", amendment.kind);
};
assert_eq!(replacement.id, TradeId::new("exec-c"));
assert_eq!(replacement.order_id, OrderId::new("ord-1"));
assert_eq!(replacement.side, Side::Buy);
assert_eq!(replacement.price, Decimal::new(10150, 2));
assert_eq!(replacement.quantity, Decimal::TWO);
assert_eq!(replacement.order_filled_quantity, None);
assert_eq!(replacement.time_exchange, amendment.time_exchange);
}
#[test]
fn process_ws_text_reports_a_correction_it_cannot_resolve() {
let price = Some(Decimal::new(10150, 2));
let cases = [
(
r#""execution_id":"exec-c","qty":"2","#,
(Some("exec-c"), None, Some(Decimal::TWO)),
),
(
r#""price":"101.50","qty":"2","#,
(None, price, Some(Decimal::TWO)),
),
(
r#""execution_id":"exec-c","price":"101.50","qty":"0","#,
(Some("exec-c"), price, Some(Decimal::ZERO)),
),
(
r#""execution_id":"exec-c","price":"101.50","#,
(Some("exec-c"), price, None),
),
];
for (fields, (id, price, quantity)) in cases {
let amendment = amendment_of(amendment_frame("trade_correct", fields));
assert_eq!(
amendment.kind,
TradeAmendmentKind::CorrectedUnresolved {
id: id.map(TradeId::new),
price,
quantity,
},
"{fields}"
);
}
}
#[test]
fn a_fill_after_a_bust_is_delivered_at_a_cumulative_seen_before() {
let fill = |execution_id: &str| {
format!(
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"}}}}"#
)
};
let events = stream_events(&[
fill("exec-a"),
amendment_frame(
"trade_bust",
r#""execution_id":"exec-x","previous_execution_id":"exec-a","qty":"-2","#,
),
fill("exec-b"),
]);
let kinds: Vec<_> = events
.iter()
.filter_map(|event| match &event.kind {
AccountEventKind::Trade(trade) => Some(format!("trade {}", trade.id)),
AccountEventKind::TradeAmended(amendment) => {
Some(format!("amended {:?}", amendment.original))
}
_ => None,
})
.collect();
assert_eq!(
kinds,
vec![
"trade exec-a",
"amended Some(TradeId(\"exec-a\"))",
"trade exec-b"
]
);
}
#[test]
fn decimal_zero_normalize_is_zero_str() {
assert_eq!(Decimal::ZERO.normalize().to_string(), "0");
}
#[test]
fn decimal_normalize_strips_trailing_zeros() {
let from_rest = Decimal::from_str("1.00").unwrap().normalize();
let from_ws = Decimal::from_str("1").unwrap().normalize();
assert_eq!(from_rest.to_string(), from_ws.to_string());
assert_eq!(from_rest.to_string(), "1");
assert_eq!(
Decimal::from_str("100.000")
.unwrap()
.normalize()
.to_string(),
"100"
);
assert_eq!(
Decimal::from_str("0.10").unwrap().normalize().to_string(),
"0.1"
);
assert_eq!(
Decimal::from_str("0.100").unwrap().normalize().to_string(),
"0.1"
);
}
#[test]
fn map_position_intent_open_buy_maps_to_buy_to_open() {
assert_eq!(
map_position_intent(Side::Buy, ReduceOnly::No),
AlpacaPositionIntent::BuyToOpen
);
}
#[test]
fn map_position_intent_open_sell_maps_to_sell_to_open() {
assert_eq!(
map_position_intent(Side::Sell, ReduceOnly::No),
AlpacaPositionIntent::SellToOpen
);
}
#[test]
fn map_position_intent_reduce_buy_maps_to_buy_to_close() {
assert_eq!(
map_position_intent(Side::Buy, ReduceOnly::Position),
AlpacaPositionIntent::BuyToClose
);
}
#[test]
fn map_position_intent_reduce_sell_maps_to_sell_to_close() {
assert_eq!(
map_position_intent(Side::Sell, ReduceOnly::Position),
AlpacaPositionIntent::SellToClose
);
}
#[test]
fn map_position_intent_maps_a_free_base_close_as_any_other_close() {
assert_eq!(
map_position_intent(Side::Sell, ReduceOnly::PositionAndFreeBase),
AlpacaPositionIntent::SellToClose
);
assert_eq!(
map_position_intent(Side::Buy, ReduceOnly::PositionAndFreeBase),
AlpacaPositionIntent::BuyToClose
);
}
#[test]
fn parse_order_error_401_maps_to_unauthenticated() {
assert!(matches!(
parse_order_error(reqwest::StatusCode::UNAUTHORIZED, None, "bad credentials"),
UnindexedOrderError::Rejected(ApiError::Unauthenticated(_))
));
}
#[test]
fn parse_order_error_403_without_code_maps_to_unauthenticated() {
assert!(matches!(
parse_order_error(reqwest::StatusCode::FORBIDDEN, None, "forbidden."),
UnindexedOrderError::Rejected(ApiError::Unauthenticated(_))
));
}
#[test]
fn parse_order_error_403_with_code_and_insufficient_maps_to_balance_insufficient() {
for message in [
"insufficient buying power",
"insufficient day trading buying power",
"insufficient qty available for order (requested: 10, available: 0)",
] {
let err = parse_order_error(reqwest::StatusCode::FORBIDDEN, Some(40310000), message);
let UnindexedOrderError::Rejected(ApiError::BalanceInsufficient(None, text)) = err
else {
panic!("expected BalanceInsufficient for {message:?}, got {err:?}");
};
assert_eq!(text, format!("40310000 {message}"));
}
}
#[test]
fn parse_order_error_reads_a_refused_short_by_its_text() {
for (status, code, message, reason) in [
(
reqwest::StatusCode::FORBIDDEN,
Some(40310000),
"account is not allowed to short",
BorrowRejectReason::BorrowDisabled,
),
(
reqwest::StatusCode::FORBIDDEN,
Some(40310000),
"asset XYZ cannot be sold short",
BorrowRejectReason::NotShortable,
),
(
reqwest::StatusCode::UNPROCESSABLE_ENTITY,
Some(42210000),
"asset XYZ cannot be sold short",
BorrowRejectReason::NotShortable,
),
(
reqwest::StatusCode::UNPROCESSABLE_ENTITY,
None,
"account is not allowed to short",
BorrowRejectReason::BorrowDisabled,
),
] {
assert_eq!(
parse_order_error(status, code, message),
UnindexedOrderError::Rejected(ApiError::BorrowRejected(BorrowReject::new(
reason,
code.map(|code| code.to_string()),
message.to_owned(),
))),
"{status} {message:?}"
);
}
for (status, code) in [
(reqwest::StatusCode::UNPROCESSABLE_ENTITY, 42210000),
(reqwest::StatusCode::FORBIDDEN, 40310000),
] {
assert_eq!(
parse_order_error(status, Some(code), "fractional orders cannot be sold short"),
UnindexedOrderError::Rejected(ApiError::OrderRejected(format!(
"{code} fractional orders cannot be sold short"
))),
"{status}"
);
}
assert!(matches!(
parse_order_error(
reqwest::StatusCode::FORBIDDEN,
Some(40310000),
"insufficient qty: account is not allowed to short",
),
UnindexedOrderError::Rejected(ApiError::BorrowRejected(_))
));
assert!(matches!(
parse_order_error(
reqwest::StatusCode::FORBIDDEN,
None,
"account is not allowed to short"
),
UnindexedOrderError::Rejected(ApiError::Unauthenticated(_))
));
}
#[test]
fn parse_order_error_403_with_code_maps_to_order_rejected() {
for (code, message) in [
(40310000, "account is restricted to liquidation only"),
(
40310100,
"trade denied due to pattern day trading protection",
),
] {
let err = parse_order_error(reqwest::StatusCode::FORBIDDEN, Some(code), message);
let UnindexedOrderError::Rejected(ApiError::OrderRejected(text)) = err else {
panic!("expected OrderRejected for {message:?}, got {err:?}");
};
assert_eq!(text, format!("{code} {message}"));
}
}
#[test]
fn alpaca_api_error_reads_code_and_falls_back_to_raw_text() {
let coded = AlpacaApiError::from_body(
br#"{"code": 40310000, "message": "insufficient buying power"}"#,
);
assert_eq!(coded.code, Some(40310000));
assert_eq!(coded.message, "insufficient buying power");
let bare = AlpacaApiError::from_body(br#"{"message": "forbidden."}"#);
assert_eq!(bare.code, None);
assert_eq!(bare.message, "forbidden.");
let raw = AlpacaApiError::from_body(b"upstream connect error");
assert_eq!(raw.code, None);
assert_eq!(raw.message, "upstream connect error");
}
#[test]
fn parse_order_error_404_maps_to_order_rejected_with_not_found_prefix() {
let err = parse_order_error(reqwest::StatusCode::NOT_FOUND, None, "order not found");
let UnindexedOrderError::Rejected(ApiError::OrderRejected(msg)) = err else {
panic!("expected OrderRejected, got {err:?}");
};
assert!(
msg.contains("order not found"),
"message should contain 'order not found': {msg}"
);
}
#[test]
fn parse_order_error_422_insufficient_only_maps_to_balance_insufficient() {
assert!(matches!(
parse_order_error(
reqwest::StatusCode::UNPROCESSABLE_ENTITY,
None,
"insufficient funds for this order"
),
UnindexedOrderError::Rejected(ApiError::BalanceInsufficient(None, _))
));
}
#[test]
fn parse_order_error_429_maps_to_rate_limit() {
assert!(matches!(
parse_order_error(reqwest::StatusCode::TOO_MANY_REQUESTS, None, "rate limited"),
UnindexedOrderError::Rejected(ApiError::RateLimit {
retry_after: None,
..
})
));
}
#[test]
fn a_rate_limit_delay_is_capped_at_alpacas_window() {
let window = Some(Duration::from_secs(DEFAULT_RATE_LIMIT_DELAY_SECS));
for (name, value) in [("retry-after", u64::MAX), ("x-ratelimit-reset", u64::MAX)] {
let mut headers = reqwest::header::HeaderMap::new();
headers.insert(name, value.to_string().parse().unwrap());
assert_eq!(parse_rate_limit_delay(&headers), window, "{name}");
RateLimitTracker::new().on_rate_limited(parse_rate_limit_delay(&headers));
}
let mut headers = reqwest::header::HeaderMap::new();
headers.insert("retry-after", "0".parse().unwrap());
assert_eq!(
parse_rate_limit_delay(&headers),
Some(Duration::ZERO),
"taken as given"
);
}
#[test]
fn parse_time_in_force_unknown_value_falls_back_to_good_until_end_of_day() {
assert_eq!(
parse_time_in_force("opg"),
TimeInForce::GoodUntilEndOfDay,
"unknown TIF must fall back to GoodUntilEndOfDay (with a warn! in production)"
);
}
#[test]
fn build_instrument_snapshots_output_order_matches_instruments_slice() {
let orders = vec![
make_order_response("o1", "SPY"),
make_order_response("o2", "AAPL"),
make_order_response("o3", "MSFT"),
];
let instruments = vec![
InstrumentNameExchange::new("MSFT"),
InstrumentNameExchange::new("AAPL"),
];
let snapshots = build_instrument_snapshots(orders, Vec::new(), &instruments);
assert_eq!(snapshots.len(), 2);
assert_eq!(
snapshots[0].instrument.name().as_str(),
"MSFT",
"first snapshot must be MSFT (first in instruments slice)"
);
assert_eq!(
snapshots[1].instrument.name().as_str(),
"AAPL",
"second snapshot must be AAPL (second in instruments slice)"
);
}
mod http_tests {
use super::super::*;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, Request, Respond, ResponseTemplate};
struct Sequential {
call: std::sync::atomic::AtomicU32,
pages: Vec<serde_json::Value>,
}
impl Sequential {
fn new(pages: Vec<serde_json::Value>) -> Self {
Self {
call: std::sync::atomic::AtomicU32::new(0),
pages,
}
}
}
impl Respond for Sequential {
fn respond(&self, _: &Request) -> ResponseTemplate {
let i = self.call.fetch_add(1, std::sync::atomic::Ordering::Relaxed) as usize;
let body = self.pages.get(i).unwrap_or_else(|| {
panic!(
"Sequential: request #{i} has no configured response \
(only {} page(s) supplied)",
self.pages.len()
)
});
ResponseTemplate::new(200).set_body_json(body)
}
}
fn make_activities_json(count: usize, id_prefix: &str) -> serde_json::Value {
serde_json::Value::Array(
(0..count)
.map(|i| {
serde_json::json!({
"id": format!("{id_prefix}-{i:05}"),
"order_id": "ord-1",
"symbol": "SPY",
"side": "buy",
"price": "100.00",
"qty": "1",
"transaction_time": "2025-04-18T14:30:00Z"
})
})
.collect(),
)
}
fn make_orders_json(count: usize) -> serde_json::Value {
serde_json::Value::Array(
(0..count)
.map(|i| {
serde_json::json!({
"id": format!("order-{i:05}"),
"client_order_id": null,
"symbol": "SPY",
"qty": "1",
"filled_qty": "0",
"side": "buy",
"type": "limit",
"time_in_force": "day",
"limit_price": "100.00",
"created_at": "2025-04-18T14:30:00Z"
})
})
.collect(),
)
}
#[tokio::test]
async fn paginate_activities_single_page_below_max_returns_all_not_truncated() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(
ResponseTemplate::new(200).set_body_json(make_activities_json(5, "act")),
)
.mount(&server)
.await;
let http = reqwest::Client::new();
let rl = RateLimitTracker::new();
let result = paginate_activities::<AlpacaActivity>(
&http,
&rl,
&server.uri(),
FILL_ACTIVITIES,
recovery_after(),
None,
)
.await
.unwrap();
assert_eq!(result.activities.len(), 5);
assert_eq!(result.resume, None);
assert_eq!(server.received_requests().await.unwrap().len(), 1);
}
#[tokio::test]
async fn paginate_activities_exactly_page_size_items_fetches_second_page() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(Sequential::new(vec![
make_activities_json(ALPACA_MAX_ACTIVITIES, "act"),
serde_json::json!([]), ]))
.mount(&server)
.await;
let http = reqwest::Client::new();
let rl = RateLimitTracker::new();
let result = paginate_activities::<AlpacaActivity>(
&http,
&rl,
&server.uri(),
FILL_ACTIVITIES,
recovery_after(),
None,
)
.await
.unwrap();
assert_eq!(result.activities.len(), ALPACA_MAX_ACTIVITIES);
assert_eq!(result.resume, None);
assert_eq!(
server.received_requests().await.unwrap().len(),
2,
"exactly 2 requests: first full page + second empty page"
);
}
#[tokio::test]
async fn paginate_activities_two_pages_returns_combined_activities_not_truncated() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(Sequential::new(vec![
make_activities_json(ALPACA_MAX_ACTIVITIES, "p1"),
make_activities_json(37, "p2"),
]))
.mount(&server)
.await;
let http = reqwest::Client::new();
let rl = RateLimitTracker::new();
let result = paginate_activities::<AlpacaActivity>(
&http,
&rl,
&server.uri(),
FILL_ACTIVITIES,
recovery_after(),
None,
)
.await
.unwrap();
assert_eq!(result.activities.len(), ALPACA_MAX_ACTIVITIES + 37);
assert_eq!(result.resume, None);
assert_eq!(server.received_requests().await.unwrap().len(), 2);
}
#[tokio::test]
async fn paginate_activities_at_max_pages_sets_truncated_true() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(make_activities_json(ALPACA_MAX_ACTIVITIES, "act")),
)
.mount(&server)
.await;
let http = reqwest::Client::new();
let rl = RateLimitTracker::new();
let result = paginate_activities::<AlpacaActivity>(
&http,
&rl,
&server.uri(),
FILL_ACTIVITIES,
recovery_after(),
None,
)
.await
.unwrap();
assert!(
result.resume.is_some(),
"must say where to read on after MAX_ACTIVITY_PAGES pages"
);
assert_eq!(
result.activities.len(),
MAX_ACTIVITY_PAGES * ALPACA_MAX_ACTIVITIES,
"must accumulate exactly MAX_ACTIVITY_PAGES * page_size activities"
);
assert_eq!(
server.received_requests().await.unwrap().len(),
MAX_ACTIVITY_PAGES,
"loop must issue exactly MAX_ACTIVITY_PAGES requests then stop"
);
}
#[tokio::test]
async fn fetch_raw_open_orders_499_results_returns_ok() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/orders"))
.respond_with(
ResponseTemplate::new(200).set_body_json(make_orders_json(MAX_OPEN_ORDERS - 1)),
)
.mount(&server)
.await;
let http = reqwest::Client::new();
let rl = RateLimitTracker::new();
let result = fetch_raw_open_orders(&http, &rl, &server.uri(), &[]).await;
assert!(
result.is_ok(),
"499 orders must not trigger truncation: {result:?}"
);
assert_eq!(result.unwrap().len(), MAX_OPEN_ORDERS - 1);
}
#[tokio::test]
async fn fetch_raw_open_orders_500_results_returns_truncated_snapshot_error() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/orders"))
.respond_with(
ResponseTemplate::new(200).set_body_json(make_orders_json(MAX_OPEN_ORDERS)),
)
.mount(&server)
.await;
let http = reqwest::Client::new();
let rl = RateLimitTracker::new();
let result = fetch_raw_open_orders(&http, &rl, &server.uri(), &[]).await;
assert!(
matches!(
result,
Err(UnindexedClientError::TruncatedSnapshot { limit }) if limit == MAX_OPEN_ORDERS
),
"500 orders must return TruncatedSnapshot, got: {result:?}"
);
}
async fn mount_account_and_positions(server: &MockServer) {
mount_account_and_positions_answering_after(server, Duration::ZERO).await;
}
async fn mount_account_and_positions_answering_after(server: &MockServer, delay: Duration) {
Mock::given(method("GET"))
.and(path("/v2/account"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!({
"status": "ACTIVE",
"currency": "USD",
"cash": "-1500.50",
"equity": "4000.00",
"buying_power": "5000.00",
"options_buying_power": "2000.00",
"non_marginable_buying_power": "1000.00",
}))
.set_delay(delay),
)
.mount(server)
.await;
Mock::given(method("GET"))
.and(path("/v2/positions"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!([
{
"symbol": "XYZ",
"asset_class": "us_equity",
"side": "short",
"qty": "-10",
"qty_available": "-10",
"avg_entry_price": "20.00",
"unrealized_pl": "15.50",
},
{
"symbol": "XYZ271217C00025000",
"asset_class": "us_option",
"side": "long",
"qty": "3",
"qty_available": "3",
"avg_entry_price": "0.85",
"unrealized_pl": "-45",
},
{
"symbol": "ABC",
"asset_class": "us_equity",
"side": "long",
"qty": "2.5",
"qty_available": "2.5",
"avg_entry_price": "40.00",
"unrealized_pl": "1.25",
},
{
"symbol": "BTCUSD",
"asset_class": "crypto",
"side": "long",
"qty": "0.75",
"qty_available": "0.5",
"avg_entry_price": "50000",
"unrealized_pl": "0",
},
]))
.set_delay(delay),
)
.mount(server)
.await;
Mock::given(method("GET"))
.and(path("/v2/orders"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([])))
.mount(server)
.await;
}
fn client_for(server: &MockServer) -> AlpacaClient {
AlpacaClient::new(AlpacaConfig::with_base_url(
"test-key".into(),
"test-secret".into(),
server.uri(),
))
}
#[tokio::test]
async fn account_snapshot_reports_cash_positions_and_crypto_balances() {
use crate::client::ExecutionClient;
use rust_decimal_macros::dec;
let server = MockServer::start().await;
mount_account_and_positions(&server).await;
let snapshot = client_for(&server)
.account_snapshot(&[], &[])
.await
.unwrap();
let balances: Vec<_> = snapshot
.balances
.iter()
.map(|b| (b.asset.name().as_str(), b.balance.total, b.balance.free))
.collect();
assert_eq!(
balances,
vec![
("USD", dec!(-1500.50), dec!(-1500.50)),
("BTC", dec!(0.75), dec!(0.5)),
],
"crypto stays a balance; equities and options do not appear as balances"
);
let positions: Vec<_> = snapshot
.instruments
.iter()
.map(|i| {
let p = i.position.open().unwrap();
(
i.instrument.name().as_str(),
p.quantity,
p.entry_price,
p.unrealized_pnl,
)
})
.collect();
assert_eq!(
positions,
vec![
("XYZ", dec!(-10), Some(dec!(20.00)), Some(dec!(15.50))),
(
"XYZ271217C00025000",
dec!(3),
Some(dec!(0.85)),
Some(dec!(-45))
),
("ABC", dec!(2.5), Some(dec!(40.00)), Some(dec!(1.25))),
],
"short equity, option and fractional positions; no crypto position"
);
}
#[tokio::test]
async fn balances_are_stamped_before_their_reads_were_sent() {
use crate::client::ExecutionClient;
const ANSWER_DELAY: Duration = Duration::from_millis(200);
let server = MockServer::start().await;
mount_account_and_positions_answering_after(&server, ANSWER_DELAY).await;
let client = client_for(&server);
let before = Utc::now();
let fetched = client.fetch_balances(&[]).await.unwrap();
let between = Utc::now();
let snapshot = client.account_snapshot(&[], &[]).await.unwrap();
let after = Utc::now();
assert_eq!((fetched.len(), snapshot.balances.len()), (2, 2));
assert_eq!(fetched[0].asset.name().as_str(), USD);
assert!(fetched[0].time_exchange + 2 * ANSWER_DELAY <= between);
for (balances, from, to) in [
(&fetched, before, between),
(&snapshot.balances, between, after),
] {
for balance in balances {
assert!(
from <= balance.time_exchange && balance.time_exchange + ANSWER_DELAY <= to,
"{balance:?} not within {from} ..= {to} - {ANSWER_DELAY:?}"
);
}
}
}
#[tokio::test]
async fn account_snapshot_usd_only_still_reports_positions() {
use crate::client::ExecutionClient;
use rust_decimal_macros::dec;
let server = MockServer::start().await;
mount_account_and_positions(&server).await;
let usd = [AssetNameExchange::new("USD")];
let instruments = [InstrumentNameExchange::new("XYZ")];
let snapshot = client_for(&server)
.account_snapshot(&usd, &instruments)
.await
.unwrap();
assert_eq!(snapshot.balances.len(), 1);
assert_eq!(snapshot.balances[0].asset, AssetNameExchange::new("USD"));
assert_eq!(snapshot.balances[0].balance.total, dec!(-1500.50));
assert_eq!(snapshot.instruments.len(), 1);
assert_eq!(snapshot.instruments[0].position.quantity(), Some(dec!(-10)));
}
#[tokio::test]
async fn account_snapshot_non_usd_only_skips_account_request() {
use crate::client::ExecutionClient;
let server = MockServer::start().await;
mount_account_and_positions(&server).await;
let btc = [AssetNameExchange::new("btc")];
let snapshot = client_for(&server)
.account_snapshot(&btc, &[])
.await
.unwrap();
assert_eq!(snapshot.balances.len(), 1);
assert_eq!(snapshot.balances[0].asset, AssetNameExchange::new("BTC"));
let requests = server.received_requests().await.unwrap();
assert!(
requests.iter().all(|r| r.url.path() != "/v2/account"),
"no USD requested, so /v2/account must not be fetched"
);
}
#[tokio::test]
async fn open_order_refuses_every_margin_effect_without_sending() {
use crate::client::ExecutionClient;
use crate::error::OrderError;
use crate::margin::MarginEffect;
use crate::order::request::{OrderRequestOpen, RequestOpen};
use crate::order::state::InactiveOrderState;
use crate::order::{
OrderKey, OrderKind, TimeInForce,
id::{ClientOrderId, StrategyId},
};
use rust_decimal::Decimal;
use rustrade_instrument::Side;
use rustrade_instrument::exchange::ExchangeId;
use rustrade_instrument::instrument::name::InstrumentNameExchange;
let server = MockServer::start().await;
let client = client_for(&server);
let instrument = InstrumentNameExchange::new("AAPL");
for effect in [
MarginEffect::NoSideEffect,
MarginEffect::Borrow,
MarginEffect::Repay,
] {
let order = client
.open_order(OrderRequestOpen {
key: OrderKey {
exchange: ExchangeId::AlpacaBroker,
instrument: &instrument,
strategy: StrategyId::new("test-strategy"),
cid: ClientOrderId::new("test-cid"),
},
state: RequestOpen::new(
Side::Buy,
None,
Decimal::new(10, 0),
OrderKind::Market,
TimeInForce::ImmediateOrCancel,
)
.with_margin_effect(effect),
})
.await;
assert!(
matches!(
order.state,
OrderState::Inactive(InactiveOrderState::OpenFailed(
OrderError::UnsupportedMarginEffect(refused)
)) if refused == effect
),
"{order:?}"
);
}
assert!(
server.received_requests().await.unwrap().is_empty(),
"a refused order is never sent"
);
}
#[tokio::test]
async fn open_order_reduce_only_sell_sends_sell_to_close_intent() {
use crate::client::ExecutionClient;
use crate::order::request::{OrderRequestOpen, RequestOpen};
use crate::order::{
OrderKey, OrderKind, TimeInForce,
id::{ClientOrderId, StrategyId},
};
use rust_decimal::Decimal;
use rustrade_instrument::Side;
use rustrade_instrument::exchange::ExchangeId;
use rustrade_instrument::instrument::name::InstrumentNameExchange;
use wiremock::matchers::{method, path};
let server = MockServer::start().await;
let captured_body = std::sync::Arc::new(parking_lot::Mutex::new(None));
let captured_clone = captured_body.clone();
Mock::given(method("POST"))
.and(path("/v2/orders"))
.respond_with(move |req: &Request| {
let body: serde_json::Value = serde_json::from_slice(&req.body).unwrap();
*captured_clone.lock() = Some(body);
ResponseTemplate::new(200).set_body_json(serde_json::json!({
"id": "test-order-id",
"client_order_id": "test-cid",
"symbol": "AAPL",
"qty": "10",
"filled_qty": "0",
"side": "sell",
"type": "market",
"time_in_force": "ioc",
"limit_price": null,
"created_at": "2025-04-18T14:30:00Z"
}))
})
.mount(&server)
.await;
let config =
AlpacaConfig::with_base_url("test-key".into(), "test-secret".into(), server.uri());
let client = AlpacaClient::new(config);
let request = OrderRequestOpen {
key: OrderKey {
exchange: ExchangeId::AlpacaBroker,
instrument: InstrumentNameExchange::new("AAPL"),
strategy: StrategyId::new("test-strategy"),
cid: ClientOrderId::new("test-cid"),
},
state: RequestOpen::new(
Side::Sell,
None,
Decimal::new(10, 0),
OrderKind::Market,
TimeInForce::ImmediateOrCancel,
)
.with_reduce_only(ReduceOnly::Position), };
let order = client
.open_order(OrderRequestOpen {
key: OrderKey {
exchange: request.key.exchange,
instrument: &request.key.instrument,
strategy: request.key.strategy.clone(),
cid: request.key.cid.clone(),
},
state: request.state.clone(),
})
.await;
assert!(
order.state.is_accepted(),
"order should be accepted: {:?}",
order.state
);
let body = captured_body
.lock()
.take()
.expect("request body should be captured");
assert_eq!(
body.get("position_intent").and_then(|v| v.as_str()),
Some("sell_to_close"),
"reduce_only Position + Side::Sell should produce position_intent=sell_to_close, got: {body}"
);
}
#[tokio::test]
async fn open_order_not_reduce_only_buy_sends_buy_to_open_intent() {
use crate::client::ExecutionClient;
use crate::order::request::{OrderRequestOpen, RequestOpen};
use crate::order::{
OrderKey, OrderKind, TimeInForce,
id::{ClientOrderId, StrategyId},
};
use rust_decimal::Decimal;
use rustrade_instrument::Side;
use rustrade_instrument::exchange::ExchangeId;
use rustrade_instrument::instrument::name::InstrumentNameExchange;
let server = MockServer::start().await;
let captured_body = std::sync::Arc::new(parking_lot::Mutex::new(None));
let captured_clone = captured_body.clone();
Mock::given(method("POST"))
.and(path("/v2/orders"))
.respond_with(move |req: &Request| {
let body: serde_json::Value = serde_json::from_slice(&req.body).unwrap();
*captured_clone.lock() = Some(body);
ResponseTemplate::new(200).set_body_json(serde_json::json!({
"id": "test-order-id",
"client_order_id": "test-cid",
"symbol": "AAPL",
"qty": "10",
"filled_qty": "0",
"side": "buy",
"type": "market",
"time_in_force": "ioc",
"limit_price": null,
"created_at": "2025-04-18T14:30:00Z"
}))
})
.mount(&server)
.await;
let config =
AlpacaConfig::with_base_url("test-key".into(), "test-secret".into(), server.uri());
let client = AlpacaClient::new(config);
let instrument = InstrumentNameExchange::new("AAPL");
let request = OrderRequestOpen {
key: OrderKey {
exchange: ExchangeId::AlpacaBroker,
instrument: &instrument,
strategy: StrategyId::new("test-strategy"),
cid: ClientOrderId::new("test-cid"),
},
state: RequestOpen::new(
Side::Buy,
None,
Decimal::new(10, 0),
OrderKind::Market,
TimeInForce::ImmediateOrCancel,
),
};
let order = client.open_order(request).await;
assert!(
order.state.is_accepted(),
"order should be accepted: {:?}",
order.state
);
let body = captured_body
.lock()
.take()
.expect("request body should be captured");
assert_eq!(
body.get("position_intent").and_then(|v| v.as_str()),
Some("buy_to_open"),
"reduce_only No + Side::Buy should produce position_intent=buy_to_open, got: {body}"
);
}
#[tokio::test]
async fn rest_with_retry_reports_an_undecodable_2xx_body_as_internal() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account"))
.respond_with(ResponseTemplate::new(200).set_body_string("not json"))
.mount(&server)
.await;
let http = reqwest::Client::new();
let rl = RateLimitTracker::new();
let url = format!("{}/v2/account", server.uri());
let result: Result<serde_json::Value, _> =
rest_with_retry(&rl, || http.get(&url)).await;
assert!(
matches!(result, Err(UnindexedClientError::Internal(_))),
"{result:?}"
);
}
async fn holds_back(rl: &RateLimitTracker) -> bool {
tokio::time::timeout(Duration::from_millis(50), rl.wait_if_blocked())
.await
.is_err()
}
fn with_rate_limit(
template: ResponseTemplate,
remaining: u32,
reset: bool,
) -> ResponseTemplate {
let template = template.insert_header("x-ratelimit-remaining", remaining.to_string());
if !reset {
return template;
}
let now_secs = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs();
template.insert_header("x-ratelimit-reset", (now_secs + 30).to_string())
}
#[tokio::test]
async fn rest_with_retry_pauses_once_no_requests_remain() {
for (remaining, reset, pauses) in [(0, true, true), (1, true, false), (0, false, false)]
{
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account"))
.respond_with(with_rate_limit(
ResponseTemplate::new(200).set_body_json(serde_json::json!({})),
remaining,
reset,
))
.mount(&server)
.await;
let http = reqwest::Client::new();
let rl = RateLimitTracker::new();
let url = format!("{}/v2/account", server.uri());
let _: serde_json::Value = rest_with_retry(&rl, || http.get(&url)).await.unwrap();
assert_eq!(
holds_back(&rl).await,
pauses,
"remaining {remaining}, reset {reset}"
);
}
}
#[tokio::test]
async fn a_pause_lasts_a_minute_at_most() {
tokio::time::pause();
let now_secs = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs();
let mut headers = reqwest::header::HeaderMap::new();
headers.insert("x-ratelimit-remaining", "0".parse().unwrap());
headers.insert(
"x-ratelimit-reset",
(now_secs + 3_600).to_string().parse().unwrap(),
);
let rl = RateLimitTracker::new();
observe_rate_limit_remaining(&rl, &headers);
assert!(holds_back(&rl).await);
tokio::time::advance(Duration::from_secs(DEFAULT_RATE_LIMIT_DELAY_SECS)).await;
assert!(!holds_back(&rl).await);
}
#[tokio::test]
async fn rest_delete_with_retry_pauses_once_no_requests_remain() {
let server = MockServer::start().await;
Mock::given(method("DELETE"))
.and(path("/v2/orders/abc"))
.respond_with(with_rate_limit(ResponseTemplate::new(204), 0, true))
.mount(&server)
.await;
let http = reqwest::Client::new();
let rl = RateLimitTracker::new();
let url = format!("{}/v2/orders/abc", server.uri());
rest_delete_with_retry(&rl, || http.delete(&url))
.await
.unwrap();
assert!(holds_back(&rl).await);
}
#[tokio::test]
async fn open_order_with_an_undecodable_2xx_response_is_status_unknown() {
use crate::client::ExecutionClient;
use crate::error::OrderError;
use crate::order::request::{OrderRequestOpen, RequestOpen};
use crate::order::state::{InactiveOrderState, OrderState};
use crate::order::{
OrderKey, OrderKind, TimeInForce,
id::{ClientOrderId, StrategyId},
};
use rustrade_instrument::Side;
use rustrade_instrument::exchange::ExchangeId;
use rustrade_instrument::instrument::name::InstrumentNameExchange;
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v2/orders"))
.respond_with(
ResponseTemplate::new(200).set_body_json(serde_json::json!({ "id": 1 })),
)
.mount(&server)
.await;
let instrument = InstrumentNameExchange::new("AAPL");
let order = client_for(&server)
.open_order(OrderRequestOpen {
key: OrderKey {
exchange: ExchangeId::AlpacaBroker,
instrument: &instrument,
strategy: StrategyId::new("test-strategy"),
cid: ClientOrderId::new("test-cid"),
},
state: RequestOpen::new(
Side::Buy,
None,
Decimal::new(10, 0),
OrderKind::Market,
TimeInForce::ImmediateOrCancel,
),
})
.await;
assert!(
matches!(
order.state,
OrderState::Inactive(InactiveOrderState::OpenFailed(OrderError::Connectivity(
_
)))
),
"{:?}",
order.state
);
}
#[tokio::test]
async fn open_order_reports_a_403_by_its_code() {
use crate::client::ExecutionClient;
use crate::error::OrderError;
use crate::order::request::{OrderRequestOpen, RequestOpen};
use crate::order::state::{InactiveOrderState, OrderState};
use crate::order::{
OrderKey, OrderKind, TimeInForce,
id::{ClientOrderId, StrategyId},
};
use rustrade_instrument::Side;
use rustrade_instrument::exchange::ExchangeId;
use rustrade_instrument::instrument::name::InstrumentNameExchange;
async fn open_against(
status: u16,
body: serde_json::Value,
) -> OrderState<AssetNameExchange, InstrumentNameExchange> {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v2/orders"))
.respond_with(ResponseTemplate::new(status).set_body_json(body))
.mount(&server)
.await;
let instrument = InstrumentNameExchange::new("AAPL");
client_for(&server)
.open_order(OrderRequestOpen {
key: OrderKey {
exchange: ExchangeId::AlpacaBroker,
instrument: &instrument,
strategy: StrategyId::new("test-strategy"),
cid: ClientOrderId::new("test-cid"),
},
state: RequestOpen::new(
Side::Sell,
None,
Decimal::new(10, 0),
OrderKind::Market,
TimeInForce::ImmediateOrCancel,
),
})
.await
.state
}
let rejected =
|state: &OrderState<AssetNameExchange, InstrumentNameExchange>| match state {
OrderState::Inactive(InactiveOrderState::OpenFailed(OrderError::Rejected(
e,
))) => e.clone(),
other => panic!("expected a rejection, got {other:?}"),
};
let state = open_against(
403,
serde_json::json!({
"code": 40310000,
"message": "insufficient buying power"
}),
)
.await;
assert_eq!(
rejected(&state),
ApiError::BalanceInsufficient(None, "40310000 insufficient buying power".into())
);
let state = open_against(
403,
serde_json::json!({
"code": 40310000,
"message": "account is not allowed to short"
}),
)
.await;
assert_eq!(
rejected(&state),
ApiError::BorrowRejected(BorrowReject::new(
BorrowRejectReason::BorrowDisabled,
Some("40310000".into()),
"account is not allowed to short".into(),
))
);
let state = open_against(
422,
serde_json::json!({
"code": 42210000,
"message": "asset XYZ cannot be sold short"
}),
)
.await;
assert_eq!(
rejected(&state),
ApiError::BorrowRejected(BorrowReject::new(
BorrowRejectReason::NotShortable,
Some("42210000".into()),
"asset XYZ cannot be sold short".into(),
))
);
let state = open_against(403, serde_json::json!({ "message": "forbidden." })).await;
assert!(
matches!(rejected(&state), ApiError::Unauthenticated(_)),
"{state:?}"
);
}
#[tokio::test]
async fn open_bracket_order_with_an_undecodable_2xx_response_is_status_unknown() {
use crate::error::OrderError;
use crate::order::state::{InactiveOrderState, OrderState};
use rustrade_instrument::instrument::name::InstrumentNameExchange;
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v2/orders"))
.respond_with(
ResponseTemplate::new(200).set_body_json(serde_json::json!({ "id": 1 })),
)
.mount(&server)
.await;
let result = client_for(&server)
.open_bracket_order(AlpacaBracketOrderRequest::new(
InstrumentNameExchange::new("SPY"),
crate::order::id::StrategyId::new("test"),
crate::order::id::ClientOrderId::new("test-bracket"),
Side::Buy,
Decimal::ONE,
Decimal::new(100, 0),
Decimal::new(120, 0),
Decimal::new(90, 0),
TimeInForce::GoodUntilCancelled { post_only: false },
))
.await;
assert!(
matches!(
result.parent.state,
OrderState::Inactive(InactiveOrderState::OpenFailed(OrderError::Connectivity(
_
)))
),
"{:?}",
result.parent.state
);
}
fn recovery_after() -> DateTime<Utc> {
"2025-01-01T00:00:00Z".parse().expect("valid time")
}
fn activity_json(
id: &str,
order_id: &str,
qty: &str,
cum_qty: Option<&str>,
) -> serde_json::Value {
let mut v = serde_json::json!({
"id": id,
"order_id": order_id,
"symbol": "SPY",
"side": "buy",
"price": "100.00",
"qty": qty,
"transaction_time": "2025-04-18T14:30:00Z"
});
if let Some(c) = cum_qty {
v["cum_qty"] = serde_json::Value::String(c.to_string());
}
v
}
#[tokio::test(start_paused = true)]
async fn the_reconnect_budget_resets_only_once_a_connection_proves_stable() {
use futures::{SinkExt as _, StreamExt as _};
use tokio_tungstenite::tungstenite::Message;
const STABLE: Duration = Duration::from_secs(HEARTBEAT_TIMEOUT_SECS + 3);
const DROPPED: Duration = Duration::from_millis(100);
let lifetime = |n: usize| match n {
0 | 1 => Some(DROPPED),
2 => Some(STABLE),
_ => None,
};
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let ws_url = format!("ws://{}", listener.local_addr().unwrap());
let listens = Arc::new(parking_lot::Mutex::new(Vec::new()));
let seen = Arc::clone(&listens);
tokio::spawn(async move {
let mut connections = 0..;
while let Ok((stream, _)) = listener.accept().await {
stream.set_nodelay(true).unwrap();
let seen = Arc::clone(&seen);
let lifetime = connections.next().and_then(lifetime);
tokio::spawn(async move {
let Ok(mut ws) = tokio_tungstenite::accept_async(stream).await else {
return;
};
for answer in [
r#"{"stream":"authorization","data":{"status":"authorized","action":"authenticate"}}"#,
r#"{"stream":"listening","data":{"streams":["trade_updates"]}}"#,
] {
let Some(Ok(_)) = ws.next().await else {
return;
};
if ws.send(Message::text(answer)).await.is_err() {
return;
}
}
seen.lock().push(tokio::time::Instant::now());
let closes = async {
match lifetime {
Some(after) => tokio::time::sleep(after).await,
None => std::future::pending().await,
}
};
tokio::pin!(closes);
let mut pings = tokio::time::interval(Duration::from_secs(1));
loop {
tokio::select! {
() = &mut closes => {
let _ = ws.close(None).await;
return;
}
_ = pings.tick() => {
if ws.send(Message::Ping(Default::default())).await.is_err() {
return;
}
}
message = ws.next() => {
if !matches!(message, Some(Ok(_))) {
return;
}
}
}
}
});
}
});
let rest_url = crate::client::test_http::serve(|_, _| (404, String::new())).await;
let mut config = AlpacaConfig::with_base_url("key".into(), "secret".into(), rest_url);
config.ws_url_override = Some(ws_url);
let (tx, _rx) = mpsc::unbounded_channel();
tokio::spawn(connection_manager(
tx,
new_dedup_cache(),
Arc::new(config),
reqwest::Client::new(),
Arc::new(RateLimitTracker::new()),
KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
Vec::new(),
None,
));
let deadline = tokio::time::Instant::now() + STABLE + Duration::from_secs(15);
while listens.lock().len() < 4 {
assert!(
tokio::time::Instant::now() < deadline,
"{} connections",
listens.lock().len()
);
tokio::time::sleep(Duration::from_millis(50)).await;
}
let initial = BackoffLimits::ACCOUNT_STREAM.initial;
let waits: Vec<_> = listens.lock().windows(2).map(|w| w[1] - w[0]).collect();
assert_eq!(
waits,
[DROPPED + initial, DROPPED + 2 * initial, STABLE + initial],
"the waits between listens"
);
}
#[tokio::test(start_paused = true)]
async fn failed_connects_end_the_stream_once_the_budget_is_spent() {
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
let ws_url = format!("ws://{}", listener.local_addr().unwrap());
drop(listener);
let mut config = AlpacaConfig::paper("key".into(), "secret".into());
config.ws_url_override = Some(ws_url);
let (tx, mut rx) = mpsc::unbounded_channel();
let started = tokio::time::Instant::now();
connection_manager(
tx,
new_dedup_cache(),
Arc::new(config),
reqwest::Client::new(),
Arc::new(RateLimitTracker::new()),
KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
Vec::new(),
None,
)
.await;
let Ok(event) = rx.try_recv() else {
panic!("the stream ended without reporting why");
};
let AccountEventKind::StreamTerminated(
StreamTerminationReason::ReconnectBudgetExhausted {
attempts,
last_error,
},
) = event.kind
else {
panic!("unexpected event: {event:?}");
};
assert_eq!(attempts, BackoffLimits::ACCOUNT_STREAM.max_attempts);
assert!(
!last_error.is_empty(),
"the last connect failure is reported"
);
assert!(
started.elapsed() >= Duration::from_secs(181),
"{:?}",
started.elapsed()
);
assert!(rx.try_recv().is_err(), "nothing follows the termination");
}
#[tokio::test]
async fn a_reconnect_reads_missed_fills_from_alpacas_time_when_the_host_is_ahead() {
use futures::{SinkExt as _, StreamExt as _};
use tokio_tungstenite::tungstenite::Message;
let stamped = Utc::now() - TimeDelta::seconds(120);
let update = format!(
r#"{{"stream":"trade_updates","data":{{"event":"new","timestamp":"{}","order":{{"id":"id-a","client_order_id":"a","symbol":"SPY","qty":"2","filled_qty":"0","side":"buy","type":"limit","time_in_force":"gtc","limit_price":"100","status":"new"}}}}}}"#,
stamped.to_rfc3339_opts(SecondsFormat::Millis, true)
);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let ws_url = format!("ws://{}", listener.local_addr().unwrap());
tokio::spawn(async move {
let mut connections = 0..;
while let Ok((stream, _)) = listener.accept().await {
let first = connections.next() == Some(0);
let update = update.clone();
tokio::spawn(async move {
let Ok(mut ws) = tokio_tungstenite::accept_async(stream).await else {
return;
};
for answer in [
r#"{"stream":"authorization","data":{"status":"authorized","action":"authenticate"}}"#,
r#"{"stream":"listening","data":{"streams":["trade_updates"]}}"#,
] {
let Some(Ok(_)) = ws.next().await else {
return;
};
if ws.send(Message::text(answer)).await.is_err() {
return;
}
}
if first {
let _ = ws.send(Message::text(update)).await;
let _ = ws.close(None).await;
}
while let Some(Ok(_)) = ws.next().await {}
});
}
});
let rest = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([])))
.mount(&rest)
.await;
let mut config = AlpacaConfig::with_base_url("key".into(), "secret".into(), rest.uri());
config.ws_url_override = Some(ws_url);
let (tx, _rx) = mpsc::unbounded_channel();
let task = tokio::spawn(connection_manager(
tx,
new_dedup_cache(),
Arc::new(config),
reqwest::Client::new(),
Arc::new(RateLimitTracker::new()),
KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
Vec::new(),
None,
));
let after = tokio::time::timeout(Duration::from_secs(10), async {
loop {
let requests = rest.received_requests().await.unwrap_or_default();
if let Some(after) = requests
.iter()
.filter(|request| request.url.path() == "/v2/account/activities")
.find_map(|request| {
request
.url
.query_pairs()
.find(|(key, _)| key == "after")
.map(|(_, after)| after.into_owned())
})
{
break after;
}
tokio::time::sleep(Duration::from_millis(20)).await;
}
})
.await
.expect("the reconnect reads the missed fills");
let after = DateTime::parse_from_rfc3339(&after)
.unwrap()
.with_timezone(&Utc);
let lookback = TimeDelta::milliseconds(SIGNAL_RECOVERY_LOOKBACK_MS);
assert!(
after <= stamped + TimeDelta::seconds(1) - lookback,
"{after}"
);
assert!(
after >= stamped - TimeDelta::seconds(1) - lookback,
"{after}"
);
task.abort();
}
async fn drive_recover_fills_with(
activities: Vec<serde_json::Value>,
dedup: SharedDedupCache,
known: &SharedKnownLiveOrders,
) -> Vec<UnindexedAccountEvent> {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(
ResponseTemplate::new(200).set_body_json(serde_json::Value::Array(activities)),
)
.mount(&server)
.await;
let http = reqwest::Client::new();
let rl = RateLimitTracker::new();
let (tx, mut rx) = mpsc::unbounded_channel();
let outcome = recover_fills(
&http,
&rl,
&[],
&server.uri(),
recovery_after(),
&tx,
&dedup,
known,
)
.await;
assert_eq!(outcome, Ok(()), "a full read leaves nothing unrecovered");
drop(tx);
let mut out = Vec::new();
while let Ok(event) = rx.try_recv() {
out.push(event);
}
out
}
async fn drive_recover_fills(
activities: Vec<serde_json::Value>,
) -> Vec<UnindexedAccountEvent> {
drive_recover_fills_with(
activities,
new_dedup_cache(),
&KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
)
.await
}
async fn recover_fills_from(
server: &MockServer,
) -> (Result<(), UnreadFills>, Vec<UnindexedAccountEvent>) {
let (tx, mut rx) = mpsc::unbounded_channel();
let outcome = recover_fills(
&reqwest::Client::new(),
&RateLimitTracker::new(),
&[],
&server.uri(),
recovery_after(),
&tx,
&new_dedup_cache(),
&KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
)
.await;
drop(tx);
(outcome, std::iter::from_fn(|| rx.try_recv().ok()).collect())
}
#[tokio::test]
async fn a_failed_recovery_read_leaves_every_fill_since_the_disconnect_unread() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(ResponseTemplate::new(500).set_body_string("unavailable"))
.mount(&server)
.await;
let (outcome, events) = recover_fills_from(&server).await;
let Err(UnreadFills {
start,
reason: FillRecoveryFailure::Request(_),
}) = outcome
else {
panic!("expected the request's failure, got {outcome:?}");
};
assert_eq!(start, recovery_after(), "from the disconnect");
assert!(events.is_empty(), "nothing forwarded: {events:?}");
}
#[tokio::test]
async fn a_truncated_recovery_read_delivers_what_it_read_and_leaves_the_rest_unread() {
let time = |s: &str| s.parse::<DateTime<Utc>>().expect("valid time");
let cases = [
(
Some("2025-04-18T14:30:01.234567Z"),
time("2025-04-18T14:30:01.234Z"),
),
(Some("not a time"), time("2025-04-18T14:30:00Z")),
(None, recovery_after()),
];
for (last_time, expected_start) in cases {
let mut page = make_activities_json(ALPACA_MAX_ACTIVITIES, "act");
let Some(activities) = page.as_array_mut() else {
panic!("an array of activities");
};
match last_time {
Some(last_time) => {
if let Some(last) = activities.last_mut() {
last["transaction_time"] =
serde_json::Value::String(last_time.to_string());
}
}
None => {
for activity in activities.iter_mut() {
activity["transaction_time"] =
serde_json::Value::String("not a time".to_string());
}
}
}
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(ResponseTemplate::new(200).set_body_json(page))
.mount(&server)
.await;
let (outcome, events) = recover_fills_from(&server).await;
let fills_read = MAX_ACTIVITY_PAGES * ALPACA_MAX_ACTIVITIES;
assert_eq!(
outcome,
Err(UnreadFills {
start: expected_start,
reason: FillRecoveryFailure::Truncated { fills_read },
}),
"last time {last_time:?}"
);
assert_eq!(events.len(), fills_read, "every fill read is forwarded");
}
}
#[test]
fn a_fill_recovery_give_up_covers_every_instrument_the_stream_does() {
let start = recovery_after();
let end = start + chrono::Duration::minutes(5);
let spy = InstrumentNameExchange::new("SPY");
let qqq = InstrumentNameExchange::new("QQQ");
let cases = [
(vec![], FillRecoveryScope::AllInstruments),
(
vec![spy.clone(), qqq.clone()],
FillRecoveryScope::Instruments(vec![spy, qqq]),
),
];
for (instruments, scope) in cases {
let reason = FillRecoveryFailure::TimedOut { timeout_secs: 30 };
let unread = UnreadFills {
start,
reason: reason.clone(),
};
assert_eq!(
fill_recovery_gave_up(&instruments, unread, end),
UnindexedAccountEvent::new(
ExchangeId::AlpacaBroker,
AccountEventKind::FillRecoveryGaveUp(FillRecoveryGap::new(
scope, start, end, 1, reason,
)),
)
);
}
}
#[test]
fn a_read_that_reached_past_the_recovery_start_still_reports() {
let end = recovery_after();
let start = end + chrono::Duration::seconds(1);
let unread = UnreadFills {
start,
reason: FillRecoveryFailure::Truncated { fills_read: 5_000 },
};
let event = fill_recovery_gave_up(&[], unread, end);
let AccountEventKind::FillRecoveryGaveUp(gave_up) = event.kind else {
panic!("expected FillRecoveryGaveUp, got {event:?}");
};
assert_eq!((gave_up.start, gave_up.end), (start, start));
}
async fn recover_fills_or_report_from(
server: &MockServer,
instruments: &[InstrumentNameExchange],
timeout: Duration,
) -> Vec<UnindexedAccountEvent> {
let (tx, mut rx) = mpsc::unbounded_channel();
recover_fills_or_report(
&reqwest::Client::new(),
&RateLimitTracker::new(),
instruments,
&server.uri(),
recovery_after(),
Utc::now(),
timeout,
&tx,
&new_dedup_cache(),
&KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
)
.await;
drop(tx);
std::iter::from_fn(|| rx.try_recv().ok()).collect()
}
#[tokio::test]
async fn a_timed_out_recovery_reports_every_fill_since_the_disconnect() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::json!([]))
.set_delay(Duration::from_secs(3_600)),
)
.mount(&server)
.await;
let spy = InstrumentNameExchange::new("SPY");
let events = recover_fills_or_report_from(
&server,
std::slice::from_ref(&spy),
Duration::from_millis(50),
)
.await;
let [event] = events.as_slice() else {
panic!("only the give-up is sent: {events:?}");
};
let AccountEventKind::FillRecoveryGaveUp(gave_up) = &event.kind else {
panic!("expected FillRecoveryGaveUp, got {event:?}");
};
assert_eq!(gave_up.scope, FillRecoveryScope::Instruments(vec![spy]));
assert_eq!(gave_up.start, recovery_after(), "from the disconnect");
assert!(gave_up.end >= gave_up.start);
assert_eq!(gave_up.attempts, 1);
assert!(
matches!(gave_up.reason, FillRecoveryFailure::TimedOut { .. }),
"{:?}",
gave_up.reason
);
}
#[tokio::test]
async fn a_truncated_recovery_reports_after_the_fills_it_read() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(make_activities_json(ALPACA_MAX_ACTIVITIES, "act")),
)
.mount(&server)
.await;
let events = recover_fills_or_report_from(&server, &[], Duration::from_secs(30)).await;
let Some((last, fills)) = events.split_last() else {
panic!("events are sent");
};
assert_eq!(fills.len(), MAX_ACTIVITY_PAGES * ALPACA_MAX_ACTIVITIES);
assert!(
fills
.iter()
.all(|event| matches!(event.kind, AccountEventKind::Trade(_))),
"every fill comes first"
);
let AccountEventKind::FillRecoveryGaveUp(gave_up) = &last.kind else {
panic!("the give-up comes last, got {last:?}");
};
assert_eq!(gave_up.scope, FillRecoveryScope::AllInstruments);
assert!(matches!(
gave_up.reason,
FillRecoveryFailure::Truncated { .. }
));
}
#[test]
fn an_activity_id_names_its_execution_after_the_separator() {
assert_eq!(
activity_execution_id("20261005041849348::524b1902-817e-446c-825b-a9fcfebbc17e"),
"524b1902-817e-446c-825b-a9fcfebbc17e"
);
assert_eq!(activity_execution_id("act-1"), "act-1");
assert_eq!(
activity_execution_id("20261005041849348::"),
"20261005041849348::"
);
}
#[tokio::test]
async fn a_fill_has_one_trade_id_on_the_stream_in_recovery_and_from_fetch_trades() {
use crate::client::ExecutionClient;
let execution_id = "524b1902-817e-446c-825b-a9fcfebbc17e";
let activity = activity_json(
&format!("20250418103000000::{execution_id}"),
"ord-1",
"2",
Some("2"),
);
let [streamed, _snapshot] = convert_trade_update(AlpacaTradeUpdate {
event: SmolStr::new("fill"),
execution_id: Some(execution_id),
previous_execution_id: None,
order: super::make_order_ws("ord-1", "SPY", "buy", "2"),
price: Some("100.00"),
qty: Some("2"),
timestamp: Some("2025-04-18T14:30:00Z"),
});
let Some(streamed) = streamed else {
panic!("a fill converts to a Trade");
};
let recovered = drive_recover_fills(vec![activity.clone()]).await;
let [recovered] = recovered.as_slice() else {
panic!("one fill recovered: {recovered:?}");
};
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(serde_json::Value::Array(vec![activity])),
)
.mount(&server)
.await;
let Ok(fetched) = client_for(&server)
.fetch_trades(recovery_after(), Utc::now(), &[])
.await
else {
panic!("fetch_trades reads the activity");
};
let [fetched] = fetched.trades.as_slice() else {
panic!("one fill fetched: {fetched:?}");
};
assert_eq!(trade_of(&streamed).id.0.as_str(), execution_id);
assert_eq!(trade_of(recovered).id.0.as_str(), execution_id);
assert_eq!(fetched.id.0.as_str(), execution_id);
}
fn time(s: &str) -> DateTime<Utc> {
s.parse().expect("valid time")
}
struct ActivitiesVenue {
activities: Vec<serde_json::Value>,
}
impl Respond for ActivitiesVenue {
fn respond(&self, request: &Request) -> ResponseTemplate {
let query: std::collections::HashMap<String, String> =
request.url.query_pairs().into_owned().collect();
let bound = |name: &str| query.get(name).map(|s| time(s));
let (after, until) = (bound("after"), bound("until"));
let size: usize = query
.get("page_size")
.and_then(|s| s.parse().ok())
.unwrap_or(ALPACA_MAX_ACTIVITIES);
let in_bounds = self.activities.iter().filter(|activity| {
let at =
floor_millis(time(activity["transaction_time"].as_str().expect("a time")));
after.is_none_or(|after| at >= after) && until.is_none_or(|until| at < until)
});
let page: Vec<_> = match query.get("page_token") {
Some(token) => in_bounds
.skip_while(|activity| activity["id"] != token.as_str())
.skip(1)
.take(size)
.cloned()
.collect(),
None => in_bounds.take(size).cloned().collect(),
};
ResponseTemplate::new(200).set_body_json(page)
}
}
fn fill_at(id: &str, symbol: &str, transaction_time: DateTime<Utc>) -> serde_json::Value {
serde_json::json!({
"id": id,
"order_id": format!("ord-{id}"),
"symbol": symbol,
"side": "buy",
"price": "100.00",
"qty": "1",
"cum_qty": "1",
"transaction_time": transaction_time.to_rfc3339_opts(SecondsFormat::Micros, true),
})
}
async fn serve_activities(activities: Vec<serde_json::Value>) -> MockServer {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(ActivitiesVenue { activities })
.mount(&server)
.await;
server
}
fn non_trade(
id: &str,
activity_type: &str,
sub_type: Option<&str>,
net_amount: &str,
created_at: &str,
) -> serde_json::Value {
let mut activity = serde_json::json!({
"id": id, "activity_type": activity_type, "net_amount": net_amount,
"created_at": created_at, "date": &created_at[..10], "currency": "USD",
"status": "executed",
});
if let Some(sub_type) = sub_type {
activity["activity_sub_type"] = serde_json::json!(sub_type);
}
activity
}
async fn query_pairs(server: &MockServer) -> Vec<Vec<(String, String)>> {
let Some(requests) = server.received_requests().await else {
panic!("requests are recorded");
};
requests
.iter()
.map(|request| request.url.query_pairs().into_owned().collect())
.collect()
}
#[tokio::test]
async fn fetch_cash_flows_maps_each_activity_type_and_leaves_out_what_it_cannot_use() {
let at = "2025-04-18T05:00:00Z";
let mut fee = non_trade("fee", "FEE", Some("TAF"), "-0.02", at);
fee["symbol"] = serde_json::json!("SPY");
let mut canceled = non_trade("canceled", "FEE", Some("REG"), "-0.01", at);
canceled["status"] = serde_json::json!("canceled");
let mut no_amount = non_trade("no-amount", "FEE", None, "0", at);
no_amount.as_object_mut().unwrap().remove("net_amount");
let mut no_created_at = non_trade("no-time", "PTC", None, "-0.50", at);
no_created_at.as_object_mut().unwrap().remove("created_at");
let mut bad_time = non_trade("bad-time", "PTC", None, "-0.50", at);
bad_time["created_at"] = serde_json::json!("not-a-time");
let mut corrected = non_trade("correct", "PTC", None, "-0.40", at);
corrected["status"] = serde_json::json!("correct");
let mut euro = non_trade("euro", "PTR", None, "0.30", at);
euro["currency"] = serde_json::json!("EUR");
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([
non_trade("mgn", "INT", Some("MGN"), "-12.50", at),
non_trade("int-paid", "INT", None, "-3.00", at),
non_trade("int-credit", "INT", Some("CDT"), "1.00", at),
fee,
canceled,
no_amount,
no_created_at,
bad_time,
corrected,
euro,
non_trade("late", "FEE", None, "-0.01", "2025-04-20T05:00:00Z"),
])))
.mount(&server)
.await;
let client = client_for(&server);
let read = client
.fetch_cash_flows(time("2025-04-18T00:00:00Z"), time("2025-04-19T00:00:00Z"))
.await
.unwrap();
assert_eq!(read.resume, None);
let kinds: Vec<_> = read
.flows
.iter()
.map(|flow| (flow.id.clone().unwrap().0.to_string(), flow.kind.clone()))
.collect();
let interest = CashFlowKind::MarginInterest {
rate: None,
principal: None,
timing: None,
};
assert_eq!(
kinds,
[
("mgn", interest.clone()),
("int-paid", interest),
("int-credit", CashFlowKind::Other("INT.CDT".into())),
("fee", CashFlowKind::Other("FEE.TAF".into())),
("correct", CashFlowKind::Other("PTC".into())),
("euro", CashFlowKind::Rebate),
]
.map(|(id, kind)| (id.to_owned(), kind))
);
let by_id = |id: &str| {
read.flows
.iter()
.find(|flow| flow.id == Some(CashFlowId::new(id)))
.unwrap()
};
assert_eq!(by_id("mgn").amount, Decimal::new(-1250, 2));
assert_eq!(by_id("mgn").asset, AssetNameExchange::new("USD"));
assert_eq!(by_id("mgn").instrument, None);
assert_eq!(by_id("mgn").time_exchange, time(at));
assert_eq!(
by_id("fee").instrument,
Some(InstrumentNameExchange::new("SPY"))
);
assert_eq!(
by_id("correct").amount,
Decimal::new(-40, 2),
"a correction is returned as it is"
);
assert_eq!(by_id("euro").asset, AssetNameExchange::new("EUR"));
let queries = query_pairs(&server).await;
let [query] = queries.as_slice() else {
panic!("one request: {queries:?}");
};
assert!(
query.contains(&("activity_types".into(), "INT,FEE,PTC,PTR".into())),
"{query:?}"
);
}
#[tokio::test]
async fn fetch_trades_asks_for_fills_by_the_documented_parameter() {
use crate::client::ExecutionClient;
let server = serve_activities(Vec::new()).await;
let client = client_for(&server);
let read = client
.fetch_trades(
time("2025-04-18T00:00:00Z"),
time("2025-04-19T00:00:00Z"),
&[],
)
.await
.unwrap();
assert!(read.trades.is_empty());
let queries = query_pairs(&server).await;
let [query] = queries.as_slice() else {
panic!("one request: {queries:?}");
};
assert!(
query.contains(&("activity_types".into(), "FILL".into())),
"{query:?}"
);
assert!(
!query.iter().any(|(name, _)| name == "activity_type"),
"{query:?}"
);
}
#[tokio::test]
async fn fetch_cash_flows_reads_nothing_for_an_empty_span() {
let server = MockServer::start().await;
let client = client_for(&server);
let read = client
.fetch_cash_flows(time("2025-04-19T00:00:00Z"), time("2025-04-18T00:00:00Z"))
.await
.unwrap();
assert!(read.flows.is_empty() && read.resume.is_none());
assert!(query_pairs(&server).await.is_empty());
}
#[tokio::test]
async fn fetch_cash_flows_that_stops_at_its_bound_says_where_to_read_on() {
let base = time("2025-04-18T05:00:00Z");
let activities: Vec<_> = (0..6_000_i64)
.map(|n| {
let created =
(base + TimeDelta::seconds(n)).to_rfc3339_opts(SecondsFormat::Micros, true);
non_trade(&format!("{n:017}::fee"), "FEE", None, "-0.01", &created)
})
.collect();
struct Paged(Vec<serde_json::Value>);
impl Respond for Paged {
fn respond(&self, request: &Request) -> ResponseTemplate {
let token = request
.url
.query_pairs()
.find(|(name, _)| name == "page_token")
.map(|(_, value)| value.into_owned());
let page: Vec<_> = self
.0
.iter()
.skip_while(|activity| {
token
.as_deref()
.is_some_and(|token| activity["id"] != token)
})
.skip(usize::from(token.is_some()))
.take(ALPACA_MAX_ACTIVITIES)
.cloned()
.collect();
ResponseTemplate::new(200).set_body_json(page)
}
}
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(Paged(activities))
.mount(&server)
.await;
let client = client_for(&server);
let read = client
.fetch_cash_flows(base, base + TimeDelta::days(1))
.await
.unwrap();
assert_eq!(read.flows.len(), 5_000);
assert_eq!(read.resume, Some(base + TimeDelta::seconds(4_999)));
}
#[tokio::test]
async fn fetch_cash_flows_that_cannot_advance_fails() {
let at = "2025-04-18T05:00:00Z";
let full: Vec<_> = (0..ALPACA_MAX_ACTIVITIES)
.map(|n| non_trade(&format!("fee-{n}"), "FEE", None, "-0.01", at))
.collect();
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(ResponseTemplate::new(200).set_body_json(full))
.mount(&server)
.await;
let client = client_for(&server);
let result = client
.fetch_cash_flows(time(at), time("2025-04-19T00:00:00Z"))
.await;
assert!(
matches!(result, Err(UnindexedClientError::Internal(_))),
"{result:?}"
);
}
fn option_activity(
id: &str,
activity_type: &str,
symbol: &str,
qty: &str,
date: &str,
) -> serde_json::Value {
serde_json::json!({
"id": id, "activity_type": activity_type, "symbol": symbol, "qty": qty,
"date": date, "net_amount": "0", "status": "executed",
})
}
fn share_leg(
id: &str,
symbol: &str,
qty: &str,
price: &str,
date: &str,
) -> serde_json::Value {
let mut activity = option_activity(id, "OPTRD", symbol, qty, date);
activity["price"] = serde_json::json!(price);
activity
}
async fn serve_json(activities: serde_json::Value) -> MockServer {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(ResponseTemplate::new(200).set_body_json(activities))
.mount(&server)
.await;
server
}
#[test]
fn occ_contract_reads_the_root_kind_and_strike() {
assert_eq!(
occ_contract("QS240301C00006500"),
Some(("QS", true, Decimal::new(65, 1)))
);
assert_eq!(
occ_contract("AAPL240119P00150000"),
Some(("AAPL", false, Decimal::from(150)))
);
for not_occ in [
"QS",
"240301C00006500",
"QS240301X00006500",
"QS2403O1C00006500",
] {
assert_eq!(occ_contract(not_occ), None, "{not_occ}");
}
}
#[tokio::test]
async fn fetch_option_events_maps_each_activity_and_pairs_the_share_legs() {
let day = "2024-03-01";
let mut expiry = option_activity("exp", "OPEXP", "QS240301C00010500", "-1", day);
expiry["created_at"] = serde_json::json!("2024-03-01T21:30:00Z");
let mut canceled = option_activity("canceled", "OPEXP", "QS240301P00000500", "-1", day);
canceled["status"] = serde_json::json!("canceled");
let server = serve_json(serde_json::json!([
expiry,
option_activity("exc", "OPEXC", "QS240301C00006500", "-1", day),
share_leg("exc-shares", "QS", "100", "6.5", day),
option_activity("asn", "OPASN", "QS240301P00009000", "1", day),
share_leg("asn-shares", "QS", "100", "9", day),
share_leg("orphan", "QS", "-100", "7", day),
canceled,
option_activity("no-price", "OPTRD", "QS", "100", day),
option_activity("no-qty", "OPEXP", "QS240301C00011000", "0", day),
option_activity("later", "OPEXP", "QS240308C00010500", "-1", "2024-03-02"),
]))
.await;
let client = client_for(&server);
let read = client
.fetch_option_events(time("2024-03-01T00:00:00Z"), time("2024-03-01T23:00:00Z"))
.await
.unwrap();
assert_eq!(read.resume, None);
let summary: Vec<_> = read
.trades
.iter()
.map(|trade| {
(
trade.id.0.to_string(),
trade.instrument.name().to_string(),
trade.origin,
trade.side,
trade.price,
trade.quantity,
)
})
.collect();
let row = |id: &str, symbol: &str, origin, side, price: Decimal, quantity: i64| {
(
id.to_owned(),
symbol.to_owned(),
origin,
side,
price,
Decimal::from(quantity),
)
};
assert_eq!(
summary,
[
row(
"exp",
"QS240301C00010500",
TradeOrigin::Expiry,
Side::Sell,
Decimal::ZERO,
1
),
row(
"exc",
"QS240301C00006500",
TradeOrigin::Exercise,
Side::Sell,
Decimal::ZERO,
1
),
row(
"exc-shares",
"QS",
TradeOrigin::Exercise,
Side::Buy,
Decimal::new(65, 1),
100
),
row(
"asn",
"QS240301P00009000",
TradeOrigin::Assignment,
Side::Buy,
Decimal::ZERO,
1
),
row(
"asn-shares",
"QS",
TradeOrigin::Assignment,
Side::Buy,
Decimal::from(9),
100
),
row(
"orphan",
"QS",
TradeOrigin::External,
Side::Sell,
Decimal::from(7),
100
),
]
);
let exp = &read.trades[0];
assert_eq!(
exp.time_exchange,
time("2024-03-01T21:30:00Z"),
"created_at"
);
assert_eq!(exp.order_id, OrderId::new("exp"));
assert_eq!(exp.fees.fees, Decimal::ZERO);
assert_eq!(
read.trades[1].time_exchange,
time("2024-03-01T00:00:00Z"),
"no created_at: the start of its date"
);
let queries = query_pairs(&server).await;
let [query] = queries.as_slice() else {
panic!("one request: {queries:?}");
};
assert!(
query.contains(&("activity_types".into(), "OPASN,OPEXC,OPEXP,OPTRD".into())),
"{query:?}"
);
}
#[tokio::test]
async fn fetch_option_events_pairs_a_share_leg_by_its_side() {
let day = "2024-03-01";
let server = serve_json(serde_json::json!([
option_activity("exc-call", "OPEXC", "QS240301C00006500", "-1", day),
option_activity("asn-call", "OPASN", "QS240301C00006500", "1", day),
share_leg("sold", "QS", "-100", "6.5", day),
share_leg("bought", "QS", "100", "6.5", day),
share_leg("bought-again", "QS", "100", "6.5", day),
share_leg("other-day", "QS", "100", "6.5", "2024-02-29"),
]))
.await;
let client = client_for(&server);
let read = client
.fetch_option_events(time("2024-02-29T00:00:00Z"), time("2024-03-01T23:00:00Z"))
.await
.unwrap();
let origin = |id: &str| {
read.trades
.iter()
.find(|trade| trade.id == TradeId::new(id))
.map(|trade| trade.origin)
};
assert_eq!(origin("sold"), Some(TradeOrigin::Assignment));
assert_eq!(origin("bought"), Some(TradeOrigin::Exercise));
assert_eq!(origin("bought-again"), Some(TradeOrigin::External));
assert_eq!(origin("other-day"), Some(TradeOrigin::External));
}
#[tokio::test]
async fn fetch_option_events_pairs_a_share_leg_by_its_size() {
let day = "2024-03-01";
let server = serve_json(serde_json::json!([
option_activity("asn-two", "OPASN", "QS240301P00006500", "2", day),
option_activity("exc-one", "OPEXC", "QS240301C00006500", "-1", day),
option_activity("exc-adjusted", "OPEXC", "QS240301C00006500", "-1", day),
share_leg("hundred", "QS", "100", "6.5", day),
share_leg("two-hundred", "QS", "200", "6.5", day),
share_leg("hundred-fifty", "QS", "150", "6.5", day),
]))
.await;
let client = client_for(&server);
let read = client
.fetch_option_events(time("2024-03-01T00:00:00Z"), time("2024-03-01T23:00:00Z"))
.await
.unwrap();
let origin = |id: &str| {
read.trades
.iter()
.find(|trade| trade.id == TradeId::new(id))
.map(|trade| trade.origin)
};
assert_eq!(origin("hundred"), Some(TradeOrigin::Exercise));
assert_eq!(origin("two-hundred"), Some(TradeOrigin::Assignment));
assert_eq!(origin("hundred-fifty"), Some(TradeOrigin::Exercise));
}
#[tokio::test]
async fn fetch_option_events_reads_whole_days_and_returns_the_span() {
let mut early =
option_activity("early", "OPEXP", "QS240301C00010500", "-1", "2024-03-01");
early["created_at"] = serde_json::json!("2024-03-01T10:00:00Z");
let mut inside =
option_activity("inside", "OPEXP", "QS240301C00011500", "-1", "2024-03-01");
inside["created_at"] = serde_json::json!("2024-03-01T16:00:00Z");
let server = serve_json(serde_json::json!([
option_activity(
"dated-before",
"OPEXP",
"QS240229C00010500",
"-1",
"2024-02-29"
),
early,
option_activity("dated", "OPEXP", "QS240301C00012500", "-1", "2024-03-01"),
inside,
]))
.await;
let client = client_for(&server);
let read = client
.fetch_option_events(time("2024-03-01T15:00:00Z"), time("2024-03-01T17:00:00Z"))
.await
.unwrap();
let ids: Vec<_> = read
.trades
.iter()
.map(|trade| trade.id.0.to_string())
.collect();
assert_eq!(ids, ["dated", "inside"]);
let queries = query_pairs(&server).await;
let [query] = queries.as_slice() else {
panic!("one request: {queries:?}");
};
assert!(
query.contains(&("after".into(), "2024-03-01T00:00:00.000Z".into())),
"{query:?}"
);
assert!(
query.contains(&("until".into(), "2024-03-02T00:00:00.000Z".into())),
"{query:?}"
);
}
#[tokio::test]
async fn fetch_option_events_reads_nothing_for_an_empty_span() {
let server = MockServer::start().await;
let client = client_for(&server);
let read = client
.fetch_option_events(time("2024-03-02T00:00:00Z"), time("2024-03-01T00:00:00Z"))
.await
.unwrap();
assert!(read.trades.is_empty() && read.resume.is_none());
assert!(query_pairs(&server).await.is_empty());
}
async fn read_span(
client: &AlpacaClient,
start: DateTime<Utc>,
end: DateTime<Utc>,
instruments: &[InstrumentNameExchange],
) -> (std::collections::HashSet<String>, usize) {
use crate::client::ExecutionClient;
let mut ids = std::collections::HashSet::new();
let mut from = start;
for calls in 1..=10 {
let read = match client.fetch_trades(from, end, instruments).await {
Ok(read) => read,
Err(error) => panic!("call {calls} failed: {error:?}"),
};
ids.extend(read.trades.iter().map(|trade| trade.id.0.to_string()));
match read.resume {
Some(resume) => from = resume,
None => return (ids, calls),
}
}
panic!("the reads did not reach the span's end");
}
#[tokio::test]
async fn fetch_trades_reads_a_busy_span_completely_across_calls() {
let base = time("2025-04-18T14:30:00Z");
let mut activities = Vec::new();
for ms in 0..4_000_i64 {
let symbol = if ms % 2 == 0 { "SPY" } else { "QQQ" };
for (k, micros) in [("a", 900), ("b", 500), ("c", 100)] {
let at = base + TimeDelta::milliseconds(ms) + TimeDelta::microseconds(micros);
activities.push(fill_at(&format!("{ms:017}::{ms}-{k}"), symbol, at));
}
}
let server = serve_activities(activities).await;
let client = client_for(&server);
let end = time("2025-04-18T15:00:00Z");
let (ids, calls) = read_span(&client, recovery_after(), end, &[]).await;
assert_eq!(ids.len(), 12_000, "every fill is read once or more");
assert_eq!(calls, 3, "5,000 activities per call");
let spy = [InstrumentNameExchange::new("SPY")];
let (ids, _) = read_span(&client, recovery_after(), end, &spy).await;
assert_eq!(ids.len(), 6_000, "every SPY fill, and only those");
let Some(requests) = server.received_requests().await else {
panic!("requests are recorded");
};
assert!(
requests
.iter()
.all(|request| request.url.query_pairs().any(
|(name, value)| name == "until" && value == "2025-04-18T15:00:00.001Z"
)),
"every request is bounded by the span's end"
);
}
#[tokio::test]
async fn fetch_trades_says_where_to_read_on_after_a_call_without_matches() {
use crate::client::ExecutionClient;
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(make_activities_json(ALPACA_MAX_ACTIVITIES, "act")),
)
.mount(&server)
.await;
let qqq = [InstrumentNameExchange::new("QQQ")];
let Ok(read) = client_for(&server)
.fetch_trades(recovery_after(), Utc::now(), &qqq)
.await
else {
panic!("the read succeeds");
};
assert!(read.trades.is_empty(), "every fill read is SPY's");
assert_eq!(read.resume, Some(time("2025-04-18T14:30:00Z")));
}
#[tokio::test]
async fn fetch_trades_that_cannot_advance_is_truncated() {
use crate::client::ExecutionClient;
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(make_activities_json(ALPACA_MAX_ACTIVITIES, "act")),
)
.mount(&server)
.await;
let result = client_for(&server)
.fetch_trades(time("2025-04-18T14:30:00.000500Z"), Utc::now(), &[])
.await;
assert!(
matches!(
result,
Err(UnindexedClientError::Truncated { fills_read: 5_000 })
),
"got {result:?}"
);
}
#[tokio::test]
async fn fetch_trades_returns_only_the_span() {
let (start, end) = (
time("2025-04-18T14:30:00.123456Z"),
time("2025-04-18T14:30:05.678901Z"),
);
let micro = TimeDelta::microseconds(1);
let server = serve_activities(vec![
fill_at("1::before", "SPY", start - micro),
fill_at("2::start", "SPY", start),
fill_at("3::end", "SPY", end),
fill_at("4::after", "SPY", end + micro),
])
.await;
let (ids, calls) = read_span(&client_for(&server), start, end, &[]).await;
assert_eq!(calls, 1);
let mut ids: Vec<_> = ids.into_iter().collect();
ids.sort();
assert_eq!(ids, ["end", "start"]);
}
#[tokio::test]
async fn fetch_trades_that_read_past_the_end_is_complete() {
use crate::client::ExecutionClient;
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.respond_with(
ResponseTemplate::new(200)
.set_body_json(make_activities_json(ALPACA_MAX_ACTIVITIES, "act")),
)
.mount(&server)
.await;
let Ok(read) = client_for(&server)
.fetch_trades(recovery_after(), time("2025-04-18T14:29:00Z"), &[])
.await
else {
panic!("the read succeeds");
};
assert_eq!(read, TradesRead::complete(Vec::new()));
}
#[tokio::test]
async fn fetch_trades_accepts_the_extreme_times() {
use crate::client::ExecutionClient;
let server = serve_activities(vec![fill_at(
"1::fill",
"SPY",
time("2025-04-18T14:30:00Z"),
)])
.await;
let Ok(read) = client_for(&server)
.fetch_trades(DateTime::<Utc>::MIN_UTC, DateTime::<Utc>::MAX_UTC, &[])
.await
else {
panic!("the read succeeds");
};
assert_eq!(read.trades.len(), 1);
assert_eq!(read.resume, None);
let Some(requests) = server.received_requests().await else {
panic!("requests are recorded");
};
let [request] = requests.as_slice() else {
panic!("one request: {requests:?}");
};
let query: std::collections::HashMap<_, _> =
request.url.query_pairs().into_owned().collect();
assert_eq!(
query.get("after").map(String::as_str),
Some("1970-01-01T00:00:00.000Z")
);
assert!(!query.contains_key("until"), "{query:?}");
let before_epoch = DateTime::UNIX_EPOCH - TimeDelta::days(1);
let Ok(read) = client_for(&server)
.fetch_trades(DateTime::<Utc>::MIN_UTC, before_epoch, &[])
.await
else {
panic!("the read succeeds");
};
assert_eq!(read, TradesRead::complete(Vec::new()));
assert_eq!(server.received_requests().await.map(|r| r.len()), Some(1));
}
#[tokio::test]
async fn fetch_trades_reads_nothing_for_an_empty_span() {
use crate::client::ExecutionClient;
let server = MockServer::start().await;
let start = recovery_after();
let Ok(read) = client_for(&server)
.fetch_trades(start, start - TimeDelta::seconds(1), &[])
.await
else {
panic!("an empty span reads");
};
assert_eq!(read, TradesRead::complete(Vec::new()));
assert_eq!(server.received_requests().await.map(|r| r.len()), Some(0));
}
#[tokio::test]
async fn recovery_reads_from_the_start_of_the_disconnects_millisecond() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account/activities"))
.and(wiremock::matchers::query_param(
"after",
"2025-01-01T00:00:00.123Z",
))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!([])))
.expect(1)
.mount(&server)
.await;
let (tx, _rx) = mpsc::unbounded_channel();
let outcome = recover_fills(
&reqwest::Client::new(),
&RateLimitTracker::new(),
&[],
&server.uri(),
time("2025-01-01T00:00:00.123456Z"),
&tx,
&new_dedup_cache(),
&KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
)
.await;
assert_eq!(outcome, Ok(()));
server.verify().await;
}
fn dedup_keys(dedup: &SharedDedupCache) -> Vec<String> {
let mut keys: Vec<_> = dedup
.lock()
.iter()
.filter_map(|(key, _)| match key {
FillKey::Cumulative(key) => Some(key.to_string()),
FillKey::Execution(_) | FillKey::CumulativeWithoutId(_) => None,
})
.collect();
keys.reverse();
keys
}
fn trade_of(
event: &UnindexedAccountEvent,
) -> &Trade<AssetNameExchange, InstrumentNameExchange> {
match &event.kind {
AccountEventKind::Trade(t) => t,
other => panic!("expected a Trade event, got {other:?}"),
}
}
#[tokio::test]
async fn a_recovered_fill_reports_the_cumulative_alpaca_sent() {
let events = drive_recover_fills(vec![
activity_json("act-1", "ord-1", "2", Some("2")),
activity_json("act-2", "ord-1", "3", Some("5")),
])
.await;
let reported: Vec<_> = events
.iter()
.map(|e| trade_of(e).order_filled_quantity)
.collect();
assert_eq!(
reported,
vec![
Some(Decimal::from_str("2").unwrap()),
Some(Decimal::from_str("5").unwrap())
]
);
}
#[tokio::test]
async fn the_dedup_key_uses_the_venue_cumulative_not_an_intra_batch_count() {
let dedup = new_dedup_cache();
drive_recover_fills_with(
vec![activity_json("act-1", "ord-1", "2", Some("5"))],
dedup.clone(),
&KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
)
.await;
assert_eq!(
dedup_keys(&dedup),
vec!["ord-1:5"],
"counting executions within the batch would key this as ord-1:2"
);
}
async fn recover_after_ws_fill(
execution_id: Option<&str>,
activity_id: &str,
) -> Vec<UnindexedAccountEvent> {
let dedup = new_dedup_cache();
let update = AlpacaTradeUpdate {
event: SmolStr::new("partial_fill"),
execution_id,
previous_execution_id: None,
order: super::make_order_ws("ord-1", "SPY", "buy", "5"),
price: Some("100.00"),
qty: Some("2"),
timestamp: None,
};
assert!(
!is_duplicate_fill(&dedup, ws_execution_id(&update), early_dedup_key(&update)),
"precondition: first sighting"
);
drive_recover_fills_with(
vec![activity_json(activity_id, "ord-1", "2", Some("5"))],
dedup,
&KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
)
.await
}
#[tokio::test]
async fn a_fill_already_delivered_over_websocket_is_not_recovered_twice() {
for (ws_execution_id, activity_id) in [
(Some("exec-a"), "20250418103000000::exec-a"),
(None, "20250418103000000::exec-a"),
(Some("exec-a"), "act-1"),
] {
let events = recover_after_ws_fill(ws_execution_id, activity_id).await;
assert!(
events.is_empty(),
"a fill already delivered over WebSocket ({ws_execution_id:?}) must not be \
re-delivered by recovery ({activity_id}), got {events:?}"
);
}
}
async fn recover_after_correction(fields: &str) -> Vec<UnindexedAccountEvent> {
let (tx, _rx) = mpsc::unbounded_channel();
let dedup = new_dedup_cache();
let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
process_ws_text(
&super::amendment_frame("trade_correct", fields),
&tx,
&dedup,
&known,
&mut VenueClock::new(),
);
drive_recover_fills_with(
vec![activity_json(
"20250418103100000::exec-c",
"ord-1",
"2",
Some("2"),
)],
dedup,
&known,
)
.await
}
#[tokio::test]
async fn a_corrected_execution_is_not_recovered_again() {
let events = recover_after_correction(
r#""execution_id":"exec-c","previous_execution_id":"exec-a","price":"101.50","qty":"2","#,
)
.await;
assert!(
events.is_empty(),
"the corrected execution was delivered as the replacement, got {events:?}"
);
}
#[tokio::test]
async fn an_unresolved_correction_does_not_keep_its_execution_from_recovery() {
let events = recover_after_correction(
r#""execution_id":"exec-c","previous_execution_id":"exec-a","qty":"2","#,
)
.await;
let ids: Vec<_> = events.iter().map(|e| trade_of(e).id.0.as_str()).collect();
assert_eq!(ids, vec!["exec-c"]);
}
#[tokio::test]
async fn a_different_execution_at_a_cumulative_already_seen_is_recovered() {
let events = recover_after_ws_fill(Some("exec-a"), "20250418103000000::exec-b").await;
let ids: Vec<_> = events.iter().map(|e| trade_of(e).id.0.as_str()).collect();
assert_eq!(ids, vec!["exec-b"]);
}
#[tokio::test]
async fn without_a_reported_cumulative_the_previous_behaviour_is_preserved() {
let dedup = new_dedup_cache();
let events = drive_recover_fills_with(
vec![
activity_json("act-1", "ord-1", "2", None),
activity_json("act-2", "ord-1", "3", None),
],
dedup.clone(),
&KnownLiveOrders::shared(ExchangeId::AlpacaBroker),
)
.await;
assert_eq!(
dedup_keys(&dedup),
vec!["ord-1:2", "ord-1:5"],
"intra-batch accumulation"
);
let ids: Vec<_> = events.iter().map(|e| trade_of(e).id.0.as_str()).collect();
assert_eq!(ids, vec!["act-1", "act-2"], "each keeps its activity's id");
assert!(
events
.iter()
.all(|e| trade_of(e).order_filled_quantity.is_none()),
"a venue that reported no cumulative must not have one invented for it"
);
}
fn order_json(cid: &str, symbol: &str, status: &str, filled: &str) -> serde_json::Value {
serde_json::json!({
"id": format!("id-{cid}"),
"client_order_id": cid,
"symbol": symbol,
"qty": "2",
"filled_qty": filled,
"filled_avg_price": if filled == "0" { None } else { Some("101.5") },
"status": status,
"side": "buy",
"type": "limit",
"time_in_force": "gtc",
"limit_price": "100",
"created_at": "2026-10-01T14:30:00Z",
"updated_at": "2026-10-01T15:00:00Z"
})
}
async fn mount_lookup(server: &MockServer, cid: &str, order: ResponseTemplate, calls: u64) {
use wiremock::matchers::query_param;
Mock::given(method("GET"))
.and(path("/v2/orders:by_client_order_id"))
.and(query_param("client_order_id", cid))
.respond_with(order)
.expect(calls)
.mount(server)
.await;
}
fn ok(body: serde_json::Value) -> ResponseTemplate {
ResponseTemplate::new(200).set_body_json(body)
}
fn alpaca_key(instrument: &str, cid: &str) -> UnindexedOrderKey {
OrderKey::new(
ExchangeId::AlpacaBroker,
InstrumentNameExchange::new(instrument),
StrategyId::new("strategy"),
ClientOrderId::new(cid),
)
}
fn hold(known: &SharedKnownLiveOrders, key: &UnindexedOrderKey) {
let open = Open::new(
VenueOrderId::Assigned(OrderId::new(format!("id-{}", key.cid))),
Utc::now(),
Decimal::ZERO,
);
known
.lock()
.live(key, Decimal::TWO, OrderKind::Limit, &open);
}
#[tokio::test]
async fn fetch_ended_orders_reads_each_end_alpaca_reports() {
use rust_decimal_macros::dec;
let server = MockServer::start().await;
for (cid, status, filled) in [
("filled", "filled", "2"),
("canceled", "canceled", "1"),
("replaced", "replaced", "0"),
("expired", "expired", "1"),
("rejected", "rejected", "0"),
("done", "done_for_day", "1"),
("live", "new", "0"),
("unknown-status", "frozen", "0"),
] {
mount_lookup(&server, cid, ok(order_json(cid, "SPY", status, filled)), 1).await;
}
mount_lookup(
&server,
"elsewhere",
ok(order_json("elsewhere", "QQQ", "canceled", "0")),
1,
)
.await;
mount_lookup(
&server,
"gone",
ResponseTemplate::new(404).set_body_json(serde_json::json!({
"code": 40410000,
"message": "order not found for gone"
})),
1,
)
.await;
let keys: Vec<_> = [
"filled",
"canceled",
"replaced",
"expired",
"rejected",
"done",
"live",
"unknown-status",
"elsewhere",
"gone",
]
.into_iter()
.map(|cid| alpaca_key("spy", cid))
.collect();
let ended = client_for(&server).fetch_ended_orders(&keys).await.unwrap();
let mut found: Vec<_> = ended
.iter()
.map(|order| (order.key.cid.0.as_str(), &order.state))
.collect();
found.sort_by_key(|(cid, _)| *cid);
let [
(canceled, InactiveOrderState::Cancelled(cancelled)),
(expired, InactiveOrderState::Expired(expiry)),
(filled, InactiveOrderState::FullyFilled(fill)),
(rejected, InactiveOrderState::OpenFailed(failed)),
(replaced, InactiveOrderState::Cancelled(replacement)),
] = found.as_slice()
else {
panic!("filled, canceled, replaced, expired and rejected: {found:?}");
};
assert_eq!(
[*canceled, *expired, *filled, *rejected, *replaced],
["canceled", "expired", "filled", "rejected", "replaced"]
);
assert_eq!(
(fill.filled_quantity, fill.avg_price),
(dec!(2), Some(dec!(101.5)))
);
assert_eq!(
cancelled.filled_quantity,
Some(dec!(1)),
"what filled before"
);
assert_eq!(expiry.filled_quantity, Some(dec!(1)));
assert_eq!(replacement.filled_quantity, Some(Decimal::ZERO));
assert_eq!(
cancelled.time_exchange,
parse_timestamp("2026-10-01T15:00:00Z").unwrap(),
"stamped when the order last changed"
);
assert!(matches!(
failed,
OrderError::Rejected(ApiError::OrderRejected(message))
if message.contains("id-rejected")
));
assert!(
ended
.iter()
.all(|order| order.key.strategy == StrategyId::new("strategy")
&& order.key.instrument == InstrumentNameExchange::new("spy")),
"each carries the key it was asked for"
);
}
#[tokio::test]
async fn fetch_ended_orders_fails_when_a_lookup_fails() {
let server = MockServer::start().await;
mount_lookup(&server, "a", ok(order_json("a", "SPY", "canceled", "0")), 1).await;
mount_lookup(&server, "b", ResponseTemplate::new(500), 1).await;
let result = client_for(&server)
.fetch_ended_orders(&[alpaca_key("SPY", "a"), alpaca_key("SPY", "b")])
.await;
assert!(
matches!(result, Err(UnindexedClientError::Connectivity(_))),
"{result:?}"
);
}
#[tokio::test]
async fn a_reconnect_check_lists_once_and_reports_the_orders_that_ended() {
use wiremock::matchers::query_param;
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/orders"))
.and(query_param("status", "open"))
.and(query_param("symbols", "QQQ,SPY"))
.respond_with(ok(serde_json::json!([order_json(
"spy-listed",
"SPY",
"new",
"0"
)])))
.expect(1)
.mount(&server)
.await;
mount_lookup(&server, "spy-listed", ResponseTemplate::new(500), 0).await;
mount_lookup(
&server,
"spy-cancelled",
ok(order_json("spy-cancelled", "SPY", "canceled", "1")),
1,
)
.await;
mount_lookup(
&server,
"spy-done",
ok(order_json("spy-done", "SPY", "done_for_day", "1")),
1,
)
.await;
mount_lookup(
&server,
"qqq-filled",
ok(order_json("qqq-filled", "QQQ", "filled", "2")),
1,
)
.await;
let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
let [listed, cancelled, done, filled] = [
("SPY", "spy-listed"),
("SPY", "spy-cancelled"),
("SPY", "spy-done"),
("QQQ", "qqq-filled"),
]
.map(|(instrument, cid)| alpaca_key(instrument, cid));
for key in [&listed, &cancelled, &done, &filled] {
hold(&known, key);
}
let mut unchecked = UncheckedOrders::default();
unchecked.open(known.lock().instruments());
let config = Arc::new(AlpacaConfig::with_base_url(
"test-key".into(),
"test-secret".into(),
server.uri(),
));
let (tx, mut rx) = mpsc::unbounded_channel();
recover_alpaca_ended_orders(
&reqwest::Client::new(),
&Arc::new(RateLimitTracker::new()),
&config,
&known,
&mut unchecked,
&tx,
)
.await;
let mut reported: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok())
.map(|event| {
let AccountEventKind::OrderSnapshot(
rustrade_integration::collection::snapshot::Snapshot(order),
) = event.kind
else {
panic!("an order snapshot: {event:?}");
};
assert!(matches!(order.state, OrderState::Inactive(_)), "{order:?}");
order.key.cid
})
.collect();
reported.sort();
assert_eq!(reported, [filled.cid.clone(), cancelled.cid.clone()]);
let known = known.lock();
assert!(known.contains(&listed.cid) && known.contains(&done.cid));
assert!(!known.contains(&cancelled.cid) && !known.contains(&filled.cid));
assert!(unchecked.is_empty(), "both instruments checked");
}
#[tokio::test]
async fn a_failed_listing_is_retried_for_every_instrument_in_it() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/orders"))
.respond_with(ResponseTemplate::new(503))
.expect(1)
.mount(&server)
.await;
let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
hold(&known, &alpaca_key("SPY", "a"));
hold(&known, &alpaca_key("QQQ", "b"));
let mut unchecked = UncheckedOrders::default();
unchecked.open(known.lock().instruments());
let config = Arc::new(AlpacaConfig::with_base_url(
"test-key".into(),
"test-secret".into(),
server.uri(),
));
let (tx, mut rx) = mpsc::unbounded_channel();
recover_alpaca_ended_orders(
&reqwest::Client::new(),
&Arc::new(RateLimitTracker::new()),
&config,
&known,
&mut unchecked,
&tx,
)
.await;
assert!(rx.try_recv().is_err(), "nothing reported");
let now = tokio::time::Instant::now();
assert!(
unchecked.ready(&NoPendingFills, now).is_empty(),
"both wait"
);
for instrument in ["QQQ", "SPY"] {
assert_eq!(
unchecked.failed(
&InstrumentNameExchange::new(instrument),
now,
Duration::ZERO
),
Some(crate::client::order_recovery::GapFailure::Retry(
Duration::from_secs(crate::client::order_recovery::GAP_RETRY_BASE_SECS * 2)
)),
"{instrument} has failed once already"
);
}
assert_eq!(known.lock().instruments().len(), 2, "both still held");
}
#[tokio::test]
async fn placing_listing_and_cancelling_keep_the_held_orders_in_step() {
use crate::order::request::{OrderRequestCancel, RequestCancel, RequestOpen};
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v2/orders"))
.respond_with(ok(order_json("resting", "SPY", "new", "0")))
.up_to_n_times(1)
.mount(&server)
.await;
Mock::given(method("POST"))
.and(path("/v2/orders"))
.respond_with(ok(order_json("filled", "SPY", "filled", "2")))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/v2/orders"))
.respond_with(ok(serde_json::json!([order_json(
"listed", "QQQ", "new", "0"
)])))
.mount(&server)
.await;
Mock::given(method("DELETE"))
.and(path("/v2/orders/id-resting"))
.respond_with(ResponseTemplate::new(204))
.mount(&server)
.await;
let client = client_for(&server);
let spy = InstrumentNameExchange::new("SPY");
for cid in ["resting", "filled"] {
let request = OrderRequestOpen {
key: OrderKey::new(
ExchangeId::AlpacaBroker,
&spy,
StrategyId::new("strategy"),
ClientOrderId::new(cid),
),
state: RequestOpen::new(
Side::Buy,
Some(Decimal::ONE_HUNDRED),
Decimal::TWO,
OrderKind::Limit,
TimeInForce::GoodUntilCancelled { post_only: false },
),
};
client.open_order(request).await;
}
client.fetch_open_orders(&[]).await.unwrap();
{
let known = client.known_live.lock();
assert!(known.contains(&ClientOrderId::new("resting")));
assert!(
!known.contains(&ClientOrderId::new("filled")),
"filled in its placement response"
);
assert!(known.contains(&ClientOrderId::new("listed")));
}
let cancel = client
.cancel_order(OrderRequestCancel {
key: OrderKey::new(
ExchangeId::AlpacaBroker,
&spy,
StrategyId::new("strategy"),
ClientOrderId::new("resting"),
),
state: RequestCancel {
id: Some(VenueOrderId::Assigned(OrderId::new("id-resting"))),
},
})
.await;
let Ok(cancelled) = &cancel.state else {
panic!("expected Ok, got {cancel:?}");
};
assert_eq!(
cancelled.filled_quantity, None,
"the DELETE response has no body, so what filled is unknown"
);
assert!(
client
.known_live
.lock()
.contains(&ClientOrderId::new("resting")),
"a cancel Alpaca has only accepted has not ended the order"
);
}
#[tokio::test]
async fn an_order_that_ended_in_its_placement_response_is_reported_as_it_ended() {
use crate::order::request::RequestOpen;
let Some(time) = parse_timestamp("2026-10-01T15:00:00Z") else {
panic!("order_json's updated_at parses");
};
let id = |cid: &str| OrderId::new(format!("id-{cid}"));
let open = |cid: &str, filled| {
OrderState::active(Open::new(VenueOrderId::Assigned(id(cid)), time, filled))
};
let cases: [(&str, &str, &str, UnindexedOrderState, bool); 13] = [
(
"ioc-unfilled",
"expired",
"0",
OrderState::expired(Expired::new(
id("ioc-unfilled"),
time,
Some(Decimal::ZERO),
)),
false,
),
(
"ioc-partial",
"expired",
"1",
OrderState::expired(Expired::new(id("ioc-partial"), time, Some(Decimal::ONE))),
false,
),
(
"fill-unknown",
"expired",
"not-a-number",
OrderState::expired(Expired::new(id("fill-unknown"), time, None)),
false,
),
(
"cancelled",
"canceled",
"1",
OrderState::inactive(Cancelled::new(id("cancelled"), time, Some(Decimal::ONE))),
false,
),
(
"rejected",
"rejected",
"0",
OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
"Alpaca rejected order id-rejected after accepting it".to_string(),
))),
false,
),
(
"filled",
"filled",
"2",
OrderState::fully_filled(Filled::new(
id("filled"),
time,
Decimal::TWO,
Some(Decimal::new(1015, 1)),
)),
false,
),
(
"replaced",
"replaced",
"0",
OrderState::inactive(Cancelled::new(id("replaced"), time, Some(Decimal::ZERO))),
false,
),
(
"live-covered",
"partially_filled",
"2",
OrderState::fully_filled(Filled::new(
id("live-covered"),
time,
Decimal::TWO,
Some(Decimal::new(1015, 1)),
)),
false,
),
("new", "new", "0", open("new", Decimal::ZERO), true),
(
"live-fill-unknown",
"new",
"not-a-number",
open("live-fill-unknown", Decimal::ZERO),
true,
),
(
"partial",
"partially_filled",
"1",
open("partial", Decimal::ONE),
true,
),
(
"for-the-day",
"done_for_day",
"1",
open("for-the-day", Decimal::ONE),
true,
),
(
"unknown",
"a_status_not_yet_documented",
"0",
open("unknown", Decimal::ZERO),
true,
),
];
for (cid, status, filled, expected, held) in cases {
for bracket in [false, true] {
let server = MockServer::start().await;
Mock::given(method("POST"))
.and(path("/v2/orders"))
.respond_with(ok(order_json(cid, "SPY", status, filled)))
.expect(1)
.mount(&server)
.await;
let client = client_for(&server);
let spy = InstrumentNameExchange::new("SPY");
let time_in_force = TimeInForce::GoodUntilCancelled { post_only: false };
let state = if bracket {
client
.open_bracket_order(AlpacaBracketOrderRequest::new(
spy,
StrategyId::new("strategy"),
ClientOrderId::new(cid),
Side::Buy,
Decimal::TWO,
Decimal::ONE_HUNDRED,
Decimal::new(120, 0),
Decimal::new(90, 0),
time_in_force,
))
.await
.parent
.state
} else {
let request = OrderRequestOpen {
key: OrderKey::new(
ExchangeId::AlpacaBroker,
&spy,
StrategyId::new("strategy"),
ClientOrderId::new(cid),
),
state: RequestOpen::new(
Side::Buy,
Some(Decimal::ONE_HUNDRED),
Decimal::TWO,
OrderKind::Limit,
time_in_force,
),
};
client.open_order(request).await.state
};
assert_eq!(state, expected, "{status} {filled}, bracket: {bracket}");
assert_eq!(
client.known_live.lock().contains(&ClientOrderId::new(cid)),
held,
"{status} {filled}, bracket: {bracket}"
);
}
}
}
#[test]
fn the_stream_keeps_the_held_orders_in_step() {
let (tx, mut rx) = mpsc::unbounded_channel();
let dedup = new_dedup_cache();
let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
let frame = |event: &str, filled: &str| {
format!(
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}"}}}}}}"#
)
};
let cid = ClientOrderId::new("a");
process_ws_text(
&frame("new", "0"),
&tx,
&dedup,
&known,
&mut VenueClock::new(),
);
assert!(known.lock().contains(&cid), "acknowledged");
process_ws_text(
&frame("done_for_day", "1"),
&tx,
&dedup,
&known,
&mut VenueClock::new(),
);
assert!(known.lock().contains(&cid), "done for the day is not ended");
process_ws_text(
&frame("canceled", "1"),
&tx,
&dedup,
&known,
&mut VenueClock::new(),
);
assert!(!known.lock().contains(&cid), "cancelled");
let events: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
let [_, done, cancelled] = events.as_slice() else {
panic!("three events: {events:?}");
};
let AccountEventKind::OrderSnapshot(
rustrade_integration::collection::snapshot::Snapshot(order),
) = &done.kind
else {
panic!("done_for_day reports an order snapshot: {done:?}");
};
let OrderState::Active(ActiveOrderState::Open(open)) = &order.state else {
panic!("open: {order:?}");
};
assert_eq!(open.filled_quantity, Decimal::ONE, "with what it filled");
assert!(matches!(
cancelled.kind,
AccountEventKind::OrderCancelled(_)
));
}
#[tokio::test]
async fn a_recovered_fill_that_completes_an_order_stops_it_being_held() {
let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
let key = alpaca_key("SPY", "a");
known.lock().live(
&key,
Decimal::TWO,
OrderKind::Limit,
&Open::new(
VenueOrderId::Assigned(OrderId::new("ord-1")),
Utc::now(),
Decimal::ZERO,
),
);
drive_recover_fills_with(
vec![activity_json("act-1", "ord-1", "1", Some("1"))],
new_dedup_cache(),
&known,
)
.await;
assert!(known.lock().contains(&key.cid), "half filled");
drive_recover_fills_with(
vec![activity_json("act-2", "ord-1", "1", Some("2"))],
new_dedup_cache(),
&known,
)
.await;
assert!(!known.lock().contains(&key.cid), "filled");
}
#[tokio::test]
async fn the_check_loop_runs_at_once_then_waits_out_a_failure() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/orders"))
.respond_with(ResponseTemplate::new(503))
.expect(1)
.mount(&server)
.await;
let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
hold(&known, &alpaca_key("SPY", "a"));
let mut unchecked = UncheckedOrders::default();
unchecked.open(known.lock().instruments());
let config = Arc::new(AlpacaConfig::with_base_url(
"test-key".into(),
"test-secret".into(),
server.uri(),
));
let (tx, _rx) = mpsc::unbounded_channel();
let ran = tokio::time::timeout(
Duration::from_secs(1),
run_order_checks(
&reqwest::Client::new(),
&Arc::new(RateLimitTracker::new()),
&config,
&known,
&mut unchecked,
&tx,
),
)
.await;
assert!(ran.is_err(), "the loop never completes");
assert!(
unchecked.contains(&InstrumentNameExchange::new("SPY")),
"SPY waits for its retry"
);
assert_eq!(
server.received_requests().await.unwrap().len(),
1,
"the failed check is not run again at once"
);
}
#[tokio::test]
async fn the_check_loop_waits_once_the_consumer_has_gone() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.respond_with(ResponseTemplate::new(503))
.expect(0)
.mount(&server)
.await;
let known = KnownLiveOrders::shared(ExchangeId::AlpacaBroker);
hold(&known, &alpaca_key("SPY", "a"));
let mut unchecked = UncheckedOrders::default();
unchecked.open(known.lock().instruments());
let config = Arc::new(AlpacaConfig::with_base_url(
"test-key".into(),
"test-secret".into(),
server.uri(),
));
let (tx, rx) = mpsc::unbounded_channel();
drop(rx);
let ran = tokio::time::timeout(
Duration::from_millis(200),
run_order_checks(
&reqwest::Client::new(),
&Arc::new(RateLimitTracker::new()),
&config,
&known,
&mut unchecked,
&tx,
),
)
.await;
assert!(ran.is_err(), "the loop never completes");
assert!(
server.received_requests().await.unwrap().is_empty(),
"nothing asked once the consumer has gone"
);
}
async fn shortability_with_asset(
symbol: &str,
asset: serde_json::Value,
) -> (Result<Shortability, UnindexedClientError>, Vec<String>) {
use crate::client::ShortabilityClient;
let server = MockServer::start().await;
Mock::given(method("GET"))
.respond_with(ResponseTemplate::new(200).set_body_json(asset))
.mount(&server)
.await;
let result = client_for(&server)
.fetch_shortability(&InstrumentNameExchange::new(symbol))
.await;
let paths = server
.received_requests()
.await
.unwrap()
.iter()
.map(|request| request.url.path().to_owned())
.collect();
(result, paths)
}
#[tokio::test]
async fn an_easy_to_borrow_stock_is_shortable_at_no_fee() {
let (result, paths) = shortability_with_asset(
"AAPL",
serde_json::json!({
"class": "us_equity",
"symbol": "AAPL",
"shortable": true,
"easy_to_borrow": true,
"borrow_status": "easy_to_borrow",
}),
)
.await;
assert_eq!(
result.unwrap(),
Shortability::new()
.with_shortable(true)
.with_easy_to_borrow(true)
.with_fee_rate(Decimal::ZERO)
);
assert_eq!(paths, vec!["/v2/assets/AAPL"]);
}
#[tokio::test]
async fn a_hard_to_borrow_stock_has_no_known_fee() {
let (result, _) = shortability_with_asset(
"GME",
serde_json::json!({
"class": "us_equity",
"shortable": true,
"easy_to_borrow": true,
"borrow_status": "hard_to_borrow",
}),
)
.await;
assert_eq!(
result.unwrap(),
Shortability::new()
.with_shortable(true)
.with_easy_to_borrow(false),
"borrow_status wins over the easy_to_borrow flag, and a hard-to-borrow fee is unknown"
);
}
#[tokio::test]
async fn without_a_borrow_status_the_easy_to_borrow_flag_is_read() {
let (result, _) = shortability_with_asset(
"AAPL",
serde_json::json!({ "class": "us_equity", "shortable": false, "easy_to_borrow": false }),
)
.await;
assert_eq!(
result.unwrap(),
Shortability::new()
.with_shortable(false)
.with_easy_to_borrow(false)
);
}
#[tokio::test]
async fn an_asset_without_borrow_fields_is_unknown_but_for_shortable() {
let (result, paths) = shortability_with_asset(
"BTC/USD",
serde_json::json!({ "class": "crypto", "symbol": "BTC/USD", "shortable": false }),
)
.await;
assert_eq!(result.unwrap(), Shortability::new().with_shortable(false));
assert_eq!(
paths,
vec!["/v2/assets/BTC%2FUSD"],
"a crypto pair's slash is escaped, not read as a path separator"
);
}
#[tokio::test]
async fn an_option_has_no_shortability() {
let (result, _) = shortability_with_asset(
"AAPL250117C00150000",
serde_json::json!({ "class": "us_option", "shortable": true }),
)
.await;
assert!(
matches!(
result,
Err(UnindexedClientError::Api(ApiError::InstrumentInvalid(ref instrument, ref reason)))
if instrument.name().as_str() == "AAPL250117C00150000"
&& reason.contains("option")
),
"{result:?}"
);
}
#[tokio::test]
async fn an_unknown_symbol_is_an_invalid_instrument() {
use crate::client::ShortabilityClient;
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/assets/NOPE"))
.respond_with(
ResponseTemplate::new(404).set_body_json(
serde_json::json!({ "message": "asset not found for NOPE" }),
),
)
.mount(&server)
.await;
let result = client_for(&server)
.fetch_shortability(&InstrumentNameExchange::new("NOPE"))
.await;
assert!(
matches!(
result,
Err(UnindexedClientError::Api(ApiError::InstrumentInvalid(ref instrument, _)))
if instrument.name().as_str() == "NOPE"
),
"{result:?}"
);
}
struct RateLimited {
call: std::sync::atomic::AtomicU32,
last: Option<(&'static str, String)>,
}
impl RateLimited {
fn mount(last: Option<(&'static str, String)>) -> Self {
Self {
call: std::sync::atomic::AtomicU32::new(0),
last,
}
}
}
impl Respond for RateLimited {
fn respond(&self, _: &Request) -> ResponseTemplate {
let call = self.call.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
let response = ResponseTemplate::new(429)
.set_body_json(serde_json::json!({"message": "too many requests"}));
if call < MAX_RATE_LIMIT_ATTEMPTS {
return response.insert_header("retry-after", "0");
}
match &self.last {
Some((name, value)) => response.insert_header(*name, value.as_str()),
None => response,
}
}
}
fn unix_now() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs()
}
#[tokio::test]
async fn exhausted_429_retries_report_alpacas_wait() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account"))
.respond_with(RateLimited::mount(Some(("retry-after", "7".to_string()))))
.mount(&server)
.await;
let http = reqwest::Client::new();
let rl = RateLimitTracker::new();
let url = format!("{}/v2/account", server.uri());
let error = rest_with_retry::<serde_json::Value>(&rl, || http.get(&url))
.await
.unwrap_err();
let UnindexedClientError::Api(error) = error else {
panic!("expected Api, got {error:?}");
};
assert_eq!(error.retry_after(), Some(Duration::from_secs(7)));
let blocked_until = rl
.blocked_until
.lock()
.expect("the last refusal pauses the client");
assert!(
blocked_until >= tokio::time::Instant::now() + Duration::from_secs(6),
"the client's next request waits for Alpaca's wait"
);
assert_eq!(
server.received_requests().await.unwrap().len(),
MAX_RATE_LIMIT_ATTEMPTS as usize
);
}
#[tokio::test]
async fn exhausted_429_retries_read_the_reset_time_without_retry_after() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account"))
.respond_with(RateLimited::mount(Some((
"x-ratelimit-reset",
(unix_now() + 30).to_string(),
))))
.mount(&server)
.await;
let http = reqwest::Client::new();
let rl = RateLimitTracker::new();
let url = format!("{}/v2/account", server.uri());
let error = rest_with_retry::<serde_json::Value>(&rl, || http.get(&url))
.await
.unwrap_err();
let UnindexedClientError::Api(ApiError::RateLimit {
retry_after: Some(wait),
..
}) = error
else {
panic!("expected a RateLimit with a wait, got {error:?}");
};
assert!(
(Duration::from_secs(29)..=Duration::from_secs(30)).contains(&wait),
"{wait:?}"
);
}
#[tokio::test]
async fn exhausted_429_retries_without_a_hint_report_none() {
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/v2/account"))
.respond_with(RateLimited::mount(None))
.mount(&server)
.await;
let http = reqwest::Client::new();
let rl = RateLimitTracker::new();
let url = format!("{}/v2/account", server.uri());
let error = rest_with_retry::<serde_json::Value>(&rl, || http.get(&url))
.await
.unwrap_err();
assert!(
matches!(
error,
UnindexedClientError::Api(ApiError::RateLimit {
retry_after: None,
..
})
),
"{error:?}"
);
}
#[tokio::test]
async fn an_exhausted_cancel_reports_alpacas_wait() {
let server = MockServer::start().await;
Mock::given(method("DELETE"))
.and(path("/v2/orders/abc"))
.respond_with(RateLimited::mount(Some(("retry-after", "5".to_string()))))
.mount(&server)
.await;
let http = reqwest::Client::new();
let rl = RateLimitTracker::new();
let url = format!("{}/v2/orders/abc", server.uri());
let error = rest_delete_with_retry(&rl, || http.delete(&url))
.await
.unwrap_err();
let UnindexedOrderError::Rejected(error) = error else {
panic!("expected Rejected, got {error:?}");
};
assert_eq!(error.retry_after(), Some(Duration::from_secs(5)));
assert!(
rl.blocked_until.lock().is_some(),
"the last refusal pauses the client"
);
}
}
}