diff --git a/alembic/versions/g8a9b0c1d2e3_add_8k_parsing.py b/alembic/versions/g8a9b0c1d2e3_add_8k_parsing.py
new file mode 100644
index 0000000..754bad7
--- /dev/null
+++ b/alembic/versions/g8a9b0c1d2e3_add_8k_parsing.py
@@ -0,0 +1,81 @@
+"""add 8-K parsing: parsed_status/items_json + sec_filing_events
+
+Revision ID: g8a9b0c1d2e3
+Revises: f7a8b9c0d1e2
+Create Date: 2026-04-08
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+revision: str = "g8a9b0c1d2e3"
+down_revision: Union[str, Sequence[str], None] = "f7a8b9c0d1e2"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+ conn = op.get_bind()
+
+ # Add parsed_status + items_json to sec_filings
+ col_names = [row[0] for row in conn.execute(
+ sa.text("SELECT column_name FROM information_schema.columns WHERE table_name='sec_filings'")
+ )]
+ if "parsed_status" not in col_names:
+ op.add_column(
+ "sec_filings",
+ sa.Column("parsed_status", sa.String(20), server_default="pending", nullable=True),
+ )
+ if "items_json" not in col_names:
+ op.add_column(
+ "sec_filings",
+ sa.Column("items_json", postgresql.JSON(), nullable=True),
+ )
+
+ # Mark all existing 8-K filings as 'pending' so the parser can backfill them
+ conn.execute(
+ sa.text(
+ "UPDATE sec_filings SET parsed_status = 'pending' "
+ "WHERE form_type IN ('8-K', '8-K/A') AND parsed_status IS NULL"
+ )
+ )
+
+ # Create sec_filing_events table
+ if not conn.dialect.has_table(conn, "sec_filing_events"):
+ op.create_table(
+ "sec_filing_events",
+ sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
+ sa.Column("ticker", sa.String(10), nullable=False),
+ sa.Column("accession_number", sa.String(30), nullable=False),
+ sa.Column("form_type", sa.String(20), nullable=False),
+ sa.Column("filing_date", postgresql.TIMESTAMP(timezone=True), nullable=False),
+ sa.Column("item_number", sa.String(10), nullable=False),
+ sa.Column("event_type", sa.String(50), nullable=False),
+ sa.Column("title", sa.String(512), nullable=True),
+ sa.Column("summary", sa.Text(), nullable=True),
+ sa.Column("content_source", sa.String(20), nullable=True),
+ sa.Column("created_at", postgresql.TIMESTAMP(timezone=True), nullable=True),
+ sa.Column("updated_at", postgresql.TIMESTAMP(timezone=True), nullable=True),
+ sa.UniqueConstraint("accession_number", "item_number", name="uq_filing_event"),
+ )
+ op.create_index("ix_filing_events_ticker", "sec_filing_events", ["ticker"])
+ op.create_index("ix_filing_events_accession", "sec_filing_events", ["accession_number"])
+ op.create_index(
+ "idx_filing_events_ticker_date",
+ "sec_filing_events",
+ ["ticker", "filing_date"],
+ )
+ op.create_index(
+ "idx_filing_events_ticker_type",
+ "sec_filing_events",
+ ["ticker", "event_type"],
+ )
+ op.create_index("idx_filing_events_date", "sec_filing_events", ["filing_date"])
+
+
+def downgrade() -> None:
+ op.drop_table("sec_filing_events")
+ op.drop_column("sec_filings", "items_json")
+ op.drop_column("sec_filings", "parsed_status")
diff --git a/app/api/v1/endpoints/filings.py b/app/api/v1/endpoints/filings.py
index f9a5187..8ead426 100644
--- a/app/api/v1/endpoints/filings.py
+++ b/app/api/v1/endpoints/filings.py
@@ -18,14 +18,21 @@ from app.schemas.filing import (
BulkFilingSearchItem,
BulkFilingSearchRequest,
BulkFilingSearchResponse,
+ BulkParseRequest,
+ BulkParseResponse,
ExhibitContentResponse,
FilingDocumentInfo,
FilingDocumentListResponse,
+ FilingEventResponse,
+ FilingEventsSearchResponse,
FilingSearchResponse,
FilingSummary,
)
+from app.models.filing_event import SECFilingEvent
from app.services.sec_filings_service import sec_filings_service
+from app.services.sec_8k_parser import sec_8k_parser
from app.utils.cache import with_cache, build_cache_key, get_negative_cached, set_negative_cached
+from sqlalchemy import select, and_, func as sql_func
router = APIRouter()
logger = logging.getLogger("app.api.v1.filings")
@@ -126,6 +133,8 @@ async def search_filings(
primary_document=f.primary_document,
filing_description=f.filing_description,
documents_count=docs_count,
+ parsed_status=f.parsed_status,
+ items=f.items_json if isinstance(f.items_json, list) else None,
)
)
@@ -431,3 +440,139 @@ async def get_exhibit_bulk(
failed_count=len(results) - successful,
query_time_seconds=round(elapsed, 3),
)
+
+
+# ── Filing Events ────────────────────────────────────────────────────────────
+
+@router.get(
+ "/events/{ticker}",
+ response_model=FilingEventsSearchResponse,
+ summary="Get parsed 8-K events for a ticker",
+ description=(
+ "Returns structured events parsed from 8-K filings. Each event corresponds to one "
+ "8-K Item (e.g., Item 8.01 → other_material_event, Item 2.02 → earnings_result).\\n\\n"
+ "If there are unprocessed (pending) filings, they are lazily parsed on first request.\\n\\n"
+ "**Example**: `GET /filings/events/AVGO?start_date=2026-04-01&event_type=other_material_event`"
+ ),
+)
+@with_cache(namespace="filings:events", ttl=1800, key_params=["ticker", "start_date", "end_date", "event_type", "limit", "offset"])
+async def get_filing_events(
+ ticker: str,
+ response: Response,
+ start_date: Optional[str] = Query(None, description="Start date YYYY-MM-DD"),
+ end_date: Optional[str] = Query(None, description="End date YYYY-MM-DD"),
+ event_type: Optional[str] = Query(None, description="Filter by event type (e.g. other_material_event)"),
+ limit: int = Query(20, ge=1, le=200),
+ offset: int = Query(0, ge=0),
+ db: AsyncSession = Depends(get_db),
+):
+ """Return parsed 8-K events for a ticker. Lazy-parses any pending filings first."""
+ ticker_upper = ticker.upper()
+
+ start_dt = end_dt = None
+ try:
+ if start_date:
+ start_dt = datetime.strptime(start_date, "%Y-%m-%d").replace(tzinfo=timezone.utc)
+ if end_date:
+ end_dt = datetime.strptime(end_date, "%Y-%m-%d").replace(
+ hour=23, minute=59, second=59, tzinfo=timezone.utc
+ )
+ except ValueError:
+ raise HTTPException(status_code=400, detail="Invalid date format. Use YYYY-MM-DD")
+
+ # Lazy-parse any pending 8-Ks for this ticker (limit 20 to avoid long waits)
+ try:
+ await asyncio.wait_for(
+ sec_8k_parser.parse_bulk(db, tickers=[ticker_upper], limit=20),
+ timeout=60,
+ )
+ except (asyncio.TimeoutError, Exception) as e:
+ logger.warning(f"Lazy parse timed out/failed for {ticker_upper}: {e}")
+
+ # Query events
+ conditions = [SECFilingEvent.ticker == ticker_upper]
+ if start_dt:
+ conditions.append(SECFilingEvent.filing_date >= start_dt)
+ if end_dt:
+ conditions.append(SECFilingEvent.filing_date <= end_dt)
+ if event_type:
+ conditions.append(SECFilingEvent.event_type == event_type)
+
+ where_clause = and_(*conditions)
+
+ count_result = await db.execute(
+ select(sql_func.count()).select_from(SECFilingEvent).where(where_clause)
+ )
+ total_count = count_result.scalar() or 0
+
+ rows_result = await db.execute(
+ select(SECFilingEvent)
+ .where(where_clause)
+ .order_by(SECFilingEvent.filing_date.desc())
+ .limit(limit)
+ .offset(offset)
+ )
+ rows = rows_result.scalars().all()
+
+ events = [
+ FilingEventResponse(
+ id=str(row.id),
+ ticker=row.ticker,
+ accession_number=row.accession_number,
+ form_type=row.form_type,
+ filing_date=row.filing_date.strftime("%Y-%m-%d"),
+ item_number=row.item_number,
+ event_type=row.event_type,
+ title=row.title,
+ summary=row.summary,
+ content_source=row.content_source,
+ )
+ for row in rows
+ ]
+
+ return FilingEventsSearchResponse(
+ ticker=ticker_upper,
+ events=events,
+ total_count=total_count,
+ metadata={
+ "limit": limit,
+ "offset": offset,
+ "event_type": event_type,
+ },
+ )
+
+
+@router.post(
+ "/events/parse/bulk",
+ response_model=BulkParseResponse,
+ summary="Bulk parse pending 8-K filings",
+ description=(
+ "Parse pending 8-K filings and create structured events. "
+ "Use this to backfill events for existing DB records.\\n\\n"
+ "**Example body**: `{\"tickers\": [\"AVGO\", \"AAPL\"], \"limit\": 50}`\\n"
+ "Omit `tickers` to parse all pending filings (up to `limit`)."
+ ),
+)
+async def parse_8k_bulk(
+ request: BulkParseRequest,
+ db: AsyncSession = Depends(get_db),
+):
+ """Bulk parse pending 8-K filings and persist events."""
+ try:
+ result = await asyncio.wait_for(
+ sec_8k_parser.parse_bulk(db, tickers=request.tickers, limit=request.limit),
+ timeout=600,
+ )
+ except asyncio.TimeoutError:
+ raise HTTPException(status_code=504, detail="Bulk parse timed out after 600s")
+ except Exception as e:
+ logger.error(f"Bulk parse failed: {e}")
+ raise HTTPException(status_code=502, detail=f"Bulk parse failed: {e}")
+
+ return BulkParseResponse(
+ succeeded=result["succeeded"],
+ failed=result["failed"],
+ skipped=result["skipped"],
+ total=result["total"],
+ query_time_seconds=result["elapsed"],
+ )
diff --git a/app/models/__init__.py b/app/models/__init__.py
index cc11285..130a1be 100644
--- a/app/models/__init__.py
+++ b/app/models/__init__.py
@@ -1,6 +1,7 @@
from app.models.financial import Company, FinancialData, CalculatedMetrics, PriceData, DataUpdateLog
from app.models.etf import CusipMap, ETFCIKMap, ETFSeriesMap, ETFHoldingsSnapshot, ETFHolding
from app.models.filing import SECFiling
+from app.models.filing_event import SECFilingEvent
from app.models.finra_short_volume import FinraShortVolume
from app.models.alpaca_price import AlpacaPriceData
from app.models.overlay_registry import CompanyAlias, YouTubeChannelRegistry, WikiPageMap, ThemeTopicMap
@@ -9,6 +10,7 @@ from app.models.overlay_feature import OverlayFeatureRecord, OverlayJobLog
from app.models.insider_transaction import InsiderTransaction
from app.models.earnings_surprise import EarningsSurprise
from app.models.universe_snapshot import UniverseTickerRegistry, UniverseSnapshot
+from app.models.dividend_calendar import DividendCalendar
__all__ = [
"Company",
@@ -22,6 +24,7 @@ __all__ = [
"ETFHoldingsSnapshot",
"ETFHolding",
"SECFiling",
+ "SECFilingEvent",
"FinraShortVolume",
"AlpacaPriceData",
# Overlay
@@ -41,4 +44,6 @@ __all__ = [
# Universe
"UniverseTickerRegistry",
"UniverseSnapshot",
+ # Dividends
+ "DividendCalendar",
]
\ No newline at end of file
diff --git a/app/models/filing.py b/app/models/filing.py
index b682938..165bf44 100644
--- a/app/models/filing.py
+++ b/app/models/filing.py
@@ -2,7 +2,7 @@
Database model for SEC filings (8-K, 6-K, 20-F, 40-F, 10-K, 10-Q, etc.)
"""
-from sqlalchemy import Column, String, JSON, Index
+from sqlalchemy import Column, String, JSON, Index, Text
from sqlalchemy.dialects.postgresql import UUID, TIMESTAMP
from datetime import datetime, timezone
import uuid
@@ -23,6 +23,8 @@ class SECFiling(Base):
primary_document_url = Column(String(512))
filing_description = Column(String(512))
documents_json = Column(JSON) # Cached document list
+ parsed_status = Column(String(20), default="pending") # pending|succeeded|failed|skipped
+ items_json = Column(JSON) # Extracted item numbers, e.g. ["8.01", "9.01"]
indexed_at = Column(TIMESTAMP(timezone=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))
diff --git a/app/models/filing_event.py b/app/models/filing_event.py
new file mode 100644
index 0000000..5867014
--- /dev/null
+++ b/app/models/filing_event.py
@@ -0,0 +1,39 @@
+"""
+Database model for parsed SEC 8-K filing events.
+Each row represents one Item extracted from an 8-K primary document.
+"""
+
+from datetime import datetime, timezone
+import uuid
+
+from sqlalchemy import Column, String, Text, Index
+from sqlalchemy.dialects.postgresql import UUID, TIMESTAMP, JSON
+
+from app.core.database import Base
+
+
+class SECFilingEvent(Base):
+ __tablename__ = "sec_filing_events"
+
+ id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+ ticker = Column(String(10), nullable=False, index=True)
+ accession_number = Column(String(30), nullable=False, index=True)
+ form_type = Column(String(20), nullable=False)
+ filing_date = Column(TIMESTAMP(timezone=True), nullable=False)
+ item_number = Column(String(10), nullable=False) # e.g. "8.01"
+ event_type = Column(String(50), nullable=False) # e.g. "other_material_event"
+ title = Column(String(512)) # Item title from document
+ summary = Column(Text) # Extracted text content (truncated)
+ content_source = Column(String(20)) # "primary_doc" or "exhibit"
+ 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__ = (
+ Index("idx_filing_events_ticker_date", "ticker", "filing_date"),
+ Index("idx_filing_events_ticker_type", "ticker", "event_type"),
+ Index("idx_filing_events_date", "filing_date"),
+ )
diff --git a/app/schemas/filing.py b/app/schemas/filing.py
index b5074ae..ec4f6a7 100644
--- a/app/schemas/filing.py
+++ b/app/schemas/filing.py
@@ -35,6 +35,8 @@ class FilingSummary(BaseModel):
"primary_document": "a8-k20240201.htm",
"filing_description": "Results of Operations and Financial Condition",
"documents_count": 4,
+ "parsed_status": "succeeded",
+ "items": ["2.02", "9.01"],
}
})
@@ -45,6 +47,8 @@ class FilingSummary(BaseModel):
primary_document: Optional[str] = None
filing_description: Optional[str] = None
documents_count: Optional[int] = None
+ parsed_status: Optional[str] = None
+ items: Optional[List[str]] = None
class FilingSearchResponse(BaseModel):
@@ -192,3 +196,57 @@ class BulkExhibitResponse(BaseModel):
successful_count: int
failed_count: int
query_time_seconds: float
+
+
+# ── Filing Events ────────────────────────────────────────────────────────────
+
+class FilingEventResponse(BaseModel):
+ model_config = ConfigDict(json_schema_extra={
+ "example": {
+ "id": "550e8400-e29b-41d4-a716-446655440000",
+ "ticker": "AVGO",
+ "accession_number": "0001193125-26-144028",
+ "form_type": "8-K",
+ "filing_date": "2026-04-06",
+ "item_number": "8.01",
+ "event_type": "other_material_event",
+ "title": "Other Events",
+ "summary": "Broadcom Inc. and Google LLC have entered into a Long Term Agreement...",
+ "content_source": "primary_doc",
+ }
+ })
+
+ id: str
+ ticker: str
+ accession_number: str
+ form_type: str
+ filing_date: str
+ item_number: str
+ event_type: str
+ title: Optional[str] = None
+ summary: Optional[str] = None
+ content_source: Optional[str] = None
+
+
+class FilingEventsSearchResponse(BaseModel):
+ ticker: str
+ events: List[FilingEventResponse]
+ total_count: int
+ metadata: Dict[str, Any] = Field(default_factory=dict)
+
+
+class BulkParseRequest(BaseModel):
+ model_config = ConfigDict(json_schema_extra={
+ "example": {"tickers": ["AVGO", "AAPL"], "limit": 50}
+ })
+
+ tickers: Optional[List[str]] = None # None = all pending
+ limit: int = Field(default=100, ge=1, le=1000)
+
+
+class BulkParseResponse(BaseModel):
+ succeeded: int
+ failed: int
+ skipped: int
+ total: int
+ query_time_seconds: float
diff --git a/app/services/sec_8k_parser.py b/app/services/sec_8k_parser.py
new file mode 100644
index 0000000..9331786
--- /dev/null
+++ b/app/services/sec_8k_parser.py
@@ -0,0 +1,431 @@
+"""
+SEC 8-K Parser Service
+
+Parses 8-K primary document HTML to extract Items and create structured events.
+Supports Item 8.01 standalone (no exhibit required) by reading primary doc body.
+"""
+
+import asyncio
+import logging
+import re
+import time as _time
+from datetime import datetime, timezone
+from typing import Dict, List, Optional, Tuple
+
+import uuid
+from bs4 import BeautifulSoup
+from sqlalchemy import select, and_, func
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.filing import SECFiling
+from app.models.filing_event import SECFilingEvent
+from app.services.sec_http_client import SECHttpClient
+
+logger = logging.getLogger(__name__)
+
+# Maximum text size for summary storage
+MAX_SUMMARY_CHARS = 10_000
+# Maximum size for primary document fetch
+MAX_PRIMARY_DOC_SIZE = 5 * 1024 * 1024 # 5 MB
+
+# Item regex: matches "Item X.XX" with optional punctuation and title
+ITEM_RE = re.compile(
+ r"Item\s+(\d+\.\d+)\s*[.\:\u2014\u2013\u2012\-]?\s*(.*)",
+ re.IGNORECASE,
+)
+# End-of-items marker
+END_MARKER_RE = re.compile(r"\bSIGNATURE[S]?\b", re.IGNORECASE)
+
+# Batch size for bulk parse queries
+_BULK_CHUNK = 50
+
+
+class SEC8KParser:
+ """Parse 8-K primary documents and persist structured events to DB."""
+
+ # Maps 8-K Item numbers to semantic event types
+ ITEM_EVENT_MAP: Dict[str, Optional[str]] = {
+ "1.01": "material_contract",
+ "1.02": "contract_termination",
+ "1.03": "bankruptcy",
+ "1.04": "mine_safety",
+ "2.01": "acquisition_disposition",
+ "2.02": "earnings_result",
+ "2.03": "financial_obligation",
+ "2.04": "triggering_event",
+ "2.05": "exit_activity",
+ "2.06": "material_impairment",
+ "3.01": "delisting_notice",
+ "3.02": "unregistered_equity_sale",
+ "3.03": "rights_modification",
+ "4.01": "accountant_change",
+ "4.02": "financial_restatement",
+ "5.01": "control_change",
+ "5.02": "management_change",
+ "5.03": "articles_amendment",
+ "5.05": "bylaws_amendment",
+ "5.06": "shell_status_change",
+ "5.07": "shareholder_vote",
+ "5.08": "shareholder_nomination",
+ "7.01": "regulation_fd",
+ "8.01": "other_material_event",
+ "9.01": None, # Exhibits listing — not a separate event
+ }
+
+ # Items that may have richer content in EX-99.1 (press release)
+ EXHIBIT_ENRICHABLE = {"2.02", "7.01", "8.01"}
+
+ def __init__(self) -> None:
+ self._http = SECHttpClient("Stock Oracle 8K Parser")
+
+ # ------------------------------------------------------------------
+ # Public: parse one filing by accession number
+ # ------------------------------------------------------------------
+
+ async def parse_filing(self, db: AsyncSession, accession_number: str) -> int:
+ """Parse a single filing. Returns number of events created/updated."""
+ result = await db.execute(
+ select(SECFiling).where(SECFiling.accession_number == accession_number)
+ )
+ filing = result.scalar_one_or_none()
+ if not filing:
+ raise ValueError(f"Filing {accession_number} not found in DB")
+ return await self._parse_one(db, filing)
+
+ # ------------------------------------------------------------------
+ # Public: parse multiple filings by accession numbers
+ # ------------------------------------------------------------------
+
+ async def parse_filings_by_accessions(
+ self, db: AsyncSession, accession_numbers: List[str]
+ ) -> Dict[str, int]:
+ """Parse multiple filings. Returns {accession: events_count}."""
+ result = await db.execute(
+ select(SECFiling).where(SECFiling.accession_number.in_(accession_numbers))
+ )
+ filings = result.scalars().all()
+ results: Dict[str, int] = {}
+ for filing in filings:
+ try:
+ n = await self._parse_one(db, filing)
+ results[filing.accession_number] = n
+ except Exception as e:
+ logger.warning(f"Parse failed for {filing.accession_number}: {e}")
+ results[filing.accession_number] = 0
+ return results
+
+ # ------------------------------------------------------------------
+ # Public: bulk parse pending 8-K filings
+ # ------------------------------------------------------------------
+
+ async def parse_bulk(
+ self,
+ db: AsyncSession,
+ tickers: Optional[List[str]] = None,
+ limit: int = 100,
+ ) -> Dict:
+ """Parse pending 8-K filings, optionally filtered by ticker.
+
+ Returns summary dict with succeeded/failed/skipped counts.
+ """
+ t0 = _time.monotonic()
+
+ conditions = [
+ SECFiling.parsed_status == "pending",
+ SECFiling.form_type.in_(["8-K", "8-K/A"]),
+ ]
+ if tickers:
+ conditions.append(SECFiling.ticker.in_([t.upper() for t in tickers]))
+
+ result = await db.execute(
+ select(SECFiling)
+ .where(and_(*conditions))
+ .order_by(SECFiling.filing_date.desc())
+ .limit(limit)
+ )
+ filings = result.scalars().all()
+
+ if not filings:
+ return {
+ "succeeded": 0,
+ "failed": 0,
+ "skipped": 0,
+ "total": 0,
+ "elapsed": 0.0,
+ }
+
+ sem = asyncio.Semaphore(4)
+ succeeded = failed = skipped = 0
+
+ async def _parse_limited(f: SECFiling) -> str:
+ async with sem:
+ from app.core.database import AsyncSessionLocal
+ async with AsyncSessionLocal() as session:
+ try:
+ n = await self._parse_one(session, f)
+ return "succeeded" if n > 0 else "skipped"
+ except Exception as e:
+ logger.warning(f"Bulk parse failed {f.accession_number}: {e}")
+ return "failed"
+
+ outcomes = await asyncio.gather(*[_parse_limited(f) for f in filings])
+ for outcome in outcomes:
+ if outcome == "succeeded":
+ succeeded += 1
+ elif outcome == "failed":
+ failed += 1
+ else:
+ skipped += 1
+
+ return {
+ "succeeded": succeeded,
+ "failed": failed,
+ "skipped": skipped,
+ "total": len(filings),
+ "elapsed": round(_time.monotonic() - t0, 3),
+ }
+
+ # ------------------------------------------------------------------
+ # Internal: parse a single SECFiling object
+ # ------------------------------------------------------------------
+
+ async def _parse_one(self, db: AsyncSession, filing: SECFiling) -> int:
+ """Fetch and parse the primary document of a filing. Returns event count."""
+ self._http.set_deadline(60.0)
+ now = datetime.now(timezone.utc)
+ accession_number = filing.accession_number
+
+ # Re-fetch filing in the current session to ensure it is tracked for updates
+ result = await db.execute(
+ select(SECFiling).where(SECFiling.accession_number == accession_number)
+ )
+ filing = result.scalar_one_or_none()
+ if not filing:
+ raise ValueError(f"Filing {accession_number} not found in DB")
+
+ try:
+ # Get document list (cached in documents_json)
+ from app.services.sec_filings_service import sec_filings_service
+ docs = await sec_filings_service.get_filing_documents(db, filing.accession_number)
+
+ # Find primary document URL
+ primary_url = _find_primary_doc_url(docs, filing)
+ if not primary_url:
+ logger.warning(
+ f"No primary document found for {filing.accession_number}, skipping"
+ )
+ filing.parsed_status = "skipped"
+ filing.updated_at = now
+ await db.commit()
+ return 0
+
+ # Fetch and parse HTML
+ html = await self._http.fetch_text(primary_url, max_bytes=MAX_PRIMARY_DOC_SIZE)
+ items = extract_items(html)
+
+ if not items:
+ logger.info(f"No items found in {filing.accession_number}, marking skipped")
+ filing.parsed_status = "skipped"
+ filing.items_json = []
+ filing.updated_at = now
+ await db.commit()
+ return 0
+
+ # Build event rows
+ item_numbers = [item["number"] for item in items]
+ events_to_upsert = []
+ for item in items:
+ event_type = self.ITEM_EVENT_MAP.get(item["number"], "other")
+ if event_type is None:
+ continue # skip 9.01 (exhibits listing)
+
+ content = item["content"]
+ content_source = "primary_doc"
+
+ # Optionally enrich with exhibit content for press-release items
+ if item["number"] in self.EXHIBIT_ENRICHABLE:
+ exhibit_content = await _try_exhibit_content(
+ db, filing.accession_number, docs
+ )
+ if exhibit_content and len(exhibit_content) > len(content):
+ content = exhibit_content
+ content_source = "exhibit"
+
+ summary = content[:MAX_SUMMARY_CHARS] if content else None
+ title = item.get("title") or filing.filing_description or ""
+
+ events_to_upsert.append({
+ "id": uuid.uuid4(),
+ "ticker": filing.ticker,
+ "accession_number": filing.accession_number,
+ "form_type": filing.form_type,
+ "filing_date": filing.filing_date,
+ "item_number": item["number"],
+ "event_type": event_type,
+ "title": title[:512] if title else None,
+ "summary": summary,
+ "content_source": content_source,
+ "created_at": now,
+ "updated_at": now,
+ })
+
+ if events_to_upsert:
+ stmt = pg_insert(SECFilingEvent).values(events_to_upsert)
+ stmt = stmt.on_conflict_do_update(
+ constraint="uq_filing_event",
+ set_={
+ "event_type": stmt.excluded.event_type,
+ "title": stmt.excluded.title,
+ "summary": stmt.excluded.summary,
+ "content_source": stmt.excluded.content_source,
+ "updated_at": stmt.excluded.updated_at,
+ },
+ )
+ await db.execute(stmt)
+
+ # Update filing parse status
+ filing.parsed_status = "succeeded" if events_to_upsert else "skipped"
+ filing.items_json = item_numbers
+ filing.updated_at = now
+ await db.commit()
+
+ logger.info(
+ f"Parsed {filing.accession_number} ({filing.ticker}): "
+ f"{len(events_to_upsert)} events from items {item_numbers}"
+ )
+ return len(events_to_upsert)
+
+ except Exception as e:
+ logger.error(f"Parse error for {filing.accession_number}: {e}")
+ try:
+ filing.parsed_status = "failed"
+ filing.updated_at = now
+ await db.commit()
+ except Exception:
+ await db.rollback()
+ raise
+ finally:
+ self._http.clear_deadline()
+
+
+# ------------------------------------------------------------------
+# Module-level helpers
+# ------------------------------------------------------------------
+
+def extract_items(html: str) -> List[Dict]:
+ """Extract 8-K items from primary document HTML.
+
+ Returns list of {"number": "8.01", "title": "Other Events", "content": "..."}.
+ Deduplicates item numbers (table of contents entries overwritten by body entries).
+ """
+ soup = BeautifulSoup(html, "html.parser")
+ # Remove noise tags
+ for tag in soup.find_all(["script", "style", "ix:header"]):
+ tag.decompose()
+ text = soup.get_text(separator="\n")
+
+ # Find all Item headers
+ all_matches = list(ITEM_RE.finditer(text))
+ if not all_matches:
+ return []
+
+ # Deduplicate: keep the *last* occurrence of each item number
+ # (earlier occurrences are usually the table of contents)
+ seen: Dict[str, re.Match] = {}
+ for m in all_matches:
+ num = m.group(1)
+ seen[num] = m # last match wins
+
+ deduped = sorted(seen.values(), key=lambda m: m.start())
+
+ # Find end-of-body marker (SIGNATURES section)
+ last_item_end = deduped[-1].end() if deduped else 0
+ sig_match = END_MARKER_RE.search(text, last_item_end)
+ text_end = sig_match.start() if sig_match else len(text)
+
+ items: List[Dict] = []
+ for i, m in enumerate(deduped):
+ number = m.group(1)
+ title = m.group(2).strip().rstrip(".").strip()
+ content_start = m.end()
+ content_end = deduped[i + 1].start() if i + 1 < len(deduped) else text_end
+ content = text[content_start:content_end].strip()
+ # Clean up excessive whitespace
+ content = re.sub(r"\n{3,}", "\n\n", content)
+ items.append({"number": number, "title": title, "content": content})
+
+ return items
+
+
+def _find_primary_doc_url(docs: List[Dict], filing: SECFiling) -> Optional[str]:
+ """Find the primary 8-K document URL.
+
+ Prefers the stored primary_document_url (always a direct link from SEC EDGAR
+ submissions JSON). Falls back to documents_json with iXBRL viewer URL stripping.
+ """
+ # Primary URL from the submissions JSON is always a direct link — prefer it
+ if filing.primary_document_url:
+ return filing.primary_document_url
+
+ if not docs:
+ return None
+
+ form_type_upper = filing.form_type.upper()
+ # Exact type match (e.g., "8-K")
+ for doc in docs:
+ doc_type = (doc.get("type") or "").upper()
+ if doc_type == form_type_upper:
+ return _strip_ixbrl_viewer(doc.get("url") or "")
+
+ # Fallback: type contains form type base (strip "/A")
+ for doc in docs:
+ doc_type = (doc.get("type") or "").upper()
+ if form_type_upper.replace("/A", "") in doc_type:
+ return _strip_ixbrl_viewer(doc.get("url") or "")
+
+ return None
+
+
+def _strip_ixbrl_viewer(url: str) -> str:
+ """Convert '/ix?doc=/Archives/...' URLs to direct document URLs."""
+ # Pattern: https://www.sec.gov/ix?doc=/Archives/edgar/data/...
+ if "/ix?doc=" in url:
+ idx = url.index("/ix?doc=")
+ path = url[idx + len("/ix?doc="):] # e.g. /Archives/edgar/...
+ if path.startswith("/"):
+ return "https://www.sec.gov" + path
+ return url
+
+
+async def _try_exhibit_content(
+ db: AsyncSession,
+ accession_number: str,
+ docs: List[Dict],
+) -> Optional[str]:
+ """Try to fetch EX-99.1 exhibit content. Returns None on failure."""
+ has_exhibit = any(
+ (d.get("type") or "").upper() == "EX-99.1" for d in docs
+ )
+ if not has_exhibit:
+ return None
+ try:
+ from app.services.sec_filings_service import sec_filings_service
+ result = await asyncio.wait_for(
+ sec_filings_service.get_exhibit_content(db, accession_number, "EX-99.1"),
+ timeout=20,
+ )
+ content = result.get("content", "")
+ if content:
+ # Strip HTML tags from exhibit content
+ soup = BeautifulSoup(content, "html.parser")
+ for tag in soup.find_all(["script", "style"]):
+ tag.decompose()
+ return soup.get_text(separator="\n").strip()
+ return None
+ except Exception:
+ return None
+
+
+# Singleton
+sec_8k_parser = SEC8KParser()
diff --git a/app/services/sec_filings_service.py b/app/services/sec_filings_service.py
index 5d033ee..d70fd6e 100644
--- a/app/services/sec_filings_service.py
+++ b/app/services/sec_filings_service.py
@@ -9,6 +9,8 @@ import time
from datetime import datetime, timezone
from typing import Dict, List, Optional, Set, Tuple
+from app.core.config import settings
+
from bs4 import BeautifulSoup
from sqlalchemy import select, and_, func
from sqlalchemy.ext.asyncio import AsyncSession
@@ -32,6 +34,7 @@ class SECFilingsService:
def __init__(self):
self._http = SECHttpClient("Stock Oracle Filings Service")
self._doc_fetch_locks: Dict[str, asyncio.Lock] = {}
+ self._reindex_locks: Dict[str, asyncio.Lock] = {}
# ------------------------------------------------------------------
# index_filings: fetch from SEC submissions and upsert to DB
@@ -43,12 +46,20 @@ class SECFilingsService:
ticker: str,
form_types: Optional[Set[str]] = None,
force_refresh: bool = False,
+ skip_cache: bool = False,
) -> int:
"""Index filings from SEC submissions endpoint into sec_filings table.
Returns the number of filings indexed (inserted or updated).
+
+ Args:
+ force_refresh: Update existing DB records in addition to inserting new ones.
+ Also implies skip_cache=True.
+ skip_cache: Bypass HTTP cache reads to fetch fresh data from SEC EDGAR.
"""
ticker = ticker.upper()
+ # force_refresh always implies fresh data from SEC
+ effective_skip_cache = skip_cache or force_refresh
self._http.set_deadline(120.0)
try:
cik = await self._http.get_company_cik(ticker)
@@ -57,7 +68,7 @@ class SECFilingsService:
cik_int = int(cik)
url = f"{self._http.sec_base_data}/submissions/CIK{cik_int:010d}.json"
- data = await self._http.fetch_json(url)
+ data = await self._http.fetch_json(url, skip_cache=effective_skip_cache)
# Determine which form types to index
target_forms = form_types or self.SUPPORTED_FORM_TYPES
@@ -120,7 +131,7 @@ class SECFilingsService:
continue
older_url = f"{self._http.sec_base_data}/submissions/{name}"
try:
- older = await self._http.fetch_json(older_url)
+ older = await self._http.fetch_json(older_url, skip_cache=effective_skip_cache)
add_from_block(older)
except Exception:
continue
@@ -147,6 +158,9 @@ class SECFilingsService:
existing_accs = {row[0] for row in acc_result.fetchall()}
existing_map = {}
+ # Snapshot pre-existing accessions to identify truly new inserts later
+ _pre_existing_accs: Set[str] = set(existing_accs)
+
# Upsert in chunks to avoid all-or-nothing transaction failures
for chunk_start in range(0, len(raw_filings), self.CHUNK_SIZE):
chunk = raw_filings[chunk_start: chunk_start + self.CHUNK_SIZE]
@@ -180,6 +194,7 @@ class SECFilingsService:
primary_document=rf["primary_document"],
primary_document_url=rf["primary_document_url"],
filing_description=rf["filing_description"],
+ parsed_status="pending" if rf["form_type"] in ("8-K", "8-K/A") else None,
indexed_at=now,
created_at=now,
updated_at=now,
@@ -199,6 +214,21 @@ class SECFilingsService:
await db.rollback()
logger.info(f"Indexed {indexed_count} filings for {ticker}")
+
+ # Auto-parse newly inserted 8-K filings (non-blocking; failures are logged)
+ truly_new_8k = [
+ rf["accession_number"]
+ for rf in raw_filings
+ if rf["form_type"] in ("8-K", "8-K/A")
+ and rf["accession_number"] not in _pre_existing_accs
+ ]
+ if truly_new_8k:
+ from app.services.sec_8k_parser import sec_8k_parser
+ try:
+ await sec_8k_parser.parse_filings_by_accessions(db, truly_new_8k)
+ except Exception as e:
+ logger.warning(f"Auto-parse failed for {ticker} 8-Ks: {e}")
+
return indexed_count
finally:
self._http.clear_deadline()
@@ -252,6 +282,56 @@ class SECFilingsService:
if total_count == 0:
return [], 0
+ else:
+ # Ticker already has filings — check for staleness and incrementally re-index
+ # if the data is older than SEC_DATA_REFRESH_HOURS.
+ staleness_result = await db.execute(
+ select(func.max(SECFiling.indexed_at)).where(SECFiling.ticker == ticker)
+ )
+ last_indexed = staleness_result.scalar()
+ now_utc = datetime.now(timezone.utc)
+ refresh_seconds = settings.SEC_DATA_REFRESH_HOURS * 3600
+ is_stale = last_indexed is None or (
+ (now_utc - last_indexed).total_seconds() > refresh_seconds
+ )
+ if is_stale:
+ # Use per-ticker lock to prevent thundering herd.
+ # If another coroutine is already re-indexing this ticker, skip and
+ # serve stale data rather than stacking up duplicate SEC requests.
+ if ticker not in self._reindex_locks:
+ self._reindex_locks[ticker] = asyncio.Lock()
+ lock = self._reindex_locks[ticker]
+ if not lock.locked():
+ async with lock:
+ try:
+ new_count = await self.index_filings(
+ db, ticker, form_types=form_types, skip_cache=True
+ )
+ if new_count == 0:
+ # No new filings from SEC — touch indexed_at on one row so
+ # the next query doesn't immediately re-trigger staleness.
+ await db.execute(
+ SECFiling.__table__.update()
+ .where(
+ SECFiling.id.in_(
+ select(SECFiling.id)
+ .where(SECFiling.ticker == ticker)
+ .order_by(SECFiling.filing_date.desc())
+ .limit(1)
+ )
+ )
+ .values(indexed_at=now_utc)
+ )
+ await db.commit()
+ # Re-count to include any newly inserted filings
+ count_result = await db.execute(
+ select(func.count()).select_from(SECFiling).where(where_clause)
+ )
+ total_count = count_result.scalar() or 0
+ except Exception as e:
+ logger.warning(f"Staleness re-index failed for {ticker}: {e}")
+ # Serve existing stale data rather than returning empty
+
# Query with pagination
result = await db.execute(
select(SECFiling)
@@ -446,24 +526,44 @@ class SECFilingsService:
indexed_counts = {row.ticker: row.cnt for row in count_result}
missing = [t for t in tickers if indexed_counts.get(t, 0) == 0]
- # Index missing tickers in parallel (Semaphore(4) to not overwhelm SEC)
+ # Check staleness for tickers that already have filings
+ stale: List[str] = []
+ present_tickers = [t for t in tickers if indexed_counts.get(t, 0) > 0]
+ if present_tickers:
+ staleness_rows = await db.execute(
+ select(SECFiling.ticker, func.max(SECFiling.indexed_at))
+ .where(SECFiling.ticker.in_(present_tickers))
+ .group_by(SECFiling.ticker)
+ )
+ now_utc = datetime.now(timezone.utc)
+ refresh_seconds = settings.SEC_DATA_REFRESH_HOURS * 3600
+ for row in staleness_rows:
+ t_name, last_indexed = row[0], row[1]
+ if last_indexed is None or (now_utc - last_indexed).total_seconds() > refresh_seconds:
+ stale.append(t_name)
+
+ # Index missing tickers and re-index stale tickers in parallel (Semaphore(4))
# Each coroutine uses its own session to avoid concurrent-session conflicts
- if missing:
+ to_index = [(t, False) for t in missing] + [(t, True) for t in stale]
+ if to_index:
sem = asyncio.Semaphore(4)
- async def _index_one(ticker: str) -> None:
+ async def _index_one(ticker: str, is_stale: bool) -> None:
async with sem:
try:
from app.core.database import AsyncSessionLocal
async with AsyncSessionLocal() as session:
await asyncio.wait_for(
- self.index_filings(session, ticker, form_types=form_types),
+ self.index_filings(
+ session, ticker, form_types=form_types,
+ skip_cache=is_stale,
+ ),
timeout=120,
)
except Exception as e:
logger.warning(f"Bulk index failed for {ticker}: {e}")
- await asyncio.gather(*[_index_one(t) for t in missing])
+ await asyncio.gather(*[_index_one(t, s) for t, s in to_index])
# Fetch results for all tickers from DB
results = []
diff --git a/tests/test_8k_parser.py b/tests/test_8k_parser.py
new file mode 100644
index 0000000..5cc83ff
--- /dev/null
+++ b/tests/test_8k_parser.py
@@ -0,0 +1,395 @@
+"""
+Unit tests for SEC 8-K parser service.
+
+Tests:
+- extract_items(): HTML parsing, item extraction, deduplication
+- parse_filing(): end-to-end with mocked HTTP
+- _find_primary_doc_url(): iXBRL URL stripping
+- _strip_ixbrl_viewer(): URL conversion
+"""
+
+import asyncio
+import pytest
+import uuid
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from app.services.sec_8k_parser import (
+ SEC8KParser,
+ extract_items,
+ _find_primary_doc_url,
+ _strip_ixbrl_viewer,
+)
+from app.models.filing import SECFiling
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+def _make_filing(**kwargs):
+ defaults = dict(
+ id=uuid.uuid4(),
+ ticker="AVGO",
+ cik="1730168",
+ accession_number="0001193125-26-144028",
+ form_type="8-K",
+ filing_date=datetime(2026, 4, 6, tzinfo=timezone.utc),
+ primary_document="d87999d8k.htm",
+ primary_document_url="https://www.sec.gov/Archives/edgar/data/1730168/000119312526144028/d87999d8k.htm",
+ filing_description=None,
+ documents_json=None,
+ parsed_status="pending",
+ items_json=None,
+ )
+ defaults.update(kwargs)
+ f = MagicMock(spec=SECFiling)
+ for k, v in defaults.items():
+ setattr(f, k, v)
+ return f
+
+
+# ---------------------------------------------------------------------------
+# TestExtractItems: HTML parsing
+# ---------------------------------------------------------------------------
+
+class TestExtractItems:
+
+ def test_single_item_8_01(self):
+ html = """
+
+ Item 8.01 Other Events.
+ Broadcom and Google signed a TPU supply agreement.
+ SIGNATURES
+
+ """
+ items = extract_items(html)
+ assert len(items) == 1
+ assert items[0]["number"] == "8.01"
+ assert "Broadcom" in items[0]["content"]
+ assert "SIGNATURES" not in items[0]["content"]
+
+ def test_multiple_items(self):
+ html = """
+
+ Item 2.02 Results of Operations.
+ Q1 revenue was $14.9 billion.
+ Item 9.01 Financial Statements and Exhibits.
+ (d) Exhibits.
+ SIGNATURES
+
+ """
+ items = extract_items(html)
+ assert len(items) == 2
+ assert items[0]["number"] == "2.02"
+ assert "revenue" in items[0]["content"]
+ assert items[1]["number"] == "9.01"
+
+ def test_deduplication_removes_toc_entries(self):
+ """Table of contents entries appear before body — last occurrence wins."""
+ html = """
+
+ TABLE OF CONTENTS
+ Item 8.01 Other Events...1
+ Item 9.01 Financial Statements...2
+ Item 8.01 Other Events.
+ This is the actual body content.
+ SIGNATURES
+
+ """
+ items = extract_items(html)
+ # Only body occurrences should survive
+ numbers = [i["number"] for i in items]
+ assert numbers.count("8.01") == 1
+ # The body occurrence should have actual content
+ item_8 = next(i for i in items if i["number"] == "8.01")
+ assert "body content" in item_8["content"]
+
+ def test_content_stops_at_signatures(self):
+ html = """
+
+ Item 8.01 Other Events.
+ Material contract signed.
+ SIGNATURES
+ John Doe, CEO
+
+ """
+ items = extract_items(html)
+ assert "SIGNATURES" not in items[0]["content"]
+ assert "John Doe" not in items[0]["content"]
+
+ def test_empty_html_returns_empty(self):
+ items = extract_items("No items here.
")
+ assert items == []
+
+ def test_item_title_extracted(self):
+ html = """
+
+ Item 8.01 Other Events.
+ Some content.
+ SIGNATURES
+
+ """
+ items = extract_items(html)
+ assert items[0]["title"] == "Other Events"
+
+ def test_script_tags_removed(self):
+ html = """
+
+
+ Item 8.01 Other Events.
+ Real content here.
+ SIGNATURES
+
+ """
+ items = extract_items(html)
+ # Script tag contents should not produce items
+ assert len(items) == 1
+ assert items[0]["number"] == "8.01"
+
+ def test_three_items_boundaries(self):
+ html = """
+
+ Item 2.02 Results of Operations.
+ Revenue $10B.
+ Item 8.01 Other Events.
+ Partnership signed.
+ Item 9.01 Financial Statements.
+ Exhibit list.
+ SIGNATURES
+
+ """
+ items = extract_items(html)
+ assert len(items) == 3
+ assert "Revenue" in items[0]["content"]
+ assert "Partnership" in items[1]["content"]
+ assert "Exhibit" in items[2]["content"]
+ # No cross-contamination
+ assert "Partnership" not in items[0]["content"]
+ assert "Revenue" not in items[1]["content"]
+
+
+# ---------------------------------------------------------------------------
+# TestStripIxbrlViewer
+# ---------------------------------------------------------------------------
+
+class TestStripIxbrlViewer:
+
+ def test_strips_ix_doc_prefix(self):
+ url = "https://www.sec.gov/ix?doc=/Archives/edgar/data/1730168/000119312526144028/d87999d8k.htm"
+ result = _strip_ixbrl_viewer(url)
+ assert result == "https://www.sec.gov/Archives/edgar/data/1730168/000119312526144028/d87999d8k.htm"
+
+ def test_passthrough_for_direct_url(self):
+ url = "https://www.sec.gov/Archives/edgar/data/320193/000032019324000006/a8k.htm"
+ assert _strip_ixbrl_viewer(url) == url
+
+ def test_empty_string(self):
+ assert _strip_ixbrl_viewer("") == ""
+
+
+# ---------------------------------------------------------------------------
+# TestFindPrimaryDocUrl
+# ---------------------------------------------------------------------------
+
+class TestFindPrimaryDocUrl:
+
+ def test_prefers_primary_document_url(self):
+ filing = _make_filing(
+ primary_document_url="https://www.sec.gov/Archives/edgar/data/123/000123/doc.htm"
+ )
+ docs = [
+ {"type": "8-K", "url": "https://www.sec.gov/ix?doc=/Archives/edgar/data/123/000123/doc.htm"}
+ ]
+ result = _find_primary_doc_url(docs, filing)
+ # Should use the stored primary_document_url, not the ix?doc= URL
+ assert result == "https://www.sec.gov/Archives/edgar/data/123/000123/doc.htm"
+
+ def test_falls_back_to_docs_when_no_primary_url(self):
+ filing = _make_filing(primary_document_url=None)
+ docs = [
+ {"type": "8-K", "url": "https://www.sec.gov/Archives/edgar/data/123/000123/doc.htm"}
+ ]
+ result = _find_primary_doc_url(docs, filing)
+ assert result == "https://www.sec.gov/Archives/edgar/data/123/000123/doc.htm"
+
+ def test_strips_ixbrl_in_fallback(self):
+ filing = _make_filing(primary_document_url=None)
+ docs = [
+ {"type": "8-K", "url": "https://www.sec.gov/ix?doc=/Archives/edgar/data/123/000123/doc.htm"}
+ ]
+ result = _find_primary_doc_url(docs, filing)
+ assert "/ix?doc=" not in result
+
+ def test_returns_none_for_empty_docs_and_no_url(self):
+ filing = _make_filing(primary_document_url=None)
+ result = _find_primary_doc_url([], filing)
+ assert result is None
+
+ def test_form_type_8k_a_matching(self):
+ filing = _make_filing(form_type="8-K/A", primary_document_url=None)
+ docs = [
+ {"type": "8-K/A", "url": "https://www.sec.gov/Archives/edgar/data/123/000123/doc.htm"}
+ ]
+ result = _find_primary_doc_url(docs, filing)
+ assert result is not None
+
+
+# ---------------------------------------------------------------------------
+# TestSEC8KParserItemEventMap
+# ---------------------------------------------------------------------------
+
+class TestItemEventMap:
+
+ def test_key_items_present(self):
+ parser = SEC8KParser()
+ assert parser.ITEM_EVENT_MAP["8.01"] == "other_material_event"
+ assert parser.ITEM_EVENT_MAP["2.02"] == "earnings_result"
+ assert parser.ITEM_EVENT_MAP["5.02"] == "management_change"
+ assert parser.ITEM_EVENT_MAP["1.01"] == "material_contract"
+ assert parser.ITEM_EVENT_MAP["9.01"] is None # skip
+
+ def test_exhibit_enrichable_set(self):
+ parser = SEC8KParser()
+ assert "8.01" in parser.EXHIBIT_ENRICHABLE
+ assert "2.02" in parser.EXHIBIT_ENRICHABLE
+ assert "9.01" not in parser.EXHIBIT_ENRICHABLE
+
+
+# ---------------------------------------------------------------------------
+# TestParseFiling: end-to-end with mocks
+# ---------------------------------------------------------------------------
+
+AVGO_8K_HTML = """
+
+FORM 8-K CURRENT REPORT
+Date of Report: April 6, 2026
+Broadcom Inc.
+Item 8.01 Other Events.
+Broadcom Inc. and Google LLC have entered into a Long Term Agreement
+for Broadcom to develop and supply custom Tensor Processing Units (TPUs)
+for Google's future generations of TPUs.
+Cautionary Note Regarding Forward-Looking Statements
+SIGNATURES
+John Doe, CEO
+
+"""
+
+MULTI_ITEM_HTML = """
+
+Item 2.02 Results of Operations.
+Q1 2026 revenue was $14.9 billion.
+Item 8.01 Other Events.
+Google partnership announced.
+Item 9.01 Financial Statements and Exhibits.
+(d) Exhibits. See exhibit index.
+SIGNATURES
+
+"""
+
+
+class TestParseFiling:
+
+ @pytest.mark.asyncio
+ async def test_parse_item_8_01_standalone(self):
+ """Item 8.01 standalone — no exhibit needed, content from primary doc."""
+ parser = SEC8KParser()
+ filing = _make_filing()
+
+ docs = [
+ {"type": "8-K", "url": "https://www.sec.gov/Archives/edgar/data/1730168/000119312526144028/d87999d8k.htm"},
+ ]
+
+ # lazy import patched at the source module
+ with patch("app.services.sec_filings_service.sec_filings_service") as mock_svc:
+ mock_svc.get_filing_documents = AsyncMock(return_value=docs)
+ with patch.object(parser._http, "fetch_text", new_callable=AsyncMock) as mock_fetch:
+ mock_fetch.return_value = AVGO_8K_HTML
+ with patch("app.services.sec_8k_parser.pg_insert") as mock_pg_insert:
+ mock_stmt = MagicMock()
+ mock_stmt.on_conflict_do_update.return_value = mock_stmt
+ mock_pg_insert.return_value = mock_stmt
+ db = AsyncMock()
+ db.execute = AsyncMock(side_effect=[
+ _make_db_result(filing), # re-fetch filing
+ MagicMock(), # pg_insert execute
+ ])
+ db.commit = AsyncMock()
+
+ n = await parser._parse_one(db, filing)
+
+ assert n == 1
+ assert filing.parsed_status == "succeeded"
+ assert filing.items_json == ["8.01"]
+
+ @pytest.mark.asyncio
+ async def test_parse_multiple_items_skips_9_01(self):
+ """Multiple items: 2.02+8.01+9.01, 9.01 should be skipped (no event created)."""
+ parser = SEC8KParser()
+ filing = _make_filing(accession_number="0001730168-26-000011")
+ docs = [
+ {"type": "8-K", "url": "https://www.sec.gov/Archives/edgar/data/1730168/000173016826000011/avgo.htm"},
+ ]
+
+ with patch("app.services.sec_filings_service.sec_filings_service") as mock_svc:
+ mock_svc.get_filing_documents = AsyncMock(return_value=docs)
+ with patch.object(parser._http, "fetch_text", new_callable=AsyncMock) as mock_fetch:
+ mock_fetch.return_value = MULTI_ITEM_HTML
+ with patch("app.services.sec_8k_parser.pg_insert") as mock_pg_insert:
+ mock_stmt = MagicMock()
+ mock_stmt.on_conflict_do_update.return_value = mock_stmt
+ mock_pg_insert.return_value = mock_stmt
+ db = AsyncMock()
+ db.execute = AsyncMock(side_effect=[
+ _make_db_result(filing),
+ MagicMock(),
+ ])
+ db.commit = AsyncMock()
+
+ n = await parser._parse_one(db, filing)
+
+ # 2 events (2.02 and 8.01), 9.01 skipped
+ assert n == 2
+ assert filing.parsed_status == "succeeded"
+ assert "9.01" in filing.items_json # items_json includes 9.01 (for reference)
+ assert "2.02" in filing.items_json
+ assert "8.01" in filing.items_json
+
+ @pytest.mark.asyncio
+ async def test_parse_filing_sets_failed_on_error(self):
+ """HTTP fetch failure → parsed_status = 'failed'."""
+ parser = SEC8KParser()
+ filing = _make_filing()
+ docs = [{"type": "8-K", "url": "https://www.sec.gov/Archives/doc.htm"}]
+
+ with patch("app.services.sec_filings_service.sec_filings_service") as mock_svc:
+ mock_svc.get_filing_documents = AsyncMock(return_value=docs)
+ with patch.object(parser._http, "fetch_text", new_callable=AsyncMock) as mock_fetch:
+ mock_fetch.side_effect = RuntimeError("network error")
+ db = AsyncMock()
+ db.execute = AsyncMock(return_value=_make_db_result(filing))
+ db.commit = AsyncMock()
+
+ with pytest.raises(RuntimeError, match="network error"):
+ await parser._parse_one(db, filing)
+
+ assert filing.parsed_status == "failed"
+
+ @pytest.mark.asyncio
+ async def test_parse_filing_not_found_raises(self):
+ """Filing not in DB → ValueError."""
+ parser = SEC8KParser()
+ filing = _make_filing()
+ db = AsyncMock()
+ db.execute = AsyncMock(return_value=_make_db_result(None))
+
+ with pytest.raises(ValueError, match="not found in DB"):
+ await parser._parse_one(db, filing)
+
+
+def _make_db_result(value):
+ # SQLAlchemy result methods (scalar_one_or_none, scalars, etc.) are synchronous,
+ # so use MagicMock (not AsyncMock) to avoid returning unawaited coroutines.
+ result = MagicMock()
+ result.scalar_one_or_none.return_value = value
+ return result