yirugi
/
stock-oracle
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
main
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from '7e1a4699e9'
${ noResults }
stock-oracle/app/services/sec_8k_parser.py

450 lines
16 KiB
Python
Raw Blame History

"""
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,
force_reparse: bool = False,
) -> Dict:
"""Parse 8-K filings, optionally filtered by ticker.
Args:
force_reparse: If True, also reparse succeeded/failed filings (resets to pending).
Returns summary dict with succeeded/failed/skipped counts.
"""
t0 = _time.monotonic()
# Reset failed/succeeded filings to pending when force_reparse is requested
if force_reparse:
reset_conditions = [
SECFiling.parsed_status.in_(["succeeded", "failed"]),
SECFiling.form_type.in_(["8-K", "8-K/A"]),
]
if tickers:
reset_conditions.append(SECFiling.ticker.in_([t.upper() for t in tickers]))
await db.execute(
SECFiling.__table__.update()
.where(and_(*reset_conditions))
.values(parsed_status="pending")
)
await db.commit()
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()
Reference in New Issue View Git Blame Copy Permalink