CtrlK
BlogDocsLog inGet started
Tessl Logo

tessl/pypi-alpaca-py

The Official Python SDK for Alpaca APIs providing access to trading, market data, and broker services

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview
Eval results
Files

broker-client.mddocs/

Broker Client

The Broker Client provides comprehensive functionality for managing customer brokerage accounts, including account creation, KYC/AML compliance, funding operations, document management, and portfolio rebalancing. It's designed for broker-dealers and fintech platforms building investment services.

Core Client

BrokerClient

from alpaca.broker import BrokerClient

class BrokerClient(RESTClient):
    def __init__(
        self,
        api_key: Optional[str] = None,
        secret_key: Optional[str] = None,
        oauth_token: Optional[str] = None,
        use_basic_auth: bool = False,
        raw_data: bool = False,
        url_override: Optional[str] = None,
        sandbox: bool = False,
    ) -> None:
        """
        Initialize broker client for account management.
        
        Args:
            api_key: Broker API key
            secret_key: Broker API secret key
            oauth_token: OAuth token for user-on-behalf operations
            use_basic_auth: Use basic authentication headers
            raw_data: Return raw API responses instead of models
            url_override: Override base URL for testing
            sandbox: Use sandbox environment for testing
        """

Basic Setup

# Production broker client
broker_client = BrokerClient(
    api_key="your-broker-api-key",
    secret_key="your-broker-secret-key",
    use_basic_auth=True
)

# Sandbox broker client for testing
broker_client = BrokerClient(
    api_key="sandbox-api-key",
    secret_key="sandbox-secret-key", 
    use_basic_auth=True,
    sandbox=True
)

# OAuth-based client for user-on-behalf operations
broker_client = BrokerClient(
    oauth_token="user-oauth-token",
    sandbox=True
)

Account Management

Account Creation

create_account()

def create_account(
    self, 
    account_data: CreateAccountRequest
) -> Union[Account, RawData]:
    """
    Create a new customer brokerage account.
    
    Args:
        account_data: Account creation parameters including identity, contact, and disclosures
        
    Returns:
        Account: Created account with assigned ID and status
    """

CreateAccountRequest

from alpaca.broker.requests import CreateAccountRequest
from alpaca.broker.models import Contact, Identity, Disclosures, TrustedContact
from datetime import date

class CreateAccountRequest(NonEmptyRequest):
    def __init__(
        self,
        contact: Contact,                              # Required: Contact information
        identity: Identity,                            # Required: Identity information  
        disclosures: Disclosures,                     # Required: Regulatory disclosures
        agreements: List[AgreementType],              # Required: Signed agreements
        documents: Optional[List[AccountDocument]] = None,  # Supporting documents
        trusted_contact: Optional[TrustedContact] = None,   # Trusted contact person
        enabled_assets: Optional[List[AssetClass]] = None,  # Tradeable asset classes
        currency: Optional[str] = None,                     # Account currency (USD)
        max_options_trading_level: Optional[int] = None     # Options trading level (0-4)
    ):

Core Account Models

Contact Model

from alpaca.broker.models import Contact

class Contact:
    """Customer contact information."""
    
    email_address: str              # Email address
    phone_number: str               # Phone number with country code
    street_address: List[str]       # Street address lines
    unit: Optional[str]            # Apartment/unit number
    city: str                      # City
    state: str                     # State/province
    postal_code: str               # ZIP/postal code  
    country: Optional[str]         # Country code (defaults to USA)

Identity Model

from alpaca.broker.models import Identity
from alpaca.broker.enums import TaxIdType, EmploymentStatus, FundingSource

class Identity:
    """Customer identity information."""
    
    given_name: str                           # First name
    middle_name: Optional[str]                # Middle name
    family_name: str                          # Last name
    date_of_birth: date                       # Date of birth
    tax_id: Optional[str]                     # SSN/ITIN/EIN
    tax_id_type: Optional[TaxIdType]          # Type of tax ID
    country_of_citizenship: Optional[str]     # Citizenship country
    country_of_birth: Optional[str]           # Birth country
    country_of_tax_residence: Optional[str]   # Tax residence country
    funding_source: Optional[List[FundingSource]]  # Source of funds
    annual_income_min: Optional[int]          # Minimum annual income
    annual_income_max: Optional[int]          # Maximum annual income
    liquid_net_worth_min: Optional[int]       # Minimum liquid net worth
    liquid_net_worth_max: Optional[int]       # Maximum liquid net worth
    total_net_worth_min: Optional[int]        # Minimum total net worth
    total_net_worth_max: Optional[int]        # Maximum total net worth
    employment_status: Optional[EmploymentStatus]  # Employment status
    employer_name: Optional[str]              # Employer name
    employer_address: Optional[str]           # Employer address
    employment_position: Optional[str]        # Job title/position

Disclosures Model

from alpaca.broker.models import Disclosures

