"""N4 - two runs, two 10-K segmenters, cohort confounding, fiscal_year audit, 10-K funnel (cluster D). Read-only."""
from __future__ import annotations
import json, sys
from pathlib import Path
import numpy as np
sys.path.insert(0, str(Path(__file__).resolve().parent))
from common import *

r = Run("runs")
con = connect(SPID_DB)
tmin = con.execute("select min(extracted_at) from extracted_sections").fetchone()[0]
con.execute(f"""CREATE TEMP TABLE led AS
SELECT e.section_id, e.accession_number, e.n_mentions, CASE WHEN e.extracted_at = TIMESTAMP '{tmin}' THEN 'backfilled_june' ELSE 'october' END AS run,
       f.form_type, f.cik, f.filing_date, f.fiscal_year AS filing_fiscal_year, co.in_sp500, co.in_russell3000
FROM extracted_sections e JOIN src_filings f ON f.accession_number = e.accession_number LEFT JOIN src_companies co ON f.cik = co.cik""")
con.execute("""CREATE TEMP TABLE secmeta AS
SELECT section_id, section_name, word_count, 'parsed' AS segmenter FROM src_sections
UNION ALL SELECT section_id, section_name, word_count, 'legacy_ingest' FROM spid_sections""")
con.execute("CREATE TEMP TABLE ledm AS SELECT l.*, s.section_name, s.word_count, s.segmenter FROM led l LEFT JOIN secmeta s USING (section_id)")

# ---- (b) legacy (ingest.py) vs parsed (src/parsing) text of the same 10-K items ----------------
MONEY = r"\$\s?\d[\d,]*(\.\d+)?\s?(billion|million|thousand|bn|mm)?"
YEAR = r"\b(19[89]\d|20[0-3]\d)\b"
feat = lambda src: f"""SELECT section_id, accession_number, section_name, word_count, length(text) AS nchar,
       length(text) - length(replace(text, '$', '')) AS dollar_tokens,
       len(regexp_extract_all(text, '{YEAR}')) AS year_tokens,
       len(regexp_extract_all(text, '{MONEY}')) AS money_tokens,
       len(regexp_extract_all(text, '\\b\\d[\\d,]*(\\.\\d+)?\\b')) AS numeric_tokens
FROM {src}"""
print("features: legacy", flush=True)
con.execute(f"CREATE TEMP TABLE leg AS {feat('spid_sections')}")
print("features: parsed 10-K of the same accessions", flush=True)
con.execute(f"CREATE TEMP TABLE par AS {feat('src_sections')} WHERE accession_number IN (SELECT DISTINCT accession_number FROM spid_sections)")
r.dump("segmenter_item_comparison.csv", con, """
WITH j AS (SELECT l.section_name, l.word_count lw, p.word_count pw, l.dollar_tokens ld, p.dollar_tokens pd, l.year_tokens ly, p.year_tokens py, l.money_tokens lm, p.money_tokens pm, l.numeric_tokens ln, p.numeric_tokens pn
           FROM leg l JOIN par p ON l.accession_number = p.accession_number AND l.section_name = p.section_name)
SELECT section_name, count(*) pairs,
       round(median(lw)) legacy_median_words, round(median(pw)) parsed_median_words,
       round(quantile_cont(lw*1.0/nullif(pw,0), 0.10),3) ratio_p10, round(quantile_cont(lw*1.0/nullif(pw,0), 0.50),3) ratio_p50, round(quantile_cont(lw*1.0/nullif(pw,0), 0.90),3) ratio_p90,
       round(avg(ld),1) legacy_mean_dollar, round(avg(pd),1) parsed_mean_dollar,
       round(avg(ly),1) legacy_mean_years, round(avg(py),1) parsed_mean_years,
       round(avg(lm),1) legacy_mean_money, round(avg(pm),1) parsed_mean_money,
       round(avg(ln),1) legacy_mean_numeric, round(avg(pn),1) parsed_mean_numeric,
       round(100.0*count(*) FILTER (WHERE lm < pm)/count(*),2) pct_legacy_fewer_money_tokens,
       round(100.0*count(*) FILTER (WHERE lm = 0 AND pm > 0)/count(*),2) pct_legacy_zero_money_parsed_positive,
       round(100.0*count(*) FILTER (WHERE ld < pd)/count(*),2) pct_legacy_fewer_dollar,
       round(100.0*count(*) FILTER (WHERE lw < 0.5*pw)/count(*),2) pct_legacy_lt_half_words,
       round(100.0*count(*) FILTER (WHERE lw > 2*pw)/count(*),2) pct_legacy_gt_double_words
FROM j GROUP BY 1 ORDER BY 2 DESC""")
S = {}
S["legacy_accessions"] = con.execute("select count(distinct accession_number) from spid_sections").fetchone()[0]
S["legacy_rows"] = con.execute("select count(*) from spid_sections").fetchone()[0]
S["legacy_accessions_with_parsed_counterpart"] = con.execute("select count(distinct accession_number) from par").fetchone()[0]
S["legacy_rows_matched_by_item"] = con.execute("select count(*) from leg l join par p on l.accession_number=p.accession_number and l.section_name=p.section_name").fetchone()[0]
S["legacy_section_names"] = con.execute("select section_name, count(*), round(median(word_count)) from spid_sections group by 1 order by 2 desc").fetchall()
S["parsed_section_names_same_accessions"] = con.execute("select section_name, count(*), round(median(word_count)) from par group by 1 order by 2 desc").fetchall()
S["legacy_items_per_accession"] = con.execute("select n, count(*) from (select accession_number, count(*) n from spid_sections group by 1) group by 1 order by 1").fetchall()
S["legacy_accessions_missing_item"] = con.execute("""select section_name, count(*) from (select a.accession_number, i.section_name from (select distinct accession_number from spid_sections) a cross join (select unnest(['business','risk_factors','properties','md_and_a','quantitative_disclosures']) section_name) i
   where not exists (select 1 from spid_sections s where s.accession_number=a.accession_number and s.section_name=i.section_name)) group by 1 order by 2 desc""").fetchall()

