#!/usr/bin/env python3 """American Worker Index (AWI v3) — family-v1 production engine (national level). The worker-conditions member of the index family. Same Hazen percentile-of-own-history transform as ADI, same orientation rule (every input flagged so higher = worse for the worker), same registry-derived equal-domain mean. Four deliberate differences from ADI: 1. FOUR equal domains, not five: the_paycheck / the_job / the_power / the_divide. 2. Publication rule: a quarter publishes when >= 3 of the 4 domains have at least one present member (ADI requires ALL domains). The composite is the equal-weight mean of the domains PRESENT that quarter. 3. Trailing-edge trim: after publication, the tail is trimmed to the last quarter that has the FULL current member set present, so the latest published reading is never a ragged 1-2-member edge. Interior partial-member quarters (deep history, before a member's series begins) are kept. 4. History spans ~1948-present and carries two ANNUAL members (family income Gini, union membership). An annual observation for year Y is step-carried across Y's four quarters and up to CARRY_QUARTERS further quarters, until the next annual observation supersedes it — a documented carry of a real published value (never an interpolation). This keeps the composite current between annual releases; the carry is capped so a long-stale annual value eventually drops the member and the trailing-edge trim pulls the published tail back rather than resting on year-old data. Run from the repo root: PYTHONPATH=. python3 scripts/indexes/compute_worker_index.py Reads only committed files under data/indicators/**; writes data/indexes/worker_index.json + data/indexes/worker_index.csv + a byte-identical site/public/data/indexes/ mirror. Deterministic, stdlib-only, no network, no wall-clock (last_updated = max of input last_updated). Refuses to write on any validation-gate failure. """ from __future__ import annotations import json import logging import math import sys from dataclasses import dataclass from pathlib import Path from typing import Dict, List, Optional, Sequence, Tuple THIS_DIR = Path(__file__).resolve().parent REPO_ROOT = THIS_DIR.parents[1] OUTPUT_DIR = REPO_ROOT / "data" / "indexes" if str(REPO_ROOT) not in sys.path: sys.path.insert(0, str(REPO_ROOT)) try: from scripts.indexes.family_normalization import hazen_percentiles_series # noqa: E402 from scripts.indicators.schema import date_to_quarter # noqa: E402 except ImportError as exc: # pragma: no cover raise SystemExit( "Cannot import scripts.indexes.family_normalization / " "scripts.indicators.schema. Run from the repo root with PYTHONPATH=." ) from exc logger = logging.getLogger("worker_index") # --------------------------------------------------------------------------- # Registry — the only place members, domains, and paths are declared. # Orientation is read from each JSON's `direction` field at load; the build # fails closed if the field is absent (same rule as compute_adi). # --------------------------------------------------------------------------- TAXONOMY_VERSION = "family-v1" SERIES_ID = "worker" SERIES_NAME = "American Worker Index (AWI)" DOMAIN_IDS: Tuple[str, ...] = ( "the_paycheck", # D1 "the_job", # D2 "the_power", # D3 "the_divide", # D4 ) DOMAIN_LABELS: Dict[str, str] = { "the_paycheck": "The Paycheck", "the_job": "The Job", "the_power": "The Power", "the_divide": "The Divide", } # >= this many of the 4 domains must be present for a quarter to publish. MIN_DOMAINS_TO_PUBLISH = 3 # Quarters an annual member's value is carried FORWARD past its release year's # Q4 (in addition to filling the release year itself), until the next annual # value supersedes it. 4 quarters = one year, so an annual member is never # carried more than a year past its last release before the trailing-edge trim # drops it. CARRY_QUARTERS = 4 N_BANDS = 5 # Uniform 20-point segments; higher band = worse for the worker. National # time-series indexes share one locked label set; place indexes do not use it. BAND_LABELS: Tuple[str, ...] = ("Minimal", "Low", "Typical", "High", "Severe") # Split for the same reason as compute_adi.py — see the note there. # ORIENTABLE gates members and sign math; DECLARED gates the candidate sweep. ORIENTABLE_DIRECTIONS = ("higher_is_worse", "lower_is_worse") DECLARED_DIRECTIONS = ORIENTABLE_DIRECTIONS + ("bidirectional", "not_applicable") ANNUAL_FREQUENCY = "annual" GROWTH_WINDOW_QUARTERS = 20 # 5 years, for the Paycheck growth transform @dataclass(frozen=True) class MemberSpec: input_id: str rel_path: str domain: str transform: str = "level" # "level" | "growth_5y" (5-year annualized % growth) # The composite members, grouped by domain. Counts are NOT written here — they # are derived from this tuple wherever they are published, because a hand-typed # count survives a membership change (F1 dropped a member and "seven members" # stayed in the published methodology until review caught it). Directions are # NOT declared here (read from JSON); the trailing comment records the EXPECTED # direction so a wiring error is obvious in review. # # The three Paycheck members use a 5-year annualized GROWTH transform rather # than the raw level. Two reasons: (1) a level percentile of a slowly-rising # real wage would read as favorable at a record real level and hide the story the # domain exists to tell — that pay growth has decoupled from productivity; and # (2) growth de-trends the series, which is the direct answer to the "a trending # series sits at its percentile extreme by construction" critique. The # direction field still applies unchanged: lower wage GROWTH is worse # (lower_is_worse), a faster-rising wedge is worse (higher_is_worse). The # remaining members are level percentiles (rates, shares, and the union/labor-share # power measures, which carry their meaning as levels). _ENGLISH_NUMBERS: Dict[int, str] = { 1: "one", 2: "two", 3: "three", 4: "four", 5: "five", 6: "six", 7: "seven", 8: "eight", 9: "nine", 10: "ten", } def _english_count(n: int) -> str: """Spell a small count for published methodology prose; digits past ten.""" return _ENGLISH_NUMBERS.get(n, str(n)) WORKER_REGISTRY: Tuple[MemberSpec, ...] = ( # D1 The Paycheck MemberSpec("real_production_wage", "data/indicators/awi/real_production_wage.json", "the_paycheck", "growth_5y"), # lower_is_worse MemberSpec("median_weekly_earnings_real", "data/indicators/awi/median_weekly_earnings_real.json", "the_paycheck", "growth_5y"), # lower_is_worse MemberSpec("productivity_pay_wedge", "data/indicators/awi/productivity_pay_wedge.json", "the_paycheck", "growth_5y"), # higher_is_worse # D2 The Job (unemployment_rate is REUSED from the labor door dir, not a new file) # # longterm_unemployed_share was a third member here until 2026-07-23 (unit F1, # owner decision (7) in docs/plans/INDEX_FAMILY_BUILD_PLAN_2026-07.md). It is # the long-term unemployed as a share OF THE UNEMPLOYED, so its denominator is # itself the distress variable: when a shock throws millions of newly-jobless # people into the denominator, the share COLLAPSES. In April 2020 it fell # 20.0% -> 4.0% while unemployment tripled 3.6% -> 14.8%, so the composite # read the worst jobs month in modern history as one of the best readings the # domain had ever produced. It has no distress pole, and a percentile # composite cannot orient a member that has none. It is now a dispositioned # satellite; the_job rebalances 3 members -> 2 automatically, because # DOMAIN_MEMBERS and member_weight both derive from this tuple. MemberSpec("unemployment_rate", "data/indicators/door4_income/unemployment_rate.json", "the_job"), # higher_is_worse MemberSpec("involuntary_parttime_share", "data/indicators/awi/involuntary_parttime_share.json", "the_job"), # higher_is_worse # D3 The Power MemberSpec("union_membership", "data/indicators/awi/union_membership.json", "the_power"), # lower_is_worse MemberSpec("labor_share", "data/indicators/awi/labor_share.json", "the_power"), # lower_is_worse # D4 The Divide MemberSpec("gini_families", "data/indicators/awi/gini_families.json", "the_divide"), # higher_is_worse MemberSpec("bottom50_wealth_share", "data/indicators/awi/bottom50_wealth_share.json", "the_divide"), # lower_is_worse ) DOMAIN_MEMBERS: Dict[str, Tuple[str, ...]] = { d: tuple(s.input_id for s in WORKER_REGISTRY if s.domain == d) for d in DOMAIN_IDS } # Non-member satellites in data/indicators/awi/ (adi_role="supporting_evidence"), # dispositioned so gate_dispositions can prove exhaustiveness (mirror ADI EXCLUDED). SATELLITES: Tuple[str, ...] = ( "prime_age_epop", "multiple_jobholders", "real_minimum_wage", # Retired from WORKER_REGISTRY 2026-07-23 (unit F1) — share-of-the-unemployed, # so its denominator is the distress variable and it carries no pole. See the # note at its former the_job entry above. Still published as an indicator; # only its composite membership changed. "longterm_unemployed_share", ) # The AI Displacement Tracker series also live in data/indicators/awi/ (door 5) # but belong to a different index; dispositioned so gate_dispositions does not # flag them as undispositioned worker candidates. AI_TRACKER_SERIES: Tuple[str, ...] = ( "ai_adoption", "ai_capability", "ai_job_posting_share", "ai_layoffs", "tech_openings", "youth_unemployment", ) VINTAGE_HONESTY = ( "Inputs are revised series (BLS, BEA, and Federal Reserve constructs via " "FRED, plus the spliced union-membership series). Historical AWI values are " "computed on today's revised vintages. Every run restates the full history: " "the percentile yardstick grows by one quarter per refresh and upstream " "revisions re-rank past quarters. No out-of-sample claim is made for any " "historical reading, and the index makes no forecast." ) # --------------------------------------------------------------------------- # Loading and aggregation # --------------------------------------------------------------------------- @dataclass(frozen=True) class LoadedInput: spec: MemberSpec direction: str frequency: str last_updated: str source: str fred_series_id: Optional[str] name: str unit: str quarterly_raw: Dict[str, float] @dataclass(frozen=True) class Perturbation: input_id: str quarter: str mode: str # "unit" | "extreme" def quarter_index(q: str) -> int: year, qq = q.split("-Q") return int(year) * 4 + int(qq) - 1 def index_quarter(i: int) -> str: return f"{i // 4}-Q{i % 4 + 1}" def _annual_step_carry(by_quarter: Dict[str, float]) -> Dict[str, float]: """Step-carry an annual series across each release year's quarters plus CARRY_QUARTERS further quarters, superseded by the next annual observation. Input: a dict of {Y-Q1: value} (annual observations land in Q1 after date_to_quarter). Output: the same values carried across quarters. This is a documented carry of a real published value, not an interpolation — every filled quarter equals the most recent actual annual reading. """ if not by_quarter: return {} observed = sorted(by_quarter, key=quarter_index) out: Dict[str, float] = {} for idx, q in enumerate(observed): start = quarter_index(q) value = by_quarter[q] # carry until the next observation, capped at the release year's 4 # quarters + CARRY_QUARTERS if idx + 1 < len(observed): hard_stop = quarter_index(observed[idx + 1]) else: hard_stop = start + 4 + CARRY_QUARTERS cap = start + 4 + CARRY_QUARTERS stop = min(hard_stop, cap) for i in range(start, stop): out[index_quarter(i)] = value return out def _growth_5y(quarterly: Dict[str, float]) -> Dict[str, float]: """5-year (20-quarter) annualized % growth of a quarterly level series. g[q] = ((v[q] / v[q-20]) ** (1/5) - 1) * 100, defined only where the quarter exactly 20 quarters earlier is present and positive. The first 20 quarters of the level series have no growth value and drop out — honest, and the trailing edge is unaffected. """ out: Dict[str, float] = {} for q, v in quarterly.items(): prior = index_quarter(quarter_index(q) - GROWTH_WINDOW_QUARTERS) base = quarterly.get(prior) if base is not None and base > 0 and v > 0: out[q] = ((v / base) ** (1.0 / 5.0) - 1.0) * 100.0 return out def load_inputs(repo_root: Path) -> Dict[str, LoadedInput]: loaded: Dict[str, LoadedInput] = {} for spec in WORKER_REGISTRY: raw = json.loads((repo_root / spec.rel_path).read_text()) direction = raw.get("direction") if direction not in ORIENTABLE_DIRECTIONS: raise SystemExit( f"FAIL CLOSED: {spec.input_id} direction is {direction!r}; " f"expected one of {ORIENTABLE_DIRECTIONS}. A composite member must " f"carry a distress pole — a bidirectional or not_applicable series " f"cannot be oriented into the index. Refusing to build." ) frequency = str(raw.get("frequency", "")) by_quarter: Dict[str, List[float]] = {} for obs in raw["data"]: value = obs.get("value") if value is None: continue by_quarter.setdefault(date_to_quarter(obs["date"]), []).append(float(value)) if not by_quarter: raise SystemExit(f"FAIL CLOSED: {spec.input_id} has zero usable observations.") quarterly = {q: sum(vs) / len(vs) for q, vs in by_quarter.items()} if frequency == ANNUAL_FREQUENCY: quarterly = _annual_step_carry(quarterly) if spec.transform == "growth_5y": quarterly = _growth_5y(quarterly) if not quarterly: raise SystemExit(f"FAIL CLOSED: {spec.input_id} has no 5y-growth points.") loaded[spec.input_id] = LoadedInput( spec=spec, direction=direction, frequency=frequency, last_updated=str(raw.get("last_updated", "")), source=str(raw.get("source", "")), fred_series_id=raw.get("fred_series_id"), name=str(raw.get("name", spec.input_id)), unit=str(raw.get("unit", "")), quarterly_raw=quarterly, ) return loaded # --------------------------------------------------------------------------- # Build: orientation -> percentiles -> domain scores -> composite -> trim # --------------------------------------------------------------------------- @dataclass class BuildResult: published_quarters: List[str] composite: Dict[str, float] domain_scores: Dict[str, Dict[str, float]] members_present: Dict[str, Dict[str, int]] member_pct: Dict[str, Dict[str, float]] oriented: Dict[str, Dict[str, float]] def domain_weight_full() -> float: """Equal weight when all domains are present. The published composite uses the mean of the domains PRESENT (1/k for k present), which equals this when k == len(DOMAIN_IDS).""" return 1.0 / len(DOMAIN_IDS) def member_weight(domain: str) -> float: return 1.0 / len(DOMAIN_MEMBERS[domain]) def build(loaded: Dict[str, LoadedInput], perturb: Optional[Perturbation] = None) -> BuildResult: # Orientation (negate lower_is_worse so higher = worse for the worker) oriented: Dict[str, Dict[str, float]] = {} for input_id, li in loaded.items(): quarterly = dict(li.quarterly_raw) # Resolve the pole ONCE — see the matching note in compute_adi.py. Three # sites in this function each made their own two-way `else` decision; # deriving all three from one guarded value locks the class rather than # one instance (the 2026-05-14 A-0b sibling-escape shape). if li.direction == "lower_is_worse": sign = -1.0 elif li.direction == "higher_is_worse": sign = 1.0 else: raise SystemExit( f"FAIL CLOSED: {input_id} reached orientation with direction " f"{li.direction!r}; sign math requires one of {ORIENTABLE_DIRECTIONS}." ) if perturb is not None and perturb.input_id == input_id: if perturb.quarter not in quarterly: raise ValueError(f"perturbation quarter {perturb.quarter} absent from {input_id}") if perturb.mode == "unit": quarterly[perturb.quarter] = quarterly[perturb.quarter] + sign elif perturb.mode == "extreme": if sign > 0: quarterly[perturb.quarter] = max(quarterly.values()) + 1.0 else: quarterly[perturb.quarter] = min(quarterly.values()) - 1.0 else: # pragma: no cover raise ValueError(f"unknown perturbation mode {perturb.mode!r}") oriented[input_id] = {q: sign * v for q, v in quarterly.items()} # Hazen: one full-own-history pass per series member_pct: Dict[str, Dict[str, float]] = {} for input_id, series in oriented.items(): quarters = sorted(series, key=quarter_index) pcts = hazen_percentiles_series([series[q] for q in quarters]) member_pct[input_id] = dict(zip(quarters, pcts)) # Domain scores + composite (mean of PRESENT domains) + publication (>= MIN) all_quarters = sorted({q for series in member_pct.values() for q in series}, key=quarter_index) published: List[str] = [] composite: Dict[str, float] = {} domain_scores: Dict[str, Dict[str, float]] = {} members_present: Dict[str, Dict[str, int]] = {} for q in all_quarters: per_domain: Dict[str, float] = {} per_domain_count: Dict[str, int] = {} for d in DOMAIN_IDS: present = [m for m in DOMAIN_MEMBERS[d] if q in member_pct[m]] if present: per_domain[d] = sum(member_pct[m][q] for m in present) / len(present) per_domain_count[d] = len(present) if len(per_domain) >= MIN_DOMAINS_TO_PUBLISH: published.append(q) composite[q] = sum(per_domain.values()) / len(per_domain) # equal-weight mean of present domains domain_scores[q] = per_domain members_present[q] = per_domain_count published = _trailing_trim(published, member_pct) return BuildResult(published, composite, domain_scores, members_present, member_pct, oriented) def _has_full_member_set(q: str, member_pct: Dict[str, Dict[str, float]]) -> bool: return all(q in member_pct[s.input_id] for s in WORKER_REGISTRY) def _trailing_trim(published: List[str], member_pct: Dict[str, Dict[str, float]]) -> List[str]: """Trim the tail to the last quarter with the FULL current member set, so the latest published reading is never a ragged partial edge. Interior partial-member quarters (before a member's series begins) are kept.""" full = [q for q in published if _has_full_member_set(q, member_pct)] if not full: return published # coverage gate flags this cutoff = quarter_index(full[-1]) return [q for q in published if quarter_index(q) <= cutoff] # --------------------------------------------------------------------------- # Gates — any failure => no output written (same guard as compute_adi.main) # --------------------------------------------------------------------------- def quantile_linear(values: Sequence[float], p: float) -> float: s = sorted(values) n = len(s) if n == 1: return s[0] h = (n - 1) * p lo = math.floor(h) if lo + 1 >= n: return s[-1] return s[lo] + (s[lo + 1] - s[lo]) * (h - lo) def gate_weights() -> Dict[str, object]: w_d = domain_weight_full() ok = abs(w_d * len(DOMAIN_IDS) - 1.0) < 1e-12 member_ws = {} for d in DOMAIN_IDS: w_m = member_weight(d) member_ws[d] = w_m ok = ok and abs(w_m * len(DOMAIN_MEMBERS[d]) - 1.0) < 1e-12 return {"pass": ok, "domain_weight_full": w_d, "member_weights": member_ws, "rule": "equal domains: full-coverage weight = 1/len(DOMAIN_IDS); " "published composite = unweighted mean of present domains; zero hand-typed weights"} def gate_labels() -> Dict[str, object]: expected = ("Minimal", "Low", "Typical", "High", "Severe") ok = (BAND_LABELS == expected and len(BAND_LABELS) == N_BANDS and band_for(0.0) == (1, "0-20", "Minimal") and band_for(99.99) == (N_BANDS, "80-100", "Severe")) return {"pass": ok, "labels": list(BAND_LABELS), "rule": "national time-series labels are Minimal/Low/Typical/High/Severe; " "severity ascends with the band number"} def gate_orientation(loaded: Dict[str, LoadedInput], baseline: BuildResult) -> Dict[str, object]: rows = [] all_pass = True for spec in WORKER_REGISTRY: input_id = spec.input_id candidates = [q for q in baseline.published_quarters if q in loaded[input_id].quarterly_raw] if not candidates: continue # member's data ends before the trimmed publication tail q_star = candidates[-1] base_c = baseline.composite[q_star] row = {"input_id": input_id, "direction": loaded[input_id].direction, "quarter": q_star} for mode in ("unit", "extreme"): pert_c = build(loaded, Perturbation(input_id, q_star, mode)).composite[q_star] if mode == "unit": ok = pert_c >= base_c - 1e-9 else: series = baseline.oriented[input_id] v_star = series[q_star] top = max(series.values()) already_unique_top = v_star == top and sum(1 for v in series.values() if v == top) == 1 ok = (pert_c >= base_c - 1e-9) if already_unique_top else (pert_c > base_c + 1e-9) row[f"{mode}_pass"] = ok all_pass = all_pass and ok rows.append(row) return {"pass": all_pass, "per_input": rows, "rule": "perturbing any member's latest published quarter in the distress " "direction must not lower the composite"} def gate_dispositions(repo_root: Path) -> Dict[str, object]: members = {s.input_id for s in WORKER_REGISTRY if s.input_id != "unemployment_rate"} excused = set(SATELLITES) | set(AI_TRACKER_SERIES) universe = set() for path in sorted(repo_root.glob("data/indicators/awi/*.json")): try: raw = json.loads(path.read_text()) except (OSError, json.JSONDecodeError): continue # DECLARED, not ORIENTABLE — a poleless awi/ series is still a committed # candidate owing a disposition. Filtering on the poled pair drops the # universe 18 -> 13 with the gate still reporting pass. if raw.get("direction") in DECLARED_DIRECTIONS: universe.add(path.stem) undispositioned = sorted(universe - members - excused) stale_excusals = sorted(excused - universe) return {"pass": not undispositioned and not stale_excusals, "n_candidates": len(universe), "stale_excusals": stale_excusals, "undispositioned": undispositioned, "rule": "every committed data/indicators/awi/ series with a direction field is a " "worker-index member, a dispositioned satellite, or an AI-tracker series"} def gate_coverage(result: BuildResult) -> Dict[str, object]: qs = result.published_quarters tail_full = _has_full_member_set(qs[-1], result.member_pct) if qs else False # Publication must be contiguous from the first full-coverage quarter onward. # Deep-history quarters (before some members exist) publish under the >=3-of-4 # rule and are also contiguous by construction of all_quarters, so a gap # anywhere is a real defect. contiguous = all(quarter_index(b) - quarter_index(a) == 1 for a, b in zip(qs, qs[1:])) return {"pass": bool(qs) and tail_full and contiguous, "published": f"{qs[0]} to {qs[-1]} ({len(qs)} quarters)" if qs else "EMPTY", "published_range_contiguous": contiguous, "tail_has_full_member_set": tail_full, "rule": "published range contiguous AND the last published quarter carries the " "full current member set (trailing-edge trim invariant)"} # Worker-distress structural sanity gate. Replaces ADI's FFIEC-seam / GFC gates # (which test the national delinquency data) with a worker-appropriate check: # Structural sanity check: the GFC/jobless-recovery era (2008-2015) must score # far worse for the American worker than the postwar-deal era (pre-1980). This # is anchored to two FIXED historical windows, so it verifies a permanent fact # — the orientation is correct (a known-bad era ranks high) and the composite # is not sign-flipped — without ever constraining the LIVE reading. # # The index itself is symmetric and unbiased: each input is a percentile of its # own history, so a series that improves lowers the composite and one that # worsens raises it (the empirical series falls in booms and rises in # downturns; trough 1969, and the Job domain is low today). An earlier version # of this gate required the GLOBAL worst 8-quarter window to stay inside # 2008-2015. That was wrong: a percentile-of-history composite has no ceiling, # so if a future downturn ever produced a worse stretch, that gate would have # refused to publish the real reading — suppressing a true extreme, not # preventing a false one. The fix anchors to fixed history instead. The # worst-window position is still reported for eyeballing, never gated on. # (Cold review 2026-07-07 P1.) DISTRESS_ERA = ("2008-Q1", "2015-Q4") POSTWAR_DEAL_END = "1980-Q1" # exclusive: pre-1980 is the postwar-deal baseline DISTRESS_MARGIN = 20.0 # the distress era must outscore the postwar deal by this much WORST_WINDOW_LEN = 8 def gate_worker_distress(result: BuildResult) -> Dict[str, object]: qs = result.published_quarters composite = result.composite lo, hi = quarter_index(DISTRESS_ERA[0]), quarter_index(DISTRESS_ERA[1]) era = [composite[q] for q in qs if lo <= quarter_index(q) <= hi] postwar = [composite[q] for q in qs if q < POSTWAR_DEAL_END] if not era or not postwar: return {"pass": False, "rule": "need both the 2008-2015 distress era and the pre-1980 baseline"} era_mean = sum(era) / len(era) postwar_mean = sum(postwar) / len(postwar) margin = era_mean - postwar_mean # informational: where the worst sustained 8-quarter window actually sits best_start, best_mean = None, -1.0 for i in range(len(qs) - WORST_WINDOW_LEN + 1): window = qs[i:i + WORST_WINDOW_LEN] if quarter_index(window[-1]) - quarter_index(window[0]) != WORST_WINDOW_LEN - 1: continue m = sum(composite[q] for q in window) / WORST_WINDOW_LEN if m > best_mean: best_mean, best_start = m, window[0] ok = margin > DISTRESS_MARGIN return {"pass": ok, "distress_era_mean": round(era_mean, 2), "postwar_deal_mean": round(postwar_mean, 2), "margin": round(margin, 2), "worst_8q_window_start": best_start, "worst_8q_window_mean": round(best_mean, 2) if best_start else None, "rule": f"the 2008-2015 GFC/jobless-recovery era scores at least {DISTRESS_MARGIN} points " "worse for the worker than the pre-1980 postwar-deal era (a permanent structural " "fact — never fail-closes on legitimate future worsening)"} def band_for(composite_value: float) -> Tuple[int, str, str]: width = 100.0 / N_BANDS band = min(int(composite_value // width) + 1, N_BANDS) low = round((band - 1) * width) high = round(band * width) return band, f"{low}-{high}", BAND_LABELS[band - 1] # --------------------------------------------------------------------------- # Output assembly — same key shape as adi.json # --------------------------------------------------------------------------- def assemble_output( loaded: Dict[str, LoadedInput], result: BuildResult, gates: Dict[str, Dict[str, object]], ) -> Dict[str, object]: last_updated = max(li.last_updated for li in loaded.values()) band_thresholds = [round(i * 100.0 / N_BANDS) for i in range(1, N_BANDS)] registry_block = [] for spec in WORKER_REGISTRY: li = loaded[spec.input_id] qs = sorted(li.quarterly_raw, key=quarter_index) entry = { "input_id": spec.input_id, "name": li.name, "path": spec.rel_path, "domain": spec.domain, "direction": li.direction, "frequency": li.frequency, "transform": spec.transform, "fred_series_id": li.fred_series_id, "source": li.source, "unit": li.unit, "n_quarters": len(qs), "first_quarter": qs[0], "last_quarter": qs[-1], "last_updated": li.last_updated, "member_weight_within_domain": round(member_weight(spec.domain), 6), } registry_block.append(entry) rows = [] for q in result.published_quarters: band, band_range, band_label = band_for(result.composite[q]) rows.append({ "quarter": q, "composite": round(result.composite[q], 2), "band": band, "band_range": band_range, "band_label": band_label, "domains": {d: {"score": round(result.domain_scores[q][d], 2), "members_present": result.members_present[q][d]} for d in DOMAIN_IDS if d in result.domain_scores[q]}, "members": {s.input_id: round(result.member_pct[s.input_id][q], 2) for s in WORKER_REGISTRY if q in result.member_pct[s.input_id]}, }) peak_q = max(result.published_quarters, key=lambda q: result.composite[q]) trough_q = min(result.published_quarters, key=lambda q: result.composite[q]) latest_q = result.published_quarters[-1] first_year = result.published_quarters[0].split("-")[0] latest_rank_pct = hazen_percentiles_series( [result.composite[q] for q in result.published_quarters])[-1] latest_band = band_for(result.composite[latest_q]) return { "series_id": SERIES_ID, "name": SERIES_NAME, "taxonomy_version": TAXONOMY_VERSION, "level": "national", "frequency": "quarterly", "source": ("American Default Research, computed from BLS, BEA, and Federal Reserve " "series via FRED plus a spliced union-membership series (per-input " "attribution in methodology.registry)"), "last_updated": last_updated, "summary": { "n_published_quarters": len(result.published_quarters), "first_quarter": result.published_quarters[0], "last_quarter": latest_q, "peak": {"quarter": peak_q, "composite": round(result.composite[peak_q], 2)}, "trough": {"quarter": trough_q, "composite": round(result.composite[trough_q], 2)}, "latest": { "quarter": latest_q, "composite": round(result.composite[latest_q], 2), "band": latest_band[0], "band_label": latest_band[2], "reading": (f"On average, the American worker's inputs sit worse than in " f"{round(result.composite[latest_q])}% of their own quarterly histories"), "rank_in_history": { "hazen_percentile": round(latest_rank_pct, 1), "reading": (f"The composite itself sits higher than " f"{round(latest_rank_pct)}% of all published quarters since {first_year}"), "rule": "Hazen percentile of the latest composite within the published composite series", }, "domains": {d: round(result.domain_scores[latest_q][d], 2) for d in DOMAIN_IDS if d in result.domain_scores[latest_q]}, }, }, "methodology": { "reading": ("AWI(t) is the mean of the worker domains present that quarter, each the " "mean of its members' Hazen percentiles within their own full quarterly " "history. The composite is a mean of percentiles, not itself a percentile " "of quarters. AWI measures current conditions and makes no forecast."), "domains": {d: list(DOMAIN_MEMBERS[d]) for d in DOMAIN_IDS}, "domain_labels": DOMAIN_LABELS, "domain_weight_full": round(domain_weight_full(), 6), "weight_rule": gates["weights"]["rule"], "normalization": ("Hazen percentile per series over its entire available quarterly " "history in one pass: (average_rank - 0.5) / n * 100, ties averaged. " "One yardstick, shared with the ADI/SDI/CDI family."), "member_transforms": { # Derived, not hand-typed: a membership change must not be able # to leave a stale count in published methodology (it did — # "seven members" survived F1's 10 -> 9 change until review). "level": (f"{_english_count(sum(1 for s in WORKER_REGISTRY if s.transform == 'level'))} " f"members enter as raw quarterly levels (rates, shares, " f"union and labor-share power measures)"), "growth_5y": ("the three Paycheck members (real production wage, real median weekly " "earnings, productivity-pay wedge) enter as 5-year annualized % growth " "before ranking — de-trending the wage series so a record real level " "does not read as favorable, and answering the trending-percentile critique"), }, "orientation_rule": ("Every input's direction field is read from its JSON; " "lower_is_worse values are negated before ranking so higher always " "means worse for the worker. A missing or unexpected direction " "aborts the build."), "publication_rule": (f"publish a quarter only when at least {MIN_DOMAINS_TO_PUBLISH} of " f"the {len(DOMAIN_IDS)} domains have a present member; the composite " f"is the equal-weight mean of the domains present"), "trailing_edge_rule": ("the published series is trimmed at the tail to the last quarter " "carrying the full current member set, so the latest reading is " "never a ragged partial edge; interior partial-member quarters are kept"), "annual_carry_rule": (f"annual members (family income Gini, union membership) are " f"step-carried across their release year and up to {CARRY_QUARTERS} " f"further quarters, superseded by the next annual release — a " f"documented carry of a real published value, never an interpolation"), "missing_data_rules": [ "no imputation of an absent member within a quarter; a domain scores on the members present", "no cross-series interpolation; annual members carry a real published value forward under the annual_carry_rule", "a quarter with fewer than the publication minimum of domains is not published", "the published tail is trimmed to the last full-member quarter", "orientation and direction are read from each member JSON, never hardcoded here", ], "bands": {"n_bands": N_BANDS, "thresholds": band_thresholds, "labels": list(BAND_LABELS), "derivation": "uniform segments, threshold_i = i * 100 / N_BANDS; labels are an " "editorial lock, severity ascending with the band number", "label_usage": "the labels describe the national time axis and always publish " "with the literal mean-of-input-histories reading; higher bands " "mean a worse position for the worker"}, "family_relation": ("The American Worker Index is the worker-conditions member of the " "index family. ADI ranks the nation's present quarter against its own " "history; SDI and CDI rank places; AWI ranks the American worker's " "present quarter against 1948-present. Same Hazen transform, same " "orientation rule, same equal-weight-domain shape."), "vintage_honesty": VINTAGE_HONESTY, "revision_model": ("Every refresh restates the full history on today's revised source " "vintages; the percentile yardstick grows by one quarter per run."), "registry": registry_block, }, "validation": gates, "data": rows, } def serialize(output: Dict[str, object]) -> str: return json.dumps(output, indent=2, ensure_ascii=False) + "\n" def write_csv(output: Dict[str, object], path: Path) -> None: header = ["quarter", "composite", "band", "band_range", "band_label"] for d in DOMAIN_IDS: header += [f"{d}_score", f"{d}_members"] lines = [",".join(header)] for row in output["data"]: cells = [row["quarter"], f"{row['composite']:.2f}", str(row["band"]), row["band_range"], row["band_label"]] # allow-literal: json-mirror zero-padding for d in DOMAIN_IDS: dom = row["domains"].get(d) if dom is not None: cells += [f"{dom['score']:.2f}", str(dom["members_present"])] # allow-literal: json-mirror zero-padding else: cells += ["", ""] # domain absent this quarter (>=3-of-4 publication) lines.append(",".join(cells)) path.write_text("\n".join(lines) + "\n") # --------------------------------------------------------------------------- # Main — reproducibility + write-only-on-full-pass guard (mirror compute_adi) # --------------------------------------------------------------------------- def run_once() -> Tuple[Dict[str, object], str, bool]: loaded = load_inputs(REPO_ROOT) result = build(loaded) gates: Dict[str, Dict[str, object]] = {} gates["weights"] = gate_weights() gates["labels"] = gate_labels() gates["dispositions"] = gate_dispositions(REPO_ROOT) gates["orientation"] = gate_orientation(loaded, result) gates["coverage"] = gate_coverage(result) gates["worker_distress"] = gate_worker_distress(result) output = assemble_output(loaded, result, gates) gates["vintage_honesty_present"] = { "pass": VINTAGE_HONESTY in json.dumps(output["methodology"]), "rule": "the methodology block must contain the vintage-honesty statement", } serialized = serialize(output) all_pass = all(bool(g["pass"]) for g in gates.values()) return output, serialized, all_pass def main() -> int: logging.basicConfig(level=logging.INFO, format="%(message)s") out1, ser1, pass1 = run_once() out2, ser2, pass2 = run_once() reproducible = ser1 == ser2 out1["validation"]["reproducibility"] = { "pass": reproducible, "rule": "two full in-process passes serialize byte-identically; validate re-verifies cross-process", } ser1 = serialize(out1) all_pass = pass1 and pass2 and reproducible for gate_name, g in out1["validation"].items(): logger.info("gate %-24s %s", gate_name, "PASS" if g["pass"] else "FAIL") if not all_pass: logger.error("GATE FAILURE — no output written.") return 1 OUTPUT_DIR.mkdir(parents=True, exist_ok=True) json_path = OUTPUT_DIR / f"{SERIES_ID}_index.json" csv_path = OUTPUT_DIR / f"{SERIES_ID}_index.csv" json_path.write_text(ser1) write_csv(out1, csv_path) public_dir = REPO_ROOT / "site" / "public" / "data" / "indexes" public_dir.mkdir(parents=True, exist_ok=True) (public_dir / f"{SERIES_ID}_index.json").write_text(ser1) (public_dir / f"{SERIES_ID}_index.csv").write_text(csv_path.read_text()) latest = out1["summary"]["latest"] logger.info("wrote %s — %s composite %.1f (%s)", json_path, latest["quarter"], latest["composite"], latest["band_label"]) return 0 if __name__ == "__main__": raise SystemExit(main())