class Disclosures:
    """Regulatory disclosure responses."""
    
    is_control_person: bool                   # Control person of public company
    is_affiliated_exchange_or_finra: bool     # FINRA/exchange affiliation  
    is_politically_exposed: bool             # Politically exposed person
    immediate_family_exposed: bool           # Family politically exposed
    is_discretionary_investment: bool        # Discretionary investment authority
    employment_details: Optional[str]        # Employment disclosure details
    employer_details: Optional[str]          # Employer disclosure details
    investment_experience: Optional[str]     # Investment experience level
    investment_objectives: Optional[List[str]]  # Investment objectives
    source_of_funds: Optional[List[str]]     # Source of funds details

Account Creation Example:

from alpaca.broker.requests import CreateAccountRequest
from alpaca.broker.models import Contact, Identity, Disclosures
from alpaca.broker.enums import TaxIdType, EmploymentStatus, FundingSource, AgreementType
from datetime import date

# Create account request
account_request = CreateAccountRequest(
    contact=Contact(
        email_address="customer@example.com",
        phone_number="+1-555-123-4567", 
        street_address=["123 Main Street"],
        city="New York",
        state="NY",
        postal_code="10001"
    ),
    
    identity=Identity(
        given_name="John",
        family_name="Doe",
        date_of_birth=date(1990, 5, 15),
        tax_id="123456789",
        tax_id_type=TaxIdType.USA_SSN,
        country_of_citizenship="USA",
        annual_income_min=50000,
        annual_income_max=75000,
        liquid_net_worth_min=10000,
        liquid_net_worth_max=25000,
        employment_status=EmploymentStatus.EMPLOYED,
        funding_source=[FundingSource.EMPLOYMENT_INCOME]
    ),
    
    disclosures=Disclosures(
        is_control_person=False,
        is_affiliated_exchange_or_finra=False,
        is_politically_exposed=False,
        immediate_family_exposed=False,
        is_discretionary_investment=False
    ),
    
    agreements=[
        AgreementType.ACCOUNT_AGREEMENT,
        AgreementType.CUSTOMER_AGREEMENT
    ],
    
    enabled_assets=[AssetClass.US_EQUITY],
    max_options_trading_level=0  # No options trading
)

# Create the account
account = broker_client.create_account(account_request)
print(f"Account created: {account.id}")
print(f"Account number: {account.account_number}")
print(f"Status: {account.status}")

Account Querying

list_accounts()

def list_accounts(
    self, 
    search_parameters: Optional[ListAccountsRequest] = None
) -> Union[List[Account], RawData]:
    """
    List customer accounts with optional filtering.
    
    Args:
        search_parameters: Account filtering criteria
        
    Returns:
        List[Account]: List of accounts matching criteria
    """

get_account_by_id()

def get_account_by_id(
    self, 
    account_id: Union[UUID, str]
) -> Union[Account, RawData]:
    """
    Get specific account by ID.
    
    Args:
        account_id: Account UUID
        
    Returns:
        Account: Account details
    """

ListAccountsRequest

from alpaca.broker.requests import ListAccountsRequest
from alpaca.trading.enums import AccountStatus
from datetime import datetime

class ListAccountsRequest(NonEmptyRequest):
    def __init__(
        self,
        query: Optional[str] = None,                    # Search query (name, email, etc.)
        created_after: Optional[datetime] = None,       # Created after timestamp
        created_before: Optional[datetime] = None,      # Created before timestamp
        status: Optional[AccountStatus] = None,         # Account status filter
        sort: Optional[str] = None,                     # Sort field
        entities: Optional[str] = None,                 # Additional entities to include
        limit: Optional[int] = None,                    # Maximum accounts to return
        page_token: Optional[str] = None                # Pagination token
    ):

Account Querying Examples:

# List all accounts
accounts = broker_client.list_accounts()
print(f"Total accounts: {len(accounts)}")

# Search for specific customer
search_results = broker_client.list_accounts(
    ListAccountsRequest(
        query="john.doe@example.com",
        limit=10
    )
)

# Filter by status and date
active_accounts = broker_client.list_accounts(
    ListAccountsRequest(
        status=AccountStatus.ACTIVE,
        created_after=datetime.now() - timedelta(days=30),
        sort="created_at"
    )
)

# Get specific account
account = broker_client.get_account_by_id("account-uuid-here")
print(f"Account {account.account_number}: {account.status}")

Account Updates

update_account()

def update_account(
    self,
    account_id: Union[UUID, str],
    account_data: UpdateAccountRequest,
) -> Union[Account, RawData]:
    """
    Update existing account information.
    
    Args:
        account_id: Account UUID to update
        account_data: Updated account fields
        
    Returns:
        Account: Updated account
    """

UpdateAccountRequest

from alpaca.broker.requests import UpdateAccountRequest

class UpdateAccountRequest(NonEmptyRequest):
    def __init__(
        self,
        contact: Optional[Contact] = None,              # Updated contact info
        identity: Optional[Identity] = None,            # Updated identity info
        disclosures: Optional[Disclosures] = None,      # Updated disclosures
        trusted_contact: Optional[TrustedContact] = None, # Updated trusted contact
        enabled_assets: Optional[List[AssetClass]] = None,  # Updated tradeable assets
        max_options_trading_level: Optional[int] = None     # Updated options level
    ):

