#!/usr/bin/env python3
"""Build a reproducible U.S. vacation-home statistics dataset from ACS files.

The dataset uses the Census Bureau's formal category "For seasonal,
recreational, or occasional use" as the measurable proxy often called
"vacation homes." It joins ACS Detailed Tables B25004 and B25001 to the ACS
Summary File geography labels, retains published 90% margins of error, and
calculates shares and rankings for states and county equivalents. It also builds a
separate 2024 ACS 1-year nation/state supplement so annual state figures are not
mixed with the complete all-county 5-year comparison.

Verified/rebuilt: 2026-07-29
"""
from __future__ import annotations

import hashlib
import json
import math
import os
from pathlib import Path
import zipfile

import pandas as pd

OUT = Path(os.environ.get('VACATION_HOME_DATA_DIR', Path(__file__).resolve().parent))
VERIFIED_DATE = '2026-07-29'
VINTAGE = '2020-2024 ACS 5-year estimates'
ANNUAL_VINTAGE = '2024 ACS 1-year estimates'
PRIOR_VINTAGE = '2015-2019 ACS 5-year estimates'

FILES = {
    'b25001_2024': OUT / 'acsdt5y2024-b25001.dat',
    'b25004_2024': OUT / 'acsdt5y2024-b25004.dat',
    'geos_2024': OUT / 'Geos20245YR.txt',
    'b25001_2024_1yr': OUT / 'acsdt1y2024-b25001.dat',
    'b25004_2024_1yr': OUT / 'acsdt1y2024-b25004.dat',
    'geos_2024_1yr': OUT / 'Geos20241YR.txt',
    'b25001_2019': OUT / 'acsdt5y2019-b25001.dat',
    'b25004_2019': OUT / 'acsdt5y2019-b25004.dat',
    'geos_2019': OUT / 'Geos20195YR.csv',
}

SOURCE_URLS = {
    'b25001_2024': 'https://www2.census.gov/programs-surveys/acs/summary_file/2024/table-based-SF/data/5YRData/acsdt5y2024-b25001.dat',
    'b25004_2024': 'https://www2.census.gov/programs-surveys/acs/summary_file/2024/table-based-SF/data/5YRData/acsdt5y2024-b25004.dat',
    'geos_2024': 'https://www2.census.gov/programs-surveys/acs/summary_file/2024/table-based-SF/documentation/Geos20245YR.txt',
    'b25001_2024_1yr': 'https://www2.census.gov/programs-surveys/acs/summary_file/2024/table-based-SF/data/1YRData/acsdt1y2024-b25001.dat',
    'b25004_2024_1yr': 'https://www2.census.gov/programs-surveys/acs/summary_file/2024/table-based-SF/data/1YRData/acsdt1y2024-b25004.dat',
    'geos_2024_1yr': 'https://www2.census.gov/programs-surveys/acs/summary_file/2024/table-based-SF/documentation/Geos20241YR.txt',
    'b25001_2019': 'https://www2.census.gov/programs-surveys/acs/summary_file/2019/prototype/5YRData/acsdt5y2019-b25001.dat',
    'b25004_2019': 'https://www2.census.gov/programs-surveys/acs/summary_file/2019/prototype/5YRData/acsdt5y2019-b25004.dat',
    'geos_2019': 'https://www2.census.gov/programs-surveys/acs/summary_file/2019/prototype/Geos20195YR.csv',
}

OUT_FILES = {
    'states': OUT / 'vacation-home-statistics-2020-2024-acs-states.csv',
    'annual_states': OUT / 'vacation-home-statistics-2024-acs-1-year-national-states.csv',
    'counties': OUT / 'vacation-home-statistics-2020-2024-acs-counties.csv',
    'idaho': OUT / 'vacation-home-statistics-2020-2024-acs-idaho-counties.csv',
    'local': OUT / 'vacation-home-statistics-2020-2024-acs-local.csv',
    'json': OUT / 'vacation-home-statistics-2020-2024-acs.json',
    'methodology': OUT / 'vacation-home-statistics-methodology.md',
    'checksums': OUT / 'vacation-home-statistics-sha256.txt',
    'bundle': OUT / 'vacation-home-statistics-2020-2024-acs-bundle.zip',
}


