stock-oracle/app/services/activist_ownership_service.py

"""
SC 13D / 13G Activist Ownership Service

Phase 1 (index-only): ingest from SEC full-index company.idx
Phase 2 (background enrich): parse cover-page XML/HTML for ownership_pct/shares_owned
"""

import logging
import re
from datetime import date, datetime, timezone
from typing import Dict, List, Optional, Tuple

from sqlalchemy import and_, func, select, text
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession

from app.models.activist_ownership import ActivistOwnershipEvent
from app.services.sec_full_index_service import IndexEntry, SECFullIndexService
from app.services.sec_http_client import SECHttpClient

logger = logging.getLogger(__name__)

_CHUNK = 1500

# Regex patterns for cover-page HTML parsing
_RE_OWNERSHIP_PCT = re.compile(
    r"(?:Percent\s+of\s+Class\s+Represented\s+by\s+Amount\s+in\s+Row(?:\s*\(\d+\))?|"
    r"percent(?:age)?\s+of\s+class)[^\d]*(\d+\.?\d*)\s*%?",
    re.IGNORECASE,
)
_RE_SHARES_OWNED = re.compile(
    r"Aggregate\s+Amount\s+(?:Beneficially\s+)?Owned[^\d]*(\d[\d,]*)",
    re.IGNORECASE,
)

ACTIVIST_FORM_TYPES = {
    "SC 13D", "SC 13G", "SC 13D/A", "SC 13G/A",
    "SCHEDULE 13D", "SCHEDULE 13G", "SCHEDULE 13D/A", "SCHEDULE 13G/A",
}
ENRICH_BATCH_SIZE = 200


