Skip to content

Portfolio

Portfolio Resource

composer.resources.portfolio.Portfolio

Resource for portfolio and holdings endpoints.

Source code in composer/resources/portfolio.py
class Portfolio:
    """Resource for portfolio and holdings endpoints."""

    def __init__(self, http_client):
        self._client = http_client

    def get_account_holdings(
        self,
        account_id: str,
        position_type: str | None = None,
    ) -> list[Holding]:
        """
        Get all current positions held in the specified account.

        Args:
            account_id: The account UUID
            position_type: Optional filter - 'direct', 'symphony', or 'all'

        Returns
        -------
            List of holdings in the account
        """
        params = {}
        if position_type:
            position_type_map = {
                "direct": "DEFAULT_DIRECT",
                "symphony": "SYMPHONY",
                "all": "ALL",
            }
            params["position_type"] = position_type_map.get(
                position_type.lower(), position_type.upper()
            )
        response = self._client.get(
            f"/api/v1/accounts/{account_id}/holdings",
            params=params if params else None,
        )
        return [Holding.model_validate(h) for h in response]

    def get_total_stats(self, account_id: str) -> TotalStats:
        """
        Get aggregate portfolio statistics for an account.

        Args:
            account_id: The account UUID

        Returns
        -------
            Total portfolio statistics including value, returns, cash, etc.
        """
        response = self._client.get(
            f"/api/v1/portfolio/accounts/{account_id}/total-stats",
        )
        return TotalStats.model_validate(response)

    def get_symphony_stats(self, account_id: str) -> SymphonyStatsResponse:
        """
        Get aggregate stats per symphony for an account.

        Args:
            account_id: The account UUID

        Returns
        -------
            Stats for each symphony in the account
        """
        response = self._client.get(
            f"/api/v1/portfolio/accounts/{account_id}/symphony-stats",
        )
        return SymphonyStatsResponse.model_validate(response)

    def get_symphony_stats_meta(self, account_id: str) -> SymphonyStatsMetaResponse:
        """
        Get aggregate stats per symphony with metadata for an account.

        Args:
            account_id: The account UUID

        Returns
        -------
            Stats with metadata for each symphony in the account
        """
        response = self._client.get(
            f"/api/v1/portfolio/accounts/{account_id}/symphony-stats-meta",
        )
        return SymphonyStatsMetaResponse.model_validate(response)

    def get_portfolio_history(self, account_id: str) -> PortfolioHistory:
        """
        Get the value of the account portfolio over time.

        Args:
            account_id: The account UUID

        Returns
        -------
            Portfolio value history with timestamps and values
        """
        response = self._client.get(
            f"/api/v1/portfolio/accounts/{account_id}/portfolio-history",
        )
        return PortfolioHistory.model_validate(response)

    def get_symphony_value_history(
        self,
        account_id: str,
        symphony_id: str,
    ) -> TimeSeries:
        """
        Get the value of a symphony position over time.

        Args:
            account_id: The account UUID
            symphony_id: The symphony UUID

        Returns
        -------
            Symphony value history including deposit-adjusted series
        """
        response = self._client.get(
            f"/api/v1/portfolio/accounts/{account_id}/symphonies/{symphony_id}",
        )
        return TimeSeries.model_validate(response)

    def get_symphony_holdings(
        self,
        account_id: str,
        symphony_id: str,
    ) -> SymphonyHoldings:
        """
        Get the current holdings of a symphony.

        Args:
            account_id: The account UUID
            symphony_id: The symphony UUID

        Returns
        -------
            Current holdings including cash, shares, etc.
        """
        response = self._client.get(
            f"/api/v1/portfolio/accounts/{account_id}/symphonies/{symphony_id}/holdings",
        )
        print(response)
        return SymphonyHoldings.model_validate(response)

    def get_holdings_by_position(self, position_id: str) -> SymphonyHoldings:
        """
        Get the current holdings of a symphony from position ID.

        Args:
            position_id: The position UUID

        Returns
        -------
            Current holdings including cash, shares, etc.
        """
        response = self._client.get(
            f"/api/v1/portfolio/positions/{position_id}/holdings",
        )
        return SymphonyHoldings.model_validate(response)

    def get_activity_history(
        self,
        account_id: str,
        symphony_id: str,
        limit: int,
        offset: int,
        start_date: str | None = None,
        end_date: str | None = None,
    ) -> ActivityHistoryResponse:
        """
        Get the activity history for a symphony.

        Activity types include: edit, invest, withdraw, rebalance, etc.

        Args:
            account_id: The account UUID
            symphony_id: The symphony UUID
            limit: Maximum number of results to return
            offset: Offset for pagination
            start_date: Optional start date filter (ISO format)
            end_date: Optional end date filter (ISO format)

        Returns
        -------
            Activity history for the symphony
        """
        params = {
            "start_date": start_date,
            "end_date": end_date,
            "limit": limit,
            "offset": offset,
        }
        params = {k: v for k, v in params.items() if v is not None}
        response = self._client.get(
            f"/api/v1/portfolio/accounts/{account_id}/symphonies/{symphony_id}/activity-history",
            params=params,
        )
        return ActivityHistoryResponse.model_validate(response)

    def get_deploy_details(
        self,
        account_id: str,
        symphony_id: str,
        deploy_id: str,
    ) -> DeployHistoryResponse:
        """
        Get the details about a deploy from the perspective of the symphony.

        Args:
            account_id: The account UUID
            symphony_id: The symphony UUID
            deploy_id: The deploy UUID

        Returns
        -------
            Deploy history including parent deploys
        """
        response = self._client.get(
            f"/api/v1/portfolio/accounts/{account_id}/symphonies/{symphony_id}/deploys/{deploy_id}",
        )
        return DeployHistoryResponse.model_validate(response)

    def get_holding_stats(self, account_id: str) -> HoldingStatsResponse:
        """
        Get holding statistics for an account.

        Returns the direct and symphony allocations, amounts, and values by holding.
        Includes current prices, daily change percentages, and notional values.

        Args:
            account_id: The account UUID

        Returns
        -------
            Detailed holding statistics
        """
        response = self._client.get(
            f"/api/v1/portfolio/accounts/{account_id}/holding-stats",
        )
        return HoldingStatsResponse.model_validate(response)