Account Update Examples:

# Update contact information
updated_contact = Contact(
    email_address="newemail@example.com",
    phone_number="+1-555-987-6543",
    street_address=["456 New Street"],
    city="San Francisco", 
    state="CA",
    postal_code="94102"
)

update_request = UpdateAccountRequest(contact=updated_contact)
updated_account = broker_client.update_account(account_id, update_request)

# Enable options trading
options_update = UpdateAccountRequest(
    max_options_trading_level=2,  # Level 2 options
    enabled_assets=[AssetClass.US_EQUITY, AssetClass.US_OPTION]
)
updated_account = broker_client.update_account(account_id, options_update)

Funding Operations

ACH Relationships

create_ach_relationship()

def create_ach_relationship(
    self,
    account_id: Union[UUID, str],
    ach_data: CreateACHRelationshipRequest,
) -> Union[ACHRelationship, RawData]:
    """
    Create ACH bank relationship for transfers.
    
    Args:
        account_id: Customer account UUID
        ach_data: ACH relationship parameters
        
    Returns:
        ACHRelationship: Created ACH relationship
    """

CreateACHRelationshipRequest

from alpaca.broker.requests import CreateACHRelationshipRequest
from alpaca.broker.enums import BankAccountType

class CreateACHRelationshipRequest(NonEmptyRequest):
    def __init__(
        self,
        account_owner_name: str,                    # Account owner name
        bank_account_type: BankAccountType,         # CHECKING or SAVINGS
        bank_account_number: str,                   # Bank account number
        bank_routing_number: str,                   # Bank routing number
        nickname: Optional[str] = None              # Relationship nickname
    ):

create_plaid_relationship()

def create_plaid_relationship(
    self,
    account_id: Union[UUID, str],
    plaid_data: CreatePlaidRelationshipRequest,
) -> Union[ACHRelationship, RawData]:
    """
    Create ACH relationship via Plaid Link.
    
    Args:
        account_id: Customer account UUID  
        plaid_data: Plaid integration parameters
        
    Returns:
        ACHRelationship: Created ACH relationship via Plaid
    """

get_ach_relationships()

def get_ach_relationships(
    self, 
    account_id: Union[UUID, str]
) -> Union[List[ACHRelationship], RawData]:
    """
    Get ACH relationships for account.
    
    Args:
        account_id: Customer account UUID
        
    Returns:
        List[ACHRelationship]: ACH relationships for the account
    """

delete_ach_relationship()

def delete_ach_relationship(
    self,
    account_id: Union[UUID, str],
    ach_relationship_id: Union[UUID, str],
) -> None:
    """
    Delete ACH relationship.
    
    Args:
        account_id: Customer account UUID
        ach_relationship_id: ACH relationship UUID to delete
    """

ACHRelationship Model

from alpaca.broker.models import ACHRelationship
from alpaca.broker.enums import ACHRelationshipStatus, BankAccountType

class ACHRelationship:
    """ACH bank relationship for transfers."""
    
    id: UUID                                # Relationship UUID
    account_id: UUID                        # Customer account UUID
    account_owner_name: str                 # Account owner name
    bank_account_type: BankAccountType      # CHECKING or SAVINGS
    bank_account_number: str                # Masked account number
    bank_routing_number: str                # Bank routing number
    nickname: Optional[str]                 # Relationship nickname
    status: ACHRelationshipStatus          # Relationship status
    created_at: datetime                    # Creation timestamp

ACH Relationship Examples:

from alpaca.broker.enums import BankAccountType

# Create ACH relationship manually
ach_request = CreateACHRelationshipRequest(
    account_owner_name="John Doe",
    bank_account_type=BankAccountType.CHECKING,
    bank_account_number="1234567890",
    bank_routing_number="021000021",  # Chase routing number
    nickname="Primary Checking"
)

ach_relationship = broker_client.create_ach_relationship(account_id, ach_request)
print(f"ACH relationship created: {ach_relationship.id}")
print(f"Status: {ach_relationship.status}")

# List ACH relationships for account
relationships = broker_client.get_ach_relationships(account_id)
for rel in relationships:
    print(f"{rel.nickname}: {rel.bank_account_type} ending in {rel.bank_account_number[-4:]}")

Money Transfers

create_ach_transfer()

def create_ach_transfer(
    self,
    account_id: Union[UUID, str],
    transfer_data: CreateACHTransferRequest,
) -> Union[Transfer, RawData]:
    """
    Create ACH transfer (deposit or withdrawal).
    
    Args:
        account_id: Customer account UUID
        transfer_data: Transfer parameters
        
    Returns:
        Transfer: Created transfer with status
    """

CreateACHTransferRequest

from alpaca.broker.requests import CreateACHTransferRequest
from alpaca.broker.enums import TransferDirection, TransferTiming