# ---- (c) full_body rows ----------------------------------------------------------------------------
S["full_body_rows"] = con.execute("select count(*), count(distinct accession_number), round(median(word_count)), min(word_count), max(word_count) from spid_sections where section_name='full_body'").fetchone()
S["full_body_accessions_with_other_items"] = con.execute("select count(distinct accession_number) from spid_sections where section_name='full_body' and accession_number in (select accession_number from spid_sections where section_name<>'full_body')").fetchone()[0]
S["full_body_ledger_yield"] = con.execute("""select count(*), count(*) filter (where n_mentions>0), sum(n_mentions), round(sum(n_mentions)*1.0/count(*),4), round(100.0*count(*) filter (where n_mentions>0)/count(*),2)
   from ledm where segmenter='legacy_ingest' and section_name='full_body'""").fetchone()
S["legacy_item_ledger_yield"] = con.execute("""select section_name, count(*), count(*) filter (where n_mentions>0), sum(n_mentions), round(sum(n_mentions)*1.0/count(*),4), round(100.0*count(*) filter (where n_mentions>0)/count(*),2), round(median(word_count))
   from ledm where segmenter='legacy_ingest' group by 1 order by 2 desc""").fetchall()
S["parsed_10k_item_ledger_yield"] = con.execute("""select section_name, count(*), count(*) filter (where n_mentions>0), sum(n_mentions), round(sum(n_mentions)*1.0/count(*),4), round(100.0*count(*) filter (where n_mentions>0)/count(*),2), round(median(word_count))
   from ledm where segmenter='parsed' and form_type='10-K' group by 1 order by 2 desc""").fetchall()
S["full_body_accessions_parsed_items_available"] = con.execute("""select count(distinct p.accession_number), count(*) from par p where p.accession_number in (select accession_number from spid_sections where section_name='full_body') and p.section_name in ('business','risk_factors','properties','md_and_a','quantitative_disclosures')""").fetchone()

