#!/usr/bin/env python3
"""
Field completeness in Veridion's extracted congressional transaction records,
2014-2025. Produces the PNG posted to r/dataisbeautiful.

This script is the reproducibility artifact. The counts below are frozen from a
single read-only measurement against Veridion's production warehouse at
2026-09-10 20:44:52 UTC. The extraction query that produced them is in
QUERY_FOR_COUNTS. Every rate on the chart is computed here from those
numerators and denominators, so nothing is hand-typed as a percentage.

    python3 plot_field_completeness.py

Requires matplotlib. Writes congress-field-completeness-2026-09-10.png
alongside itself.

WHAT THIS MEASURES, STATED PRECISELY
------------------------------------
The denominator is rows Veridion collected AND admitted to its serving path.
It already excludes documents and rows our pipeline failed to collect or failed
to admit. A rate here is therefore a property of *our extraction* of the public
record. It is not a measure of what the filings contain, and it is not a claim
about what any other parser could read.

A blank field is one of two things and these counts cannot tell them apart:
the field was blank on the filed document, or our parser did not read it.

"All four" tests the amount FLOOR (amount_low), not a complete range. That is
deliberate: the top disclosure band is "Over $50,000,000" and has no upper
bound on the form, so requiring one would penalise a correctly completed
filing. On this subset 49,916 rows carry a floor, 49,913 also carry an upper
bound, and the 3 that do not include the 2 rows in that open-ended band.

No cause is assigned to any movement in these series. Field presence can change
with asset mix, chamber mix, collection coverage and parser behaviour, and this
data does not separate them.
"""

from __future__ import annotations

import matplotlib

matplotlib.use("Agg")
import matplotlib.pyplot as plt  # noqa: E402
from matplotlib.ticker import FuncFormatter  # noqa: E402

MEASURED_AT = "2026-09-10 20:44:52 UTC"

# The serving-admission predicate. Printed against the alias `warehouse`; the
# physical relation name is withheld because a build gate refuses bare
# warehouse-relation references outside reviewed files. The predicate is what
# you need to check the logic.
ADMISSION_PREDICATE = """
where source in ('pdf', 'official-senate-efd')
  and superseded_at is null
  and nullif(trim(member_slug), '') is not null
  and nullif(trim(member), '')      is not null
  and nullif(trim(doc_id), '')      is not null
  and transaction_date is not null
  and filing_date      is not null
  and filing_date >= transaction_date
  and pdf_url ~ '^https://(disclosures-clerk[.]house[.]gov/.+[.]pdf([?].*)?|efdsearch[.]senate[.]gov/.+)$'
"""

QUERY_FOR_COUNTS = f"""
-- Numerators and denominators for every point on the chart.
with adm as (
  select *, extract(year from transaction_date)::int as yr
  from warehouse
  {ADMISSION_PREDICATE.strip()}
)
select yr as year,
  count(*)                                                                as rows_denominator,
  count(*) filter (where nullif(trim(coalesce(ticker, '')), '') is not null) as ticker_n,
  count(*) filter (where owner_type is not null)                          as owner_n,
  count(*) filter (where nullif(trim(coalesce(ticker, '')), '') is not null
                     and owner_type is not null
                     and amount_low is not null
                     and nullif(trim(coalesce(asset, '')), '') is not null) as all_four_n
from adm
where yr between 2014 and 2025
group by yr order by yr;

-- Corpus totals for the 2014-2025 subset. Documents and filers are DISTINCT
-- counts over the whole subset; they deliberately do not equal the sum of the
-- yearly counts, because one document and one filer can appear in more than
-- one transaction year. Run this to reproduce the three constants below.
with adm as (
  select *, extract(year from transaction_date)::int as yr
  from warehouse
  {ADMISSION_PREDICATE.strip()}
), sub as (select * from adm where yr between 2014 and 2025)
select count(*)                    as rows,       -- 49,916
       count(distinct doc_id)      as documents,  -- 6,133
       count(distinct member_slug) as filers      -- 351
from sub;

-- And the check that the distinct counts are NOT the yearly sums:
--   sum of yearly distinct documents = 6,310  (vs 6,133 distinct overall)
--   sum of yearly distinct filers    = 1,274  (vs   351 distinct overall)
with adm as (
  select *, extract(year from transaction_date)::int as yr
  from warehouse
  {ADMISSION_PREDICATE.strip()}
), sub as (select * from adm where yr between 2014 and 2025)
select sum(d) as sum_yearly_documents, sum(f) as sum_yearly_filers
from (select count(distinct doc_id) d, count(distinct member_slug) f
      from sub group by yr) per_year;

-- Amount-bound completeness, since "all four" tests the floor only:
with adm as (
  select *, extract(year from transaction_date)::int as yr
  from warehouse
  {ADMISSION_PREDICATE.strip()}
), sub as (select * from adm where yr between 2014 and 2025)
select count(*) filter (where amount_low is not null)                              as floor_present,   -- 49,916
       count(*) filter (where amount_low is not null and amount_high is not null)  as both_bounds,     -- 49,913
       count(*) filter (where amount_range = 'Over $50,000,000')                   as open_ended_band  --      2
from sub;
"""

