spb/firc0001_uqo
Public
1"""N2 - field-level grounding audit of the 308,065 mentions against their section text (cluster A, C, J).23For every mention: each year in years[] present as a 4-digit token; each location / partner string4present after case-folding and light normalisation; the amount present under scale-aware renderings5with a currency token within 40 characters; fallback upper bounds (years/technologies equal to the6regex output on the model's own description); amount tail; type-signal checks. Grouped by form, item,7confidence bin, run, segmenter, filing-date period; 1,000-replicate bootstrap clustered by section.8Read-only; section text streamed once; mention-level flags kept on the node (parquet).9"""10from __future__ import annotations11import json, re, sys, os, time, datetime as dt12from pathlib import Path13from collections import defaultdict14import numpy as np15import multiprocessing as mp16sys.path.insert(0, str(Path(__file__).resolve().parent))17from common import *18sys.path.insert(0, str(Path(__file__).resolve().parent / "spid"))19import taxonomy as tx2021NREP = 100022rng = np.random.default_rng(SEED)23YEAR_TOK = re.compile(r"(?<!\d)(19[89]\d|20[0-4]\d)(?!\d)")24CUR_NEAR = re.compile(r"\$|\busd\b|\bu\.s\. dollars?\b|\bdollars?\b", re.I)25NONUSD = re.compile(r"€|£|¥|₹|\b(eur|gbp|cad|aud|jpy|chf|cny|rmb|yuan|renminbi|pesos?|rupees?|reais?|krw|won|sek|nok|dkk|zar|rand|c\$|a\$|hk\$|nt\$|s\$|euros?|pounds? sterling|yen)\b", re.I)26SPACE = re.compile(r"\s+")27PUNCT = re.compile(r"[^\w\s]")28ALIASES = {"u.s.": "united states", "us": "united states", "usa": "united states", "u.s.a.": "united states", "united states of america": "united states",29 "u.k.": "united kingdom", "uk": "united kingdom", "uae": "united arab emirates", "eu": "european union"}30FIN_EDIT09 = re.compile(r"acqui|merger|purchase of|closing of the", re.I)3132def norm(s: str) -> str:33 s = s.casefold().strip()34 s = ALIASES.get(s, s)35 s = PUNCT.sub(" ", s)36 return SPACE.sub(" ", s).strip()3738def amount_candidates(a: float) -> list[str]:39 """Numeric strings under which an amount may appear in the text (scale-aware, 1-3 significant digits)."""40 out = set()41 if a is None or a == 0:42 return []43 a = abs(a)44 for scale in (1.0, 1e3, 1e6, 1e9, 1e12):45 v = a / scale46 if v < 0.05 or v >= 1e7:47 continue48 if scale == 1.0:49 if abs(v - round(v)) < 1e-6:50 out.add(f"{int(round(v)):,}"); out.add(str(int(round(v))))51 continue52 for fmt in ("{:.0f}", "{:.1f}", "{:.2f}", "{:.3f}", "{:,.0f}", "{:,.1f}", "{:,.2f}"):53 s = fmt.format(v)54 if "." in s:55 s2 = s.rstrip("0").rstrip(".")56 out.add(s2)57 out.add(s)58 return sorted(out, key=len, reverse=True)5960def find_number(text: str, cands: list[str]):61 for c in cands:62 start = 063 while True:64 i = text.find(c, start)65 if i < 0:66 break67 j = i + len(c)68 before = text[i-1] if i > 0 else " "69 after = text[j] if j < len(text) else " "70 if not (before.isdigit() or before in ".,") and not (after.isdigit() or (after in ".," and j+1 < len(text) and text[j+1].isdigit())):71 return i, j72 start = j73 return None7475MENTIONS: dict[str, list[dict]] = {}7677def process_section(item):78 sid, text = item79 ms = MENTIONS.get(sid)80 if not ms:81 return []82 low = text.casefold()83 lown = SPACE.sub(" ", PUNCT.sub(" ", low))84 years_in_text = set(YEAR_TOK.findall(text))85 out = []86 for m in ms:87 f = {"mention_id": m["mention_id"]}88 yrs = m["years"] or []89 f["n_years"] = len(yrs)90 f["years_found"] = sum(1 for y in yrs if str(y) in years_in_text)91 f["years_all_grounded"] = (f["years_found"] == len(yrs)) if yrs else None92 f["years_not_subset"] = (f["years_found"] < len(yrs)) if yrs else False93 locs = m["locations"] or []94 f["n_locations"] = len(locs)95 f["locations_found"] = sum(1 for s in locs if norm(s) and (" " + norm(s) + " ") in (" " + lown + " "))96 f["locations_all_grounded"] = (f["locations_found"] == len(locs)) if locs else None97 pars = m["partners"] or []98 f["n_partners"] = len(pars)99 f["partners_found"] = sum(1 for s in pars if norm(s) and (" " + norm(s) + " ") in (" " + lown + " "))100 f["partners_all_grounded"] = (f["partners_found"] == len(pars)) if pars else None101 techs = m["technologies"] or []102 f["n_technologies"] = len(techs)103 f["technologies_found"] = sum(1 for s in techs if norm(s) and (" " + norm(s) + " ") in (" " + lown + " "))104 f["technologies_all_grounded"] = (f["technologies_found"] == len(techs)) if techs else None105 a = m["amount_usd"]106 f["has_amount"] = a is not None107 f["amount_number_found"] = None; f["amount_currency_within_40"] = None; f["amount_nonusd_within_40"] = None108 if a is not None and a != 0:109 pos = find_number(text, amount_candidates(a))110 f["amount_number_found"] = pos is not None111 if pos:112 i, j = pos; win = text[max(0, i-40): j+40]113 f["amount_currency_within_40"] = bool(CUR_NEAR.search(win))114 f["amount_nonusd_within_40"] = bool(NONUSD.search(win))115 else:116 f["amount_currency_within_40"] = False; f["amount_nonusd_within_40"] = False117 elif a == 0:118 f["amount_number_found"] = False; f["amount_currency_within_40"] = False; f["amount_nonusd_within_40"] = False119 desc = m["description"] or ""120 ey = tx.extract_years(desc); et = tx.extract_technologies(desc)121 f["years_eq_regex_desc"] = bool(yrs) and sorted(yrs) == ey122 f["years_regex_desc_nonempty"] = bool(ey)123 f["techs_eq_regex_desc"] = bool(techs) and techs == et124 f["techs_regex_desc_nonempty"] = bool(et)125 nd = ((m["project_name"] or "") + " " + desc).lower()126 sigs = tx.TYPE_SIGNALS.get(m["project_type"], [])127 f["type_signal_in_name_desc"] = any(rx.search(nd) for rx in sigs)128 f["any_type_signal_in_name_desc"] = any(rx.search(nd) for rxs in tx.TYPE_SIGNALS.values() for rx in rxs)129 f["edit09_ma_regex"] = bool(FIN_EDIT09.search(nd))130 f["edit09_investment"] = "investment" in nd131 out.append(f)132 return out133134def main():135 r = Run("grounding")136 con = connect(SPID_DB)137 tmin = con.execute("select min(extracted_at) from extracted_sections").fetchone()[0]138 print("loading mentions ...", flush=True)139 cur = con.execute(f"""140 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,141 m.confidence, m.form_type, m.section_name, m.filing_date, m.cik,142 CASE WHEN e.extracted_at = TIMESTAMP '{tmin}' THEN 'backfilled_june' ELSE 'october' END AS run,143 CASE WHEN ss.section_id IS NOT NULL THEN 'legacy_ingest' ELSE 'parsed' END AS segmenter,144 co.in_sp500, co.state, co.state_of_incorporation145 FROM project_mentions m JOIN extracted_sections e USING (section_id) LEFT JOIN spid_sections ss ON ss.section_id = m.section_id146 LEFT JOIN src_companies co ON m.cik = co.cik""")147 cols = [d[0] for d in cur.description]148 meta = {}149 for row in cur.fetchall():150 d = dict(zip(cols, row))151 MENTIONS.setdefault(d["section_id"], []).append(d)152 meta[d["mention_id"]] = d153 n_m = len(meta); n_s = len(MENTIONS)154 print(f"{n_m} mentions over {n_s} sections", flush=True)155 con.execute("CREATE TEMP TABLE need AS SELECT DISTINCT section_id FROM project_mentions")156 q = """SELECT s.section_id, s.text FROM src_sections s JOIN need USING (section_id)157 UNION ALL SELECT s.section_id, s.text FROM spid_sections s JOIN need USING (section_id)"""158 cur = con.execute(q)159 flags = []160 t0 = time.time(); done = 0; chars = 0161 ctx = mp.get_context("fork")162 with ctx.Pool(4) as pool:163 def batches():164 while True:165 rows = cur.fetchmany(2000)166 if not rows:167 break168 yield rows169 for rows in batches():170 chars += sum(len(t) for _, t in rows)171 for res in pool.imap_unordered(process_section, rows, chunksize=50):172 flags.extend(res)173 done += len(rows)174 if done % 20000 < 2000:175 print(f" {done}/{n_s} sections, {len(flags)} mentions, {chars/1e9:.2f} GB, {time.time()-t0:.0f}s", flush=True)176 r.notes["sections_streamed"] = done; r.notes["chars_streamed"] = chars; r.notes["mentions_flagged"] = len(flags)177 r.notes["sections_with_mentions_not_found_in_text_sources"] = n_s - done178 # ---- assemble ---------------------------------------------------------------------------------179 import pyarrow as pa, pyarrow.parquet as pq180 recs = []181 for f in flags:182 m = meta[f["mention_id"]]183 c = m["confidence"]184 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"185 recs.append({**f, "section_id": m["section_id"], "form": m["form_type"], "item": m["section_name"], "run": m["run"], "segmenter": m["segmenter"],186 "conf_bin": cb, "period": "pre_2023-10" if (m["filing_date"] and str(m["filing_date"]) < "2023-10-01") else "post_2023-10",187 "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"],188 "amount_usd": m["amount_usd"], "amount_raw": m["amount_raw"], "confidence": c, "cik": m["cik"], "filing_date": str(m["filing_date"])})189 tbl = pa.Table.from_pylist(recs)190 pq.write_table(tbl, r.out / "mention_flags.parquet") # stays on the node (> 5 MB)191 con.register("fl", tbl)192 FIELDS = {"years": "years_all_grounded", "locations": "locations_all_grounded", "partners": "partners_all_grounded", "technologies": "technologies_all_grounded",193 "amount_number": "amount_number_found", "amount_with_currency": "amount_currency_within_40"}194 def boot(groupcol: str, fname: str, flag: str, where: str = "TRUE"):195 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()196 bygrp = defaultdict(list)197 for g, sid, n, k in rows: bygrp[g].append((n, k))198 out = []199 for g, lst in bygrp.items():200 n = np.array([x[0] for x in lst], float); k = np.array([x[1] for x in lst], float)201 share = k.sum()/n.sum()202 if len(lst) > 1:203 w = rng.poisson(1.0, size=(NREP, len(lst)))204 reps = (w @ k) / np.maximum(1e-9, (w @ n))205 lo, hi = np.percentile(reps, [2.5, 97.5])206 else:207 lo = hi = share208 out.append((fname, g, int(n.sum()), len(lst), int(k.sum()), round(share, 4), round(float(lo), 4), round(float(hi), 4)))209 return out210 cols_out = ["field", "group", "n_mentions", "n_sections", "n_grounded", "grounded_share", "ci_low", "ci_high"]211 allrows = []212 for grouping in ["'all'", "form", "run", "segmenter", "conf_bin", "period", "in_sp500::varchar", "project_type", "status"]:213 rows = []214 for fname, flag in FIELDS.items():215 rows += boot(grouping, fname, flag)216 gname = grouping.replace("'", "").replace("::varchar", "")217 allrows += [(gname,) + x for x in rows]218 r.write_csv("grounding_by_group.csv", ["grouping"] + cols_out, allrows)219 rows = []220 for fname, flag in FIELDS.items():221 rows += [(x[1].split("|")[0], x[1].split("|")[1]) + (x[0],) + x[2:] for x in boot("form || '|' || item", fname, flag)]222 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)223 # combined 'all structured fields grounded' per mention224 con.execute("""CREATE TEMP TABLE comb AS SELECT section_id, form, run, conf_bin,225 (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,226 (n_years>0 OR n_locations>0 OR n_partners>0 OR has_amount) AS any_field FROM fl""")227 r.dump("grounding_all_structured.csv", con, """228 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 comb229 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 1230 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 1231 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""")232 # partial grounding (element level)233 r.dump("grounding_element_level.csv", con, """234 SELECT 'years' field, sum(n_years) elements, sum(years_found) found, round(100.0*sum(years_found)/sum(n_years),2) pct FROM fl235 UNION ALL SELECT 'locations', sum(n_locations), sum(locations_found), round(100.0*sum(locations_found)/sum(n_locations),2) FROM fl236 UNION ALL SELECT 'partners', sum(n_partners), sum(partners_found), round(100.0*sum(partners_found)/sum(n_partners),2) FROM fl237 UNION ALL SELECT 'technologies', sum(n_technologies), sum(technologies_found), round(100.0*sum(technologies_found)/sum(n_technologies),2) FROM fl""")238 # ---- fallback upper bounds and type-signal checks ------------------------------------------------------239 FB = {}240 FB["mentions"] = con.execute("select count(*) from fl").fetchone()[0]241 FB["years_nonempty"] = con.execute("select count(*) from fl where n_years>0").fetchone()[0]242 FB["years_eq_regex_desc"] = con.execute("select count(*) from fl where years_eq_regex_desc").fetchone()[0]243 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()244 FB["years_not_subset_of_text"] = con.execute("select count(*) from fl where years_not_subset").fetchone()[0]245 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]246 FB["techs_nonempty"] = con.execute("select count(*) from fl where n_technologies>0").fetchone()[0]247 FB["techs_eq_regex_desc"] = con.execute("select count(*) from fl where techs_eq_regex_desc").fetchone()[0]248 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()249 FB["techs_literal_in_text"] = con.execute("select count(*) filter (where technologies_all_grounded), count(*) from fl where n_technologies>0").fetchone()250 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()251 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]252 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]253 FB["tech_deploy_total"] = con.execute("select count(*), count(*) filter (where status='mentioned') from fl where project_type='technology_deployment'").fetchone()254 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()255 FB["edit09_investment_infrastructure"] = con.execute("select count(*), count(*) filter (where edit09_investment) from fl where project_type='infrastructure'").fetchone()256 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]257 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()258 r.write_json("fallback_bounds.json", FB)259 # ---- amount tail ---------------------------------------------------------------------------------------260 T = {}261 T["n_with_amount"] = con.execute("select count(*) from fl where has_amount").fetchone()[0]262 T["max_amount"] = con.execute("select max(amount_usd) from fl").fetchone()[0]263 T["ge_50e9"] = con.execute("select count(*) from fl where amount_usd >= 50e9").fetchone()[0]264 T["ge_1e12"] = con.execute("select count(*) from fl where amount_usd >= 1e12").fetchone()[0]265 T["le_0"] = con.execute("select count(*) from fl where amount_usd <= 0").fetchone()[0]266 T["lt_0"] = con.execute("select count(*) from fl where amount_usd < 0").fetchone()[0]267 T["eq_0"] = con.execute("select count(*) from fl where amount_usd = 0").fetchone()[0]268 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()269 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()270 T["projects_max_amount_ge_50e9"] = con.execute("select count(*) from projects where total_amount_usd >= 50e9").fetchone()[0]271 T["projects_max_amount_max"] = con.execute("select max(total_amount_usd) from projects").fetchone()[0]272 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]273 T["nonusd_within_40_share"] = con.execute("select count(*) filter (where amount_nonusd_within_40), count(*) filter (where amount_number_found) from fl").fetchone()274 r.write_json("amount_tail.json", T)275 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""")276 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""")277 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,278 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""")279 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""")280 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""")281 r.dump("grounding_by_type.csv", con, """select project_type, count(*) n,282 round(100.0*count(*) filter (where years_all_grounded)/nullif(count(*) filter (where years_all_grounded is not null),0),2) years_pct,283 round(100.0*count(*) filter (where locations_all_grounded)/nullif(count(*) filter (where locations_all_grounded is not null),0),2) locations_pct,284 round(100.0*count(*) filter (where partners_all_grounded)/nullif(count(*) filter (where partners_all_grounded is not null),0),2) partners_pct,285 round(100.0*count(*) filter (where amount_number_found)/nullif(count(*) filter (where amount_number_found is not null),0),2) amount_pct286 from fl group by 1 order by 2 desc""")287 r.finish()288289if __name__ == "__main__":290 main()291