get_account_holdings(account_id, position_type=None)

Get all current positions held in the specified account.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
position_type str | None

Optional filter - 'direct', 'symphony', or 'all'

None
Returns
List of holdings in the account
Source code in composer/resources/portfolio.py
def get_account_holdings(
    self,
    account_id: str,
    position_type: str | None = None,
) -> list[Holding]:
    """
    Get all current positions held in the specified account.

    Args:
        account_id: The account UUID
        position_type: Optional filter - 'direct', 'symphony', or 'all'

    Returns
    -------
        List of holdings in the account
    """
    params = {}
    if position_type:
        position_type_map = {
            "direct": "DEFAULT_DIRECT",
            "symphony": "SYMPHONY",
            "all": "ALL",
        }
        params["position_type"] = position_type_map.get(
            position_type.lower(), position_type.upper()
        )
    response = self._client.get(
        f"/api/v1/accounts/{account_id}/holdings",
        params=params if params else None,
    )
    return [Holding.model_validate(h) for h in response]

get_activity_history(account_id, symphony_id, limit, offset, start_date=None, end_date=None)

Get the activity history for a symphony.

Activity types include: edit, invest, withdraw, rebalance, etc.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
symphony_id str

The symphony UUID

required
limit int

Maximum number of results to return

required
offset int

Offset for pagination

required
start_date str | None

Optional start date filter (ISO format)

None
end_date str | None

Optional end date filter (ISO format)

None
Returns
Activity history for the symphony
Source code in composer/resources/portfolio.py
def get_activity_history(
    self,
    account_id: str,
    symphony_id: str,
    limit: int,
    offset: int,
    start_date: str | None = None,
    end_date: str | None = None,
) -> ActivityHistoryResponse:
    """
    Get the activity history for a symphony.

    Activity types include: edit, invest, withdraw, rebalance, etc.

    Args:
        account_id: The account UUID
        symphony_id: The symphony UUID
        limit: Maximum number of results to return
        offset: Offset for pagination
        start_date: Optional start date filter (ISO format)
        end_date: Optional end date filter (ISO format)

    Returns
    -------
        Activity history for the symphony
    """
    params = {
        "start_date": start_date,
        "end_date": end_date,
        "limit": limit,
        "offset": offset,
    }
    params = {k: v for k, v in params.items() if v is not None}
    response = self._client.get(
        f"/api/v1/portfolio/accounts/{account_id}/symphonies/{symphony_id}/activity-history",
        params=params,
    )
    return ActivityHistoryResponse.model_validate(response)

get_deploy_details(account_id, symphony_id, deploy_id)

Get the details about a deploy from the perspective of the symphony.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
symphony_id str

The symphony UUID

required
deploy_id str

The deploy UUID

required
Returns
Deploy history including parent deploys
Source code in composer/resources/portfolio.py
def get_deploy_details(
    self,
    account_id: str,
    symphony_id: str,
    deploy_id: str,
) -> DeployHistoryResponse:
    """
    Get the details about a deploy from the perspective of the symphony.

    Args:
        account_id: The account UUID
        symphony_id: The symphony UUID
        deploy_id: The deploy UUID

    Returns
    -------
        Deploy history including parent deploys
    """
    response = self._client.get(
        f"/api/v1/portfolio/accounts/{account_id}/symphonies/{symphony_id}/deploys/{deploy_id}",
    )
    return DeployHistoryResponse.model_validate(response)

get_holding_stats(account_id)

Get holding statistics for an account.

Returns the direct and symphony allocations, amounts, and values by holding. Includes current prices, daily change percentages, and notional values.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
Returns
Detailed holding statistics
Source code in composer/resources/portfolio.py
def get_holding_stats(self, account_id: str) -> HoldingStatsResponse:
    """
    Get holding statistics for an account.

    Returns the direct and symphony allocations, amounts, and values by holding.
    Includes current prices, daily change percentages, and notional values.

    Args:
        account_id: The account UUID

    Returns
    -------
        Detailed holding statistics
    """
    response = self._client.get(
        f"/api/v1/portfolio/accounts/{account_id}/holding-stats",
    )
    return HoldingStatsResponse.model_validate(response)

get_holdings_by_position(position_id)

Get the current holdings of a symphony from position ID.

Parameters:

Name Type Description Default
position_id str

The position UUID

required
Returns
Current holdings including cash, shares, etc.
Source code in composer/resources/portfolio.py
def get_holdings_by_position(self, position_id: str) -> SymphonyHoldings:
    """
    Get the current holdings of a symphony from position ID.

    Args:
        position_id: The position UUID

    Returns
    -------
        Current holdings including cash, shares, etc.
    """
    response = self._client.get(
        f"/api/v1/portfolio/positions/{position_id}/holdings",
    )
    return SymphonyHoldings.model_validate(response)

get_portfolio_history(account_id)

