| 6 |
6 |
Both connections are opened read_only=True; threads=4; nothing is written under the repository. |
| 7 |
7 |
--skip-slow omits the candidate-gate scan over the whole ledger (gate_pass_total / calls_by_form) and the bootstrap. |
| 8 |
8 |
""" |
| 9 |
|
−import csv, json, random, re, statistics, sys |
|
9 |
+import csv, json, random, re, shutil, statistics, sys, tempfile |
| 10 |
10 |
from datetime import timedelta |
| 11 |
11 |
from pathlib import Path |
| 12 |
12 |
import duckdb |
| 18 |
18 |
root = Path(args[0]); out = Path(args[1]); out.mkdir(parents=True, exist_ok=True) |
| 19 |
19 |
edgar = duckdb.connect(str(root / "data/duckdb/edgar_research.duckdb"), read_only=True) |
| 20 |
20 |
spid = duckdb.connect(str(root / "project-db/data/db/spid.duckdb"), read_only=True) |
|
21 |
+TMP = tempfile.mkdtemp(prefix="spid_export_stats_") # DuckDB spill directory: never next to the databases |
| 21 |
22 |
for con in (edgar, spid): |
| 22 |
|
− con.execute("SET threads=4"); con.execute("SET memory_limit='16GB'") |
|
23 |
+ con.execute("SET threads=4"); con.execute("SET memory_limit='16GB'"); con.execute(f"SET temp_directory='{TMP}'") |
| 23 |
24 |
# the src_* views stored in spid.duckdb point to the master node's parquet corpus (valid here) |
| 24 |
25 |
S = {} |
| 25 |
26 |
|
| 43 |
44 |
return cast(m.group(1)) if m else None |
| 44 |
45 |
|
| 45 |
46 |
|
|
47 |
+def grep_max(path, pattern): |
|
48 |
+ """Largest integer captured by the regex group over the whole log (cumulative counters), or None.""" |
|
49 |
+ p = root / path |
|
50 |
+ if not p.exists(): return None |
|
51 |
+ vals = [int(m.group(1)) for m in re.finditer(pattern, p.read_text(errors="replace"))] |
|
52 |
+ return max(vals) if vals else None |
|
53 |
+ |
|
54 |
+ |
| 46 |
55 |
def grep_count(path, pattern): |
| 47 |
56 |
p = root / path |
| 48 |
57 |
if not p.exists(): return None |
| 106 |
115 |
gap = spid.execute("""with t as (select distinct extracted_at ts from extracted_sections where extracted_at > (select min(extracted_at) from extracted_sections)), |
| 107 |
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() |
| 108 |
117 |
S["outage"] = {"start": str(gap[0]), "end": str(gap[1]), "hours": round(gap[2], 2)} if gap else None |
| 109 |
|
−wall_h = (t1 - t0).total_seconds() / 3600.0 |
|
118 |
+# the October run starts at its first flush, i.e. the first timestamp after the backfilled block (min(extracted_at) is the backfill instant) |
|
119 |
+t_oct0 = one(spid, "select min(extracted_at) from extracted_sections where extracted_at > (select min(extracted_at) from extracted_sections)") |
|
120 |
+S["october_first_flush"] = str(t_oct0) |
|
121 |
+wall_h = (t1 - t_oct0).total_seconds() / 3600.0 |
| 110 |
122 |
S["october_wall_clock_hours"] = round(wall_h, 2) |
| 111 |
123 |
S["october_effective_sections_per_second"] = round(S["october_rows"] / ((wall_h - (gap[2] if gap else 0.0)) * 3600.0), 1) |
| 112 |
124 |
S["conf_exact_default_count"] = one(spid, "select count(*) from project_mentions where round(confidence,3) = 0.600") |
| 113 |
125 |
S["conf_min_by_run"] = spid.execute("""select min(m.confidence) filter (where e.extracted_at = (select min(extracted_at) from extracted_sections)), |
| 114 |
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() |
| 115 |
|
−S["amount_raw_deviations"] = one(spid, "select count(*) from project_mentions where amount_usd is not null and amount_raw <> '$' || format('{:,.0f}', amount_usd)") |
|
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-blind |
|
128 |
+S["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)") |
| 116 |
129 |
dump("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) overwritten |
| 117 |
130 |
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""") |
| 118 |
131 |
dump("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, |
| 123 |
136 |
"sp500_run": {"log": "project-db/logs/extract.log", "date": grep_first("project-db/logs/extract.log", r"(\d{4}-\d{2}-\d{2})"), |
| 124 |
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), |
| 125 |
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), |
| 126 |
|
− "sections_logged": grep_first("project-db/logs/extract.log", r"sections=(\d+)", int), "ledger_rows_backfilled": S["backfilled_rows"], |
|
139 |
+ "sections_logged": grep_max("project-db/logs/extract.log", r"cum_sections=(\d+)"), "ledger_rows_backfilled": S["backfilled_rows"], |
| 127 |
140 |
"mentions_backfilled": S["backfilled_mentions"], "llm_call_failed": grep_count("project-db/logs/extract.log", r"LLM call failed")}, |
| 128 |
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})"), |
| 129 |
142 |
"workers": grep_first("data/logs/spid_extract.log", r"workers=(\d+)", int), "ledger_rows": S["october_rows"], |
| 151 |
164 |
dump("projects_by_first_seen_year", spid, "select extract(year from first_seen)::int yr, count(*) n from projects group by 1 order by 1") |
| 152 |
165 |
dump("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)") |
| 153 |
166 |
dump("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)") |
| 154 |
|
−dump("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") |
|
167 |
+dump("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") |
| 155 |
168 |
dump("confidence_dist", spid, "select round(confidence,1) conf, count(*) n from project_mentions group by 1 order by 1") |
| 156 |
169 |
dump("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") |
| 157 |
170 |
S["amount_total_median_musd"] = one(spid, "select round(median(total_amount_usd)/1e6,1) from projects where total_amount_usd is not null") |
| 161 |
174 |
S["amount_le_zero"] = one(spid, "select count(*) from projects where total_amount_usd <= 0") |
| 162 |
175 |
dump("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") |
| 163 |
176 |
dump("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") |
| 164 |
|
−dump("top_issuers", spid, "select ticker, company_name, sector, count(*) projects from projects group by 1,2,3 order by 4 desc limit 15") |
|
177 |
+dump("top_issuers", spid, "select ticker, company_name, sector, count(*) projects from projects group by 1,2,3 order by 4 desc, 1 limit 15") |
| 165 |
178 |
dump("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") |
| 166 |
179 |
S["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() |
| 167 |
180 |
S["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() |
| 181 |
194 |
|
| 182 |
195 |
json.dump(S, open(out / "stats.json", "w"), indent=2, default=str) |
| 183 |
196 |
print(json.dumps(S, indent=1, default=str)[:3000]) |
|
197 |
+shutil.rmtree(TMP, ignore_errors=True) |