diff --git a/alembic/env.py b/alembic/env.py
index 64b8cfd..8ac184e 100644
--- a/alembic/env.py
+++ b/alembic/env.py
@@ -15,6 +15,7 @@ from app.models import financial, error_log, request_log, fred_data, filing, fin
from app.models import etf # ETF models
from app.models import overlay_registry, overlay_raw_event, overlay_feature # Phase 5 Overlay models
from app.models import attention # Attention subsystem (event-centric backtest-friendly)
+from app.models import insider_transaction, activist_ownership # SEC insider & activist ownership
# this is the Alembic Config object
config = context.config
diff --git a/alembic/versions/l3d4e5f6g7h8_add_form4_derived_columns.py b/alembic/versions/l3d4e5f6g7h8_add_form4_derived_columns.py
new file mode 100644
index 0000000..b913891
--- /dev/null
+++ b/alembic/versions/l3d4e5f6g7h8_add_form4_derived_columns.py
@@ -0,0 +1,63 @@
+"""add Form 4 derived columns (is_ceo, is_cfo, is_c_suite, purchase_pct_of_holding, owner_relationship)
+
+Revision ID: l3d4e5f6g7h8
+Revises: k2c3d4e5f6g7
+Create Date: 2026-04-22
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+revision = "l3d4e5f6g7h8"
+down_revision = "k2c3d4e5f6g7"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+ conn = op.get_bind()
+
+ # Add new columns
+ op.add_column("insider_transactions", sa.Column("owner_relationship", sa.Text(), nullable=True))
+ op.add_column("insider_transactions", sa.Column("is_ceo", sa.Boolean(), nullable=False, server_default="false"))
+ op.add_column("insider_transactions", sa.Column("is_cfo", sa.Boolean(), nullable=False, server_default="false"))
+ op.add_column("insider_transactions", sa.Column("is_c_suite", sa.Boolean(), nullable=False, server_default="false"))
+ op.add_column("insider_transactions", sa.Column("purchase_pct_of_holding", sa.Float(), nullable=True))
+
+ # Widen transaction_code from varchar(5) to varchar(10)
+ op.alter_column("insider_transactions", "transaction_code",
+ existing_type=sa.String(5), type_=sa.String(10))
+
+ # New indexes for PIT / by-date queries
+ op.create_index("idx_insider_filing_ticker", "insider_transactions", ["filing_date", "ticker"])
+
+ # Backfill c-suite flags from existing officer_title data
+ conn.execute(sa.text("""
+ UPDATE insider_transactions
+ SET
+ is_ceo = (officer_title ~* '\\mCEO\\M|Chief\\s+Executive\\s+Officer'),
+ is_cfo = (officer_title ~* '\\mCFO\\M|Chief\\s+Financial\\s+Officer|Principal\\s+Financial\\s+Officer'),
+ is_c_suite = (officer_title ~* '\\mCEO\\M|Chief\\s+Executive\\s+Officer|\\mCFO\\M|Chief\\s+Financial\\s+Officer|Principal\\s+Financial\\s+Officer|\\mCOO\\M|\\mCTO\\M|\\mCIO\\M|\\mCLO\\M|\\mCMO\\M|\\mPresident\\M|\\mChair(man|person|woman)?\\M|Chief\\s+\\w+\\s+Officer')
+ WHERE officer_title IS NOT NULL
+ """))
+
+ # Backfill purchase_pct_of_holding for existing buy transactions
+ conn.execute(sa.text("""
+ UPDATE insider_transactions
+ SET purchase_pct_of_holding = ABS(shares) / NULLIF(shares_owned_after, 0)
+ WHERE transaction_code IN ('P', 'A')
+ AND shares > 0
+ AND shares_owned_after IS NOT NULL
+ AND shares_owned_after > 0
+ """))
+
+
+def downgrade() -> None:
+ op.drop_index("idx_insider_filing_ticker", table_name="insider_transactions")
+ op.drop_column("insider_transactions", "purchase_pct_of_holding")
+ op.drop_column("insider_transactions", "is_c_suite")
+ op.drop_column("insider_transactions", "is_cfo")
+ op.drop_column("insider_transactions", "is_ceo")
+ op.drop_column("insider_transactions", "owner_relationship")
+ op.alter_column("insider_transactions", "transaction_code",
+ existing_type=sa.String(10), type_=sa.String(5))
diff --git a/alembic/versions/m4e5f6g7h8i9_add_activist_ownership_events.py b/alembic/versions/m4e5f6g7h8i9_add_activist_ownership_events.py
new file mode 100644
index 0000000..d4dae2f
--- /dev/null
+++ b/alembic/versions/m4e5f6g7h8i9_add_activist_ownership_events.py
@@ -0,0 +1,55 @@
+"""add activist_ownership_events table for SC 13D/13G filings
+
+Revision ID: m4e5f6g7h8i9
+Revises: l3d4e5f6g7h8
+Create Date: 2026-04-22
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+revision = "m4e5f6g7h8i9"
+down_revision = "l3d4e5f6g7h8"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+ conn = op.get_bind()
+ if conn.dialect.has_table(conn, "activist_ownership_events"):
+ return
+
+ op.create_table(
+ "activist_ownership_events",
+ sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
+ sa.Column("symbol", sa.String(10), nullable=False),
+ sa.Column("issuer_cik", sa.String(20), nullable=False),
+ sa.Column("filer_cik", sa.String(20), nullable=False),
+ sa.Column("filer_name", sa.String(500), nullable=False),
+ sa.Column("accession_number", sa.String(30), nullable=False),
+ sa.Column("form_type", sa.String(20), nullable=False),
+ sa.Column("filing_date", sa.TIMESTAMP(timezone=True), nullable=False),
+ sa.Column("is_amendment", sa.Boolean(), nullable=False, server_default="false"),
+ sa.Column("ownership_pct", sa.Float(), nullable=True),
+ sa.Column("shares_owned", sa.Float(), nullable=True),
+ sa.Column("change_pct", sa.Float(), nullable=True),
+ sa.Column("parse_status", sa.String(20), nullable=False, server_default="index_only"),
+ sa.Column("filing_url", sa.Text(), nullable=True),
+ sa.Column("created_at", sa.TIMESTAMP(timezone=True), nullable=False,
+ server_default=sa.text("NOW()")),
+ sa.Column("updated_at", sa.TIMESTAMP(timezone=True), nullable=False,
+ server_default=sa.text("NOW()")),
+ sa.UniqueConstraint("accession_number", "filer_cik", name="uq_activist_event"),
+ )
+ op.create_index("idx_activist_sym_date", "activist_ownership_events", ["symbol", "filing_date"])
+ op.create_index("idx_activist_filer_issuer", "activist_ownership_events",
+ ["filer_cik", "issuer_cik", "filing_date"])
+ op.create_index("idx_activist_filing_date", "activist_ownership_events", ["filing_date"])
+
+
+def downgrade() -> None:
+ op.drop_index("idx_activist_filing_date", table_name="activist_ownership_events")
+ op.drop_index("idx_activist_filer_issuer", table_name="activist_ownership_events")
+ op.drop_index("idx_activist_sym_date", table_name="activist_ownership_events")
+ op.drop_table("activist_ownership_events")
diff --git a/app/api/v1/api.py b/app/api/v1/api.py
index 1de9ef8..ff9cbd2 100644
--- a/app/api/v1/api.py
+++ b/app/api/v1/api.py
@@ -3,7 +3,7 @@ API v1 router
"""
from fastapi import APIRouter
-from app.api.v1.endpoints import financial, price, catalog, health, migration, database, error_logs, request_logs, news, etf, stocks, fred, filings, alpaca, finra, overlay, screener, attention, insider, earnings, universe, dividends, company
+from app.api.v1.endpoints import financial, price, catalog, health, migration, database, error_logs, request_logs, news, etf, stocks, fred, filings, alpaca, finra, overlay, screener, attention, insider, earnings, universe, dividends, company, ownership
api_router = APIRouter()
@@ -31,4 +31,5 @@ api_router.include_router(insider.router, prefix="/insider", tags=["insider"])
api_router.include_router(earnings.router, prefix="/earnings", tags=["earnings"])
api_router.include_router(universe.router, prefix="/universe", tags=["universe"])
api_router.include_router(dividends.router, prefix="/dividends", tags=["dividends"])
-api_router.include_router(company.router, prefix="/company", tags=["company"])
\ No newline at end of file
+api_router.include_router(company.router, prefix="/company", tags=["company"])
+api_router.include_router(ownership.router, prefix="/ownership", tags=["ownership"])
\ No newline at end of file
diff --git a/app/api/v1/endpoints/insider.py b/app/api/v1/endpoints/insider.py
index 5cf4ad7..78f6fb7 100644
--- a/app/api/v1/endpoints/insider.py
+++ b/app/api/v1/endpoints/insider.py
@@ -3,6 +3,7 @@ Insider Transaction endpoints — SEC Form 4 data
"""
import logging
+from datetime import date
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Query
@@ -11,6 +12,10 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.core.database import get_db
from app.schemas.insider import (
+ Form4AggregateResponse,
+ Form4ByDateResponse,
+ Form4Entry,
+ Form4Response,
InsiderTransactionEntry,
InsiderTransactionResponse,
InsiderSummaryPeriod,
@@ -108,3 +113,103 @@ async def get_insider_summary(
except Exception as e:
logger.error(f"Insider summary error for {symbol}: {e}")
raise HTTPException(status_code=502, detail=f"Failed to fetch insider summary: {e}")
+
+
+# ------------------------------------------------------------------
+# New PIT-safe Form 4 endpoints
+# ------------------------------------------------------------------
+
+@router.get(
+ "/form4/{ticker}",
+ response_model=Form4Response,
+ summary="PIT-safe Form 4 insider transactions",
+ description=(
+ "Returns Form 4 transactions for a ticker where **filing_date ≤ as_of** (point-in-time safe).\n\n"
+ "`as_of` is required to prevent lookahead in backtests.\n\n"
+ "`start`/`end` also filter by `filing_date` (not transaction_date)."
+ ),
+)
+@with_cache(namespace="insider:form4", ttl=300,
+ key_params=["ticker", "start", "end", "as_of", "buy_only", "csuite_only"])
+async def get_form4(
+ ticker: str,
+ response: Response,
+ as_of: date = Query(..., description="Point-in-time cutoff (filing_date ≤ as_of). Required."),
+ start: Optional[date] = Query(None, description="Window start (filing_date ≥ start)"),
+ end: Optional[date] = Query(None, description="Window end (filing_date ≤ end)"),
+ buy_only: bool = Query(False, description="Only return buy transactions (P/A, shares > 0)"),
+ csuite_only: bool = Query(False, description="Only return C-suite insider transactions"),
+ db: AsyncSession = Depends(get_db),
+):
+ svc = InsiderTransactionService()
+ try:
+ rows, total = await svc.get_form4_pit(
+ db, ticker=ticker, as_of=as_of, start=start, end=end,
+ buy_only=buy_only, csuite_only=csuite_only,
+ )
+ entries = [Form4Entry.from_orm_obj(r) for r in rows]
+ return Form4Response(
+ symbol=ticker.upper(),
+ as_of=as_of,
+ window={"start": start.isoformat() if start else None, "end": end.isoformat() if end else None},
+ transactions=entries,
+ total_count=total,
+ )
+ except Exception as e:
+ logger.error(f"Form4 error for {ticker}: {e}")
+ raise HTTPException(status_code=502, detail=str(e))
+
+
+@router.get(
+ "/form4/by-date/{filing_date}",
+ response_model=Form4ByDateResponse,
+ summary="Form 4 filings by a specific date (cross-ticker)",
+ description="Returns all Form 4 transactions where filing_date equals the given date. Useful for pre-market screening.",
+)
+@with_cache(namespace="insider:form4_bydate", ttl=3600, key_params=["filing_date", "buy_only"])
+async def get_form4_by_date(
+ filing_date: date,
+ response: Response,
+ buy_only: bool = Query(False, description="Only return buy transactions"),
+ db: AsyncSession = Depends(get_db),
+):
+ svc = InsiderTransactionService()
+ try:
+ rows = await svc.get_form4_by_date(db, filing_date=filing_date, buy_only=buy_only)
+ entries = [Form4Entry.from_orm_obj(r) for r in rows]
+ return Form4ByDateResponse(
+ filing_date=filing_date,
+ buy_only=buy_only,
+ transactions=entries,
+ total_count=len(entries),
+ )
+ except Exception as e:
+ logger.error(f"Form4 by-date error for {filing_date}: {e}")
+ raise HTTPException(status_code=502, detail=str(e))
+
+
+@router.get(
+ "/form4/aggregate/{ticker}",
+ response_model=Form4AggregateResponse,
+ summary="Aggregate Form 4 buy activity (PIT-safe)",
+ description=(
+ "Aggregated insider buy metrics within [as_of - window_days, as_of].\n\n"
+ "All based on `filing_date` (PIT-safe). Returns buy_count, buy_dollar_total, "
+ "cluster_size (unique insiders), csuite_count, avg_pct_of_holding, recency_days."
+ ),
+)
+@with_cache(namespace="insider:form4_agg", ttl=300, key_params=["ticker", "as_of", "window_days"])
+async def get_form4_aggregate(
+ ticker: str,
+ response: Response,
+ as_of: date = Query(..., description="Point-in-time cutoff. Required."),
+ window_days: int = Query(30, ge=1, le=365, description="Lookback window in days"),
+ db: AsyncSession = Depends(get_db),
+):
+ svc = InsiderTransactionService()
+ try:
+ agg = await svc.get_form4_aggregate(db, ticker=ticker, as_of=as_of, window_days=window_days)
+ return Form4AggregateResponse(**agg)
+ except Exception as e:
+ logger.error(f"Form4 aggregate error for {ticker}: {e}")
+ raise HTTPException(status_code=502, detail=str(e))
diff --git a/app/api/v1/endpoints/ownership.py b/app/api/v1/endpoints/ownership.py
new file mode 100644
index 0000000..81fe29d
--- /dev/null
+++ b/app/api/v1/endpoints/ownership.py
@@ -0,0 +1,92 @@
+"""
+Activist Ownership endpoints — SC 13D / 13G
+"""
+
+import logging
+from datetime import date
+from typing import Optional
+
+from fastapi import APIRouter, Depends, HTTPException, Query
+from fastapi.responses import Response
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.database import get_db
+from app.schemas.ownership import (
+ ActivistEventEntry,
+ ActivistEventsResponse,
+ ActivistActiveResponse,
+)
+from app.services.activist_ownership_service import ActivistOwnershipService
+from app.utils.cache import with_cache
+
+router = APIRouter()
+logger = logging.getLogger("app.api.v1.ownership")
+
+
+@router.get(
+ "/13dg/active",
+ response_model=ActivistActiveResponse,
+ summary="Active activist positions as-of a date",
+ description=(
+ "Returns the latest SC 13D/13G filing per (filer, issuer) pair where **filing_date ≤ as_of** "
+ "and **ownership_pct ≥ min_ownership_pct**.\n\n"
+ "Positions with `ownership_pct = null` (not yet enriched) are excluded.\n\n"
+ "`as_of` is required."
+ ),
+)
+@with_cache(namespace="ownership:13dg_active", ttl=300,
+ key_params=["as_of", "min_ownership_pct"])
+async def get_13dg_active(
+ response: Response,
+ as_of: date = Query(..., description="Point-in-time cutoff. Required."),
+ min_ownership_pct: float = Query(5.0, ge=0.0, le=100.0, description="Minimum ownership %"),
+ db: AsyncSession = Depends(get_db),
+):
+ svc = ActivistOwnershipService()
+ try:
+ rows = await svc.get_active_positions(db, as_of=as_of, min_ownership_pct=min_ownership_pct)
+ entries = [ActivistEventEntry.from_orm_obj(r) for r in rows]
+ return ActivistActiveResponse(
+ as_of=as_of,
+ min_ownership_pct=min_ownership_pct,
+ positions=entries,
+ total_count=len(entries),
+ )
+ except Exception as e:
+ logger.error(f"13D/G active positions error: {e}")
+ raise HTTPException(status_code=502, detail=str(e))
+
+
+@router.get(
+ "/13dg/{ticker}",
+ response_model=ActivistEventsResponse,
+ summary="SC 13D/13G activist ownership events for a ticker",
+ description=(
+ "Returns SC 13D and SC 13G filings (including amendments) where **filing_date ≤ as_of**.\n\n"
+ "`as_of` is required for PIT safety in backtests.\n\n"
+ "Note: `ownership_pct` / `shares_owned` will be `null` until background enrichment runs (~30 min)."
+ ),
+)
+@with_cache(namespace="ownership:13dg", ttl=300, key_params=["ticker", "start", "end", "as_of"])
+async def get_13dg_events(
+ ticker: str,
+ response: Response,
+ as_of: date = Query(..., description="Point-in-time cutoff (filing_date ≤ as_of). Required."),
+ start: Optional[date] = Query(None, description="Window start (filing_date ≥ start)"),
+ end: Optional[date] = Query(None, description="Window end (filing_date ≤ end)"),
+ db: AsyncSession = Depends(get_db),
+):
+ svc = ActivistOwnershipService()
+ try:
+ rows = await svc.get_events(db, ticker=ticker, as_of=as_of, start=start, end=end)
+ entries = [ActivistEventEntry.from_orm_obj(r) for r in rows]
+ return ActivistEventsResponse(
+ symbol=ticker.upper(),
+ as_of=as_of,
+ window={"start": start.isoformat() if start else None, "end": end.isoformat() if end else None},
+ events=entries,
+ total_count=len(entries),
+ )
+ except Exception as e:
+ logger.error(f"13D/G events error for {ticker}: {e}")
+ raise HTTPException(status_code=502, detail=str(e))
diff --git a/app/core/config.py b/app/core/config.py
index 98d73a7..bc62343 100644
--- a/app/core/config.py
+++ b/app/core/config.py
@@ -52,6 +52,8 @@ class Settings(BaseSettings):
SEC_EMAIL: str = os.getenv("SEC_EMAIL", "example@example.com")
SEC_DATA_REFRESH_HOURS: int = 24
SEC_DATA_START_YEAR: int = 1994 # SEC EDGAR data available from 1994
+ SEC_INGEST_TIMEZONE: str = "America/New_York"
+ SEC_FORM4_BOOTSTRAP_QUARTERS: int = 8 # how many recent quarters to backfill
# Security
SECRET_KEY: str = os.getenv("SECRET_KEY", "development-secret-key-change-in-production")
diff --git a/app/main.py b/app/main.py
index 645d4b7..33135dc 100644
--- a/app/main.py
+++ b/app/main.py
@@ -30,6 +30,12 @@ async def lifespan(app: FastAPI):
start_scheduler()
except Exception:
pass
+ # Start SEC ingest scheduler (Form 4 + 13D/G daily/weekly)
+ try:
+ from app.services.sec_ingest.scheduler import start_sec_ingest_scheduler
+ start_sec_ingest_scheduler()
+ except Exception:
+ pass
yield
# Shutdown
try:
@@ -37,6 +43,11 @@ async def lifespan(app: FastAPI):
stop_scheduler()
except Exception:
pass
+ try:
+ from app.services.sec_ingest.scheduler import stop_sec_ingest_scheduler
+ stop_sec_ingest_scheduler()
+ except Exception:
+ pass
try:
from app.core.http_client import close_http_session
await close_http_session()
@@ -152,6 +163,10 @@ app.openapi_tags = [
"name": "fred",
"description": "FRED (Federal Reserve Economic Data) — macroeconomic series via FRED API proxy"
},
+ {
+ "name": "ownership",
+ "description": "SEC 13D/13G activist ownership events — activist filings, active positions (PIT-safe)"
+ },
{
"name": "error-logs",
"description": "Error log management — browse and clear server-side error records"
diff --git a/app/models/activist_ownership.py b/app/models/activist_ownership.py
new file mode 100644
index 0000000..f6ef045
--- /dev/null
+++ b/app/models/activist_ownership.py
@@ -0,0 +1,50 @@
+"""
+SC 13D / 13G Activist Ownership Events model
+"""
+
+from datetime import datetime, timezone
+import uuid
+
+from sqlalchemy import Column, String, Float, Boolean, Index, Text, UniqueConstraint
+from sqlalchemy.dialects.postgresql import UUID, TIMESTAMP
+
+from app.core.database import Base
+
+
+class ActivistOwnershipEvent(Base):
+ __tablename__ = "activist_ownership_events"
+
+ id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+ symbol = Column(String(10), nullable=False)
+ issuer_cik = Column(String(20), nullable=False)
+ filer_cik = Column(String(20), nullable=False)
+ filer_name = Column(String(500), nullable=False)
+ accession_number = Column(String(30), nullable=False)
+ form_type = Column(String(20), nullable=False) # SC 13D, SC 13G, SC 13D/A, SC 13G/A
+ filing_date = Column(TIMESTAMP(timezone=True), nullable=False)
+ is_amendment = Column(Boolean, nullable=False, default=False)
+ ownership_pct = Column(Float, nullable=True)
+ shares_owned = Column(Float, nullable=True)
+ change_pct = Column(Float, nullable=True)
+ # index_only | parsed | parse_failed
+ parse_status = Column(String(20), nullable=False, default="index_only")
+ filing_url = Column(Text, nullable=True)
+
+ created_at = Column(
+ TIMESTAMP(timezone=True), default=lambda: datetime.now(timezone.utc)
+ )
+ updated_at = Column(
+ TIMESTAMP(timezone=True),
+ default=lambda: datetime.now(timezone.utc),
+ onupdate=lambda: datetime.now(timezone.utc),
+ )
+
+ __table_args__ = (
+ UniqueConstraint(
+ "accession_number", "filer_cik",
+ name="uq_activist_event",
+ ),
+ Index("idx_activist_sym_date", "symbol", "filing_date"),
+ Index("idx_activist_filer_issuer", "filer_cik", "issuer_cik", "filing_date"),
+ Index("idx_activist_filing_date", "filing_date"),
+ )
diff --git a/app/models/insider_transaction.py b/app/models/insider_transaction.py
index c6fb093..f492e48 100644
--- a/app/models/insider_transaction.py
+++ b/app/models/insider_transaction.py
@@ -5,7 +5,7 @@ SEC Form 4 — Insider Transaction model
from datetime import datetime, timezone
import uuid
-from sqlalchemy import Column, String, Float, Boolean, Index, UniqueConstraint
+from sqlalchemy import Column, String, Float, Boolean, Index, Text, UniqueConstraint
from sqlalchemy.dialects.postgresql import UUID, TIMESTAMP
from app.core.database import Base
@@ -23,19 +23,24 @@ class InsiderTransaction(Base):
# Reporting owner
owner_name = Column(String(255), nullable=False)
owner_cik = Column(String(20), nullable=True)
+ owner_relationship = Column(Text, nullable=True)
is_officer = Column(Boolean, default=False)
is_director = Column(Boolean, default=False)
is_ten_percent_owner = Column(Boolean, default=False)
officer_title = Column(String(255), nullable=True)
+ is_ceo = Column(Boolean, nullable=False, default=False)
+ is_cfo = Column(Boolean, nullable=False, default=False)
+ is_c_suite = Column(Boolean, nullable=False, default=False)
# Transaction details
security_title = Column(String(255), nullable=True)
transaction_date = Column(TIMESTAMP(timezone=True), nullable=False)
- transaction_code = Column(String(5), nullable=False) # P, S, A, M, G, etc.
+ transaction_code = Column(String(10), nullable=False) # P, S, A, M, G, J/K, etc.
shares = Column(Float, nullable=False)
price_per_share = Column(Float, nullable=True)
total_value = Column(Float, nullable=True)
shares_owned_after = Column(Float, nullable=True)
+ purchase_pct_of_holding = Column(Float, nullable=True)
is_derivative = Column(Boolean, default=False)
created_at = Column(
@@ -56,4 +61,5 @@ class InsiderTransaction(Base):
Index("idx_insider_ticker_date", "ticker", "transaction_date"),
Index("idx_insider_ticker_code", "ticker", "transaction_code"),
Index("idx_insider_filing_date", "filing_date"),
+ Index("idx_insider_filing_ticker", "filing_date", "ticker"),
)
diff --git a/app/schemas/insider.py b/app/schemas/insider.py
index 26c815d..3232427 100644
--- a/app/schemas/insider.py
+++ b/app/schemas/insider.py
@@ -89,3 +89,84 @@ class InsiderSummaryResponse(BaseModel):
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
+
+
+class Form4ByDateResponse(BaseModel):
+ filing_date: date
+ buy_only: bool
+ transactions: List[Form4Entry]
+ total_count: int
+
+
+class Form4AggregateResponse(BaseModel):
+ 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
diff --git a/app/schemas/ownership.py b/app/schemas/ownership.py
new file mode 100644
index 0000000..7f3723f
--- /dev/null
+++ b/app/schemas/ownership.py
@@ -0,0 +1,55 @@
+"""
+Activist ownership schemas — SC 13D / 13G
+"""
+
+from datetime import date
+from typing import Any, Dict, List, Optional
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class ActivistEventEntry(BaseModel):
+ model_config = ConfigDict(from_attributes=True)
+
+ symbol: str
+ filing_date: date
+ filer_name: str
+ filer_cik: str
+ form_type: str
+ ownership_pct: Optional[float] = None
+ shares_owned: Optional[float] = None
+ is_amendment: bool = False
+ change_pct: Optional[float] = None
+ accession_number: str
+ parse_status: str
+
+ @classmethod
+ def from_orm_obj(cls, obj) -> "ActivistEventEntry":
+ return cls(
+ symbol=obj.symbol,
+ filing_date=obj.filing_date.date() if hasattr(obj.filing_date, "date") else obj.filing_date,
+ filer_name=obj.filer_name,
+ filer_cik=obj.filer_cik,
+ form_type=obj.form_type,
+ ownership_pct=obj.ownership_pct,
+ shares_owned=obj.shares_owned,
+ is_amendment=obj.is_amendment or False,
+ change_pct=obj.change_pct,
+ accession_number=obj.accession_number,
+ parse_status=obj.parse_status,
+ )
+
+
+class ActivistEventsResponse(BaseModel):
+ symbol: str
+ as_of: date
+ window: Dict[str, Any] = Field(default_factory=dict)
+ events: List[ActivistEventEntry]
+ total_count: int
+
+
+class ActivistActiveResponse(BaseModel):
+ as_of: date
+ min_ownership_pct: float
+ positions: List[ActivistEventEntry]
+ total_count: int
diff --git a/app/services/activist_ownership_service.py b/app/services/activist_ownership_service.py
new file mode 100644
index 0000000..8e81f7e
--- /dev/null
+++ b/app/services/activist_ownership_service.py
@@ -0,0 +1,398 @@
+"""
+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
diff --git a/app/services/insider_transaction_service.py b/app/services/insider_transaction_service.py
index 189c9b7..1ac8bac 100644
--- a/app/services/insider_transaction_service.py
+++ b/app/services/insider_transaction_service.py
@@ -6,8 +6,9 @@ individual transactions in the insider_transactions table.
"""
import logging
+import re
import xml.etree.ElementTree as ET
-from datetime import datetime, timedelta, timezone
+from datetime import date, datetime, timedelta, timezone
from typing import Dict, List, Optional, Set, Tuple
from sqlalchemy import select, and_, func, desc, distinct, case
@@ -16,11 +17,23 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert
from app.models.insider_transaction import InsiderTransaction
from app.services.sec_http_client import SECHttpClient
+from app.services.sec_full_index_service import IndexEntry
logger = logging.getLogger(__name__)
# Chunk size for batch insert (asyncpg 32767-param limit)
-_CHUNK = 1500 # ~20 cols × 1500 = 30000 params
+_CHUNK = 1500 # ~22 cols × 1500 = 33000 → stay under 32767 with ~20 populated
+
+# C-suite title regex patterns (case-insensitive)
+_RE_CEO = re.compile(r"\b(CEO|Chief\s+Executive\s+Officer)\b", re.IGNORECASE)
+_RE_CFO = re.compile(r"\b(CFO|Chief\s+Financial\s+Officer|Principal\s+Financial\s+Officer)\b", re.IGNORECASE)
+_RE_CSUITE = re.compile(
+ r"\b(CEO|Chief\s+Executive\s+Officer"
+ r"|CFO|Chief\s+Financial\s+Officer|Principal\s+Financial\s+Officer"
+ r"|COO|CTO|CIO|CLO|CMO|President|Chair(?:man|person|woman)?"
+ r"|Chief\s+\w+\s+Officer)\b",
+ re.IGNORECASE,
+)
class InsiderTransactionService:
@@ -28,6 +41,20 @@ class InsiderTransactionService:
def __init__(self):
self._http = SECHttpClient("Stock Oracle Insider Service")
+ # ------------------------------------------------------------------
+ # C-suite title classification
+ # ------------------------------------------------------------------
+
+ @staticmethod
+ def _derive_title_flags(title: Optional[str]) -> Dict[str, bool]:
+ if not title:
+ return {"is_ceo": False, "is_cfo": False, "is_c_suite": False}
+ return {
+ "is_ceo": bool(_RE_CEO.search(title)),
+ "is_cfo": bool(_RE_CFO.search(title)),
+ "is_c_suite": bool(_RE_CSUITE.search(title)),
+ }
+
# ------------------------------------------------------------------
# Index Form 4s from SEC EDGAR
# ------------------------------------------------------------------
@@ -162,6 +189,235 @@ class InsiderTransactionService:
finally:
self._http.clear_deadline()
+ # ------------------------------------------------------------------
+ # Bulk ingestion from full-index entries (daily/quarterly)
+ # ------------------------------------------------------------------
+
+ async def _get_form4_xml_url(self, cik_int: int, acc: str, acc_clean: str) -> Optional[str]:
+ """Resolve the correct Form 4 XML URL by fetching the filing index JSON.
+
+ SEC Form 4 XML files use filer-defined names (e.g., 'form4.xml',
+ 'wk-form4_*.xml', 'tm*_*.xml'). We fetch the filing index JSON to get
+ the primary document filename, then build the correct XML URL.
+ """
+ idx_url = (
+ f"https://www.sec.gov/Archives/edgar/data/{cik_int}/{acc_clean}/{acc}-index.json"
+ )
+ try:
+ idx_data = await self._http.fetch_json(idx_url)
+ primary = idx_data.get("primary_document", "")
+ if not primary:
+ docs = idx_data.get("documents", [])
+ for doc in docs:
+ url = doc.get("document_url", "")
+ if url.endswith(".xml") and "xsl" not in url:
+ primary = url.rsplit("/", 1)[-1]
+ break
+ if primary:
+ filename = primary.rsplit("/", 1)[-1]
+ return (
+ f"https://www.sec.gov/Archives/edgar/data/{cik_int}/{acc_clean}/{filename}"
+ )
+ except Exception:
+ pass
+ return None
+
+ async def index_form4_from_index_entries(
+ self,
+ db: AsyncSession,
+ entries: List[IndexEntry],
+ commit_every: int = 3000,
+ ) -> int:
+ """Ingest Form 4 filings from SEC full-index IndexEntry list.
+
+ For each new accession: fetch the filing index JSON to get the correct
+ XML filename, then parse the XML (extracting ticker from issuerTradingSymbol).
+ Skips already-indexed accession numbers. Returns new transaction count.
+ Commits to DB every commit_every accumulated rows to show progress and limit memory.
+ """
+ if not entries:
+ return 0
+
+ acc_set = {e.accession_number for e in entries}
+ existing = await db.execute(
+ select(InsiderTransaction.accession_number).distinct().where(
+ InsiderTransaction.accession_number.in_(acc_set)
+ )
+ )
+ existing_accs: Set[str] = {r[0] for r in existing.fetchall()}
+ new_entries = [e for e in entries if e.accession_number not in existing_accs]
+
+ if not new_entries:
+ return 0
+
+ total = len(new_entries)
+ logger.info(f"SEC full-index: processing {total} new Form 4 accessions")
+
+ inserted = 0
+ pending_rows: List[Dict] = []
+
+ async def _flush(rows: List[Dict]) -> int:
+ n = 0
+ for i in range(0, len(rows), _CHUNK):
+ chunk = rows[i:i + _CHUNK]
+ stmt = pg_insert(InsiderTransaction).values(chunk)
+ stmt = stmt.on_conflict_do_nothing(constraint="uq_insider_transaction")
+ result = await db.execute(stmt)
+ n += result.rowcount
+ await db.commit()
+ return n
+
+ for idx, entry in enumerate(new_entries, 1):
+ acc_clean = entry.accession_number.replace("-", "")
+ # entry.cik is the ISSUER CIK in company.idx for Form 4.
+ # Form 4 filings are stored under the issuer's CIK directory.
+ cik_int = int(entry.cik)
+ filing_date = datetime.combine(entry.filing_date, datetime.min.time()).replace(tzinfo=timezone.utc)
+
+ # Resolve XML URL via filing index JSON (avoids guessing the filename)
+ xml_url = await self._get_form4_xml_url(cik_int, entry.accession_number, acc_clean)
+ if not xml_url:
+ logger.debug(f"SEC full-index: no XML URL for {entry.accession_number}")
+ else:
+ try:
+ xml_text = await self._http.fetch_text(xml_url, accept="application/xml")
+ rows = self.parse_form4_xml(xml_text, None, entry.cik, entry.accession_number, filing_date)
+ pending_rows.extend(rows)
+ except Exception as e:
+ logger.warning(f"SEC full-index: failed {entry.accession_number}: {e}")
+
+ if len(pending_rows) >= commit_every:
+ n = await _flush(pending_rows)
+ inserted += n
+ pending_rows = []
+ logger.info(f"SEC full-index Form 4: progress {idx}/{total} accessions, {inserted} rows inserted so far")
+
+ if pending_rows:
+ inserted += await _flush(pending_rows)
+
+ logger.info(f"SEC full-index Form 4: done — inserted {inserted} transactions from {total} accessions")
+ return inserted
+
+ # ------------------------------------------------------------------
+ # PIT-safe queries (filing_date-based)
+ # ------------------------------------------------------------------
+
+ async def get_form4_pit(
+ self,
+ db: AsyncSession,
+ ticker: str,
+ as_of: date,
+ start: Optional[date] = None,
+ end: Optional[date] = None,
+ buy_only: bool = False,
+ csuite_only: bool = False,
+ limit: int = 500,
+ ) -> Tuple[List[InsiderTransaction], int]:
+ """PIT-safe Form 4 query. filing_date <= as_of is the lookahead guard."""
+ from datetime import date
+ ticker = ticker.upper()
+ as_of_dt = datetime(as_of.year, as_of.month, as_of.day, 23, 59, 59, tzinfo=timezone.utc)
+
+ conditions = [
+ InsiderTransaction.ticker == ticker,
+ InsiderTransaction.filing_date <= as_of_dt,
+ ]
+ if start:
+ conditions.append(InsiderTransaction.filing_date >= datetime(start.year, start.month, start.day, tzinfo=timezone.utc))
+ if end:
+ conditions.append(InsiderTransaction.filing_date <= datetime(end.year, end.month, end.day, 23, 59, 59, tzinfo=timezone.utc))
+ if buy_only:
+ conditions.append(InsiderTransaction.transaction_code.in_(["P", "A"]))
+ conditions.append(InsiderTransaction.shares > 0)
+ if csuite_only:
+ conditions.append(InsiderTransaction.is_c_suite == True)
+
+ count_q = await db.execute(
+ select(func.count(InsiderTransaction.id)).where(and_(*conditions))
+ )
+ total = count_q.scalar() or 0
+
+ result = await db.execute(
+ select(InsiderTransaction)
+ .where(and_(*conditions))
+ .order_by(desc(InsiderTransaction.filing_date))
+ .limit(limit)
+ )
+ return result.scalars().all(), total
+
+ async def get_form4_by_date(
+ self,
+ db: AsyncSession,
+ filing_date: date,
+ buy_only: bool = False,
+ ) -> List[InsiderTransaction]:
+ """Return all Form 4 transactions with filing_date == the given date (cross-ticker)."""
+ from sqlalchemy import cast, Date as SADate
+ conditions = [
+ func.cast(InsiderTransaction.filing_date, SADate) == filing_date,
+ ]
+ if buy_only:
+ conditions.append(InsiderTransaction.transaction_code.in_(["P", "A"]))
+ conditions.append(InsiderTransaction.shares > 0)
+
+ result = await db.execute(
+ select(InsiderTransaction)
+ .where(and_(*conditions))
+ .order_by(desc(InsiderTransaction.filing_date), InsiderTransaction.ticker)
+ )
+ return result.scalars().all()
+
+ async def get_form4_aggregate(
+ self,
+ db: AsyncSession,
+ ticker: str,
+ as_of: date,
+ window_days: int = 30,
+ ) -> Dict:
+ """Aggregate Form 4 buy activity for a ticker within [as_of-window_days, as_of]."""
+ from datetime import date, timedelta
+ ticker = ticker.upper()
+ as_of_dt = datetime(as_of.year, as_of.month, as_of.day, 23, 59, 59, tzinfo=timezone.utc)
+ window_start = as_of_dt - timedelta(days=window_days)
+
+ base = and_(
+ InsiderTransaction.ticker == ticker,
+ InsiderTransaction.filing_date > window_start,
+ InsiderTransaction.filing_date <= as_of_dt,
+ InsiderTransaction.transaction_code.in_(["P", "A"]),
+ InsiderTransaction.shares > 0,
+ InsiderTransaction.is_derivative == False,
+ )
+
+ agg = await db.execute(
+ select(
+ func.count(InsiderTransaction.id).label("buy_count"),
+ func.sum(InsiderTransaction.total_value).label("buy_dollar_total"),
+ func.count(distinct(InsiderTransaction.owner_cik)).label("cluster_size"),
+ func.sum(case((InsiderTransaction.is_c_suite == True, 1), else_=0)).label("csuite_count"),
+ func.avg(InsiderTransaction.purchase_pct_of_holding).label("avg_pct_of_holding"),
+ func.max(InsiderTransaction.filing_date).label("last_filing_date"),
+ ).where(base)
+ )
+ row = agg.one()
+
+ recency_days = window_days
+ if row.last_filing_date:
+ delta = as_of_dt - row.last_filing_date.replace(tzinfo=timezone.utc) if row.last_filing_date.tzinfo is None else as_of_dt - row.last_filing_date
+ recency_days = max(0, delta.days)
+
+ return {
+ "symbol": ticker,
+ "as_of": as_of.isoformat(),
+ "window_days": window_days,
+ "buy_count": int(row.buy_count or 0),
+ "buy_dollar_total": round(float(row.buy_dollar_total or 0), 2),
+ "cluster_size": int(row.cluster_size or 0),
+ "csuite_count": int(row.csuite_count or 0),
+ "avg_pct_of_holding": round(float(row.avg_pct_of_holding), 4) if row.avg_pct_of_holding else None,
+ "recency_days": recency_days,
+ }
+
# ------------------------------------------------------------------
# Form 4 XML parsing
# ------------------------------------------------------------------
@@ -169,12 +425,16 @@ class InsiderTransactionService:
def parse_form4_xml(
self,
xml_text: str,
- ticker: str,
+ ticker: Optional[str],
cik: str,
accession_number: str,
filing_date: datetime,
) -> List[Dict]:
- """Parse a Form 4 XML document and return transaction dicts."""
+ """Parse a Form 4 XML document and return transaction dicts.
+
+ ticker may be None when called from full-index ingestion; in that case
+ issuerTradingSymbol from the XML is used as the ticker.
+ """
rows: List[Dict] = []
try:
root = ET.fromstring(xml_text)
@@ -182,6 +442,16 @@ class InsiderTransactionService:
logger.warning(f"Insider XML parse error for {accession_number}: {e}")
return []
+ # Extract issuer ticker from XML if not provided (full-index ingestion path)
+ if not ticker:
+ issuer_el = root.find(".//issuer")
+ if issuer_el is not None:
+ sym = _xml_text(issuer_el, "issuerTradingSymbol")
+ ticker = sym.upper() if sym else None
+ if not ticker:
+ logger.debug(f"Insider: no ticker for {accession_number}, skipping")
+ return []
+
# Extract reporting owners
owners = []
for ro in root.findall(".//reportingOwner"):
@@ -190,13 +460,29 @@ class InsiderTransactionService:
name = _xml_text(owner_id, "rptOwnerName") if owner_id is not None else None
if not name:
continue
+ is_officer = _xml_bool(rel, "isOfficer")
+ is_director = _xml_bool(rel, "isDirector")
+ is_ten_pct = _xml_bool(rel, "isTenPercentOwner")
+ officer_title = _xml_text(rel, "officerTitle") if rel is not None else None
+ # Build human-readable relationship string
+ parts = []
+ if is_officer:
+ parts.append(officer_title or "Officer")
+ if is_director:
+ parts.append("Director")
+ if is_ten_pct:
+ parts.append("10% Owner")
+ owner_rel = ", ".join(parts) if parts else None
+ flags = self._derive_title_flags(officer_title)
owners.append({
"owner_name": name,
"owner_cik": _xml_text(owner_id, "rptOwnerCik"),
- "is_officer": _xml_bool(rel, "isOfficer"),
- "is_director": _xml_bool(rel, "isDirector"),
- "is_ten_percent_owner": _xml_bool(rel, "isTenPercentOwner"),
- "officer_title": _xml_text(rel, "officerTitle") if rel is not None else None,
+ "owner_relationship": owner_rel,
+ "is_officer": is_officer,
+ "is_director": is_director,
+ "is_ten_percent_owner": is_ten_pct,
+ "officer_title": officer_title,
+ **flags,
})
if not owners:
@@ -267,6 +553,11 @@ class InsiderTransactionService:
if price is not None and shares is not None:
total_value = round(abs(shares) * price, 2)
+ # purchase_pct_of_holding: only for open-market buys/awards with known post-holding
+ purchase_pct = None
+ if code in ("P", "A") and shares is not None and shares > 0 and shares_after and shares_after > 0:
+ purchase_pct = abs(shares) / shares_after
+
return {
"ticker": ticker,
"cik": cik,
@@ -279,6 +570,7 @@ class InsiderTransactionService:
"price_per_share": price,
"total_value": total_value,
"shares_owned_after": shares_after,
+ "purchase_pct_of_holding": purchase_pct,
"is_derivative": is_derivative,
}
diff --git a/app/services/sec_full_index_service.py b/app/services/sec_full_index_service.py
new file mode 100644
index 0000000..ef3d02a
--- /dev/null
+++ b/app/services/sec_full_index_service.py
@@ -0,0 +1,253 @@
+"""
+SEC EDGAR full-index parser.
+
+Parses company.idx (fixed-width) from both quarterly full-index archives
+and daily-index files. Emits IndexEntry objects filtered by form type.
+
+Empirically verified column offsets (0-based) in real EDGAR company.idx files:
+ Company Name : 0 – 61 (62 chars, left-aligned)
+ Form Type : 62 – 73 (12 chars, left-aligned)
+ CIK : 74 – 90 (17 chars; 5 mandatory leading spaces + CIK digits + trailing spaces)
+ Date Filed : 91 – 100 (10 chars, YYYY-MM-DD)
+ Separator : 101 – 102 (2 spaces)
+ Filename : 103 – end
+
+NOTE: The header line labels "CIK" at 74 and "Date Filed" at 86, but the actual
+data has CIK digits starting at 79 (after 5 mandatory spaces) and date at 91.
+
+form.idx (inside form345.zip) has the SAME offsets from position 74 onward,
+but swaps the first two fields:
+ Form Type : 0 – 11 (12 chars)
+ Company Name : 12 – 73 (62 chars)
+ CIK / Date / Filename: same offsets as company.idx (74, 91, 103)
+"""
+
+import io
+import logging
+import zipfile
+from dataclasses import dataclass
+from datetime import date
+from typing import List, Optional, Set
+
+from app.services.sec_http_client import SECHttpClient
+
+logger = logging.getLogger(__name__)
+
+FORM4_TYPES: Set[str] = {"4", "4/A"}
+# SEC EDGAR uses both abbreviated (older) and full (newer) form type names for 13D/G.
+ACTIVIST_13DG_TYPES: Set[str] = {
+ "SC 13D", "SC 13G", "SC 13D/A", "SC 13G/A",
+ "SCHEDULE 13D", "SCHEDULE 13G", "SCHEDULE 13D/A", "SCHEDULE 13G/A",
+}
+
+# Header lines in company.idx to skip
+_HEADER_LINES = 10
+
+
+@dataclass
+class IndexEntry:
+ company_name: str
+ form_type: str
+ cik: str # zero-padded 10 digits
+ filing_date: date
+ filename: str # e.g. edgar/data/12345/0001234500-26-000001.txt
+ accession_number: str # dashed form, e.g. 0001234500-26-000001
+
+
+def _parse_idx_line(line: str) -> Optional[IndexEntry]:
+ """Parse one fixed-width line from company.idx. Returns None on error.
+
+ Real EDGAR column layout (empirically verified):
+ [0:62] company name
+ [62:74] form type
+ [74:91] CIK (17-char field: 5 mandatory leading spaces + digits + trailing spaces)
+ [91:101] date filed (YYYY-MM-DD, 10 chars)
+ [101:103] separator (2 spaces)
+ [103:] filename
+ """
+ if len(line) < 103:
+ return None
+ company_name = line[0:62].strip()
+ form_type = line[62:74].strip()
+ cik_raw = line[74:91].strip()
+ date_raw = line[91:101].strip()
+ filename = line[103:].strip()
+ if not (form_type and cik_raw and date_raw and filename):
+ return None
+ try:
+ cik = str(int(cik_raw)).zfill(10)
+ filing_date = date.fromisoformat(date_raw)
+ except (ValueError, TypeError):
+ return None
+ # Derive accession_number from filename
+ basename = filename.rsplit("/", 1)[-1]
+ acc_part = basename.split("-index")[0].split(".")[0]
+ if len(acc_part) < 5:
+ return None
+ return IndexEntry(
+ company_name=company_name,
+ form_type=form_type,
+ cik=cik,
+ filing_date=filing_date,
+ filename=filename,
+ accession_number=acc_part,
+ )
+
+
+def parse_company_idx(text: str, form_types: Optional[Set[str]] = None) -> List[IndexEntry]:
+ """Parse a company.idx text blob and return matching IndexEntry list.
+
+ The header format varies by year. We skip all lines until we see the dash
+ separator row (---...), then start parsing data from the next line.
+ """
+ entries: List[IndexEntry] = []
+ lines = text.splitlines()
+
+ # Find the dash separator line; data starts immediately after
+ data_start = 0
+ for i, line in enumerate(lines):
+ stripped = line.strip()
+ if stripped and all(c in "-" for c in stripped) and len(stripped) > 20:
+ data_start = i + 1
+ break
+ else:
+ # Fallback: skip fixed number of header lines
+ data_start = _HEADER_LINES
+
+ body_lines = lines[data_start:]
+ for line in body_lines:
+ if not line.strip():
+ continue
+ entry = _parse_idx_line(line)
+ if entry is None:
+ continue
+ if form_types and entry.form_type not in form_types:
+ continue
+ entries.append(entry)
+ return entries
+
+
+def _parse_form_idx_line(line: str) -> Optional[IndexEntry]:
+ """Parse one fixed-width line from form.idx (inside form345.zip).
+
+ form.idx has Form Type as the FIRST column (0-12), unlike company.idx.
+ CIK/Date/Filename share the same offsets as company.idx from position 74 onward:
+ [74:91] CIK, [91:101] date, [103:] filename.
+ """
+ if len(line) < 103:
+ return None
+ form_type = line[0:12].strip()
+ company_name = line[12:74].strip()
+ cik_raw = line[74:91].strip()
+ date_raw = line[91:101].strip()
+ filename = line[103:].strip()
+ if not (form_type and cik_raw and date_raw and filename):
+ return None
+ try:
+ cik = str(int(cik_raw)).zfill(10)
+ filing_date = date.fromisoformat(date_raw)
+ except (ValueError, TypeError):
+ return None
+ basename = filename.rsplit("/", 1)[-1]
+ acc_part = basename.split("-index")[0].split(".")[0]
+ if len(acc_part) < 5:
+ return None
+ return IndexEntry(
+ company_name=company_name,
+ form_type=form_type,
+ cik=cik,
+ filing_date=filing_date,
+ filename=filename,
+ accession_number=acc_part,
+ )
+
+
+def _parse_form_idx(text: str, form_types: Optional[Set[str]] = None) -> List[IndexEntry]:
+ """Parse a form.idx text blob (from form345.zip) using form.idx column layout."""
+ entries: List[IndexEntry] = []
+ lines = text.splitlines()
+
+ data_start = 0
+ for i, line in enumerate(lines):
+ stripped = line.strip()
+ if stripped and all(c in "-" for c in stripped) and len(stripped) > 20:
+ data_start = i + 1
+ break
+ else:
+ data_start = _HEADER_LINES
+
+ for line in lines[data_start:]:
+ if not line.strip():
+ continue
+ entry = _parse_form_idx_line(line)
+ if entry is None:
+ continue
+ if form_types and entry.form_type not in form_types:
+ continue
+ entries.append(entry)
+ return entries
+
+
+def parse_form345_zip(zip_path: str, form_types: Optional[Set[str]] = None) -> List[IndexEntry]:
+ """Parse the form.idx inside a form345.zip file.
+
+ form345.zip contains 'form.idx' with Form Type as the first column
+ (different from company.idx where Company Name is first).
+ """
+ entries: List[IndexEntry] = []
+ try:
+ with zipfile.ZipFile(zip_path, "r") as zf:
+ names = zf.namelist()
+ idx_name = next((n for n in names if n.endswith(".idx")), None)
+ if not idx_name:
+ logger.warning(f"No .idx file found in {zip_path}")
+ return entries
+ with zf.open(idx_name) as f:
+ text = f.read().decode("latin-1", errors="replace")
+ entries = _parse_form_idx(text, form_types=form_types)
+ except Exception as e:
+ logger.error(f"Error parsing form345.zip {zip_path}: {e}")
+ return entries
+
+
+class SECFullIndexService:
+ """Fetches and parses SEC EDGAR full-index and daily-index files."""
+
+ def __init__(self):
+ self._http = SECHttpClient("Stock Oracle SEC Index Service")
+
+ async def fetch_quarterly_form4_entries(self, year: int, quarter: int) -> List[IndexEntry]:
+ """Download and parse quarterly company.idx for Form 4 entries."""
+ try:
+ text = await self._http.fetch_quarterly_company_idx(year, quarter)
+ return parse_company_idx(text, form_types=FORM4_TYPES)
+ except Exception as e:
+ logger.error(f"Failed to fetch quarterly Form 4 index {year}/Q{quarter}: {e}")
+ return []
+
+ async def fetch_quarterly_activist_entries(self, year: int, quarter: int) -> List[IndexEntry]:
+ """Download and parse quarterly company.idx for SC 13D/13G entries."""
+ try:
+ text = await self._http.fetch_quarterly_company_idx(year, quarter)
+ return parse_company_idx(text, form_types=ACTIVIST_13DG_TYPES)
+ except Exception as e:
+ logger.error(f"Failed to fetch quarterly 13D/G index {year}/Q{quarter}: {e}")
+ return []
+
+ async def fetch_daily_form4_entries(self, date_str: str) -> List[IndexEntry]:
+ """Download and parse a daily index file for Form 4 entries. date_str = YYYYMMDD."""
+ try:
+ text = await self._http.fetch_daily_index(date_str)
+ return parse_company_idx(text, form_types=FORM4_TYPES)
+ except Exception as e:
+ logger.error(f"Failed to fetch daily Form 4 index {date_str}: {e}")
+ return []
+
+ async def fetch_daily_activist_entries(self, date_str: str) -> List[IndexEntry]:
+ """Download and parse a daily index file for SC 13D/13G entries."""
+ try:
+ text = await self._http.fetch_daily_index(date_str)
+ return parse_company_idx(text, form_types=ACTIVIST_13DG_TYPES)
+ except Exception as e:
+ logger.error(f"Failed to fetch daily 13D/G index {date_str}: {e}")
+ return []
diff --git a/app/services/sec_http_client.py b/app/services/sec_http_client.py
index 62bb949..8613a03 100644
--- a/app/services/sec_http_client.py
+++ b/app/services/sec_http_client.py
@@ -319,6 +319,70 @@ class SECHttpClient:
finally:
self._pending -= 1
+ # ------------------------------------------------------------------
+ # CIK reverse-lookup (numeric CIK → ticker)
+ # ------------------------------------------------------------------
+
+ async def get_ticker_for_cik(self, cik: str) -> Optional[str]:
+ """Return the primary ticker for a numeric CIK, or None if not found."""
+ cik_str = str(int(cik)).zfill(10)
+ url = f"{self.sec_base_www}/files/company_tickers.json"
+ try:
+ data = await self.fetch_json(url)
+ for _, info in data.items():
+ if str(info.get("cik_str", "")).zfill(10) == cik_str:
+ return info.get("ticker", "").upper() or None
+ except Exception as e:
+ logger.error(f"Error in get_ticker_for_cik({cik}): {e}")
+ return None
+
+ # ------------------------------------------------------------------
+ # Full-index / daily-index helpers
+ # ------------------------------------------------------------------
+
+ async def fetch_quarterly_company_idx(self, year: int, quarter: int) -> str:
+ """Fetch the full-index company.idx for a given year/quarter (fixed-width text)."""
+ url = f"{self.sec_base_www}/Archives/edgar/full-index/{year}/QTR{quarter}/company.idx"
+ return await self.fetch_text(url, accept="text/plain")
+
+ async def fetch_daily_index(self, date_str: str) -> str:
+ """Fetch the daily company index for a given date string (YYYYMMDD).
+
+ Short TTL: the disk cache key is unique per date, so yesterday's file is
+ cached forever. Today's file may be growing — callers should use skip_cache
+ if they need the freshest data.
+ """
+ from datetime import datetime
+ dt = datetime.strptime(date_str, "%Y%m%d")
+ quarter = (dt.month - 1) // 3 + 1
+ url = f"{self.sec_base_www}/Archives/edgar/daily-index/{dt.year}/QTR{quarter}/company.{date_str}.idx"
+ return await self.fetch_text(url, accept="text/plain")
+
+ async def fetch_form345_zip(self, year: int, quarter: int, dest_path: str) -> str:
+ """Stream the form345.zip full-index archive to dest_path. Returns dest_path."""
+ import os
+ url = f"{self.sec_base_www}/Archives/edgar/full-index/{year}/QTR{quarter}/form.idx"
+ # form.idx is a fixed-width index file filtered to form types; for zip download
+ # use the actual form345.zip:
+ zip_url = f"{self.sec_base_www}/Archives/edgar/full-index/{year}/QTR{quarter}/form345.zip"
+ if self._pending >= self._MAX_PENDING:
+ raise RuntimeError(f"SEC request queue full ({self._pending}/{self._MAX_PENDING} pending)")
+ self._pending += 1
+ try:
+ import aiohttp as _aiohttp
+ session = await self._get_session()
+ async with self._req_sem:
+ await self._rate_limiter.acquire()
+ async with session.get(zip_url, timeout=aiohttp.ClientTimeout(total=300)) as resp:
+ resp.raise_for_status()
+ os.makedirs(os.path.dirname(dest_path) or ".", exist_ok=True)
+ with open(dest_path, "wb") as f:
+ async for chunk in resp.content.iter_chunked(65536):
+ f.write(chunk)
+ return dest_path
+ finally:
+ self._pending -= 1
+
async def fetch_text(self, url: str, accept: str = "text/html", max_bytes: Optional[int] = None) -> str:
"""Fetch text content with retry, backoff, block page detection, and caching."""
# In-memory cache
diff --git a/app/services/sec_ingest/__init__.py b/app/services/sec_ingest/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/app/services/sec_ingest/scheduler.py b/app/services/sec_ingest/scheduler.py
new file mode 100644
index 0000000..2717044
--- /dev/null
+++ b/app/services/sec_ingest/scheduler.py
@@ -0,0 +1,201 @@
+"""
+APScheduler-based daily/weekly SEC ingest scheduler.
+
+Jobs:
+ form4_daily_ingest — Tue-Sat 09:00 ET: previous business day Form 4
+ activist_13dg_daily_ingest — same schedule: SC 13D/G index-only ingest
+ form4_weekly_reindex — Sat 03:00 ET: current quarter company.idx rescan
+ activist_13dg_enrich — every 30 min: enrich index_only rows with ownership_pct
+
+Wire up via start_sec_ingest_scheduler() / stop_sec_ingest_scheduler() in app lifespan.
+"""
+
+import logging
+from datetime import date, datetime, timedelta, timezone
+
+logger = logging.getLogger(__name__)
+
+_scheduler = None
+
+
+def _get_scheduler():
+ global _scheduler
+ if _scheduler is None:
+ try:
+ from apscheduler.schedulers.asyncio import AsyncIOScheduler
+ from app.core.config import settings
+ _scheduler = AsyncIOScheduler(timezone=settings.SEC_INGEST_TIMEZONE)
+ except ImportError:
+ logger.warning("apscheduler not installed; SEC ingest scheduling disabled")
+ return None
+ return _scheduler
+
+
+def _prev_business_day() -> date:
+ """Return the most recent business day (Mon-Fri) relative to today."""
+ today = datetime.now(timezone.utc).date()
+ delta = timedelta(days=1)
+ # If today is Mon, go back to Fri; otherwise go back 1 day
+ candidate = today - delta
+ while candidate.weekday() >= 5: # 5=Sat, 6=Sun
+ candidate -= delta
+ return candidate
+
+
+async def _run_form4_daily_ingest() -> None:
+ """Ingest prior business day Form 4 entries from SEC daily full-index."""
+ from app.core.database import AsyncSessionLocal
+ from app.services.sec_full_index_service import SECFullIndexService
+ from app.services.insider_transaction_service import InsiderTransactionService
+
+ target_date = _prev_business_day()
+ date_str = target_date.strftime("%Y%m%d")
+ logger.info(f"[SEC Ingest] Form 4 daily ingest for {date_str}")
+
+ index_svc = SECFullIndexService()
+ txn_svc = InsiderTransactionService()
+
+ entries = await index_svc.fetch_daily_form4_entries(date_str)
+ if not entries:
+ logger.info(f"[SEC Ingest] No Form 4 entries in daily index for {date_str}")
+ return
+
+ async with AsyncSessionLocal() as db:
+ try:
+ inserted = await txn_svc.index_form4_from_index_entries(db, entries)
+ logger.info(f"[SEC Ingest] Form 4 daily: {inserted} new transactions for {date_str}")
+ except Exception as e:
+ logger.error(f"[SEC Ingest] Form 4 daily ingest failed ({date_str}): {e}")
+
+
+async def _run_activist_daily_ingest() -> None:
+ """Ingest prior business day SC 13D/G entries from SEC daily full-index."""
+ from app.core.database import AsyncSessionLocal
+ from app.services.sec_full_index_service import SECFullIndexService
+ from app.services.activist_ownership_service import ActivistOwnershipService
+
+ target_date = _prev_business_day()
+ date_str = target_date.strftime("%Y%m%d")
+ logger.info(f"[SEC Ingest] Activist 13D/G daily ingest for {date_str}")
+
+ index_svc = SECFullIndexService()
+ activist_svc = ActivistOwnershipService()
+
+ entries = await index_svc.fetch_daily_activist_entries(date_str)
+ if not entries:
+ logger.info(f"[SEC Ingest] No 13D/G entries in daily index for {date_str}")
+ return
+
+ async with AsyncSessionLocal() as db:
+ try:
+ inserted = await activist_svc.ingest_from_index_entries(db, entries)
+ logger.info(f"[SEC Ingest] Activist 13D/G daily: {inserted} new rows for {date_str}")
+ except Exception as e:
+ logger.error(f"[SEC Ingest] Activist daily ingest failed ({date_str}): {e}")
+
+
+async def _run_form4_weekly_reindex() -> None:
+ """Re-scan current quarter's company.idx to catch corrections and amendments."""
+ from app.core.database import AsyncSessionLocal
+ from app.services.sec_full_index_service import SECFullIndexService
+ from app.services.insider_transaction_service import InsiderTransactionService
+
+ today = datetime.now(timezone.utc).date()
+ quarter = (today.month - 1) // 3 + 1
+ logger.info(f"[SEC Ingest] Form 4 weekly reindex {today.year}/Q{quarter}")
+
+ index_svc = SECFullIndexService()
+ txn_svc = InsiderTransactionService()
+
+ entries = await index_svc.fetch_quarterly_form4_entries(today.year, quarter)
+ if not entries:
+ return
+
+ async with AsyncSessionLocal() as db:
+ try:
+ inserted = await txn_svc.index_form4_from_index_entries(db, entries)
+ logger.info(f"[SEC Ingest] Form 4 weekly reindex: {inserted} new transactions")
+ except Exception as e:
+ logger.error(f"[SEC Ingest] Form 4 weekly reindex failed: {e}")
+
+
+async def _run_activist_enrich() -> None:
+ """Enrich a batch of index_only activist rows with ownership_pct / shares_owned."""
+ from app.core.database import AsyncSessionLocal
+ from app.services.activist_ownership_service import ActivistOwnershipService
+
+ activist_svc = ActivistOwnershipService()
+ async with AsyncSessionLocal() as db:
+ try:
+ enriched = await activist_svc.enrich_pending(db)
+ if enriched:
+ logger.info(f"[SEC Ingest] Activist enrich: {enriched} rows enriched")
+ except Exception as e:
+ logger.error(f"[SEC Ingest] Activist enrich job failed: {e}")
+
+
+def start_sec_ingest_scheduler() -> None:
+ """Start the SEC ingest scheduler. Call from FastAPI lifespan startup."""
+ sched = _get_scheduler()
+ if sched is None:
+ return
+
+ try:
+ from apscheduler.triggers.cron import CronTrigger
+ from apscheduler.triggers.interval import IntervalTrigger
+
+ sched.add_job(
+ _run_form4_daily_ingest,
+ trigger=CronTrigger(day_of_week="tue-sat", hour=9, minute=0),
+ id="form4_daily_ingest",
+ replace_existing=True,
+ max_instances=1,
+ misfire_grace_time=600,
+ coalesce=True,
+ )
+ sched.add_job(
+ _run_activist_daily_ingest,
+ trigger=CronTrigger(day_of_week="tue-sat", hour=9, minute=5),
+ id="activist_13dg_daily_ingest",
+ replace_existing=True,
+ max_instances=1,
+ misfire_grace_time=600,
+ coalesce=True,
+ )
+ sched.add_job(
+ _run_form4_weekly_reindex,
+ trigger=CronTrigger(day_of_week="sat", hour=3, minute=0),
+ id="form4_weekly_reindex",
+ replace_existing=True,
+ max_instances=1,
+ misfire_grace_time=600,
+ coalesce=True,
+ )
+ sched.add_job(
+ _run_activist_enrich,
+ trigger=IntervalTrigger(minutes=30),
+ id="activist_13dg_enrich",
+ replace_existing=True,
+ max_instances=1,
+ misfire_grace_time=120,
+ coalesce=True,
+ )
+
+ sched.start()
+ logger.info(
+ "SEC ingest scheduler started: "
+ "Form 4 daily @ Tue-Sat 09:00 ET, "
+ "13D/G daily @ Tue-Sat 09:05 ET, "
+ "Form 4 weekly reindex @ Sat 03:00 ET, "
+ "Activist enrich every 30 min"
+ )
+ except Exception as e:
+ logger.error(f"SEC ingest scheduler start failed: {e}")
+
+
+def stop_sec_ingest_scheduler() -> None:
+ """Stop the scheduler. Call from FastAPI lifespan shutdown."""
+ sched = _get_scheduler()
+ if sched and sched.running:
+ sched.shutdown(wait=False)
+ logger.info("SEC ingest scheduler stopped")
diff --git a/docker-compose.yml b/docker-compose.yml
index c3b6b20..42be550 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -68,7 +68,7 @@ services:
mem_limit: 3g
memswap_limit: 3g
restart: unless-stopped
- command: ["sh", "-c", "cp /app/yfinance_plus/yfinance_plus.py /usr/local/lib/python3.11/site-packages/yfinance_plus.py && python -m uvicorn app.main:app --host 0.0.0.0 --port 18000 --limit-concurrency 25"]
+ command: ["sh", "-c", "cp /app/yfinance_plus/yfinance_plus.py /usr/local/lib/python3.11/site-packages/yfinance_plus.py && python -m uvicorn app.main:app --host 0.0.0.0 --port 18000 --limit-concurrency 50"]
# Frontend Application
frontend:
diff --git a/docs/DATA_COVERAGE.md b/docs/DATA_COVERAGE.md
index d18f5ef..fc33c68 100644
--- a/docs/DATA_COVERAGE.md
+++ b/docs/DATA_COVERAGE.md
@@ -2,7 +2,7 @@
각 API 엔드포인트의 **실제 DB 보유 데이터 범위**와 **과거 데이터 백필 방법**을 정리한 문서입니다.
-> 마지막 업데이트: 2026-04-20
+> 마지막 업데이트: 2026-04-23
> DB 실측 기준
---
@@ -25,6 +25,11 @@
| `/overlay/{symbol}/headlines` | Yahoo Finance RSS | ✅ | 2026-02-24 ~ 현재 | 서비스 시작 이후 | 과거 백필 불가 |
| `/overlay/{symbol}/wiki` | Wikipedia Pageviews API | ✅ | 2015-12-26 ~ 현재 | 2015년~ | 자동 수집됨 |
| `/insider/transactions` | SEC EDGAR Form 4 | ✅ | 요청 기반 자동 누적 | 2004년~ | 요청 기반 자동 누적 |
+| `/insider/form4/{ticker}` | SEC EDGAR Form 4 | ✅ | **최근 2년 (8개 quarter) 사전 bootstrap** | 2024년 Q3 ~ 현재, 676,858건 / 4,814 티커 | `bootstrap_form4_by_ticker.py` 완료 |
+| `/insider/form4/by-date/{date}` | SEC EDGAR Form 4 | ✅ | 동상 | 동상 | 동상 |
+| `/insider/form4/aggregate/{ticker}` | SEC EDGAR Form 4 | ✅ | 동상 | 동상 | 동상 |
+| `/ownership/13dg/{ticker}` | SEC EDGAR SC 13D/G | ✅ | **최근 2년 (8개 quarter) 사전 bootstrap** | 2024년 Q3 ~ 현재, 42,329행 | `bootstrap_13dg.py` 완료 |
+| `/ownership/13dg/active` | SEC EDGAR SC 13D/G | ✅ | 동상 | 동상 | 동상 |
| `/earnings/surprise` | yfinance-plus earnings_dates | ✅ | 요청 기반 자동 누적 | ~25분기 (6년+) | 요청 기반 자동 누적 |
| `/universe/screen` | SEC EDGAR + yfinance 월별 스냅샷 | ✅ (사전 빌드 필요) | admin 빌드 후 사용 가능 | 2010년~ | **⚠️ 사전 빌드 필요** |
| `/company/{ticker}` | yfinance-plus + universe_ticker_registry | ✅ (Redis 24h + DB 영구) | 모든 yfinance 지원 티커 | 즉시 | 요청 기반 자동 누적 |
@@ -225,17 +230,17 @@ GET /api/v1/overlay/admin/health
### `/api/v1/insider` — 내부자 거래 (SEC Form 4)
-**현재 DB 보유**: 요청 기반 자동 누적 (첫 조회 시 자동 인덱싱)
+**현재 DB 보유**:
+- **기존 `/transactions`, `/summary`**: 요청 기반 자동 누적 (첫 조회 시 자동 인덱싱)
+- **신규 `/form4/*`**: 최근 2년 사전 bootstrap 완료 — 676,858건 / 4,814 티커 / 2024 Q3 ~ 2026-04-23
-**이론적 범위**: 2004년~ (EDGAR 전자 파일링 이후). 실제 커버리지는 기업마다 다름.
+**자동 갱신**: 매 영업일 09:00 ET — 직전 영업일 daily full-index → Form 4 upsert (scheduler)
-**조회 파라미터**:
-- `GET /insider/transactions/{symbol}?days=90&transaction_type=P-Purchase`
-- `GET /insider/summary/{symbol}?period=90d`: 집계 요약 (매수/매도 금액, 순매수)
+**⚠️ PIT 주의**: 신규 `/form4/*` 엔드포인트는 `as_of` 파라미터 **필수**. 누락 시 422.
-**지원 거래 유형**: `P-Purchase`, `S-Sale`, `A-Award`, `D-Return`, `F-TaxWithholding`, `G-Gift`, `M-OptionExercise`
+---
-**동작 방식**: 첫 조회 시 SEC EDGAR Form 4 XML 자동 파싱 → DB 저장. 이후 캐시.
+#### 기존 엔드포인트 (Lazy on-demand)
```bash
# 최근 90일 내부자 거래 조회 (자동 인덱싱)
@@ -245,6 +250,138 @@ curl "http://localhost:18001/api/v1/insider/transactions/NVDA?days=90"
curl "http://localhost:18001/api/v1/insider/summary/AAPL?period=90d"
```
+**지원 거래 유형**: `P-Purchase`, `S-Sale`, `A-Award`, `D-Return`, `F-TaxWithholding`, `G-Gift`, `M-OptionExercise`
+
+---
+
+#### 신규 PIT-safe 엔드포인트 (2026-04-23)
+
+**`GET /insider/form4/{ticker}`** — 티커별 Form 4 거래 목록
+
+| 파라미터 | 필수 | 설명 |
+|---|---|---|
+| `as_of` | ✅ | 기준 날짜 (YYYY-MM-DD). `filing_date <= as_of` 필터. |
+| `start` | - | 시작 날짜 (filing_date 기준) |
+| `end` | - | 종료 날짜 (filing_date 기준) |
+| `buy_only` | - | `true` → 매수(`P`,`A`) 거래만 |
+| `csuite_only` | - | `true` → CEO·CFO·COO·CTO 등 C-suite만 |
+
+반환 필드: `symbol`, `filing_date`, `transaction_date`, `owner_cik`, `owner_name`, `owner_relationship`, `is_officer`, `is_director`, `is_ten_percent_owner`, `is_ceo`, `is_cfo`, `is_c_suite`, `shares`, `price`, `total_value`, `shares_owned_following`, `purchase_pct_of_holding`, `transaction_code`, `accession_number`
+
+```bash
+# TSLA CEO·CFO 매수 거래 조회
+curl "http://localhost:18001/api/v1/insider/form4/TSLA?as_of=2026-04-20&start=2026-01-01&csuite_only=true&buy_only=true"
+
+# NVDA 최근 30일 내부자 거래 전체
+curl "http://localhost:18001/api/v1/insider/form4/NVDA?as_of=2026-04-20&start=2026-03-20"
+```
+
+**`GET /insider/form4/by-date/{date}`** — 특정 공시일 전체 거래 (cross-ticker)
+
+| 파라미터 | 필수 | 설명 |
+|---|---|---|
+| `buy_only` | - | 매수 거래만 반환 |
+
+```bash
+# 2026-04-17 공시 전체 매수 거래
+curl "http://localhost:18001/api/v1/insider/form4/by-date/2026-04-17?buy_only=true"
+```
+
+**`GET /insider/form4/aggregate/{ticker}`** — 집계 요약 (PIT-safe)
+
+| 파라미터 | 필수 | 설명 |
+|---|---|---|
+| `as_of` | ✅ | 기준 날짜 |
+| `window_days` | - | 집계 윈도우 (기본 30일). `filing_date ∈ (as_of - window_days, as_of]` |
+
+반환 필드: `buy_count`, `buy_dollar_total`, `cluster_size`, `csuite_count`, `avg_pct_of_holding`, `recency_days`
+
+```bash
+# AAPL 최근 30일 내부자 매수 집계
+curl "http://localhost:18001/api/v1/insider/form4/aggregate/AAPL?as_of=2026-04-20&window_days=30"
+```
+
+---
+
+#### C-suite 판별 기준
+
+`officer_title` 에 대해 case-insensitive 정규식 적용:
+- `is_ceo`: `CEO`, `Chief Executive Officer`
+- `is_cfo`: `CFO`, `Chief Financial Officer`, `Principal Financial Officer`
+- `is_c_suite`: 위 둘 + `COO`, `CTO`, `CIO`, `CLO`, `CMO`, `President`, `Chairman/person/woman`, `Chief * Officer`
+
+---
+
+---
+
+### `/api/v1/ownership` — Activist Ownership (SEC SC 13D/G) (신규, 2026-04-23)
+
+**현재 DB 보유**: 42,329행, 최근 2년 (2024 Q3 ~ 2026-04-23) 사전 bootstrap 완료
+
+**자동 갱신**:
+- 매 영업일 09:00 ET — daily full-index → SC 13D/G index-only upsert
+- 30분 주기 background enrich — `parse_status='index_only'` 200행씩 cover-page XML/HTML 파싱 → `ownership_pct`, `shares_owned` 보강
+
+**parse_status 의미**:
+- `index_only`: EDGAR 인덱스에서 가져온 기본 메타만 있음. `ownership_pct=NULL`
+- `parsed`: cover-page XML/HTML 파싱 완료. `ownership_pct` 채워짐
+- `parse_failed`: 파싱 시도했으나 문서 구조 불명확
+
+**⚠️ PIT 주의**: `as_of` 파라미터 **필수**. 누락 시 422. 모든 날짜 필터는 `filing_date` 기준.
+
+---
+
+**`GET /ownership/13dg/{ticker}`** — 티커별 activist 이벤트 목록
+
+| 파라미터 | 필수 | 설명 |
+|---|---|---|
+| `as_of` | ✅ | 기준 날짜. `filing_date <= as_of` 필터. |
+| `start` | - | 시작 날짜 (filing_date 기준) |
+| `end` | - | 종료 날짜 (filing_date 기준) |
+
+반환 필드: `symbol`, `filing_date`, `filer_name`, `filer_cik`, `form_type`, `ownership_pct`, `shares_owned`, `is_amendment`, `change_pct`, `accession_number`, `parse_status`
+
+`form_type` 값: `SC 13D`, `SC 13G`, `SC 13D/A`, `SC 13G/A` (또는 `SCHEDULE 13D/G` 등 변형 포함)
+
+```bash
+# AAPL activist filing 전체 (as_of 기준 이전)
+curl "http://localhost:18001/api/v1/ownership/13dg/AAPL?as_of=2026-04-23"
+
+# RLGT 2025년 이후 activist 이벤트
+curl "http://localhost:18001/api/v1/ownership/13dg/RLGT?as_of=2026-04-23&start=2025-01-01"
+```
+
+---
+
+**`GET /ownership/13dg/active`** — 현재 활성 activist 포지션 목록
+
+| 파라미터 | 필수 | 설명 |
+|---|---|---|
+| `as_of` | ✅ | 기준 날짜 |
+| `min_ownership_pct` | - | 최소 지분율 (기본 5.0) |
+
+**쿼리 로직**: `(filer_cik, issuer_cik)` 쌍별 최신 filing (`filing_date DESC`) 한 행씩, `ownership_pct >= min_ownership_pct` 필터.
+
+```bash
+# 현재 5% 이상 activist 포지션 전체 (파싱된 행만)
+curl "http://localhost:18001/api/v1/ownership/13dg/active?as_of=2026-04-23&min_ownership_pct=5.0"
+
+# 10% 이상 대형 activist
+curl "http://localhost:18001/api/v1/ownership/13dg/active?as_of=2026-04-23&min_ownership_pct=10.0"
+```
+
+---
+
+#### Bootstrap (1회성, 이미 완료)
+
+```bash
+# 최근 8개 quarter SC 13D/G index-only 수집
+docker exec stock_oracle_api python scripts/bootstrap_13dg.py --quarters 8
+
+# Form 4 bootstrap (최근 2년)
+docker exec stock_oracle_api python scripts/bootstrap_form4_by_ticker.py
+```
+
---
### `/api/v1/earnings` — 어닝 서프라이즈 (yfinance-plus)
@@ -474,6 +611,16 @@ UNION ALL
SELECT 'earnings_surprise', MIN(earnings_date)::date, MAX(earnings_date)::date, COUNT(DISTINCT earnings_date::date), COUNT(DISTINCT ticker) FROM earnings_surprise
ORDER BY tbl;
+-- Form 4 PIT 데이터 상태
+SELECT MIN(filing_date)::date, MAX(filing_date)::date, COUNT(*) AS txns, COUNT(DISTINCT ticker) AS tickers
+FROM insider_transactions;
+
+-- SC 13D/G activist 데이터 상태
+SELECT parse_status, COUNT(*),
+ COUNT(*) FILTER (WHERE ownership_pct IS NOT NULL) AS has_pct,
+ MIN(filing_date)::date, MAX(filing_date)::date
+FROM activist_ownership_events GROUP BY parse_status ORDER BY parse_status;
+
-- Universe 스냅샷 상태 확인
SELECT
COUNT(DISTINCT ticker) AS tickers,
diff --git a/scripts/bootstrap_13dg.py b/scripts/bootstrap_13dg.py
new file mode 100644
index 0000000..91ec498
--- /dev/null
+++ b/scripts/bootstrap_13dg.py
@@ -0,0 +1,104 @@
+"""
+Bootstrap SC 13D/G activist ownership events for the last N quarters.
+Skips Form 4 (already bootstrapped separately).
+
+Usage (inside the container):
+ python scripts/bootstrap_13dg.py [--quarters N]
+
+Idempotent: already-indexed accessions are skipped via upsert ON CONFLICT DO NOTHING.
+"""
+
+import asyncio
+import fcntl
+import logging
+import os
+import sys
+from datetime import datetime, timezone
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+logger = logging.getLogger("bootstrap_13dg")
+
+_LOCK_FILE = "/tmp/bootstrap_13dg.lock"
+
+
+def _quarters_to_process(n: int):
+ now = datetime.now(timezone.utc)
+ current_q = (now.month - 1) // 3 + 1
+ results = []
+ year, q = now.year, current_q
+ for _ in range(n):
+ results.append((year, q))
+ q -= 1
+ if q == 0:
+ q = 4
+ year -= 1
+ return results
+
+
+async def bootstrap(n_quarters: int = 8) -> None:
+ from app.core.database import AsyncSessionLocal
+ from app.services.sec_full_index_service import SECFullIndexService
+ from app.services.activist_ownership_service import ActivistOwnershipService
+
+ quarters = _quarters_to_process(n_quarters)
+ logger.info(f"Bootstrapping 13D/G for {n_quarters} quarters: {quarters}")
+
+ index_svc = SECFullIndexService()
+ activist_svc = ActivistOwnershipService()
+
+ _BATCH = 5000 # entries per sub-batch within a quarter (prevents OOM on 20k+ quarters)
+
+ total_inserted = 0
+ for year, quarter in quarters:
+ logger.info(f"Processing {year}/Q{quarter} ...")
+ try:
+ activist_entries = await index_svc.fetch_quarterly_activist_entries(year, quarter)
+ logger.info(f" 13D/G entries from company.idx: {len(activist_entries)}")
+ quarter_inserted = 0
+ for batch_start in range(0, len(activist_entries), _BATCH):
+ batch = activist_entries[batch_start:batch_start + _BATCH]
+ async with AsyncSessionLocal() as db:
+ inserted = await activist_svc.ingest_from_index_entries(db, batch)
+ quarter_inserted += inserted
+ # Clear HTTP cache between batches to prevent OOM on large quarters
+ activist_svc._http._text_cache.clear()
+ activist_svc._http._json_cache.clear()
+ import gc; gc.collect()
+ logger.info(f" batch {batch_start// _BATCH + 1}: {inserted} inserted")
+ total_inserted += quarter_inserted
+ logger.info(f" 13D/G index-only inserted: {quarter_inserted}")
+ except Exception as e:
+ logger.warning(f" {year}/Q{quarter} failed: {e}")
+ finally:
+ activist_svc._http._text_cache.clear()
+ activist_svc._http._json_cache.clear()
+ import gc; gc.collect()
+
+ logger.info(f"Bootstrap complete: {total_inserted} total 13D/G events inserted across {n_quarters} quarters.")
+
+
+if __name__ == "__main__":
+ import argparse
+
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--quarters", type=int, default=8)
+ args = parser.parse_args()
+
+ lock_fh = open(_LOCK_FILE, "w")
+ try:
+ fcntl.flock(lock_fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ except OSError:
+ logger.error("Bootstrap is already running (lock file held). Exiting.")
+ sys.exit(1)
+
+ try:
+ asyncio.run(bootstrap(n_quarters=args.quarters))
+ finally:
+ fcntl.flock(lock_fh, fcntl.LOCK_UN)
+ lock_fh.close()
+ try:
+ os.unlink(_LOCK_FILE)
+ except OSError:
+ pass
diff --git a/scripts/bootstrap_form4_2y.py b/scripts/bootstrap_form4_2y.py
new file mode 100644
index 0000000..92198b6
--- /dev/null
+++ b/scripts/bootstrap_form4_2y.py
@@ -0,0 +1,138 @@
+"""
+One-time bootstrap: ingest Form 4 + SC 13D/G for the last N quarters.
+
+Usage (inside the container):
+ python scripts/bootstrap_form4_2y.py
+
+Quarters bootstrapped = settings.SEC_FORM4_BOOTSTRAP_QUARTERS (default 8 = 2 years).
+Idempotent: already-indexed accessions are skipped via upsert ON CONFLICT DO NOTHING.
+"""
+
+import asyncio
+import fcntl
+import logging
+import os
+import sys
+import tempfile
+from datetime import datetime, timezone
+
+# Ensure project root is on sys.path when invoked directly
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+logger = logging.getLogger("bootstrap")
+
+_LOCK_FILE = "/tmp/bootstrap_form4.lock"
+# Use Archives endpoint for probe — browse-edgar can be 200 while Archives is still 429
+_SEC_PROBE_URL = "https://www.sec.gov/Archives/edgar/full-index/2026/QTR1/company.idx"
+
+
+def _quarters_to_process(n: int):
+ """Return list of (year, quarter) tuples for the last n quarters."""
+ now = datetime.now(timezone.utc)
+ current_q = (now.month - 1) // 3 + 1
+ results = []
+ year, q = now.year, current_q
+ for _ in range(n):
+ results.append((year, q))
+ q -= 1
+ if q == 0:
+ q = 4
+ year -= 1
+ return results
+
+
+async def _check_sec_available() -> bool:
+ """Return True if SEC EDGAR is responding (not rate-limiting us)."""
+ import aiohttp
+ headers = {"User-Agent": "Stock Oracle bootstrap@stockoracle.internal"}
+ try:
+ async with aiohttp.ClientSession() as session:
+ async with session.get(
+ _SEC_PROBE_URL, headers=headers,
+ timeout=aiohttp.ClientTimeout(total=10),
+ allow_redirects=True,
+ ) as resp:
+ if resp.status == 429:
+ logger.warning(f"SEC EDGAR is rate-limiting us (429). Try again in 30+ minutes.")
+ return False
+ return resp.status < 500
+ except Exception as e:
+ logger.warning(f"SEC EDGAR probe failed: {e}")
+ return False
+
+
+async def bootstrap() -> None:
+ from app.core.config import settings
+ from app.core.database import AsyncSessionLocal
+ from app.services.sec_full_index_service import SECFullIndexService
+ from app.services.insider_transaction_service import InsiderTransactionService
+ from app.services.activist_ownership_service import ActivistOwnershipService
+
+ # Pre-flight: abort if SEC is rate-limiting us to avoid wasted retries
+ if not await _check_sec_available():
+ logger.error("Aborting bootstrap — SEC EDGAR is rate-limiting this IP. Re-run in 30-60 minutes.")
+ return
+
+ n_quarters = settings.SEC_FORM4_BOOTSTRAP_QUARTERS
+ quarters = _quarters_to_process(n_quarters)
+ logger.info(f"Bootstrapping {n_quarters} quarters: {quarters}")
+
+ index_svc = SECFullIndexService()
+ txn_svc = InsiderTransactionService()
+ activist_svc = ActivistOwnershipService()
+
+ for year, quarter in quarters:
+ logger.info(f"Processing {year}/Q{quarter} ...")
+
+ # Form 4 via form345.zip
+ zip_path = os.path.join(tempfile.gettempdir(), f"form345_{year}_Q{quarter}.zip")
+ if not os.path.exists(zip_path):
+ logger.info(f" Downloading form345.zip for {year}/Q{quarter} ...")
+ try:
+ await index_svc._http.fetch_form345_zip(year, quarter, zip_path)
+ except Exception as e:
+ logger.warning(f" form345.zip download failed: {e}; falling back to company.idx")
+ zip_path = None
+
+ form4_entries = []
+ if zip_path and os.path.exists(zip_path):
+ from app.services.sec_full_index_service import parse_form345_zip, FORM4_TYPES
+ form4_entries = parse_form345_zip(zip_path, form_types=FORM4_TYPES)
+ logger.info(f" Form 4 entries from zip: {len(form4_entries)}")
+ else:
+ form4_entries = await index_svc.fetch_quarterly_form4_entries(year, quarter)
+ logger.info(f" Form 4 entries from company.idx: {len(form4_entries)}")
+
+ async with AsyncSessionLocal() as db:
+ inserted = await txn_svc.index_form4_from_index_entries(db, form4_entries)
+ logger.info(f" Form 4 inserted: {inserted}")
+
+ # 13D/G via company.idx
+ activist_entries = await index_svc.fetch_quarterly_activist_entries(year, quarter)
+ logger.info(f" 13D/G entries from company.idx: {len(activist_entries)}")
+ async with AsyncSessionLocal() as db:
+ inserted = await activist_svc.ingest_from_index_entries(db, activist_entries)
+ logger.info(f" 13D/G index-only inserted: {inserted}")
+
+ logger.info("Bootstrap complete. Run the activist enrich job to populate ownership_pct.")
+
+
+if __name__ == "__main__":
+ # Prevent multiple concurrent runs via file lock
+ lock_fh = open(_LOCK_FILE, "w")
+ try:
+ fcntl.flock(lock_fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ except OSError:
+ logger.error("Bootstrap is already running (lock file held). Exiting.")
+ sys.exit(1)
+
+ try:
+ asyncio.run(bootstrap())
+ finally:
+ fcntl.flock(lock_fh, fcntl.LOCK_UN)
+ lock_fh.close()
+ try:
+ os.unlink(_LOCK_FILE)
+ except OSError:
+ pass
diff --git a/scripts/bootstrap_form4_by_ticker.py b/scripts/bootstrap_form4_by_ticker.py
new file mode 100644
index 0000000..5279545
--- /dev/null
+++ b/scripts/bootstrap_form4_by_ticker.py
@@ -0,0 +1,138 @@
+"""
+Bootstrap Form 4 insider transactions by iterating over all universe tickers.
+
+Uses index_form4s(ticker, days=730) — the proven submissions-API path — rather
+than the company.idx + per-accession Archives index.json approach (which is
+aggressively rate-limited by SEC EDGAR).
+
+Usage (inside the container):
+ python scripts/bootstrap_form4_by_ticker.py [--resume-from TICKER]
+
+Options:
+ --resume-from TICKER Skip tickers alphabetically before TICKER (for resuming
+ after an interruption or rate-limit pause).
+ --days N Days of history to fetch per ticker (default: 730).
+ --delay-ms N Per-ticker delay in milliseconds (default: 300).
+
+Idempotent: already-indexed accessions are skipped via upsert ON CONFLICT DO NOTHING.
+"""
+
+import asyncio
+import fcntl
+import logging
+import os
+import sys
+import time
+from typing import Optional
+
+from sqlalchemy import text
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
+logger = logging.getLogger("bootstrap_ticker")
+
+_LOCK_FILE = "/tmp/bootstrap_form4_ticker.lock"
+
+
+async def bootstrap(
+ resume_from: Optional[str] = None,
+ days: int = 730,
+ delay_ms: int = 300,
+) -> None:
+ from app.core.database import AsyncSessionLocal
+ from app.services.insider_transaction_service import InsiderTransactionService
+ from sqlalchemy import text
+
+ txn_svc = InsiderTransactionService()
+
+ async with AsyncSessionLocal() as db:
+ rows = await db.execute(
+ text("SELECT ticker FROM universe_ticker_registry ORDER BY ticker")
+ )
+ tickers = [r[0] for r in rows.fetchall()]
+
+ total = len(tickers)
+ logger.info(f"Universe: {total} tickers, fetching {days} days of Form 4 history each")
+
+ if resume_from:
+ resume_from = resume_from.upper()
+ start_idx = next((i for i, t in enumerate(tickers) if t >= resume_from), 0)
+ logger.info(f"Resuming from {resume_from} (index {start_idx}/{total})")
+ tickers = tickers[start_idx:]
+
+ inserted_total = 0
+ errors = 0
+ delay_s = delay_ms / 1000.0
+ _SESSION_RECYCLE = 50 # close/reopen aiohttp session every N tickers
+
+ for i, ticker in enumerate(tickers, 1):
+ # Recycle the aiohttp session periodically to release connection pool memory
+ if i % _SESSION_RECYCLE == 1 and i > 1:
+ await txn_svc._http.close()
+ txn_svc = InsiderTransactionService()
+ import gc
+ gc.collect()
+ logger.info(f"Session recycled at ticker {i} to free memory")
+
+ # SECHttpClient has unbounded _text_cache and _json_cache — clear before each ticker
+ # (disk cache still provides persistence across calls)
+ txn_svc._http._text_cache.clear()
+ txn_svc._http._json_cache.clear()
+
+ try:
+ async with AsyncSessionLocal() as db:
+ # Disable idle-in-transaction timeout: HTTP fetches for 730 days of
+ # filings can take several minutes, exceeding the default 90s limit.
+ await db.execute(text("SET SESSION idle_in_transaction_session_timeout = 0"))
+ await db.commit()
+ n = await txn_svc.index_form4s(db, ticker, days=days, force_refresh=False)
+ inserted_total += n
+ except Exception as e:
+ errors += 1
+ logger.warning(f"[{i}/{len(tickers)}] {ticker}: error — {e}")
+
+ if i % 100 == 0 or i == len(tickers):
+ logger.info(
+ f"Progress: {i}/{len(tickers)} tickers processed, "
+ f"{inserted_total} transactions inserted, {errors} errors"
+ )
+
+ if delay_s > 0:
+ await asyncio.sleep(delay_s)
+
+ logger.info(
+ f"Bootstrap complete: {len(tickers)} tickers, "
+ f"{inserted_total} total transactions inserted, {errors} errors"
+ )
+
+
+if __name__ == "__main__":
+ import argparse
+
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--resume-from", default=None)
+ parser.add_argument("--days", type=int, default=730)
+ parser.add_argument("--delay-ms", type=int, default=300)
+ args = parser.parse_args()
+
+ lock_fh = open(_LOCK_FILE, "w")
+ try:
+ fcntl.flock(lock_fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ except OSError:
+ logger.error("Bootstrap is already running (lock file held). Exiting.")
+ sys.exit(1)
+
+ try:
+ asyncio.run(bootstrap(
+ resume_from=args.resume_from,
+ days=args.days,
+ delay_ms=args.delay_ms,
+ ))
+ finally:
+ fcntl.flock(lock_fh, fcntl.LOCK_UN)
+ lock_fh.close()
+ try:
+ os.unlink(_LOCK_FILE)
+ except OSError:
+ pass
diff --git a/tests/test_activist_ownership_parse.py b/tests/test_activist_ownership_parse.py
new file mode 100644
index 0000000..b569847
--- /dev/null
+++ b/tests/test_activist_ownership_parse.py
@@ -0,0 +1,88 @@
+"""
+Unit tests for activist ownership cover-page parsing (HTML + XML).
+"""
+
+import pytest
+
+from app.services.activist_ownership_service import _parse_cover_html, _parse_cover_xml
+
+
+# Minimal 13D HTML cover page (simplified)
+SAMPLE_13D_HTML = """
+
+
+ | Item 11. Aggregate Amount Beneficially Owned by Each Reporting Person | 12,500,000 |
+ | Item 13. Percent of Class Represented by Amount in Row (11) | 7.2% |
+
+
+"""
+
+SAMPLE_13G_HTML = """
+
+Percent of Class Represented by Amount in Row (11): 5.4 %
+Aggregate Amount Beneficially Owned: 8,000,000
+
+"""
+
+SAMPLE_COVER_XML = """
+
+ 15000000
+ 9.1
+
+"""
+
+SAMPLE_COVER_XML_NAMESPACED = """
+
+ 3000000
+ 3.5
+
+"""
+
+
+class TestParseCoverHtml:
+ def test_extract_ownership_pct(self):
+ pct, shares = _parse_cover_html(SAMPLE_13D_HTML)
+ assert pct == pytest.approx(7.2)
+
+ def test_extract_shares_owned(self):
+ pct, shares = _parse_cover_html(SAMPLE_13D_HTML)
+ assert shares == pytest.approx(12_500_000)
+
+ def test_13g_format(self):
+ pct, shares = _parse_cover_html(SAMPLE_13G_HTML)
+ assert pct == pytest.approx(5.4)
+ assert shares == pytest.approx(8_000_000)
+
+ def test_empty_html(self):
+ pct, shares = _parse_cover_html("")
+ assert pct is None
+ assert shares is None
+
+ def test_no_pct_in_html(self):
+ pct, shares = _parse_cover_html("Aggregate Amount Beneficially Owned: 1,000,000
")
+ assert pct is None
+ assert shares == pytest.approx(1_000_000)
+
+
+class TestParseCoverXml:
+ def test_structured_xml(self):
+ pct, shares = _parse_cover_xml(SAMPLE_COVER_XML)
+ assert pct == pytest.approx(9.1)
+ assert shares == pytest.approx(15_000_000)
+
+ def test_namespaced_xml(self):
+ # Namespace-prefixed elements should be found via {*} wildcard search
+ pct, shares = _parse_cover_xml(SAMPLE_COVER_XML_NAMESPACED)
+ # Either finds or gracefully returns None — no crash
+ assert pct is None or isinstance(pct, float)
+ assert shares is None or isinstance(shares, float)
+
+ def test_invalid_xml(self):
+ pct, shares = _parse_cover_xml(">>")
+ assert pct is None
+ assert shares is None
+
+ def test_empty_xml(self):
+ pct, shares = _parse_cover_xml("")
+ assert pct is None
+ assert shares is None
diff --git a/tests/test_form4_pit.py b/tests/test_form4_pit.py
new file mode 100644
index 0000000..877595a
--- /dev/null
+++ b/tests/test_form4_pit.py
@@ -0,0 +1,191 @@
+"""
+PIT (point-in-time) correctness + business logic tests for Form 4.
+
+Pure unit tests (no DB) for c-suite derivation, purchase_pct formula,
+and service query-building logic.
+"""
+
+import pytest
+from datetime import date, datetime, timezone
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from app.services.insider_transaction_service import InsiderTransactionService
+
+
+# ------------------------------------------------------------------
+# C-suite title derivation
+# ------------------------------------------------------------------
+
+class TestDeriveTitleFlags:
+ def test_ceo_full_title(self):
+ flags = InsiderTransactionService._derive_title_flags("Chief Executive Officer")
+ assert flags == {"is_ceo": True, "is_cfo": False, "is_c_suite": True}
+
+ def test_ceo_abbrev(self):
+ flags = InsiderTransactionService._derive_title_flags("CEO")
+ assert flags["is_ceo"] is True
+ assert flags["is_c_suite"] is True
+
+ def test_cfo_full_title(self):
+ flags = InsiderTransactionService._derive_title_flags("Chief Financial Officer")
+ assert flags["is_cfo"] is True
+ assert flags["is_c_suite"] is True
+
+ def test_principal_financial_officer(self):
+ flags = InsiderTransactionService._derive_title_flags("Principal Financial Officer")
+ assert flags["is_cfo"] is True
+
+ def test_president(self):
+ flags = InsiderTransactionService._derive_title_flags("President, Americas Division")
+ assert flags["is_c_suite"] is True
+ assert flags["is_ceo"] is False
+
+ def test_chairman(self):
+ flags = InsiderTransactionService._derive_title_flags("Chairman of the Board")
+ assert flags["is_c_suite"] is True
+
+ def test_coo(self):
+ flags = InsiderTransactionService._derive_title_flags("COO")
+ assert flags["is_c_suite"] is True
+
+ def test_evp_is_not_csuite(self):
+ # "EVP Sales" does not match CEO/CFO/COO/... or "Chief X Officer"
+ flags = InsiderTransactionService._derive_title_flags("EVP Sales")
+ assert flags["is_c_suite"] is False
+
+ def test_director_is_not_csuite(self):
+ flags = InsiderTransactionService._derive_title_flags("Director")
+ assert all(not v for v in flags.values())
+
+ def test_none_title(self):
+ flags = InsiderTransactionService._derive_title_flags(None)
+ assert flags == {"is_ceo": False, "is_cfo": False, "is_c_suite": False}
+
+ def test_empty_string(self):
+ flags = InsiderTransactionService._derive_title_flags("")
+ assert all(not v for v in flags.values())
+
+ def test_case_insensitive(self):
+ flags = InsiderTransactionService._derive_title_flags("ceo")
+ assert flags["is_ceo"] is True
+
+ def test_ceo_and_president(self):
+ flags = InsiderTransactionService._derive_title_flags("President and CEO")
+ assert flags["is_ceo"] is True
+ assert flags["is_c_suite"] is True
+
+ def test_chief_marketing_officer(self):
+ flags = InsiderTransactionService._derive_title_flags("Chief Marketing Officer")
+ assert flags["is_c_suite"] is True
+ assert flags["is_ceo"] is False
+
+
+# ------------------------------------------------------------------
+# purchase_pct_of_holding formula (tested via _parse_transaction_element mock)
+# ------------------------------------------------------------------
+
+class TestPurchasePct:
+ """Test the purchase_pct_of_holding calculation logic directly."""
+
+ def _compute_pct(self, code, shares, shares_after):
+ """Mirrors the formula in _parse_transaction_element."""
+ if code in ("P", "A") and shares is not None and shares > 0 and shares_after and shares_after > 0:
+ return abs(shares) / shares_after
+ return None
+
+ def test_open_market_buy(self):
+ pct = self._compute_pct("P", 100.0, 1000.0)
+ assert pct == pytest.approx(0.1, rel=1e-6)
+
+ def test_award(self):
+ pct = self._compute_pct("A", 500.0, 5000.0)
+ assert pct == pytest.approx(0.1, rel=1e-6)
+
+ def test_sell_is_none(self):
+ assert self._compute_pct("S", -100.0, 900.0) is None
+
+ def test_zero_shares_after(self):
+ assert self._compute_pct("P", 100.0, 0.0) is None
+
+ def test_none_shares_after(self):
+ assert self._compute_pct("P", 100.0, None) is None
+
+ def test_negative_buy_shares_is_none(self):
+ # Negative shares on a "P" code shouldn't happen, but guard
+ assert self._compute_pct("P", -100.0, 1000.0) is None
+
+ def test_small_buy(self):
+ pct = self._compute_pct("P", 1.0, 1_000_000.0)
+ assert pct == pytest.approx(1e-6, rel=1e-4)
+
+
+# ------------------------------------------------------------------
+# PIT filter logic (mocked DB)
+# ------------------------------------------------------------------
+
+class TestGetForm4Pit:
+ """Verify that get_form4_pit builds correct date conditions."""
+
+ @pytest.mark.asyncio
+ async def test_as_of_is_used_as_upper_bound(self):
+ svc = InsiderTransactionService()
+ mock_db = AsyncMock()
+ mock_result = MagicMock()
+ mock_result.scalar.return_value = 0
+ mock_result.scalars.return_value.all.return_value = []
+ mock_db.execute = AsyncMock(return_value=mock_result)
+
+ rows, total = await svc.get_form4_pit(
+ mock_db, ticker="AAPL", as_of=date(2024, 1, 15)
+ )
+ assert total == 0
+ assert rows == []
+ # Verify execute was called (conditions were built)
+ assert mock_db.execute.called
+
+ @pytest.mark.asyncio
+ async def test_buy_only_adds_code_filter(self):
+ svc = InsiderTransactionService()
+ mock_db = AsyncMock()
+ mock_result = MagicMock()
+ mock_result.scalar.return_value = 0
+ mock_result.scalars.return_value.all.return_value = []
+ mock_db.execute = AsyncMock(return_value=mock_result)
+
+ # Should not raise
+ rows, total = await svc.get_form4_pit(
+ mock_db, ticker="NVDA", as_of=date(2024, 3, 5), buy_only=True
+ )
+ assert mock_db.execute.called
+
+
+# ------------------------------------------------------------------
+# Aggregate query
+# ------------------------------------------------------------------
+
+class TestGetForm4Aggregate:
+
+ @pytest.mark.asyncio
+ async def test_returns_dict_structure(self):
+ svc = InsiderTransactionService()
+ mock_db = AsyncMock()
+ mock_row = MagicMock()
+ mock_row.buy_count = 3
+ mock_row.buy_dollar_total = 150000.0
+ mock_row.cluster_size = 2
+ mock_row.csuite_count = 1
+ mock_row.avg_pct_of_holding = 0.05
+ mock_row.last_filing_date = datetime(2024, 1, 10, tzinfo=timezone.utc)
+ mock_result = MagicMock()
+ mock_result.one.return_value = mock_row
+ mock_db.execute = AsyncMock(return_value=mock_result)
+
+ agg = await svc.get_form4_aggregate(
+ mock_db, ticker="AAPL", as_of=date(2024, 1, 15), window_days=30
+ )
+ assert agg["symbol"] == "AAPL"
+ assert agg["buy_count"] == 3
+ assert agg["cluster_size"] == 2
+ assert agg["csuite_count"] == 1
+ assert "recency_days" in agg
+ assert "buy_dollar_total" in agg
diff --git a/tests/test_sec_full_index_parser.py b/tests/test_sec_full_index_parser.py
new file mode 100644
index 0000000..e30a09c
--- /dev/null
+++ b/tests/test_sec_full_index_parser.py
@@ -0,0 +1,149 @@
+"""
+Unit tests for SEC full-index (company.idx and form.idx) parsers.
+"""
+
+import pytest
+from datetime import date
+
+from app.services.sec_full_index_service import (
+ parse_company_idx,
+ _parse_form_idx,
+ FORM4_TYPES,
+ ACTIVIST_13DG_TYPES,
+)
+
+
+# Realistic company.idx sample — fixed-width, matches real SEC EDGAR format.
+# Column offsets: Company(0-61), Form(62-73), CIK(74-85), Date(86-97), File(98+)
+# The dashes line is the sentinel that marks the start of data rows.
+_H = "Company Name Form Type CIK Date Filed Filename"
+_D = "-" * 120
+
+def _row(company, form_type, cik, filing_date, acc):
+ """Build a fixed-width company.idx data row matching real EDGAR format.
+
+ Real EDGAR layout: company(62) + form_type(12) + 5_spaces + cik(12, left-pad) + date(10) + 2_spaces + filename
+ CIK always has 5 mandatory leading spaces; date is at position 91, filename at 103.
+ """
+ filename = f"edgar/data/{int(cik)}/{acc}-index.htm"
+ cik_str = f"{str(int(cik)):<12}" # 12-char left-aligned CIK (trailing spaces)
+ # 5 mandatory leading spaces + 12-char CIK + date(10) + 2 spaces = positions 74-102
+ line = f"{company:<62}{form_type:<12} {cik_str}{filing_date} {filename}"
+ return line
+
+SAMPLE_IDX = "\n".join([
+ "Full-Index of EDGAR Filings Submitted to the Commission",
+ "",
+ _H,
+ _D,
+ _row("APPLE INC", "4", "320193", "2026-01-15", "0000320193-26-000001"),
+ _row("BERKSHIRE HATHAWAY INC", "SC 13G", "1067983", "2026-02-14", "0001067983-26-000042"),
+ _row("BLACKROCK INC", "SC 13D/A", "1364742", "2026-01-20", "0001364742-26-000010"),
+ _row("TESLA INC", "4/A", "1318605", "2026-01-18", "0001318605-26-000005"),
+ _row("MICROSOFT CORP", "10-K", "789019", "2026-02-01", "0000789019-26-000020"),
+])
+
+
+def test_parse_form4_entries():
+ entries = parse_company_idx(SAMPLE_IDX, form_types=FORM4_TYPES)
+ assert len(entries) == 2
+ form_types = {e.form_type for e in entries}
+ assert "4" in form_types
+ assert "4/A" in form_types
+
+
+def test_parse_activist_entries():
+ entries = parse_company_idx(SAMPLE_IDX, form_types=ACTIVIST_13DG_TYPES)
+ assert len(entries) == 2
+ form_types = {e.form_type for e in entries}
+ assert "SC 13G" in form_types
+ assert "SC 13D/A" in form_types
+
+
+def test_parse_all_entries_no_filter():
+ entries = parse_company_idx(SAMPLE_IDX)
+ # 4, SC 13G, SC 13D/A, 4/A, 10-K
+ assert len(entries) == 5
+
+
+def test_parse_entry_fields():
+ entries = parse_company_idx(SAMPLE_IDX, form_types=FORM4_TYPES)
+ apple = next((e for e in entries if "APPLE" in e.company_name), None)
+ assert apple is not None, f"APPLE not found in {[e.company_name for e in entries]}"
+ assert apple.cik == "0000320193"
+ assert apple.filing_date == date(2026, 1, 15)
+ assert "0000320193-26-000001" in apple.accession_number
+ assert "edgar/data" in apple.filename
+
+
+def test_parse_skips_short_lines():
+ idx = "short\n" + SAMPLE_IDX
+ entries = parse_company_idx(idx, form_types=FORM4_TYPES)
+ # Should not crash; just skip the malformed line
+ assert isinstance(entries, list)
+
+
+def test_parse_empty_text():
+ entries = parse_company_idx("", form_types=FORM4_TYPES)
+ assert entries == []
+
+
+def test_parse_header_only():
+ # Only the 4 header lines (no data rows after the dash separator)
+ lines = SAMPLE_IDX.splitlines()
+ dash_idx = next(i for i, l in enumerate(lines) if set(l.strip()) == {"-"} and len(l.strip()) > 20)
+ header_only = "\n".join(lines[:dash_idx + 1]) # up to and including dash line
+ entries = parse_company_idx(header_only, form_types=FORM4_TYPES)
+ assert entries == []
+
+
+# ------------------------------------------------------------------
+# form.idx parser (form345.zip column layout: Form Type is FIRST)
+# ------------------------------------------------------------------
+
+def _form_row(form_type, company, cik, filing_date, acc):
+ """Build a fixed-width form.idx row (Form Type first, then Company Name).
+
+ Same EDGAR offsets from position 74 onward as company.idx.
+ """
+ filename = f"edgar/data/{int(cik)}/{acc}-index.htm"
+ cik_str = f"{str(int(cik)):<12}" # 12-char left-aligned CIK
+ line = f"{form_type:<12}{company:<62} {cik_str}{filing_date} {filename}"
+ return line
+
+_FORM_IDX = "\n".join([
+ "Form Type Company Name CIK Date Filed Filename",
+ "-" * 120,
+ _form_row("4", "TIM COOK", "320193", "2026-01-15", "0000320193-26-000001"),
+ _form_row("4/A", "ELON MUSK", "1318605", "2026-01-18", "0001318605-26-000005"),
+ _form_row("10-K", "MICROSOFT CORP", "789019", "2026-02-01", "0000789019-26-000020"),
+])
+
+
+def test_form_idx_parse_form4():
+ entries = _parse_form_idx(_FORM_IDX, form_types=FORM4_TYPES)
+ assert len(entries) == 2
+ form_types = {e.form_type for e in entries}
+ assert "4" in form_types
+ assert "4/A" in form_types
+
+
+def test_form_idx_company_name_correct():
+ entries = _parse_form_idx(_FORM_IDX, form_types=FORM4_TYPES)
+ cook = next((e for e in entries if "TIM COOK" in e.company_name), None)
+ assert cook is not None
+ assert cook.cik == "0000320193"
+ assert cook.filing_date == date(2026, 1, 15)
+ assert "0000320193-26-000001" in cook.accession_number
+
+
+def test_form_idx_does_not_mix_offsets_with_company_idx():
+ # Same data fed to company.idx parser would produce garbled output;
+ # form_idx parser must correctly extract form_type from cols 0-12.
+ company_idx_entries = parse_company_idx(_FORM_IDX, form_types=FORM4_TYPES)
+ form_idx_entries = _parse_form_idx(_FORM_IDX, form_types=FORM4_TYPES)
+ # company.idx parser treats "4 TIM COOK..." as company_name="4"
+ # and form_type from offset 62 — misses "4" form type → 0 entries
+ # form.idx parser correctly gets 2 Form 4 entries
+ assert len(form_idx_entries) == 2
+ assert len(company_idx_entries) == 0 # wrong parser on wrong format = empty