# year: (rows_denominator, ticker_n, owner_n, all_four_n)
# Frozen 2026-09-10 20:44:52 UTC. The corpus grows daily; a later run of the
# query above will return larger denominators.
COUNTS: dict[int, tuple[int, int, int, int]] = {
    2014: (2125, 1315, 889, 664),
    2015: (3152, 2179, 1872, 1423),
    2016: (3393, 2199, 2021, 1439),
    2017: (3394, 2178, 1824, 1333),
    2018: (3826, 2358, 2240, 1609),
    2019: (5109, 3140, 2433, 1564),
    2020: (5355, 3596, 2993, 2030),
    2021: (4537, 3413, 2634, 1938),
    2022: (3638, 2898, 2217, 1804),
    2023: (4456, 4048, 2398, 2146),
    2024: (2917, 2593, 1995, 1810),
    2025: (8014, 7291, 4326, 3867),
}

TOTAL_ROWS = 49_916
DISTINCT_DOCUMENTS = 6_133          # NOT the sum of yearly counts, which is 6,310
DISTINCT_FILERS = 351               # NOT the sum of yearly counts, which is 1,274
SUM_OF_YEARLY_DOCUMENTS = 6_310
SUM_OF_YEARLY_FILERS = 1_274

# Amount-bound completeness on the 2014-2025 subset, measured with the third
# query above. "All four" tests the floor because the top band is open-ended.
AMOUNT_FLOOR_PRESENT = 49_916
AMOUNT_BOTH_BOUNDS = 49_913
OPEN_ENDED_BAND_ROWS = 2

OBSIDIAN = "#0d0f12"
IVORY = "#efe9dd"
GOLD = "#c9a227"
MUTED = "#8b8f96"
DIM = "#5a5f66"
RULE = "#22262b"