def require_inputs() -> None:
    missing = [str(path) for path in FILES.values() if not path.exists()]
    if missing:
        raise FileNotFoundError('Missing input files:\n' + '\n'.join(missing))


def read_table(path: Path, columns: list[str]) -> pd.DataFrame:
    df = pd.read_csv(path, sep='|', dtype={'GEO_ID': 'string'}, usecols=columns, low_memory=False)
    for c in columns:
        if c != 'GEO_ID':
            df[c] = pd.to_numeric(df[c], errors='coerce').astype('Int64')
    return df


def read_geos_2024() -> pd.DataFrame:
    use = ['STUSAB', 'SUMLEVEL', 'COMPONENT', 'STATE', 'COUNTY', 'PLACE', 'GEO_ID', 'NAME']
    df = pd.read_csv(FILES['geos_2024'], sep='|', dtype='string', usecols=use, low_memory=False)
    return df[df['COMPONENT'].eq('00')].copy()


def read_geos_2019() -> pd.DataFrame:
    use = ['STUSAB', 'SUMLEVEL', 'COMPONENT', 'STATE', 'COUNTY', 'PLACE', 'DADSID', 'NAME']
    df = pd.read_csv(FILES['geos_2019'], dtype='string', usecols=use, low_memory=False, encoding='latin-1')
    df = df[df['COMPONENT'].eq('00')].copy()
    return df.rename(columns={'DADSID': 'GEO_ID'})


def read_geos_2024_1yr() -> pd.DataFrame:
    use = ['STUSAB', 'SUMLEVEL', 'COMPONENT', 'STATE', 'COUNTY', 'PLACE', 'GEO_ID', 'NAME']
    df = pd.read_csv(FILES['geos_2024_1yr'], sep='|', dtype='string', usecols=use, low_memory=False)
    return df[df['COMPONENT'].eq('00')].copy()


def load_annual_2024() -> pd.DataFrame:
    b1 = read_table(FILES['b25001_2024_1yr'], ['GEO_ID', 'B25001_E001', 'B25001_M001'])
    b4 = read_table(
        FILES['b25004_2024_1yr'],
        ['GEO_ID', 'B25004_E001', 'B25004_M001', 'B25004_E006', 'B25004_M006'],
    )
    geos = read_geos_2024_1yr()
    df = geos.merge(b1, on='GEO_ID', how='inner', validate='one_to_one')
    return df.merge(b4, on='GEO_ID', how='inner', validate='one_to_one')


def load_vintage(year: int) -> pd.DataFrame:
    if year == 2024:
        b1_path = FILES['b25001_2024']
        b4_path = FILES['b25004_2024']
        geos = read_geos_2024()
    elif year == 2019:
        b1_path = FILES['b25001_2019']
        b4_path = FILES['b25004_2019']
        geos = read_geos_2019()
    else:
        raise ValueError(year)

    b1 = read_table(b1_path, ['GEO_ID', 'B25001_E001', 'B25001_M001'])
    b4 = read_table(b4_path, ['GEO_ID', 'B25004_E001', 'B25004_M001', 'B25004_E006', 'B25004_M006'])
    df = geos.merge(b1, on='GEO_ID', how='inner', validate='one_to_one')
    df = df.merge(b4, on='GEO_ID', how='inner', validate='one_to_one')
    return df


def ratio_pct(numerator: pd.Series, denominator: pd.Series) -> pd.Series:
    out = numerator.astype('Float64').div(denominator.astype('Float64')).mul(100)
    return out.where(denominator.ne(0))