class CreateACHTransferRequest(NonEmptyRequest):
    def __init__(
        self,
        relationship_id: Union[UUID, str],              # ACH relationship UUID
        amount: float,                                  # Transfer amount
        direction: TransferDirection,                   # INCOMING or OUTGOING
        timing: Optional[TransferTiming] = None,        # IMMEDIATE or NEXT_BUSINESS_DAY
        fee_payment_method: Optional[str] = None        # Fee payment method
    ):

get_transfers()

def get_transfers(
    self,
    account_id: Union[UUID, str],
    transfer_data: Optional[GetTransfersRequest] = None,
) -> Union[List[Transfer], RawData]:
    """
    Get transfers for account with optional filtering.
    
    Args:
        account_id: Customer account UUID
        transfer_data: Transfer query parameters
        
    Returns:
        List[Transfer]: Account transfers
    """

cancel_transfer()

def cancel_transfer(
    self,
    account_id: Union[UUID, str],
    transfer_id: Union[UUID, str],
) -> None:
    """
    Cancel pending transfer.
    
    Args:
        account_id: Customer account UUID
        transfer_id: Transfer UUID to cancel
    """

Transfer Model

from alpaca.broker.models import Transfer
from alpaca.broker.enums import TransferDirection, TransferStatus, TransferType

class Transfer:
    """Money transfer information."""
    
    id: UUID                            # Transfer UUID
    account_id: UUID                    # Customer account UUID
    relationship_id: UUID               # ACH relationship UUID
    amount: float                       # Transfer amount
    direction: TransferDirection        # INCOMING or OUTGOING
    type: TransferType                  # ACH, WIRE, CHECK
    status: TransferStatus              # Transfer status
    reason: Optional[str]               # Transfer reason/description
    requested_at: datetime              # Request timestamp
    effective_at: Optional[datetime]    # Effective date
    expires_at: Optional[datetime]      # Expiration date
    fee: Optional[float]                # Transfer fee
    additional_information: Optional[str]  # Additional details

Transfer Examples:

from alpaca.broker.enums import TransferDirection

# Deposit funds (incoming transfer)
deposit_request = CreateACHTransferRequest(
    relationship_id=ach_relationship.id,
    amount=5000.00,
    direction=TransferDirection.INCOMING
)

deposit = broker_client.create_ach_transfer(account_id, deposit_request)
print(f"Deposit initiated: ${deposit.amount} (Status: {deposit.status})")

# Withdraw funds (outgoing transfer)
withdrawal_request = CreateACHTransferRequest(
    relationship_id=ach_relationship.id,
    amount=1000.00, 
    direction=TransferDirection.OUTGOING
)

withdrawal = broker_client.create_ach_transfer(account_id, withdrawal_request)
print(f"Withdrawal initiated: ${withdrawal.amount}")

# Get transfer history
transfers = broker_client.get_transfers(account_id)
for transfer in transfers:
    direction_str = "Deposit" if transfer.direction == TransferDirection.INCOMING else "Withdrawal"
    print(f"{direction_str}: ${transfer.amount} - {transfer.status} ({transfer.requested_at.date()})")

Trading Operations

Order Management for Broker Accounts

submit_order()

def submit_order(
    self,
    account_id: Union[UUID, str],
    order_data: OrderRequest,
) -> Union[Order, RawData]:
    """
    Submit order for customer account.
    
    Args:
        account_id: Customer account UUID
        order_data: Order parameters (same as trading client)
        
    Returns:
        Order: Submitted order
    """

get_orders_for_account()

def get_orders_for_account(
    self, 
    account_id: Union[UUID, str]
) -> Union[List[Order], RawData]:
    """
    Get orders for specific customer account.
    
    Args:
        account_id: Customer account UUID
        
    Returns:
        List[Order]: Orders for the account
    """

Broker Trading Examples:

from alpaca.trading.requests import MarketOrderRequest
from alpaca.trading.enums import OrderSide, TimeInForce

# Submit order for customer account
customer_order = MarketOrderRequest(
    symbol="AAPL",
    qty=100,
    side=OrderSide.BUY,
    time_in_force=TimeInForce.DAY
)

order = broker_client.submit_order(account_id, customer_order)
print(f"Order submitted for customer: {order.id}")

# Get customer's orders
customer_orders = broker_client.get_orders_for_account(account_id)
print(f"Customer has {len(customer_orders)} orders")

Document Management

Document Upload and Retrieval

upload_document()

def upload_document(
    self,
    account_id: Union[UUID, str],
    document_data: UploadDocumentRequest,
) -> Union[AccountDocument, RawData]:
    """
    Upload document for account.
    
    Args:
        account_id: Customer account UUID
        document_data: Document upload parameters
        
    Returns:
        AccountDocument: Uploaded document information
    """

get_documents()

def get_documents(
    self, 
    account_id: Union[UUID, str]
) -> Union[List[AccountDocument], RawData]:
    """
    Get documents for account.
    
    Args:
        account_id: Customer account UUID
        
    Returns:
        List[AccountDocument]: Account documents
    """

