#!/usr/bin/env python3 """Verify and query the 2026-08-27 US federal procurement forecasts release. The recipe pins the release manifest by byte size and SHA-256 before trusting its declarations. It then downloads and verifies the machine schema, atomic source-health report, current-snapshot CSV, and change CSV. Both CSV products are loaded as VARCHAR so agency-authored codes, value ranges, quarters, dates, and placeholder text are never silently coerced. The outputs include a bounded NAICS-prefix watchlist. ``naics_code_6`` is a derived leading-six-digit comparison key; the original ``naics_code`` remains unchanged because DOJ values preserve ``code--description`` text. New/updated records and records not in the current snapshot are exported separately. Forecasts are non-binding planning information, not solicitations, committed spending, or proof that an acquisition will proceed. Verify decision-critical facts with the official agency source and SAM.gov. Requires Python 3.10+ and duckdb==1.5.5. SPDX-License-Identifier: MIT """ from __future__ import annotations import argparse import csv import hashlib import json import os import platform import re import tempfile from pathlib import Path from typing import Any from urllib.request import Request, urlopen import duckdb REPOSITORY = "webtruffle/us-federal-procurement-forecasts" RELEASE_TAG = "2026-08-27" RELEASE_URL = f"https://github.com/{REPOSITORY}/releases/tag/{RELEASE_TAG}" ASSET_BASE_URL = f"https://github.com/{REPOSITORY}/releases/download/{RELEASE_TAG}" MANIFEST_SHA256 = "507f08dd3f2ddcac46d0612fa130c3dc3718babfbbfc0e7c3f2fcbe2ad4cdd02" MANIFEST_BYTES = 17_403 EXPECTED_SCHEMA_VERSION = "1.0" EXPECTED_DUCKDB_VERSION = "1.5.5" DEFAULT_NAICS_PREFIX = "5415" DEFAULT_WATCHLIST_LIMIT = 50 MAX_WATCHLIST_LIMIT = 500 USER_AGENT = ( "WebTruffle-US-federal-procurement-forecasts-Python-recipe/1.0 " "(+https://www.webtruffle.com/)" ) CURRENT_FILE = "us-federal-procurement-forecasts.csv" CHANGES_FILE = "changes.csv" SCHEMA_FILE = "schema.json" SOURCE_HEALTH_FILE = "source-health.json" INPUT_FILES = (SCHEMA_FILE, SOURCE_HEALTH_FILE, CURRENT_FILE, CHANGES_FILE) PINNED_INPUT_ASSETS = { SCHEMA_FILE: { "bytes": 5_576, "sha256": "5b41eb451d476b0f4bf1dbcccec4731d2edefd63233b1b70084c1b773540cb8d", "product": "metadata", }, SOURCE_HEALTH_FILE: { "bytes": 10_079, "sha256": "15d5092829fed68d8969cbfc30c7aea1d0e17720209f4549181af9207934fae6", "product": "metadata", }, CURRENT_FILE: { "bytes": 2_985_263, "sha256": "1076227ca33fcb46810f4dc0ae68039690db96c0b8b662ac4070ae42ca599c3d", "product": "forecasts", "record_count": 2_501, }, CHANGES_FILE: { "bytes": 377_363, "sha256": "3f2f8be431f76a400134f74a6dc2dfa649ebc2cb0cc4d87e1e057333d2fd0707", "product": "changes", "record_count": 347, }, } EXPECTED_PRODUCTS = { "forecasts": { "product_id": "forecasts", "record_count": 2_501, "grain": "one record in the complete current four-source snapshot", "change_types": ["new", "updated", "unchanged"], "files": [ "us-federal-procurement-forecasts.csv", "us-federal-procurement-forecasts.json", "us-federal-procurement-forecasts.jsonl", ], }, "changes": { "product_id": "changes", "record_count": 347, "grain": "one record whose snapshot membership or content changed", "change_types": ["new", "updated", "not_in_current_snapshot"], "files": ["changes.csv", "changes.jsonl"], }, } EXPECTED_SOURCES = ( "dhs_apfs", "doe_acquisition_forecast", "education_acquisition_forecast", "doj_acquisition_forecast", ) EXPECTED_CURRENT_SOURCE_COUNTS = { "dhs_apfs": 841, "doe_acquisition_forecast": 823, "education_acquisition_forecast": 370, "doj_acquisition_forecast": 467, } EXPECTED_CURRENT_CHANGE_COUNTS = { "new": 136, "updated": 12, "unchanged": 2_353, } EXPECTED_CHANGE_COUNTS = { "new": 136, "updated": 12, "not_in_current_snapshot": 199, } EXPECTED_SOURCE_CHANGE_COUNTS = { "dhs_apfs": { "new": 4, "updated": 0, "unchanged": 837, "not_in_current_snapshot": 6, }, "doe_acquisition_forecast": { "new": 0, "updated": 0, "unchanged": 823, "not_in_current_snapshot": 0, }, "education_acquisition_forecast": { "new": 0, "updated": 0, "unchanged": 370, "not_in_current_snapshot": 0, }, "doj_acquisition_forecast": { "new": 132, "updated": 12, "unchanged": 323, "not_in_current_snapshot": 193, }, } EXPECTED_IDENTIFIER_METHOD_COUNTS = { "current": { "apfs_number": 841, "derived_composite_no_source_row_id": 823, "tracking_number": 370, "unique_action_tracking_number": 449, "derived_composite_missing_placeholder_or_duplicate_tracking_number": 18, }, "changes": { "apfs_number": 10, "unique_action_tracking_number": 296, "derived_composite_missing_placeholder_or_duplicate_tracking_number": 41, }, } EXPECTED_HEADERS = ( "edition_date", "id", "source", "source_forecast_id", "source_reference", "source_identifier_method", "agency_name", "subagency_name", "organization_name", "funding_office", "contracting_office", "program_office", "fiscal_year", "title", "description", "requirement_type", "category", "subcategory", "naics_code", "naics_description", "product_service_code", "contract_type", "award_type", "contract_vehicle", "competition_strategy", "small_business_program", "set_aside_type", "business_size_selection", "estimated_value_range", "estimated_current_fiscal_year_value_range", "current_contract_number", "incumbent_contractor", "contract_status", "award_quarter", "estimated_solicitation_date", "estimated_solicitation_timing", "anticipated_award_date", "anticipated_award_timing", "performance_start_date", "performance_end_date", "performance_end_month", "period_of_performance", "contract_availability", "contract_length", "rfi_planned", "place_of_performance", "place_of_performance_city", "place_of_performance_state", "place_of_performance_country", "acquisition_history", "solicitation_url", "additional_information", "source_created_date", "source_published_date", "source_last_updated_date", "source_url", "source_document_url", "source_terms_url", "first_seen_at", "last_seen_at", "content_hash", "change_type", ) SOURCE_AWARE_FIELDS = ( "title", "description", "fiscal_year", "naics_code", "naics_description", "product_service_code", "estimated_value_range", "estimated_current_fiscal_year_value_range", "competition_strategy", "small_business_program", "set_aside_type", "business_size_selection", "estimated_solicitation_date", "estimated_solicitation_timing", "anticipated_award_date", "anticipated_award_timing", "award_quarter", "place_of_performance_state", "solicitation_url", "current_contract_number", "incumbent_contractor", "source_url", ) EXPECTED_SOURCE_FIELD_PRESENCE = { "dhs_apfs": { "title": 841, "description": 841, "fiscal_year": 841, "naics_code": 841, "naics_description": 841, "product_service_code": 0, "estimated_value_range": 841, "estimated_current_fiscal_year_value_range": 0, "competition_strategy": 841, "small_business_program": 840, "set_aside_type": 353, "business_size_selection": 0, "estimated_solicitation_date": 841, "estimated_solicitation_timing": 0, "anticipated_award_date": 841, "anticipated_award_timing": 0, "award_quarter": 841, "place_of_performance_state": 841, "solicitation_url": 0, "current_contract_number": 274, "incumbent_contractor": 274, "source_url": 841, }, "doe_acquisition_forecast": { "title": 823, "description": 823, "fiscal_year": 0, "naics_code": 823, "naics_description": 823, "product_service_code": 0, "estimated_value_range": 823, "estimated_current_fiscal_year_value_range": 0, "competition_strategy": 0, "small_business_program": 0, "set_aside_type": 823, "business_size_selection": 823, "estimated_solicitation_date": 0, "estimated_solicitation_timing": 0, "anticipated_award_date": 0, "anticipated_award_timing": 0, "award_quarter": 0, "place_of_performance_state": 823, "solicitation_url": 0, "current_contract_number": 823, "incumbent_contractor": 823, "source_url": 823, }, "education_acquisition_forecast": { "title": 370, "description": 0, "fiscal_year": 370, "naics_code": 370, "naics_description": 370, "product_service_code": 0, "estimated_value_range": 370, "estimated_current_fiscal_year_value_range": 370, "competition_strategy": 370, "small_business_program": 0, "set_aside_type": 0, "business_size_selection": 0, "estimated_solicitation_date": 0, "estimated_solicitation_timing": 0, "anticipated_award_date": 0, "anticipated_award_timing": 0, "award_quarter": 370, "place_of_performance_state": 0, "solicitation_url": 0, "current_contract_number": 0, "incumbent_contractor": 370, "source_url": 370, }, "doj_acquisition_forecast": { "title": 467, "description": 467, "fiscal_year": 467, "naics_code": 467, "naics_description": 0, "product_service_code": 467, "estimated_value_range": 467, "estimated_current_fiscal_year_value_range": 0, "competition_strategy": 467, "small_business_program": 467, "set_aside_type": 0, "business_size_selection": 0, "estimated_solicitation_date": 0, "estimated_solicitation_timing": 467, "anticipated_award_date": 0, "anticipated_award_timing": 467, "award_quarter": 467, "place_of_performance_state": 341, "solicitation_url": 19, "current_contract_number": 280, "incumbent_contractor": 281, "source_url": 467, }, } def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description=( "Verify the WebTruffle 2026-08-27 federal procurement forecasts " "release and export a bounded, source-aware NAICS watchlist." ) ) parser.add_argument( "--data-dir", type=Path, default=Path(f"us-federal-procurement-forecasts-{RELEASE_TAG}"), help="Directory for verified inputs and deterministic query outputs.", ) parser.add_argument( "--naics-prefix", default=DEFAULT_NAICS_PREFIX, help="A 2-6 digit prefix matched against a derived leading six-digit key.", ) parser.add_argument( "--watchlist-limit", type=int, default=DEFAULT_WATCHLIST_LIMIT, help=f"Maximum watchlist rows (1-{MAX_WATCHLIST_LIMIT}).", ) args = parser.parse_args() if not re.fullmatch(r"[0-9]{2,6}", args.naics_prefix): parser.error("--naics-prefix must contain exactly 2-6 ASCII digits") if not 1 <= args.watchlist_limit <= MAX_WATCHLIST_LIMIT: parser.error(f"--watchlist-limit must be between 1 and {MAX_WATCHLIST_LIMIT}") return args 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(1024 * 1024), b""): digest.update(chunk) size += len(chunk) return digest.hexdigest(), size def assert_file( path: Path, expected_sha256: str, expected_bytes: int, ) -> tuple[str, int]: 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}. " "Move or delete the unexpected file before retrying." ) return actual_sha256, actual_bytes def download_missing( url: str, destination: Path, expected_sha256: str, expected_bytes: int, ) -> None: if destination.exists(): assert_file(destination, expected_sha256, expected_bytes) print(f"verified existing {destination.name}") return 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) try: downloaded_bytes = 0 digest = hashlib.sha256() with ( urlopen(request, timeout=120) as response, os.fdopen(descriptor, "wb") as target, ): while chunk := response.read(1024 * 1024): downloaded_bytes += len(chunk) if downloaded_bytes > expected_bytes: raise RuntimeError( f"Download exceeded the declared {expected_bytes} bytes for " f"{destination.name}." ) digest.update(chunk) target.write(chunk) if downloaded_bytes != expected_bytes or digest.hexdigest() != expected_sha256: raise RuntimeError( f"Verification failed for downloaded {destination.name}: expected " f"{expected_bytes} bytes / {expected_sha256}, got {downloaded_bytes} " f"bytes / {digest.hexdigest()}." ) os.replace(partial, destination) finally: try: os.close(descriptor) except OSError: pass partial.unlink(missing_ok=True) print(f"downloaded + verified {destination.name}") def load_manifest(data_dir: Path) -> dict[str, Any]: path = data_dir / "manifest.json" download_missing( f"{ASSET_BASE_URL}/manifest.json", path, MANIFEST_SHA256, MANIFEST_BYTES, ) manifest = json.loads(path.read_text(encoding="utf-8")) expected_scalars = { "dataset_id": "us-federal-procurement-forecasts", "schema_version": EXPECTED_SCHEMA_VERSION, "target_date": RELEASE_TAG, "edition_date": RELEASE_TAG, "snapshot_type": "full_current_snapshot", "atomic_complete": True, "record_count": 2_501, "change_count": 347, } observed_scalars = {key: manifest.get(key) for key in expected_scalars} if observed_scalars != expected_scalars: raise RuntimeError(f"Pinned manifest checkpoints changed: {observed_scalars}.") if tuple(manifest.get("required_sources", [])) != EXPECTED_SOURCES: raise RuntimeError( f"Required source set changed: {manifest.get('required_sources')!r}." ) if manifest.get("record_counts_by_source") != EXPECTED_CURRENT_SOURCE_COUNTS: raise RuntimeError("Pinned manifest source counts changed.") if manifest.get("change_counts") != EXPECTED_CHANGE_COUNTS: raise RuntimeError("Pinned manifest change counts changed.") if manifest.get("products") != EXPECTED_PRODUCTS: raise RuntimeError("Pinned manifest product declarations changed.") if tuple(manifest.get("record_fields", [])) != EXPECTED_HEADERS: raise RuntimeError("Manifest record_fields changed from the pinned 62 fields.") declared_files = manifest.get("files", {}) for file_name, pinned in PINNED_INPUT_ASSETS.items(): declaration = declared_files.get(file_name, {}) observed = { key: declaration.get(key) for key in ("bytes", "sha256", "product", "record_count") if key in pinned } if observed != pinned: raise RuntimeError( f"Pinned asset declaration changed for {file_name}: {observed}." ) return manifest def download_inputs(data_dir: Path, manifest: dict[str, Any]) -> None: for file_name in INPUT_FILES: declaration = manifest["files"][file_name] download_missing( f"{ASSET_BASE_URL}/{file_name}", data_dir / file_name, declaration["sha256"], int(declaration["bytes"]), ) def read_csv_header(path: Path) -> list[str]: with path.open("r", encoding="utf-8-sig", newline="") as source: return next(csv.reader(source)) def validate_schema_and_headers(data_dir: Path) -> dict[str, Any]: schema = json.loads((data_dir / SCHEMA_FILE).read_text(encoding="utf-8")) properties = schema.get("properties", {}) if schema.get("additionalProperties") is not False: raise RuntimeError("Pinned schema must reject additional properties.") if tuple(properties) != EXPECTED_HEADERS: raise RuntimeError("Schema property order changed from the pinned 62 fields.") if tuple(schema.get("required", [])) != EXPECTED_HEADERS: raise RuntimeError("Schema required fields changed from the pinned 62 fields.") if len(EXPECTED_HEADERS) != 62: raise RuntimeError("Recipe header contract must contain exactly 62 fields.") evidence: dict[str, Any] = {} for file_name in (CURRENT_FILE, CHANGES_FILE): headers = read_csv_header(data_dir / file_name) if len(headers) != len(set(headers)): raise RuntimeError(f"{file_name} contains duplicate column names.") if tuple(headers) != EXPECTED_HEADERS: missing = sorted(set(EXPECTED_HEADERS) - set(headers)) extra = sorted(set(headers) - set(EXPECTED_HEADERS)) raise RuntimeError( f"Exact 62-field header validation failed for {file_name}: " f"missing={missing}, extra={extra}, " f"order_matches={tuple(headers) == EXPECTED_HEADERS}." ) evidence[file_name] = { "header_count": len(headers), "exact_schema_and_manifest_order": True, "csv_transport_type": "text", } return evidence def validate_source_health( data_dir: Path, manifest: dict[str, Any], ) -> dict[str, Any]: report = json.loads((data_dir / SOURCE_HEALTH_FILE).read_text(encoding="utf-8")) for key in ("dataset_id", "schema_version", "target_date", "generated_at"): if report.get(key) != manifest.get(key): raise RuntimeError( f"Source-health {key} does not match the verified manifest." ) if report.get("atomic_complete") is not True: raise RuntimeError("Source-health report is not atomic and complete.") if report.get("all_required_sources_succeeded") is not True: raise RuntimeError("Not every required source succeeded.") if tuple(report.get("required_sources", [])) != EXPECTED_SOURCES: raise RuntimeError("Source-health required source set changed.") if report.get("record_counts_by_source") != EXPECTED_CURRENT_SOURCE_COUNTS: raise RuntimeError("Source-health record counts changed.") entries = report.get("sources", []) by_source = {entry.get("source"): entry for entry in entries} if len(entries) != len(by_source) or set(by_source) != set(EXPECTED_SOURCES): raise RuntimeError( "Source-health entries are missing, duplicated, or unexpected." ) evidence: dict[str, Any] = {} manifest_sources = { entry.get("source"): entry for entry in manifest.get("sources", []) } for source in EXPECTED_SOURCES: entry = by_source[source] expected = EXPECTED_SOURCE_CHANGE_COUNTS[source] observed = { "new": int(entry.get("new_count", -1)), "updated": int(entry.get("updated_count", -1)), "unchanged": int(entry.get("unchanged_count", -1)), "not_in_current_snapshot": int( entry.get("not_in_current_snapshot_count", -1) ), } if entry.get("status") != "succeeded" or entry.get("error") != "": raise RuntimeError(f"Source {source} did not report clean success.") if int(entry.get("records_seen", -1)) != EXPECTED_CURRENT_SOURCE_COUNTS[source]: raise RuntimeError(f"Source-health record count changed for {source}.") if observed != expected: raise RuntimeError( f"Source-health change counts changed for {source}: {observed}." ) if manifest_sources.get(source, {}).get("status") != "succeeded": raise RuntimeError(f"Manifest source status changed for {source}.") metadata = entry.get("source_metadata", {}) if metadata.get("raw_download_retained") is not False: raise RuntimeError(f"Unexpected raw-retention state for {source}.") evidence[source] = { "status": "succeeded", "records_seen": EXPECTED_CURRENT_SOURCE_COUNTS[source], "change_counts": expected, "raw_download_retained": False, } return { "atomic_complete": True, "all_required_sources_succeeded": True, "sources": evidence, } def register_csv_tables( connection: duckdb.DuckDBPyConnection, data_dir: Path, ) -> dict[str, Any]: evidence: dict[str, Any] = {} for file_name, table_name in ( (CURRENT_FILE, "forecasts"), (CHANGES_FILE, "changes"), ): source_view = f"{table_name}_source" connection.read_csv( str(data_dir / file_name), header=True, all_varchar=True, ).create_view(source_view) connection.execute( f"CREATE TEMP TABLE {table_name} AS SELECT * FROM {source_view}" ) connection.execute(f"DROP VIEW {source_view}") table_info = connection.execute(f"PRAGMA table_info('{table_name}')").fetchall() if len(table_info) != 62: raise RuntimeError(f"{table_name} did not materialize 62 columns.") types = {row[2] for row in table_info} if types != {"VARCHAR"}: raise RuntimeError( f"{table_name} contains inferred non-VARCHAR columns: {sorted(types)}" ) if tuple(row[1] for row in table_info) != EXPECTED_HEADERS: raise RuntimeError(f"{table_name} column order changed after loading.") evidence[table_name] = { "columns": len(table_info), "source_type": "VARCHAR", "source_values_preserved_as_text": True, } return evidence def fetch_named_row( connection: duckdb.DuckDBPyConnection, sql: str, ) -> dict[str, Any]: cursor = connection.execute(sql) row = cursor.fetchone() if row is None: raise RuntimeError("Expected one validation row, got none.") return dict(zip((column[0] for column in cursor.description), row)) def assert_counts_keys_and_product_relationships( connection: duckdb.DuckDBPyConnection, ) -> dict[str, Any]: checks = fetch_named_row( connection, f""" SELECT (SELECT count(*) FROM forecasts) AS current_rows, (SELECT count(*) FROM changes) AS change_rows, (SELECT count(*) FROM forecasts WHERE nullif(trim(id), '') IS NULL) AS current_missing_ids, (SELECT count(*) - count(DISTINCT id) FROM forecasts) AS current_duplicate_ids, (SELECT count(*) FROM changes WHERE nullif(trim(id), '') IS NULL) AS change_missing_ids, (SELECT count(*) - count(DISTINCT id) FROM changes) AS change_duplicate_ids, (SELECT count(*) FROM forecasts WHERE edition_date <> '{RELEASE_TAG}') AS wrong_current_editions, (SELECT count(*) FROM changes WHERE edition_date <> '{RELEASE_TAG}') AS wrong_change_editions, (SELECT count(*) FROM forecasts WHERE NOT starts_with(id, source || ':')) AS malformed_current_ids, (SELECT count(*) FROM changes WHERE NOT starts_with(id, source || ':')) AS malformed_change_ids, (SELECT count(*) FROM forecasts WHERE NOT regexp_full_match(content_hash, '[0-9a-f]{{64}}')) AS malformed_current_hashes, (SELECT count(*) FROM changes WHERE NOT regexp_full_match(content_hash, '[0-9a-f]{{64}}')) AS malformed_change_hashes, (SELECT count(*) FROM changes change LEFT JOIN forecasts current USING (id) WHERE change.change_type IN ('new', 'updated') AND current.id IS NULL) AS active_changes_missing_current_rows, (SELECT count(*) FROM changes change JOIN forecasts current USING (id) WHERE change.change_type IN ('new', 'updated') AND ( change.content_hash IS DISTINCT FROM current.content_hash OR change.change_type IS DISTINCT FROM current.change_type OR change.source IS DISTINCT FROM current.source )) AS active_change_content_mismatches, (SELECT count(*) FROM changes change JOIN forecasts current USING (id) WHERE change.change_type = 'not_in_current_snapshot') AS departed_ids_still_current, (SELECT count(*) FROM forecasts current LEFT JOIN changes change USING (id) WHERE current.change_type IN ('new', 'updated') AND change.id IS NULL) AS current_changes_missing_change_rows, (SELECT count(*) FROM forecasts current JOIN changes change USING (id) WHERE current.change_type = 'unchanged') AS unchanged_rows_in_change_feed, (SELECT count(*) FROM forecasts WHERE regexp_extract(naics_code, '^([0-9]{{6}})', 1) = '') AS rows_without_leading_six_digit_naics, (SELECT count(*) FROM forecasts WHERE regexp_full_match(naics_code, '[0-9]{{6}}')) AS exact_six_digit_naics_rows, (SELECT count(*) FROM forecasts WHERE regexp_full_match(naics_code, '[0-9]{{6}}--.+')) AS naics_rows_with_preserved_description_suffix, (SELECT count(*) FROM forecasts WHERE source_identifier_method LIKE 'derived%') AS current_rows_with_derived_identity, (SELECT count(*) FROM forecasts WHERE nullif(solicitation_url, '') IS NOT NULL) AS current_rows_with_solicitation_url, (SELECT count(DISTINCT source_url) FROM forecasts) AS distinct_current_source_urls """, ) expected = { "current_rows": 2_501, "change_rows": 347, "current_missing_ids": 0, "current_duplicate_ids": 0, "change_missing_ids": 0, "change_duplicate_ids": 0, "wrong_current_editions": 0, "wrong_change_editions": 0, "malformed_current_ids": 0, "malformed_change_ids": 0, "malformed_current_hashes": 0, "malformed_change_hashes": 0, "active_changes_missing_current_rows": 0, "active_change_content_mismatches": 0, "departed_ids_still_current": 0, "current_changes_missing_change_rows": 0, "unchanged_rows_in_change_feed": 0, "rows_without_leading_six_digit_naics": 0, "exact_six_digit_naics_rows": 2_034, "naics_rows_with_preserved_description_suffix": 467, "current_rows_with_derived_identity": 841, "current_rows_with_solicitation_url": 19, "distinct_current_source_urls": 844, } if checks != expected: raise RuntimeError( f"Pinned row/key/product checks failed: expected {expected}, got {checks}." ) return checks def query_group_counts( connection: duckdb.DuckDBPyConnection, table: str, field: str, ) -> dict[str, int]: return { str(key): int(count) for key, count in connection.execute( f"SELECT {field}, count(*) FROM {table} GROUP BY {field}" ).fetchall() } def assert_group_counts(connection: duckdb.DuckDBPyConnection) -> dict[str, Any]: observed = { "current_sources": query_group_counts(connection, "forecasts", "source"), "current_change_types": query_group_counts( connection, "forecasts", "change_type" ), "change_types": query_group_counts(connection, "changes", "change_type"), "current_identifier_methods": query_group_counts( connection, "forecasts", "source_identifier_method" ), "change_identifier_methods": query_group_counts( connection, "changes", "source_identifier_method" ), } expected = { "current_sources": EXPECTED_CURRENT_SOURCE_COUNTS, "current_change_types": EXPECTED_CURRENT_CHANGE_COUNTS, "change_types": EXPECTED_CHANGE_COUNTS, "current_identifier_methods": EXPECTED_IDENTIFIER_METHOD_COUNTS["current"], "change_identifier_methods": EXPECTED_IDENTIFIER_METHOD_COUNTS["changes"], } if observed != expected: raise RuntimeError( f"Pinned categorical counts failed: expected {expected}, got {observed}." ) return observed def build_source_field_coverage_sql() -> str: unions = [] for field in SOURCE_AWARE_FIELDS: unions.append( f"""SELECT source, '{field}' AS field_name, count_if(nullif(trim({field}), '') IS NOT NULL) AS present_records, count(*) AS total_records FROM forecasts GROUP BY source""" ) return f""" WITH coverage AS ( {" UNION ALL ".join(unions)} ) SELECT '{RELEASE_TAG}' AS release_tag, source, field_name, present_records, total_records, round(present_records * 100.0 / total_records, 2) AS presence_percent FROM coverage ORDER BY source, field_name """.strip() SOURCE_FIELD_COVERAGE_SQL = build_source_field_coverage_sql() def assert_source_field_coverage( connection: duckdb.DuckDBPyConnection, ) -> dict[str, Any]: rows = connection.execute(SOURCE_FIELD_COVERAGE_SQL).fetchall() observed = {source: {} for source in EXPECTED_SOURCES} for _, source, field, present, total, _ in rows: if source not in observed or field not in SOURCE_AWARE_FIELDS: raise RuntimeError( f"Unexpected source/field coverage row: {source}/{field}." ) if int(total) != EXPECTED_CURRENT_SOURCE_COUNTS[source]: raise RuntimeError(f"Coverage denominator changed for {source}/{field}.") observed[source][field] = int(present) if observed != EXPECTED_SOURCE_FIELD_PRESENCE: raise RuntimeError( "Pinned source-aware field coverage changed: " f"expected {EXPECTED_SOURCE_FIELD_PRESENCE}, got {observed}." ) return { "field_count": len(SOURCE_AWARE_FIELDS), "row_count": len(rows), "presence_counts": observed, "interpretation": ( "Presence is reported within each source. Empty normalized fields often " "mean the agency does not publish that concept." ), } RELEASE_CHECKPOINTS_SQL = f""" SELECT '{RELEASE_TAG}' AS release_tag, (SELECT count(*) FROM forecasts) AS current_snapshot_records, (SELECT count(*) FROM changes) AS observed_change_records, (SELECT count(DISTINCT source) FROM forecasts) AS successful_sources, (SELECT count(*) FROM forecasts WHERE change_type = 'new') AS current_new, (SELECT count(*) FROM forecasts WHERE change_type = 'updated') AS current_updated, (SELECT count(*) FROM forecasts WHERE change_type = 'unchanged') AS current_unchanged, (SELECT count(*) FROM changes WHERE change_type = 'new') AS change_feed_new, (SELECT count(*) FROM changes WHERE change_type = 'updated') AS change_feed_updated, (SELECT count(*) FROM changes WHERE change_type = 'not_in_current_snapshot') AS change_feed_not_in_current_snapshot, (SELECT count(*) FROM forecasts WHERE source_identifier_method LIKE 'derived%') AS current_records_with_derived_identity, (SELECT count(*) FROM forecasts WHERE regexp_full_match(naics_code, '[0-9]{{6}}')) AS exact_six_digit_naics_values, (SELECT count(*) FROM forecasts WHERE regexp_full_match(naics_code, '[0-9]{{6}}--.+')) AS naics_values_with_preserved_description_suffix, (SELECT count(*) FROM forecasts WHERE nullif(solicitation_url, '') IS NOT NULL) AS records_with_solicitation_url """.strip() CHANGE_COUNTS_BY_SOURCE_SQL = f""" WITH source_names(source, source_order) AS ( VALUES ('dhs_apfs', 1), ('doe_acquisition_forecast', 2), ('education_acquisition_forecast', 3), ('doj_acquisition_forecast', 4) ), change_names(change_type, change_order) AS ( VALUES ('new', 1), ('updated', 2), ('not_in_current_snapshot', 3) ) SELECT '{RELEASE_TAG}' AS release_tag, source_names.source, change_names.change_type, count(changes.id) AS change_records FROM source_names CROSS JOIN change_names LEFT JOIN changes ON changes.source = source_names.source AND changes.change_type = change_names.change_type GROUP BY source_names.source, source_names.source_order, change_names.change_type, change_names.change_order ORDER BY source_names.source_order, change_names.change_order """.strip() def build_watchlist_sql(naics_prefix: str, limit: int) -> str: return f""" SELECT '{RELEASE_TAG}' AS release_tag, regexp_extract(naics_code, '^([0-9]{{6}})', 1) AS naics_code_6, forecasts.* FROM forecasts WHERE starts_with( regexp_extract(naics_code, '^([0-9]{{6}})', 1), '{naics_prefix}' ) ORDER BY naics_code_6, source, id LIMIT {limit} """.strip() NEW_UPDATED_SQL = f""" SELECT '{RELEASE_TAG}' AS release_tag, regexp_extract(naics_code, '^([0-9]{{6}})', 1) AS naics_code_6, changes.* FROM changes WHERE change_type IN ('new', 'updated') ORDER BY source, change_type, id """.strip() NOT_IN_CURRENT_SNAPSHOT_SQL = f""" SELECT '{RELEASE_TAG}' AS release_tag, regexp_extract(naics_code, '^([0-9]{{6}})', 1) AS naics_code_6, changes.* FROM changes WHERE change_type = 'not_in_current_snapshot' ORDER BY source, id """.strip() def write_query( connection: duckdb.DuckDBPyConnection, sql: str, destination: Path, ) -> dict[str, Any]: rows = connection.execute(f"SELECT count(*) FROM ({sql}) AS result").fetchone()[0] connection.sql(sql).write_csv( str(destination), header=True, overwrite=True, use_tmp_file=True, ) digest, size = sha256_and_size(destination) print(f"wrote {destination.name:<48} {rows:>7,} rows") return {"rows": int(rows), "bytes": size, "sha256": digest} def reassert_input_files( data_dir: Path, manifest: dict[str, Any], ) -> dict[str, dict[str, Any]]: evidence: dict[str, dict[str, Any]] = {} digest, size = assert_file( data_dir / "manifest.json", MANIFEST_SHA256, MANIFEST_BYTES, ) evidence["manifest.json"] = {"bytes": size, "sha256": digest} for file_name in INPUT_FILES: declaration = manifest["files"][file_name] digest, size = assert_file( data_dir / file_name, declaration["sha256"], int(declaration["bytes"]), ) evidence[file_name] = {"bytes": size, "sha256": digest} return evidence def main() -> None: args = parse_args() if duckdb.__version__ != EXPECTED_DUCKDB_VERSION: raise RuntimeError( f"This recipe requires duckdb=={EXPECTED_DUCKDB_VERSION}; " f"found {duckdb.__version__}." ) data_dir = args.data_dir.expanduser().resolve() data_dir.mkdir(parents=True, exist_ok=True) manifest = load_manifest(data_dir) download_inputs(data_dir, manifest) schema_evidence = validate_schema_and_headers(data_dir) source_health_evidence = validate_source_health(data_dir, manifest) connection = duckdb.connect() varchar_evidence = register_csv_tables(connection, data_dir) key_product_evidence = assert_counts_keys_and_product_relationships(connection) group_count_evidence = assert_group_counts(connection) field_coverage_evidence = assert_source_field_coverage(connection) input_evidence = reassert_input_files(data_dir, manifest) watchlist_sql = build_watchlist_sql( args.naics_prefix, args.watchlist_limit, ) matching_current_rows = connection.execute( """ SELECT count(*) FROM forecasts WHERE starts_with( regexp_extract(naics_code, '^([0-9]{6})', 1), ? ) """, [args.naics_prefix], ).fetchone()[0] with tempfile.TemporaryDirectory( dir=data_dir, prefix=".forecast-results.", ) as temporary_directory: staging_dir = Path(temporary_directory) queries = { "forecast-release-checkpoints.csv": RELEASE_CHECKPOINTS_SQL, "forecast-change-counts-by-source.csv": CHANGE_COUNTS_BY_SOURCE_SQL, "forecast-source-field-coverage.csv": SOURCE_FIELD_COVERAGE_SQL, "forecast-naics-watchlist.csv": watchlist_sql, "forecast-new-updated.csv": NEW_UPDATED_SQL, "forecast-not-in-current-snapshot.csv": NOT_IN_CURRENT_SNAPSHOT_SQL, } outputs = { file_name: write_query( connection, sql, staging_dir / file_name, ) for file_name, sql in queries.items() } expected_output_rows = { "forecast-release-checkpoints.csv": 1, "forecast-change-counts-by-source.csv": 12, "forecast-source-field-coverage.csv": 88, "forecast-naics-watchlist.csv": min( int(matching_current_rows), args.watchlist_limit, ), "forecast-new-updated.csv": 148, "forecast-not-in-current-snapshot.csv": 199, } observed_output_rows = { file_name: evidence["rows"] for file_name, evidence in outputs.items() } if observed_output_rows != expected_output_rows: raise RuntimeError( "Pinned-release output checkpoints failed: expected " f"{expected_output_rows}, got {observed_output_rows}." ) recipe_path = Path(__file__).resolve() recipe_sha256, recipe_bytes = sha256_and_size(recipe_path) provenance = { "dataset_id": "us-federal-procurement-forecasts", "recipe": "us-federal-procurement-forecasts-python", "recipe_version": "1.0", "recipe_file": { "name": recipe_path.name, "bytes": recipe_bytes, "sha256": recipe_sha256, }, "release_tag": RELEASE_TAG, "release_url": RELEASE_URL, "schema_version": manifest["schema_version"], "release_generated_at": manifest["generated_at"], "runtime_contract": { "python_minimum_version": "3.10", "python_implementation": platform.python_implementation(), "duckdb_version": duckdb.__version__, }, "selection": { "naics_prefix": args.naics_prefix, "derived_comparison_field": "naics_code_6", "source_field_preserved": "naics_code", "matching_current_rows_before_limit": int(matching_current_rows), "watchlist_limit": args.watchlist_limit, }, "product_counts": { "current_snapshot": 2_501, "changes": 347, }, "product_grains": { product: declaration["grain"] for product, declaration in EXPECTED_PRODUCTS.items() }, "schema_and_headers": schema_evidence, "source_health": source_health_evidence, "materialized_source_types": varchar_evidence, "key_and_product_checks": key_product_evidence, "categorical_counts": group_count_evidence, "source_aware_field_coverage": field_coverage_evidence, "inputs": input_evidence, "outputs": outputs, "queries": queries, "receipt_determinism": ( "No wall-clock execution timestamp or machine path is recorded. " "Sorted JSON, pinned inputs, ordered queries, fixed runtime contract, " "and explicit selection parameters make the receipt reproducible for " "the same recipe and options." ), "interpretation": { "snapshot": ( "The current product is one accepted four-source snapshot, not a " "government-wide inventory." ), "changes": ( "New and updated rows remain active in the pinned current snapshot. " "not_in_current_snapshot rows are exported separately and do not " "prove cancellation, deletion, or award." ), "naics": ( "naics_code_6 is derived only for comparison. The source-authored " "naics_code is retained because DOJ preserves code--description text." ), "values_and_timing": ( "Value ranges, dates, quarters, competition, and set-aside fields " "remain source text. Do not infer spend, invent dates, or treat empty " "cross-source fields as factual absence." ), "authority": ( "Forecasts are non-binding planning signals. Verify each candidate " "at the official source and confirm live notices on SAM.gov." ), }, } staged_receipt = staging_dir / "forecast-query-provenance.json" staged_receipt.write_text( json.dumps(provenance, indent=2, sort_keys=True) + "\n", encoding="utf-8", ) final_receipt = data_dir / staged_receipt.name final_receipt.unlink(missing_ok=True) for file_name in queries: os.replace(staging_dir / file_name, data_dir / file_name) os.replace(staged_receipt, final_receipt) print(f"wrote {final_receipt.name}") connection.close() print( "verified checkpoint: 2,501 current records; 347 changes; " "148 new/updated; 199 not in current snapshot; four atomic sources; " f"{matching_current_rows:,} NAICS {args.naics_prefix}* matches before the " f"{args.watchlist_limit}-row watchlist limit" ) if __name__ == "__main__": main()