def base_transform(df: pd.DataFrame, geography_type: str) -> pd.DataFrame:
    out = pd.DataFrame({
        'verification_tier': '★',
        'geo_id': df['GEO_ID'],
        'geography_type': geography_type,
        'name': df['NAME'],
        'state_abbreviation': df['STUSAB'].replace({'US': pd.NA}),
        'state_fips': df['STATE'],
        'county_fips': df['COUNTY'],
        'place_fips': df['PLACE'],
        'total_housing_units_est': df['B25001_E001'],
        'total_housing_units_moe_90': df['B25001_M001'],
        'vacant_housing_units_est': df['B25004_E001'],
        'vacant_housing_units_moe_90': df['B25004_M001'],
        'seasonal_recreational_occasional_units_est': df['B25004_E006'],
        'seasonal_recreational_occasional_units_moe_90': df['B25004_M006'],
    })
    out['share_of_all_housing_pct'] = ratio_pct(
        out['seasonal_recreational_occasional_units_est'], out['total_housing_units_est']
    )
    out['share_of_vacant_housing_pct'] = ratio_pct(
        out['seasonal_recreational_occasional_units_est'], out['vacant_housing_units_est']
    )
    out['seasonal_count_relative_moe_pct'] = ratio_pct(
        out['seasonal_recreational_occasional_units_moe_90'],
        out['seasonal_recreational_occasional_units_est'],
    )
    out['acs_vintage'] = VINTAGE
    out['table_total_housing'] = 'B25001'
    out['table_vacancy_status'] = 'B25004'
    out['seasonal_variable'] = 'B25004_006E'
    out['seasonal_moe_variable'] = 'B25004_006M'
    out['verified_date'] = VERIFIED_DATE
    out['source_b25001_url'] = SOURCE_URLS['b25001_2024']
    out['source_b25004_url'] = SOURCE_URLS['b25004_2024']
    out['source_geography_url'] = SOURCE_URLS['geos_2024']
    return out


def base_transform_annual(df: pd.DataFrame, geography_type: str) -> pd.DataFrame:
    out = pd.DataFrame({
        'verification_tier': '★',
        'geo_id': df['GEO_ID'],
        'geography_type': geography_type,
        'name': df['NAME'],
        'state_abbreviation': df['STUSAB'].replace({'US': pd.NA}),
        'state_fips': df['STATE'],
        'county_fips': df['COUNTY'],
        'place_fips': df['PLACE'],
        'total_housing_units_est': df['B25001_E001'],
        'total_housing_units_moe_90': df['B25001_M001'],
        'vacant_housing_units_est': df['B25004_E001'],
        'vacant_housing_units_moe_90': df['B25004_M001'],
        'seasonal_recreational_occasional_units_est': df['B25004_E006'],
        'seasonal_recreational_occasional_units_moe_90': df['B25004_M006'],
    })
    out['share_of_all_housing_pct'] = ratio_pct(
        out['seasonal_recreational_occasional_units_est'], out['total_housing_units_est']
    )
    out['share_of_vacant_housing_pct'] = ratio_pct(
        out['seasonal_recreational_occasional_units_est'], out['vacant_housing_units_est']
    )
    out['seasonal_count_relative_moe_pct'] = ratio_pct(
        out['seasonal_recreational_occasional_units_moe_90'],
        out['seasonal_recreational_occasional_units_est'],
    )
    out['acs_vintage'] = ANNUAL_VINTAGE
    out['table_total_housing'] = 'B25001'
    out['table_vacancy_status'] = 'B25004'
    out['seasonal_variable'] = 'B25004_006E'
    out['seasonal_moe_variable'] = 'B25004_006M'
    out['verified_date'] = VERIFIED_DATE
    out['source_b25001_url'] = SOURCE_URLS['b25001_2024_1yr']
    out['source_b25004_url'] = SOURCE_URLS['b25004_2024_1yr']
    out['source_geography_url'] = SOURCE_URLS['geos_2024_1yr']
    return out


def add_ranks(df: pd.DataFrame, county: bool = False) -> pd.DataFrame:
    out = df.copy()
    out['rank_by_seasonal_count'] = (
        out['seasonal_recreational_occasional_units_est']
        .rank(method='min', ascending=False)
        .astype('Int64')
    )
    out['rank_by_share_of_all_housing'] = (
        out['share_of_all_housing_pct'].rank(method='min', ascending=False).astype('Int64')
    )
    if county:
        eligible = out['total_housing_units_est'].ge(10_000)
        out['rank_by_share_housing_units_ge_10000'] = pd.Series(pd.NA, index=out.index, dtype='Int64')
        out.loc[eligible, 'rank_by_share_housing_units_ge_10000'] = (
            out.loc[eligible, 'share_of_all_housing_pct']
            .rank(method='min', ascending=False)
            .astype('Int64')
        )
    return out