# ---- (d) parser-fallback full_document 10-Ks never submitted ---------------------------------------------
S["full_document_10k"] = con.execute("""select count(*), count(distinct s.accession_number) from src_sections s join src_filings f using(accession_number) where f.form_type='10-K' and s.section_name='full_document'""").fetchone()
S["full_document_10k_in_legacy"] = con.execute("""select count(distinct s.accession_number) from src_sections s join src_filings f using(accession_number) where f.form_type='10-K' and s.section_name='full_document' and s.accession_number in (select accession_number from spid_sections)""").fetchone()[0]
S["full_document_10k_in_ledger"] = con.execute("""select count(*) from src_sections s join src_filings f using(accession_number) where f.form_type='10-K' and s.section_name='full_document' and s.section_id in (select section_id from extracted_sections)""").fetchone()[0]
S["full_document_10k_accessions_with_no_other_section"] = con.execute("""select count(*) from (select s.accession_number from src_sections s join src_filings f using(accession_number) where f.form_type='10-K' group by 1 having bool_and(s.section_name='full_document'))""").fetchone()[0]
S["full_document_10k_invisible_accessions"] = con.execute("""select count(*) from (select s.accession_number from src_sections s join src_filings f using(accession_number) where f.form_type='10-K' and s.accession_number not in (select accession_number from spid_sections) group by 1 having bool_and(s.section_name='full_document'))""").fetchone()[0]
S["full_document_by_form"] = con.execute("""select f.form_type, count(*), count(distinct s.accession_number), count(*) filter (where s.section_id in (select section_id from extracted_sections)) in_ledger from src_sections s join src_filings f using(accession_number) where s.section_name='full_document' group by 1""").fetchall()
S["sections_10k_by_item_all_parsed"] = con.execute("""select s.section_name, count(*), round(median(s.word_count)), count(*) filter (where s.word_count<30) from src_sections s join src_filings f using(accession_number) where f.form_type='10-K' group by 1 order by 2 desc""").fetchall()
r.write_json("segmenters.json", S)

