Skip to content

schemas.balance_snapshot_schemas

src.schemas.balance_snapshot_schemas

Balance snapshot request and response schemas.

Pydantic schemas for balance snapshot API endpoints. Includes: - Request schemas (client → API) - Response schemas (API → client) - DTO-to-schema conversion methods

Reference
  • docs/architecture/api-design-patterns.md

Classes

BalanceSnapshotResponse

Bases: BaseModel

Single balance snapshot response.

Attributes:

Name Type Description
id UUID

Snapshot unique identifier.
account_id UUID

Account FK.
balance Decimal

Total account balance.
available_balance Decimal | None

Available balance (nullable).
holdings_value Decimal | None

Total holdings value (nullable).
cash_value Decimal | None

Cash balance (nullable).
currency str

ISO 4217 currency code.
source str

How snapshot was captured.
captured_at datetime

When balance was captured.
created_at datetime

Record creation timestamp.
change_amount Decimal | None

Change from previous snapshot (nullable).
change_percent float | None

Percentage change (nullable).
Source code in src/schemas/balance_snapshot_schemas.py 
30
31
32
33
34
35
36
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
class BalanceSnapshotResponse(BaseModel):
    """Single balance snapshot response.

    Attributes:
        id: Snapshot unique identifier.
        account_id: Account FK.
        balance: Total account balance.
        available_balance: Available balance (nullable).
        holdings_value: Total holdings value (nullable).
        cash_value: Cash balance (nullable).
        currency: ISO 4217 currency code.
        source: How snapshot was captured.
        captured_at: When balance was captured.
        created_at: Record creation timestamp.
        change_amount: Change from previous snapshot (nullable).
        change_percent: Percentage change (nullable).
    """

    id: UUID = Field(..., description="Snapshot unique identifier")
    account_id: UUID = Field(..., description="Account FK")
    balance: Decimal = Field(..., description="Total account balance")
    available_balance: Decimal | None = Field(
        None, description="Available balance (may differ from total)"
    )
    holdings_value: Decimal | None = Field(
        None, description="Total market value of holdings"
    )
    cash_value: Decimal | None = Field(None, description="Cash/money market balance")
    currency: str = Field(..., description="ISO 4217 currency code", examples=["USD"])
    source: str = Field(
        ...,
        description="How snapshot was captured",
        examples=["account_sync", "manual_sync", "scheduled_sync"],
    )
    captured_at: datetime = Field(..., description="When balance was captured")
    created_at: datetime = Field(..., description="Record creation timestamp")
    change_amount: Decimal | None = Field(
        None, description="Change from previous snapshot"
    )
    change_percent: float | None = Field(
        None, description="Percentage change from previous"
    )

    @classmethod
    def from_dto(cls, dto: BalanceSnapshotResult) -> "BalanceSnapshotResponse":
        """Convert application DTO to response schema.

        Args:
            dto: BalanceSnapshotResult from handler.

        Returns:
            BalanceSnapshotResponse for API response.
        """
        return cls(
            id=dto.id,
            account_id=dto.account_id,
            balance=dto.balance,
            available_balance=dto.available_balance,
            holdings_value=dto.holdings_value,
            cash_value=dto.cash_value,
            currency=dto.currency,
            source=dto.source,
            captured_at=dto.captured_at,
            created_at=dto.created_at,
            change_amount=dto.change_amount,
            change_percent=dto.change_percent,
        )
Functions
from_dto classmethod 
from_dto(
    dto: BalanceSnapshotResult,
) -> BalanceSnapshotResponse

Convert application DTO to response schema.

Parameters:

Name Type Description Default
dto BalanceSnapshotResult

BalanceSnapshotResult from handler.

 required

Returns:

Type Description
BalanceSnapshotResponse

BalanceSnapshotResponse for API response.
Source code in src/schemas/balance_snapshot_schemas.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
@classmethod
def from_dto(cls, dto: BalanceSnapshotResult) -> "BalanceSnapshotResponse":
    """Convert application DTO to response schema.

    Args:
        dto: BalanceSnapshotResult from handler.

    Returns:
        BalanceSnapshotResponse for API response.
    """
    return cls(
        id=dto.id,
        account_id=dto.account_id,
        balance=dto.balance,
        available_balance=dto.available_balance,
        holdings_value=dto.holdings_value,
        cash_value=dto.cash_value,
        currency=dto.currency,
        source=dto.source,
        captured_at=dto.captured_at,
        created_at=dto.created_at,
        change_amount=dto.change_amount,
        change_percent=dto.change_percent,
    )

BalanceHistoryResponse

Bases: BaseModel

Balance history response for charting.

Includes computed metrics for change tracking.

Attributes:

Name Type Description
snapshots list[BalanceSnapshotResponse]

List of snapshot responses (ordered by time).
total_count int

Total number of snapshots in range.
start_balance Decimal | None

Balance at start of period (nullable).
end_balance Decimal | None

Balance at end of period (nullable).
total_change_amount Decimal | None

Change over period (nullable).
total_change_percent float | None

Percentage change over period (nullable).
currency str | None

Currency of the values (nullable).
Source code in src/schemas/balance_snapshot_schemas.py 
 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
