Balance Tracking Architecture¶

BalanceSnapshot (frozen=True)
├── id: UUID                           # Unique identifier
├── account_id: UUID                   # FK to Account
├── balance: Money                     # Total account balance
├── available_balance: Money | None    # Available for trading/withdrawal
├── holdings_value: Money | None       # Total securities value
├── cash_value: Money | None           # Cash/money market balance
├── currency: str                      # ISO 4217 code
├── source: SnapshotSource             # How/why captured
├── provider_metadata: dict | None     # Raw provider data at capture
├── captured_at: datetime              # When balance was captured
└── created_at: datetime               # Record creation timestamp

┌─────────────────────────────────────────┐
│            Total Balance                │
│           (balance.amount)              │
├───────────────────┬─────────────────────┤
│   Holdings Value  │     Cash Value      │
│ (holdings_value)  │   (cash_value)      │
│                   │                     │
│  Stocks, ETFs,    │  Money Market,      │
│  Bonds, etc.      │  Sweep Account      │
└───────────────────┴─────────────────────┘

Relationship: balance ≈ holdings_value + cash_value

class SnapshotSource(StrEnum):
    """Source/trigger of a balance snapshot."""

    ACCOUNT_SYNC = "account_sync"
    """Captured during account data sync."""

    HOLDINGS_SYNC = "holdings_sync"
    """Captured during holdings sync operation."""

    MANUAL_SYNC = "manual_sync"
    """User-initiated sync request."""

    SCHEDULED_SYNC = "scheduled_sync"
    """Automated background sync job."""

    INITIAL_CONNECTION = "initial_connection"
    """First sync after provider connection."""

    # Query Methods
    def is_automated(self) -> bool:
        """Check if snapshot was captured automatically."""
        return self in {ACCOUNT_SYNC, HOLDINGS_SYNC, SCHEDULED_SYNC}

    def is_user_initiated(self) -> bool:
        """Check if snapshot was triggered by user action."""
        return self in {MANUAL_SYNC, INITIAL_CONNECTION}

class BalanceSnapshot:
    def has_value_breakdown(self) -> bool:
        """Check if both holdings_value and cash_value are present."""
        return self.holdings_value is not None and self.cash_value is not None

    def get_holdings_percentage(self) -> float | None:
        """Calculate percentage of portfolio in holdings.

        Returns:
            Percentage (0-100) or None if breakdown unavailable.
        """

    def get_cash_percentage(self) -> float | None:
        """Calculate percentage of portfolio in cash."""

    def is_automated_capture(self) -> bool:
        """Check if snapshot was captured automatically."""
        return self.source.is_automated()

    def is_user_initiated_capture(self) -> bool:
        """Check if snapshot was triggered by user."""
        return self.source.is_user_initiated()

    def calculate_change_from(
        self, previous: "BalanceSnapshot"
    ) -> tuple[Money, float] | None:
        """Calculate change from previous snapshot.

        Returns:
            Tuple of (change_amount, change_percent), or None if currencies differ.
        """

class BalanceSnapshotRepository(Protocol):
    """Port for balance snapshot persistence."""

    async def find_by_id(self, snapshot_id: UUID) -> BalanceSnapshot | None:
        """Find snapshot by ID."""

    async def find_by_account_id(
        self,
        account_id: UUID,
        limit: int | None = None,
    ) -> list[BalanceSnapshot]:
        """Find snapshots for account (newest first)."""

    async def find_by_account_and_date_range(
        self,
        account_id: UUID,
        start_date: datetime,
        end_date: datetime,
    ) -> list[BalanceSnapshot]:
        """Find snapshots within date range (chronological)."""

    async def find_latest_by_account(
        self, account_id: UUID
    ) -> BalanceSnapshot | None:
        """Find most recent snapshot for account."""

    async def find_latest_by_user(
        self, user_id: UUID
    ) -> list[BalanceSnapshot]:
        """Find latest snapshot for each of user's accounts."""

    async def save(self, snapshot: BalanceSnapshot) -> None:
        """Persist snapshot (insert only - no updates)."""

query = GetBalanceHistory(
    account_id=account_id,
    user_id=user_id,
    start_date=datetime(2024, 1, 1),
    end_date=datetime(2024, 12, 31),
)
result = await handler.handle(query)

# Result contains:
# - snapshots: List ordered chronologically
# - start_balance: First snapshot balance
# - end_balance: Last snapshot balance
# - total_change_amount: end - start
# - total_change_percent: percentage change

query = GetLatestBalanceSnapshots(user_id=user_id)
result = await handler.handle(query)

# Result contains:
# - snapshots: One per account (most recent)
# - total_count: Number of accounts
# - total_balance_by_currency: Aggregated balances

# Get first and last snapshot in period
first = snapshots[0]
last = snapshots[-1]

change, percent = last.calculate_change_from(first)
print(f"Period return: {change.amount} ({percent:.2f}%)")

for snapshot in snapshots:
    if snapshot.has_value_breakdown():
        holdings_pct = snapshot.get_holdings_percentage()
        cash_pct = snapshot.get_cash_percentage()
        print(f"{snapshot.captured_at}: {holdings_pct}% holdings, {cash_pct}% cash")

# In SyncAccountsHandler or SyncHoldingsHandler
from uuid_extensions import uuid7
from src.domain.entities import BalanceSnapshot
from src.domain.enums import SnapshotSource

# After syncing data, capture snapshot
snapshot = BalanceSnapshot(
    id=uuid7(),
    account_id=account.id,
    balance=account.balance,
    available_balance=account.available_balance,
    holdings_value=total_holdings_market_value,  # From sum of holdings
    cash_value=account.cash_balance,
    currency=account.currency,
    source=SnapshotSource.HOLDINGS_SYNC,
    provider_metadata={"sync_id": sync_id, "provider": connection.provider_slug},
)
await snapshot_repo.save(snapshot)

src/domain/
├── entities/
│   └── balance_snapshot.py         # BalanceSnapshot entity
├── enums/
│   └── snapshot_source.py          # SnapshotSource enum
├── errors/
│   └── balance_snapshot_error.py   # Error constants
└── protocols/
    └── balance_snapshot_repository.py  # Repository protocol

src/infrastructure/
└── persistence/
    ├── models/
    │   └── balance_snapshot.py     # SQLAlchemy model
    └── repositories/
        └── balance_snapshot_repository.py  # PostgreSQL adapter

src/application/
├── queries/
│   └── balance_snapshot_queries.py  # Query definitions
└── handlers/
    └── balance_snapshot_handlers.py  # Query handlers

tests/
├── unit/
│   └── test_domain_balance_snapshot.py
└── integration/
    └── test_balance_snapshot_repository.py

-- Essential indexes for balance_snapshots table
CREATE INDEX idx_balance_snapshots_account_id 
    ON balance_snapshots(account_id);

CREATE INDEX idx_balance_snapshots_captured_at 
    ON balance_snapshots(captured_at DESC);

CREATE INDEX idx_balance_snapshots_account_captured 
    ON balance_snapshots(account_id, captured_at DESC);