def add_prior_state_data(current: pd.DataFrame, prior_raw: pd.DataFrame) -> pd.DataFrame:
    prior = base_transform(prior_raw, 'state')
    prior = prior.rename(columns={
        'total_housing_units_est': 'prior_total_housing_units_est',
        'total_housing_units_moe_90': 'prior_total_housing_units_moe_90',
        'vacant_housing_units_est': 'prior_vacant_housing_units_est',
        'vacant_housing_units_moe_90': 'prior_vacant_housing_units_moe_90',
        'seasonal_recreational_occasional_units_est': 'prior_seasonal_units_est',
        'seasonal_recreational_occasional_units_moe_90': 'prior_seasonal_units_moe_90',
        'share_of_all_housing_pct': 'prior_share_of_all_housing_pct',
        'share_of_vacant_housing_pct': 'prior_share_of_vacant_housing_pct',
    })
    keep = [
        'geo_id', 'prior_total_housing_units_est', 'prior_total_housing_units_moe_90',
        'prior_vacant_housing_units_est', 'prior_vacant_housing_units_moe_90',
        'prior_seasonal_units_est', 'prior_seasonal_units_moe_90',
        'prior_share_of_all_housing_pct', 'prior_share_of_vacant_housing_pct',
    ]
    out = current.merge(prior[keep], on='geo_id', how='left', validate='one_to_one')
    out['seasonal_count_change_2015_2019_to_2020_2024'] = (
        out['seasonal_recreational_occasional_units_est'] - out['prior_seasonal_units_est']
    )
    out['seasonal_count_pct_change_2015_2019_to_2020_2024'] = ratio_pct(
        out['seasonal_count_change_2015_2019_to_2020_2024'], out['prior_seasonal_units_est']
    )
    out['share_point_change_2015_2019_to_2020_2024'] = (
        out['share_of_all_housing_pct'] - out['prior_share_of_all_housing_pct']
    )
    out['comparison_period'] = f'{PRIOR_VINTAGE} to {VINTAGE}'
    return out


def reorder(df: pd.DataFrame, geography_type: str) -> pd.DataFrame:
    base = [
        'verification_tier', 'geo_id', 'geography_type', 'name', 'state_abbreviation',
        'state_fips', 'county_fips', 'place_fips', 'total_housing_units_est',
        'total_housing_units_moe_90', 'vacant_housing_units_est',
        'vacant_housing_units_moe_90', 'seasonal_recreational_occasional_units_est',
        'seasonal_recreational_occasional_units_moe_90', 'share_of_all_housing_pct',
        'share_of_vacant_housing_pct', 'seasonal_count_relative_moe_pct',
        'rank_by_seasonal_count', 'rank_by_share_of_all_housing',
    ]
    if geography_type == 'county':
        base += ['rank_by_share_housing_units_ge_10000']
    prior_cols = [
        'prior_total_housing_units_est', 'prior_total_housing_units_moe_90',
        'prior_vacant_housing_units_est', 'prior_vacant_housing_units_moe_90',
        'prior_seasonal_units_est', 'prior_seasonal_units_moe_90',
        'prior_share_of_all_housing_pct', 'prior_share_of_vacant_housing_pct',
        'seasonal_count_change_2015_2019_to_2020_2024',
        'seasonal_count_pct_change_2015_2019_to_2020_2024',
        'share_point_change_2015_2019_to_2020_2024', 'comparison_period',
    ]
    base += [c for c in prior_cols if c in df.columns]
    base += [
        'acs_vintage', 'table_total_housing', 'table_vacancy_status',
        'seasonal_variable', 'seasonal_moe_variable', 'verified_date',
        'source_b25001_url', 'source_b25004_url', 'source_geography_url',
    ]
    return df[base]


def json_safe_records(df: pd.DataFrame) -> list[dict]:
    # Pandas serializes missing values as JSON null, including nullable dtypes.
    return json.loads(df.to_json(orient='records', date_format='iso', force_ascii=False))