141
142
143
144
145
146
147
148
class BalanceHistoryResponse(BaseModel):
    """Balance history response for charting.

    Includes computed metrics for change tracking.

    Attributes:
        snapshots: List of snapshot responses (ordered by time).
        total_count: Total number of snapshots in range.
        start_balance: Balance at start of period (nullable).
        end_balance: Balance at end of period (nullable).
        total_change_amount: Change over period (nullable).
        total_change_percent: Percentage change over period (nullable).
        currency: Currency of the values (nullable).
    """

    snapshots: list[BalanceSnapshotResponse] = Field(
        ..., description="List of balance snapshots"
    )
    total_count: int = Field(..., description="Total snapshot count")
    start_balance: Decimal | None = Field(
        None, description="Balance at start of period"
    )
    end_balance: Decimal | None = Field(None, description="Balance at end of period")
    total_change_amount: Decimal | None = Field(
        None, description="Total change over period"
    )
    total_change_percent: float | None = Field(
        None, description="Percentage change over period"
    )
    currency: str | None = Field(None, description="Currency of values")

    @classmethod
    def from_dto(cls, dto: BalanceHistoryResult) -> "BalanceHistoryResponse":
        """Convert application DTO to response schema.

        Args:
            dto: BalanceHistoryResult from handler.

        Returns:
            BalanceHistoryResponse for API response.
        """
        return cls(
            snapshots=[BalanceSnapshotResponse.from_dto(s) for s in dto.snapshots],
            total_count=dto.total_count,
            start_balance=dto.start_balance,
            end_balance=dto.end_balance,
            total_change_amount=dto.total_change_amount,
            total_change_percent=dto.total_change_percent,
            currency=dto.currency,
        )
Functions
from_dto classmethod 
from_dto(
    dto: BalanceHistoryResult,
) -> BalanceHistoryResponse

Convert application DTO to response schema.

Parameters:

Name Type Description Default
dto BalanceHistoryResult

BalanceHistoryResult from handler.

 required

Returns:

Type Description
BalanceHistoryResponse

BalanceHistoryResponse for API response.
Source code in src/schemas/balance_snapshot_schemas.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
@classmethod
def from_dto(cls, dto: BalanceHistoryResult) -> "BalanceHistoryResponse":
    """Convert application DTO to response schema.

    Args:
        dto: BalanceHistoryResult from handler.

    Returns:
        BalanceHistoryResponse for API response.
    """
    return cls(
        snapshots=[BalanceSnapshotResponse.from_dto(s) for s in dto.snapshots],
        total_count=dto.total_count,
        start_balance=dto.start_balance,
        end_balance=dto.end_balance,
        total_change_amount=dto.total_change_amount,
        total_change_percent=dto.total_change_percent,
        currency=dto.currency,
    )

LatestSnapshotsResponse

Bases: BaseModel

Latest snapshots response for portfolio summary.

Attributes:

Name Type Description
snapshots list[BalanceSnapshotResponse]

List of latest snapshot responses (one per account).
total_count int

Number of accounts with snapshots.
total_balance_by_currency dict[str, str]

Aggregate balance by currency.
Source code in src/schemas/balance_snapshot_schemas.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
class LatestSnapshotsResponse(BaseModel):
    """Latest snapshots response for portfolio summary.

    Attributes:
        snapshots: List of latest snapshot responses (one per account).
        total_count: Number of accounts with snapshots.
        total_balance_by_currency: Aggregate balance by currency.
    """

    snapshots: list[BalanceSnapshotResponse] = Field(
        ..., description="Latest snapshot per account"
    )
    total_count: int = Field(..., description="Number of accounts with snapshots")
    total_balance_by_currency: dict[str, str] = Field(
        ..., description="Aggregate balance by currency"
    )

    @classmethod
    def from_dto(cls, dto: LatestSnapshotsResult) -> "LatestSnapshotsResponse":
        """Convert application DTO to response schema.

        Args:
            dto: LatestSnapshotsResult from handler.

        Returns:
            LatestSnapshotsResponse for API response.
        """
        return cls(
            snapshots=[BalanceSnapshotResponse.from_dto(s) for s in dto.snapshots],
            total_count=dto.total_count,
            total_balance_by_currency=dto.total_balance_by_currency,
        )
Functions
from_dto classmethod 
from_dto(
    dto: LatestSnapshotsResult,
) -> LatestSnapshotsResponse

Convert application DTO to response schema.

Parameters:

Name Type Description Default
dto LatestSnapshotsResult

LatestSnapshotsResult from handler.

 required

Returns:

Type Description
LatestSnapshotsResponse

LatestSnapshotsResponse for API response.
Source code in src/schemas/balance_snapshot_schemas.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
@classmethod
def from_dto(cls, dto: LatestSnapshotsResult) -> "LatestSnapshotsResponse":
    """Convert application DTO to response schema.

    Args:
        dto: LatestSnapshotsResult from handler.

    Returns:
        LatestSnapshotsResponse for API response.
    """
    return cls(
        snapshots=[BalanceSnapshotResponse.from_dto(s) for s in dto.snapshots],
        total_count=dto.total_count,
        total_balance_by_currency=dto.total_balance_by_currency,
    )

BalanceHistoryRequest

Bases: BaseModel

Request parameters for balance history query.

Attributes:

Name Type Description
start_date datetime

Start of date range (inclusive).
end_date datetime

End of date range (inclusive).
source str | None

Optional filter by snapshot source.
Source code in src/schemas/balance_snapshot_schemas.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
class BalanceHistoryRequest(BaseModel):
    """Request parameters for balance history query.

    Attributes:
        start_date: Start of date range (inclusive).
        end_date: End of date range (inclusive).
        source: Optional filter by snapshot source.
    """

    start_date: datetime = Field(..., description="Start of date range")
    end_date: datetime = Field(..., description="End of date range")
    source: str | None = Field(
        None,
        description="Filter by snapshot source",
        examples=["account_sync", "manual_sync"],
    )