diff --git a/atlas-terminal/apps/web/src/app/filings/page.tsx b/atlas-terminal/apps/web/src/app/filings/page.tsx index dcc9ae6..c9ad65a 100644 --- a/atlas-terminal/apps/web/src/app/filings/page.tsx +++ b/atlas-terminal/apps/web/src/app/filings/page.tsx @@ -18,6 +18,14 @@ const SECTIONS_SEC: FilingSectionTab[] = [ { key: "item9a", label: "Item 9A: Controls & Procedures", short: "Controls", anchorId: "sec-item-9a" }, ]; +const SECTIONS_SEC_FOREIGN: FilingSectionTab[] = [ + { key: "item1a", label: "Item 3 / 3D: Risk Factors", short: "Risk Factors", anchorId: "sec-item-1a" }, + { key: "item3", label: "Legal / Material Proceedings", short: "Legal", anchorId: "sec-item-3" }, + { key: "item7", label: "Item 5: Operating & Financial Review", short: "OFR", anchorId: "sec-item-7" }, + { key: "item8", label: "Item 18: Financial Statements", short: "Financials", anchorId: "sec-item-8" }, + { key: "item9a", label: "Item 15: Controls & Procedures", short: "Controls", anchorId: "sec-item-9a" }, +]; + const SECTIONS_DART: FilingSectionTab[] = [ { key: "item1a", label: "Investment Risk (II)", short: "Risk", anchorId: "dart-item-1a" }, { key: "item3", label: "Litigation", short: "Legal", anchorId: "dart-item-3" }, @@ -34,9 +42,10 @@ const SECTIONS_EDINET: FilingSectionTab[] = [ { key: "item9a", label: "内部統制", short: "内部統制", anchorId: "edinet-item-9a" }, ]; -function sectionsForJurisdiction(j: FilingJurisdiction): FilingSectionTab[] { +function sectionsForJurisdiction(j: FilingJurisdiction, secForm: string | null): FilingSectionTab[] { if (j === "DART") return SECTIONS_DART; if (j === "EDINET") return SECTIONS_EDINET; + if (secForm && secForm !== "10-K") return SECTIONS_SEC_FOREIGN; return SECTIONS_SEC; } @@ -60,6 +69,7 @@ export default function FilingsPage() { const [error, setError] = useState(""); const [htmlVersion, setHtmlVersion] = useState(0); const [filingSource, setFilingSource] = useState(null); + const [secFilingForm, setSecFilingForm] = useState(null); const [linkMap, setLinkMap] = useState | null>(null); const [infoMessage, setInfoMessage] = useState(""); const [translatedText, setTranslatedText] = useState(""); @@ -68,7 +78,10 @@ export default function FilingsPage() { const previewJ = inferFilingJurisdiction(ticker); const activeJurisdiction = filingSource ?? previewJ; - const sectionTabs = useMemo(() => sectionsForJurisdiction(activeJurisdiction), [activeJurisdiction]); + const sectionTabs = useMemo( + () => sectionsForJurisdiction(activeJurisdiction, activeJurisdiction === "SEC" ? secFilingForm : null), + [activeJurisdiction, secFilingForm], + ); async function loadFiling() { setLoading(true); @@ -78,6 +91,7 @@ export default function FilingsPage() { setAiSummary(""); setLinkMap(null); setInfoMessage(""); + setSecFilingForm(null); const j = inferFilingJurisdiction(ticker); try { @@ -90,6 +104,7 @@ export default function FilingsPage() { if (res.ok) { const data = await res.json(); setFilingSource(mapApiSource(data.source)); + setSecFilingForm(typeof data.filing_form === "string" ? data.filing_form : "10-K"); setSections({ item1a: data.item1a || "", item3: data.item3 || "", @@ -101,6 +116,9 @@ export default function FilingsPage() { if (data.links && typeof data.links === "object") { setLinkMap(data.links as Record); } + if (typeof data.message === "string" && data.message) { + setInfoMessage(data.message); + } setActiveSection("item7"); setHtmlVersion((v) => v + 1); setLoaded(true); @@ -269,8 +287,8 @@ export default function FilingsPage() { }; } return { - title: "10-K Annual Report (SEC)", - body: "Downloads the latest 10-K from SEC EDGAR. The filing is shown with original HTML tables and emphasis, restyled for the terminal dark theme. Section tabs scroll to Item 1A, MD&A, and more.", + title: "Annual Report (SEC)", + body: "Downloads the latest annual SEC filing from EDGAR. US issuers typically use 10-K, while ADRs and foreign issuers often use 20-F or 40-F. The filing is shown with original HTML tables and emphasis, with section tabs for risk factors, management discussion, financials, and controls.", }; }, [previewJ]); @@ -319,7 +337,7 @@ export default function FilingsPage() { {loading ? "Loading..." : previewJ === "SEC" - ? "Load 10-K Filing" + ? "Load SEC Annual Filing" : previewJ === "DART" ? "Load DART Report" : "Load EDINET filing"} diff --git a/atlas-terminal/server/routers/edgar.py b/atlas-terminal/server/routers/edgar.py index 5d55041..29707d7 100644 --- a/atlas-terminal/server/routers/edgar.py +++ b/atlas-terminal/server/routers/edgar.py @@ -1,4 +1,4 @@ -"""SEC EDGAR router -- 10-K section download, cache lookup, and comparison.""" +"""SEC EDGAR router -- annual filing section download, cache lookup, and comparison.""" from fastapi import APIRouter, HTTPException, Query @@ -14,7 +14,7 @@ router = APIRouter() @router.get( "/sections/{ticker}", response_model=EdgarSectionsResponse, - summary="Get 10-K sections (cached or download)", + summary="Get annual SEC filing sections (cached or download)", ) async def get_sections( ticker: str, @@ -24,41 +24,48 @@ async def get_sections( description="Include original HTML (with section anchors) for iframe viewing", ), ): - """Return cleaned Item 1A, 3, 7, 8, 9A texts for *ticker*. + """Return cleaned annual filing sections for *ticker*. If the sections are already cached locally the download is skipped. - When *include_html* is true, the response includes ``html`` (wrapped 10-K + When *include_html* is true, the response includes ``html`` (wrapped filing slice with ``id=atlas-item7`` etc. for in-page scrolling). """ try: from server.services.sec_parser import ( - download_and_extract_all_items, - get_10k_sections, + download_and_extract_all_items_with_form, + get_annual_sections_with_form, get_sec_filing_url, load_10k_html_slice, ) - sections, status = get_10k_sections(ticker.upper(), email) + sections, status, filing_form = get_annual_sections_with_form(ticker.upper(), email) html_payload = "" if include_html: html_payload = load_10k_html_slice(ticker.upper()) or "" if not html_payload: - sections = download_and_extract_all_items(ticker.upper(), email) + sections, filing_form = download_and_extract_all_items_with_form(ticker.upper(), email) status = "downloaded" html_payload = load_10k_html_slice(ticker.upper()) or "" # Resolve actual filing document URL from SEC EDGAR - filing_url = get_sec_filing_url(ticker.upper()) + filing_url = get_sec_filing_url(ticker.upper(), preferred_forms=[filing_form]) links = {} if filing_url: - links["View Original 10-K Filing"] = filing_url + links[f"View Original {filing_form} Filing"] = filing_url links["SEC EDGAR Filings"] = ( f"https://www.sec.gov/cgi-bin/browse-edgar?action=getcompany" - f"&CIK={ticker.upper()}&type=10-K&dateb=&owner=include&count=5" + f"&CIK={ticker.upper()}&type={filing_form}&dateb=&owner=include&count=5" ) return EdgarSectionsResponse( status=status, + filing_form=filing_form, + filing_label=f"{filing_form} Annual Report (SEC)", + message=( + f"Loaded {filing_form} annual filing for {ticker.upper()}." + if filing_form != "10-K" + else None + ), item1a=sections.get("item1a", ""), item3=sections.get("item3", ""), item7=sections.get("item7", ""), @@ -78,13 +85,13 @@ async def get_sections( @router.get( "/item7/{ticker}", response_model=Item7Response, - summary="Get Item 7 (MD&A) text", + summary="Get annual filing MD&A / OFR text", ) async def get_item7( ticker: str, email: str = Query(..., description="SEC EDGAR fair-access email"), ): - """Return only the Item 7 Management Discussion & Analysis text.""" + """Return the main management discussion section from the latest annual filing.""" try: from server.services.sec_parser import download_and_extract_item7_and_1a @@ -99,13 +106,13 @@ async def get_item7( @router.get( "/compare/{ticker}", response_model=CompareResponse, - summary="Latest vs 3-year-ago Item 7 comparison", + summary="Latest vs 3-year-ago annual filing comparison", ) async def compare_item7( ticker: str, email: str = Query(..., description="SEC EDGAR fair-access email"), ): - """Download up to 5 10-Ks and return the latest and 3-year-ago Item 7 for + """Download up to 5 annual filings and return the latest and 3-year-ago Item 7/OFR for comparative analysis. Also returns the latest Item 1A. """ try: diff --git a/atlas-terminal/server/services/sec_parser.py b/atlas-terminal/server/services/sec_parser.py index d7e2ea6..35400d8 100644 --- a/atlas-terminal/server/services/sec_parser.py +++ b/atlas-terminal/server/services/sec_parser.py @@ -1,8 +1,8 @@ -"""SEC EDGAR 10-K download, parsing, section extraction, and caching. +"""SEC EDGAR annual filing download, parsing, section extraction, and caching. -Handles the full pipeline from downloading a 10-K filing via +Handles the full pipeline from downloading an annual filing (10-K / 20-F / 40-F) via ``sec_edgar_downloader`` through HTML stripping to isolating individual -Item sections (1A, 3, 7, 8, 9A) and persisting the cleaned text to a +Item sections (risk, legal/other, MD&A/OFR, financials, controls) and persisting the cleaned text to a local JSON cache under ``data/``. """ @@ -22,6 +22,8 @@ from server.services.text_chunker import clean_text_for_llm, smart_chunk logger = logging.getLogger(__name__) +ANNUAL_SEC_FORMS: List[str] = ["10-K", "20-F", "40-F"] + # --------------------------------------------------------------------------- # Paths # --------------------------------------------------------------------------- @@ -60,12 +62,13 @@ def _resolve_cik(ticker: str) -> Optional[int]: return None -def get_sec_filing_url(ticker: str) -> Optional[str]: - """Return the URL of the latest 10-K filing document on SEC EDGAR.""" +def get_sec_filing_url(ticker: str, preferred_forms: Optional[List[str]] = None) -> Optional[str]: + """Return the URL of the latest annual filing document on SEC EDGAR.""" cik = _resolve_cik(ticker) if cik is None: return None cik_padded = str(cik).zfill(10) + forms_to_match = preferred_forms or ANNUAL_SEC_FORMS try: resp = httpx.get( f"https://data.sec.gov/submissions/CIK{cik_padded}.json", @@ -79,7 +82,7 @@ def get_sec_filing_url(ticker: str) -> Optional[str]: accessions = recent.get("accessionNumber", []) docs = recent.get("primaryDocument", []) for i, form in enumerate(forms): - if form in ("10-K", "10-K/A"): + if form in forms_to_match or form.replace("/A", "") in forms_to_match: acc_no_dash = accessions[i].replace("-", "") return ( f"https://www.sec.gov/Archives/edgar/data" @@ -120,6 +123,36 @@ ITEM9A_PATTERNS: List[str] = [ r"ITEM\s+9A\s*[.:]\s*Controls", ] +ITEM20F_RISK_PATTERNS: List[str] = [ + r"Item\s+3\.?\s*D\s*[.:]\s*Risk\s+Factors", + r"ITEM\s+3\.?\s*D\s*[.:]\s*Risk\s+Factors", + r"Item\s+3\s*[.:][^\n]*Risk\s+Factors", +] + +ITEM20F_MDA_PATTERNS: List[str] = [ + r"Item\s+5\s*[.:]\s*Operating\s+and\s+Financial\s+Review\s+and\s+Prospects", + r"ITEM\s+5\s*[.:]\s*Operating\s+and\s+Financial\s+Review", +] + +ITEM20F_FIN_PATTERNS: List[str] = [ + r"Item\s+18\s*[.:]\s*Financial\s+Statements", + r"ITEM\s+18\s*[.:]\s*Financial\s+Statements", + r"Item\s+17\s*[.:]\s*Financial\s+Statements", + r"Item\s+8\s*[.:]\s*Financial\s+Information", +] + +ITEM20F_CONTROLS_PATTERNS: List[str] = [ + r"Item\s+15\s*[.:]\s*Controls\s+and\s+Procedures", + r"ITEM\s+15\s*[.:]\s*Controls\s+and\s+Procedures", + r"Item\s+15\s*[.:]\s*Disclosure\s+Controls", +] + +ITEM20F_LEGAL_PATTERNS: List[str] = [ + r"Legal\s+Proceedings", + r"Litigation", + r"Arbitration", +] + # --------------------------------------------------------------------------- # HTML helpers @@ -262,14 +295,14 @@ def _get_edgar_downloader() -> type: return Downloader -def find_downloaded_10k_path(download_root: Path, ticker: str) -> Optional[Path]: - """Locate the most recent 10-K filing directory on disk.""" +def find_downloaded_filing_path(download_root: Path, ticker: str, form_type: str) -> Optional[Path]: + """Locate the most recent filing directory for *form_type* on disk.""" ticker_upper = ticker.upper() for base in (download_root / "sec-edgar-filings", download_root): - path_10k = base / ticker_upper / "10-K" - if path_10k.exists(): + filing_path = base / ticker_upper / form_type + if filing_path.exists(): subdirs = sorted( - [d for d in path_10k.iterdir() if d.is_dir()], + [d for d in filing_path.iterdir() if d.is_dir()], key=lambda x: x.name, reverse=True, ) @@ -281,10 +314,10 @@ def find_downloaded_10k_path(download_root: Path, ticker: str) -> Optional[Path] for company_dir in base.iterdir(): if not company_dir.is_dir(): continue - path_10k = company_dir / "10-K" - if path_10k.exists(): + filing_path = company_dir / form_type + if filing_path.exists(): subdirs = sorted( - [d for d in path_10k.iterdir() if d.is_dir()], + [d for d in filing_path.iterdir() if d.is_dir()], key=lambda x: x.name, reverse=True, ) @@ -293,20 +326,30 @@ def find_downloaded_10k_path(download_root: Path, ticker: str) -> Optional[Path] return None -def find_all_10k_filing_dirs(download_root: Path, ticker: str) -> List[Path]: - """Return all 10-K filing directories sorted newest-first.""" +def find_downloaded_10k_path(download_root: Path, ticker: str) -> Optional[Path]: + """Backwards-compatible alias for 10-K directory lookup.""" + return find_downloaded_filing_path(download_root, ticker, "10-K") + + +def find_all_filing_dirs(download_root: Path, ticker: str, form_type: str) -> List[Path]: + """Return all filing directories for *form_type* sorted newest-first.""" ticker_upper = ticker.upper() for base in (download_root / "sec-edgar-filings", download_root): - path_10k = base / ticker_upper / "10-K" - if path_10k.exists(): + filing_path = base / ticker_upper / form_type + if filing_path.exists(): return sorted( - [d for d in path_10k.iterdir() if d.is_dir()], + [d for d in filing_path.iterdir() if d.is_dir()], key=lambda x: x.name, reverse=True, ) return [] +def find_all_10k_filing_dirs(download_root: Path, ticker: str) -> List[Path]: + """Backwards-compatible alias for 10-K directory lookup.""" + return find_all_filing_dirs(download_root, ticker, "10-K") + + def get_main_10k_text(filing_dir: Path) -> str: """Return the longest extracted text from all files in *filing_dir*.""" all_text: List[tuple] = [] @@ -367,11 +410,31 @@ def _strip_scripts_keep_html(html: str) -> str: # Regex bundles for DOM anchor injection (first match in document order wins). _SEC_ITEM_INJECT_SPECS: List[tuple[str, List[re.Pattern]]] = [ - ("sec-item-1a", [re.compile(p, re.I) for p in ITEM1A_PATTERNS] + [re.compile(r"Item\s+1A\s*[.:]", re.I)]), - ("sec-item-3", [re.compile(p, re.I) for p in ITEM3_PATTERNS] + [re.compile(r"Item\s+3\s*[.:]", re.I)]), - ("sec-item-7", [re.compile(p, re.I) for p in ITEM7_PATTERNS] + [re.compile(r"Item\s+7\s*[.:]", re.I)]), - ("sec-item-8", [re.compile(p, re.I) for p in ITEM8_PATTERNS] + [re.compile(r"Item\s+8\s*[.:]", re.I)]), - ("sec-item-9a", [re.compile(p, re.I) for p in ITEM9A_PATTERNS] + [re.compile(r"Item\s+9A\s*[.:]", re.I)]), + ( + "sec-item-1a", + [re.compile(p, re.I) for p in ITEM1A_PATTERNS + ITEM20F_RISK_PATTERNS] + + [re.compile(r"Item\s+1A\s*[.:]", re.I), re.compile(r"Item\s+3\.?\s*D\s*[.:]", re.I)], + ), + ( + "sec-item-3", + [re.compile(p, re.I) for p in ITEM3_PATTERNS + ITEM20F_LEGAL_PATTERNS] + + [re.compile(r"Legal\s+Proceedings", re.I)], + ), + ( + "sec-item-7", + [re.compile(p, re.I) for p in ITEM7_PATTERNS + ITEM20F_MDA_PATTERNS] + + [re.compile(r"Item\s+7\s*[.:]", re.I), re.compile(r"Item\s+5\s*[.:]", re.I)], + ), + ( + "sec-item-8", + [re.compile(p, re.I) for p in ITEM8_PATTERNS + ITEM20F_FIN_PATTERNS] + + [re.compile(r"Item\s+8\s*[.:]", re.I), re.compile(r"Item\s+18\s*[.:]", re.I)], + ), + ( + "sec-item-9a", + [re.compile(p, re.I) for p in ITEM9A_PATTERNS + ITEM20F_CONTROLS_PATTERNS] + + [re.compile(r"Item\s+9A\s*[.:]", re.I), re.compile(r"Item\s+15\s*[.:]", re.I)], + ), ] @@ -548,6 +611,12 @@ def _get_10k_cache_path(ticker: str) -> Path: return _DATA_DIR / f"{ticker.upper()}_latest.json" +def _get_filing_meta_cache_path(ticker: str) -> Path: + """Path for cached filing metadata alongside the section cache.""" + _DATA_DIR.mkdir(parents=True, exist_ok=True) + return _DATA_DIR / f"{ticker.upper()}_latest_meta.json" + + def _load_10k_from_cache(ticker: str) -> Optional[Dict[str, str]]: """Load cached sections or return ``None`` if absent.""" path = _get_10k_cache_path(ticker) @@ -568,51 +637,130 @@ def _save_10k_to_cache(ticker: str, data: Dict[str, str]) -> None: json.dump(data, f, ensure_ascii=False, indent=0) +def _load_filing_meta_from_cache(ticker: str) -> Dict[str, str]: + """Load cached filing metadata, defaulting legacy caches to 10-K.""" + path = _get_filing_meta_cache_path(ticker) + if not path.exists(): + return {"filing_form": "10-K", "filing_label": "10-K Annual Report (SEC)"} + try: + with open(path, "r", encoding="utf-8") as f: + raw = json.load(f) + if isinstance(raw, dict): + filing_form = str(raw.get("filing_form") or "10-K") + filing_label = str(raw.get("filing_label") or f"{filing_form} Annual Report (SEC)") + return {"filing_form": filing_form, "filing_label": filing_label} + except Exception: + pass + return {"filing_form": "10-K", "filing_label": "10-K Annual Report (SEC)"} + + +def _save_filing_meta_to_cache(ticker: str, filing_form: str) -> None: + """Persist latest filing metadata.""" + path = _get_filing_meta_cache_path(ticker) + payload = { + "filing_form": filing_form, + "filing_label": f"{filing_form} Annual Report (SEC)", + } + with open(path, "w", encoding="utf-8") as f: + json.dump(payload, f, ensure_ascii=False, indent=0) + + # --------------------------------------------------------------------------- # High-level download + extract # --------------------------------------------------------------------------- -def download_and_extract_all_items(ticker: str, email: str) -> Dict[str, str]: - """Download latest 10-K, extract Items 1A/3/7/8/9A, clean and cache.""" - Downloader = _get_edgar_downloader() - raw_html: Optional[str] = None - with tempfile.TemporaryDirectory() as tmpdir: - download_root = Path(tmpdir) - dl = Downloader("FQDC-10K-Analyzer", email, str(download_root)) - dl.get("10-K", ticker.upper(), limit=1, download_details=True) - filing_dir = find_downloaded_10k_path(download_root, ticker) - if not filing_dir: - raise FileNotFoundError(f"Could not find 10-K for ticker '{ticker}'.") - full_text = get_main_10k_text(filing_dir) - if not full_text: - raise ValueError("Could not extract text from the 10-K.") - # Read raw HTML while tempdir still exists - raw_html = read_main_10k_html_raw(filing_dir) +def _extract_sections_for_form(full_text: str, filing_form: str) -> Dict[str, str]: + """Extract normalised section buckets for a specific SEC annual form.""" + if filing_form == "20-F": + item1a = find_item_section_generic(full_text, ITEM20F_RISK_PATTERNS, 3, ["Risk", "Factors"], max_chars=80_000) + item3 = find_item_section_generic(full_text, ITEM20F_LEGAL_PATTERNS, 8, ["Legal", "Proceedings", "Arbitration"], max_chars=40_000) + start7 = _find_section_start(full_text, ITEM20F_MDA_PATTERNS, 5) + text_after_7 = full_text[start7:] if start7 >= 0 else full_text + item7 = find_item_section_generic( + text_after_7, + ITEM20F_MDA_PATTERNS, + 5, + ["Operating", "Financial", "Review", "Prospects"], + max_chars=100_000, + ) + if not item7 and text_after_7: + item7 = text_after_7[:120_000] + item8 = find_item_section_generic( + full_text, + ITEM20F_FIN_PATTERNS, + 18, + ["Financial Statements", "Financial Information"], + max_chars=200_000, + ) + item9a = find_item_section_generic( + full_text, + ITEM20F_CONTROLS_PATTERNS, + 15, + ["Controls", "Procedures", "Internal"], + max_chars=50_000, + ) + else: + item1a = find_item_section_generic(full_text, ITEM1A_PATTERNS, 1, ["Risk", "Factors"], max_chars=80_000) + item3 = _extract_item_from_full(full_text, ITEM3_PATTERNS, 3, ["Legal", "Proceedings"], max_chars=40_000) + item9a = _extract_item_from_full(full_text, ITEM9A_PATTERNS, 9, ["Controls", "Procedures", "Internal"], max_chars=40_000) + start7 = _find_section_start(full_text, ITEM7_PATTERNS, 7) + text_after_7 = full_text[start7:] if start7 >= 0 else full_text + item7 = find_item_section_generic(text_after_7, ITEM7_PATTERNS, 7, ["Management's Discussion", "MD&A", "Analysis"], max_chars=100_000) + if not item7 and text_after_7: + item7 = text_after_7[:120_000] + item8 = _extract_item_from_full(full_text, ITEM8_PATTERNS, 8, ["Financial Statements", "Supplementary Data"], max_chars=200_000) - item1a = find_item_section_generic(full_text, ITEM1A_PATTERNS, 1, ["Risk", "Factors"], max_chars=80_000) - item3 = _extract_item_from_full(full_text, ITEM3_PATTERNS, 3, ["Legal", "Proceedings"], max_chars=40_000) - item9a = _extract_item_from_full(full_text, ITEM9A_PATTERNS, 9, ["Controls", "Procedures", "Internal"], max_chars=40_000) - - start7 = _find_section_start(full_text, ITEM7_PATTERNS, 7) - text_after_7 = full_text[start7:] if start7 >= 0 else full_text - item7 = find_item_section_generic(text_after_7, ITEM7_PATTERNS, 7, ["Management's Discussion", "MD&A", "Analysis"], max_chars=100_000) - if not item7 and text_after_7: - item7 = text_after_7[:120_000] - item8 = _extract_item_from_full(full_text, ITEM8_PATTERNS, 8, ["Financial Statements", "Supplementary Data"], max_chars=200_000) - - data: Dict[str, str] = { + return { "item1a": clean_text_for_llm(item1a or ""), "item3": clean_text_for_llm(item3 or ""), "item9a": clean_text_for_llm(item9a or ""), "item7": clean_text_for_llm(item7 or ""), "item8": clean_text_for_llm(item8 or ""), } + + +def _download_latest_annual_filing_dir(download_root: Path, ticker: str, email: str, limit: int = 1) -> tuple[str, Path]: + """Download the latest available annual SEC filing directory for a ticker.""" + Downloader = _get_edgar_downloader() + dl = Downloader("FQDC-10K-Analyzer", email, str(download_root)) + for filing_form in ANNUAL_SEC_FORMS: + dl.get(filing_form, ticker.upper(), limit=limit, download_details=True) + filing_dir = find_downloaded_filing_path(download_root, ticker, filing_form) + if filing_dir: + return filing_form, filing_dir + raise FileNotFoundError( + f"Could not find annual SEC filing (10-K / 20-F / 40-F) for ticker '{ticker}'." + ) + + +def download_and_extract_all_items_with_form(ticker: str, email: str) -> tuple[Dict[str, str], str]: + """Download latest annual filing, extract sections, and return filing form.""" + Downloader = _get_edgar_downloader() + raw_html: Optional[str] = None + with tempfile.TemporaryDirectory() as tmpdir: + download_root = Path(tmpdir) + _ = Downloader + filing_form, filing_dir = _download_latest_annual_filing_dir(download_root, ticker, email, limit=1) + full_text = get_main_10k_text(filing_dir) + if not full_text: + raise ValueError(f"Could not extract text from the {filing_form}.") + # Read raw HTML while tempdir still exists + raw_html = read_main_10k_html_raw(filing_dir) + + data = _extract_sections_for_form(full_text, filing_form) _save_10k_to_cache(ticker, data) + _save_filing_meta_to_cache(ticker, filing_form) if raw_html: fragment = prepare_native_html_fragment_from_10k_raw(raw_html) if fragment: save_10k_html_slice(ticker, fragment) + return data, filing_form + + +def download_and_extract_all_items(ticker: str, email: str) -> Dict[str, str]: + """Backwards-compatible wrapper returning only the section payload.""" + data, _ = download_and_extract_all_items_with_form(ticker, email) return data @@ -624,9 +772,19 @@ def get_10k_sections(ticker: str, email: str) -> tuple[Dict[str, str], str]: return download_and_extract_all_items(ticker, email), "downloaded" +def get_annual_sections_with_form(ticker: str, email: str) -> tuple[Dict[str, str], str, str]: + """Return sections, cache status, and detected annual SEC filing form.""" + cached = _load_10k_from_cache(ticker) + if cached is not None: + meta = _load_filing_meta_from_cache(ticker) + return cached, "cache", meta.get("filing_form", "10-K") + data, filing_form = download_and_extract_all_items_with_form(ticker, email) + return data, "downloaded", filing_form + + def download_and_extract_item7_and_1a(ticker: str, email: str) -> tuple[str, str, str]: - """Fetch 10-K and return ``(full_text, item1a, item7)``.""" - sections, _ = get_10k_sections(ticker, email) + """Fetch annual filing and return ``(full_text, item1a, item7)``.""" + sections, _, _ = get_annual_sections_with_form(ticker, email) return "", sections.get("item1a", "") or "", sections.get("item7", "") or "" @@ -634,38 +792,34 @@ def download_item7_latest_and_3y_ago( ticker: str, email: str, ) -> tuple[Optional[str], Optional[str], Optional[str], bool]: - """Download up to 5 10-Ks; return item1a (latest), item7 latest, item7 3y ago, has_comparison.""" - Downloader = _get_edgar_downloader() + """Download up to 5 annual filings and compare latest vs 3-years-ago MD&A/OFR.""" with tempfile.TemporaryDirectory() as tmpdir: download_root = Path(tmpdir) - dl = Downloader("FQDC-10K-Analyzer", email, str(download_root)) - dl.get("10-K", ticker.upper(), limit=5, download_details=True) - filing_dirs = find_all_10k_filing_dirs(download_root, ticker) + filing_form, _ = _download_latest_annual_filing_dir(download_root, ticker, email, limit=5) + filing_dirs = find_all_filing_dirs(download_root, ticker, filing_form) if not filing_dirs: - raise FileNotFoundError(f"Could not find 10-K for ticker '{ticker}'.") + raise FileNotFoundError( + f"Could not find annual SEC filing (10-K / 20-F / 40-F) for ticker '{ticker}'." + ) full_latest = get_main_10k_text(filing_dirs[0]) if not full_latest: - raise ValueError("Could not extract text from the latest 10-K.") + raise ValueError(f"Could not extract text from the latest {filing_form}.") - item1a = find_item_section_generic(full_latest, ITEM1A_PATTERNS, 1, ["Risk", "Factors"], max_chars=80_000) - - s7 = _find_section_start(full_latest, ITEM7_PATTERNS, 7) - text_after_7 = full_latest[s7:] if s7 >= 0 else full_latest - item7_latest = find_item_section_generic(text_after_7, ITEM7_PATTERNS, 7, ["Management's Discussion", "MD&A", "Analysis"], max_chars=100_000) - if not item7_latest and text_after_7: - item7_latest = smart_chunk(text_after_7[:120_000], max_chars=20_000) + latest_sections = _extract_sections_for_form(full_latest, filing_form) + item1a = latest_sections.get("item1a", "") + item7_latest = latest_sections.get("item7", "") + if not item7_latest and full_latest: + item7_latest = smart_chunk(full_latest[:120_000], max_chars=20_000) item7_3y_ago: Optional[str] = None has_comparison = False if len(filing_dirs) >= 4: full_3y = get_main_10k_text(filing_dirs[3]) if full_3y: - s7_3y = _find_section_start(full_3y, ITEM7_PATTERNS, 7) - text_3y = full_3y[s7_3y:] if s7_3y >= 0 else full_3y - item7_3y_ago = find_item_section_generic(text_3y, ITEM7_PATTERNS, 7, ["Management's Discussion", "MD&A", "Analysis"], max_chars=100_000) - if not item7_3y_ago and text_3y: - item7_3y_ago = smart_chunk(text_3y[:120_000], max_chars=20_000) + item7_3y_ago = _extract_sections_for_form(full_3y, filing_form).get("item7", "") + if not item7_3y_ago: + item7_3y_ago = smart_chunk(full_3y[:120_000], max_chars=20_000) has_comparison = bool(item7_3y_ago) return item1a or "", item7_latest or "", item7_3y_ago, has_comparison diff --git a/atlas-terminal/tests/test_sec_parser.py b/atlas-terminal/tests/test_sec_parser.py new file mode 100644 index 0000000..0f684b4 --- /dev/null +++ b/atlas-terminal/tests/test_sec_parser.py @@ -0,0 +1,56 @@ +from pathlib import Path + +from server.services import sec_parser + + +def test_download_and_extract_all_items_with_form_falls_back_to_20f(monkeypatch) -> None: + calls: list[str] = [] + saved_meta: dict[str, str] = {} + + class FakeDownloader: + def __init__(self, *_args, **_kwargs) -> None: + pass + + def get(self, form_type: str, ticker: str, limit: int = 1, download_details: bool = True) -> None: + calls.append(form_type) + + def fake_find_downloaded_filing_path(_download_root: Path, _ticker: str, form_type: str): + if form_type == "20-F": + return Path("/tmp/nio-20f") + return None + + def fake_get_main_text(_filing_dir: Path) -> str: + return "\n".join( + [ + "Item 3.D Risk Factors", + "Battery supply and geopolitical risks remain material.", + "Item 5. Operating and Financial Review and Prospects", + "Management discusses margin recovery and delivery outlook.", + "Item 15. Controls and Procedures", + "Disclosure controls were effective.", + "Item 18. Financial Statements", + "Consolidated financial statements follow.", + ] + ) + + monkeypatch.setattr(sec_parser, "_get_edgar_downloader", lambda: FakeDownloader) + monkeypatch.setattr(sec_parser, "find_downloaded_filing_path", fake_find_downloaded_filing_path) + monkeypatch.setattr(sec_parser, "get_main_10k_text", fake_get_main_text) + monkeypatch.setattr(sec_parser, "read_main_10k_html_raw", lambda _path: "20-F filing") + monkeypatch.setattr(sec_parser, "_save_10k_to_cache", lambda *_args, **_kwargs: None) + monkeypatch.setattr(sec_parser, "save_10k_html_slice", lambda *_args, **_kwargs: None) + monkeypatch.setattr(sec_parser, "prepare_native_html_fragment_from_10k_raw", lambda raw_html: raw_html) + monkeypatch.setattr( + sec_parser, + "_save_filing_meta_to_cache", + lambda _ticker, filing_form: saved_meta.update({"filing_form": filing_form}), + ) + + sections, filing_form = sec_parser.download_and_extract_all_items_with_form("NIO", "test@example.com") + + assert calls[:2] == ["10-K", "20-F"] + assert filing_form == "20-F" + assert saved_meta["filing_form"] == "20-F" + assert "Battery supply" in sections["item1a"] + assert "margin recovery" in sections["item7"] + assert "financial statements" in sections["item8"].lower()