You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
77 lines
2.3 KiB
Python
77 lines
2.3 KiB
Python
"""
|
|
PIT Dividend Calendar schemas
|
|
"""
|
|
|
|
from datetime import date, datetime
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
|
|
class DividendCalendarEntry(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
ticker: str
|
|
ex_dividend_date: date
|
|
amount: float
|
|
declaration_date: Optional[date] = None
|
|
record_date: Optional[date] = None
|
|
payment_date: Optional[date] = None
|
|
currency: str = "USD"
|
|
dividend_type: str = "regular"
|
|
frequency: Optional[str] = None
|
|
as_of_date: date
|
|
source: str
|
|
|
|
@classmethod
|
|
def from_orm_obj(cls, obj) -> "DividendCalendarEntry":
|
|
def _to_date(val):
|
|
if val is None:
|
|
return None
|
|
return val.date() if hasattr(val, "date") else val
|
|
|
|
return cls(
|
|
ticker=obj.ticker,
|
|
ex_dividend_date=_to_date(obj.ex_dividend_date),
|
|
amount=obj.amount,
|
|
declaration_date=_to_date(obj.declaration_date),
|
|
record_date=_to_date(obj.record_date),
|
|
payment_date=_to_date(obj.payment_date),
|
|
currency=obj.currency or "USD",
|
|
dividend_type=obj.dividend_type or "regular",
|
|
frequency=obj.frequency,
|
|
as_of_date=_to_date(obj.as_of_date),
|
|
source=obj.source,
|
|
)
|
|
|
|
|
|
class DividendUpcomingResponse(BaseModel):
|
|
"""Response for PIT upcoming dividends query."""
|
|
dividends: List[DividendCalendarEntry]
|
|
total_count: int
|
|
metadata: Dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class DividendHistoryResponse(BaseModel):
|
|
"""Response for single-symbol dividend history."""
|
|
symbol: str
|
|
dividends: List[DividendCalendarEntry]
|
|
total_count: int
|
|
annual_yield_estimate: Optional[float] = None
|
|
metadata: Dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class DividendIngestRequest(BaseModel):
|
|
"""Request body for bulk backfill ingest."""
|
|
symbols: List[str] = Field(..., min_length=1, max_length=200)
|
|
force_refresh: bool = Field(False, description="Re-ingest even if data exists")
|
|
|
|
|
|
class DividendIngestResponse(BaseModel):
|
|
"""Response for admin ingest endpoint."""
|
|
symbols_processed: int
|
|
total_records_upserted: int
|
|
failed_symbols: List[str] = Field(default_factory=list)
|
|
status: str
|
|
metadata: Dict[str, Any] = Field(default_factory=dict)
|