Get the value of the account portfolio over time.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
Returns
Portfolio value history with timestamps and values
Source code in composer/resources/portfolio.py
def get_portfolio_history(self, account_id: str) -> PortfolioHistory:
    """
    Get the value of the account portfolio over time.

    Args:
        account_id: The account UUID

    Returns
    -------
        Portfolio value history with timestamps and values
    """
    response = self._client.get(
        f"/api/v1/portfolio/accounts/{account_id}/portfolio-history",
    )
    return PortfolioHistory.model_validate(response)

get_symphony_holdings(account_id, symphony_id)

Get the current holdings of a symphony.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
symphony_id str

The symphony UUID

required
Returns
Current holdings including cash, shares, etc.
Source code in composer/resources/portfolio.py
def get_symphony_holdings(
    self,
    account_id: str,
    symphony_id: str,
) -> SymphonyHoldings:
    """
    Get the current holdings of a symphony.

    Args:
        account_id: The account UUID
        symphony_id: The symphony UUID

    Returns
    -------
        Current holdings including cash, shares, etc.
    """
    response = self._client.get(
        f"/api/v1/portfolio/accounts/{account_id}/symphonies/{symphony_id}/holdings",
    )
    print(response)
    return SymphonyHoldings.model_validate(response)

get_symphony_stats(account_id)

Get aggregate stats per symphony for an account.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
Returns
Stats for each symphony in the account
Source code in composer/resources/portfolio.py
def get_symphony_stats(self, account_id: str) -> SymphonyStatsResponse:
    """
    Get aggregate stats per symphony for an account.

    Args:
        account_id: The account UUID

    Returns
    -------
        Stats for each symphony in the account
    """
    response = self._client.get(
        f"/api/v1/portfolio/accounts/{account_id}/symphony-stats",
    )
    return SymphonyStatsResponse.model_validate(response)

get_symphony_stats_meta(account_id)

Get aggregate stats per symphony with metadata for an account.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
Returns
Stats with metadata for each symphony in the account
Source code in composer/resources/portfolio.py
def get_symphony_stats_meta(self, account_id: str) -> SymphonyStatsMetaResponse:
    """
    Get aggregate stats per symphony with metadata for an account.

    Args:
        account_id: The account UUID

    Returns
    -------
        Stats with metadata for each symphony in the account
    """
    response = self._client.get(
        f"/api/v1/portfolio/accounts/{account_id}/symphony-stats-meta",
    )
    return SymphonyStatsMetaResponse.model_validate(response)

get_symphony_value_history(account_id, symphony_id)

Get the value of a symphony position over time.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
symphony_id str

The symphony UUID

required
Returns
Symphony value history including deposit-adjusted series
Source code in composer/resources/portfolio.py
def get_symphony_value_history(
    self,
    account_id: str,
    symphony_id: str,
) -> TimeSeries:
    """
    Get the value of a symphony position over time.

    Args:
        account_id: The account UUID
        symphony_id: The symphony UUID

    Returns
    -------
        Symphony value history including deposit-adjusted series
    """
    response = self._client.get(
        f"/api/v1/portfolio/accounts/{account_id}/symphonies/{symphony_id}",
    )
    return TimeSeries.model_validate(response)

get_total_stats(account_id)

Get aggregate portfolio statistics for an account.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
Returns
Total portfolio statistics including value, returns, cash, etc.
Source code in composer/resources/portfolio.py
def get_total_stats(self, account_id: str) -> TotalStats:
    """
    Get aggregate portfolio statistics for an account.

    Args:
        account_id: The account UUID

    Returns
    -------
        Total portfolio statistics including value, returns, cash, etc.
    """
    response = self._client.get(
        f"/api/v1/portfolio/accounts/{account_id}/total-stats",
    )
    return TotalStats.model_validate(response)

Accounts Resource

composer.resources.accounts.Accounts

Resource for account-related endpoints.