# ---- (e) within S&P 500, within form: backfilled vs October ----------------------------------------------
r.dump("within_sp500_by_run_form.csv", con, """
SELECT run, form_type, count(*) rows, count(DISTINCT accession_number) filings, count(DISTINCT cik) issuers,
       count(*) FILTER (WHERE n_mentions>0) with_mentions, sum(n_mentions) mentions,
       round(100.0*count(*) FILTER (WHERE n_mentions>0)/count(*),2) productive_pct, round(sum(n_mentions)*1.0/count(*),4) mentions_per_row,
       round(median(word_count)) median_words, min(filing_date) first_filing, max(filing_date) last_filing
FROM ledm WHERE in_sp500 GROUP BY 1,2 ORDER BY 2,1""")
r.dump("within_sp500_10q_by_item_run.csv", con, """
SELECT section_name item, run, count(*) rows, count(*) FILTER (WHERE n_mentions>0) with_mentions, sum(n_mentions) mentions,
       round(100.0*count(*) FILTER (WHERE n_mentions>0)/count(*),2) productive_pct, round(sum(n_mentions)*1.0/count(*),4) mentions_per_row, round(median(word_count)) median_words
FROM ledm WHERE in_sp500 AND form_type='10-Q' GROUP BY 1,2 ORDER BY 1,2""")
con.execute("""CREATE TEMP TABLE pm AS SELECT m.*, l.run, l.in_sp500, l.in_russell3000 FROM project_mentions m JOIN led l USING (section_id)""")
r.dump("within_sp500_10q_mentions_by_run.csv", con, """
SELECT run, count(*) mentions, round(avg(confidence),4) mean_conf, round(median(confidence),3) median_conf,
       round(100.0*count(*) FILTER (WHERE confidence>=0.9)/count(*),2) pct_conf_ge_09, round(100.0*count(*) FILTER (WHERE amount_usd IS NOT NULL)/count(*),2) pct_with_amount,
       round(100.0*count(*) FILTER (WHERE len(locations)>0)/count(*),2) pct_with_location, round(100.0*count(*) FILTER (WHERE len(partners)>0)/count(*),2) pct_with_partner,
       round(100.0*count(*) FILTER (WHERE len(years)>0)/count(*),2) pct_with_years, round(avg(length(description)),1) mean_desc_chars
FROM pm WHERE in_sp500 AND form_type='10-Q' GROUP BY 1""")
r.dump("within_sp500_10q_type_mix_by_run.csv", con, """
SELECT project_type, count(*) FILTER (WHERE run='backfilled_june') june_n, count(*) FILTER (WHERE run='october') oct_n,
       round(100.0*count(*) FILTER (WHERE run='backfilled_june')/sum(count(*) FILTER (WHERE run='backfilled_june')) OVER (),2) june_pct,
       round(100.0*count(*) FILTER (WHERE run='october')/sum(count(*) FILTER (WHERE run='october')) OVER (),2) oct_pct
FROM pm WHERE in_sp500 AND form_type='10-Q' GROUP BY 1 ORDER BY 2+3 DESC""")
r.dump("mentions_by_run_form_all.csv", con, """
SELECT run, form_type, count(*) mentions, round(avg(confidence),4) mean_conf, round(100.0*count(*) FILTER (WHERE amount_usd IS NOT NULL)/count(*),2) pct_with_amount,
       round(100.0*count(*) FILTER (WHERE len(locations)>0)/count(*),2) pct_with_location, round(100.0*count(*) FILTER (WHERE fiscal_year IS NULL)/count(*),2) pct_fiscal_year_null
FROM pm GROUP BY 1,2 ORDER BY 2,1""")
# paired within-issuer comparison (10-Q, S&P 500 issuers present in both runs)
pairs = con.execute("""
WITH a AS (SELECT cik, run, count(*) rows, count(*) FILTER (WHERE n_mentions>0) prod, sum(n_mentions) men FROM ledm WHERE in_sp500 AND form_type='10-Q' GROUP BY 1,2)
SELECT j.cik, j.rows, j.prod, j.men, o.rows, o.prod, o.men FROM a j JOIN a o ON j.cik=o.cik AND j.run='backfilled_june' AND o.run='october'""").fetchall()
P = {"issuers_paired": len(pairs)}
if pairs:
    d_prod = np.array([o_p/o_r - j_p/j_r for _, j_r, j_p, j_m, o_r, o_p, o_m in pairs])
    d_men = np.array([o_m/o_r - j_m/j_r for _, j_r, j_p, j_m, o_r, o_p, o_m in pairs])
    P["mean_diff_productive_share_oct_minus_june"] = float(d_prod.mean()); P["se"] = float(d_prod.std(ddof=1)/np.sqrt(len(d_prod)))
    P["mean_diff_mentions_per_row_oct_minus_june"] = float(d_men.mean()); P["se_m"] = float(d_men.std(ddof=1)/np.sqrt(len(d_men)))
    P["june_mean_productive_share"] = float(np.mean([j_p/j_r for _, j_r, j_p, *_ in pairs])); P["oct_mean_productive_share"] = float(np.mean([o_p/o_r for _, j_r, j_p, j_m, o_r, o_p, o_m in pairs]))
r.write_json("within_issuer_paired_10q.json", P)

# ---- (f) Table 6 within cohort + cohort dummy regression ------------------------------------------------
con.execute("""CREATE TEMP TABLE _m2 AS SELECT *, lower(coalesce(try(locations[1]), '')) AS loc1,
   regexp_extract(lower(regexp_replace(project_name, '[^A-Za-z ]', '', 'g')), '([a-z]{4,})', 1) AS name_tok FROM pm""")
