alpaca-sdk 0.1.0

//! The [trading API](https://docs.alpaca.markets/us/docs/trading-api) client.

use reqwest::Method;
use uuid::Uuid;

use crate::auth::Credentials;
use crate::config::BaseUrl;
use crate::error::Result;
use crate::rest::{Empty, RestClient, RestConfig};
use crate::sse::EventStreamRequest;
use crate::trading::enums::ActivityType;
use crate::trading::locates::{
    CreateLocateRequest, GetLocateQuotesRequest, GetLocatesRequest, Locate, LocateQuotes,
    LocatesPage,
};
use crate::trading::markets::{GetMarketCalendarRequest, Market, MarketCalendar};
use crate::trading::models::{
    AccountConfiguration, Activity, Asset, Calendar, CancelOrderResponse, Clock,
    ClosePositionResponse, CorporateActionAnnouncement, OptionContract, OptionContractsResponse,
    Order, PortfolioHistory, Position, TradeAccount, Watchlist,
};
use crate::trading::requests::{
    ClosePositionRequest, CreateWatchlistRequest, GetAccountActivitiesRequest, GetAssetsRequest,
    GetCalendarRequest, GetCorporateAnnouncementsRequest, GetOptionContractsRequest,
    GetOrderByIdRequest, GetOrdersRequest, GetPortfolioHistoryRequest, OrderRequest,
    ReplaceOrderRequest, UpdateWatchlistRequest,
};
use crate::trading::tokenization::{
    ByClientRequestId, GetTokenizationRequestsRequest, MintTokenRequest, TokenizationRequest,
};
use crate::trading::wallets::{
    CreateWhitelistedAddressRequest, CryptoTransfer, CryptoWallet, GetCryptoWalletsRequest,
    TransferFeeEstimate, TransferFeeEstimateRequest, WhitelistedAddress,
};
use crate::types::AssetIdent;
use crate::types::Sort;
use crate::types::path::segment;
use crate::types::serde_util::OneOrMany;

/// A client for Alpaca's trading API.
///
/// ```no_run
/// # use alpaca_sdk::{Credentials, trading::TradingClient};
/// # async fn example() -> alpaca_sdk::Result<()> {
/// let credentials = Credentials::from_env()?;
/// let client = TradingClient::new(&credentials, true)?;
///
/// let account = client.get_account().await?;
/// println!("{:?}", account.buying_power);
/// # Ok(())
/// # }
/// ```
#[derive(Debug, Clone)]
pub struct TradingClient {
    rest: RestClient,
    /// A second HTTP client, for the activity event stream: its response body
    /// is read incrementally rather than decoded whole.
    raw: reqwest::Client,
}

impl TradingClient {
    /// A client targeting the paper or live trading environment.
    ///
    /// # Errors
    /// Returns an error if the credentials cannot be encoded as headers or the
    /// underlying HTTP client fails to build.
    pub fn new(credentials: &Credentials, paper: bool) -> Result<Self> {
        Self::with_config(credentials, RestConfig::from(BaseUrl::trading(paper)))
    }

    /// A client with a custom endpoint, retry policy, or timeout.
    ///
    /// # Errors
    /// Returns an error if the credentials cannot be encoded as headers or the
    /// underlying HTTP client fails to build.
    pub fn with_config(credentials: &Credentials, config: RestConfig) -> Result<Self> {
        Ok(Self {
            raw: crate::sse::streaming_client(credentials)?,
            rest: RestClient::new(credentials, config)?,
        })
    }

    /// The underlying transport, for routes this client does not wrap.
    ///
    /// The transport is public and returns the body undecoded, so a route this
    /// crate has not wrapped is still one call away.
    #[must_use]
    pub fn rest(&self) -> &RestClient {
        &self.rest
    }

    /// Issues a request whose response body is discarded.
    ///
    /// Several routes answer `204 No Content`, and `exercise_options_position`
    /// answers with a bare string that is not JSON.
    async fn send_void(&self, method: Method, path: &str) -> Result<()> {
        self.rest
            .request_raw(
                method,
                crate::rest::Replay::ByMethod,
                path,
                None::<&Empty>,
                None::<&Empty>,
            )
            .await?;
        Ok(())
    }

    // ------------------------------------------------------------- orders