download_document()

def download_document(
    self,
    account_id: Union[UUID, str],
    document_id: Union[UUID, str],
) -> bytes:
    """
    Download document content.
    
    Args:
        account_id: Customer account UUID
        document_id: Document UUID
        
    Returns:
        bytes: Document content
    """

Trade Documents

get_trade_documents()

def get_trade_documents(
    self,
    account_id: Union[UUID, str],
    request: Optional[GetTradeDocumentsRequest] = None,
) -> Union[List[TradeDocument], RawData]:
    """
    Get trade confirmation and statement documents.
    
    Args:
        account_id: Customer account UUID
        request: Trade document filtering parameters
        
    Returns:
        List[TradeDocument]: Trade documents
    """

Document Management Examples:

# Upload identity document
with open("customer_id.pdf", "rb") as f:
    document_content = f.read()

upload_request = UploadDocumentRequest(
    document_type="identity_verification",
    content=base64.b64encode(document_content).decode(),
    mime_type="application/pdf"
)

document = broker_client.upload_document(account_id, upload_request)
print(f"Document uploaded: {document.id}")

# Get all documents for account
documents = broker_client.get_documents(account_id)
for doc in documents:
    print(f"{doc.document_type}: {doc.name} (uploaded: {doc.created_at})")

# Get trade confirmations
trade_docs = broker_client.get_trade_documents(account_id)
for doc in trade_docs:
    print(f"Trade document: {doc.name} ({doc.date})")

Journal Entries

Journal entries allow moving cash or securities between accounts.

Cash Journals

create_journal()

def create_journal(
    self, 
    journal_data: CreateJournalRequest
) -> Union[Journal, RawData]:
    """
    Create journal entry to move cash between accounts.
    
    Args:
        journal_data: Journal entry parameters
        
    Returns:
        Journal: Created journal entry
    """

CreateJournalRequest

from alpaca.broker.requests import CreateJournalRequest
from alpaca.broker.enums import JournalEntryType
from datetime import date

class CreateJournalRequest(NonEmptyRequest):
    def __init__(
        self,
        entry_type: JournalEntryType,               # JNLC (cash) or JNLS (securities)
        from_account: Union[UUID, str],             # Source account UUID  
        to_account: Union[UUID, str],               # Destination account UUID
        amount: Optional[float] = None,             # Cash amount (for JNLC)
        symbol: Optional[str] = None,               # Securities symbol (for JNLS)
        qty: Optional[float] = None,                # Securities quantity (for JNLS)
        price: Optional[float] = None,              # Securities price (for JNLS)
        settle_date: Optional[date] = None,         # Settlement date
        description: Optional[str] = None           # Journal description
    ):

Batch Journals

create_batch_journal()

def create_batch_journal(
    self, 
    journal_data: CreateBatchJournalRequest
) -> Union[BatchJournalResponse, RawData]:
    """
    Create multiple journal entries in a batch.
    
    Args:
        journal_data: Batch journal parameters
        
    Returns:
        BatchJournalResponse: Batch creation results
    """

get_journals()

def get_journals(
    self, 
    journal_data: Optional[GetJournalsRequest] = None
) -> Union[List[Journal], RawData]:
    """
    Get journal entries with optional filtering.
    
    Args:
        journal_data: Journal query parameters
        
    Returns:
        List[Journal]: Journal entries
    """

Journal Model

from alpaca.broker.models import Journal
from alpaca.broker.enums import JournalEntryType, JournalStatus

class Journal:
    """Journal entry for moving cash or securities between accounts."""
    
    id: UUID                           # Journal UUID
    entry_type: JournalEntryType       # JNLC (cash) or JNLS (securities)
    from_account: UUID                 # Source account UUID
    to_account: UUID                   # Destination account UUID
    status: JournalStatus              # Journal status
    settle_date: Optional[date]        # Settlement date
    system_date: Optional[date]        # System processing date
    net_amount: Optional[float]        # Net cash amount
    amount: Optional[float]            # Gross amount
    symbol: Optional[str]              # Securities symbol
    qty: Optional[float]               # Securities quantity  
    price: Optional[float]             # Securities price
    description: Optional[str]         # Description
    created_at: datetime               # Creation timestamp

Journal Examples:

from alpaca.broker.enums import JournalEntryType

# Transfer cash between accounts
cash_journal = CreateJournalRequest(
    entry_type=JournalEntryType.JNLC,  # Cash journal
    from_account="source-account-uuid",
    to_account="destination-account-uuid", 
    amount=1000.00,
    description="Customer cash transfer"
)

journal = broker_client.create_journal(cash_journal)
print(f"Cash journal created: {journal.id} (${journal.amount})")

# Transfer securities between accounts  
securities_journal = CreateJournalRequest(
    entry_type=JournalEntryType.JNLS,  # Securities journal
    from_account="source-account-uuid",
    to_account="destination-account-uuid",
    symbol="AAPL",
    qty=100,
    price=175.00,
    description="AAPL share transfer"
)