def write_methodology(n_states: int, n_counties: int, n_idaho: int) -> None:
    text = f"""# Vacation Home Statistics by State and County — Dataset Methodology

**Version / verification date:** {VERIFIED_DATE}  
**Complete state-and-county vintage:** {VINTAGE}  
**Supplemental state-only annual vintage:** {ANNUAL_VINTAGE}  
**Comparison vintage:** {PRIOR_VINTAGE}  
**Geographic scope:** United States; 50 states and the District of Columbia; {n_counties:,} counties and county equivalents in those jurisdictions; all {n_idaho} Idaho counties; Island Park city, Idaho  
**Verification tier:** ★ Every published row was read directly from the cited U.S. Census Bureau files or calculated from those rows. No provisional ● rows are included.

## What this dataset measures

The measurable category is the American Community Survey vacancy-status estimate **“For seasonal, recreational, or occasional use”** (Detailed Table B25004, estimate variable `B25004_006E`; 90% margin of error variable `B25004_006M`). The Census Bureau has historically described this class as housing often referred to as vacation homes. It is not a count of property owners, second-home-owning households, bookings, or currently active Airbnb/Vrbo listings.

The denominator for share of all housing is Detailed Table B25001, `B25001_001E` (total housing units). The denominator for share of vacant housing is B25004, `B25004_001E` (all vacant housing units).

The complete national, state, and county ranking uses the 2020–2024 ACS 5-year estimates because that is the single current Census product covering every county and county equivalent. A separate 2024 ACS 1-year nation/state file is published for readers who need the latest annual state-only estimates. The two products must not be mixed in one ranking or treated as identical measurements.

## Official input files

Current 2020–2024 ACS 5-year files:

- B25001: {SOURCE_URLS['b25001_2024']}
- B25004: {SOURCE_URLS['b25004_2024']}
- Geography labels: {SOURCE_URLS['geos_2024']}

Supplemental 2024 ACS 1-year nation/state files:

- B25001: {SOURCE_URLS['b25001_2024_1yr']}
- B25004: {SOURCE_URLS['b25004_2024_1yr']}
- Geography labels: {SOURCE_URLS['geos_2024_1yr']}

Prior 2015–2019 ACS 5-year files used only for nation/state comparison:

- B25001: {SOURCE_URLS['b25001_2019']}
- B25004: {SOURCE_URLS['b25004_2019']}
- Geography labels: {SOURCE_URLS['geos_2019']}

## Processing steps

1. Read B25001 and B25004 table-based Summary File records.
2. Retain standard geography component `00` from the geography-label files.
3. Join each table to the geography labels on the Census `GEO_ID`.
4. Retain summary levels `010` (nation), `040` (state), `050` (county/county equivalent), and the Island Park row from `160` (place).
5. Exclude Puerto Rico from the 50-state-plus-D.C. rankings and from the county-equivalent file. Puerto Rico remains available from the underlying Census source but is outside the stated ranking universe.
6. Calculate `share_of_all_housing_pct = B25004_006E / B25001_001E × 100`.
7. Calculate `share_of_vacant_housing_pct = B25004_006E / B25004_001E × 100`.
8. Retain the Census-published 90% margin of error for each count. The dataset does not publish a derived ratio margin of error; it instead provides the source count MOE and its relative size.
9. Rank state and county rows separately by count and by share. Ties receive the same minimum rank.
10. Publish a second county share rank limited to county equivalents with at least 10,000 total housing units. This is an editorial display filter—not a Census classification—and is included alongside, not instead of, the unfiltered ranking.
11. Join the non-overlapping {PRIOR_VINTAGE} estimates to nation/state rows for a point-estimate comparison. Interpret changes cautiously and account for Census comparability guidance.
12. Build a separate nation/state supplement from the {ANNUAL_VINTAGE}. Do not combine its values with the all-county 5-year ranking.

## Output coverage

- Complete-comparison state/D.C. file: {n_states} rows.
- Supplemental 2024 annual nation/state file: 52 rows (nation plus 50 states and D.C.).
- County/county-equivalent file: {n_counties:,} rows.
- Idaho county file: {n_idaho} rows.
- Local file: United States, Idaho, Fremont County, and Island Park city.
- JSON file: metadata plus the annual nation/state supplement and the complete state, county, Idaho, and local records.

## Ranking and precision rules

- Rankings use unrounded values; displayed percentages may be rounded afterward.
- All Census estimates are sample estimates, not exact administrative counts.
- ACS margins of error are published at the 90% confidence level.
- A high share based on a small housing denominator can be real but unstable. The unfiltered list remains available, while the 10,000-unit display rank offers a larger-denominator comparison.
- `seasonal_count_relative_moe_pct` is the published count MOE divided by the count estimate. It is a diagnostic, not a confidence interval for the calculated share.

## Reproduction and integrity

Run `build_vacation_home_dataset.py` after placing the nine official input files in `/mnt/data`. The script checks row counts, national totals, local spot checks, ratio bounds, populated rank fields, and required source/date metadata. SHA-256 hashes in `vacation-home-statistics-sha256.txt` identify the exact published files.

## Limitations

- The 2020–2024 ACS 5-year estimate summarizes survey responses collected across five years; it is not a July 2026 point-in-time inventory.
- The 2024 ACS 1-year supplement and 2020–2024 ACS 5-year tables are different survey products. Differences between them do not, by themselves, represent change over time.
- The formal vacancy-status category is broader than active short-term-rental inventory and does not prove a unit is listed, rentable, licensed, or available.
- A housing unit can be categorized as seasonal/recreational/occasional use without establishing ownership structure or the owner’s primary residence.
- County-equivalent definitions differ by state and can change. Connecticut’s current ACS county-equivalent geography uses planning regions; historical county comparisons require matched-boundary treatment.
- Point-estimate differences between ACS periods do not establish causes.
- The Census Bureau can issue errata. Rebuild the dataset after a relevant revision.

## Neutral attribution reference

Organization: Island Park Property Management Research  
Page title: Vacation Home Statistics by State and County: Latest U.S. Census Data  
Last verified: July 29, 2026  
Underlying data: U.S. Census Bureau, {VINTAGE}, Detailed Tables B25001 and B25004; supplemental state-only table from {ANNUAL_VINTAGE}.
"""
    OUT_FILES['methodology'].write_text(text, encoding='utf-8')


