Skip to content

schemas.holding_schemas

src.schemas.holding_schemas

Holding request and response schemas.

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

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

Classes

HoldingResponse

Bases: BaseModel

Single holding response.

Attributes:

Name Type Description
id UUID

Holding unique identifier.
account_id UUID

Account FK.
provider_holding_id str

Provider's unique ID for position.
symbol str

Security ticker symbol.
security_name str

Full security name.
asset_type str

Asset type (equity, etf, option, etc.).
quantity Decimal

Number of shares/units.
cost_basis Decimal

Total cost paid.
market_value Decimal

Current market value.
currency str

ISO 4217 currency code.
average_price Decimal | None

Average cost per share.
current_price Decimal | None

Current market price per share.
unrealized_gain_loss Decimal | None

Computed gain/loss.
unrealized_gain_loss_percent Decimal | None

Computed gain/loss percentage.
is_active bool

Whether position is active.
is_profitable bool

Whether position is profitable.
last_synced_at datetime | None

Last sync timestamp.
created_at datetime

Creation timestamp.
updated_at datetime

Last update timestamp.
Source code in src/schemas/holding_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
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
class HoldingResponse(BaseModel):
    """Single holding response.

    Attributes:
        id: Holding unique identifier.
        account_id: Account FK.
        provider_holding_id: Provider's unique ID for position.
        symbol: Security ticker symbol.
        security_name: Full security name.
        asset_type: Asset type (equity, etf, option, etc.).
        quantity: Number of shares/units.
        cost_basis: Total cost paid.
        market_value: Current market value.
        currency: ISO 4217 currency code.
        average_price: Average cost per share.
        current_price: Current market price per share.
        unrealized_gain_loss: Computed gain/loss.
        unrealized_gain_loss_percent: Computed gain/loss percentage.
        is_active: Whether position is active.
        is_profitable: Whether position is profitable.
        last_synced_at: Last sync timestamp.
        created_at: Creation timestamp.
        updated_at: Last update timestamp.
    """

    id: UUID = Field(..., description="Holding unique identifier")
    account_id: UUID = Field(..., description="Account FK")
    provider_holding_id: str = Field(
        ..., description="Provider's unique ID for position"
    )
    symbol: str = Field(..., description="Security ticker symbol", examples=["AAPL"])
    security_name: str = Field(
        ..., description="Full security name", examples=["Apple Inc."]
    )
    asset_type: str = Field(
        ...,
        description="Asset type",
        examples=["equity", "etf", "option", "mutual_fund"],
    )
    quantity: Decimal = Field(..., description="Number of shares/units")
    cost_basis: Decimal = Field(..., description="Total cost paid")
    market_value: Decimal = Field(..., description="Current market value")
    currency: str = Field(..., description="ISO 4217 currency code", examples=["USD"])
    average_price: Decimal | None = Field(None, description="Average cost per share")
    current_price: Decimal | None = Field(
        None, description="Current market price per share"
    )
    unrealized_gain_loss: Decimal | None = Field(
        None, description="Computed unrealized gain/loss"
    )
    unrealized_gain_loss_percent: Decimal | None = Field(
        None, description="Computed gain/loss percentage"
    )
    is_active: bool = Field(..., description="Whether position is active")
    is_profitable: bool = Field(..., description="Whether position is profitable")
    last_synced_at: datetime | None = Field(None, description="Last sync timestamp")
    created_at: datetime = Field(..., description="Creation timestamp")
    updated_at: datetime = Field(..., description="Last update timestamp")

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

        Args:
            dto: HoldingResult from handler.

        Returns:
            HoldingResponse for API response.
        """
        return cls(
            id=dto.id,
            account_id=dto.account_id,
            provider_holding_id=dto.provider_holding_id,
            symbol=dto.symbol,
            security_name=dto.security_name,
            asset_type=dto.asset_type,
            quantity=dto.quantity,
            cost_basis=dto.cost_basis,
            market_value=dto.market_value,
            currency=dto.currency,
            average_price=dto.average_price,
            current_price=dto.current_price,
            unrealized_gain_loss=dto.unrealized_gain_loss,
            unrealized_gain_loss_percent=dto.unrealized_gain_loss_percent,
            is_active=dto.is_active,
            is_profitable=dto.is_profitable,
            last_synced_at=dto.last_synced_at,
            created_at=dto.created_at,
            updated_at=dto.updated_at,
        )
Functions
from_dto classmethod 
from_dto(dto: HoldingResult) -> HoldingResponse

Convert application DTO to response schema.

Parameters:

Name Type Description Default
dto HoldingResult

HoldingResult from handler.

 required

Returns:

Type Description
HoldingResponse

HoldingResponse for API response.
Source code in src/schemas/holding_schemas.py 
 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
@classmethod
def from_dto(cls, dto: HoldingResult) -> "HoldingResponse":
    """Convert application DTO to response schema.

    Args:
        dto: HoldingResult from handler.

    Returns:
        HoldingResponse for API response.
    """
    return cls(
        id=dto.id,
        account_id=dto.account_id,
        provider_holding_id=dto.provider_holding_id,
        symbol=dto.symbol,
        security_name=dto.security_name,
        asset_type=dto.asset_type,
        quantity=dto.quantity,
        cost_basis=dto.cost_basis,
        market_value=dto.market_value,
        currency=dto.currency,
        average_price=dto.average_price,
        current_price=dto.current_price,
        unrealized_gain_loss=dto.unrealized_gain_loss,
        unrealized_gain_loss_percent=dto.unrealized_gain_loss_percent,
        is_active=dto.is_active,
        is_profitable=dto.is_profitable,
        last_synced_at=dto.last_synced_at,
        created_at=dto.created_at,
        updated_at=dto.updated_at,
    )

HoldingListResponse

Bases: BaseModel

Holding list response with aggregates.

Attributes:

Name Type Description
holdings list[HoldingResponse]

List of holdings.
total_count int

Total number of holdings.
active_count int

Number of active holdings.
total_market_value_by_currency dict[str, str]

Aggregated market values.
total_cost_basis_by_currency dict[str, str]

Aggregated cost basis.
total_unrealized_gain_loss_by_currency dict[str, str]

Aggregated gain/loss.
Source code in src/schemas/holding_schemas.py 
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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
class HoldingListResponse(BaseModel):
    """Holding list response with aggregates.

    Attributes:
        holdings: List of holdings.
        total_count: Total number of holdings.
        active_count: Number of active holdings.
        total_market_value_by_currency: Aggregated market values.
        total_cost_basis_by_currency: Aggregated cost basis.
        total_unrealized_gain_loss_by_currency: Aggregated gain/loss.
    """

    holdings: list[HoldingResponse] = Field(..., description="List of holdings")
    total_count: int = Field(..., description="Total holding count")
    active_count: int = Field(..., description="Active holding count")
    total_market_value_by_currency: dict[str, str] = Field(
        ..., description="Aggregated market values by currency"
    )
    total_cost_basis_by_currency: dict[str, str] = Field(
        ..., description="Aggregated cost basis by currency"
    )
    total_unrealized_gain_loss_by_currency: dict[str, str] = Field(
        ..., description="Aggregated unrealized gain/loss by currency"
    )

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

        Args:
            dto: HoldingListResult from handler.

        Returns:
            HoldingListResponse for API response.
        """
        return cls(
            holdings=[HoldingResponse.from_dto(h) for h in dto.holdings],
            total_count=dto.total_count,
            active_count=dto.active_count,
            total_market_value_by_currency=dto.total_market_value_by_currency,
            total_cost_basis_by_currency=dto.total_cost_basis_by_currency,
            total_unrealized_gain_loss_by_currency=dto.total_unrealized_gain_loss_by_currency,
        )
