#!/usr/bin/env python3 """Verify and compare two pinned WebTruffle US federal grants snapshots. The recipe uses only the Python 3.11+ standard library. It downloads the tagged 2026-07-22 and 2026-08-11 manifests and CSV assets, verifies their exact byte counts and SHA-256 digests while streaming, checks the complete 43-column schema, and writes deterministic analysis products: * agency-summary.csv * applicant-label-matches.csv * funding-estimate-watchlist.csv * posted-deadline-watchlist.csv * pipeline-changes.csv * query-provenance.json Funding figures are source-reported planning estimates, not awards, payments, obligations, or guarantees. Candidate sets still require review of the complete official announcement at source_url. SPDX-License-Identifier: MIT """ from __future__ import annotations import argparse import csv import hashlib import json import os import platform import sys import tempfile from collections import Counter, defaultdict from dataclasses import dataclass from datetime import date from decimal import ROUND_HALF_UP, Decimal, InvalidOperation from pathlib import Path from typing import Any, Iterable, Mapping, Sequence from urllib.request import Request, urlopen if sys.version_info < (3, 11): raise RuntimeError("This recipe requires Python 3.11 or newer.") REPOSITORY = "webtruffle/us-federal-grants" REPOSITORY_URL = f"https://github.com/{REPOSITORY}" SOURCE_LICENSE = "https://www.grants.gov/api/terms-conditions" SOURCE_NOTICE = ( "This product uses the Grants.gov API but is not endorsed or certified by " "the U.S. Department of Health and Human Services." ) USER_AGENT = "WebTruffle-federal-grants-Python-recipe/1.0 (+https://www.webtruffle.com/)" CHUNK_BYTES = 1024 * 1024 EXPECTED_HEADERS = ( "id", "source", "source_opportunity_id", "record_type", "opportunity_number", "title", "agency_code", "agency_name", "opportunity_category_code", "opportunity_category", "opportunity_status", "status_basis", "version", "posted_date", "close_date", "archive_date", "forecasted_post_date", "forecasted_close_date", "estimated_award_date", "estimated_project_start_date", "fiscal_year", "last_updated_date", "extract_date", "estimated_total_program_funding_usd", "expected_number_of_awards", "award_ceiling_usd", "award_ceiling_unlimited", "award_floor_usd", "cost_sharing_required", "funding_instrument_codes", "funding_instrument_types", "funding_activity_category_codes", "funding_activity_categories", "assistance_listing_numbers", "eligible_applicant_codes", "eligible_applicant_types", "additional_information_url", "source_url", "source_license", "first_seen_at", "last_seen_at", "content_hash", "change_type", ) if len(EXPECTED_HEADERS) != 43: raise RuntimeError("The pinned federal-grants schema must contain exactly 43 fields.") JSON_ARRAY_FIELDS = ( "funding_instrument_codes", "funding_instrument_types", "funding_activity_category_codes", "funding_activity_categories", "assistance_listing_numbers", "eligible_applicant_codes", "eligible_applicant_types", ) @dataclass(frozen=True) class ReleaseSpec: tag: str manifest_bytes: int manifest_sha256: str csv_bytes: int csv_sha256: str records: int posted: int forecasted: int closing_within_7_days: int closing_within_30_days: int @property def release_url(self) -> str: return f"{REPOSITORY_URL}/releases/tag/{self.tag}" @property def asset_base_url(self) -> str: return f"{REPOSITORY_URL}/releases/download/{self.tag}" RELEASES = ( ReleaseSpec( tag="2026-07-22", manifest_bytes=4_625, manifest_sha256="0555203a7723ea7da586bc8a2e7c9317d4ac9d4134bfc02195b35cc4d4b55db7", csv_bytes=1_987_836, csv_sha256="c70ab8bdddea9c8724317dd6f3d949d8b1a6f105aa3ce30e29c7cc605e50b844", records=1_750, posted=1_251, forecasted=499, closing_within_7_days=120, closing_within_30_days=406, ), ReleaseSpec( tag="2026-08-11", manifest_bytes=4_627, manifest_sha256="d92af401bf7cc547457c595ef2c57a28fda2091c62d836af1541248e22603cb5", csv_bytes=1_951_757, csv_sha256="3759daf0fca66d8708f6c8dcbc786b48b8c41a715b3144316438055ccd383d4a", records=1_691, posted=1_152, forecasted=539, closing_within_7_days=111, closing_within_30_days=313, ), ) EXPECTED_ORACLES: dict[str, int] = { "start_records": 1_750, "start_posted": 1_251, "start_forecasted": 499, "end_records": 1_691, "end_posted": 1_152, "end_forecasted": 539, "retained": 1_484, "entered_snapshot": 207, "exited_snapshot": 266, "posted_to_posted": 995, "forecasted_to_forecasted": 459, "forecasted_to_posted": 30, "posted_to_forecasted": 0, "end_agencies": 192, "end_nih_records": 684, "end_health_records": 853, "end_selected_applicant_type_label_matches": 763, "end_funding_estimates_reported": 952, "end_cost_sharing_required": 126, "end_discretionary": 1_629, "end_effective_deadline_0_7": 111, "end_effective_deadline_0_30": 313, "end_forecast_deadline_0_30": 12, "end_posted_deadline_0_30": 301, "pipeline_changes": 503, } AGENCY_SUMMARY_FIELDS = ( "edition_date", "agency_code", "agency_name", "records", "posted", "forecasted", "selected_applicant_type_label_matches", "funding_estimate_reported", "funding_estimate_coverage", "cost_sharing_required", "posted_closing_0_30_days", "source_license", ) CANDIDATE_FIELDS = ( "edition_date", "id", "source", "source_opportunity_id", "opportunity_number", "title", "agency_code", "agency_name", "opportunity_status", "posted_date", "close_date", "forecasted_post_date", "forecasted_close_date", "estimated_award_date", "estimated_project_start_date", "estimated_total_program_funding_usd", "expected_number_of_awards", "award_ceiling_usd", "award_ceiling_unlimited", "award_floor_usd", "cost_sharing_required", "funding_instrument_types", "funding_activity_categories", "assistance_listing_numbers", "eligible_applicant_codes", "eligible_applicant_types", "source_url", "source_license", ) DEADLINE_FIELDS = ( "edition_date", "days_to_close", *CANDIDATE_FIELDS[1:], ) SNAPSHOT_CHANGE_FIELDS = ( "comparison_start", "comparison_end", "change_type", "id", "source", "source_opportunity_id", "opportunity_number", "title", "agency_code", "agency_name", "start_status", "end_status", "start_close_date", "end_close_date", "start_forecasted_close_date", "end_forecasted_close_date", "start_estimated_total_program_funding_usd", "end_estimated_total_program_funding_usd", "source_url", "source_license", "start_source_url", "end_source_url", ) OUTPUT_FILES = ( "agency-summary.csv", "applicant-label-matches.csv", "funding-estimate-watchlist.csv", "posted-deadline-watchlist.csv", "pipeline-changes.csv", ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Download, verify, and compare the pinned 2026-07-22 and 2026-08-11 " "WebTruffle US federal grants snapshots." ) ) parser.add_argument( "--data-dir", type=Path, default=Path("federal-grants-data-2026-07-22-to-2026-08-11"), help="Directory for verified inputs and deterministic outputs.", ) parser.add_argument( "--applicant-type", type=nonempty_label, default="Small businesses", help=( "Structured eligible-applicant label to select. Matching is an exact " "label comparison after trimming and case folding." ), ) return parser.parse_args() def nonempty_label(value: str) -> str: label = value.strip() if not label: raise argparse.ArgumentTypeError("applicant type must not be empty") return label def sha256_and_size(path: Path) -> tuple[str, int]: digest = hashlib.sha256() size = 0 with path.open("rb") as source: for chunk in iter(lambda: source.read(CHUNK_BYTES), b""): digest.update(chunk) size += len(chunk) return digest.hexdigest(), size def assert_file( path: Path, expected_sha256: str, expected_bytes: int, ) -> dict[str, int | str]: actual_sha256, actual_bytes = sha256_and_size(path) if actual_bytes != expected_bytes or actual_sha256 != expected_sha256: raise RuntimeError( f"Verification failed for {path}: expected {expected_bytes} bytes / " f"{expected_sha256}, got {actual_bytes} bytes / {actual_sha256}." ) return {"bytes": actual_bytes, "sha256": actual_sha256} def download_verified( url: str, destination: Path, expected_sha256: str, expected_bytes: int, ) -> dict[str, int | str]: destination.parent.mkdir(parents=True, exist_ok=True) if destination.exists(): evidence = assert_file(destination, expected_sha256, expected_bytes) print(f"verified existing {destination.name:<26} {expected_bytes:>9,} bytes") return evidence request = Request(url, headers={"User-Agent": USER_AGENT, "Accept": "*/*"}) descriptor, temporary_name = tempfile.mkstemp( dir=destination.parent, prefix=f".{destination.name}.", suffix=".part", ) partial = Path(temporary_name) descriptor_open = True try: downloaded_bytes = 0 digest = hashlib.sha256() with ( urlopen(request, timeout=120) as response, os.fdopen(descriptor, "wb") as target, ): descriptor_open = False while chunk := response.read(CHUNK_BYTES): downloaded_bytes += len(chunk) if downloaded_bytes > expected_bytes: raise RuntimeError( f"Download exceeded the pinned {expected_bytes} bytes for " f"{destination.name}." ) digest.update(chunk) target.write(chunk) target.flush() os.fsync(target.fileno()) actual_sha256 = digest.hexdigest() if downloaded_bytes != expected_bytes or actual_sha256 != expected_sha256: raise RuntimeError( f"Verification failed for downloaded {destination.name}: expected " f"{expected_bytes} bytes / {expected_sha256}, got {downloaded_bytes} " f"bytes / {actual_sha256}." ) os.replace(partial, destination) finally: if descriptor_open: try: os.close(descriptor) except OSError: pass partial.unlink(missing_ok=True) print(f"downloaded + verified {destination.name:<26} {expected_bytes:>9,} bytes") return {"bytes": expected_bytes, "sha256": expected_sha256} def validate_manifest(spec: ReleaseSpec, manifest: Mapping[str, Any]) -> None: expected_scalars = { "schema_version": "1.0", "dataset_id": "us-federal-grants", "target_date": spec.tag, "record_count": spec.records, "record_grain": "One official federal funding opportunity per source ID.", } observed_scalars = {key: manifest.get(key) for key in expected_scalars} if observed_scalars != expected_scalars: raise RuntimeError( f"Manifest scalar mismatch for {spec.tag}: expected {expected_scalars}, " f"got {observed_scalars}." ) expected_statuses = {"posted": spec.posted, "forecasted": spec.forecasted} if manifest.get("status_counts") != expected_statuses: raise RuntimeError( f"Manifest status mismatch for {spec.tag}: {manifest.get('status_counts')!r}." ) expected_deadlines = { "closing_within_7_days": spec.closing_within_7_days, "closing_within_30_days": spec.closing_within_30_days, } if manifest.get("deadline_counts") != expected_deadlines: raise RuntimeError( f"Manifest deadline mismatch for {spec.tag}: " f"{manifest.get('deadline_counts')!r}." ) if manifest.get("record_fields") != list(EXPECTED_HEADERS): raise RuntimeError( f"Manifest schema mismatch for {spec.tag}; expected the pinned 43 fields " "in their declared order." ) csv_declaration = manifest.get("files", {}).get("funding-opportunities.csv") expected_csv_declaration = { "path": f"daily/{spec.tag}/funding-opportunities.csv", "bytes": spec.csv_bytes, "sha256": spec.csv_sha256, } if csv_declaration != expected_csv_declaration: raise RuntimeError( f"Manifest CSV declaration mismatch for {spec.tag}: {csv_declaration!r}." ) if manifest.get("source_licenses") != {"grants_gov": SOURCE_LICENSE}: raise RuntimeError( f"Manifest source-license mismatch for {spec.tag}: " f"{manifest.get('source_licenses')!r}." ) if manifest.get("warnings") != []: raise RuntimeError( f"Pinned manifest for {spec.tag} contains warnings: {manifest.get('warnings')!r}." ) def parse_json_string_list(row: Mapping[str, str], field: str) -> list[str]: raw = row[field] try: value = json.loads(raw) except json.JSONDecodeError as error: raise RuntimeError( f"Invalid JSON array in {field} for opportunity " f"{row['source_opportunity_id']}: {error}." ) from error if not isinstance(value, list) or any(not isinstance(item, str) for item in value): raise RuntimeError( f"Expected a string array in {field} for opportunity " f"{row['source_opportunity_id']}." ) return value def parse_iso_date(value: str, field: str, opportunity_id: str) -> date | None: if value == "": return None try: return date.fromisoformat(value) except ValueError as error: raise RuntimeError( f"Invalid ISO date in {field} for opportunity {opportunity_id}: {value!r}." ) from error def parse_decimal(value: str, field: str, opportunity_id: str) -> Decimal | None: if value == "": return None try: return Decimal(value) except InvalidOperation as error: raise RuntimeError( f"Invalid decimal in {field} for opportunity {opportunity_id}: {value!r}." ) from error def parse_integer(value: str, field: str, opportunity_id: str) -> int | None: if value == "": return None try: return int(value) except ValueError as error: raise RuntimeError( f"Invalid integer in {field} for opportunity {opportunity_id}: {value!r}." ) from error def validate_rows(spec: ReleaseSpec, rows: Sequence[dict[str, str]]) -> None: if len(rows) != spec.records: raise RuntimeError( f"Row-count mismatch for {spec.tag}: expected {spec.records}, got {len(rows)}." ) ids = [row["id"] for row in rows] source_ids = [row["source_opportunity_id"] for row in rows] if len(ids) != len(set(ids)): raise RuntimeError(f"Duplicate stable id values in {spec.tag}.") if len(source_ids) != len(set(source_ids)): raise RuntimeError(f"Duplicate source_opportunity_id values in {spec.tag}.") expected_statuses = Counter(posted=spec.posted, forecasted=spec.forecasted) observed_statuses = Counter(row["opportunity_status"] for row in rows) if observed_statuses != expected_statuses: raise RuntimeError( f"CSV status mismatch for {spec.tag}: expected {expected_statuses}, " f"got {observed_statuses}." ) for row in rows: opportunity_id = row["source_opportunity_id"] if set(row) != set(EXPECTED_HEADERS): raise RuntimeError( f"Row-level schema mismatch in {spec.tag} for opportunity {opportunity_id}." ) if row["source"] != "grants_gov": raise RuntimeError(f"Unexpected source in {spec.tag}: {row['source']!r}.") if row["extract_date"] != spec.tag: raise RuntimeError( f"Unexpected extract_date for opportunity {opportunity_id}: " f"{row['extract_date']!r}." ) if row["status_basis"] != "derived_from_snapshot_dates": raise RuntimeError( f"Unexpected status basis for opportunity {opportunity_id}: " f"{row['status_basis']!r}." ) if row["source_license"] != SOURCE_LICENSE: raise RuntimeError( f"Missing or unexpected source license for opportunity {opportunity_id}." ) expected_source_url = ( f"https://www.grants.gov/search-results-detail/{opportunity_id}" ) if row["source_url"] != expected_source_url: raise RuntimeError( f"Unexpected source URL for opportunity {opportunity_id}: " f"{row['source_url']!r}." ) for field in JSON_ARRAY_FIELDS: parse_json_string_list(row, field) for field in ( "posted_date", "close_date", "archive_date", "forecasted_post_date", "forecasted_close_date", "estimated_award_date", "estimated_project_start_date", "last_updated_date", "extract_date", ): parse_iso_date(row[field], field, opportunity_id) for field in ( "estimated_total_program_funding_usd", "award_ceiling_usd", "award_floor_usd", ): parse_decimal(row[field], field, opportunity_id) for field in ("expected_number_of_awards", "fiscal_year"): parse_integer(row[field], field, opportunity_id) def load_release( data_dir: Path, spec: ReleaseSpec, ) -> tuple[dict[str, Any], list[dict[str, str]], dict[str, dict[str, int | str]]]: release_dir = data_dir / "inputs" / spec.tag manifest_path = release_dir / "manifest.json" csv_path = release_dir / "funding-opportunities.csv" manifest_evidence = download_verified( f"{spec.asset_base_url}/manifest.json", manifest_path, spec.manifest_sha256, spec.manifest_bytes, ) manifest = json.loads(manifest_path.read_text(encoding="utf-8")) validate_manifest(spec, manifest) csv_evidence = download_verified( f"{spec.asset_base_url}/funding-opportunities.csv", csv_path, spec.csv_sha256, spec.csv_bytes, ) with csv_path.open("r", encoding="utf-8-sig", newline="") as source: reader = csv.DictReader(source) if tuple(reader.fieldnames or ()) != EXPECTED_HEADERS: raise RuntimeError( f"CSV header mismatch for {spec.tag}; expected the pinned 43 columns " "in their declared order." ) rows = list(reader) validate_rows(spec, rows) return manifest, rows, { "manifest.json": manifest_evidence, "funding-opportunities.csv": csv_evidence, } def has_list_value(row: Mapping[str, str], field: str, expected: str) -> bool: return expected in parse_json_string_list(row, field) def normalize_label(value: str) -> str: return value.strip().casefold() def has_applicant_type(row: Mapping[str, str], normalized_label: str) -> bool: return any( normalize_label(label) == normalized_label for label in parse_json_string_list(row, "eligible_applicant_types") ) def days_from(edition: date, value: str, field: str, opportunity_id: str) -> int | None: parsed = parse_iso_date(value, field, opportunity_id) return None if parsed is None else (parsed - edition).days def candidate_projection(row: Mapping[str, str], edition_date: str) -> dict[str, str]: result = {field: row[field] for field in CANDIDATE_FIELDS if field != "edition_date"} return {"edition_date": edition_date, **result} def build_applicant_type_candidates( rows: Sequence[dict[str, str]], edition_date: str, normalized_applicant_type: str, ) -> list[dict[str, str]]: candidates = [ candidate_projection(row, edition_date) for row in rows if has_applicant_type(row, normalized_applicant_type) ] candidates.sort( key=lambda row: ( 0 if row["opportunity_status"] == "posted" else 1, row["close_date"] or row["forecasted_close_date"] or "9999-12-31", row["agency_name"].casefold(), row["source_opportunity_id"], ) ) return candidates def build_funding_estimate_watchlist( rows: Sequence[dict[str, str]], edition_date: str, ) -> list[dict[str, str]]: watchlist = [ candidate_projection(row, edition_date) for row in rows if row["estimated_total_program_funding_usd"] != "" ] watchlist.sort( key=lambda row: ( -parse_decimal( row["estimated_total_program_funding_usd"], "estimated_total_program_funding_usd", row["source_opportunity_id"], ), row["agency_name"].casefold(), row["source_opportunity_id"], ) ) return watchlist def build_posted_deadline_watchlist( rows: Sequence[dict[str, str]], edition_date: str, ) -> list[dict[str, str | int]]: edition = date.fromisoformat(edition_date) watchlist: list[dict[str, str | int]] = [] for row in rows: if row["opportunity_status"] != "posted": continue days = days_from( edition, row["close_date"], "close_date", row["source_opportunity_id"], ) if days is None or not 0 <= days <= 30: continue watchlist.append( { "edition_date": edition_date, "days_to_close": days, **{ field: row[field] for field in CANDIDATE_FIELDS if field != "edition_date" }, } ) watchlist.sort( key=lambda row: ( int(row["days_to_close"]), str(row["close_date"]), str(row["agency_code"]), str(row["source_opportunity_id"]), str(row["opportunity_number"]), ) ) return watchlist def build_agency_summary( rows: Sequence[dict[str, str]], edition_date: str, normalized_applicant_type: str, ) -> list[dict[str, str | int]]: edition = date.fromisoformat(edition_date) groups: dict[tuple[str, str, str], Counter[str]] = defaultdict(Counter) for row in rows: key = (row["agency_code"], row["agency_name"], row["source_license"]) metrics = groups[key] metrics["records"] += 1 metrics[row["opportunity_status"]] += 1 if has_applicant_type(row, normalized_applicant_type): metrics["selected_applicant_type_label_matches"] += 1 if row["estimated_total_program_funding_usd"] != "": metrics["funding_estimate_reported"] += 1 if row["cost_sharing_required"] == "True": metrics["cost_sharing_required"] += 1 if row["opportunity_status"] == "posted": days = days_from( edition, row["close_date"], "close_date", row["source_opportunity_id"], ) if days is not None and 0 <= days <= 30: metrics["posted_closing_0_30_days"] += 1 summary = [] for (agency_code, agency_name, source_license), metrics in groups.items(): coverage = ( Decimal(metrics["funding_estimate_reported"]) / Decimal(metrics["records"]) ).quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP) summary.append({ "edition_date": edition_date, "agency_code": agency_code, "agency_name": agency_name, "records": metrics["records"], "posted": metrics["posted"], "forecasted": metrics["forecasted"], "selected_applicant_type_label_matches": metrics[ "selected_applicant_type_label_matches" ], "funding_estimate_reported": metrics["funding_estimate_reported"], "funding_estimate_coverage": format(coverage, ".4f"), "cost_sharing_required": metrics["cost_sharing_required"], "posted_closing_0_30_days": metrics["posted_closing_0_30_days"], "source_license": source_license, }) summary.sort( key=lambda row: ( -int(row["records"]), str(row["agency_name"]).casefold(), str(row["agency_code"]), ) ) return summary def build_snapshot_changes( start_rows: Sequence[dict[str, str]], end_rows: Sequence[dict[str, str]], start_tag: str, end_tag: str, ) -> list[dict[str, str]]: start_by_id = {row["source_opportunity_id"]: row for row in start_rows} end_by_id = {row["source_opportunity_id"]: row for row in end_rows} changes: list[dict[str, str]] = [] for source_id in sorted(set(start_by_id) | set(end_by_id)): start = start_by_id.get(source_id) end = end_by_id.get(source_id) if start is None: change_type = "entered_snapshot" elif end is None: change_type = "exited_snapshot" elif start["opportunity_status"] != end["opportunity_status"]: change_type = ( f"{start['opportunity_status']}_to_{end['opportunity_status']}" ) else: continue preferred = end or start if preferred is None: raise RuntimeError(f"Missing both snapshots for source ID {source_id}.") if start is not None and end is not None and start["id"] != end["id"]: raise RuntimeError(f"Stable id changed for source opportunity {source_id}.") if start is not None and end is not None: if start["source_license"] != end["source_license"]: raise RuntimeError(f"Source license changed for opportunity {source_id}.") changes.append( { "comparison_start": start_tag, "comparison_end": end_tag, "change_type": change_type, "id": preferred["id"], "source": preferred["source"], "source_opportunity_id": source_id, "opportunity_number": preferred["opportunity_number"], "title": preferred["title"], "agency_code": preferred["agency_code"], "agency_name": preferred["agency_name"], "start_status": "" if start is None else start["opportunity_status"], "end_status": "" if end is None else end["opportunity_status"], "start_close_date": "" if start is None else start["close_date"], "end_close_date": "" if end is None else end["close_date"], "start_forecasted_close_date": ( "" if start is None else start["forecasted_close_date"] ), "end_forecasted_close_date": ( "" if end is None else end["forecasted_close_date"] ), "start_estimated_total_program_funding_usd": ( "" if start is None else start["estimated_total_program_funding_usd"] ), "end_estimated_total_program_funding_usd": ( "" if end is None else end["estimated_total_program_funding_usd"] ), "source_url": preferred["source_url"], "source_license": preferred["source_license"], "start_source_url": "" if start is None else start["source_url"], "end_source_url": "" if end is None else end["source_url"], } ) change_order = { "entered_snapshot": 0, "exited_snapshot": 1, "forecasted_to_posted": 2, } changes.sort( key=lambda row: ( change_order.get(row["change_type"], 99), row["source_opportunity_id"], ) ) return changes def effective_deadline_count( rows: Sequence[dict[str, str]], edition_date: str, maximum_days: int, ) -> tuple[int, int, int]: edition = date.fromisoformat(edition_date) posted = 0 forecasted = 0 for row in rows: field = ( "close_date" if row["opportunity_status"] == "posted" else "forecasted_close_date" ) days = days_from(edition, row[field], field, row["source_opportunity_id"]) if days is None or not 0 <= days <= maximum_days: continue if row["opportunity_status"] == "posted": posted += 1 else: forecasted += 1 return posted + forecasted, posted, forecasted def compute_oracles( start_rows: Sequence[dict[str, str]], end_rows: Sequence[dict[str, str]], agency_summary: Sequence[Mapping[str, Any]], applicant_type_candidates: Sequence[Mapping[str, Any]], funding_estimate_watchlist: Sequence[Mapping[str, Any]], deadline_watchlist: Sequence[Mapping[str, Any]], pipeline_changes: Sequence[Mapping[str, Any]], ) -> dict[str, int]: start_by_id = {row["source_opportunity_id"]: row for row in start_rows} end_by_id = {row["source_opportunity_id"]: row for row in end_rows} start_ids = set(start_by_id) end_ids = set(end_by_id) transitions = Counter( ( start_by_id[source_id]["opportunity_status"], end_by_id[source_id]["opportunity_status"], ) for source_id in start_ids & end_ids ) effective_7, _, _ = effective_deadline_count(end_rows, RELEASES[1].tag, 7) effective_30, posted_30, forecast_30 = effective_deadline_count( end_rows, RELEASES[1].tag, 30 ) observed = { "start_records": len(start_rows), "start_posted": sum( row["opportunity_status"] == "posted" for row in start_rows ), "start_forecasted": sum( row["opportunity_status"] == "forecasted" for row in start_rows ), "end_records": len(end_rows), "end_posted": sum(row["opportunity_status"] == "posted" for row in end_rows), "end_forecasted": sum( row["opportunity_status"] == "forecasted" for row in end_rows ), "retained": len(start_ids & end_ids), "entered_snapshot": len(end_ids - start_ids), "exited_snapshot": len(start_ids - end_ids), "posted_to_posted": transitions[("posted", "posted")], "forecasted_to_forecasted": transitions[("forecasted", "forecasted")], "forecasted_to_posted": transitions[("forecasted", "posted")], "posted_to_forecasted": transitions[("posted", "forecasted")], "end_agencies": len(agency_summary), "end_nih_records": sum( row["agency_name"] == "National Institutes of Health" for row in end_rows ), "end_health_records": sum( has_list_value(row, "funding_activity_categories", "Health") for row in end_rows ), "end_selected_applicant_type_label_matches": len(applicant_type_candidates), "end_funding_estimates_reported": len(funding_estimate_watchlist), "end_cost_sharing_required": sum( row["cost_sharing_required"] == "True" for row in end_rows ), "end_discretionary": sum( row["opportunity_category"] == "Discretionary" for row in end_rows ), "end_effective_deadline_0_7": effective_7, "end_effective_deadline_0_30": effective_30, "end_forecast_deadline_0_30": forecast_30, "end_posted_deadline_0_30": posted_30, "pipeline_changes": len(pipeline_changes), } if len(deadline_watchlist) != posted_30: raise RuntimeError( "Deadline builder and independent effective-deadline calculation disagree: " f"{len(deadline_watchlist)} != {posted_30}." ) return observed def assert_oracles( observed: Mapping[str, int], expected: Mapping[str, int], ) -> None: if dict(observed) != dict(expected): differences = { key: {"expected": expected.get(key), "observed": observed.get(key)} for key in sorted(set(expected) | set(observed)) if expected.get(key) != observed.get(key) } raise RuntimeError(f"Pinned regression-oracle failure: {differences}") def validate_output_rows( fieldnames: Sequence[str], rows: Iterable[Mapping[str, Any]], ) -> None: expected = set(fieldnames) for index, row in enumerate(rows, start=1): if set(row) != expected: raise RuntimeError( f"Output row {index} has unexpected fields. Expected {fieldnames}, " f"got {tuple(row)}." ) def write_csv( destination: Path, fieldnames: Sequence[str], rows: Sequence[Mapping[str, Any]], ) -> dict[str, int | str]: validate_output_rows(fieldnames, rows) with destination.open("w", encoding="utf-8", newline="") as target: writer = csv.DictWriter( target, fieldnames=fieldnames, extrasaction="raise", lineterminator="\n", ) writer.writeheader() writer.writerows(rows) target.flush() os.fsync(target.fileno()) digest, size = sha256_and_size(destination) return {"rows": len(rows), "bytes": size, "sha256": digest} def write_json(destination: Path, value: Mapping[str, Any]) -> None: with destination.open("w", encoding="utf-8", newline="\n") as target: json.dump(value, target, indent=2, sort_keys=True) target.write("\n") target.flush() os.fsync(target.fileno()) def reassert_inputs( data_dir: Path, ) -> dict[str, dict[str, dict[str, int | str]]]: evidence: dict[str, dict[str, dict[str, int | str]]] = {} for spec in RELEASES: release_dir = data_dir / "inputs" / spec.tag evidence[spec.tag] = { "manifest.json": assert_file( release_dir / "manifest.json", spec.manifest_sha256, spec.manifest_bytes, ), "funding-opportunities.csv": assert_file( release_dir / "funding-opportunities.csv", spec.csv_sha256, spec.csv_bytes, ), } return evidence def acquire_promotion_lock(data_dir: Path) -> tuple[int, Path]: lock_path = data_dir / ".federal-grants-results.lock" try: descriptor = os.open( lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600, ) except FileExistsError as error: raise RuntimeError( f"Result promotion is locked by {lock_path}. If no recipe process is " "running, inspect the existing outputs before removing this stale lock." ) from error try: os.write(descriptor, b"federal-grants-data-python result promotion\n") os.fsync(descriptor) except BaseException: os.close(descriptor) lock_path.unlink(missing_ok=True) raise return descriptor, lock_path def release_promotion_lock(descriptor: int, lock_path: Path) -> None: os.close(descriptor) lock_path.unlink(missing_ok=True) def main() -> None: args = parse_args() data_dir = args.data_dir.expanduser().resolve() data_dir.mkdir(parents=True, exist_ok=True) manifests: dict[str, dict[str, Any]] = {} releases: dict[str, list[dict[str, str]]] = {} initial_input_evidence: dict[str, dict[str, dict[str, int | str]]] = {} for spec in RELEASES: manifest, rows, evidence = load_release(data_dir, spec) manifests[spec.tag] = manifest releases[spec.tag] = rows initial_input_evidence[spec.tag] = evidence start_spec, end_spec = RELEASES start_rows = releases[start_spec.tag] end_rows = releases[end_spec.tag] normalized_applicant_type = normalize_label(args.applicant_type) applicant_type_candidates = build_applicant_type_candidates( end_rows, end_spec.tag, normalized_applicant_type, ) if not applicant_type_candidates: raise RuntimeError( f"Applicant-type filter {args.applicant_type!r} matched zero records. " "Use an exact structured label from eligible_applicant_types." ) matched_source_labels = sorted( { label.strip() for row in end_rows for label in parse_json_string_list(row, "eligible_applicant_types") if normalize_label(label) == normalized_applicant_type }, key=lambda label: (label.casefold(), label), ) agency_summary = build_agency_summary( end_rows, end_spec.tag, normalized_applicant_type, ) funding_estimate_watchlist = build_funding_estimate_watchlist( end_rows, end_spec.tag ) deadline_watchlist = build_posted_deadline_watchlist(end_rows, end_spec.tag) pipeline_changes = build_snapshot_changes( start_rows, end_rows, start_spec.tag, end_spec.tag, ) observed_oracles = compute_oracles( start_rows, end_rows, agency_summary, applicant_type_candidates, funding_estimate_watchlist, deadline_watchlist, pipeline_changes, ) expected_oracles = dict(EXPECTED_ORACLES) if normalized_applicant_type != normalize_label("Small businesses"): expected_oracles["end_selected_applicant_type_label_matches"] = len( applicant_type_candidates ) assert_oracles(observed_oracles, expected_oracles) with tempfile.TemporaryDirectory( dir=data_dir, prefix=".federal-grants-results.", ) as temporary_directory: staging_dir = Path(temporary_directory) outputs = { "agency-summary.csv": write_csv( staging_dir / "agency-summary.csv", AGENCY_SUMMARY_FIELDS, agency_summary, ), "applicant-label-matches.csv": write_csv( staging_dir / "applicant-label-matches.csv", CANDIDATE_FIELDS, applicant_type_candidates, ), "funding-estimate-watchlist.csv": write_csv( staging_dir / "funding-estimate-watchlist.csv", CANDIDATE_FIELDS, funding_estimate_watchlist, ), "posted-deadline-watchlist.csv": write_csv( staging_dir / "posted-deadline-watchlist.csv", DEADLINE_FIELDS, deadline_watchlist, ), "pipeline-changes.csv": write_csv( staging_dir / "pipeline-changes.csv", SNAPSHOT_CHANGE_FIELDS, pipeline_changes, ), } expected_output_rows = { "agency-summary.csv": 192, "applicant-label-matches.csv": len(applicant_type_candidates), "funding-estimate-watchlist.csv": 952, "posted-deadline-watchlist.csv": 301, "pipeline-changes.csv": 503, } observed_output_rows = { name: int(evidence["rows"]) for name, evidence in outputs.items() } if observed_output_rows != expected_output_rows: raise RuntimeError( f"Output row-count failure: expected {expected_output_rows}, " f"got {observed_output_rows}." ) final_input_evidence = reassert_inputs(data_dir) if final_input_evidence != initial_input_evidence: raise RuntimeError("Input evidence changed during analysis.") recipe_path = Path(__file__).resolve() recipe_sha256, recipe_bytes = sha256_and_size(recipe_path) license_path = recipe_path.with_suffix(".LICENSE.txt") license_evidence: dict[str, int | str] | None = None if license_path.exists(): license_sha256, license_bytes = sha256_and_size(license_path) license_evidence = { "name": license_path.name, "bytes": license_bytes, "sha256": license_sha256, } provenance = { "recipe": { "name": "federal-grants-data-python", "version": "1.0", "file": { "name": recipe_path.name, "bytes": recipe_bytes, "sha256": recipe_sha256, }, "license_file": license_evidence, }, "runtime": { "implementation": platform.python_implementation(), "python_version": platform.python_version(), "platform": platform.platform(), }, "source": { "repository": REPOSITORY_URL, "license": SOURCE_LICENSE, "required_notice": SOURCE_NOTICE, }, "comparison": { "start_tag": start_spec.tag, "end_tag": end_spec.tag, "elapsed_calendar_days": ( date.fromisoformat(end_spec.tag) - date.fromisoformat(start_spec.tag) ).days, "change_definition": ( "Stable source_opportunity_id population entries, exits, and " "opportunity_status transitions between the two complete active " "snapshots; same-status retained rows are omitted." ), }, "filters": { "applicant_type": { "requested_label": args.applicant_type, "normalized_label": normalized_applicant_type, "match_rule": ( "Case-insensitive exact-label match after trimming leading " "and trailing whitespace from both values." ), "matched_source_labels": matched_source_labels, "matched_records": len(applicant_type_candidates), } }, "releases": { spec.tag: { "release_url": spec.release_url, "target_date": manifests[spec.tag]["target_date"], "generated_at": manifests[spec.tag]["generated_at"], "record_grain": manifests[spec.tag]["record_grain"], "record_count": spec.records, "status_counts": manifests[spec.tag]["status_counts"], "deadline_counts": manifests[spec.tag]["deadline_counts"], "source_extract": manifests[spec.tag]["source_extract"], "inputs": final_input_evidence[spec.tag], } for spec in RELEASES }, "schema_check": { "header_count": len(EXPECTED_HEADERS), "headers": list(EXPECTED_HEADERS), "manifests_match": True, "csv_headers_match": True, }, "regression_oracles": { "expected": expected_oracles, "observed": observed_oracles, "passed": True, }, "outputs": outputs, "interpretation": { "agency_summary": ( "Counts from the 2026-08-11 active snapshot; agencies are keyed " "by the source agency code and name. funding_estimate_coverage is " "the source-reported estimate count divided by agency records, " "written as a deterministic four-decimal fraction. The selected " "applicant metric uses the recorded applicant_type filter." ), "applicant_type_label_matches": ( "Records whose structured eligible_applicant_types array matches " "the recorded applicant_type filter; this is not an exclusivity " "or eligibility determination." ), "funding_estimate_watchlist": ( "Records with a source-reported estimated total program funding " "value. Values are planning estimates, not awards or payments, and " "are not summed by this recipe." ), "posted_deadline_watchlist": ( "Posted records with a close_date from 0 through 30 calendar days " "after the 2026-08-11 edition date, inclusive." ), "snapshot_exits": ( "exited_snapshot means present in the 2026-07-22 active snapshot " "and absent from the 2026-08-11 active snapshot. It does not by " "itself prove closure, cancellation, archival, or removal." ), "source_review": ( "Every row-level output preserves source_url and source_license. " "Review the full official announcement before relying on a date, " "amount, or eligibility code." ), }, } provenance_path = staging_dir / "query-provenance.json" write_json(provenance_path, provenance) # Every destination is written and verified in the private staging directory # first. Promotion holds an exclusive directory lock so concurrent filters # cannot interleave shared filenames. Provenance is invalidated first and # promoted last, so it acts as the completed-run commit record. promotion_descriptor, promotion_lock_path = acquire_promotion_lock(data_dir) try: final_provenance_path = data_dir / provenance_path.name final_provenance_path.unlink(missing_ok=True) for file_name in OUTPUT_FILES: os.replace(staging_dir / file_name, data_dir / file_name) evidence = outputs[file_name] assert_file( data_dir / file_name, str(evidence["sha256"]), int(evidence["bytes"]), ) os.replace(provenance_path, final_provenance_path) finally: release_promotion_lock(promotion_descriptor, promotion_lock_path) for file_name, evidence in outputs.items(): print(f"wrote {file_name:<34} {int(evidence['rows']):>5,} rows") print("wrote query-provenance.json") print( "verified checkpoints: 1,484 retained; 207 entered; 266 exited; " f"30 forecasted-to-posted; {len(applicant_type_candidates):,} " f"{args.applicant_type} explicit label matches; 301 posted deadlines in 0-30 days" ) if __name__ == "__main__": main()