def sha256(path: Path) -> str:
    h = hashlib.sha256()
    with path.open('rb') as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b''):
            h.update(chunk)
    return h.hexdigest()


def validate(nation: pd.DataFrame, states: pd.DataFrame, annual_nation_states: pd.DataFrame,
             counties: pd.DataFrame, idaho: pd.DataFrame, local: pd.DataFrame) -> list[str]:
    checks: list[str] = []
    def assert_check(condition: bool, label: str) -> None:
        if not condition:
            raise AssertionError(label)
        checks.append(label)

    assert_check(len(nation) == 1, 'one national row')
    assert_check(len(states) == 51, '51 state/D.C. rows')
    assert_check(len(annual_nation_states) == 52, '52 annual nation/state rows')
    assert_check(len(counties) == 3144, '3,144 county/county-equivalent rows excluding Puerto Rico')
    assert_check(len(idaho) == 44, '44 Idaho county rows')
    assert_check(len(local) == 4, 'four local-summary rows')
    us = nation.iloc[0]
    assert_check(int(us['seasonal_recreational_occasional_units_est']) == 4_743_227,
                 'national seasonal estimate matches B25004')
    assert_check(int(us['total_housing_units_est']) == 143_775_355,
                 'national total housing estimate matches B25001')
    annual_us = annual_nation_states.loc[annual_nation_states['geography_type'].eq('nation')].iloc[0]
    assert_check(int(annual_us['seasonal_recreational_occasional_units_est']) == 4_342_127,
                 '2024 annual national seasonal estimate spot check')
    annual_fl = annual_nation_states.loc[annual_nation_states['state_abbreviation'].eq('FL')].iloc[0]
    assert_check(int(annual_fl['seasonal_recreational_occasional_units_est']) == 741_429,
                 '2024 annual Florida spot check')
    fl = states.loc[states['state_abbreviation'].eq('FL')].iloc[0]
    assert_check(int(fl['seasonal_recreational_occasional_units_est']) == 802_265,
                 'Florida spot check')
    me = states.loc[states['state_abbreviation'].eq('ME')].iloc[0]
    assert_check(round(float(me['share_of_all_housing_pct']), 3) == 15.303,
                 'Maine share spot check')
    fremont = counties.loc[(counties['state_abbreviation'].eq('ID')) & counties['name'].str.startswith('Fremont County')].iloc[0]
    assert_check(int(fremont['seasonal_recreational_occasional_units_est']) == 3_844,
                 'Fremont County spot check')
    island = local.loc[local['geography_type'].eq('place')].iloc[0]
    assert_check(int(island['seasonal_recreational_occasional_units_est']) == 426,
                 'Island Park city spot check')
    annual_state_rows = annual_nation_states.loc[annual_nation_states['geography_type'].eq('state')]
    assert_check(
        states['rank_by_seasonal_count'].notna().all()
        and states['rank_by_share_of_all_housing'].notna().all()
        and counties['rank_by_seasonal_count'].notna().all()
        and counties['rank_by_share_of_all_housing'].notna().all()
        and annual_state_rows['rank_by_seasonal_count'].notna().all()
        and annual_state_rows['rank_by_share_of_all_housing'].notna().all(),
        'rank fields populated',
    )

    frames = {
        'nation': nation,
        'states': states,
        'annual nation/states': annual_nation_states,
        'counties': counties,
        'Idaho counties': idaho,
        'local summary': local,
    }
    for label, frame in frames.items():
        for share_col in ['share_of_all_housing_pct', 'share_of_vacant_housing_pct']:
            valid = frame[share_col].isna() | frame[share_col].between(0, 100, inclusive='both')
            assert_check(bool(valid.all()), f'{label} {share_col} within 0–100')
        assert_check(frame['verified_date'].eq(VERIFIED_DATE).all(), f'{label} verification date populated')
        required_source_cols = ['source_b25001_url', 'source_b25004_url', 'source_geography_url']
        source_ok = frame[required_source_cols].notna().all(axis=None)
        assert_check(bool(source_ok), f'{label} source URLs populated')
    return checks


