Skip to content

infrastructure.providers.alpaca.api.transactions_api

src.infrastructure.providers.alpaca.api.transactions_api

Alpaca Transactions API client.

HTTP client for Alpaca Trading API account activities endpoint. Uses API Key authentication headers.

Endpoints

GET /v2/account/activities - Get account activities (Alpaca's term for transactions)

Activity Types

FILL - Order fill (trade execution) DIV - Dividend DIVCGL/DIVCGS/DIVNRA/DIVFT/DIVTXEX - Various dividend types INT - Interest JNLC - Journal entry (cash) JNLS - Journal entry (stock) MA - Merger/Acquisition NC - Name change PTC - Pass-through charge REO - Reorg fee SC - Symbol change SSO - Stock spinoff SSP - Stock split

Reference
  • https://docs.alpaca.markets/reference/getaccountactivities-1

Attributes

Classes

AlpacaTransactionsAPI

Bases: BaseProviderAPIClient

HTTP client for Alpaca Trading API transactions (activities) endpoint.

Uses API Key authentication (not OAuth Bearer tokens). Returns raw JSON responses - mapping to domain types is done by mappers.

Note: Alpaca calls transactions "activities" in their API.

Attributes:

Name Type Description
base_url

Alpaca Trading API base URL (paper or live).
timeout

HTTP request timeout in seconds.
Example

api = AlpacaTransactionsAPI( ... base_url="https://paper-api.alpaca.markets", ... timeout=30.0, ... ) result = await api.get_transactions(api_key="...", api_secret="...")

Source code in src/infrastructure/providers/alpaca/api/transactions_api.py 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
class AlpacaTransactionsAPI(BaseProviderAPIClient):
    """HTTP client for Alpaca Trading API transactions (activities) endpoint.

    Uses API Key authentication (not OAuth Bearer tokens).
    Returns raw JSON responses - mapping to domain types is done by mappers.

    Note: Alpaca calls transactions "activities" in their API.

    Attributes:
        base_url: Alpaca Trading API base URL (paper or live).
        timeout: HTTP request timeout in seconds.

    Example:
        >>> api = AlpacaTransactionsAPI(
        ...     base_url="https://paper-api.alpaca.markets",
        ...     timeout=30.0,
        ... )
        >>> result = await api.get_transactions(api_key="...", api_secret="...")
    """

    def __init__(
        self,
        *,
        base_url: str,
        timeout: float = PROVIDER_TIMEOUT_DEFAULT,
    ) -> None:
        """Initialize Alpaca Activities API client.

        Args:
            base_url: Alpaca Trading API base URL.
            timeout: HTTP request timeout in seconds.
        """
        super().__init__(
            base_url=base_url,
            provider_name="alpaca",
            timeout=timeout,
        )

    async def get_transactions(
        self,
        api_key: str,
        api_secret: str,
        *,
        activity_types: list[str] | None = None,
        start_date: date | None = None,
        end_date: date | None = None,
        page_size: int = 100,
    ) -> Result[list[dict[str, Any]], ProviderError]:
        """Fetch account transactions (called 'activities' in Alpaca API).

        Args:
            api_key: Alpaca API Key ID.
            api_secret: Alpaca API Secret Key.
            activity_types: Filter by Alpaca activity types (e.g., ["FILL", "DIV"]).
            start_date: Get transactions after this date.
            end_date: Get transactions until this date.
            page_size: Number of transactions per page (max 100).

        Returns:
            Success(list[dict]): List of transaction JSON objects.
            Failure(ProviderAuthenticationError): If credentials are invalid.
            Failure(ProviderUnavailableError): If Alpaca API is unreachable.
        """
        self._logger.debug(
            "alpaca_transactions_api_get_transactions_started",
            activity_types=activity_types,
            start_date=str(start_date) if start_date else None,
            end_date=str(end_date) if end_date else None,
        )

        # Build query parameters
        params: dict[str, Any] = {"page_size": page_size}
        if activity_types:
            params["activity_types"] = ",".join(activity_types)
        if start_date:
            params["after"] = start_date.isoformat()
        if end_date:
            params["until"] = end_date.isoformat()

        headers = self._build_alpaca_headers(api_key, api_secret)

        return await self._execute_and_parse_list(
            method="GET",
            path="/v2/account/activities",
            headers=headers,
            params=params,
            operation="get_transactions",
        )

    def _build_alpaca_headers(self, api_key: str, api_secret: str) -> dict[str, str]:
        """Build HTTP headers for Alpaca API requests.

        Args:
            api_key: Alpaca API Key ID.
            api_secret: Alpaca API Secret Key.

        Returns:
            Headers dict with API key authentication.
        """
        return {
            "APCA-API-KEY-ID": api_key,
            "APCA-API-SECRET-KEY": api_secret,
            "Accept": "application/json",
        }
Functions
__init__ 
__init__(
    *,
    base_url: str,
    timeout: float = PROVIDER_TIMEOUT_DEFAULT
) -> None

Parameters:

Name Type Description Default
base_url str

Alpaca Trading API base URL.

 required
timeout float

HTTP request timeout in seconds.

 PROVIDER_TIMEOUT_DEFAULT
Source code in src/infrastructure/providers/alpaca/api/transactions_api.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def __init__(
    self,
    *,
    base_url: str,
    timeout: float = PROVIDER_TIMEOUT_DEFAULT,
) -> None:
    """Initialize Alpaca Activities API client.

    Args:
        base_url: Alpaca Trading API base URL.
        timeout: HTTP request timeout in seconds.
    """
    super().__init__(
        base_url=base_url,
        provider_name="alpaca",
        timeout=timeout,
    )
get_transactions async 
get_transactions(
    api_key: str,
    api_secret: str,
    *,
    activity_types: list[str] | None = None,
    start_date: date | None = None,
    end_date: date | None = None,
    page_size: int = 100
) -> Result[list[dict[str, Any]], ProviderError]

Fetch account transactions (called 'activities' in Alpaca API).

Parameters:

Name Type Description Default
api_key str

Alpaca API Key ID.

 required
api_secret str

Alpaca API Secret Key.

 required
activity_types list[str] | None

Filter by Alpaca activity types (e.g., ["FILL", "DIV"]).

 None
start_date date | None

Get transactions after this date.

 None
end_date date | None

Get transactions until this date.

 None
page_size int

Number of transactions per page (max 100).

 100

Returns:

Name Type Description
Success list[dict]

List of transaction JSON objects.
Failure ProviderAuthenticationError

If credentials are invalid.
Failure ProviderUnavailableError

If Alpaca API is unreachable.
Source code in src/infrastructure/providers/alpaca/api/transactions_api.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
async def get_transactions(
    self,
    api_key: str,
    api_secret: str,
    *,
    activity_types: list[str] | None = None,
    start_date: date | None = None,
    end_date: date | None = None,
    page_size: int = 100,
) -> Result[list[dict[str, Any]], ProviderError]:
    """Fetch account transactions (called 'activities' in Alpaca API).

    Args:
        api_key: Alpaca API Key ID.
        api_secret: Alpaca API Secret Key.
        activity_types: Filter by Alpaca activity types (e.g., ["FILL", "DIV"]).
        start_date: Get transactions after this date.
        end_date: Get transactions until this date.
        page_size: Number of transactions per page (max 100).

    Returns:
        Success(list[dict]): List of transaction JSON objects.
        Failure(ProviderAuthenticationError): If credentials are invalid.
        Failure(ProviderUnavailableError): If Alpaca API is unreachable.
    """
    self._logger.debug(
        "alpaca_transactions_api_get_transactions_started",
        activity_types=activity_types,
        start_date=str(start_date) if start_date else None,
        end_date=str(end_date) if end_date else None,
    )

    # Build query parameters
    params: dict[str, Any] = {"page_size": page_size}
    if activity_types:
        params["activity_types"] = ",".join(activity_types)
    if start_date:
        params["after"] = start_date.isoformat()
    if end_date:
        params["until"] = end_date.isoformat()

    headers = self._build_alpaca_headers(api_key, api_secret)

    return await self._execute_and_parse_list(
        method="GET",
        path="/v2/account/activities",
        headers=headers,
        params=params,
        operation="get_transactions",
    )