securities_journal_entry = broker_client.create_journal(securities_journal)
print(f"Securities journal created: {securities_journal_entry.id}")

# Get journal history
journals = broker_client.get_journals()
for journal in journals:
    if journal.entry_type == JournalEntryType.JNLC:
        print(f"Cash: ${journal.amount} from {journal.from_account} to {journal.to_account}")
    else:
        print(f"Securities: {journal.qty} {journal.symbol} @ ${journal.price}")

Portfolio Management & Rebalancing

Portfolio Creation

create_portfolio()

def create_portfolio(
    self, 
    portfolio_data: CreatePortfolioRequest
) -> Union[Portfolio, RawData]:
    """
    Create portfolio strategy for rebalancing.
    
    Args:
        portfolio_data: Portfolio creation parameters
        
    Returns:
        Portfolio: Created portfolio strategy
    """

CreatePortfolioRequest

from alpaca.broker.requests import CreatePortfolioRequest
from alpaca.broker.enums import WeightType

class CreatePortfolioRequest(NonEmptyRequest):
    def __init__(
        self,
        name: str,                              # Portfolio name
        description: Optional[str] = None,      # Portfolio description
        weights: Dict[str, float],              # Asset allocation weights
        weight_type: WeightType = WeightType.PERCENT,  # PERCENT or DOLLAR
        cooldown_days: Optional[int] = None,    # Rebalancing cooldown period
        schedule: Optional[str] = None,         # Rebalancing schedule
        max_accounts_per_day: Optional[int] = None  # Max accounts to rebalance per day
    ):

get_portfolios()

def get_portfolios(
    self, 
    request: Optional[GetPortfoliosRequest] = None
) -> Union[List[Portfolio], RawData]:
    """
    Get portfolio strategies with optional filtering.
    
    Args:
        request: Portfolio query parameters
        
    Returns:
        List[Portfolio]: Portfolio strategies
    """

Account Subscriptions

create_subscription()

def create_subscription(
    self, 
    subscription_data: CreateSubscriptionRequest
) -> Union[Subscription, RawData]:
    """
    Subscribe account to portfolio strategy.
    
    Args:
        subscription_data: Subscription parameters
        
    Returns:
        Subscription: Created subscription
    """

CreateSubscriptionRequest

from alpaca.broker.requests import CreateSubscriptionRequest

class CreateSubscriptionRequest(NonEmptyRequest):
    def __init__(
        self,
        account_id: Union[UUID, str],           # Customer account UUID
        portfolio_id: Union[UUID, str],         # Portfolio strategy UUID
        max_investment_amount: Optional[float] = None,  # Maximum investment
        cash_percentage: Optional[float] = None         # Cash allocation percentage
    ):

Rebalancing Runs

create_run()

def create_run(
    self, 
    run_data: CreateRunRequest
) -> Union[RebalancingRun, RawData]:
    """
    Execute portfolio rebalancing run.
    
    Args:
        run_data: Rebalancing run parameters
        
    Returns:
        RebalancingRun: Rebalancing execution results
    """

CreateRunRequest

from alpaca.broker.requests import CreateRunRequest
from alpaca.broker.enums import RunType

class CreateRunRequest(NonEmptyRequest):
    def __init__(
        self,
        type: RunType,                          # FULL_REBALANCE or DRIFT_REBALANCE
        portfolio_id: Union[UUID, str],         # Portfolio strategy UUID
        account_ids: Optional[List[Union[UUID, str]]] = None,  # Specific accounts to rebalance
        notify: Optional[bool] = None,          # Send notifications
        max_accounts_per_day: Optional[int] = None  # Limit accounts per day
    ):

Portfolio Management Examples:

from alpaca.broker.enums import WeightType, RunType

# Create balanced portfolio strategy
portfolio_request = CreatePortfolioRequest(
    name="Conservative Balanced",
    description="60% stocks, 40% bonds conservative allocation",
    weights={
        "SPY": 0.30,    # S&P 500 ETF - 30%
        "VTI": 0.30,    # Total Stock Market - 30%  
        "BND": 0.25,    # Bond ETF - 25%
        "VTEB": 0.15    # Municipal Bond ETF - 15%
    },
    weight_type=WeightType.PERCENT,
    cooldown_days=30  # Rebalance at most once per month
)

portfolio = broker_client.create_portfolio(portfolio_request)
print(f"Portfolio created: {portfolio.id} - {portfolio.name}")

# Subscribe customer account to portfolio
subscription_request = CreateSubscriptionRequest(
    account_id="customer-account-uuid",
    portfolio_id=portfolio.id,
    max_investment_amount=50000.00,  # Max $50k investment
    cash_percentage=0.05  # Keep 5% in cash
)

subscription = broker_client.create_subscription(subscription_request)
print(f"Account subscribed to portfolio: {subscription.id}")