Source code in composer/resources/accounts.py
class Accounts:
    """Resource for account-related endpoints."""

    def __init__(self, http_client):
        self._client = http_client

    def list(self) -> AccountsListResponse:
        """
        Get a list of all accounts associated with the authenticated user.

        Returns
        -------
            List of accounts with status, asset classes, and key dates
        """
        response = self._client.get("/api/v1/accounts/list")
        return AccountsListResponse.model_validate(response)

    def get_holdings(
        self,
        account_id: str,
        position_type: str | None = None,
    ) -> builtins.list[Holding]:
        """
        Get all current positions held in the specified account.

        Args:
            account_id: The account UUID
            position_type: Optional filter - 'direct', 'symphony', or 'all'

        Returns
        -------
            List of holdings in the account
        """
        params = {}
        if position_type:
            position_type_map = {
                "direct": "DEFAULT_DIRECT",
                "symphony": "SYMPHONY",
                "all": "ALL",
            }
            params["position_type"] = position_type_map.get(
                position_type.lower(), position_type.upper()
            )
        response = self._client.get(
            f"/api/v1/accounts/{account_id}/holdings",
            params=params if params else None,
        )
        return [Holding.model_validate(h) for h in response]

    def get_available_types(
        self,
        country: str | None = None,
        state: str | None = None,
        white_label_footprint: bool | None = None,
    ) -> dict[str, builtins.list[AvailableAccountType]]:
        """
        Get available account types for a user to create.

        Args:
            country: Country code
            state: State code
            white_label_footprint: Whether to use white label footprint

        Returns
        -------
            Map of region to available account types
        """
        params = {}
        if country:
            params["country"] = country
        if state:
            params["state"] = state
        if white_label_footprint is not None:
            params["white_label_footprint"] = white_label_footprint
        response = self._client.get(
            "/api/v1/accounts/available-types/v2",
            params=params if params else None,
        )
        return {
            region: [AvailableAccountType.model_validate(at) for at in types]
            for region, types in response.items()
        }

    def get_supported_regions(self) -> SupportedRegionsResponse:
        """
        Get a list of countries and states that support equity/crypto trading.

        Returns
        -------
            Supported regions for EQUITIES and CRYPTO trading
        """
        response = self._client.get("/api/v1/accounts/supported-regions")
        return SupportedRegionsResponse.model_validate(response)

    def get_activities_trades(
        self,
        account_id: str,
        asset_class: str | None = None,
        limit: int | None = None,
        offset: int | None = None,
        order_request_id: str | None = None,
    ) -> builtins.list[Trade]:
        """
        Get record of historical trades.

        Args:
            account_id: The account UUID
            asset_class: Filter by asset class (CRYPTO, EQUITIES, OPTIONS)
            limit: Maximum number of results
            offset: Pagination offset
            order_request_id: Filter by order request ID

        Returns
        -------
            List of historical trades
        """
        params = {}
        if asset_class:
            params["asset_class"] = asset_class
        if limit is not None:
            params["limit"] = limit
        if offset is not None:
            params["offset"] = offset
        if order_request_id:
            params["order_request_id"] = order_request_id
        response = self._client.get(
            f"/api/v1/accounts/{account_id}/activities/trades",
            params=params if params else None,
        )
        return [Trade.model_validate(t) for t in response]

    def get_activities_trades_volume(
        self,
        account_id: str,
        start_time: str,
        asset_class: str | None = None,
        direct_trading_only: bool | None = None,
    ) -> TradeVolumeResponse:
        """
        Get trade volume for the account.

        Args:
            account_id: The account UUID
            start_time: Start time for the volume calculation (ISO format with
                timezone, e.g., "2025-12-10T09:00:00-05:00")
            asset_class: Filter by asset class (CRYPTO, EQUITIES, OPTIONS)
            direct_trading_only: Only include direct trading volume

        Returns
        -------
            Trade volume information
        """
        params: dict[str, Any] = {"start_time": start_time}
        if asset_class:
            params["asset_class"] = asset_class
        if direct_trading_only is not None:
            params["direct_trading_only"] = direct_trading_only
        response = self._client.get(
            f"/api/v1/accounts/{account_id}/activities/trades/volume",
            params=params,
        )
        return TradeVolumeResponse.model_validate(response)

    def get_activities_trade_history(
        self,
        account_id: str,
        asset_class: str | None = None,
        limit: int | None = None,
        offset: int | None = None,
        order_request_id: str | None = None,
    ) -> builtins.list[TradeHistoryItem]:
        """
        Get record of historical trades and option events.

        Args:
            account_id: The account UUID
            asset_class: Filter by asset class (CRYPTO, EQUITIES, OPTIONS)
            limit: Maximum number of results
            offset: Pagination offset
            order_request_id: Filter by order request ID

        Returns
        -------
            List of trade history items including option events
        """
        params = {}
        if asset_class:
            params["asset_class"] = asset_class
        if limit is not None:
            params["limit"] = limit
        if offset is not None:
            params["offset"] = offset
        if order_request_id:
            params["order_request_id"] = order_request_id
        response = self._client.get(
            f"/api/v1/accounts/{account_id}/activities/trade-history",
            params=params if params else None,
        )
        return [TradeHistoryItem.model_validate(t) for t in response]

    def get_info(self, account_id: str) -> AccountInfo:
        """
        Get basic account owner information.

        Args:
            account_id: The account UUID

        Returns
        -------
            Account owner information including identity, contact, and profile
        """
        response = self._client.get(f"/api/v1/accounts/{account_id}/info")
        return AccountInfo.model_validate(response)

    def get_buying_power(self, account_id: str) -> builtins.list[BuyingPower]:
        """
        Get account buying power by asset class.

        Args:
            account_id: The account UUID

        Returns
        -------
            List of buying power info for each asset class
        """
        response = self._client.get(f"/api/v1/accounts/{account_id}/info/buying-power")
        return [BuyingPower.model_validate(bp) for bp in response]

    def get_investor_documents(
        self,
        account_id: str,
        category: str,
        year: int,
    ) -> builtins.list[InvestorDocument]:
        """
        Get investor documents for Apex and Alpaca accounts.

        Args:
            account_id: The account UUID
            category: Document category (STATEMENT, TAX_FORM, etc.)
            year: Year of documents to retrieve

        Returns
        -------
            List of investor documents
        """
        params = {"category": category, "year": year}
        response = self._client.get(
            f"/api/v1/accounts/{account_id}/info/investor-documents",
            params=params,
        )
        return [InvestorDocument.model_validate(d) for d in response]

    def download_document(self, document_id: str) -> str:
        """
        Download a single document.

        Args:
            document_id: The document ID to download

        Returns
        -------
            URL to download the document (redirect)
        """
        response = self._client.get(f"/api/v1/accounts/documents/{document_id}/download")
        return response

download_document(document_id)

Download a single document.

Parameters:

Name Type Description Default
document_id str

The document ID to download

required
Returns
URL to download the document (redirect)
Source code in composer/resources/accounts.py
def download_document(self, document_id: str) -> str:
    """
    Download a single document.

    Args:
        document_id: The document ID to download

    Returns
    -------
        URL to download the document (redirect)
    """
    response = self._client.get(f"/api/v1/accounts/documents/{document_id}/download")
    return response

get_activities_trade_history(account_id, asset_class=None, limit=None, offset=None, order_request_id=None)

Get record of historical trades and option events.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
asset_class str | None

Filter by asset class (CRYPTO, EQUITIES, OPTIONS)

None
limit int | None

Maximum number of results

None
offset int | None

Pagination offset

None
order_request_id str | None

Filter by order request ID

