stock-oracle/app/schemas/insider.py

"""
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)


# ------------------------------------------------------------------
# New PIT-safe Form 4 schemas
# ------------------------------------------------------------------

class Form4Entry(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    symbol: str
    filing_date: date
    transaction_date: date
    owner_cik: Optional[str] = None
    owner_name: str
    owner_relationship: Optional[str] = None
    is_officer: bool = False
    is_director: bool = False
    is_ten_percent_owner: bool = False
    is_ceo: bool = False
    is_cfo: bool = False
    is_c_suite: bool = False
    transaction_code: str
    transaction_type: str
    shares: float
    price: Optional[float] = None
    total_value: Optional[float] = None
    shares_owned_following: Optional[float] = None
    purchase_pct_of_holding: Optional[float] = None
    accession_number: str

    @classmethod
    def from_orm_obj(cls, obj) -> "Form4Entry":
        return cls(
            symbol=obj.ticker,
            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_cik=obj.owner_cik,
            owner_name=obj.owner_name,
            owner_relationship=obj.owner_relationship,
            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,
            is_ceo=obj.is_ceo or False,
            is_cfo=obj.is_cfo or False,
            is_c_suite=obj.is_c_suite or False,
            transaction_code=obj.transaction_code,
            transaction_type=TRANSACTION_CODE_MAP.get(obj.transaction_code, obj.transaction_code),
            shares=obj.shares,
            price=obj.price_per_share,
            total_value=obj.total_value,
            shares_owned_following=obj.shares_owned_after,
            purchase_pct_of_holding=obj.purchase_pct_of_holding,
            accession_number=obj.accession_number,
        )


class Form4Response(BaseModel):
    symbol: str
    as_of: date
    window: Dict[str, Any] = Field(default_factory=dict)
    transactions: List[Form4Entry]
    total_count: int
    metadata: Dict[str, Any] = Field(default_factory=dict)


class Form4ByDateResponse(BaseModel):
    filing_date: date
    buy_only: bool
    transactions: List[Form4Entry]
    total_count: int


class Form4AggregateResponse(BaseModel):
    """Aggregate Form 4 insider activity over a rolling window.

    All fields are computed over open-market purchases only (transaction_code='P',
    shares > 0, non-derivative). Awards/grants (A-code) are excluded.
    """
    symbol: str
    as_of: date
    window_days: int
    buy_count: int
    buy_dollar_total: float
    cluster_size: int
    csuite_count: int
    avg_pct_of_holding: Optional[float] = None
    recency_days: int
    metadata: Dict[str, Any] = Field(default_factory=dict)