class ActivistOwnershipService:

    def __init__(self):
        self._http = SECHttpClient("Stock Oracle Activist Service")
        self._index_svc = SECFullIndexService()

    # ------------------------------------------------------------------
    # Phase 1: index-only ingest from full-index entries
    # ------------------------------------------------------------------

    async def ingest_from_index_entries(
        self,
        db: AsyncSession,
        entries: List[IndexEntry],
    ) -> int:
        """Upsert SC 13D/G rows from index entries (parse_status='index_only').

        Returns count of newly inserted rows.
        """
        if not entries:
            return 0

        acc_set = {e.accession_number for e in entries}
        existing = await db.execute(
            select(ActivistOwnershipEvent.accession_number).distinct().where(
                ActivistOwnershipEvent.accession_number.in_(acc_set)
            )
        )
        existing_accs = {r[0] for r in existing.fetchall()}
        # Commit immediately to release the DB connection before HTTP fetches.
        # Without this, the connection sits idle-in-transaction for the duration
        # of all HTTP calls (potentially minutes), triggering the 90s timeout.
        await db.commit()
        new_entries = [e for e in entries if e.accession_number not in existing_accs]

        if not new_entries:
            return 0

        # Resolve issuer CIK → ticker for each entry via SGML header parse.
        # For 13D/G, the company.idx filer is the BENEFICIAL OWNER (hedge fund).
        # The issuer (subject company) is in the filing's SGML SUBJECT COMPANY block.
        rows: List[Dict] = []
        for entry in new_entries:
            issuer_cik, symbol = await self._resolve_issuer(entry)
            if not issuer_cik or not symbol:
                continue

            filing_dt = datetime(
                entry.filing_date.year, entry.filing_date.month, entry.filing_date.day,
                tzinfo=timezone.utc
            )
            is_amend = entry.form_type.endswith("/A")
            filing_url = f"https://www.sec.gov/Archives/{entry.filename}"

            rows.append({
                "symbol": symbol,
                "issuer_cik": issuer_cik,
                "filer_cik": entry.cik,
                "filer_name": entry.company_name,
                "accession_number": entry.accession_number,
                "form_type": entry.form_type,
                "filing_date": filing_dt,
                "is_amendment": is_amend,
                "parse_status": "index_only",
                "filing_url": filing_url,
            })

        if not rows:
            return 0

        inserted = 0
        for i in range(0, len(rows), _CHUNK):
            chunk = rows[i:i + _CHUNK]
            stmt = pg_insert(ActivistOwnershipEvent).values(chunk)
            stmt = stmt.on_conflict_do_nothing(constraint="uq_activist_event")
            result = await db.execute(stmt)
            inserted += result.rowcount
        await db.commit()
        logger.info(f"Activist: inserted {inserted} new index-only rows")
        return inserted

    async def _resolve_issuer(self, entry: IndexEntry) -> Tuple[Optional[str], Optional[str]]:
        """Parse SGML header of the filing .txt to find subject company CIK, then ticker.

        SC 13D/G filings use flat-file SGML format: the SUBJECT COMPANY block in the
        first few KB contains CENTRAL INDEX KEY of the issuer.
        """
        txt_url = f"https://www.sec.gov/Archives/{entry.filename}"
        try:
            text = await self._http.fetch_text(txt_url)
            # Parse only the SGML header portion (~4KB is sufficient)
            header = text[:4000]
            in_subject = False
            issuer_cik = None
            for line in header.splitlines():
                if "SUBJECT COMPANY:" in line:
                    in_subject = True
                elif "FILED BY:" in line:
                    in_subject = False
                elif in_subject and "CENTRAL INDEX KEY:" in line:
                    raw_cik = line.split("CENTRAL INDEX KEY:")[1].strip()
                    if raw_cik:
                        issuer_cik = str(int(raw_cik)).zfill(10)
                    break
            if not issuer_cik:
                return None, None
            ticker = await self._http.get_ticker_for_cik(issuer_cik)
            return issuer_cik, ticker
        except Exception as e:
            logger.debug(f"Activist: could not resolve issuer for {entry.accession_number}: {e}")
            return None, None

    # ------------------------------------------------------------------
    # Phase 2: enrich index-only rows with ownership_pct / shares_owned
    # ------------------------------------------------------------------

    async def enrich_pending(self, db: AsyncSession, batch_size: int = ENRICH_BATCH_SIZE) -> int:
        """Enrich up to batch_size index-only rows with cover-page parse."""
        pending = await db.execute(
            select(ActivistOwnershipEvent)
            .where(ActivistOwnershipEvent.parse_status == "index_only")
            .order_by(ActivistOwnershipEvent.filing_date.desc())
            .limit(batch_size)
        )
        rows = pending.scalars().all()
        if not rows:
            return 0

        enriched = 0
        for row in rows:
            try:
                ownership_pct, shares_owned = await self._parse_cover_page(row)
                change_pct = None
                if ownership_pct is not None:
                    change_pct = await self._compute_change_pct(db, row, ownership_pct)

                row.ownership_pct = ownership_pct
                row.shares_owned = shares_owned
                row.change_pct = change_pct
                row.parse_status = "parsed"
                db.add(row)
                enriched += 1
            except Exception as e:
                logger.warning(f"Activist enrich failed {row.accession_number}: {e}")
                row.parse_status = "parse_failed"
                db.add(row)

        await db.commit()
        logger.info(f"Activist enrich: {enriched}/{len(rows)} rows enriched")
        return enriched

    async def _parse_cover_page(
        self, event: ActivistOwnershipEvent
    ) -> Tuple[Optional[float], Optional[float]]:
        """Best-effort parse of cover-page XML or HTML for ownership_pct and shares_owned."""
        if not event.filing_url:
            raise ValueError("No filing_url")

        # Build the primary document URL from filing index
        acc_clean = event.accession_number.replace("-", "")
        cik_int = int(event.filer_cik)
        idx_url = (
            f"https://www.sec.gov/Archives/edgar/data/{cik_int}/{acc_clean}"
            f"/{event.accession_number}-index.json"
        )
        try:
            idx_data = await self._http.fetch_json(idx_url)
            primary_doc = idx_data.get("primary_document", "")
            if not primary_doc:
                # Fall back to the first listed document
                docs = idx_data.get("documents", [])
                primary_doc = docs[0].get("document_url", "") if docs else ""
        except Exception:
            primary_doc = ""

        # Attempt XML cover page (post-Oct-2023 structured 13D/G)
        xml_url = (
            f"https://www.sec.gov/Archives/edgar/data/{cik_int}/{acc_clean}"
            "/primary_doc.xml"
        )
        try:
            xml_text = await self._http.fetch_text(xml_url, accept="application/xml", max_bytes=512_000)
            pct, shares = _parse_cover_xml(xml_text)
            if pct is not None or shares is not None:
                return pct, shares
        except Exception:
            pass

        # Fall back to HTML primary document
        if primary_doc:
            doc_url = primary_doc if primary_doc.startswith("http") else (
                f"https://www.sec.gov/Archives/edgar/data/{cik_int}/{acc_clean}/{primary_doc}"
            )
        else:
            doc_url = (
                f"https://www.sec.gov/Archives/edgar/data/{cik_int}/{acc_clean}"
                f"/{event.accession_number}.txt"
            )
        try:
            html = await self._http.fetch_text(doc_url, max_bytes=1_000_000)
            return _parse_cover_html(html)
        except Exception as e:
            raise ValueError(f"HTML parse failed: {e}") from e

    async def _compute_change_pct(
        self, db: AsyncSession, event: ActivistOwnershipEvent, new_pct: float
    ) -> Optional[float]:
        """Compute change in ownership_pct vs the prior filing by the same filer/issuer."""
        prev = await db.execute(
            select(ActivistOwnershipEvent.ownership_pct)
            .where(
                and_(
                    ActivistOwnershipEvent.filer_cik == event.filer_cik,
                    ActivistOwnershipEvent.issuer_cik == event.issuer_cik,
                    ActivistOwnershipEvent.filing_date < event.filing_date,
                    ActivistOwnershipEvent.ownership_pct.isnot(None),
                )
            )
            .order_by(ActivistOwnershipEvent.filing_date.desc())
            .limit(1)
        )
        prev_pct = prev.scalar()
        if prev_pct is None:
            return None
        return round(new_pct - float(prev_pct), 4)

    # ------------------------------------------------------------------
    # Query methods
    # ------------------------------------------------------------------

    async def get_events(
        self,
        db: AsyncSession,
        ticker: str,
        as_of: date,
        start: Optional[date] = None,
        end: Optional[date] = None,
    ) -> List[ActivistOwnershipEvent]:
        """PIT-safe query: filing_date <= as_of."""
        ticker = ticker.upper()
        as_of_dt = datetime(as_of.year, as_of.month, as_of.day, 23, 59, 59, tzinfo=timezone.utc)

        conditions = [
            ActivistOwnershipEvent.symbol == ticker,
            ActivistOwnershipEvent.filing_date <= as_of_dt,
        ]
        if start:
            conditions.append(ActivistOwnershipEvent.filing_date >= datetime(start.year, start.month, start.day, tzinfo=timezone.utc))
        if end:
            conditions.append(ActivistOwnershipEvent.filing_date <= datetime(end.year, end.month, end.day, 23, 59, 59, tzinfo=timezone.utc))

        result = await db.execute(
            select(ActivistOwnershipEvent)
            .where(and_(*conditions))
            .order_by(ActivistOwnershipEvent.filing_date.desc())
        )
        return result.scalars().all()

    async def get_active_positions(
        self,
        db: AsyncSession,
        as_of: date,
        min_ownership_pct: float = 5.0,
    ) -> List[ActivistOwnershipEvent]:
        """Return latest-per-(filer, issuer) positions where ownership_pct >= min."""
        as_of_dt = datetime(as_of.year, as_of.month, as_of.day, 23, 59, 59, tzinfo=timezone.utc)

        # Window function: latest filing per (filer_cik, issuer_cik) as-of as_of
        subq = (
            select(
                ActivistOwnershipEvent,
                func.row_number().over(
                    partition_by=[
                        ActivistOwnershipEvent.filer_cik,
                        ActivistOwnershipEvent.issuer_cik,
                    ],
                    order_by=ActivistOwnershipEvent.filing_date.desc(),
                ).label("rn"),
            )
            .where(ActivistOwnershipEvent.filing_date <= as_of_dt)
            .subquery()
        )
        result = await db.execute(
            select(ActivistOwnershipEvent)
            .join(subq, ActivistOwnershipEvent.id == subq.c.id)
            .where(
                and_(
                    subq.c.rn == 1,
                    ActivistOwnershipEvent.ownership_pct >= min_ownership_pct,
                )
            )
            .order_by(ActivistOwnershipEvent.ownership_pct.desc())
        )
        return result.scalars().all()