def main() -> None:
    assert sum(v[0] for v in COUNTS.values()) == TOTAL_ROWS, "denominators must sum to the stated total"
    for year, (denom, *nums) in COUNTS.items():
        for n in nums:
            assert 0 <= n <= denom, f"{year}: numerator {n} outside denominator {denom}"

    years = sorted(COUNTS)
    pct = lambda idx: [100.0 * COUNTS[y][idx] / COUNTS[y][0] for y in years]  # noqa: E731
    ticker, owner, all_four = pct(1), pct(2), pct(3)

    fig, ax = plt.subplots(figsize=(12, 7.2), dpi=180)
    fig.patch.set_facecolor(OBSIDIAN)
    ax.set_facecolor(OBSIDIAN)
    for spine in ax.spines.values():
        spine.set_visible(False)
    ax.grid(axis="y", color=RULE, linewidth=0.9, zorder=0)
    ax.set_axisbelow(True)

    series = [
        (ticker, GOLD, 3.0, "-", "Ticker symbol"),
        (owner, MUTED, 2.4, "-", "Owner type"),
        (all_four, DIM, 2.0, (0, (5, 3)), "All four fields"),
    ]
    for values, colour, width, style, _ in series:
        ax.plot(
            years, values, color=colour, lw=width, ls=style, marker="o", ms=5,
            markerfacecolor=OBSIDIAN, markeredgecolor=colour, markeredgewidth=1.8, zorder=5,
        )

    ax.set_ylim(0, 100)
    ax.set_xlim(2013.4, 2026.6)
    ax.set_yticks([0, 20, 40, 60, 80, 100])
    ax.yaxis.set_major_formatter(FuncFormatter(lambda v, _: f"{int(v)}%"))
    ax.tick_params(colors=MUTED, labelsize=11.5, length=0)
    ax.set_xticks(years)
    ax.set_xticklabels([str(y) for y in years])

    # Owner and all-four end within six points of each other, so label anchors
    # are placed on a fixed ladder rather than at the series value.
    label_y = {0: 91.0, 1: 60.0, 2: 44.0}
    for idx, (values, colour, _, _, label) in enumerate(series):
        weight = "bold" if colour == GOLD else "normal"
        y = label_y[idx]
        ax.annotate(label, xy=(2025.30, y + 2.4), color=colour, fontsize=12.5, fontweight=weight, va="center")
        ax.annotate(f"{values[-1]:.1f}%", xy=(2025.30, y - 2.4), color=colour, fontsize=11, va="center")
        ax.plot([2025, 2025.22], [values[-1], y], color=colour, lw=0.9, alpha=0.55, zorder=3)

    fig.text(
        0.048, 0.952,
        "Field completeness in Veridion's extracted congressional transaction records",
        color=IVORY, fontsize=16.2, fontweight="bold", va="top",
    )
    fig.text(
        0.048, 0.902,
        "Share of extracted rows where each field is populated, by year of transaction. "
        f"{TOTAL_ROWS:,} rows, {DISTINCT_DOCUMENTS:,} documents, {DISTINCT_FILERS} filers.",
        color=MUTED, fontsize=11.6, va="top",
    )

    fig.text(
        0.048, 0.030,
        "Source: Veridion Markets' own extraction of U.S. House Clerk periodic transaction reports and U.S. Senate eFD filings.\n"
        f"Measured {MEASURED_AT}. Denominator is rows we collected and admitted, so it already excludes what our pipeline missed;\n"
        "these are rates for our extraction, not for the filings themselves and not for what another parser could read. A blank field\n"
        "may be blank on the filing or unread by our parser, and this chart cannot separate those. No cause is assigned to any movement.\n"
        "Document URLs are matched by pattern; they are not checked for a live response. Counts and query: see the linked comment.",
        color=DIM, fontsize=9.4, va="bottom", linespacing=1.55,
    )

    plt.subplots_adjust(left=0.062, right=0.845, top=0.855, bottom=0.205)
    out = "congress-field-completeness-2026-09-10.png"
    fig.savefig(out, facecolor=OBSIDIAN)
    print(f"wrote {out}")
    print(f"rows {TOTAL_ROWS:,} | documents {DISTINCT_DOCUMENTS:,} | filers {DISTINCT_FILERS}")
    print(f"  documents/filers are DISTINCT over the subset; yearly sums are "
          f"{SUM_OF_YEARLY_DOCUMENTS:,} and {SUM_OF_YEARLY_FILERS:,} and are NOT the same thing")
    print(f"  amount floor present {AMOUNT_FLOOR_PRESENT:,} | both bounds {AMOUNT_BOTH_BOUNDS:,} "
          f"| open-ended top band {OPEN_ENDED_BAND_ROWS}")
    for y in years:
        d, t, o, a = COUNTS[y]
        print(f"{y}  n={d:>5}  ticker {t:>5} ({100*t/d:4.1f}%)  owner {o:>5} ({100*o/d:4.1f}%)  all4 {a:>5} ({100*a/d:4.1f}%)")


if __name__ == "__main__":
    main()