"""N2 - field-level grounding audit of the 308,065 mentions against their section text (cluster A, C, J). For every mention: each year in years[] present as a 4-digit token; each location / partner string present after case-folding and light normalisation; the amount present under scale-aware renderings with a currency token within 40 characters; fallback upper bounds (years/technologies equal to the regex output on the model's own description); amount tail; type-signal checks. Grouped by form, item, confidence bin, run, segmenter, filing-date period; 1,000-replicate bootstrap clustered by section. Read-only; section text streamed once; mention-level flags kept on the node (parquet). """ from __future__ import annotations import json, re, sys, os, time, datetime as dt from pathlib import Path from collections import defaultdict import numpy as np import multiprocessing as mp sys.path.insert(0, str(Path(__file__).resolve().parent)) from common import * sys.path.insert(0, str(Path(__file__).resolve().parent / "spid")) import taxonomy as tx NREP = 1000 rng = np.random.default_rng(SEED) YEAR_TOK = re.compile(r"(? str: s = s.casefold().strip() s = ALIASES.get(s, s) s = PUNCT.sub(" ", s) return SPACE.sub(" ", s).strip() def amount_candidates(a: float) -> list[str]: """Numeric strings under which an amount may appear in the text (scale-aware, 1-3 significant digits).""" out = set() if a is None or a == 0: return [] a = abs(a) for scale in (1.0, 1e3, 1e6, 1e9, 1e12): v = a / scale if v < 0.05 or v >= 1e7: continue if scale == 1.0: if abs(v - round(v)) < 1e-6: out.add(f"{int(round(v)):,}"); out.add(str(int(round(v)))) continue for fmt in ("{:.0f}", "{:.1f}", "{:.2f}", "{:.3f}", "{:,.0f}", "{:,.1f}", "{:,.2f}"): s = fmt.format(v) if "." in s: s2 = s.rstrip("0").rstrip(".") out.add(s2) out.add(s) return sorted(out, key=len, reverse=True) def find_number(text: str, cands: list[str]): for c in cands: start = 0 while True: i = text.find(c, start) if i < 0: break j = i + len(c) before = text[i-1] if i > 0 else " " after = text[j] if j < len(text) else " " if not (before.isdigit() or before in ".,") and not (after.isdigit() or (after in ".," and j+1 < len(text) and text[j+1].isdigit())): return i, j start = j return None MENTIONS: dict[str, list[dict]] = {} def process_section(item): sid, text = item ms = MENTIONS.get(sid) if not ms: return [] low = text.casefold() lown = SPACE.sub(" ", PUNCT.sub(" ", low)) years_in_text = set(YEAR_TOK.findall(text)) out = [] for m in ms: f = {"mention_id": m["mention_id"]} yrs = m["years"] or [] f["n_years"] = len(yrs) f["years_found"] = sum(1 for y in yrs if str(y) in years_in_text) f["years_all_grounded"] = (f["years_found"] == len(yrs)) if yrs else None f["years_not_subset"] = (f["years_found"] < len(yrs)) if yrs else False locs = m["locations"] or [] f["n_locations"] = len(locs) f["locations_found"] = sum(1 for s in locs if norm(s) and (" " + norm(s) + " ") in (" " + lown + " ")) f["locations_all_grounded"] = (f["locations_found"] == len(locs)) if locs else None pars = m["partners"] or [] f["n_partners"] = len(pars) f["partners_found"] = sum(1 for s in pars if norm(s) and (" " + norm(s) + " ") in (" " + lown + " ")) f["partners_all_grounded"] = (f["partners_found"] == len(pars)) if pars else None techs = m["technologies"] or [] f["n_technologies"] = len(techs) f["technologies_found"] = sum(1 for s in techs if norm(s) and (" " + norm(s) + " ") in (" " + lown + " ")) f["technologies_all_grounded"] = (f["technologies_found"] == len(techs)) if techs else None a = m["amount_usd"] f["has_amount"] = a is not None f["amount_number_found"] = None; f["amount_currency_within_40"] = None; f["amount_nonusd_within_40"] = None if a is not None and a != 0: pos = find_number(text, amount_candidates(a)) f["amount_number_found"] = pos is not None if pos: i, j = pos; win = text[max(0, i-40): j+40] f["amount_currency_within_40"] = bool(CUR_NEAR.search(win)) f["amount_nonusd_within_40"] = bool(NONUSD.search(win)) else: f["amount_currency_within_40"] = False; f["amount_nonusd_within_40"] = False elif a == 0: f["amount_number_found"] = False; f["amount_currency_within_40"] = False; f["amount_nonusd_within_40"] = False desc = m["description"] or "" ey = tx.extract_years(desc); et = tx.extract_technologies(desc) f["years_eq_regex_desc"] = bool(yrs) and sorted(yrs) == ey f["years_regex_desc_nonempty"] = bool(ey) f["techs_eq_regex_desc"] = bool(techs) and techs == et f["techs_regex_desc_nonempty"] = bool(et) nd = ((m["project_name"] or "") + " " + desc).lower() sigs = tx.TYPE_SIGNALS.get(m["project_type"], []) f["type_signal_in_name_desc"] = any(rx.search(nd) for rx in sigs) f["any_type_signal_in_name_desc"] = any(rx.search(nd) for rxs in tx.TYPE_SIGNALS.values() for rx in rxs) f["edit09_ma_regex"] = bool(FIN_EDIT09.search(nd)) f["edit09_investment"] = "investment" in nd out.append(f) return out def main(): r = Run("grounding") con = connect(SPID_DB) tmin = con.execute("select min(extracted_at) from extracted_sections").fetchone()[0] print("loading mentions ...", flush=True) cur = con.execute(f""" SELECT m.mention_id, m.section_id, m.years, m.locations, m.partners, m.technologies, m.amount_usd, m.amount_raw, m.description, m.project_name, m.project_type, m.status, m.confidence, m.form_type, m.section_name, m.filing_date, m.cik, CASE WHEN e.extracted_at = TIMESTAMP '{tmin}' THEN 'backfilled_june' ELSE 'october' END AS run, CASE WHEN ss.section_id IS NOT NULL THEN 'legacy_ingest' ELSE 'parsed' END AS segmenter, co.in_sp500, co.state, co.state_of_incorporation FROM project_mentions m JOIN extracted_sections e USING (section_id) LEFT JOIN spid_sections ss ON ss.section_id = m.section_id LEFT JOIN src_companies co ON m.cik = co.cik""") cols = [d[0] for d in cur.description] meta = {} for row in cur.fetchall(): d = dict(zip(cols, row)) MENTIONS.setdefault(d["section_id"], []).append(d) meta[d["mention_id"]] = d n_m = len(meta); n_s = len(MENTIONS) print(f"{n_m} mentions over {n_s} sections", flush=True) con.execute("CREATE TEMP TABLE need AS SELECT DISTINCT section_id FROM project_mentions") q = """SELECT s.section_id, s.text FROM src_sections s JOIN need USING (section_id) UNION ALL SELECT s.section_id, s.text FROM spid_sections s JOIN need USING (section_id)""" cur = con.execute(q) flags = [] t0 = time.time(); done = 0; chars = 0 ctx = mp.get_context("fork") with ctx.Pool(4) as pool: def batches(): while True: rows = cur.fetchmany(2000) if not rows: break yield rows for rows in batches(): chars += sum(len(t) for _, t in rows) for res in pool.imap_unordered(process_section, rows, chunksize=50): flags.extend(res) done += len(rows) if done % 20000 < 2000: print(f" {done}/{n_s} sections, {len(flags)} mentions, {chars/1e9:.2f} GB, {time.time()-t0:.0f}s", flush=True) r.notes["sections_streamed"] = done; r.notes["chars_streamed"] = chars; r.notes["mentions_flagged"] = len(flags) r.notes["sections_with_mentions_not_found_in_text_sources"] = n_s - done # ---- assemble --------------------------------------------------------------------------------- import pyarrow as pa, pyarrow.parquet as pq recs = [] for f in flags: m = meta[f["mention_id"]] c = m["confidence"] cb = "0.45-0.6" if c < 0.6 else "0.6-0.7" if c < 0.7 else "0.7-0.8" if c < 0.8 else "0.8-0.9" if c < 0.9 else "0.9-1.0" recs.append({**f, "section_id": m["section_id"], "form": m["form_type"], "item": m["section_name"], "run": m["run"], "segmenter": m["segmenter"], "conf_bin": cb, "period": "pre_2023-10" if (m["filing_date"] and str(m["filing_date"]) < "2023-10-01") else "post_2023-10", "in_sp500": bool(m["in_sp500"]), "state": m["state"] or "", "state_inc": m["state_of_incorporation"] or "", "project_type": m["project_type"], "status": m["status"], "amount_usd": m["amount_usd"], "amount_raw": m["amount_raw"], "confidence": c, "cik": m["cik"], "filing_date": str(m["filing_date"])}) tbl = pa.Table.from_pylist(recs) pq.write_table(tbl, r.out / "mention_flags.parquet") # stays on the node (> 5 MB) con.register("fl", tbl) FIELDS = {"years": "years_all_grounded", "locations": "locations_all_grounded", "partners": "partners_all_grounded", "technologies": "technologies_all_grounded", "amount_number": "amount_number_found", "amount_with_currency": "amount_currency_within_40"} def boot(groupcol: str, fname: str, flag: str, where: str = "TRUE"): rows = con.execute(f"""SELECT {groupcol} g, section_id, count(*) n, count(*) FILTER (WHERE {flag}) k FROM fl WHERE {flag} IS NOT NULL AND {where} GROUP BY 1,2""").fetchall() bygrp = defaultdict(list) for g, sid, n, k in rows: bygrp[g].append((n, k)) out = [] for g, lst in bygrp.items(): n = np.array([x[0] for x in lst], float); k = np.array([x[1] for x in lst], float) share = k.sum()/n.sum() if len(lst) > 1: w = rng.poisson(1.0, size=(NREP, len(lst))) reps = (w @ k) / np.maximum(1e-9, (w @ n)) lo, hi = np.percentile(reps, [2.5, 97.5]) else: lo = hi = share out.append((fname, g, int(n.sum()), len(lst), int(k.sum()), round(share, 4), round(float(lo), 4), round(float(hi), 4))) return out cols_out = ["field", "group", "n_mentions", "n_sections", "n_grounded", "grounded_share", "ci_low", "ci_high"] allrows = [] for grouping in ["'all'", "form", "run", "segmenter", "conf_bin", "period", "in_sp500::varchar", "project_type", "status"]: rows = [] for fname, flag in FIELDS.items(): rows += boot(grouping, fname, flag) gname = grouping.replace("'", "").replace("::varchar", "") allrows += [(gname,) + x for x in rows] r.write_csv("grounding_by_group.csv", ["grouping"] + cols_out, allrows) rows = [] for fname, flag in FIELDS.items(): rows += [(x[1].split("|")[0], x[1].split("|")[1]) + (x[0],) + x[2:] for x in boot("form || '|' || item", fname, flag)] r.write_csv("grounding_by_field_form_item.csv", ["form", "item", "field", "n_mentions", "n_sections", "n_grounded", "grounded_share", "ci_low", "ci_high"], rows) # combined 'all structured fields grounded' per mention con.execute("""CREATE TEMP TABLE comb AS SELECT section_id, form, run, conf_bin, (coalesce(years_all_grounded, TRUE) AND coalesce(locations_all_grounded, TRUE) AND coalesce(partners_all_grounded, TRUE) AND coalesce(amount_number_found, TRUE)) AS all_grounded, (n_years>0 OR n_locations>0 OR n_partners>0 OR has_amount) AS any_field FROM fl""") r.dump("grounding_all_structured.csv", con, """ SELECT 'all' g, count(*) FILTER (WHERE any_field) n_with_fields, count(*) FILTER (WHERE any_field AND all_grounded) all_grounded, round(100.0*count(*) FILTER (WHERE any_field AND all_grounded)/count(*) FILTER (WHERE any_field),2) pct FROM comb UNION ALL SELECT form, count(*) FILTER (WHERE any_field), count(*) FILTER (WHERE any_field AND all_grounded), round(100.0*count(*) FILTER (WHERE any_field AND all_grounded)/count(*) FILTER (WHERE any_field),2) FROM comb GROUP BY 1 UNION ALL SELECT run, count(*) FILTER (WHERE any_field), count(*) FILTER (WHERE any_field AND all_grounded), round(100.0*count(*) FILTER (WHERE any_field AND all_grounded)/count(*) FILTER (WHERE any_field),2) FROM comb GROUP BY 1 UNION ALL SELECT conf_bin, count(*) FILTER (WHERE any_field), count(*) FILTER (WHERE any_field AND all_grounded), round(100.0*count(*) FILTER (WHERE any_field AND all_grounded)/count(*) FILTER (WHERE any_field),2) FROM comb GROUP BY 1 ORDER BY 1""") # partial grounding (element level) r.dump("grounding_element_level.csv", con, """ SELECT 'years' field, sum(n_years) elements, sum(years_found) found, round(100.0*sum(years_found)/sum(n_years),2) pct FROM fl UNION ALL SELECT 'locations', sum(n_locations), sum(locations_found), round(100.0*sum(locations_found)/sum(n_locations),2) FROM fl UNION ALL SELECT 'partners', sum(n_partners), sum(partners_found), round(100.0*sum(partners_found)/sum(n_partners),2) FROM fl UNION ALL SELECT 'technologies', sum(n_technologies), sum(technologies_found), round(100.0*sum(technologies_found)/sum(n_technologies),2) FROM fl""") # ---- fallback upper bounds and type-signal checks ------------------------------------------------------ FB = {} FB["mentions"] = con.execute("select count(*) from fl").fetchone()[0] FB["years_nonempty"] = con.execute("select count(*) from fl where n_years>0").fetchone()[0] FB["years_eq_regex_desc"] = con.execute("select count(*) from fl where years_eq_regex_desc").fetchone()[0] FB["years_eq_regex_desc_by_run"] = con.execute("select run, count(*) filter (where years_eq_regex_desc), count(*) filter (where n_years>0) from fl group by 1").fetchall() FB["years_not_subset_of_text"] = con.execute("select count(*) from fl where years_not_subset").fetchone()[0] FB["years_not_subset_and_eq_regex_desc"] = con.execute("select count(*) from fl where years_not_subset and years_eq_regex_desc").fetchone()[0] FB["techs_nonempty"] = con.execute("select count(*) from fl where n_technologies>0").fetchone()[0] FB["techs_eq_regex_desc"] = con.execute("select count(*) from fl where techs_eq_regex_desc").fetchone()[0] FB["techs_eq_regex_desc_by_run"] = con.execute("select run, count(*) filter (where techs_eq_regex_desc), count(*) filter (where n_technologies>0) from fl group by 1").fetchall() FB["techs_literal_in_text"] = con.execute("select count(*) filter (where technologies_all_grounded), count(*) from fl where n_technologies>0").fetchone() FB["type_signal_in_name_desc_by_type"] = con.execute("select project_type, count(*), count(*) filter (where type_signal_in_name_desc), round(100.0*count(*) filter (where type_signal_in_name_desc)/count(*),2) from fl group by 1 order by 2 desc").fetchall() FB["tech_deploy_mentioned_no_signal"] = con.execute("select count(*) from fl where project_type='technology_deployment' and status='mentioned' and not type_signal_in_name_desc").fetchone()[0] FB["tech_deploy_mentioned_no_any_signal"] = con.execute("select count(*) from fl where project_type='technology_deployment' and status='mentioned' and not any_type_signal_in_name_desc").fetchone()[0] FB["tech_deploy_total"] = con.execute("select count(*), count(*) filter (where status='mentioned') from fl where project_type='technology_deployment'").fetchone() FB["edit09_ma_regex_by_type"] = con.execute("select project_type, count(*), count(*) filter (where edit09_ma_regex), round(100.0*count(*) filter (where edit09_ma_regex)/count(*),2) from fl group by 1 order by 2 desc").fetchall() FB["edit09_investment_infrastructure"] = con.execute("select count(*), count(*) filter (where edit09_investment) from fl where project_type='infrastructure'").fetchone() FB["amount_raw_null_with_amount"] = con.execute("select count(*) from fl where amount_usd is not null and amount_raw is null").fetchone()[0] FB["amount_raw_null_amount_values"] = con.execute("select amount_usd, count(*) from fl where amount_usd is not null and amount_raw is null group by 1 order by 2 desc limit 5").fetchall() r.write_json("fallback_bounds.json", FB) # ---- amount tail --------------------------------------------------------------------------------------- T = {} T["n_with_amount"] = con.execute("select count(*) from fl where has_amount").fetchone()[0] T["max_amount"] = con.execute("select max(amount_usd) from fl").fetchone()[0] T["ge_50e9"] = con.execute("select count(*) from fl where amount_usd >= 50e9").fetchone()[0] T["ge_1e12"] = con.execute("select count(*) from fl where amount_usd >= 1e12").fetchone()[0] T["le_0"] = con.execute("select count(*) from fl where amount_usd <= 0").fetchone()[0] T["lt_0"] = con.execute("select count(*) from fl where amount_usd < 0").fetchone()[0] T["eq_0"] = con.execute("select count(*) from fl where amount_usd = 0").fetchone()[0] T["top200_grounded_number"] = con.execute("select count(*) filter (where amount_number_found), count(*) filter (where amount_currency_within_40), count(*) from (select * from fl where has_amount order by amount_usd desc limit 200)").fetchone() T["ge_50e9_grounded_number"] = con.execute("select count(*) filter (where amount_number_found), count(*) filter (where amount_currency_within_40), count(*) from fl where amount_usd >= 50e9").fetchone() T["projects_max_amount_ge_50e9"] = con.execute("select count(*) from projects where total_amount_usd >= 50e9").fetchone()[0] T["projects_max_amount_max"] = con.execute("select max(total_amount_usd) from projects").fetchone()[0] T["amount_quantiles_mentions"] = con.execute("select quantile_cont(amount_usd, [0.5,0.9,0.99,0.999]) from fl where has_amount").fetchone()[0] T["nonusd_within_40_share"] = con.execute("select count(*) filter (where amount_nonusd_within_40), count(*) filter (where amount_number_found) from fl").fetchone() r.write_json("amount_tail.json", T) r.dump("amount_top200.csv", con, """select mention_id, cik, form, item, project_type, amount_usd, amount_raw, amount_number_found, amount_currency_within_40, amount_nonusd_within_40, confidence, run from fl where has_amount order by amount_usd desc limit 200""") r.dump("amount_nonpositive.csv", con, """select mention_id, cik, form, item, project_type, amount_usd, amount_raw, confidence, run from fl where amount_usd <= 0 order by amount_usd limit 700""") r.dump("amount_grounding_by_conf_bin.csv", con, """select conf_bin, count(*) filter (where has_amount) with_amount, count(*) filter (where amount_number_found) number_found, count(*) filter (where amount_currency_within_40) with_currency, round(100.0*count(*) filter (where amount_number_found)/nullif(count(*) filter (where has_amount),0),2) pct_number, round(100.0*count(*) filter (where amount_currency_within_40)/nullif(count(*) filter (where has_amount),0),2) pct_currency from fl group by 1 order by 1""") r.dump("nonusd_by_state.csv", con, """select coalesce(nullif(state,''),'(none)') state, count(*) filter (where amount_number_found) number_found, count(*) filter (where amount_nonusd_within_40) nonusd, round(100.0*count(*) filter (where amount_nonusd_within_40)/nullif(count(*) filter (where amount_number_found),0),2) pct from fl group by 1 having count(*) filter (where amount_number_found) >= 100 order by 4 desc limit 40""") r.dump("nonusd_by_state_inc.csv", con, """select coalesce(nullif(state_inc,''),'(none)') state_of_incorporation, count(*) filter (where amount_number_found) number_found, count(*) filter (where amount_nonusd_within_40) nonusd, round(100.0*count(*) filter (where amount_nonusd_within_40)/nullif(count(*) filter (where amount_number_found),0),2) pct from fl group by 1 having count(*) filter (where amount_number_found) >= 100 order by 4 desc limit 40""") r.dump("grounding_by_type.csv", con, """select project_type, count(*) n, round(100.0*count(*) filter (where years_all_grounded)/nullif(count(*) filter (where years_all_grounded is not null),0),2) years_pct, round(100.0*count(*) filter (where locations_all_grounded)/nullif(count(*) filter (where locations_all_grounded is not null),0),2) locations_pct, round(100.0*count(*) filter (where partners_all_grounded)/nullif(count(*) filter (where partners_all_grounded is not null),0),2) partners_pct, round(100.0*count(*) filter (where amount_number_found)/nullif(count(*) filter (where amount_number_found is not null),0),2) amount_pct from fl group by 1 order by 2 desc""") r.finish() if __name__ == "__main__": main()