    /// Submits an order.
    ///
    /// # Errors
    /// Returns [`crate::Error::InvalidRequest`] if the order fails
    /// [`OrderRequest::validate`], or an API error if Alpaca rejects it.
    pub async fn submit_order(&self, order: &OrderRequest) -> Result<Order> {
        order.validate()?;
        self.rest.post("/orders", order).await
    }

    /// Lists orders, optionally filtered.
    ///
    /// **One page.** Alpaca returns 50 orders by default and at most 500, and
    /// this sends exactly what it is given — so an account with more history
    /// than that gets a silently truncated list rather than an error. Use
    /// [`get_all_orders`](Self::get_all_orders) to walk the whole history.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_orders(&self, filter: Option<&GetOrdersRequest>) -> Result<Vec<Order>> {
        match filter {
            Some(filter) => self.rest.get("/orders", filter).await,
            None => self.rest.get("/orders", &Empty).await,
        }
    }

    /// Every order matching `filter`, following the cursor across pages.
    ///
    /// `/v2/orders` has no `next_page_token`; it is walked with
    /// [`before_order_id`](GetOrdersRequest::before_order_id), which needs the
    /// response sorted newest-first. This therefore fixes `direction` to
    /// [`Sort::Desc`] and `limit` to the API maximum of 500, overriding whatever
    /// the filter carried for those two — every other filter is passed through.
    ///
    /// `max_items` bounds the total returned; `None` means every order there is.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_all_orders(
        &self,
        filter: Option<&GetOrdersRequest>,
        max_items: Option<usize>,
    ) -> Result<Vec<Order>> {
        let mut request = filter.cloned().unwrap_or_default();
        request.limit = Some(crate::config::ORDERS_MAX_LIMIT);
        request.direction = Some(Sort::Desc);

        walk_orders(
            &self.rest,
            "/orders",
            request,
            max_items,
            |order: &Order| order.id,
        )
        .await
    }

    /// Fetches one order by its Alpaca id.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_order_by_id(
        &self,
        order_id: Uuid,
        filter: Option<&GetOrderByIdRequest>,
    ) -> Result<Order> {
        let path = format!("/orders/{order_id}");
        match filter {
            Some(filter) => self.rest.get(&path, filter).await,
            None => self.rest.get(&path, &Empty).await,
        }
    }

    /// Fetches one order by the client order id supplied when it was submitted.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_order_by_client_id(&self, client_order_id: &str) -> Result<Order> {
        self.rest
            .get(
                "/orders:by_client_order_id",
                &[("client_order_id", client_order_id)],
            )
            .await
    }

    /// Replaces an open order, returning the new order.
    ///
    /// # Errors
    /// Returns [`crate::Error::InvalidRequest`] if the replacement fails
    /// [`ReplaceOrderRequest::validate`], or an API error if Alpaca rejects it.
    pub async fn replace_order_by_id(
        &self,
        order_id: Uuid,
        replacement: Option<&ReplaceOrderRequest>,
    ) -> Result<Order> {
        let path = format!("/orders/{order_id}");
        match replacement {
            Some(replacement) => {
                replacement.validate()?;
                self.rest.patch(&path, replacement).await
            }
            None => self.rest.patch(&path, &Empty).await,
        }
    }

    /// Cancels every open order, reporting the outcome for each.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn cancel_orders(&self) -> Result<Vec<CancelOrderResponse>> {
        self.rest.delete("/orders", &Empty).await
    }

    /// Cancels one open order.
    ///
    /// # Errors
    /// Propagates transport and API failures.
    pub async fn cancel_order_by_id(&self, order_id: Uuid) -> Result<()> {
        self.send_void(Method::DELETE, &format!("/orders/{order_id}"))
            .await
    }

    // ---------------------------------------------------------- positions

    /// Lists every open position.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_all_positions(&self) -> Result<Vec<Position>> {
        self.rest.get("/positions", &Empty).await
    }

    /// Fetches one open position by symbol or asset id.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_open_position(&self, asset: &AssetIdent) -> Result<Position> {
        self.rest
            .get(&format!("/positions/{}", asset.as_path_segment()?), &Empty)
            .await
    }

    /// Liquidates every open position, reporting the outcome for each.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn close_all_positions(
        &self,
        cancel_orders: Option<bool>,
    ) -> Result<Vec<ClosePositionResponse>> {
        match cancel_orders {
            Some(cancel) => {
                self.rest
                    .delete_effectful("/positions", &[("cancel_orders", cancel)])
                    .await
            }
            None => self.rest.delete_effectful("/positions", &Empty).await,
        }
    }

    /// Liquidates one position, in full or in part.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn close_position(
        &self,
        asset: &AssetIdent,
        close: Option<&ClosePositionRequest>,
    ) -> Result<Order> {
        let path = format!("/positions/{}", asset.as_path_segment()?);
        match close {
            Some(close) => self.rest.delete_effectful(&path, &close.to_query()).await,
            None => self.rest.delete_effectful(&path, &Empty).await,
        }
    }

    /// Exercises a held option contract.
    ///
    /// # Errors
    /// Propagates transport and API failures.
    pub async fn exercise_options_position(&self, contract: &AssetIdent) -> Result<()> {
        self.send_void(
            Method::POST,
            &format!("/positions/{}/exercise", contract.as_path_segment()?),
        )
        .await
    }

    // ------------------------------------------------------------ account

    /// Fetches the trading account.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_account(&self) -> Result<TradeAccount> {
        self.rest.get("/account", &Empty).await
    }

    /// Fetches the account's configuration.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_account_configurations(&self) -> Result<AccountConfiguration> {
        self.rest.get("/account/configurations", &Empty).await
    }

    /// Updates the account's configuration.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn set_account_configurations(
        &self,
        configuration: &AccountConfiguration,
    ) -> Result<AccountConfiguration> {
        self.rest
            .patch("/account/configurations", configuration)
            .await
    }

    /// Fetches the account's value over time.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_portfolio_history(
        &self,
        filter: Option<&GetPortfolioHistoryRequest>,
    ) -> Result<PortfolioHistory> {
        let path = "/account/portfolio/history";
        match filter {
            Some(filter) => self.rest.get(path, filter).await,
            None => self.rest.get(path, &Empty).await,
        }
    }

    /// Lists account activities, optionally filtered.
    ///
    /// The endpoint returns a heterogeneous array; see [`Activity`].
    ///
    /// # Errors
    /// Returns [`crate::Error::InvalidRequest`] if the filter combines
    /// `activity_types` with `category`, which the reference forbids.
    /// Otherwise propagates transport, API, and decoding failures.
    pub async fn get_account_activities(
        &self,
        filter: Option<&GetAccountActivitiesRequest>,
    ) -> Result<Vec<Activity>> {
        match filter {
            Some(filter) => {
                filter.validate()?;
                self.rest.get("/account/activities", filter).await
            }
            None => self.rest.get("/account/activities", &Empty).await,
        }
    }

    // ------------------------------------------------------------- assets

    /// Lists assets, optionally filtered.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_all_assets(&self, filter: Option<&GetAssetsRequest>) -> Result<Vec<Asset>> {
        match filter {
            Some(filter) => self.rest.get("/assets", filter).await,
            None => self.rest.get("/assets", &Empty).await,
        }
    }

    /// Fetches one asset by symbol or id.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_asset(&self, asset: &AssetIdent) -> Result<Asset> {
        self.rest
            .get(&format!("/assets/{}", asset.as_path_segment()?), &Empty)
            .await
    }

    // ------------------------------------------------------ market status

    /// Fetches the current market clock.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_clock(&self) -> Result<Clock> {
        self.rest.get("/clock", &Empty).await
    }

    /// Fetches the market calendar.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_calendar(&self, filter: Option<&GetCalendarRequest>) -> Result<Vec<Calendar>> {
        match filter {
            Some(filter) => self.rest.get("/calendar", filter).await,
            None => self.rest.get("/calendar", &Empty).await,
        }
    }

    // --------------------------------------------------------- watchlists

    /// Lists the account's watchlists.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_watchlists(&self) -> Result<Vec<Watchlist>> {
        self.rest.get("/watchlists", &Empty).await
    }

    /// Fetches one watchlist.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_watchlist_by_id(&self, watchlist_id: Uuid) -> Result<Watchlist> {
        self.rest
            .get(&format!("/watchlists/{watchlist_id}"), &Empty)
            .await
    }

    /// Creates a watchlist.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn create_watchlist(&self, watchlist: &CreateWatchlistRequest) -> Result<Watchlist> {
        self.rest.post("/watchlists", watchlist).await
    }

    /// Updates a watchlist's name, symbols, or both.
    ///
    /// # Errors
    /// Returns [`crate::Error::InvalidRequest`] if neither field is set.
    pub async fn update_watchlist_by_id(
        &self,
        watchlist_id: Uuid,
        update: &UpdateWatchlistRequest,
    ) -> Result<Watchlist> {
        update.validate()?;
        self.rest
            .put(&format!("/watchlists/{watchlist_id}"), update)
            .await
    }

    /// Adds one asset to a watchlist.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn add_asset_to_watchlist_by_id(
        &self,
        watchlist_id: Uuid,
        symbol: &str,
    ) -> Result<Watchlist> {
        self.rest
            .post(
                &format!("/watchlists/{watchlist_id}"),
                &serde_json::json!({ "symbol": symbol }),
            )
            .await
    }

    /// Removes one asset from a watchlist.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn remove_asset_from_watchlist_by_id(
        &self,
        watchlist_id: Uuid,
        symbol: &str,
    ) -> Result<Watchlist> {
        self.rest
            .delete(
                &format!("/watchlists/{watchlist_id}/{}", segment(symbol)?),
                &Empty,
            )
            .await
    }

    /// Deletes a watchlist.
    ///
    /// # Errors
    /// Propagates transport and API failures.
    pub async fn delete_watchlist_by_id(&self, watchlist_id: Uuid) -> Result<()> {
        self.send_void(Method::DELETE, &format!("/watchlists/{watchlist_id}"))
            .await
    }

    // -------------------------------------------------- corporate actions

    /// Searches corporate action announcements.
    ///
    /// # Errors
    /// Returns [`crate::Error::InvalidRequest`] if the date window exceeds 90
    /// days, or an API error if Alpaca rejects the request.
    #[deprecated(
        since = "0.1.0",
        note = "Alpaca deprecated this route; use `CorporateActionsClient::get_corporate_actions` \
                (the /v1/corporate-actions market data route) instead. No sunset date is published"
    )]
    pub async fn get_corporate_announcements(
        &self,
        filter: &GetCorporateAnnouncementsRequest,
    ) -> Result<Vec<CorporateActionAnnouncement>> {
        filter.validate()?;
        self.rest
            .get("/corporate_actions/announcements", &filter.to_query())
            .await
    }

    /// Fetches one corporate action announcement.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    #[deprecated(
        since = "0.1.0",
        note = "Alpaca deprecated this route; use `CorporateActionsClient::get_corporate_actions` \
                (the /v1/corporate-actions market data route) instead. No sunset date is published"
    )]
    pub async fn get_corporate_announcement_by_id(
        &self,
        announcement_id: Uuid,
    ) -> Result<CorporateActionAnnouncement> {
        self.rest
            .get(
                &format!("/corporate_actions/announcements/{announcement_id}"),
                &Empty,
            )
            .await
    }

    // ------------------------------------------------------------ options

    /// Lists option contracts matching the filter.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_option_contracts(
        &self,
        filter: &GetOptionContractsRequest,
    ) -> Result<OptionContractsResponse> {
        self.rest.get("/options/contracts", filter).await
    }

    /// Fetches one option contract by symbol or id.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_option_contract(&self, contract: &AssetIdent) -> Result<OptionContract> {
        self.rest
            .get(
                &format!("/options/contracts/{}", contract.as_path_segment()?),
                &Empty,
            )
            .await
    }

    /// Declines to exercise an in-the-money option position at expiry.
    ///
    /// Answers `204 No Content`, so there is nothing to return.
    ///
    /// # Errors
    /// Propagates transport and API failures.
    pub async fn exercise_do_not_exercise(&self, contract: &AssetIdent) -> Result<()> {
        self.send_void(
            Method::POST,
            &format!("/positions/{}/do-not-exercise", contract.as_path_segment()?),
        )
        .await
    }

    // ------------------------------------------------ activities by type

    /// Account activities of one type.
    ///
    /// The narrowed counterpart to
    /// [`get_account_activities`](Self::get_account_activities): the type moves
    /// from the query string into the path, which is the only way to ask for
    /// exactly one kind.
    ///
    /// The reference documents neither `activity_types` nor `category` here —
    /// the one type being asked for is in the path. Setting either is not
    /// rejected, because the reference states no rule against it and a guess is
    /// not a rule; it is simply undefined what Alpaca does with it.
    ///
    /// # Errors
    /// Returns [`crate::Error::InvalidRequest`] if the filter combines
    /// `activity_types` with `category`. Otherwise propagates transport, API,
    /// and decoding failures.
    pub async fn get_account_activities_by_type(
        &self,
        activity_type: &ActivityType,
        filter: Option<&GetAccountActivitiesRequest>,
    ) -> Result<Vec<Activity>> {
        let path = format!("/account/activities/{}", segment(activity_type)?);
        match filter {
            Some(filter) => {
                filter.validate()?;
                self.rest.get(&path, filter).await
            }
            None => self.rest.get(&path, &Empty).await,
        }
    }

    // ---------------------------------------------------- watchlists by name

    /// Fetches one watchlist by name.
    ///
    /// The name goes in the query string, not the path: the route is literally
    /// `/v2/watchlists:by_name`, colon and all.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_watchlist_by_name(&self, name: &str) -> Result<Watchlist> {
        self.rest
            .get("/watchlists:by_name", &[("name", name)])
            .await
    }

    /// Updates a watchlist by name.
    ///
    /// # Errors
    /// Returns [`crate::Error::InvalidRequest`] if the update changes nothing.
    pub async fn update_watchlist_by_name(
        &self,
        name: &str,
        update: &UpdateWatchlistRequest,
    ) -> Result<Watchlist> {
        update.validate()?;
        self.rest
            .request(
                Method::PUT,
                crate::rest::Replay::ByMethod,
                "/watchlists:by_name",
                Some(&[("name", name)]),
                Some(update),
            )
            .await
    }

    /// Adds one asset to a watchlist by name.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn add_asset_to_watchlist_by_name(
        &self,
        name: &str,
        symbol: &str,
    ) -> Result<Watchlist> {
        self.rest
            .request(
                Method::POST,
                crate::rest::Replay::ByMethod,
                "/watchlists:by_name",
                Some(&[("name", name)]),
                Some(&serde_json::json!({ "symbol": symbol })),
            )
            .await
    }

    /// Deletes a watchlist by name.
    ///
    /// # Errors
    /// Propagates transport and API failures.
    pub async fn delete_watchlist_by_name(&self, name: &str) -> Result<()> {
        self.rest
            .delete("/watchlists:by_name", &[("name", name)])
            .await
    }

    // ---------------------------------------------------------- calendar

    /// A named market's calendar.
    ///
    /// **A `v3` route**, where [`get_calendar`](Self::get_calendar) is `v2` and
    /// the broker's equivalent is `v2` again. It answers with pre-market, core,
    /// lunch and post-market windows as absolute instants rather than with the
    /// naive eastern-time open and close [`Calendar`] carries.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_market_calendar(
        &self,
        market: &Market,
        filter: Option<&GetMarketCalendarRequest>,
    ) -> Result<MarketCalendar> {
        let path = format!("/calendar/{}", segment(market)?);
        match filter {
            Some(filter) => self.rest.at_version("v3").get(&path, filter).await,
            None => self.rest.at_version("v3").get(&path, &Empty).await,
        }
    }

    // ----------------------------------------------------------- locates

    /// Lists locates, optionally filtered.
    ///
    /// **A `v1` route**, unlike the rest of this client.
    ///
    /// Returns one page. The response carries a `next_page_token`; pass it back
    /// in [`GetLocatesRequest::page_token`] for the next.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_locates(&self, filter: Option<&GetLocatesRequest>) -> Result<LocatesPage> {
        match filter {
            Some(filter) => self.rest.at_version("v1").get("/locates", filter).await,
            None => self.rest.at_version("v1").get("/locates", &Empty).await,
        }
    }

    /// Fetches one locate.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_locate_by_id(&self, locate_id: Uuid) -> Result<Locate> {
        self.rest
            .at_version("v1")
            .get(&format!("/locates/{locate_id}"), &Empty)
            .await
    }

    /// What the named symbols currently cost to borrow.
    ///
    /// Partial success is normal: a symbol that is easy to borrow needs no
    /// locate and comes back under
    /// [`errors`](crate::trading::LocateQuotes::errors) rather than as a failed request.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_locate_quotes(
        &self,
        request: &GetLocateQuotesRequest,
    ) -> Result<LocateQuotes> {
        self.rest
            .at_version("v1")
            .get("/locates/quotes", request)
            .await
    }

    /// Requests a locate.
    ///
    /// # Errors
    /// Returns [`crate::Error::InvalidRequest`] if the request contradicts
    /// itself; see [`CreateLocateRequest::validate`].
    pub async fn create_locate(&self, request: &CreateLocateRequest) -> Result<Locate> {
        request.validate()?;
        self.rest.at_version("v1").post("/locates", request).await
    }

    // ------------------------------------------------------ tokenization

    /// Mints a tokenized asset from a position.
    ///
    /// # Errors
    /// Returns [`crate::Error::InvalidRequest`] if `qty` is not positive.
    pub async fn mint_token(&self, request: &MintTokenRequest) -> Result<TokenizationRequest> {
        request.validate()?;
        self.rest.post("/tokenization/mint", request).await
    }

    /// Lists tokenization requests.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_tokenization_requests(
        &self,
        filter: Option<&GetTokenizationRequestsRequest>,
    ) -> Result<Vec<TokenizationRequest>> {
        match filter {
            Some(filter) => self.rest.get("/tokenization/requests", filter).await,
            None => self.rest.get("/tokenization/requests", &Empty).await,
        }
    }

    /// Fetches one tokenization request.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_tokenization_request(&self, request_id: Uuid) -> Result<TokenizationRequest> {
        self.rest
            .get(&format!("/tokenization/requests/{request_id}"), &Empty)
            .await
    }

    /// Fetches one tokenization request by the caller's own id.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_tokenization_request_by_client_id(
        &self,
        request: &ByClientRequestId,
    ) -> Result<TokenizationRequest> {
        self.rest
            .get("/tokenization/requests:by_client_request_id", request)
            .await
    }

    // ----------------------------------------------------- crypto funding

    /// The account's crypto deposit wallets.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_crypto_wallets(
        &self,
        filter: Option<&GetCryptoWalletsRequest>,
    ) -> Result<Vec<CryptoWallet>> {
        let wallets: OneOrMany<CryptoWallet> = match filter {
            Some(filter) => self.rest.get("/wallets", filter).await?,
            None => self.rest.get("/wallets", &Empty).await?,
        };
        Ok(wallets.into_vec())
    }

    /// The account's on-chain transfers.
    ///
    /// The withdrawal route that would create one is deliberately absent: it is
    /// deprecated with a sunset of 2026-10-09 and the reference's replacement
    /// is the Alpaca web application, not another endpoint. The broker API's
    /// equivalent withdrawal is neither deprecated nor absent.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_crypto_transfers(&self) -> Result<Vec<CryptoTransfer>> {
        let transfers: OneOrMany<CryptoTransfer> =
            self.rest.get("/wallets/transfers", &Empty).await?;
        Ok(transfers.into_vec())
    }

    /// Fetches one on-chain transfer.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_crypto_transfer(&self, transfer_id: &str) -> Result<CryptoTransfer> {
        self.rest
            .get(
                &format!("/wallets/transfers/{}", segment(transfer_id)?),
                &Empty,
            )
            .await
    }

    /// The addresses withdrawals may be sent to.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn get_whitelisted_addresses(&self) -> Result<Vec<WhitelistedAddress>> {
        let addresses: OneOrMany<WhitelistedAddress> =
            self.rest.get("/wallets/whitelists", &Empty).await?;
        Ok(addresses.into_vec())
    }

    /// Allowlists a withdrawal address.
    ///
    /// New entries land [`Pending`](crate::trading::WhitelistStatus::Pending) and become usable
    /// after Alpaca's cooling-off period.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn create_whitelisted_address(
        &self,
        request: &CreateWhitelistedAddressRequest,
    ) -> Result<WhitelistedAddress> {
        self.rest.post("/wallets/whitelists", request).await
    }

    /// Removes an allowlisted withdrawal address.
    ///
    /// # Errors
    /// Propagates transport and API failures.
    pub async fn delete_whitelisted_address(&self, address_id: &str) -> Result<()> {
        self.rest
            .delete(
                &format!("/wallets/whitelists/{}", segment(address_id)?),
                &Empty,
            )
            .await
    }

    /// What a proposed transfer would cost in gas.
    ///
    /// # Errors
    /// Propagates transport, API, and decoding failures.
    pub async fn estimate_transfer_fee(
        &self,
        request: &TransferFeeEstimateRequest,
    ) -> Result<TransferFeeEstimate> {
        self.rest.get("/wallets/fees/estimate", request).await
    }

    // ------------------------------------------------------------- events

    /// Streams account activity events as they happen.
    ///
    /// A `v2beta1` route, and the push counterpart to
    /// [`get_account_activities`](Self::get_account_activities). The broker API
    /// carries the same stream; this is the one a trading-only build can reach.
    ///
    /// # Errors
    /// Propagates transport failures and any non-success status the server
    /// answers the subscription with.
    pub async fn get_activity_events(
        &self,
        filter: Option<&EventStreamRequest>,
    ) -> Result<impl futures_util::Stream<Item = Result<crate::sse::Event>> + use<>> {
        let path = "/events/activities";
        let url = format!(
            "{}/v2beta1{path}",
            self.rest.config().base_url.trim_end_matches('/')
        );
        let query = filter.map(EventStreamRequest::query).unwrap_or_default();
        crate::sse::subscribe(&self.raw, &url, path, &query).await
    }
}

