spb/firc0001_uqo
Public
1"""N1 - ledger reconciliation (clusters C, G, J-throughput, H-log numbers). Read-only.23(a) _PROJECT_GATE over every ledger section (DuckDB RE2, case-insensitive) -> gate pass and the4 true call count by form x item x run; productive rate / mentions / USD per call.5(b) backfilled block (rows at min(extracted_at)), word_count < 30 rows, outage-window rows,6 zero-mention share per flush timestamp and per 5-minute bin.7(c) sections where ledger n_mentions > stored mentions (id collisions).8(d) run-log parser (October data/logs/spid_extract.log, June project-db/logs/extract.log).9(e) wall-clock / outage / effective throughput from ledger timestamps.10(f) backend values, confidence = 0.600 count, amount_raw formatting rule, min confidence by run.11"""12from __future__ import annotations13import json, re, sys, datetime as dt14from pathlib import Path15from collections import Counter, defaultdict16sys.path.insert(0, str(Path(__file__).resolve().parent))17from common import *1819GATE = (r"\b(project|initiative|program|invest|expan|construct|build|launch|develop|"20 r"deploy|implement|acqui|partner|facilit|capacity|capital|migrat|transform|"21 r"pipeline|plant|data ?cent|automat|sustainab|renewable|restructur)\w*")22OUTAGE_START = "2026-10-28 01:10:00" # window named in the consolidated review (THEO-05 / DATA-02)23OUTAGE_END = "2026-10-28 06:22:00"24PAPER_USD_TOTAL = 775.0 # construct.tex:35 'roughly 775 USD'25PAPER_USD_PER_SECTION = 0.0008 # construct.tex:3526JUNE_LOG_SECTIONS = 153894 # project-db/logs/extract.log FULL EXTRACT COMPLETE27JUNE_LOG_MENTIONS = 39930 # TOTAL mentions in DB after the June run2829r = Run("ledger")30con = connect(SPID_DB)3132# ---- (a) section metadata + gate flag for every section visible to the pipeline --------------33print("building sec_meta (regex over the parquet text, streamed) ...", flush=True)34con.execute(f"""35CREATE TEMP TABLE sec_meta AS36SELECT section_id, accession_number, section_name, word_count, length(text) AS nchar,37 regexp_matches(text, '{GATE}', 'i') AS gate, 'parsed' AS segmenter38FROM src_sections39UNION ALL40SELECT section_id, accession_number, section_name, word_count, length(text) AS nchar,41 regexp_matches(text, '{GATE}', 'i') AS gate, 'legacy_ingest' AS segmenter42FROM spid_sections43""")44r.notes["sec_meta_rows"] = con.execute("select count(*) from sec_meta").fetchone()[0]4546con.execute("""47CREATE TEMP TABLE stored AS48SELECT section_id, count(*) AS stored_mentions,49 list(project_type || ':' || left(project_name, 40)) AS stored_keys50FROM project_mentions GROUP BY 151""")52tmin = con.execute("select min(extracted_at) from extracted_sections").fetchone()[0]53r.notes["min_extracted_at"] = str(tmin)54con.execute(f"""55CREATE TEMP TABLE ledger AS56SELECT e.section_id, e.accession_number, e.backend, e.n_mentions, e.extracted_at,57 CASE WHEN e.extracted_at = TIMESTAMP '{tmin}' THEN 'backfilled_june' ELSE 'october' END AS run,58 m.section_name, m.word_count, m.nchar, m.gate, m.segmenter,59 f.form_type, f.filing_date, f.cik, f.fiscal_year,60 co.in_sp500, co.in_russell3000,61 coalesce(s.stored_mentions, 0) AS stored_mentions, s.stored_keys62FROM extracted_sections e63LEFT JOIN sec_meta m USING (section_id)64LEFT JOIN src_filings f ON f.accession_number = e.accession_number65LEFT JOIN src_companies co ON f.cik = co.cik66LEFT JOIN stored s USING (section_id)67""")68n_ledger = con.execute("select count(*) from ledger").fetchone()[0]69n_unmatched = con.execute("select count(*) from ledger where section_name is null").fetchone()[0]70r.notes["ledger_rows"] = n_ledger71r.notes["ledger_rows_without_section_metadata"] = n_unmatched7273agg = """74 count(*) AS ledger_rows,75 count(*) FILTER (WHERE word_count < 30) AS lt30_words,76 count(*) FILTER (WHERE gate) AS gate_pass,77 count(*) FILTER (WHERE gate AND word_count >= 30) AS calls,78 count(*) FILTER (WHERE NOT gate AND n_mentions = 0) AS gate_rejected_zero,79 count(*) FILTER (WHERE NOT gate AND n_mentions > 0) AS gate_fail_but_mentions,80 count(*) FILTER (WHERE n_mentions > 0) AS with_mentions,81 sum(n_mentions) AS ledger_mentions,82 sum(stored_mentions) AS stored_mentions,83 count(DISTINCT cik) AS issuers,84 round(100.0 * count(*) FILTER (WHERE n_mentions > 0) / count(*), 2) AS productive_pct_ledger,85 round(100.0 * count(*) FILTER (WHERE n_mentions > 0) / nullif(count(*) FILTER (WHERE gate AND word_count >= 30),0), 2) AS productive_pct_calls,86 round(sum(n_mentions) * 1.0 / count(*), 4) AS mentions_per_ledger_row,87 round(sum(n_mentions) * 1.0 / nullif(count(*) FILTER (WHERE gate AND word_count >= 30),0), 4) AS mentions_per_call88"""89r.dump("gate_pass_by_form_item.csv", con, f"SELECT form_type, section_name AS item, run, {agg} FROM ledger GROUP BY 1,2,3 ORDER BY 1,2,3")90r.dump("gate_pass_by_form.csv", con, f"SELECT form_type, {agg} FROM ledger GROUP BY 1 ORDER BY 1")91r.dump("gate_pass_by_run.csv", con, f"SELECT run, {agg} FROM ledger GROUP BY 1 ORDER BY 1")92r.dump("gate_pass_by_form_run.csv", con, f"SELECT form_type, run, {agg} FROM ledger GROUP BY 1,2 ORDER BY 1,2")93tot = con.execute(f"SELECT {agg} FROM ledger").fetchone()94cols = ["ledger_rows","lt30_words","gate_pass","calls","gate_rejected_zero","gate_fail_but_mentions","with_mentions","ledger_mentions","stored_mentions","issuers","productive_pct_ledger","productive_pct_calls","mentions_per_ledger_row","mentions_per_call"]95T = dict(zip(cols, tot))96oct_calls = con.execute("select count(*) filter (where gate and word_count>=30) from ledger where run='october'").fetchone()[0]97oct_rows = con.execute("select count(*) from ledger where run='october'").fetchone()[0]98T["usd_per_call_paper_total_775"] = round(PAPER_USD_TOTAL / oct_calls, 6)99T["usd_per_october_row_paper_total_775"] = round(PAPER_USD_TOTAL / oct_rows, 6)100T["usd_per_call_from_0_0008_x_ledger"] = round(PAPER_USD_PER_SECTION * n_ledger / T["calls"], 6)101T["october_rows"] = oct_rows; T["october_calls"] = oct_calls102T["headline_before"] = {"sections_submitted": 1123098, "productive_pct": 15.0, "mentions_per_section": 0.27, "usd_per_section": 0.0008}103r.write_json("gate_pass_totals.json", T)104105# ---- (b) backfilled block, < 30 words, outage window, flush timestamps -----------------------106B = {}107B["min_extracted_at"] = str(tmin)108B["rows_at_min_timestamp"] = con.execute("select count(*) from ledger where run='backfilled_june'").fetchone()[0]109B["june_log_sections"] = JUNE_LOG_SECTIONS110B["gap_rows_minus_log"] = B["rows_at_min_timestamp"] - JUNE_LOG_SECTIONS111B["block_ledger_mentions"] = con.execute("select sum(n_mentions) from ledger where run='backfilled_june'").fetchone()[0]112B["block_stored_mentions"] = con.execute("select sum(stored_mentions) from ledger where run='backfilled_june'").fetchone()[0]113B["june_log_mentions"] = JUNE_LOG_MENTIONS114B["block_lt30_words"] = con.execute("select count(*) from ledger where run='backfilled_june' and word_count<30").fetchone()[0]115B["block_gate_fail"] = con.execute("select count(*) from ledger where run='backfilled_june' and not gate").fetchone()[0]116B["block_lt30_or_gate_fail"] = con.execute("select count(*) from ledger where run='backfilled_june' and (word_count<30 or not gate)").fetchone()[0]117B["block_ge30_and_gate"] = con.execute("select count(*) from ledger where run='backfilled_june' and word_count>=30 and gate").fetchone()[0]118B["october_lt30_words"] = con.execute("select count(*) from ledger where run='october' and word_count<30").fetchone()[0]119B["block_by_form"] = con.execute("select form_type, count(*), sum(n_mentions), count(*) filter (where word_count<30) from ledger where run='backfilled_june' group by 1 order by 1").fetchall()120B["block_by_segmenter"] = con.execute("select segmenter, count(*) from ledger where run='backfilled_june' group by 1").fetchall()121B["block_sp500_share"] = con.execute("select count(*) filter (where in_sp500), count(*) from ledger where run='backfilled_june'").fetchone()122B["block_ciks"] = con.execute("select count(distinct cik) from ledger where run='backfilled_june'").fetchone()[0]123B["outage_window"] = [OUTAGE_START, OUTAGE_END]124B["rows_in_outage_window"] = con.execute(f"select count(*), sum(n_mentions) from ledger where extracted_at between TIMESTAMP '{OUTAGE_START}' and TIMESTAMP '{OUTAGE_END}'").fetchone()125B["rows_0110_0640"] = con.execute("select count(*), sum(n_mentions) from ledger where extracted_at between TIMESTAMP '2026-10-28 01:10:00' and TIMESTAMP '2026-10-28 06:40:53'").fetchone()126B["rows_by_hour_1028"] = con.execute("select hour(extracted_at) h, count(*), count(*) filter (where n_mentions=0) zeros, sum(n_mentions) from ledger where run='october' and extracted_at::date = DATE '2026-10-28' group by 1 order by 1").fetchall()127B["rows_by_hour_1027"] = con.execute("select hour(extracted_at) h, count(*), count(*) filter (where n_mentions=0) zeros, sum(n_mentions) from ledger where run='october' and extracted_at::date = DATE '2026-10-27' group by 1 order by 1").fetchall()128B["paper_outage_rows_149352"] = 149352129r.write_json("backfill_block.json", B)130131r.dump("ledger_by_5min.csv", con, """132SELECT time_bucket(INTERVAL '5 minutes', extracted_at) AS bin5, count(*) AS rows, count(*) FILTER (WHERE n_mentions>0) AS with_mentions,133 sum(n_mentions) AS mentions, round(100.0*count(*) FILTER (WHERE n_mentions=0)/count(*),2) AS zero_pct,134 count(*) FILTER (WHERE gate AND word_count>=30) AS calls, count(DISTINCT extracted_at) AS flushes135FROM ledger WHERE run='october' GROUP BY 1 ORDER BY 1""")136r.dump("ledger_by_minute.csv", con, """137SELECT time_bucket(INTERVAL '1 minute', extracted_at) AS bin1, count(*) AS rows, count(*) FILTER (WHERE n_mentions>0) AS with_mentions,138 sum(n_mentions) AS mentions, round(100.0*count(*) FILTER (WHERE n_mentions=0)/count(*),2) AS zero_pct,139 count(*) FILTER (WHERE gate AND word_count>=30) AS calls,140 round(100.0*count(*) FILTER (WHERE n_mentions=0 AND gate AND word_count>=30)/nullif(count(*) FILTER (WHERE gate AND word_count>=30),0),2) AS zero_pct_among_calls141FROM ledger WHERE run='october' GROUP BY 1 ORDER BY 1""")142B["october_distinct_timestamps"], B["october_rows_check"] = con.execute("select count(distinct extracted_at), count(*) from ledger where run='october'").fetchone()143B["timestamp_granularity_note"] = "extracted_at is stamped per row (executemany INSERT OR REPLACE, DEFAULT current_timestamp), not per flush: distinct timestamps = rows"144r.dump("flush_zero_groups_after_0622.csv", con, f"""145SELECT time_bucket(INTERVAL '1 minute', extracted_at) AS bin1, count(*) AS rows, count(*) FILTER (WHERE n_mentions=0) AS zeros,146 round(100.0*count(*) FILTER (WHERE n_mentions=0)/count(*),2) AS zero_pct,147 count(*) FILTER (WHERE gate AND word_count>=30) AS calls,148 round(100.0*count(*) FILTER (WHERE n_mentions=0 AND gate AND word_count>=30)/nullif(count(*) FILTER (WHERE gate AND word_count>=30),0),2) AS zero_pct_among_calls149FROM ledger WHERE run='october' AND extracted_at > TIMESTAMP '{OUTAGE_END}'150GROUP BY 1 HAVING count(*) >= 50 AND count(*) FILTER (WHERE n_mentions=0 AND gate AND word_count>=30) >= 0.99*count(*) FILTER (WHERE gate AND word_count>=30) ORDER BY 1""")151# the same test before the outage, for comparison152r.dump("flush_zero_groups_before_0110.csv", con, f"""153SELECT time_bucket(INTERVAL '1 minute', extracted_at) AS bin1, count(*) AS rows, count(*) FILTER (WHERE n_mentions=0) AS zeros,154 round(100.0*count(*) FILTER (WHERE n_mentions=0)/count(*),2) AS zero_pct,155 count(*) FILTER (WHERE gate AND word_count>=30) AS calls,156 round(100.0*count(*) FILTER (WHERE n_mentions=0 AND gate AND word_count>=30)/nullif(count(*) FILTER (WHERE gate AND word_count>=30),0),2) AS zero_pct_among_calls157FROM ledger WHERE run='october' AND extracted_at < TIMESTAMP '{OUTAGE_START}'158GROUP BY 1 HAVING count(*) >= 50 AND count(*) FILTER (WHERE n_mentions=0 AND gate AND word_count>=30) >= 0.99*count(*) FILTER (WHERE gate AND word_count>=30) ORDER BY 1""")159160# ---- (c) id collisions: ledger n_mentions > stored ------------------------------------------161r.dump("ledger_collisions.csv", con, """162SELECT section_id, accession_number, form_type, section_name AS item, cik, run, n_mentions AS ledger_n, stored_mentions AS stored_n,163 n_mentions - stored_mentions AS deficit, stored_keys164FROM ledger WHERE n_mentions > stored_mentions ORDER BY deficit DESC, section_id""")165r.dump("ledger_collisions_by_form.csv", con, """166SELECT form_type, count(*) AS sections, sum(n_mentions - stored_mentions) AS excess, sum(n_mentions) AS ledger_mentions, sum(stored_mentions) AS stored_mentions167FROM ledger WHERE n_mentions > stored_mentions GROUP BY 1 ORDER BY 1""")168C = {}169C["sections_ledger_gt_stored"], C["excess_mentions"] = con.execute("select count(*), sum(n_mentions-stored_mentions) from ledger where n_mentions > stored_mentions").fetchone()170C["sections_ledger_lt_stored"], C["deficit_mentions"] = con.execute("select count(*), sum(stored_mentions-n_mentions) from ledger where n_mentions < stored_mentions").fetchone()171C["sum_ledger_n_mentions"] = con.execute("select sum(n_mentions) from ledger").fetchone()[0]172C["stored_project_mentions"] = con.execute("select count(*) from project_mentions").fetchone()[0]173C["mentions_whose_section_not_in_ledger"] = con.execute("select count(*) from project_mentions m where not exists (select 1 from extracted_sections e where e.section_id=m.section_id)").fetchone()[0]174C["ledger_vs_stored_by_form"] = con.execute("select form_type, sum(n_mentions), sum(stored_mentions), sum(n_mentions)-sum(stored_mentions) from ledger group by 1 order by 1").fetchall()175C["duplicate_key_sections_in_stored"] = con.execute("""176 select count(*) from (select section_id from project_mentions group by section_id, project_type, left(project_name,40) having count(*)>1)""").fetchone()[0]177r.write_json("collisions.json", C)178179# ---- (f) misc code-vs-data checks ---------------------------------------------------------------180F = {}181F["distinct_backend_ledger"] = [x[0] for x in con.execute("select distinct backend from extracted_sections").fetchall()]182F["distinct_backend_mentions"] = [x[0] for x in con.execute("select distinct backend from project_mentions").fetchall()]183F["confidence_exact_0600_total"] = con.execute("select count(*) from project_mentions where round(confidence,3)=0.600").fetchone()[0]184F["confidence_exact_0600_by_run"] = con.execute("""185 select l.run, count(*) from project_mentions m join ledger l using(section_id) where round(m.confidence,3)=0.600 group by 1""").fetchall()186F["confidence_bins_by_run"] = con.execute("""187 select l.run, round(m.confidence,1) c, count(*) from project_mentions m join ledger l using(section_id) group by 1,2 order by 1,2""").fetchall()188F["min_confidence_by_run"] = con.execute("select l.run, min(m.confidence), max(m.confidence), count(*) from project_mentions m join ledger l using(section_id) group by 1").fetchall()189F["mentions_below_045"] = con.execute("select count(*) from project_mentions where confidence < 0.45").fetchone()[0]190F["amount_raw_rule_mismatch"] = con.execute("""191 select count(*) from project_mentions where amount_usd is not null and amount_raw is distinct from ('$' || format('{:,.0f}', amount_usd))""").fetchone()[0]192F["amount_raw_rule_mismatch_rows"] = con.execute("""193 select mention_id, amount_usd, amount_raw from project_mentions where amount_usd is not null and amount_raw is distinct from ('$' || format('{:,.0f}', amount_usd)) limit 20""").fetchall()194F["amount_raw_null_with_amount"] = con.execute("select count(*) from project_mentions where amount_usd is not null and amount_raw is null").fetchone()[0]195F["amount_raw_notnull_without_amount"] = con.execute("select count(*) from project_mentions where amount_usd is null and amount_raw is not null").fetchone()[0]196F["amount_le_0"] = con.execute("select count(*) from project_mentions where amount_usd <= 0").fetchone()[0]197F["mentions_with_run"] = con.execute("select l.run, count(*) from project_mentions m join ledger l using(section_id) group by 1").fetchall()198F["mentions_backfilled_block_vs_june_log"] = {"block_stored": B["block_stored_mentions"], "june_log": JUNE_LOG_MENTIONS}199F["fiscal_year_null_mentions_by_run"] = con.execute("select l.run, count(*) filter (where m.fiscal_year is null), count(*) from project_mentions m join ledger l using(section_id) group by 1").fetchall()200r.write_json("code_vs_data_checks.json", F)201202# ---- (e) throughput from the ledger ----------------------------------------------------------203E = {}204ts = con.execute("select distinct extracted_at from ledger where run='october' order by 1").fetchall()205ts = [t[0] for t in ts]206E["october_first_flush"] = str(ts[0]); E["october_last_flush"] = str(ts[-1])207E["october_distinct_flush_timestamps"] = len(ts)208E["october_rows"] = oct_rows209E["october_calls"] = oct_calls210wall = (ts[-1] - ts[0]).total_seconds()211gaps = sorted(((ts[i+1]-ts[i]).total_seconds(), str(ts[i]), str(ts[i+1])) for i in range(len(ts)-1))212E["largest_gaps_seconds"] = gaps[-5:]213outage_gap = gaps[-1][0]214E["wall_clock_hours_first_to_last_flush"] = round(wall/3600, 3)215E["largest_gap_hours"] = round(outage_gap/3600, 3)216E["effective_hours_excluding_largest_gap"] = round((wall-outage_gap)/3600, 3)217E["rows_per_second_wall"] = round(oct_rows/wall, 2)218E["rows_per_second_effective"] = round(oct_rows/(wall-outage_gap), 2)219E["calls_per_second_effective"] = round(oct_calls/(wall-outage_gap), 2)220E["paper_claim"] = {"sections": 968134, "hours": "about twelve", "rate": "27-36 sections per second", "workers": 48}221E["paper_rate_implied_by_12h"] = round(968134/(12*3600), 2)222# segment rates: before the outage and after223pre = con.execute(f"select count(*), min(extracted_at), max(extracted_at) from ledger where run='october' and extracted_at < TIMESTAMP '{OUTAGE_START}'").fetchone()224post = con.execute(f"select count(*), min(extracted_at), max(extracted_at) from ledger where run='october' and extracted_at > TIMESTAMP '{OUTAGE_END}'").fetchone()225E["segment_before_outage"] = {"rows": pre[0], "from": str(pre[1]), "to": str(pre[2]), "rows_per_s": round(pre[0]/max(1,(pre[2]-pre[1]).total_seconds()),2)}226E["segment_after_outage"] = {"rows": post[0], "from": str(post[1]), "to": str(post[2]), "rows_per_s": round(post[0]/max(1,(post[2]-post[1]).total_seconds()),2)}227E["rows_by_hour_october"] = con.execute("select date_trunc('hour', extracted_at) h, count(*) from ledger where run='october' group by 1 order by 1").fetchall()228r.write_json("throughput.json", E)229230# ---- (d) log parsers ---------------------------------------------------------------------------231LINE = re.compile(r"^(\d\d):(\d\d):(\d\d) (\w+)\s+(\S+) \| (.*)$")232def parse_log(path: Path, start_date: dt.date):233 day = start_date; prev = None; out = []234 for raw in open(path, errors="replace"):235 m = LINE.match(raw.rstrip("\n"))236 if not m:237 out.append((None, None, None, raw.rstrip("\n"))); continue238 h, mi, s = int(m.group(1)), int(m.group(2)), int(m.group(3))239 t = dt.time(h, mi, s)240 if prev is not None and (dt.datetime.combine(day, prev) - dt.datetime.combine(day, t)).total_seconds() > 3600:241 day = day + dt.timedelta(days=1)242 prev = t243 out.append((dt.datetime.combine(day, t), m.group(4), m.group(5), m.group(6)))244 return out245246def classify(msg: str) -> str:247 if "retryable" in msg: return "retryable_http_429_5xx"248 if "Connection reset" in msg or "ConnectError" in msg or "ReadError" in msg or "Errno" in msg: return "connection_error"249 if "Unterminated string" in msg or "Expecting" in msg or "Extra data" in msg or "JSON" in msg: return "json_decode_error"250 if "KeyError" in msg: return "key_error"251 if "timed out" in msg.lower() or "Timeout" in msg: return "timeout"252 return "other"253254def summarize(path: Path, start_date: dt.date, window=None):255 rows = parse_log(path, start_date)256 S = {"path": str(path), "lines": len(rows), "first": str(next((r_[0] for r_ in rows if r_[0]), None)), "last": str(next((r_[0] for r_ in reversed(rows) if r_[0]), None))}257 fails = [(t, classify(msg)) for t, lvl, name, msg in rows if msg and "LLM call failed" in msg]258 S["llm_call_failed_total"] = len(fails)259 S["llm_call_failed_by_reason"] = dict(Counter(c for _, c in fails))260 byhour = Counter((t.strftime("%Y-%m-%d %H") if t else "?") for t, _ in fails)261 S["llm_call_failed_by_hour"] = dict(sorted(byhour.items()))262 if window:263 w0, w1 = window264 inside = [(t, c) for t, c in fails if t and w0 <= t <= w1]265 outside = [(t, c) for t, c in fails if not (t and w0 <= t <= w1)]266 S["window"] = [str(w0), str(w1)]267 S["llm_call_failed_inside_window"] = len(inside)268 S["llm_call_failed_outside_window"] = len(outside)269 S["outside_window_by_reason"] = dict(Counter(c for _, c in outside))270 S["outside_window_list"] = [(str(t), c) for t, c in outside]271 if fails:272 S["first_failure"] = str(min(t for t, _ in fails if t)); S["last_failure"] = str(max(t for t, _ in fails if t))273 ins = [t for t, _ in inside]274 if ins: S["first_failure_in_window"] = str(min(ins)); S["last_failure_in_window"] = str(max(ins))275 S["section_failed"] = sum(1 for t, lvl, name, msg in rows if msg and "section failed" in msg)276 S["batch_failed_lines"] = sum(1 for t, lvl, name, msg in rows if msg and "BATCH" in msg and "FAILED" in msg)277 S["pending_lines"] = [(str(t), msg) for t, lvl, name, msg in rows if msg and msg.startswith("pending:")]278 S["ciks_total_lines"] = [(str(t), msg) for t, lvl, name, msg in rows if msg and msg.startswith("CIKs total")]279 S["run_complete_lines"] = [(str(t), msg) for t, lvl, name, msg in rows if msg and ("RUN COMPLETE" in msg or "EXTRACT COMPLETE" in msg)]280 ext = [int(re.search(r"Extracting from (\d+) sections", msg).group(1)) for t, lvl, name, msg in rows if msg and "Extracting from" in msg]281 S["extracting_from_lines"] = len(ext); S["extracting_from_sum_sections"] = sum(ext)282 done = [(int(a), int(b)) for t, lvl, name, msg in rows for a, b in re.findall(r"Extraction complete: (\d+) mentions from (\d+) sections", msg or "")]283 S["extraction_complete_lines"] = len(done); S["extraction_complete_sum_sections"] = sum(b for a, b in done); S["extraction_complete_sum_mentions"] = sum(a for a, b in done)284 S["workers_values"] = sorted(set(re.findall(r"workers=(\d+)", "\n".join(msg or "" for *_, msg in rows))))285 S["batch_values"] = sorted(set(re.findall(r"batch=(\d+)", "\n".join(msg or "" for *_, msg in rows))))286 S["model_id_strings"] = sorted(set(re.findall(r"gpt-[\w\-\.]+", open(path, errors="replace").read())))287 return S, fails288289w0 = dt.datetime.fromisoformat(OUTAGE_START); w1 = dt.datetime.fromisoformat(OUTAGE_END)290oct_log, oct_fails = summarize(RDB / "data/logs/spid_extract.log", dt.date(2026, 10, 27), (w0, w1))291jun_log, jun_fails = summarize(RDB / "project-db/logs/extract.log", dt.date(2026, 6, 18), None)292# other October log files (small)293others = {}294for nm in ["spid.log", "spid_resume.log", "spid_post.log", "spid_interim.log"]:295 p = RDB / "data/logs" / nm296 if p.exists():297 txt = open(p, errors="replace").read()298 others[nm] = {"lines": txt.count("\n"), "head": txt[:1500]}299pend = RDB / "data/logs/spid_pending.json"300if pend.exists():301 try:302 pj = json.load(open(pend)); others["spid_pending.json"] = {k: (v if not isinstance(v, list) else v[:20]) for k, v in pj.items()} if isinstance(pj, dict) else str(pj)[:1500]303 except Exception as ex:304 others["spid_pending.json"] = str(ex)305# outage window measured from the log itself: the contiguous block of 'retryable' failures on 2026-10-28 between 01:00 and 07:00306retry_ts = [t for t, c in oct_fails if t and c == "retryable_http_429_5xx" and dt.datetime(2026,10,28,1,0) <= t <= dt.datetime(2026,10,28,7,0)]307oct_log["outage_window_from_log"] = [str(min(retry_ts)), str(max(retry_ts)), len(retry_ts)]308w0l, w1l = min(retry_ts), max(retry_ts)309oct_log["llm_call_failed_outside_log_window"] = [(str(t), c) for t, c in oct_fails if not (t and w0l <= t <= w1l)]310oct_log["silent_empty_results_outside_outage"] = len(oct_log["llm_call_failed_outside_log_window"])311oct_log["retryable_failures_outside_log_window"] = sum(1 for t, c in oct_fails if c == "retryable_http_429_5xx" and not (t and w0l <= t <= w1l))312oct_log["ledger_rows_in_log_window"] = con.execute(f"select count(*), sum(n_mentions) from ledger where extracted_at between TIMESTAMP '{w0l}' and TIMESTAMP '{w1l}'").fetchone()313oct_log["ledger_rows_in_largest_gap"] = con.execute(f"select count(*) from ledger where extracted_at > TIMESTAMP '{gaps[-1][1]}' and extracted_at < TIMESTAMP '{gaps[-1][2]}'").fetchone()[0]314pend2 = int(re.search(r"pending: (\d+)", oct_log["pending_lines"][1][1]).group(1)) if len(oct_log["pending_lines"]) > 1 else None315oct_log["outage_rows_reconciliation"] = {"extraction_complete_sum_sections": oct_log["extraction_complete_sum_sections"], "pending_at_second_start": pend2,316 "difference": (oct_log["extraction_complete_sum_sections"] - pend2) if pend2 else None, "paper_research_log_149352": 149352,317 "pending_at_resume": int(re.search(r"pending: (\d+)", oct_log["pending_lines"][-1][1]).group(1)),318 "rows_before_outage_plus_in_window": con.execute(f"select count(*) from ledger where run='october' and extracted_at < TIMESTAMP '{gaps[-1][2]}'").fetchone()[0]}319# batch-level productivity from the log (silent-failure test, THEO-05): mentions per section per batch320def batch_rows(rows_, run_):321 out = []322 for t, lvl, name, msg in rows_:323 m_ = re.search(r"BATCH (\d+)/(\d+) done \| batch_sec=(\d+) batch_men=(\d+)", msg or "")324 if m_:325 out.append((run_, int(m_.group(1)), int(m_.group(2)), str(t), int(m_.group(3)), int(m_.group(4)), round(int(m_.group(4))/max(1,int(m_.group(3))), 4)))326 return out327oct_rows_ = parse_log(RDB / "data/logs/spid_extract.log", dt.date(2026, 10, 27)); jun_rows_ = parse_log(RDB / "project-db/logs/extract.log", dt.date(2026, 6, 18))328br = batch_rows(oct_rows_, "october") + batch_rows(jun_rows_, "june")329import statistics330med_oct = statistics.median([b[6] for b in br if b[0] == "october"]); med_jun = statistics.median([b[6] for b in br if b[0] == "june"])331br2 = [b + (round(b[6]/(med_oct if b[0]=="october" else med_jun), 3), (b[6] < 0.25*(med_oct if b[0]=="october" else med_jun))) for b in br]332r.write_csv("batch_productivity_from_log.csv", ["run", "batch", "n_batches", "done_at", "sections", "mentions", "mentions_per_section", "ratio_to_median", "anomalous_lt_25pct_of_median"], br2)333oct_log["batch_productivity_median_mentions_per_section"] = med_oct334oct_log["anomalous_batches"] = [b for b in br2 if b[0] == "october" and b[8]]335jun_log["batch_productivity_median_mentions_per_section"] = med_jun336jun_log["anomalous_batches"] = [b for b in br2 if b[0] == "june" and b[8]]337# the first resumed batch: rows stamped between its start and end, with the expected productive count at the form x item rates of the rest of the October run338b1 = [b for b in br2 if b[0] == "october" and b[1] == 1 and b[2] == 81]339if b1:340 b1_end = b1[0][3]341 b1_start = str(dt.datetime.fromisoformat(oct_log["pending_lines"][-1][0]))342 con.execute(f"CREATE TEMP TABLE b1 AS SELECT * FROM ledger WHERE run='october' AND extracted_at BETWEEN TIMESTAMP '{b1_start}' AND TIMESTAMP '{b1_end}'")343 r.dump("resume_batch1_vs_expected.csv", con, """344 WITH rates AS (SELECT form_type, section_name, count(*) FILTER (WHERE n_mentions>0)*1.0/nullif(count(*) FILTER (WHERE gate AND word_count>=30),0) p_prod, sum(n_mentions)*1.0/nullif(count(*) FILTER (WHERE gate AND word_count>=30),0) m_per_call345 FROM ledger WHERE run='october' AND section_id NOT IN (SELECT section_id FROM b1) GROUP BY 1,2)346 SELECT b.form_type, b.section_name AS item, count(*) rows, count(*) FILTER (WHERE gate AND word_count>=30) calls, count(*) FILTER (WHERE n_mentions>0) with_mentions, sum(n_mentions) mentions,347 round(sum(CASE WHEN gate AND word_count>=30 THEN r.p_prod ELSE 0 END),1) expected_with_mentions, round(sum(CASE WHEN gate AND word_count>=30 THEN r.m_per_call ELSE 0 END),1) expected_mentions,348 count(DISTINCT b.cik) ciks349 FROM b1 b LEFT JOIN rates r ON b.form_type=r.form_type AND b.section_name=r.section_name GROUP BY 1,2 ORDER BY 1,2""")350 oct_log["resume_batch1"] = {"window": [b1_start, b1_end], "log_sections": b1[0][4], "log_mentions": b1[0][5],351 "ledger_rows": con.execute("select count(*) from b1").fetchone()[0],352 "ledger_calls": con.execute("select count(*) from b1 where gate and word_count>=30").fetchone()[0],353 "ledger_with_mentions": con.execute("select count(*) from b1 where n_mentions>0").fetchone()[0],354 "ledger_mentions": con.execute("select sum(n_mentions) from b1").fetchone()[0],355 "ciks": con.execute("select list(distinct cik order by cik) from b1").fetchone()[0],356 "expected_with_mentions": con.execute("""WITH rates AS (SELECT form_type, section_name, count(*) FILTER (WHERE n_mentions>0)*1.0/nullif(count(*) FILTER (WHERE gate AND word_count>=30),0) p FROM ledger WHERE run='october' AND section_id NOT IN (SELECT section_id FROM b1) GROUP BY 1,2)357 SELECT round(sum(CASE WHEN gate AND word_count>=30 THEN r.p ELSE 0 END),1) FROM b1 b LEFT JOIN rates r ON b.form_type=r.form_type AND b.section_name=r.section_name""").fetchone()[0],358 "failed_call_lines_in_window": sum(1 for t, c in oct_fails if t and dt.datetime.fromisoformat(b1_start) <= t <= dt.datetime.fromisoformat(b1_end))}359r.write_json("log_summary.json", {"october": oct_log, "june": jun_log, "other_logs": others,360 "june_log_date_assumed": "2026-06-18 (file mtime; the log carries time-of-day only)"})361r.write_csv("log_failures_by_hour.csv", ["run", "hour", "reason", "n"],362 [("october", h, c, n) for (h, c), n in sorted(Counter((t.strftime("%Y-%m-%d %H") if t else "?", c) for t, c in oct_fails).items())] +363 [("june", h, c, n) for (h, c), n in sorted(Counter((t.strftime("%Y-%m-%d %H") if t else "?", c) for t, c in jun_fails).items())])364365# ---- Table 3 replacement input ---------------------------------------------------------------366r.dump("table3_ledger_vs_stored.csv", con, """367SELECT form_type, count(*) AS ledger_rows, count(*) FILTER (WHERE gate AND word_count>=30) AS calls,368 count(*) FILTER (WHERE n_mentions>0) AS with_mentions, sum(n_mentions) AS returned_by_model_before_dedup,369 sum(stored_mentions) AS stored_mentions, count(DISTINCT cik) AS issuers,370 count(DISTINCT cik) FILTER (WHERE n_mentions>0) AS issuers_with_mentions371FROM ledger GROUP BY 1 ORDER BY 1""")372r.notes["issuers_total_distinct_in_ledger"] = con.execute("select count(distinct cik) from ledger").fetchone()[0]373r.notes["issuers_with_filings"] = con.execute("select count(distinct cik) from src_filings").fetchone()[0]374r.finish()375