None
Returns
List of trade history items including option events
Source code in composer/resources/accounts.py
def get_activities_trade_history(
    self,
    account_id: str,
    asset_class: str | None = None,
    limit: int | None = None,
    offset: int | None = None,
    order_request_id: str | None = None,
) -> builtins.list[TradeHistoryItem]:
    """
    Get record of historical trades and option events.

    Args:
        account_id: The account UUID
        asset_class: Filter by asset class (CRYPTO, EQUITIES, OPTIONS)
        limit: Maximum number of results
        offset: Pagination offset
        order_request_id: Filter by order request ID

    Returns
    -------
        List of trade history items including option events
    """
    params = {}
    if asset_class:
        params["asset_class"] = asset_class
    if limit is not None:
        params["limit"] = limit
    if offset is not None:
        params["offset"] = offset
    if order_request_id:
        params["order_request_id"] = order_request_id
    response = self._client.get(
        f"/api/v1/accounts/{account_id}/activities/trade-history",
        params=params if params else None,
    )
    return [TradeHistoryItem.model_validate(t) for t in response]

get_activities_trades(account_id, asset_class=None, limit=None, offset=None, order_request_id=None)

Get record of historical trades.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
asset_class str | None

Filter by asset class (CRYPTO, EQUITIES, OPTIONS)

None
limit int | None

Maximum number of results

None
offset int | None

Pagination offset

None
order_request_id str | None

Filter by order request ID

None
Returns
List of historical trades
Source code in composer/resources/accounts.py
def get_activities_trades(
    self,
    account_id: str,
    asset_class: str | None = None,
    limit: int | None = None,
    offset: int | None = None,
    order_request_id: str | None = None,
) -> builtins.list[Trade]:
    """
    Get record of historical trades.

    Args:
        account_id: The account UUID
        asset_class: Filter by asset class (CRYPTO, EQUITIES, OPTIONS)
        limit: Maximum number of results
        offset: Pagination offset
        order_request_id: Filter by order request ID

    Returns
    -------
        List of historical trades
    """
    params = {}
    if asset_class:
        params["asset_class"] = asset_class
    if limit is not None:
        params["limit"] = limit
    if offset is not None:
        params["offset"] = offset
    if order_request_id:
        params["order_request_id"] = order_request_id
    response = self._client.get(
        f"/api/v1/accounts/{account_id}/activities/trades",
        params=params if params else None,
    )
    return [Trade.model_validate(t) for t in response]

get_activities_trades_volume(account_id, start_time, asset_class=None, direct_trading_only=None)

Get trade volume for the account.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
start_time str

Start time for the volume calculation (ISO format with timezone, e.g., "2025-12-10T09:00:00-05:00")

required
asset_class str | None

Filter by asset class (CRYPTO, EQUITIES, OPTIONS)

None
direct_trading_only bool | None

Only include direct trading volume

None
Returns
Trade volume information
Source code in composer/resources/accounts.py
def get_activities_trades_volume(
    self,
    account_id: str,
    start_time: str,
    asset_class: str | None = None,
    direct_trading_only: bool | None = None,
) -> TradeVolumeResponse:
    """
    Get trade volume for the account.

    Args:
        account_id: The account UUID
        start_time: Start time for the volume calculation (ISO format with
            timezone, e.g., "2025-12-10T09:00:00-05:00")
        asset_class: Filter by asset class (CRYPTO, EQUITIES, OPTIONS)
        direct_trading_only: Only include direct trading volume

    Returns
    -------
        Trade volume information
    """
    params: dict[str, Any] = {"start_time": start_time}
    if asset_class:
        params["asset_class"] = asset_class
    if direct_trading_only is not None:
        params["direct_trading_only"] = direct_trading_only
    response = self._client.get(
        f"/api/v1/accounts/{account_id}/activities/trades/volume",
        params=params,
    )
    return TradeVolumeResponse.model_validate(response)

get_available_types(country=None, state=None, white_label_footprint=None)

Get available account types for a user to create.

Parameters:

Name Type Description Default
country str | None

Country code

None
state str | None

State code

None
white_label_footprint bool | None

Whether to use white label footprint

None
Returns
Map of region to available account types
Source code in composer/resources/accounts.py
def get_available_types(
    self,
    country: str | None = None,
    state: str | None = None,
    white_label_footprint: bool | None = None,
) -> dict[str, builtins.list[AvailableAccountType]]:
    """
    Get available account types for a user to create.

    Args:
        country: Country code
        state: State code
        white_label_footprint: Whether to use white label footprint

    Returns
    -------
        Map of region to available account types
    """
    params = {}
    if country:
        params["country"] = country
    if state:
        params["state"] = state
    if white_label_footprint is not None:
        params["white_label_footprint"] = white_label_footprint
    response = self._client.get(
        "/api/v1/accounts/available-types/v2",
        params=params if params else None,
    )
    return {
        region: [AvailableAccountType.model_validate(at) for at in types]
        for region, types in response.items()
    }

get_buying_power(account_id)

Get account buying power by asset class.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
Returns
List of buying power info for each asset class
Source code in composer/resources/accounts.py
def get_buying_power(self, account_id: str) -> builtins.list[BuyingPower]:
    """
    Get account buying power by asset class.

    Args:
        account_id: The account UUID

    Returns
    -------
        List of buying power info for each asset class
    """
    response = self._client.get(f"/api/v1/accounts/{account_id}/info/buying-power")
    return [BuyingPower.model_validate(bp) for bp in response]

get_holdings(account_id, position_type=None)

Get all current positions held in the specified account.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
position_type str | None

Optional filter - 'direct', 'symphony', or 'all'

