spb/firc0001_uqo

1"""Export every number used in the working paper from the two DuckDB databases (run on the master node).23Usage (on M3U96b):  nice -n 10 env OMP_NUM_THREADS=4 ~/UQO_RESEARCH_DB/.venv/bin/python -I export_stats.py ~/UQO_RESEARCH_DB <out_dir> [--skip-slow]4Writes CSV files (one per table) + stats.json. Nothing is hand-edited downstream, and no number is typed in this file:5every value is a query result, a log/regex read (None when absent) or an explicit None flagged in the output.6Both connections are opened read_only=True; threads=4; nothing is written under the repository.7--skip-slow omits the candidate-gate scan over the whole ledger (gate_pass_total / calls_by_form) and the bootstrap.8"""9import csv, json, random, re, shutil, statistics, sys, tempfile10from datetime import timedelta11from pathlib import Path12import duckdb1314SEED = 2026110515BOOT_B = 100016args = [a for a in sys.argv[1:] if not a.startswith("--")]17SKIP_SLOW = "--skip-slow" in sys.argv18root = Path(args[0]); out = Path(args[1]); out.mkdir(parents=True, exist_ok=True)19edgar = duckdb.connect(str(root / "data/duckdb/edgar_research.duckdb"), read_only=True)20spid = duckdb.connect(str(root / "project-db/data/db/spid.duckdb"), read_only=True)21TMP = tempfile.mkdtemp(prefix="spid_export_stats_")   # DuckDB spill directory: never next to the databases22for con in (edgar, spid):23    con.execute("SET threads=4"); con.execute("SET memory_limit='16GB'"); con.execute(f"SET temp_directory='{TMP}'")24# the src_* views stored in spid.duckdb point to the master node's parquet corpus (valid here)25S = {}262728def dump(name, con, sql):29    cur = con.execute(sql); cols = [d[0] for d in cur.description]; rows = cur.fetchall()30    with open(out / f"{name}.csv", "w", newline="") as f:31        w = csv.writer(f); w.writerow(cols); w.writerows(rows)32    print(f"{name}: {len(rows)} rows"); return rows333435def one(con, sql):36    return con.execute(sql).fetchone()[0]373839def grep_first(path, pattern, cast=str):40    """First regex group found in a log file, or None when the file or the pattern is absent."""41    p = root / path42    if not p.exists(): return None43    m = re.search(pattern, p.read_text(errors="replace"))44    return cast(m.group(1)) if m else None454647def grep_max(path, pattern):48    """Largest integer captured by the regex group over the whole log (cumulative counters), or None."""49    p = root / path50    if not p.exists(): return None51    vals = [int(m.group(1)) for m in re.finditer(pattern, p.read_text(errors="replace"))]52    return max(vals) if vals else None535455def grep_count(path, pattern):56    p = root / path57    if not p.exists(): return None58    return len(re.findall(pattern, p.read_text(errors="replace")))596061# ---- universe reconciliation -------------------------------------------------62uni = json.load(open(root / "data/metadata/universe.json"))63ru = json.load(open(root / "data/metadata/russell3000_constituents.json"))64sp = json.load(open(root / "data/metadata/sp500_constituents.json"))65S["universe"] = {"built_at": uni["built_at"], "n": uni["count"], "n_sp500": uni["n_sp500"], "n_russell": uni["n_russell3000"],66                 "russell_holdings_as_of": ru["holdings_as_of"], "russell_fetched_at": ru["fetched_at"],67                 "russell_equity_lines_mapped": ru["count"], "russell_unmatched": len(ru["unmatched"]),68                 "russell_unmatched_list": ru["unmatched"], "sp500_fetched_at": sp["fetched_at"], "sp500_rows": sp["count"],69                 "russell_distinct_ciks": len({c["cik"] for c in ru["constituents"]}),70                 "both": sum(1 for c in uni["constituents"] if c["in_sp500"] and c["in_russell3000"]),71                 "sp_only": sum(1 for c in uni["constituents"] if c["in_sp500"] and not c["in_russell3000"]),72                 "ru_only": sum(1 for c in uni["constituents"] if not c["in_sp500"] and c["in_russell3000"])}73S["sp500_ciks"] = len({c["cik"] for c in sp["constituents"]}) if "constituents" in sp else None74# the iShares holdings file is not archived with the repository: the equity-line count is not reconstructible here75holdings = sorted((root / "data/metadata").glob("*holding*"))76S["iwv_equity_lines"] = None if not holdings else sum(1 for _ in open(holdings[0]))  # replaced by a parse if the file is ever archived77S["iwv_holdings_file_archived"] = bool(holdings)78S["issuers_with_filings"] = one(edgar, "select count(distinct cik) from filings")79S["issuers_without_filings"] = one(edgar, "select count(*) from companies c where not exists (select 1 from filings f where f.cik=c.cik)")80dump("issuers_without_filings", edgar, "select ticker, company_name, sector, in_sp500, in_russell3000 from companies c where not exists (select 1 from filings f where f.cik=c.cik) order by ticker")81S["issuers_with_projects"] = one(spid, "select count(distinct cik) from projects")8283# ---- corpus -------------------------------------------------------------------84dump("filings_by_form", edgar, "select form_type, count(*) n, min(filing_date) first, max(filing_date) last, round(sum(file_size_bytes)/1e9,1) gb from filings group by 1 order by 1")85dump("filings_by_year_form", edgar, "select extract(year from filing_date)::int yr, form_type, count(*) n, count(distinct cik) issuers from filings group by 1,2 order by 1,2")86dump("issuers_by_year", edgar, "select extract(year from filing_date)::int yr, count(distinct cik) issuers, count(*) filings from filings group by 1 order by 1")87dump("parse_status", edgar, "select form_type, parse_status, count(*) n from filings group by 1,2 order by 1,2")88dump("sections_by_form", edgar, "select f.form_type, count(*) sections, round(avg(s.word_count)) mean_words, round(median(s.word_count)) median_words from filing_sections s join filings f using(accession_number) group by 1 order by 1")89dump("sections_10k_by_item", edgar, "select s.section_name, count(*) n, round(median(s.word_count)) median_words from filing_sections s join filings f using(accession_number) where f.form_type='10-K' group by 1 order by 2 desc")90S["chunks"] = edgar.execute("select count(*), round(avg(word_count),1) from filing_chunks").fetchone()91S["embeddings"] = one(edgar, "select count(*) from embeddings_metadata")92S["sections_total"] = one(edgar, "select count(*) from filing_sections")93S["filings_total"] = one(edgar, "select count(*) from filings")94S["raw_gb"] = one(edgar, "select round(sum(file_size_bytes)/1e9,1) from filings")95S["filing_date_range"] = [str(x) for x in edgar.execute("select min(filing_date), max(filing_date) from filings").fetchone()]96S["annual_panel_firms"] = one(edgar, "select count(distinct cik) from annual_panel")97dump("companies_by_sector", edgar, "select sector, count(*) n, count(*) filter (where in_sp500) sp500, count(*) filter (where in_russell3000) russell from companies group by 1 order by 2 desc")98dump("companies_by_shard", edgar, "select shard_node, count(*) n from companies group by 1 order by 1")99dump("factor_tables", edgar, "select 'chunk_scores' t, count(*) n from chunk_scores union all select 'filing_scores', count(*) from filing_scores union all select 'annual_panel', count(*) from annual_panel union all select 'quarterly_panel', count(*) from quarterly_panel")100101# ---- extraction ledger --------------------------------------------------------102dump("ledger_by_form", spid, "select v.form_type, count(*) sections, count(*) filter (where e.n_mentions>0) with_mentions, sum(e.n_mentions) mentions, count(distinct v.cik) issuers from extracted_sections e join v_sections v using(section_id) group by 1 order by 1")103S["ledger_total"] = one(spid, "select count(*) from extracted_sections")104dump("ledger_10k_items", spid, "select v.section_name, count(*) n, sum(e.n_mentions) mentions from extracted_sections e join v_sections v using(section_id) where v.form_type='10-K' group by 1 order by 2 desc")105dump("ledger_10q_items", spid, "select v.section_name, count(*) n, sum(e.n_mentions) mentions from extracted_sections e join v_sections v using(section_id) where v.form_type='10-Q' group by 1 order by 2 desc")106S["ledger_10k_mdna_median_words"] = one(spid, "select round(median(v.word_count)) from extracted_sections e join v_sections v using(section_id) where v.form_type='10-K' and v.section_name='md_and_a'")107S["ledger_backends"] = [r[0] for r in spid.execute("select distinct backend from extracted_sections order by 1").fetchall()]108# the two runs are separated by the backfilled block: every June row was re-inserted at one timestamp (min(extracted_at))109t0, t1 = spid.execute("select min(extracted_at), max(extracted_at) from extracted_sections").fetchone()110S["extracted_at_range"] = [str(t0), str(t1)]111S["backfilled_rows"], S["backfilled_mentions"] = spid.execute("select count(*), sum(n_mentions) from extracted_sections where extracted_at = (select min(extracted_at) from extracted_sections)").fetchone()112S["backfilled_rows_under_30_words"] = one(spid, "select count(*) from extracted_sections e join v_sections v using(section_id) where e.extracted_at = (select min(extracted_at) from extracted_sections) and v.word_count < 30")113S["october_rows"] = S["ledger_total"] - S["backfilled_rows"]114# outage = the largest gap between consecutive distinct ledger timestamps of the October run (no window is typed here)115gap = spid.execute("""with t as (select distinct extracted_at ts from extracted_sections where extracted_at > (select min(extracted_at) from extracted_sections)),116  g as (select ts, lag(ts) over (order by ts) prev from t) select prev, ts, epoch(ts - prev)/3600.0 h from g where prev is not null order by h desc limit 1""").fetchone()117S["outage"] = {"start": str(gap[0]), "end": str(gap[1]), "hours": round(gap[2], 2)} if gap else None118# the October run starts at its first flush, i.e. the first timestamp after the backfilled block (min(extracted_at) is the backfill instant)119t_oct0 = one(spid, "select min(extracted_at) from extracted_sections where extracted_at > (select min(extracted_at) from extracted_sections)")120S["october_first_flush"] = str(t_oct0)121wall_h = (t1 - t_oct0).total_seconds() / 3600.0122S["october_wall_clock_hours"] = round(wall_h, 2)123S["october_effective_sections_per_second"] = round(S["october_rows"] / ((wall_h - (gap[2] if gap else 0.0)) * 3600.0), 1)124S["conf_exact_default_count"] = one(spid, "select count(*) from project_mentions where round(confidence,3) = 0.600")125S["conf_min_by_run"] = spid.execute("""select min(m.confidence) filter (where e.extracted_at = (select min(extracted_at) from extracted_sections)),126  min(m.confidence) filter (where e.extracted_at > (select min(extracted_at) from extracted_sections)) from project_mentions m join extracted_sections e using(section_id)""").fetchone()127# NULL-aware: rows with amount_usd = 0.0 carry amount_raw NULL (llm.py guards the f-string with `if amount`), so `<>` would be NULL-blind128S["amount_raw_deviations"] = one(spid, "select count(*) from project_mentions where amount_usd is not null and amount_raw is distinct from '$' || format('{:,.0f}', amount_usd)")129dump("ledger_collisions", spid, """select e.section_id, e.accession_number, e.n_mentions ledger_n, count(m.mention_id) stored_n, e.n_mentions - count(m.mention_id) overwritten130  from extracted_sections e left join project_mentions m using(section_id) group by 1,2,3 having e.n_mentions > count(m.mention_id) order by 5 desc, 1""")131dump("cohort_by_run", spid, """select case when e.extracted_at = (select min(extracted_at) from extracted_sections) then 'backfilled (June run)' else 'October run' end run,132  v.form_type, count(*) sections, count(*) filter (where e.n_mentions>0) with_mentions, sum(e.n_mentions) mentions, count(distinct v.cik) issuers133  from extracted_sections e join v_sections v using(section_id) group by 1,2 order by 1,2""")134# run parameters are read from the logs (regex); None when the log does not carry them135S["spid_runs"] = {136    "sp500_run": {"log": "project-db/logs/extract.log", "date": grep_first("project-db/logs/extract.log", r"(\d{4}-\d{2}-\d{2})"),137                  "model": grep_first("project-db/logs/extract.log", r"(gpt-4o-mini[\w-]*)"), "workers": grep_first("project-db/logs/extract.log", r"workers=(\d+)", int),138                  "batch": grep_first("project-db/logs/extract.log", r"batch=(\d+)", int), "ciks": grep_first("project-db/logs/extract.log", r"CIKs total=(\d+)", int),139                  "sections_logged": grep_max("project-db/logs/extract.log", r"cum_sections=(\d+)"), "ledger_rows_backfilled": S["backfilled_rows"],140                  "mentions_backfilled": S["backfilled_mentions"], "llm_call_failed": grep_count("project-db/logs/extract.log", r"LLM call failed")},141    "russell_run": {"log": "data/logs/spid_extract.log", "model": grep_first("data/logs/spid_extract.log", r"(gpt-4o-mini-\d{4}-\d{2}-\d{2})"),142                    "workers": grep_first("data/logs/spid_extract.log", r"workers=(\d+)", int), "ledger_rows": S["october_rows"],143                    "extracted_at_first": str(t0), "extracted_at_last": str(t1), "llm_call_failed": grep_count("data/logs/spid_extract.log", r"LLM call failed"),144                    "section_failed": grep_count("data/logs/spid_extract.log", r"section failed")}}145if not SKIP_SLOW:146    # true call denominator: sections whose text passes the candidate gate (llm.py returns [] before any call otherwise)147    gate = r"\b(project|initiative|program|invest|expan|construct|build|launch|develop|deploy|implement|acqui|partner|facilit|capacity|capital|migrat|transform|pipeline|plant|data ?cent|automat|sustainab|renewable|restructur)\w*"148    cb = dump("calls_by_form", spid, f"""select v.form_type, count(*) ledgered, count(*) filter (where regexp_matches(v.text, '{gate}', 'i')) gate_pass,149      count(*) filter (where regexp_matches(v.text, '{gate}', 'i') and e.n_mentions>0) productive150      from extracted_sections e join v_sections v using(section_id) group by 1 order by 1""")151    S["gate_pass_total"] = sum(r[2] for r in cb)152    S["calls_by_form"] = {r[0]: r[2] for r in cb}153else:154    S["gate_pass_total"] = None; S["calls_by_form"] = None155156# ---- mentions / projects ------------------------------------------------------157S["mentions"] = one(spid, "select count(*) from project_mentions")158S["projects"] = one(spid, "select count(*) from projects")159dump("mentions_by_form", spid, "select form_type, count(*) n, round(avg(confidence),3) mean_conf from project_mentions group by 1 order by 2 desc")160dump("mentions_by_year", spid, "select extract(year from filing_date)::int yr, count(*) mentions, count(distinct cik) issuers from project_mentions group by 1 order by 1")161dump("projects_by_type", spid, "select project_type, count(*) n, round(100.0*count(*)/(select count(*) from projects),1) pct, count(*) filter (where total_amount_usd is not null) with_amount, round(median(total_amount_usd)/1e6,1) median_amount_musd, round(avg(n_mentions),2) mean_mentions from projects group by 1 order by 2 desc")162dump("projects_by_sector", spid, "select coalesce(sector,'(unknown)') sector, count(*) n, count(distinct cik) issuers, round(count(*)*1.0/count(distinct cik),1) per_issuer from projects group by 1 order by 2 desc")163dump("projects_by_status", spid, "select status, count(*) n, round(100.0*count(*)/(select count(*) from projects),1) pct from projects group by 1 order by 2 desc")164dump("projects_by_first_seen_year", spid, "select extract(year from first_seen)::int yr, count(*) n from projects group by 1 order by 1")165dump("mentions_per_project", spid, "select case when n_mentions=1 then '1' when n_mentions=2 then '2' when n_mentions<=5 then '3-5' when n_mentions<=10 then '6-10' else '>10' end bucket, count(*) n from projects group by 1 order by min(n_mentions)")166dump("projects_per_issuer", spid, "select quantile_cont(n, [0.1,0.25,0.5,0.75,0.9,0.99]) q, min(n) mn, max(n) mx, avg(n) mean from (select cik, count(*) n from projects group by 1)")167dump("projects_by_index", spid, "select co.in_sp500, co.in_russell3000, count(*) projects, count(distinct p.cik) issuers from projects p join src_companies co on p.cik=co.cik group by 1,2 order by 1,2")168dump("confidence_dist", spid, "select round(confidence,1) conf, count(*) n from project_mentions group by 1 order by 1")169dump("amount_dist", spid, "select quantile_cont(total_amount_usd, [0.1,0.25,0.5,0.75,0.9,0.99]) q, count(*) n, round(sum(total_amount_usd)/1e9,1) sum_gusd from projects where total_amount_usd is not null")170S["amount_total_median_musd"] = one(spid, "select round(median(total_amount_usd)/1e6,1) from projects where total_amount_usd is not null")171S["amount_max"] = one(spid, "select max(total_amount_usd) from projects")172S["amount_ge_50b"] = one(spid, "select count(*) from projects where total_amount_usd >= 50e9")173S["amount_ge_1t"] = one(spid, "select count(*) from projects where total_amount_usd >= 1e12")174S["amount_le_zero"] = one(spid, "select count(*) from projects where total_amount_usd <= 0")175dump("duration", spid, "select case when last_year is null or first_year is null then 'unknown' when last_year=first_year then '0' when last_year-first_year<=2 then '1-2' when last_year-first_year<=5 then '3-5' else '>5' end span_years, count(*) n from projects group by 1 order by 1")176dump("kg", spid, "select 'nodes' k, node_type, count(*) n from kg_nodes group by 1,2 union all select 'edges', rel, count(*) from kg_edges group by 1,2 order by 1,3 desc")177dump("top_issuers", spid, "select ticker, company_name, sector, count(*) projects from projects group by 1,2,3 order by 4 desc, 1 limit 15")178dump("example_provenance", spid, "select p.project_id, p.ticker, p.project_name, p.project_type, p.status, p.total_amount_usd, p.n_mentions, p.first_seen, p.last_seen, t.accession_number, t.filing_date, t.form_type, t.status m_status, t.amount_usd, t.confidence, t.snippet from projects p join project_timeline t using(project_id) where p.n_mentions between 4 and 8 and p.total_amount_usd > 5e8 and p.project_type in ('plant_construction','data_center') order by p.n_filings desc, p.project_id, t.filing_date limit 24")179S["mentions_fields"] = spid.execute("select count(*) filter (where amount_usd is not null) amt, count(*) filter (where len(locations)>0) loc, count(*) filter (where len(technologies)>0) tech, count(*) filter (where len(partners)>0) partner, count(*) filter (where len(years)>0) yrs, count(*) from project_mentions").fetchone()180S["projects_fields"] = spid.execute("select count(*) filter (where total_amount_usd is not null), count(*) filter (where canonical_location is not null), count(*) filter (where len(technologies)>0), count(*) from projects").fetchone()181S["fiscal_year_null_mentions"] = one(spid, "select count(*) from project_mentions where fiscal_year is null")182183# ---- bootstrap CI of the per-type median amount (percentile bootstrap, seeded) ----184if not SKIP_SLOW:185    rng = random.Random(SEED)186    with open(out / "projects_by_type_ci.csv", "w", newline="") as f:187        w = csv.writer(f); w.writerow(["project_type", "n", "median", "ci_low", "ci_high", "B", "seed"])188        for ptype, in spid.execute("select project_type from projects group by 1 order by count(*) desc").fetchall():189            vals = [r[0] for r in spid.execute("select total_amount_usd from projects where total_amount_usd is not null and project_type = ?", [ptype]).fetchall()]190            if not vals: w.writerow([ptype, 0, None, None, None, BOOT_B, SEED]); continue191            meds = sorted(statistics.median(rng.choices(vals, k=len(vals))) for _ in range(BOOT_B))192            w.writerow([ptype, len(vals), statistics.median(vals), meds[int(0.025 * BOOT_B)], meds[min(int(0.975 * BOOT_B), BOOT_B - 1)], BOOT_B, SEED])193    print("projects_by_type_ci: written")194195json.dump(S, open(out / "stats.json", "w"), indent=2, default=str)196print(json.dumps(S, indent=1, default=str)[:3000])197shutil.rmtree(TMP, ignore_errors=True)198