import sys from datetime import datetime, timezone from dotenv import load_dotenv load_dotenv() from ingestion.edgar import fetch_all_edgar_data, _extract_guidance from ingestion.transcript import fetch_transcript from ingestion.embedder import clear_ticker_data, embed_and_store_filing, embed_and_store_transcript from ingestion.guidance_parser import parse_guidance from ingestion.yf_fallback import fill_missing_metrics from storage.metrics_db import init_db, upsert_metrics, prune_old_metrics, get_all_metrics from storage.sections_db import init_sections_db, upsert_section, get_section N_ANNUAL = 3 N_QUARTERLY = 12 def _period_to_av_quarter(period: str) -> str: """Convert EDGAR period string to Alpha Vantage quarter format. EDGAR: "Q42024", "FY2024", "Q12025" AV: "2024Q4", "2024Q4", "2025Q1" """ period = period.strip() if not period or len(period) < 6: return "" if period.startswith("FY"): fy = period[2:] if len(fy) == 4 and fy.isdigit(): return f"{fy}Q4" return "" if period[0] == "Q" and period[1] in "1234": fp = period[:2] fy = period[2:] if len(fy) == 4 and fy.isdigit(): return f"{fy}{fp}" return "" return "" def ingest(ticker: str, full: bool = False) -> None: print(f"[ingest] Starting {'FULL' if full else 'delta'} ingestion for {ticker.upper()}") init_db() init_sections_db() print(f"[ingest] Fetching EDGAR data (last {N_ANNUAL} annual + {N_QUARTERLY} quarterly)...") filings = fetch_all_edgar_data(ticker, n_annual=N_ANNUAL, n_quarterly=N_QUARTERLY) if not filings: print(f"[ingest] ERROR: CIK not found for '{ticker}'. Check the ticker is a valid S&P 500 symbol.") sys.exit(1) print(f"[ingest] Found {len(filings)} filing(s) for {filings[0].company_name}") if full: print("[ingest] --full: clearing existing Chroma data for this ticker...") clear_ticker_data(ticker.upper()) # Periods already fully resolved (metrics stored + transcript resolved). # sections.db persists across runs and is the source of truth for transcript resolution: # None → never attempted (must fetch from AV) # "" → attempted, none available (skip AV, re-embed not needed) # text → transcript present (reuse, no AV call) existing_metric_periods: set[str] = { m["period"] for m in get_all_metrics(ticker.upper()) } processed = skipped = 0 for edgar in filings: label = f"{edgar.form_type} {edgar.period} ({edgar.filing_date})" # --- Delta check --- if not full: transcript_resolved = get_section(edgar.ticker, edgar.period, "transcript") profile_sections_resolved = ( edgar.form_type != "10-K" or ( get_section(edgar.ticker, edgar.period, "business") is not None and get_section(edgar.ticker, edgar.period, "segments_geography") is not None ) ) if ( edgar.period in existing_metric_periods and transcript_resolved is not None and profile_sections_resolved ): print(f"[ingest] SKIP {label} (already ingested)") skipped += 1 continue processed += 1 edgar = fill_missing_metrics(edgar) print(f"[ingest] Processing {label}...") guidance = _extract_guidance(edgar.mda_text) # Fetch transcript for both 10-Q and 10-K (annual Q4 call). # In delta mode: reuse the cached transcript from sections.db when available, # so no AV API call is made for already-resolved periods. transcript = "" if edgar.form_type in ("10-Q", "10-K"): quarter = _period_to_av_quarter(edgar.period) if not quarter: print(f"[ingest] WARNING: Cannot derive quarter from period '{edgar.period}'. Skipping transcript.") else: stored = None if full else get_section(edgar.ticker, edgar.period, "transcript") if stored is not None: # Reuse cached value (may be "" if previously unavailable) — 0 AV calls. transcript = stored if transcript: print(f"[ingest] Transcript reused from cache ({len(transcript):,} chars)") else: # First time or forced full — fetch from Alpha Vantage. transcript = fetch_transcript(edgar.ticker, quarter) or "" if not transcript: print(f"[ingest] WARNING: No transcript found for {quarter}.") guidance_struct = parse_guidance(edgar.mda_text, transcript or None, edgar.period) if guidance_struct.get("guidance_period") or guidance_struct.get("guidance_revenue_low") is not None: print(f"[ingest] Parsed guidance for {guidance_struct.get('guidance_period')}: " f"rev=[{guidance_struct.get('guidance_revenue_low')},{guidance_struct.get('guidance_revenue_high')}] " f"eps=[{guidance_struct.get('guidance_eps_low')},{guidance_struct.get('guidance_eps_high')}]") upsert_metrics({ "ticker": edgar.ticker, "period": edgar.period, "company_name": edgar.company_name, "filing_date": edgar.filing_date, "form_type": edgar.form_type, "revenue": edgar.revenue, "revenue_yoy_pct": edgar.revenue_yoy_pct, "eps": edgar.eps, "gross_margin": edgar.gross_margin, "operating_margin": edgar.operating_margin, "free_cash_flow": edgar.free_cash_flow, "guidance_text": guidance, "guidance_disclosed": 1 if guidance else 0, "ingested_at": datetime.now(timezone.utc).isoformat(), "shares_diluted": edgar.shares_diluted, "effective_tax_rate": edgar.effective_tax_rate, "interest_expense": edgar.interest_expense, "total_debt": edgar.total_debt, "dividends_paid": edgar.dividends_paid, "buybacks": edgar.buybacks, "capex": edgar.capex, "stockholders_equity": edgar.stockholders_equity, "period_basis": edgar.period_basis, "report_date": edgar.report_date, "accession": edgar.accession, "source_url": edgar.source_url, "metric_contexts": edgar.metric_contexts, "quality_warnings": edgar.quality_warnings, "data_quality_status": edgar.data_quality_status, **guidance_struct, }) # Persist raw section text for cross-period text diffing (analysis/textdiff.py) if edgar.mda_text: upsert_section(edgar.ticker, edgar.period, edgar.form_type, "mda", edgar.mda_text) if edgar.risk_factors_text: upsert_section(edgar.ticker, edgar.period, edgar.form_type, "risk_factors", edgar.risk_factors_text) if edgar.form_type == "10-K": # Store empty results too: this records that the profile sections # were attempted and prevents endless delta backfills. upsert_section(edgar.ticker, edgar.period, edgar.form_type, "business", edgar.business_text) upsert_section( edgar.ticker, edgar.period, edgar.form_type, "segments_geography", edgar.segments_geography_text, ) embed_and_store_filing( ticker=edgar.ticker, company_name=edgar.company_name, mda_text=edgar.mda_text, risk_text=edgar.risk_factors_text, filing_date=edgar.filing_date, period=edgar.period, form_type=edgar.form_type, accession=edgar.accession, source_url=edgar.source_url, business_text=edgar.business_text, segments_geography_text=edgar.segments_geography_text, ) # Always mark transcript resolution — even empty — so future delta runs skip the AV call. # embed_and_store_transcript is only called when text is non-empty. if edgar.form_type in ("10-Q", "10-K"): upsert_section(edgar.ticker, edgar.period, edgar.form_type, "transcript", transcript) if transcript: embed_and_store_transcript( ticker=edgar.ticker, company_name=edgar.company_name, transcript_text=transcript, transcript_date=edgar.filing_date, period=edgar.period, source_url="", provider="alphavantage", ) print(f"[ingest] Transcript stored ({len(transcript):,} chars)") # Prune SQLite to configured limits prune_old_metrics(ticker.upper(), "10-K", N_ANNUAL) prune_old_metrics(ticker.upper(), "10-Q", N_QUARTERLY) summary = f"{processed} processed, {skipped} skipped (already up-to-date)" print(f"[ingest] Done. {filings[0].company_name} ({ticker.upper()}) — {summary}.") if __name__ == "__main__": args = sys.argv[1:] full_flag = "--full" in args tickers = [a for a in args if not a.startswith("--")] if len(tickers) != 1: print("Usage: python ingest.py [--full]") print(" --full Wipe and re-ingest all periods (fetches all AV transcripts again)") sys.exit(1) ingest(tickers[0], full=full_flag)