# Execute rebalancing run
run_request = CreateRunRequest(
    type=RunType.FULL_REBALANCE,
    portfolio_id=portfolio.id,
    account_ids=[subscription.account_id],  # Rebalance specific account
    notify=True
)

rebalancing_run = broker_client.create_run(run_request)
print(f"Rebalancing run initiated: {rebalancing_run.id}")
print(f"Status: {rebalancing_run.status}")

CIP (Customer Identification Program)

KYC and Identity Verification

The broker client includes comprehensive KYC/AML functionality:

CIP Models

from alpaca.broker.models.cip import (
    CIPKYCInfo, CIPDocument, CIPPhoto, CIPIdentity, 
    CIPWatchlist, CIPInfo
)

class CIPInfo:
    """Overall Customer Identification Program status."""
    
    kyc: Optional[CIPKYCInfo]           # KYC verification results
    identity: Optional[CIPIdentity]     # Identity verification results  
    watchlist: Optional[CIPWatchlist]   # Watchlist screening results
    patriot_act: Optional[dict]         # USA PATRIOT Act verification

Account Activities

get_account_activities()

def get_account_activities(
    self,
    account_id: Union[UUID, str],
    request: Optional[GetAccountActivitiesRequest] = None,
) -> Union[List[BaseActivity], RawData]:
    """
    Get account activities (trades, dividends, etc.).
    
    Args:
        account_id: Customer account UUID
        request: Activity filtering parameters
        
    Returns:
        List[BaseActivity]: Account activities
    """

Error Handling

from alpaca.common.exceptions import APIError

try:
    # Create account that might fail validation
    account = broker_client.create_account(account_request)
except APIError as e:
    if e.status_code == 400:
        print(f"Invalid account data: {e.message}")
    elif e.status_code == 409:
        print("Account already exists with this information")
    elif e.status_code == 422:
        print(f"Validation error: {e.message}")
    else:
        print(f"API error: {e.message} (status: {e.status_code})")
except Exception as e:
    print(f"Unexpected error: {e}")

Complete Broker Integration Example

from datetime import date, datetime, timedelta
from alpaca.broker import BrokerClient
from alpaca.broker.requests import *
from alpaca.broker.models import *
from alpaca.broker.enums import *
from alpaca.trading.requests import MarketOrderRequest
from alpaca.trading.enums import OrderSide, TimeInForce

class BrokerPlatform:
    def __init__(self, api_key: str, secret_key: str, sandbox: bool = True):
        self.client = BrokerClient(
            api_key=api_key,
            secret_key=secret_key,
            use_basic_auth=True,
            sandbox=sandbox
        )
    
    def onboard_customer(self, customer_info: dict) -> str:
        """Complete customer onboarding workflow."""
        
        print("🚀 Starting customer onboarding...")
        
        # Step 1: Create account
        account_request = CreateAccountRequest(
            contact=Contact(
                email_address=customer_info["email"],
                phone_number=customer_info["phone"],
                street_address=[customer_info["address"]],
                city=customer_info["city"],
                state=customer_info["state"],
                postal_code=customer_info["zip_code"]
            ),
            
            identity=Identity(
                given_name=customer_info["first_name"],
                family_name=customer_info["last_name"],
                date_of_birth=customer_info["date_of_birth"],
                tax_id=customer_info["ssn"],
                tax_id_type=TaxIdType.USA_SSN,
                annual_income_min=customer_info["income_min"],
                annual_income_max=customer_info["income_max"],
                employment_status=EmploymentStatus.EMPLOYED,
                funding_source=[FundingSource.EMPLOYMENT_INCOME]
            ),
            
            disclosures=Disclosures(
                is_control_person=False,
                is_affiliated_exchange_or_finra=False,
                is_politically_exposed=False,
                immediate_family_exposed=False,
                is_discretionary_investment=False
            ),
            
            agreements=[
                AgreementType.ACCOUNT_AGREEMENT,
                AgreementType.CUSTOMER_AGREEMENT
            ],
            
            enabled_assets=[AssetClass.US_EQUITY]
        )
        
        account = self.client.create_account(account_request)
        print(f"✅ Account created: {account.account_number}")
        
        return str(account.id)
    
    def setup_funding(self, account_id: str, bank_info: dict) -> str:
        """Set up bank relationship and initial funding."""
        
        print("💰 Setting up funding...")
        
        # Create ACH relationship
        ach_request = CreateACHRelationshipRequest(
            account_owner_name=bank_info["owner_name"],
            bank_account_type=BankAccountType.CHECKING,
            bank_account_number=bank_info["account_number"],
            bank_routing_number=bank_info["routing_number"],
            nickname="Primary Checking"
        )
        
        ach_relationship = self.client.create_ach_relationship(account_id, ach_request)
        print(f"✅ ACH relationship created: {ach_relationship.id}")
        
        # Initial deposit
        if bank_info.get("initial_deposit", 0) > 0:
            deposit_request = CreateACHTransferRequest(
                relationship_id=ach_relationship.id,
                amount=bank_info["initial_deposit"],
                direction=TransferDirection.INCOMING
            )
            
            deposit = self.client.create_ach_transfer(account_id, deposit_request)
            print(f"✅ Initial deposit initiated: ${deposit.amount}")
        
        return str(ach_relationship.id)
    
    def create_portfolio_strategy(self, strategy_name: str, allocations: dict) -> str:
        """Create rebalancing portfolio strategy."""
        
        print(f"📊 Creating portfolio strategy: {strategy_name}")
        
        portfolio_request = CreatePortfolioRequest(
            name=strategy_name,
            description=f"Automated {strategy_name} allocation",
            weights=allocations,
            weight_type=WeightType.PERCENT,
            cooldown_days=30
        )
        
        portfolio = self.client.create_portfolio(portfolio_request)
        print(f"✅ Portfolio strategy created: {portfolio.id}")
        
        return str(portfolio.id)
    
    def subscribe_to_portfolio(self, account_id: str, portfolio_id: str, max_investment: float) -> None:
        """Subscribe customer to portfolio strategy."""
        
        subscription_request = CreateSubscriptionRequest(
            account_id=account_id,
            portfolio_id=portfolio_id,
            max_investment_amount=max_investment,
            cash_percentage=0.05  # Keep 5% cash
        )
        
        subscription = self.client.create_subscription(subscription_request)
        print(f"✅ Account subscribed to portfolio: {subscription.id}")
    
    def execute_trade(self, account_id: str, symbol: str, quantity: float, side: str) -> None:
        """Execute trade for customer account."""
        
        order_request = MarketOrderRequest(
            symbol=symbol,
            qty=quantity,
            side=OrderSide.BUY if side.upper() == "BUY" else OrderSide.SELL,
            time_in_force=TimeInForce.DAY
        )
        
        order = self.client.submit_order(account_id, order_request)
        print(f"📈 Order submitted: {side} {quantity} {symbol} (ID: {order.id})")
    
    def get_customer_summary(self, account_id: str) -> dict:
        """Get comprehensive customer account summary."""
        
        # Get account details
        account = self.client.get_account_by_id(account_id)
        
        # Get funding relationships
        ach_relationships = self.client.get_ach_relationships(account_id)
        
        # Get recent transfers
        transfers = self.client.get_transfers(account_id)
        recent_transfers = [t for t in transfers if t.requested_at > datetime.now() - timedelta(days=30)]
        
        # Get orders
        orders = self.client.get_orders_for_account(account_id)
        
        return {
            "account_number": account.account_number,
            "status": account.status,
            "ach_relationships": len(ach_relationships),
            "recent_transfers": len(recent_transfers),
            "total_orders": len(orders),
            "created_at": account.created_at
        }