con.execute("""CREATE TEMP TABLE k2 AS SELECT *, cik || '|' || project_type || '|' || CASE WHEN loc1 <> '' THEN loc1 ELSE coalesce(nullif(name_tok,''),'general') END AS gkey FROM _m2""")
grp = "CASE WHEN in_sp500 AND in_russell3000 THEN 'both' WHEN in_sp500 THEN 'sp500_only' WHEN in_russell3000 THEN 'russell_only' ELSE 'none' END"
r.dump("table6_cohort.csv", con, f"""
WITH allm AS (SELECT {grp} g, count(DISTINCT gkey) projects, count(DISTINCT cik) issuers, count(*) mentions FROM k2 GROUP BY 1),
     octm AS (SELECT {grp} g, count(DISTINCT gkey) projects, count(DISTINCT cik) issuers, count(*) mentions FROM k2 WHERE run='october' GROUP BY 1),
     oct10q8k AS (SELECT {grp} g, count(DISTINCT gkey) projects, count(DISTINCT cik) issuers, count(*) mentions FROM k2 WHERE run='october' AND form_type IN ('10-Q','8-K') GROUP BY 1),
     ledr AS (SELECT {grp} g, count(*) ledger_rows, count(*) FILTER (WHERE run='october') october_rows, count(DISTINCT cik) issuers_ledger FROM ledm GROUP BY 1)
SELECT a.g cohort, a.issuers, a.projects projects_all, round(a.projects*1.0/a.issuers,1) per_issuer_all,
       o.projects projects_october_only, round(o.projects*1.0/o.issuers,1) per_issuer_october_only,
       q.projects projects_oct_10q_8k, round(q.projects*1.0/q.issuers,1) per_issuer_oct_10q_8k,
       l.ledger_rows, l.october_rows, round(l.ledger_rows*1.0/l.issuers_ledger,1) ledger_rows_per_issuer, round(l.october_rows*1.0/l.issuers_ledger,1) october_rows_per_issuer
FROM allm a JOIN octm o USING (g) JOIN oct10q8k q USING (g) JOIN ledr l USING (g) ORDER BY 1""")
iss = con.execute(f"""
WITH p AS (SELECT cik, count(DISTINCT gkey) projects, count(DISTINCT gkey) FILTER (WHERE run='october') projects_oct FROM k2 GROUP BY 1),
     l AS (SELECT cik, any_value(in_sp500) sp, count(*) rows, count(*) FILTER (WHERE run='october') rows_oct, count(DISTINCT accession_number) filings FROM ledm GROUP BY 1)
SELECT l.cik, l.sp::int, coalesce(p.projects,0), coalesce(p.projects_oct,0), l.rows, l.rows_oct, l.filings FROM l LEFT JOIN p USING (cik)""").fetchall()
A = np.array([[x[1], x[2], x[3], x[4], x[5], x[6]] for x in iss], dtype=float)
def ols(y, X, names):
    X = np.column_stack([np.ones(len(y)), X]); b, *_ = np.linalg.lstsq(X, y, rcond=None); e = y - X @ b
    XtX_inv = np.linalg.inv(X.T @ X); meat = (X * (e**2)[:, None]).T @ X; V = XtX_inv @ meat @ XtX_inv * len(y)/(len(y)-X.shape[1])
    se = np.sqrt(np.diag(V)); return {n: {"coef": float(bb), "hc1_se": float(s)} for n, bb, s in zip(["const"]+names, b, se)} | {"n": int(len(y)), "r2": float(1 - (e**2).sum()/((y-y.mean())**2).sum())}
REG = {}
REG["projects_all_on_sp500"] = ols(A[:,1], A[:,[0]], ["in_sp500"])
REG["projects_october_only_on_sp500"] = ols(A[:,2], A[:,[0]], ["in_sp500"])
REG["projects_all_on_sp500_log_rows"] = ols(A[:,1], np.column_stack([A[:,0], np.log1p(A[:,3])]), ["in_sp500", "log1p_ledger_rows"])
REG["projects_october_only_on_sp500_log_rows_oct"] = ols(A[:,2], np.column_stack([A[:,0], np.log1p(A[:,4])]), ["in_sp500", "log1p_october_rows"])
REG["log_projects_all_on_sp500_log_rows"] = ols(np.log1p(A[:,1]), np.column_stack([A[:,0], np.log1p(A[:,3])]), ["in_sp500", "log1p_ledger_rows"])
REG["log_projects_oct_on_sp500_log_rows_oct"] = ols(np.log1p(A[:,2]), np.column_stack([A[:,0], np.log1p(A[:,4])]), ["in_sp500", "log1p_october_rows"])
REG["note"] = "issuer-level OLS, HC1 standard errors; 'october only' counts projects whose key is touched by at least one October-run mention; in_sp500 = cohort dummy (S&P 500 issuers were extracted in two runs and with two 10-K segmenters)"
r.write_json("table6_cohort_regression.json", REG)

