ingest.py

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 <TICKER> [--full]")
        print("  --full  Wipe and re-ingest all periods (fetches all AV transcripts again)")
        sys.exit(1)
    ingest(tickers[0], full=full_flag)