/// Walks `/orders` (or its broker twin) with the `before_order_id` cursor.
///
/// Shared by [`TradingClient::get_all_orders`] and
/// [`BrokerClient::get_all_orders_for_account`](crate::broker::BrokerClient::get_all_orders_for_account):
/// the two routes take the same request type and the same page cap, so a second
/// copy of this loop would be a second place for the cursor logic to drift.
///
/// `request` is expected to arrive with `limit` and `direction` already set —
/// the walk needs newest-first ordering to page backwards.
pub(crate) async fn walk_orders<T, F>(
    rest: &RestClient,
    path: &str,
    mut request: GetOrdersRequest,
    max_items: Option<usize>,
    id_of: F,
) -> Result<Vec<T>>
where
    T: serde::de::DeserializeOwned,
    F: Fn(&T) -> Uuid,
{
    /// Narrows the page size to what is still outstanding.
    ///
    /// Per page, not once: capping only the first request still pulled 1,000
    /// orders to satisfy `max_items: Some(600)`.
    fn narrow(request: &mut GetOrdersRequest, max_items: Option<usize>, so_far: usize) {
        let Some(max) = max_items else { return };
        let Ok(remaining) = u32::try_from(max.saturating_sub(so_far)) else {
            return;
        };
        request.limit = Some(request.limit.unwrap_or(u32::MAX).min(remaining));
    }

    // Alpaca's cursors are inclusive on some routes and exclusive on others,
    // and the reference does not say which this one is — so a page boundary is
    // deduplicated rather than trusted. If it turns out inclusive, without this
    // every page would silently repeat one order.
    let mut seen: std::collections::HashSet<Uuid> = std::collections::HashSet::new();
    let mut all: Vec<T> = Vec::new();

    loop {
        if max_items == Some(0) {
            break;
        }
        narrow(&mut request, max_items, all.len());

        let page = rest.get::<Vec<T>, _>(path, &request).await?;

        // An *empty* page ends the walk, not a short one. A short page looks
        // like the end, and usually is — but it is also what a server that
        // silently caps `limit` below what was asked returns, and this route's
        // documented default page size is 50 against the 500 we ask for. Ending
        // on short would then reproduce the truncation this walk exists to fix,
        // with no error. Ending on empty costs one extra request, which is the
        // price of an endpoint that does not say whether more remain.
        if page.is_empty() {
            break;
        }

        // The cursor is the oldest id on this page, and it has to be read
        // before the page is moved into the accumulator.
        let oldest = page.last().map(&id_of);
        let before = all.len();
        for order in page {
            if seen.insert(id_of(&order)) {
                all.push(order);
            }
        }

        if let Some(max) = max_items
            && all.len() >= max
        {
            all.truncate(max);
            break;
        }

        // Progress is measured in *new* orders, not in cursor movement. A guard
        // that only compared consecutive cursors would miss a server cycling
        // between two pages — the walk would never terminate even though it had
        // stopped learning anything. `seen` already knows the answer.
        if all.len() == before {
            tracing::warn!(
                path,
                "order walk stopped making progress; every order on the last \
                 page had already been seen"
            );
            break;
        }

        let Some(oldest) = oldest else { break };
        request.before_order_id = Some(oldest);
    }

    Ok(all)
}