Functions
from_dto classmethod 
from_dto(dto: HoldingListResult) -> HoldingListResponse

Convert application DTO to response schema.

Parameters:

Name Type Description Default
dto HoldingListResult

HoldingListResult from handler.

 required

Returns:

Type Description
HoldingListResponse

HoldingListResponse for API response.
Source code in src/schemas/holding_schemas.py 
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
@classmethod
def from_dto(cls, dto: HoldingListResult) -> "HoldingListResponse":
    """Convert application DTO to response schema.

    Args:
        dto: HoldingListResult from handler.

    Returns:
        HoldingListResponse for API response.
    """
    return cls(
        holdings=[HoldingResponse.from_dto(h) for h in dto.holdings],
        total_count=dto.total_count,
        active_count=dto.active_count,
        total_market_value_by_currency=dto.total_market_value_by_currency,
        total_cost_basis_by_currency=dto.total_cost_basis_by_currency,
        total_unrealized_gain_loss_by_currency=dto.total_unrealized_gain_loss_by_currency,
    )

SyncHoldingsResponse

Bases: SyncResponse

Response for holdings sync operation.

Extends SyncResponse with holdings-specific fields.

Attributes:

Name Type Description
holdings_created int

Number of new holdings created.
holdings_updated int

Number of existing holdings updated.
holdings_deactivated int

Number of holdings deactivated.
Source code in src/schemas/holding_schemas.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
class SyncHoldingsResponse(SyncResponse):
    """Response for holdings sync operation.

    Extends SyncResponse with holdings-specific fields.

    Attributes:
        holdings_created: Number of new holdings created.
        holdings_updated: Number of existing holdings updated.
        holdings_deactivated: Number of holdings deactivated.
    """

    holdings_created: int = Field(0, description="New holdings created")
    holdings_updated: int = Field(0, description="Existing holdings updated")
    holdings_deactivated: int = Field(0, description="Holdings deactivated (sold)")

SyncHoldingsRequest

Bases: BaseModel

Request to sync holdings from provider.

Attributes:

Name Type Description
force bool

Force sync even if recently synced.
Source code in src/schemas/holding_schemas.py 
188
189
190
191
192
193
194
195
class SyncHoldingsRequest(BaseModel):
    """Request to sync holdings from provider.

    Attributes:
        force: Force sync even if recently synced.
    """

    force: bool = Field(False, description="Force sync even if recently synced")