#!/usr/bin/env python3 """Acquire official labour-market series and build publication-ready extracts. All network calls are reproducible and point to first-party statistical sources. LinkedIn Economic Graph values are transcribed from the cited publication tables; the source PDFs are not redistributed. """ from __future__ import annotations import csv import hashlib import io import itertools import json import math import subprocess import sys from datetime import datetime, timezone from pathlib import Path from typing import Any from openpyxl import load_workbook ROOT = Path(__file__).resolve().parent RAW = ROOT / "raw" DERIVED = ROOT / "derived" RAW.mkdir(parents=True, exist_ok=True) DERIVED.mkdir(parents=True, exist_ok=True) REFRESH = "--refresh" in sys.argv def curl_bytes(url: str, *, method: str = "GET", payload: dict[str, Any] | None = None) -> bytes: cmd = [ "curl", "-L", "--fail", "--silent", "--show-error", "--max-time", "90", "-A", "Mozilla/5.0 (compatible; SchymResearch/1.0)", ] if method == "POST": cmd += ["-X", "POST", "-H", "Content-Type: application/json", "-d", json.dumps(payload)] cmd.append(url) return subprocess.check_output(cmd) def write_bytes(path: Path, data: bytes) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(data) def write_csv(path: Path, rows: list[dict[str, Any]], fieldnames: list[str] | None = None) -> None: path.parent.mkdir(parents=True, exist_ok=True) if fieldnames is None: fieldnames = [] seen = set() for row in rows: for key in row: if key not in seen: fieldnames.append(key) seen.add(key) with path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames) writer.writeheader() writer.writerows(rows) def jsonstat_rows(dataset: dict[str, Any]) -> list[dict[str, Any]]: dim_ids = dataset["id"] sizes = dataset["size"] code_lists: list[list[str]] = [] for dim_id in dim_ids: index = dataset["dimension"][dim_id]["category"]["index"] if isinstance(index, list): codes = index else: codes = [code for code, _ in sorted(index.items(), key=lambda item: item[1])] code_lists.append(codes) values = dataset.get("value", {}) statuses = dataset.get("status", {}) or {} rows: list[dict[str, Any]] = [] for coords in itertools.product(*[range(size) for size in sizes]): flat = 0 for coord, size in zip(coords, sizes): flat = flat * size + coord key = str(flat) if key not in values: continue row = {dim_id: code_lists[i][coords[i]] for i, dim_id in enumerate(dim_ids)} row["value"] = values[key] if key in statuses: row["status"] = statuses[key] rows.append(row) return rows def fetch_bls() -> list[dict[str, Any]]: series = { "CES0000000001": "payroll_employment_thousands", "LNS12000000": "household_employment_thousands", "LNS14000000": "unemployment_rate_percent", "JTS000000000000000JOL": "job_openings_thousands", "JTS000000000000000HIL": "hires_thousands", "JTS000000000000000QUL": "quits_thousands", "JTS000000000000000JOR": "job_openings_rate_percent", "JTS000000000000000HIR": "hires_rate_percent", "JTS000000000000000QUR": "quits_rate_percent", } payload = {"seriesid": list(series), "startyear": "2019", "endyear": "2026"} raw_path = RAW / "bls_timeseries_2019_2026.json" data = raw_path.read_bytes() if raw_path.exists() and not REFRESH else curl_bytes("https://api.bls.gov/publicAPI/v2/timeseries/data/", method="POST", payload=payload) write_bytes(raw_path, data) parsed = json.loads(data) if parsed.get("status") != "REQUEST_SUCCEEDED": raise RuntimeError(parsed) rows: list[dict[str, Any]] = [] for item in parsed["Results"]["series"]: label = series[item["seriesID"]] for obs in item["data"]: period = obs["period"] if period == "M13": continue try: numeric_value = float(obs["value"]) except (TypeError, ValueError): continue month = int(period[1:]) rows.append({ "series_id": item["seriesID"], "series": label, "date": f"{int(obs['year']):04d}-{month:02d}-01", "value": numeric_value, "status": "preliminary" if "P" in (obs.get("footnotes") or [{}])[0].get("code", "") else "observed", }) rows.sort(key=lambda row: (row["date"], row["series"])) write_csv(DERIVED / "bls_us_monthly.csv", rows) return rows def eurostat_url(dataset: str, geo: str, since: str) -> str: return ( f"https://ec.europa.eu/eurostat/api/dissemination/statistics/1.0/data/{dataset}" f"?lang=en&geo={geo}&sinceTimePeriod={since}" ) def fetch_eurostat() -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[dict[str, Any]]]: vacancy_rows: list[dict[str, Any]] = [] employment_rows: list[dict[str, Any]] = [] unemployment_rows: list[dict[str, Any]] = [] for geo in ["DE", "FR", "NL"]: for dataset, since, target in [ ("jvs_q_r21", "2019-Q4", vacancy_rows), ("namq_10_pe", "2019-Q4", employment_rows), ("une_rt_m", "2026-01", unemployment_rows), ]: raw_path = RAW / f"eurostat_{dataset}_{geo}.json" data = raw_path.read_bytes() if raw_path.exists() and not REFRESH else curl_bytes(eurostat_url(dataset, geo, since)) write_bytes(raw_path, data) target.extend(jsonstat_rows(json.loads(data))) vacancy_rows = [ row for row in vacancy_rows if row["sizeclas"] == "TOTAL" and row["s_adj"] == "SA" and row["indic_em"] == "JVR" and row["nace_r2_1"] in {"A-T", "B-T", "C", "F", "I", "J", "K", "L", "M", "N", "O"} ] employment_rows = [ row for row in employment_rows if row["unit"] == "THS_PER" and row["s_adj"] in {"SA", "SCA"} and row["na_item"] == "EMP_NC" ] unemployment_rows = [ row for row in unemployment_rows if row["s_adj"] == "SA" and row["age"] == "TOTAL" and row["unit"] == "PC_ACT" and row["sex"] == "T" ] vacancy_rows.sort(key=lambda row: (row["geo"], row["nace_r2_1"], row["time"])) employment_rows.sort(key=lambda row: (row["geo"], row["time"])) unemployment_rows.sort(key=lambda row: (row["geo"], row["time"])) write_csv(DERIVED / "eurostat_vacancy_rates.csv", vacancy_rows) write_csv(DERIVED / "eurostat_employment.csv", employment_rows) write_csv(DERIVED / "eurostat_unemployment.csv", unemployment_rows) return vacancy_rows, employment_rows, unemployment_rows def fetch_ons_vacancies() -> list[dict[str, Any]]: url = ( "https://www.ons.gov.uk/file?uri=%2Femploymentandlabourmarket%2Fpeoplenotinwork%2F" "unemployment%2Fdatasets%2Fvacanciesandunemploymentvacs01%2Fcurrent%2Fvacs01aug2026.xlsx" ) raw_path = RAW / "ons_vacs01_aug2026.xlsx" data = raw_path.read_bytes() if raw_path.exists() and not REFRESH else curl_bytes(url) write_bytes(raw_path, data) wb = load_workbook(io.BytesIO(data), read_only=True, data_only=True) ws = wb["VACS01"] rows: list[dict[str, Any]] = [] for values in ws.iter_rows(min_row=7, values_only=True): period = values[0] vacancies = values[2] unemployed = values[3] ratio = values[4] if not period or not isinstance(vacancies, (int, float)): continue rows.append({ "period": str(period).strip(), "vacancies_thousands": float(vacancies), "unemployed_thousands": float(unemployed) if isinstance(unemployed, (int, float)) else "", "unemployed_per_vacancy": float(ratio) if isinstance(ratio, (int, float)) else "", }) write_csv(DERIVED / "ons_uk_vacancies.csv", rows) return rows def linked_in_extracts() -> None: freeze = [ {"country":"Germany","iso":"DE","month":"2026-02","postings_per_applicant_yoy":-17.8,"hiring_rate_yoy":-17.8,"applications_per_applicant_yoy":-12.3,"hiring_index_2016":0.96}, {"country":"Germany","iso":"DE","month":"2026-05","postings_per_applicant_yoy":-18.5,"hiring_rate_yoy":-20.2,"applications_per_applicant_yoy":-23.0,"hiring_index_2016":0.89}, {"country":"France","iso":"FR","month":"2026-02","postings_per_applicant_yoy":-11.5,"hiring_rate_yoy":-17.4,"applications_per_applicant_yoy":-12.7,"hiring_index_2016":0.82}, {"country":"France","iso":"FR","month":"2026-05","postings_per_applicant_yoy":-16.3,"hiring_rate_yoy":-17.5,"applications_per_applicant_yoy":-26.0,"hiring_index_2016":0.79}, {"country":"Netherlands","iso":"NL","month":"2026-02","postings_per_applicant_yoy":-9.0,"hiring_rate_yoy":-19.8,"applications_per_applicant_yoy":-7.3,"hiring_index_2016":0.68}, {"country":"Netherlands","iso":"NL","month":"2026-05","postings_per_applicant_yoy":-20.9,"hiring_rate_yoy":-17.8,"applications_per_applicant_yoy":-20.7,"hiring_index_2016":0.65}, {"country":"United Kingdom","iso":"UK","month":"2026-02","postings_per_applicant_yoy":0.7,"hiring_rate_yoy":-9.6,"applications_per_applicant_yoy":-15.0,"hiring_index_2016":0.88}, {"country":"United Kingdom","iso":"UK","month":"2026-05","postings_per_applicant_yoy":-7.1,"hiring_rate_yoy":-10.7,"applications_per_applicant_yoy":-27.6,"hiring_index_2016":0.83}, {"country":"United States","iso":"US","month":"2026-05","postings_per_applicant_yoy":-7.0,"hiring_rate_yoy":-4.8,"applications_per_applicant_yoy":-35.0,"hiring_index_2016":""}, ] write_csv(DERIVED / "linkedin_freeze_vectors.csv", freeze) first_rung = [ {"country":"France","iso":"FR","overall_hiring_vs_prepandemic":-36,"advanced_degree_entry_hiring_vs_prepandemic":-51}, {"country":"Netherlands","iso":"NL","overall_hiring_vs_prepandemic":-38,"advanced_degree_entry_hiring_vs_prepandemic":-48}, {"country":"United States","iso":"US","overall_hiring_vs_prepandemic":-23,"advanced_degree_entry_hiring_vs_prepandemic":-41}, {"country":"Germany","iso":"DE","overall_hiring_vs_prepandemic":-23,"advanced_degree_entry_hiring_vs_prepandemic":-38}, {"country":"United Kingdom","iso":"UK","overall_hiring_vs_prepandemic":-24,"advanced_degree_entry_hiring_vs_prepandemic":-30}, ] write_csv(DERIVED / "linkedin_first_rung.csv", first_rung) remote = [ {"country":"Germany","iso":"DE","remote_posting_share":7.2,"remote_application_share":17.5}, {"country":"United Kingdom","iso":"UK","remote_posting_share":9.4,"remote_application_share":17.2}, {"country":"Ireland","iso":"IE","remote_posting_share":9.4,"remote_application_share":15.6}, {"country":"Netherlands","iso":"NL","remote_posting_share":3.1,"remote_application_share":9.8}, {"country":"France","iso":"FR","remote_posting_share":3.2,"remote_application_share":6.1}, {"country":"Italy","iso":"IT","remote_posting_share":6.7,"remote_application_share":13.0}, {"country":"Sweden","iso":"SE","remote_posting_share":4.6,"remote_application_share":9.4}, ] for row in remote: row["application_to_posting_ratio"] = round(row["remote_application_share"] / row["remote_posting_share"], 2) write_csv(DERIVED / "linkedin_remote_mismatch.csv", remote) prepandemic = [ {"country":"France","iso":"FR","hiring_vs_feb2020":-39.0}, {"country":"Netherlands","iso":"NL","hiring_vs_feb2020":-41.0}, {"country":"Germany","iso":"DE","hiring_vs_feb2020":-27.0}, {"country":"United Kingdom","iso":"UK","hiring_vs_feb2020":-27.0}, {"country":"United States","iso":"US","hiring_vs_feb2020":-25.0}, ] write_csv(DERIVED / "linkedin_hiring_vs_prepandemic.csv", prepandemic) def build_us_stock_flow(bls_rows: list[dict[str, Any]]) -> None: keep = { "payroll_employment_thousands": "Employment", "job_openings_thousands": "Job openings", "hires_thousands": "Hires", "quits_thousands": "Quits", } by_series_date = {(row["series"], row["date"]): row["value"] for row in bls_rows} base_date = "2019-12-01" bases = {series: by_series_date[(series, base_date)] for series in keep} rows: list[dict[str, Any]] = [] for row in bls_rows: series = row["series"] if series not in keep or row["date"] < base_date: continue rows.append({ "date": row["date"], "series": keep[series], "value": row["value"], "index_2019_12_100": round(row["value"] / bases[series] * 100, 2), "status": row["status"], }) rows.sort(key=lambda row: (row["date"], row["series"])) write_csv(DERIVED / "us_stock_flow_index.csv", rows) def build_vacancy_peak_paths(vacancy_rows: list[dict[str, Any]], ons_rows: list[dict[str, Any]], bls_rows: list[dict[str, Any]]) -> None: rows: list[dict[str, Any]] = [] for iso, country in [("DE", "Germany"), ("FR", "France"), ("NL", "Netherlands")]: subset = [r for r in vacancy_rows if r["geo"] == iso and r["nace_r2_1"] == "A-T" and r["time"] >= "2021-Q1"] if not subset: subset = [r for r in vacancy_rows if r["geo"] == iso and r["nace_r2_1"] == "B-T" and r["time"] >= "2021-Q1"] peak = max(float(r["value"]) for r in subset) for r in subset: rows.append({"country": country, "iso": iso, "period": r["time"], "source_measure": "Eurostat vacancy rate", "value": r["value"], "peak_2021_2026_100": round(float(r["value"]) / peak * 100, 2)}) ons_subset = [r for r in ons_rows if any(str(year) in r["period"] for year in range(2021, 2027))] ons_peak = max(float(r["vacancies_thousands"]) for r in ons_subset) for r in ons_subset: rows.append({"country":"United Kingdom","iso":"UK","period":r["period"],"source_measure":"ONS vacancies, 3-month average","value":r["vacancies_thousands"],"peak_2021_2026_100":round(float(r["vacancies_thousands"])/ons_peak*100,2)}) us_subset = [r for r in bls_rows if r["series"] == "job_openings_rate_percent" and r["date"] >= "2021-01-01"] us_peak = max(float(r["value"]) for r in us_subset) for r in us_subset: rows.append({"country":"United States","iso":"US","period":r["date"][:7],"source_measure":"BLS JOLTS job openings rate","value":r["value"],"peak_2021_2026_100":round(float(r["value"])/us_peak*100,2)}) write_csv(DERIVED / "vacancy_paths_peak_100.csv", rows) def build_sector_declines(vacancy_rows: list[dict[str, Any]]) -> None: sector_names = { "C":"Manufacturing", "F":"Construction", "I":"Accommodation & food", "J":"Information & communication", "K":"Finance & insurance", "L":"Real estate", "M":"Professional & scientific", "N":"Administrative support", "O":"Public administration", } out: list[dict[str, Any]] = [] for geo, country in [("DE","Germany"),("FR","France"),("NL","Netherlands")]: for code, sector in sector_names.items(): subset = [r for r in vacancy_rows if r["geo"] == geo and r["nace_r2_1"] == code and r["time"] >= "2021-Q1"] if not subset: continue peak_row = max(subset, key=lambda r: float(r["value"])) latest_row = max(subset, key=lambda r: r["time"]) peak = float(peak_row["value"]) latest = float(latest_row["value"]) if peak <= 0: continue out.append({ "country":country,"iso":geo,"sector":sector,"nace_r2_1":code, "peak_period":peak_row["time"],"peak_rate":peak, "latest_period":latest_row["time"],"latest_rate":latest, "change_from_peak_percent":round((latest/peak-1)*100,1), }) write_csv(DERIVED / "sector_vacancy_declines.csv", out) def build_cross_country_snapshot(employment_rows: list[dict[str, Any]], unemployment_rows: list[dict[str, Any]], bls_rows: list[dict[str, Any]]) -> None: snapshot: dict[str, dict[str, Any]] = { "DE":{"country":"Germany","iso":"DE"}, "FR":{"country":"France","iso":"FR"}, "NL":{"country":"Netherlands","iso":"NL"}, "UK":{"country":"United Kingdom","iso":"UK","employment_millions":34.25,"unemployment_rate":4.9,"employment_note":"ONS LFS, Apr-Jun 2026; rounded"}, "US":{"country":"United States","iso":"US"}, } for iso in ["DE","FR","NL"]: emp = max((r for r in employment_rows if r["geo"] == iso), key=lambda r: r["time"]) unemp = max((r for r in unemployment_rows if r["geo"] == iso and r["time"] <= "2026-05"), key=lambda r: r["time"]) snapshot[iso]["employment_millions"] = round(float(emp["value"]) / 1000, 3) snapshot[iso]["unemployment_rate"] = float(unemp["value"]) snapshot[iso]["employment_note"] = f"Eurostat {emp['time']}" us_emp = max((r for r in bls_rows if r["series"] == "household_employment_thousands" and r["date"] <= "2026-05-01"), key=lambda r: r["date"]) us_unemp = max((r for r in bls_rows if r["series"] == "unemployment_rate_percent" and r["date"] <= "2026-05-01"), key=lambda r: r["date"]) snapshot["US"]["employment_millions"] = round(float(us_emp["value"]) / 1000, 3) snapshot["US"]["unemployment_rate"] = float(us_unemp["value"]) snapshot["US"]["employment_note"] = "BLS household survey, May 2026" write_csv(DERIVED / "country_snapshot.csv", list(snapshot.values())) def write_manifest() -> None: files = [] for path in sorted(ROOT.rglob("*")): if not path.is_file() or path.name == "acquisition_manifest.json": continue files.append({ "path": str(path.relative_to(ROOT)), "bytes": path.stat().st_size, "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), }) manifest = { "created_utc": datetime.now(timezone.utc).isoformat(), "status": "observed-and-derived", "files": files, } (ROOT / "acquisition_manifest.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8") def main() -> None: bls = fetch_bls() vacancies, employment, unemployment = fetch_eurostat() ons = fetch_ons_vacancies() linked_in_extracts() build_us_stock_flow(bls) build_vacancy_peak_paths(vacancies, ons, bls) build_sector_declines(vacancies) build_cross_country_snapshot(employment, unemployment, bls) write_manifest() print(json.dumps({ "bls_rows": len(bls), "eurostat_vacancy_rows": len(vacancies), "eurostat_employment_rows": len(employment), "eurostat_unemployment_rows": len(unemployment), "ons_rows": len(ons), "output": str(ROOT), }, indent=2)) if __name__ == "__main__": try: main() except Exception as exc: print(f"ERROR: {exc}", file=sys.stderr) raise