""" Warehouse Statistics by State — Analysis Pipeline Columbus Dock Door Repair Research Version 1.0 — July 22, 2026 This script documents the extraction, filtering, and calculation methodology for the U.S. Warehouse Statistics by State, 2024 dataset. Primary source: BLS QCEW Open Data API (final 2024 annual area files) Output: us-warehouse-statistics-by-state-2024.csv IMPORTANT: This script is provided for methodological transparency. BLS QCEW data are published in CSV format at data.bls.gov/cew/data/api/. """ import csv import hashlib import urllib.request from pathlib import Path # ─── Configuration ──────────────────────────────────────────────────────────── YEAR = "2024" QTR = "A" # Annual record OWN = "5" # Private ownership INDUSTRY = "493" # NAICS 493 Warehousing and Storage SIZE = "0" # All establishment sizes BLS_API_BASE = "https://data.bls.gov/cew/data/api/{year}/a/area/{area}.csv" # FIPS codes for all 50 states, D.C., and U.S. national AREA_CODES = [ ("US000", "United States"), ("01000", "Alabama"), ("02000", "Alaska"), ("04000", "Arizona"), ("05000", "Arkansas"), ("06000", "California"), ("08000", "Colorado"), ("09000", "Connecticut"), ("10000", "Delaware"), ("11000", "District of Columbia"), ("12000", "Florida"), ("13000", "Georgia"), ("15000", "Hawaii"), ("16000", "Idaho"), ("17000", "Illinois"), ("18000", "Indiana"), ("19000", "Iowa"), ("20000", "Kansas"), ("21000", "Kentucky"), ("22000", "Louisiana"), ("23000", "Maine"), ("24000", "Maryland"), ("25000", "Massachusetts"), ("26000", "Michigan"), ("27000", "Minnesota"), ("28000", "Mississippi"), ("29000", "Missouri"), ("30000", "Montana"), ("31000", "Nebraska"), ("32000", "Nevada"), ("33000", "New Hampshire"), ("34000", "New Jersey"), ("35000", "New Mexico"), ("36000", "New York"), ("37000", "North Carolina"), ("38000", "North Dakota"), ("39000", "Ohio"), ("40000", "Oklahoma"), ("41000", "Oregon"), ("42000", "Pennsylvania"), ("44000", "Rhode Island"), ("45000", "South Carolina"), ("46000", "South Dakota"), ("47000", "Tennessee"), ("48000", "Texas"), ("49000", "Utah"), ("50000", "Vermont"), ("51000", "Virginia"), ("53000", "Washington"), ("54000", "West Virginia"), ("55000", "Wisconsin"), ("56000", "Wyoming"), ] # ─── BLS CSV column names ────────────────────────────────────────────────────── # Columns from the BLS QCEW annual area CSV: # area_fips, own_code, industry_code, agglvl_code, size_code, year, qtr, # disclosure_code, annual_avg_estabs, annual_avg_emplvl, total_annual_wages, # taxable_annual_wages, annual_contributions, annual_avg_wkly_wage, # avg_annual_pay, lq_disclosure_code, lq_annual_avg_estabs, # lq_annual_avg_emplvl, lq_annual_avg_wkly_wage, lq_avg_annual_pay, # oty_annual_avg_estabs_chg, oty_annual_avg_estabs_pct_chg, # oty_annual_avg_emplvl_chg, oty_annual_avg_emplvl_pct_chg, # oty_total_annual_wages_chg, oty_total_annual_wages_pct_chg, # oty_taxable_annual_wages_chg, oty_taxable_annual_wages_pct_chg, # oty_annual_contributions_chg, oty_annual_contributions_pct_chg, # oty_annual_avg_wkly_wage_chg, oty_annual_avg_wkly_wage_pct_chg, # oty_avg_annual_pay_chg, oty_avg_annual_pay_pct_chg SUPPRESSION_CODE = "N" def fetch_area_file(area_code: str) -> list[dict]: """Download and parse a BLS QCEW annual area file.""" url = BLS_API_BASE.format(year=YEAR, area=area_code) with urllib.request.urlopen(url) as response: content = response.read().decode("utf-8") reader = csv.DictReader(content.splitlines()) return list(reader) def filter_naics_row(rows: list[dict]) -> dict | None: """Extract the single row matching our filter criteria.""" for row in rows: if ( row.get("own_code", "").strip() == OWN and row.get("industry_code", "").strip() == INDUSTRY and row.get("qtr", "").strip() == QTR and row.get("size_code", "").strip() == SIZE ): return row return None def is_suppressed(row: dict) -> bool: """Check if employment data is protected by BLS confidentiality disclosure.""" return row.get("disclosure_code", "").strip() == SUPPRESSION_CODE def safe_int(value: str, suppressed: bool) -> str: """Return value or ND if suppressed or empty.""" if suppressed: return "ND" v = value.strip() return v if v else "ND" def safe_lq(value: str, suppressed: bool) -> str: """Return LQ value or ND if suppressed.""" if suppressed: return "ND" v = value.strip() if not v or v == "0" or v == "0.00": return "ND" return v def calculate_jpe(employment: str, establishments: str) -> str: """Calculate jobs per establishment ratio.""" if employment == "ND" or establishments == "ND": return "ND" try: emp = float(employment.replace(",", "")) estab = float(establishments.replace(",", "")) if estab == 0: return "ND" return f"{emp / estab:.1f}" except ValueError: return "ND" def build_dataset() -> list[dict]: """Fetch all area files and build the dataset.""" records = [] for area_code, state_name in AREA_CODES: print(f"Fetching {state_name} ({area_code})...") rows = fetch_area_file(area_code) row = filter_naics_row(rows) if row is None: print(f" WARNING: No matching row for {state_name}") continue suppressed = is_suppressed(row) source_url = BLS_API_BASE.format(year=YEAR, area=area_code) record = { "state": state_name, "bls_area_code": area_code, "bls_source_url": source_url, "annual_avg_establishments": row.get("annual_avg_estabs", "").strip(), "annual_avg_covered_jobs": safe_int(row.get("annual_avg_emplvl", ""), suppressed), "total_annual_wages": safe_int(row.get("total_annual_wages", ""), suppressed), "avg_annual_pay": safe_int(row.get("avg_annual_pay", ""), suppressed), "avg_weekly_pay": safe_int(row.get("annual_avg_wkly_wage", ""), suppressed), "job_change_2023_2024": safe_int(row.get("oty_annual_avg_emplvl_chg", ""), suppressed), "job_growth_pct_2023_2024": safe_int(row.get("oty_annual_avg_emplvl_pct_chg", ""), suppressed), "employment_lq": safe_lq(row.get("lq_annual_avg_emplvl", ""), suppressed), "notes": "Employment and pay not disclosed (BLS confidentiality protection)" if suppressed else "", } record["jobs_per_establishment"] = calculate_jpe( record["annual_avg_covered_jobs"], record["annual_avg_establishments"] ) records.append(record) return records def assign_ranks(records: list[dict]) -> list[dict]: """Assign employment rank among disclosed jurisdictions, excluding U.S. total.""" state_records = [r for r in records if r["state"] != "United States" and r["annual_avg_covered_jobs"] != "ND"] state_records.sort(key=lambda r: int(r["annual_avg_covered_jobs"].replace(",", "")), reverse=True) for i, r in enumerate(state_records, 1): r["job_rank"] = i for r in records: if "job_rank" not in r: r["job_rank"] = "" return records def compute_sha256(filepath: Path) -> str: """Compute SHA-256 of the output CSV file.""" h = hashlib.sha256() with open(filepath, "rb") as f: for chunk in iter(lambda: f.read(8192), b""): h.update(chunk) return h.hexdigest() def write_csv(records: list[dict], output_path: Path) -> None: """Write the dataset to CSV.""" fieldnames = [ "job_rank", "state", "bls_area_code", "bls_source_url", "annual_avg_establishments", "annual_avg_covered_jobs", "total_annual_wages", "avg_annual_pay", "avg_weekly_pay", "job_change_2023_2024", "job_growth_pct_2023_2024", "employment_lq", "jobs_per_establishment", "notes", ] with open(output_path, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="ignore") writer.writeheader() writer.writerows(records) if __name__ == "__main__": output_path = Path("us-warehouse-statistics-by-state-2024.csv") print("Building U.S. Warehouse Statistics by State dataset, 2024...") records = build_dataset() records = assign_ranks(records) write_csv(records, output_path) sha256 = compute_sha256(output_path) print(f"\nDataset written to: {output_path}") print(f"SHA-256: {sha256}") print(f"Rows: {len(records)}") print("\nVerify against published checksum:") print(" c4064f138cc8eee1cf83870cfd9d0a01ec6c8d2c8293102fa63aa58cf84fbe59")