# ------------------------------------------------------------------
# Cover-page parsers
# ------------------------------------------------------------------

def _parse_cover_xml(xml_text: str) -> Tuple[Optional[float], Optional[float]]:
    """Try to extract ownership_pct and shares_owned from structured Cover Page XML."""
    import xml.etree.ElementTree as ET
    try:
        root = ET.fromstring(xml_text)
        pct_el = root.find(".//{*}percentClass")
        if pct_el is None:
            pct_el = root.find(".//percentClass")
        if pct_el is None:
            pct_el = root.find(".//{*}classPercent")
        if pct_el is None:
            pct_el = root.find(".//classPercent")
        shares_el = root.find(".//{*}aggregateAmount")
        if shares_el is None:
            shares_el = root.find(".//aggregateAmount")
        if shares_el is None:
            shares_el = root.find(".//{*}amountBeneficiallyOwned")
        if shares_el is None:
            shares_el = root.find(".//amountBeneficiallyOwned")
        if shares_el is None:
            shares_el = root.find(".//{*}reportingPersonBeneficiallyOwnedAggregateNumberOfShares")
        if shares_el is None:
            shares_el = root.find(".//reportingPersonBeneficiallyOwnedAggregateNumberOfShares")
        pct = float(pct_el.text.strip()) if pct_el is not None and pct_el.text else None
        shares = float(shares_el.text.strip().replace(",", "")) if shares_el is not None and shares_el.text else None
        return pct, shares
    except Exception:
        return None, None


def _parse_cover_html(html: str) -> Tuple[Optional[float], Optional[float]]:
    """Best-effort regex extraction from 13D/G HTML cover page."""
    # Strip HTML tags so CSS/attribute digits don't confuse the regexes
    clean = re.sub(r"<[^>]+>", " ", html)
    clean = re.sub(r"\s+", " ", clean)

    pct: Optional[float] = None
    shares: Optional[float] = None

    m = _RE_OWNERSHIP_PCT.search(clean)
    if m:
        try:
            val = float(m.group(1))
            if val <= 100:
                pct = val
        except ValueError:
            pass

    m = _RE_SHARES_OWNED.search(clean)
    if m:
        try:
            shares = float(m.group(1).replace(",", ""))
        except ValueError:
            pass

    return pct, shares