None
Returns
List of holdings in the account
Source code in composer/resources/accounts.py
def get_holdings(
    self,
    account_id: str,
    position_type: str | None = None,
) -> builtins.list[Holding]:
    """
    Get all current positions held in the specified account.

    Args:
        account_id: The account UUID
        position_type: Optional filter - 'direct', 'symphony', or 'all'

    Returns
    -------
        List of holdings in the account
    """
    params = {}
    if position_type:
        position_type_map = {
            "direct": "DEFAULT_DIRECT",
            "symphony": "SYMPHONY",
            "all": "ALL",
        }
        params["position_type"] = position_type_map.get(
            position_type.lower(), position_type.upper()
        )
    response = self._client.get(
        f"/api/v1/accounts/{account_id}/holdings",
        params=params if params else None,
    )
    return [Holding.model_validate(h) for h in response]

get_info(account_id)

Get basic account owner information.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
Returns
Account owner information including identity, contact, and profile
Source code in composer/resources/accounts.py
def get_info(self, account_id: str) -> AccountInfo:
    """
    Get basic account owner information.

    Args:
        account_id: The account UUID

    Returns
    -------
        Account owner information including identity, contact, and profile
    """
    response = self._client.get(f"/api/v1/accounts/{account_id}/info")
    return AccountInfo.model_validate(response)

get_investor_documents(account_id, category, year)

Get investor documents for Apex and Alpaca accounts.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
category str

Document category (STATEMENT, TAX_FORM, etc.)

required
year int

Year of documents to retrieve

required
Returns
List of investor documents
Source code in composer/resources/accounts.py
def get_investor_documents(
    self,
    account_id: str,
    category: str,
    year: int,
) -> builtins.list[InvestorDocument]:
    """
    Get investor documents for Apex and Alpaca accounts.

    Args:
        account_id: The account UUID
        category: Document category (STATEMENT, TAX_FORM, etc.)
        year: Year of documents to retrieve

    Returns
    -------
        List of investor documents
    """
    params = {"category": category, "year": year}
    response = self._client.get(
        f"/api/v1/accounts/{account_id}/info/investor-documents",
        params=params,
    )
    return [InvestorDocument.model_validate(d) for d in response]

get_supported_regions()

Get a list of countries and states that support equity/crypto trading.

Returns
Supported regions for EQUITIES and CRYPTO trading
Source code in composer/resources/accounts.py
def get_supported_regions(self) -> SupportedRegionsResponse:
    """
    Get a list of countries and states that support equity/crypto trading.

    Returns
    -------
        Supported regions for EQUITIES and CRYPTO trading
    """
    response = self._client.get("/api/v1/accounts/supported-regions")
    return SupportedRegionsResponse.model_validate(response)

list()

Get a list of all accounts associated with the authenticated user.

Returns
List of accounts with status, asset classes, and key dates
Source code in composer/resources/accounts.py
def list(self) -> AccountsListResponse:
    """
    Get a list of all accounts associated with the authenticated user.

    Returns
    -------
        List of accounts with status, asset classes, and key dates
    """
    response = self._client.get("/api/v1/accounts/list")
    return AccountsListResponse.model_validate(response)

Cash Resource

composer.resources.cash.Cash

Resource for cash-related endpoints.

Source code in composer/resources/cash.py
class Cash:
    """Resource for cash-related endpoints."""

    def __init__(self, http_client):
        self._client = http_client

    def get_transfer_constraints(self) -> dict[str, TransferConstraints]:
        """
        Get all transfer constraints for a user's active accounts.

        Returns
        -------
            Map of account UUID to transfer constraints
        """
        response = self._client.get("/api/v1/cash/transfer-constraints")
        return {
            account_id: TransferConstraints.model_validate(constraints)
            for account_id, constraints in response.items()
        }

    def get_ach_relationships(
        self,
        include_plaid_account_details: bool,
    ) -> ACHRelationshipsResponse:
        """
        Get all ACH relationships for a user.

        Args:
            include_plaid_account_details: Whether to include Plaid account details

        Returns
        -------
            ACH relationships including bank accounts
        """
        params = {"include_plaid_account_details": include_plaid_account_details}
        response = self._client.get(
            "/api/v1/cash/ach-relationships",
            params=params,
        )
        return ACHRelationshipsResponse.model_validate(response)

    def get_ach_limits(self, account_id: str) -> ACHLimits:
        """
        Get limits to ACH transfers.

        Args:
            account_id: The account UUID

        Returns
        -------
            ACH transfer limits
        """
        response = self._client.get(
            f"/api/v1/cash/accounts/{account_id}/ach-limits",
        )
        return ACHLimits.model_validate(response)

    def get_tax_withholding_federal(self, account_id: str) -> TaxWithholding:
        """
        Get federal tax withholding requirements for Traditional IRA distributions.

        Args:
            account_id: The account UUID

        Returns
        -------
            Tax withholding information
        """
        response = self._client.get(
            f"/api/v1/cash/accounts/{account_id}/tax-withholding/federal",
        )
        return TaxWithholding.model_validate(response)

    def get_ach_transfers(
        self,
        account_id: str,
        year: int | None = None,
    ) -> list[ACHTransfer]:
        """
        Get an account's ACH transfers.

        Args:
            account_id: The account UUID
            year: Optional year to filter transfers

        Returns
        -------
            List of ACH transfers
        """
        params = {}
        if year is not None:
            params["year"] = year
        response = self._client.get(
            f"/api/v1/cash/accounts/{account_id}/ach-transfers",
            params=params if params else None,
        )
        return [ACHTransfer.model_validate(t) for t in response]

    def get_recurring_deposits(
        self,
        account_id: str,
        status: str | None = None,
        n: int = 10,
    ) -> RecurringDepositsResponse:
        """
        Get all recurring deposits for a broker account.

        Args:
            account_id: The account UUID
            status: Optional status filter (ACTIVE, CANCELED, etc.)
            n: Number of results to return

        Returns
        -------
            List of recurring deposits
        """
        params: dict[str, Any] = {"n": n}
        if status:
            params["status"] = status
        response = self._client.get(
            f"/api/v1/cash/accounts/{account_id}/recurring-deposits",
            params=params,
        )
        return RecurringDepositsResponse.model_validate(response)

    def get_recurring_deposits_meta(self, account_id: str) -> RecurringDepositsMeta:
        """
        Get maximum recurring deposit amount for all frequencies.

        Args:
            account_id: The account UUID

        Returns
        -------
            Max deposit amounts by frequency
        """
        response = self._client.get(
            f"/api/v1/cash/accounts/{account_id}/recurring-deposits-meta",
        )
        return RecurringDepositsMeta.model_validate(response)

    def get_recurring_deposits_projection(
        self,
        account_id: str,
        amount: float,
        frequency: str,
    ) -> RecurringDepositProjection:
        """
        Project when a retirement account would reach its annual contribution limit.

        Args:
            account_id: The account UUID
            amount: Deposit amount
            frequency: Deposit frequency (WEEKLY, SEMIMONTHLY, MONTHLY, QUARTERLY)

        Returns
        -------
            Projection of when limit will be hit
        """
        params = {"amount": amount, "frequency": frequency}
        response = self._client.get(
            f"/api/v1/cash/accounts/{account_id}/recurring-deposits-projection",
            params=params,
        )
        return RecurringDepositProjection.model_validate(response)

    def get_all_recurring_deposits(
        self,
        n: int = 10,
        status: str | None = None,
    ) -> RecurringDepositsResponse:
        """
        Get all recurring deposits for a user.

        Args:
            n: Number of results to return
            status: Optional status filter (ACTIVE, CANCELED, etc.)

        Returns
        -------
            List of all recurring deposits across accounts
        """
        params: dict[str, Any] = {"n": n}
        if status:
            params["status"] = status
        response = self._client.get(
            "/api/v1/cash/recurring-deposits",
            params=params,
        )
        return RecurringDepositsResponse.model_validate(response)