def main() -> None:
    require_inputs()
    current_raw = load_vintage(2024)
    annual_raw = load_annual_2024()
    prior_raw = load_vintage(2019)

    nation_raw = current_raw[current_raw['SUMLEVEL'].eq('010')].copy()
    state_raw = current_raw[current_raw['SUMLEVEL'].eq('040') & current_raw['STATE'].ne('72')].copy()
    county_raw = current_raw[current_raw['SUMLEVEL'].eq('050') & current_raw['STATE'].ne('72')].copy()
    place_raw = current_raw[current_raw['SUMLEVEL'].eq('160')].copy()

    annual_nation_raw = annual_raw[annual_raw['SUMLEVEL'].eq('010')].copy()
    annual_state_raw = annual_raw[annual_raw['SUMLEVEL'].eq('040') & annual_raw['STATE'].ne('72')].copy()

    prior_nation_raw = prior_raw[prior_raw['SUMLEVEL'].eq('010')].copy()
    prior_state_raw = prior_raw[prior_raw['SUMLEVEL'].eq('040') & prior_raw['STATE'].ne('72')].copy()

    nation = base_transform(nation_raw, 'nation')
    nation = add_prior_state_data(nation, prior_nation_raw)
    nation['rank_by_seasonal_count'] = pd.Series(pd.NA, index=nation.index, dtype='Int64')
    nation['rank_by_share_of_all_housing'] = pd.Series(pd.NA, index=nation.index, dtype='Int64')

    annual_nation = base_transform_annual(annual_nation_raw, 'nation')
    annual_nation['rank_by_seasonal_count'] = pd.Series(pd.NA, index=annual_nation.index, dtype='Int64')
    annual_nation['rank_by_share_of_all_housing'] = pd.Series(pd.NA, index=annual_nation.index, dtype='Int64')
    annual_states = add_ranks(base_transform_annual(annual_state_raw, 'state'))
    annual_states = annual_states.sort_values(['state_fips', 'name']).reset_index(drop=True)
    annual_nation_states = pd.concat([annual_nation, annual_states], ignore_index=True, sort=False)

    states = base_transform(state_raw, 'state')
    states = add_ranks(states)
    states = add_prior_state_data(states, prior_state_raw)
    states = states.sort_values(['state_fips', 'name']).reset_index(drop=True)

    counties = base_transform(county_raw, 'county')
    counties = add_ranks(counties, county=True)
    counties = counties.sort_values(['state_fips', 'county_fips', 'name']).reset_index(drop=True)

    idaho = counties[counties['state_abbreviation'].eq('ID')].copy().reset_index(drop=True)

    island_raw = place_raw[
        place_raw['STATE'].eq('16') & place_raw['NAME'].str.startswith('Island Park city', na=False)
    ].copy()
    if len(island_raw) != 1:
        raise AssertionError(f'Expected one Island Park city row, got {len(island_raw)}')
    island = base_transform(island_raw, 'place')

    idaho_state = states[states['state_abbreviation'].eq('ID')].copy()
    fremont = counties[(counties['state_abbreviation'].eq('ID')) & counties['name'].str.startswith('Fremont County')].copy()
    local = pd.concat([nation, idaho_state, fremont, island], ignore_index=True, sort=False)

    nation = reorder(nation, 'state')
    annual_nation_states = reorder(annual_nation_states, 'state')
    states = reorder(states, 'state')
    counties = reorder(counties, 'county')
    idaho = reorder(idaho, 'county')
    # Local intentionally uses union of current columns; preserve relevant ranks where available.
    local_cols = [c for c in counties.columns if c in local.columns]
    local = local[local_cols]

    checks = validate(nation, states, annual_nation_states, counties, idaho, local)

    # UTF-8 with BOM improves direct spreadsheet compatibility.
    states.to_csv(OUT_FILES['states'], index=False, encoding='utf-8-sig')
    annual_nation_states.to_csv(OUT_FILES['annual_states'], index=False, encoding='utf-8-sig')
    counties.to_csv(OUT_FILES['counties'], index=False, encoding='utf-8-sig')
    idaho.to_csv(OUT_FILES['idaho'], index=False, encoding='utf-8-sig')
    local.to_csv(OUT_FILES['local'], index=False, encoding='utf-8-sig')

    metadata = {
        'title': 'Vacation Home Statistics by State and County: Latest U.S. Census Data',
        'creator': 'Island Park Property Management Research',
        'verified_date': VERIFIED_DATE,
        'acs_vintage': VINTAGE,
        'supplemental_annual_vintage': ANNUAL_VINTAGE,
        'comparison_vintage': PRIOR_VINTAGE,
        'definition': 'Vacant housing units for seasonal, recreational, or occasional use.',
        'ranking_universe': '50 states and District of Columbia; 3,144 counties/county equivalents; Puerto Rico excluded.',
        'verification_tier': '★ Directly read from or calculated from cited primary Census files.',
        'row_counts': {'states_and_dc': len(states), 'annual_nation_and_states': len(annual_nation_states), 'counties_and_equivalents': len(counties), 'idaho_counties': len(idaho)},
        'variables': {
            'seasonal_estimate': 'B25004_006E',
            'seasonal_moe_90': 'B25004_006M',
            'vacant_total_estimate': 'B25004_001E',
            'total_housing_estimate': 'B25001_001E',
        },
        'source_urls': SOURCE_URLS,
        'validation_checks': checks,
    }
    payload = {
        'metadata': metadata,
        'national': json_safe_records(nation),
        'states_and_dc': json_safe_records(states),
        'annual_2024_nation_and_states': json_safe_records(annual_nation_states),
        'counties_and_equivalents': json_safe_records(counties),
        'idaho_counties': json_safe_records(idaho),
        'local_summary': json_safe_records(local),
    }
    OUT_FILES['json'].write_text(json.dumps(payload, ensure_ascii=False, indent=2, allow_nan=False), encoding='utf-8')

    write_methodology(len(states), len(counties), len(idaho))

    checksum_targets = [
        OUT_FILES['states'], OUT_FILES['annual_states'], OUT_FILES['counties'], OUT_FILES['idaho'], OUT_FILES['local'],
        OUT_FILES['json'], OUT_FILES['methodology'], OUT / 'build_vacation_home_dataset.py',
    ]
    checksum_text = ''.join(f'{sha256(p)}  {p.name}\n' for p in checksum_targets)
    OUT_FILES['checksums'].write_text(checksum_text, encoding='utf-8')

    bundle_targets = checksum_targets + [OUT_FILES['checksums']]
    with zipfile.ZipFile(OUT_FILES['bundle'], 'w', compression=zipfile.ZIP_DEFLATED, compresslevel=9) as zf:
        for p in bundle_targets:
            zf.write(p, arcname=p.name)

    print('Validation checks passed:')
    for check in checks:
        print(f'  ✓ {check}')
    print('\nOutputs:')
    for key, path in OUT_FILES.items():
        print(f'  {key}: {path} ({path.stat().st_size:,} bytes)')


if __name__ == '__main__':
    main()