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.
92 lines
2.8 KiB
Python
92 lines
2.8 KiB
Python
"""
|
|
Insider transaction schemas
|
|
"""
|
|
|
|
from datetime import date
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
|
|
|
|
TRANSACTION_CODE_MAP = {
|
|
"P": "Purchase",
|
|
"S": "Sale",
|
|
"A": "Award/Grant",
|
|
"M": "Exercise/Conversion",
|
|
"G": "Gift",
|
|
"D": "Disposition to Issuer",
|
|
"F": "Tax Withholding",
|
|
"C": "Conversion",
|
|
"J": "Other",
|
|
}
|
|
|
|
|
|
class InsiderTransactionEntry(BaseModel):
|
|
model_config = ConfigDict(from_attributes=True)
|
|
|
|
filing_date: date
|
|
transaction_date: date
|
|
owner_name: str
|
|
owner_cik: Optional[str] = None
|
|
is_officer: bool = False
|
|
is_director: bool = False
|
|
is_ten_percent_owner: bool = False
|
|
officer_title: Optional[str] = None
|
|
security_title: Optional[str] = None
|
|
transaction_code: str
|
|
transaction_type: str # human-readable
|
|
shares: float
|
|
price_per_share: Optional[float] = None
|
|
total_value: Optional[float] = None
|
|
shares_owned_after: Optional[float] = None
|
|
is_derivative: bool = False
|
|
|
|
@classmethod
|
|
def from_orm_obj(cls, obj) -> "InsiderTransactionEntry":
|
|
return cls(
|
|
filing_date=obj.filing_date.date() if hasattr(obj.filing_date, "date") else obj.filing_date,
|
|
transaction_date=obj.transaction_date.date() if hasattr(obj.transaction_date, "date") else obj.transaction_date,
|
|
owner_name=obj.owner_name,
|
|
owner_cik=obj.owner_cik,
|
|
is_officer=obj.is_officer or False,
|
|
is_director=obj.is_director or False,
|
|
is_ten_percent_owner=obj.is_ten_percent_owner or False,
|
|
officer_title=obj.officer_title,
|
|
security_title=obj.security_title,
|
|
transaction_code=obj.transaction_code,
|
|
transaction_type=TRANSACTION_CODE_MAP.get(obj.transaction_code, obj.transaction_code),
|
|
shares=obj.shares,
|
|
price_per_share=obj.price_per_share,
|
|
total_value=obj.total_value,
|
|
shares_owned_after=obj.shares_owned_after,
|
|
is_derivative=obj.is_derivative or False,
|
|
)
|
|
|
|
|
|
class InsiderTransactionResponse(BaseModel):
|
|
symbol: str
|
|
transactions: List[InsiderTransactionEntry]
|
|
total_count: int
|
|
metadata: Dict[str, Any] = Field(default_factory=dict)
|
|
|
|
|
|
class InsiderSummaryPeriod(BaseModel):
|
|
period_label: str
|
|
buy_count: int = 0
|
|
sell_count: int = 0
|
|
buy_shares: float = 0.0
|
|
sell_shares: float = 0.0
|
|
buy_value: float = 0.0
|
|
sell_value: float = 0.0
|
|
net_shares: float = 0.0
|
|
net_value: float = 0.0
|
|
unique_buyers: int = 0
|
|
unique_sellers: int = 0
|
|
|
|
|
|
class InsiderSummaryResponse(BaseModel):
|
|
symbol: str
|
|
periods: List[InsiderSummaryPeriod]
|
|
notable_transactions: List[InsiderTransactionEntry]
|
|
metadata: Dict[str, Any] = Field(default_factory=dict)
|