get_ach_limits(account_id)

Get limits to ACH transfers.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
Returns
ACH transfer limits
Source code in composer/resources/cash.py
def get_ach_limits(self, account_id: str) -> ACHLimits:
    """
    Get limits to ACH transfers.

    Args:
        account_id: The account UUID

    Returns
    -------
        ACH transfer limits
    """
    response = self._client.get(
        f"/api/v1/cash/accounts/{account_id}/ach-limits",
    )
    return ACHLimits.model_validate(response)

get_ach_relationships(include_plaid_account_details)

Get all ACH relationships for a user.

Parameters:

Name Type Description Default
include_plaid_account_details bool

Whether to include Plaid account details

required
Returns
ACH relationships including bank accounts
Source code in composer/resources/cash.py
def get_ach_relationships(
    self,
    include_plaid_account_details: bool,
) -> ACHRelationshipsResponse:
    """
    Get all ACH relationships for a user.

    Args:
        include_plaid_account_details: Whether to include Plaid account details

    Returns
    -------
        ACH relationships including bank accounts
    """
    params = {"include_plaid_account_details": include_plaid_account_details}
    response = self._client.get(
        "/api/v1/cash/ach-relationships",
        params=params,
    )
    return ACHRelationshipsResponse.model_validate(response)

get_ach_transfers(account_id, year=None)

Get an account's ACH transfers.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
year int | None

Optional year to filter transfers

None
Returns
List of ACH transfers
Source code in composer/resources/cash.py
def get_ach_transfers(
    self,
    account_id: str,
    year: int | None = None,
) -> list[ACHTransfer]:
    """
    Get an account's ACH transfers.

    Args:
        account_id: The account UUID
        year: Optional year to filter transfers

    Returns
    -------
        List of ACH transfers
    """
    params = {}
    if year is not None:
        params["year"] = year
    response = self._client.get(
        f"/api/v1/cash/accounts/{account_id}/ach-transfers",
        params=params if params else None,
    )
    return [ACHTransfer.model_validate(t) for t in response]

get_all_recurring_deposits(n=10, status=None)

Get all recurring deposits for a user.

Parameters:

Name Type Description Default
n int

Number of results to return

10
status str | None

Optional status filter (ACTIVE, CANCELED, etc.)

None
Returns
List of all recurring deposits across accounts
Source code in composer/resources/cash.py
def get_all_recurring_deposits(
    self,
    n: int = 10,
    status: str | None = None,
) -> RecurringDepositsResponse:
    """
    Get all recurring deposits for a user.

    Args:
        n: Number of results to return
        status: Optional status filter (ACTIVE, CANCELED, etc.)

    Returns
    -------
        List of all recurring deposits across accounts
    """
    params: dict[str, Any] = {"n": n}
    if status:
        params["status"] = status
    response = self._client.get(
        "/api/v1/cash/recurring-deposits",
        params=params,
    )
    return RecurringDepositsResponse.model_validate(response)

get_recurring_deposits(account_id, status=None, n=10)

Get all recurring deposits for a broker account.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
status str | None

Optional status filter (ACTIVE, CANCELED, etc.)

None
n int

Number of results to return