# ---- (g) fiscal_year audit and derived fill ------------------------------------------------------------------
G = {}
G["mentions_fiscal_year_null_by_run_form"] = con.execute("select run, form_type, count(*) filter (where fiscal_year is null), count(*) from pm group by 1,2 order by 2,1").fetchall()
G["src_filings_fiscal_year_null_by_form"] = con.execute("select form_type, count(*) filter (where fiscal_year is null), count(*) filter (where report_date is null), count(*) from src_filings group by 1").fetchall()
G["null_mentions_fillable_from_src_filings"] = con.execute("""select count(*) filter (where f.fiscal_year is not null), count(*) filter (where f.fiscal_year is null and f.report_date is not null), count(*)
   from pm m join src_filings f using(accession_number) where m.fiscal_year is null""").fetchone()
G["ledger_rows_filing_fiscal_year_null_by_run"] = con.execute("select run, count(*) filter (where filing_fiscal_year is null), count(*) from ledm group by 1").fetchall()
G["fill_rule"] = "fiscal_year_fill = coalesce(src_filings.fiscal_year, year(report_date), year(filing_date)); derived, not stored"
r.dump("fiscal_year_fill_by_accession.csv", con, """
SELECT DISTINCT m.accession_number, m.form_type, f.filing_date, f.report_date, f.fiscal_year AS fiscal_year_src,
       coalesce(f.fiscal_year, year(f.report_date), year(f.filing_date)) AS fiscal_year_fill,
       CASE WHEN f.fiscal_year IS NOT NULL THEN 'src_filings' WHEN f.report_date IS NOT NULL THEN 'year(report_date)' ELSE 'year(filing_date)' END AS fill_source
FROM pm m JOIN src_filings f USING (accession_number) WHERE m.fiscal_year IS NULL ORDER BY 1""")
G["mentions_null_by_fill_source"] = con.execute("""select case when f.fiscal_year is not null then 'src_filings' when f.report_date is not null then 'year(report_date)' else 'year(filing_date)' end s, count(*)
   from pm m join src_filings f using(accession_number) where m.fiscal_year is null group by 1""").fetchall()
G["mentions_fiscal_year_vs_filing_fiscal_year_mismatch_nonnull"] = con.execute("""select count(*) from pm m join src_filings f using(accession_number) where m.fiscal_year is not null and f.fiscal_year is not null and m.fiscal_year <> f.fiscal_year""").fetchone()[0]
r.write_json("fiscal_year_audit.json", G)

