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, statistics, sys10from 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)21for con in (edgar, spid):22    con.execute("SET threads=4"); con.execute("SET memory_limit='16GB'")23# the src_* views stored in spid.duckdb point to the master node's parquet corpus (valid here)24S = {}252627def dump(name, con, sql):28    cur = con.execute(sql); cols = [d[0] for d in cur.description]; rows = cur.fetchall()29    with open(out / f"{name}.csv", "w", newline="") as f:30        w = csv.writer(f); w.writerow(cols); w.writerows(rows)31    print(f"{name}: {len(rows)} rows"); return rows323334def one(con, sql):35    return con.execute(sql).fetchone()[0]363738def grep_first(path, pattern, cast=str):39    """First regex group found in a log file, or None when the file or the pattern is absent."""40    p = root / path41    if not p.exists(): return None42    m = re.search(pattern, p.read_text(errors="replace"))43    return cast(m.group(1)) if m else None444546def grep_count(path, pattern):47    p = root / path48    if not p.exists(): return None49    return len(re.findall(pattern, p.read_text(errors="replace")))505152# ---- universe reconciliation -------------------------------------------------53uni = json.load(open(root / "data/metadata/universe.json"))54ru = json.load(open(root / "data/metadata/russell3000_constituents.json"))55sp = json.load(open(root / "data/metadata/sp500_constituents.json"))56S["universe"] = {"built_at": uni["built_at"], "n": uni["count"], "n_sp500": uni["n_sp500"], "n_russell": uni["n_russell3000"],57                 "russell_holdings_as_of": ru["holdings_as_of"], "russell_fetched_at": ru["fetched_at"],58                 "russell_equity_lines_mapped": ru["count"], "russell_unmatched": len(ru["unmatched"]),59                 "russell_unmatched_list": ru["unmatched"], "sp500_fetched_at": sp["fetched_at"], "sp500_rows": sp["count"],60                 "russell_distinct_ciks": len({c["cik"] for c in ru["constituents"]}),61                 "both": sum(1 for c in uni["constituents"] if c["in_sp500"] and c["in_russell3000"]),62                 "sp_only": sum(1 for c in uni["constituents"] if c["in_sp500"] and not c["in_russell3000"]),63                 "ru_only": sum(1 for c in uni["constituents"] if not c["in_sp500"] and c["in_russell3000"])}64S["sp500_ciks"] = len({c["cik"] for c in sp["constituents"]}) if "constituents" in sp else None65# the iShares holdings file is not archived with the repository: the equity-line count is not reconstructible here66holdings = sorted((root / "data/metadata").glob("*holding*"))67S["iwv_equity_lines"] = None if not holdings else sum(1 for _ in open(holdings[0]))  # replaced by a parse if the file is ever archived68S["iwv_holdings_file_archived"] = bool(holdings)69S["issuers_with_filings"] = one(edgar, "select count(distinct cik) from filings")70S["issuers_without_filings"] = one(edgar, "select count(*) from companies c where not exists (select 1 from filings f where f.cik=c.cik)")71dump("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")72S["issuers_with_projects"] = one(spid, "select count(distinct cik) from projects")7374# ---- corpus -------------------------------------------------------------------75dump("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")76dump("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")77dump("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")78dump("parse_status", edgar, "select form_type, parse_status, count(*) n from filings group by 1,2 order by 1,2")79dump("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")80dump("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")81S["chunks"] = edgar.execute("select count(*), round(avg(word_count),1) from filing_chunks").fetchone()82S["embeddings"] = one(edgar, "select count(*) from embeddings_metadata")83S["sections_total"] = one(edgar, "select count(*) from filing_sections")84S["filings_total"] = one(edgar, "select count(*) from filings")85S["raw_gb"] = one(edgar, "select round(sum(file_size_bytes)/1e9,1) from filings")86S["filing_date_range"] = [str(x) for x in edgar.execute("select min(filing_date), max(filing_date) from filings").fetchone()]87S["annual_panel_firms"] = one(edgar, "select count(distinct cik) from annual_panel")88dump("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")89dump("companies_by_shard", edgar, "select shard_node, count(*) n from companies group by 1 order by 1")90dump("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")9192# ---- extraction ledger --------------------------------------------------------93dump("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")94S["ledger_total"] = one(spid, "select count(*) from extracted_sections")95dump("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")96dump("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")97S["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'")98S["ledger_backends"] = [r[0] for r in spid.execute("select distinct backend from extracted_sections order by 1").fetchall()]99# the two runs are separated by the backfilled block: every June row was re-inserted at one timestamp (min(extracted_at))100t0, t1 = spid.execute("select min(extracted_at), max(extracted_at) from extracted_sections").fetchone()101S["extracted_at_range"] = [str(t0), str(t1)]102S["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()103S["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")104S["october_rows"] = S["ledger_total"] - S["backfilled_rows"]105# outage = the largest gap between consecutive distinct ledger timestamps of the October run (no window is typed here)106gap = spid.execute("""with t as (select distinct extracted_at ts from extracted_sections where extracted_at > (select min(extracted_at) from extracted_sections)),107  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()108S["outage"] = {"start": str(gap[0]), "end": str(gap[1]), "hours": round(gap[2], 2)} if gap else None109wall_h = (t1 - t0).total_seconds() / 3600.0110S["october_wall_clock_hours"] = round(wall_h, 2)111S["october_effective_sections_per_second"] = round(S["october_rows"] / ((wall_h - (gap[2] if gap else 0.0)) * 3600.0), 1)112S["conf_exact_default_count"] = one(spid, "select count(*) from project_mentions where round(confidence,3) = 0.600")113S["conf_min_by_run"] = spid.execute("""select min(m.confidence) filter (where e.extracted_at = (select min(extracted_at) from extracted_sections)),114  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()115S["amount_raw_deviations"] = one(spid, "select count(*) from project_mentions where amount_usd is not null and amount_raw <> '$' || format('{:,.0f}', amount_usd)")116dump("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) overwritten117  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""")118dump("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,119  v.form_type, count(*) sections, count(*) filter (where e.n_mentions>0) with_mentions, sum(e.n_mentions) mentions, count(distinct v.cik) issuers120  from extracted_sections e join v_sections v using(section_id) group by 1,2 order by 1,2""")121# run parameters are read from the logs (regex); None when the log does not carry them122S["spid_runs"] = {123    "sp500_run": {"log": "project-db/logs/extract.log", "date": grep_first("project-db/logs/extract.log", r"(\d{4}-\d{2}-\d{2})"),124                  "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),125                  "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),126                  "sections_logged": grep_first("project-db/logs/extract.log", r"sections=(\d+)", int), "ledger_rows_backfilled": S["backfilled_rows"],127                  "mentions_backfilled": S["backfilled_mentions"], "llm_call_failed": grep_count("project-db/logs/extract.log", r"LLM call failed")},128    "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})"),129                    "workers": grep_first("data/logs/spid_extract.log", r"workers=(\d+)", int), "ledger_rows": S["october_rows"],130                    "extracted_at_first": str(t0), "extracted_at_last": str(t1), "llm_call_failed": grep_count("data/logs/spid_extract.log", r"LLM call failed"),131                    "section_failed": grep_count("data/logs/spid_extract.log", r"section failed")}}132if not SKIP_SLOW:133    # true call denominator: sections whose text passes the candidate gate (llm.py returns [] before any call otherwise)134    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*"135    cb = dump("calls_by_form", spid, f"""select v.form_type, count(*) ledgered, count(*) filter (where regexp_matches(v.text, '{gate}', 'i')) gate_pass,136      count(*) filter (where regexp_matches(v.text, '{gate}', 'i') and e.n_mentions>0) productive137      from extracted_sections e join v_sections v using(section_id) group by 1 order by 1""")138    S["gate_pass_total"] = sum(r[2] for r in cb)139    S["calls_by_form"] = {r[0]: r[2] for r in cb}140else:141    S["gate_pass_total"] = None; S["calls_by_form"] = None142143# ---- mentions / projects ------------------------------------------------------144S["mentions"] = one(spid, "select count(*) from project_mentions")145S["projects"] = one(spid, "select count(*) from projects")146dump("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")147dump("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")148dump("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")149dump("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")150dump("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")151dump("projects_by_first_seen_year", spid, "select extract(year from first_seen)::int yr, count(*) n from projects group by 1 order by 1")152dump("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)")153dump("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)")154dump("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")155dump("confidence_dist", spid, "select round(confidence,1) conf, count(*) n from project_mentions group by 1 order by 1")156dump("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")157S["amount_total_median_musd"] = one(spid, "select round(median(total_amount_usd)/1e6,1) from projects where total_amount_usd is not null")158S["amount_max"] = one(spid, "select max(total_amount_usd) from projects")159S["amount_ge_50b"] = one(spid, "select count(*) from projects where total_amount_usd >= 50e9")160S["amount_ge_1t"] = one(spid, "select count(*) from projects where total_amount_usd >= 1e12")161S["amount_le_zero"] = one(spid, "select count(*) from projects where total_amount_usd <= 0")162dump("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")163dump("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")164dump("top_issuers", spid, "select ticker, company_name, sector, count(*) projects from projects group by 1,2,3 order by 4 desc limit 15")165dump("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")166S["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()167S["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()168S["fiscal_year_null_mentions"] = one(spid, "select count(*) from project_mentions where fiscal_year is null")169170# ---- bootstrap CI of the per-type median amount (percentile bootstrap, seeded) ----171if not SKIP_SLOW:172    rng = random.Random(SEED)173    with open(out / "projects_by_type_ci.csv", "w", newline="") as f:174        w = csv.writer(f); w.writerow(["project_type", "n", "median", "ci_low", "ci_high", "B", "seed"])175        for ptype, in spid.execute("select project_type from projects group by 1 order by count(*) desc").fetchall():176            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()]177            if not vals: w.writerow([ptype, 0, None, None, None, BOOT_B, SEED]); continue178            meds = sorted(statistics.median(rng.choices(vals, k=len(vals))) for _ in range(BOOT_B))179            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])180    print("projects_by_type_ci: written")181182json.dump(S, open(out / "stats.json", "w"), indent=2, default=str)183print(json.dumps(S, indent=1, default=str)[:3000])184