10
Returns
List of recurring deposits
Source code in composer/resources/cash.py
def get_recurring_deposits(
    self,
    account_id: str,
    status: str | None = None,
    n: int = 10,
) -> RecurringDepositsResponse:
    """
    Get all recurring deposits for a broker account.

    Args:
        account_id: The account UUID
        status: Optional status filter (ACTIVE, CANCELED, etc.)
        n: Number of results to return

    Returns
    -------
        List of recurring deposits
    """
    params: dict[str, Any] = {"n": n}
    if status:
        params["status"] = status
    response = self._client.get(
        f"/api/v1/cash/accounts/{account_id}/recurring-deposits",
        params=params,
    )
    return RecurringDepositsResponse.model_validate(response)

get_recurring_deposits_meta(account_id)

Get maximum recurring deposit amount for all frequencies.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
Returns
Max deposit amounts by frequency
Source code in composer/resources/cash.py
def get_recurring_deposits_meta(self, account_id: str) -> RecurringDepositsMeta:
    """
    Get maximum recurring deposit amount for all frequencies.

    Args:
        account_id: The account UUID

    Returns
    -------
        Max deposit amounts by frequency
    """
    response = self._client.get(
        f"/api/v1/cash/accounts/{account_id}/recurring-deposits-meta",
    )
    return RecurringDepositsMeta.model_validate(response)

get_recurring_deposits_projection(account_id, amount, frequency)

Project when a retirement account would reach its annual contribution limit.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
amount float

Deposit amount

required
frequency str

Deposit frequency (WEEKLY, SEMIMONTHLY, MONTHLY, QUARTERLY)

required
Returns
Projection of when limit will be hit
Source code in composer/resources/cash.py
def get_recurring_deposits_projection(
    self,
    account_id: str,
    amount: float,
    frequency: str,
) -> RecurringDepositProjection:
    """
    Project when a retirement account would reach its annual contribution limit.

    Args:
        account_id: The account UUID
        amount: Deposit amount
        frequency: Deposit frequency (WEEKLY, SEMIMONTHLY, MONTHLY, QUARTERLY)

    Returns
    -------
        Projection of when limit will be hit
    """
    params = {"amount": amount, "frequency": frequency}
    response = self._client.get(
        f"/api/v1/cash/accounts/{account_id}/recurring-deposits-projection",
        params=params,
    )
    return RecurringDepositProjection.model_validate(response)

get_tax_withholding_federal(account_id)

Get federal tax withholding requirements for Traditional IRA distributions.

Parameters:

Name Type Description Default
account_id str

The account UUID

required
Returns
Tax withholding information
Source code in composer/resources/cash.py
def get_tax_withholding_federal(self, account_id: str) -> TaxWithholding:
    """
    Get federal tax withholding requirements for Traditional IRA distributions.

    Args:
        account_id: The account UUID

    Returns
    -------
        Tax withholding information
    """
    response = self._client.get(
        f"/api/v1/cash/accounts/{account_id}/tax-withholding/federal",
    )
    return TaxWithholding.model_validate(response)

get_transfer_constraints()

Get all transfer constraints for a user's active accounts.

Returns
Map of account UUID to transfer constraints
Source code in composer/resources/cash.py
def get_transfer_constraints(self) -> dict[str, TransferConstraints]:
    """
    Get all transfer constraints for a user's active accounts.

    Returns
    -------
        Map of account UUID to transfer constraints
    """
    response = self._client.get("/api/v1/cash/transfer-constraints")
    return {
        account_id: TransferConstraints.model_validate(constraints)
        for account_id, constraints in response.items()
    }

User Resource

composer.resources.user.User

Resource for user-related endpoints.

Source code in composer/resources/user.py
class User:
    """Resource for user-related endpoints."""

    def __init__(self, http_client):
        self._client = http_client

    def get_jwt(self) -> JWTResponse:
        """
        Get a user JWT.

        Returns
        -------
            JWT token response
        """
        response = self._client.get("/api/v1/user/jwt")
        return JWTResponse.model_validate(response)

    def get_profile(self) -> UserProfile:
        """
        Get a user profile.

        Returns
        -------
            User profile information
        """
        response = self._client.get("/api/v1/user/profile")
        return UserProfile.model_validate(response)

    def get_agreement_status(self, agreement_id: str) -> AgreementStatusResponse:
        """
        Check if a user has accepted a specific agreement.

        Args:
            agreement_id: The agreement ID to check

        Returns
        -------
            Whether the user has agreed to the agreement
        """
        response = self._client.get(f"/api/v1/user/agreement-status/{agreement_id}")
        return AgreementStatusResponse.model_validate(response)

get_agreement_status(agreement_id)

Check if a user has accepted a specific agreement.

Parameters:

Name Type Description Default
agreement_id str

The agreement ID to check

required
Returns
Whether the user has agreed to the agreement
Source code in composer/resources/user.py
def get_agreement_status(self, agreement_id: str) -> AgreementStatusResponse:
    """
    Check if a user has accepted a specific agreement.

    Args:
        agreement_id: The agreement ID to check

    Returns
    -------
        Whether the user has agreed to the agreement
    """
    response = self._client.get(f"/api/v1/user/agreement-status/{agreement_id}")
    return AgreementStatusResponse.model_validate(response)

get_jwt()

Get a user JWT.

Returns
JWT token response
Source code in composer/resources/user.py
def get_jwt(self) -> JWTResponse:
    """
    Get a user JWT.

    Returns
    -------
        JWT token response
    """
    response = self._client.get("/api/v1/user/jwt")
    return JWTResponse.model_validate(response)

get_profile()

Get a user profile.

Returns
User profile information
Source code in composer/resources/user.py
def get_profile(self) -> UserProfile:
    """
    Get a user profile.

    Returns
    -------
        User profile information
    """
    response = self._client.get("/api/v1/user/profile")
    return UserProfile.model_validate(response)