# Example usage
if __name__ == "__main__":
    platform = BrokerPlatform(
        api_key="sandbox-api-key",
        secret_key="sandbox-secret-key",
        sandbox=True
    )
    
    # Customer onboarding
    customer_data = {
        "first_name": "Jane",
        "last_name": "Smith", 
        "email": "jane.smith@example.com",
        "phone": "+1-555-123-4567",
        "address": "123 Investment St",
        "city": "New York",
        "state": "NY", 
        "zip_code": "10001",
        "date_of_birth": date(1985, 3, 15),
        "ssn": "123456789",
        "income_min": 75000,
        "income_max": 100000
    }
    
    account_id = platform.onboard_customer(customer_data)
    
    # Set up funding
    bank_data = {
        "owner_name": "Jane Smith",
        "account_number": "1234567890",
        "routing_number": "021000021",
        "initial_deposit": 10000.00
    }
    
    ach_id = platform.setup_funding(account_id, bank_data)
    
    # Create portfolio strategy
    conservative_allocation = {
        "VTI": 0.40,    # Total Stock Market - 40%
        "VXUS": 0.20,   # International Stocks - 20%  
        "BND": 0.30,    # Bonds - 30%
        "VNQ": 0.10     # REITs - 10%
    }
    
    portfolio_id = platform.create_portfolio_strategy(
        "Conservative Growth", 
        conservative_allocation
    )
    
    # Subscribe to portfolio
    platform.subscribe_to_portfolio(account_id, portfolio_id, 8000.00)
    
    # Execute some trades
    platform.execute_trade(account_id, "AAPL", 10, "BUY")
    platform.execute_trade(account_id, "MSFT", 15, "BUY")
    
    # Get summary
    summary = platform.get_customer_summary(account_id)
    print(f"\n📋 Customer Summary:")
    print(f"Account: {summary['account_number']}")
    print(f"Status: {summary['status']}")
    print(f"ACH Relationships: {summary['ach_relationships']}")
    print(f"Recent Transfers: {summary['recent_transfers']}")
    print(f"Total Orders: {summary['total_orders']}")
    print(f"Created: {summary['created_at'].date()}")

Install with Tessl CLI

npx tessl i tessl/pypi-alpaca-py

docs

broker-client.md

data-client.md

index.md

trading-client.md

tile.json