# ---- (h) 10-K funnel ----------------------------------------------------------------------------------------
H = {}
H["parsed_10k_rows_total"] = con.execute("select count(*) from src_sections s join src_filings f using(accession_number) where f.form_type='10-K'").fetchone()[0]
H["parsed_10k_rows_out_of_scope_item"] = con.execute("select count(*) from src_sections s join src_filings f using(accession_number) where f.form_type='10-K' and s.section_name not in ('business','risk_factors','properties','md_and_a','quantitative_disclosures')").fetchone()[0]
H["parsed_10k_rows_in_scope"] = H["parsed_10k_rows_total"] - H["parsed_10k_rows_out_of_scope_item"]
H["parsed_10k_in_scope_rows_of_legacy_accessions"] = con.execute("select count(*) from src_sections s join src_filings f using(accession_number) where f.form_type='10-K' and s.section_name in ('business','risk_factors','properties','md_and_a','quantitative_disclosures') and s.accession_number in (select accession_number from spid_sections)").fetchone()[0]
H["parsed_10k_in_scope_rows_lt30_words_non_legacy"] = con.execute("select count(*) from src_sections s join src_filings f using(accession_number) where f.form_type='10-K' and s.section_name in ('business','risk_factors','properties','md_and_a','quantitative_disclosures') and s.accession_number not in (select accession_number from spid_sections) and s.word_count < 30").fetchone()[0]
H["parsed_10k_in_scope_rows_ge30_non_legacy"] = con.execute("select count(*) from src_sections s join src_filings f using(accession_number) where f.form_type='10-K' and s.section_name in ('business','risk_factors','properties','md_and_a','quantitative_disclosures') and s.accession_number not in (select accession_number from spid_sections) and s.word_count >= 30").fetchone()[0]
H["parsed_10k_in_scope_rows_ge30_non_legacy_in_ledger"] = con.execute("select count(*) from src_sections s join src_filings f using(accession_number) where f.form_type='10-K' and s.section_name in ('business','risk_factors','properties','md_and_a','quantitative_disclosures') and s.accession_number not in (select accession_number from spid_sections) and s.word_count >= 30 and s.section_id in (select section_id from extracted_sections)").fetchone()[0]
H["legacy_rows"] = con.execute("select count(*) from spid_sections").fetchone()[0]
H["legacy_rows_in_ledger"] = con.execute("select count(*) from spid_sections where section_id in (select section_id from extracted_sections)").fetchone()[0]
H["legacy_rows_lt30"] = con.execute("select count(*) from spid_sections where word_count<30").fetchone()[0]
H["ledger_10k_rows"] = con.execute("select count(*) from ledm where form_type='10-K'").fetchone()[0]
H["ledger_10k_by_segmenter"] = con.execute("select segmenter, count(*) from ledm where form_type='10-K' group by 1").fetchall()
H["funnel"] = f"{H['parsed_10k_rows_total']} parsed 10-K rows - {H['parsed_10k_rows_out_of_scope_item']} out-of-scope items - {H['parsed_10k_in_scope_rows_of_legacy_accessions']} rows of legacy-ingested filings - {H['parsed_10k_in_scope_rows_lt30_words_non_legacy']} rows < 30 words = {H['parsed_10k_in_scope_rows_ge30_non_legacy']} (ledgered: {H['parsed_10k_in_scope_rows_ge30_non_legacy_in_ledger']}); + {H['legacy_rows_in_ledger']} legacy rows = {H['parsed_10k_in_scope_rows_ge30_non_legacy_in_ledger'] + H['legacy_rows_in_ledger']} vs ledger 10-K rows {H['ledger_10k_rows']}"
H["expected_ledger_10k"] = 143704
# 10-Q items in scope (all seven submitted?)
H["ledger_10q_items"] = con.execute("select section_name, count(*) from ledm where form_type='10-Q' group by 1 order by 2 desc").fetchall()
H["parsed_10q_items"] = con.execute("select s.section_name, count(*), count(*) filter (where s.word_count<30) from src_sections s join src_filings f using(accession_number) where f.form_type='10-Q' group by 1 order by 2 desc").fetchall()
H["ledger_8k_items"] = con.execute("select section_name, count(*) from ledm where form_type='8-K' group by 1 order by 2 desc").fetchall()
r.write_json("funnel_10k.json", H)
r.dump("legacy_vs_parsed_sp500_items.csv", con, """
SELECT 'legacy_ingest' segmenter, section_name, count(*) n, round(median(word_count)) median_words, count(*) FILTER (WHERE word_count<30) lt30 FROM spid_sections GROUP BY 1,2
UNION ALL
SELECT 'parsed_sp500_10k' segmenter, s.section_name, count(*), round(median(s.word_count)), count(*) FILTER (WHERE s.word_count<30) FROM src_sections s JOIN src_filings f USING (accession_number) JOIN src_companies co ON f.cik=co.cik WHERE f.form_type='10-K' AND co.in_sp500 GROUP BY 1,2
UNION ALL
SELECT 'parsed_nonsp500_10k' segmenter, s.section_name, count(*), round(median(s.word_count)), count(*) FILTER (WHERE s.word_count<30) FROM src_sections s JOIN src_filings f USING (accession_number) JOIN src_companies co ON f.cik=co.cik WHERE f.form_type='10-K' AND NOT co.in_sp500 GROUP BY 1,2
ORDER BY 1,3 DESC""")
r.finish()