DOES-Applications/bls-data
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add BLS client library, example scripts, and usage docs

Browse Source 
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
David Boyd
2026-06-19 08:39:47 -04:00
parent eb724aa7ff
commit 20bd3e9a2f
11 changed files with 1443 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
bls_client/__init__.py Normal file
View File

@ -0,0 +1,2 @@
from .client import BLSClient
from . import queries, series

171
bls_client/client.py Normal file
View File

@ -0,0 +1,171 @@
"""
BLSClient — thin wrapper around the BLS Public Data API v2.
Usage:
from bls_client import BLSClient
client = BLSClient("YOUR_API_KEY")
data = client.fetch(["CES0000000001"], 2020, 2025)
"""
import requests
from datetime import datetime
class BLSClient:
BASE_URL = "https://api.bls.gov/publicAPI/v2"
MAX_SERIES_PER_CALL = 50
MAX_YEARS_PER_CALL = 20
def __init__(self, api_key: str):
self.api_key = api_key
# ------------------------------------------------------------------
# Core fetch
# ------------------------------------------------------------------
def fetch(
self,
series_ids: list[str] | str,
start_year: int,
end_year: int,
catalog: bool = True,
calculations: bool = False,
annual_average: bool = False,
) -> dict[str, list[dict]]:
"""
Fetch time-series data for one or more series IDs.
Automatically batches requests when series_ids > 50.
Returns a dict keyed by series ID, value is the list of
observations (newest first).
Args:
series_ids: Single series ID or list of series IDs.
start_year: First year to retrieve.
end_year: Last year to retrieve (max 20 years from start).
catalog: Include series title/metadata in response.
calculations: Include MoM / YoY net and pct changes.
annual_average: Include M13 annual average rows.
Returns:
{series_id: [{"year":..., "period":..., "value":..., ...}, ...]}
"""
if isinstance(series_ids, str):
series_ids = [series_ids]
results = {}
for i in range(0, len(series_ids), self.MAX_SERIES_PER_CALL):
batch = series_ids[i : i + self.MAX_SERIES_PER_CALL]
payload = {
"seriesid": batch,
"startyear": str(start_year),
"endyear": str(end_year),
"registrationkey": self.api_key,
"catalog": catalog,
"calculations": calculations,
"annualaverage": annual_average,
}
r = requests.post(
f"{self.BASE_URL}/timeseries/data/",
json=payload,
timeout=30,
)
r.raise_for_status()
body = r.json()
if body["status"] != "REQUEST_SUCCEEDED":
raise RuntimeError(f"BLS API error: {body['message']}")
for s in body["Results"]["series"]:
results[s["seriesID"]] = {
"data": s["data"],
"catalog": s.get("catalog", {}),
}
return results
def fetch_latest(
self,
series_ids: list[str] | str,
years: int = 2,
**kwargs,
) -> dict[str, list[dict]]:
"""Convenience: fetch the most recent `years` years."""
end = datetime.now().year
start = end - (years - 1)
return self.fetch(series_ids, start, end, **kwargs)
def fetch_named(
self,
named: dict[str, str],
start_year: int,
end_year: int,
**kwargs,
) -> dict[str, list[dict]]:
"""
Fetch a labeled dict of series IDs.
Args:
named: {"Human label": "SERIES_ID", ...}
Returns:
{"Human label": {"data": [...], "catalog": {...}}}
"""
id_to_label = {v: k for k, v in named.items()}
raw = self.fetch(list(named.values()), start_year, end_year, **kwargs)
return {id_to_label[sid]: v for sid, v in raw.items()}
# ------------------------------------------------------------------
# Discovery endpoints
# ------------------------------------------------------------------
def surveys(self) -> list[dict]:
"""Return all BLS surveys with abbreviation and name."""
r = requests.get(f"{self.BASE_URL}/surveys", timeout=15)
r.raise_for_status()
return r.json()["Results"]["survey"]
def popular(self, survey: str) -> list[str]:
"""Return popular series IDs for a given survey abbreviation."""
r = requests.get(
f"{self.BASE_URL}/timeseries/popular",
params={"survey": survey},
timeout=15,
)
r.raise_for_status()
body = r.json()
return [
s["seriesID"]
for s in body.get("Results", {}).get("series", [])
if s
]
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
@staticmethod
def latest_obs(series_result: dict) -> dict | None:
"""Return the most recent non-null observation from a fetch result."""
for obs in series_result.get("data", []):
if obs["value"] != "-":
return obs
return None
@staticmethod
def to_rows(results: dict[str, dict]) -> list[dict]:
"""
Flatten fetch results into a list of dicts suitable for CSV or
pandas DataFrame ingestion.
Each row: {series_id, series_title, year, period, period_name, value}
"""
rows = []
for sid, s in results.items():
title = s.get("catalog", {}).get("series_title", "")
for obs in s.get("data", []):
if obs["value"] == "-":
continue
rows.append({
"series_id": sid,
"series_title": title,
"year": obs["year"],
"period": obs["period"],
"period_name": obs.get("periodName", ""),
"value": obs["value"],
})
return rows

1
bls_client/queries/__init__.py Normal file
View File

@ -0,0 +1 @@
from . import employment, prices, wages, productivity

133
bls_client/queries/employment.py Normal file
View File

@ -0,0 +1,133 @@
"""
Pre-built employment series IDs — ready to pass to BLSClient.fetch().
All functions return a single series ID string or a dict of {label: series_id}.
"""
from ..series import laus_state, laus_msa, ces_national, ces_state, jolts
# ---------------------------------------------------------------------------
# National employment
# ---------------------------------------------------------------------------
def nonfarm_payrolls(seasonal: bool = True) -> str:
"""Total nonfarm payroll employment (CES). The headline monthly jobs number."""
return ces_national("total_nonfarm", "employees", seasonal)
def private_payrolls(seasonal: bool = True) -> str:
"""Total private sector payroll employment."""
return ces_national("total_private", "employees", seasonal)
def government_employment(seasonal: bool = True) -> str:
"""Total government employment (federal + state + local)."""
return ces_national("government", "employees", seasonal)
def manufacturing_employment(seasonal: bool = True) -> str:
return ces_national("manufacturing", "employees", seasonal)
def national_unemployment_rate(seasonal: bool = True) -> str:
"""National unemployment rate from the CPS (U-3 rate)."""
return "LNS14000000" if seasonal else "LNU04000000"
def national_labor_force_participation(seasonal: bool = True) -> str:
"""National labor force participation rate."""
return "LNS11300000" if seasonal else "LNU01300000"
# ---------------------------------------------------------------------------
# LAUS — state unemployment
# ---------------------------------------------------------------------------
def dc_unemployment_rate(seasonal: bool = False) -> str:
return laus_state(11, "rate", seasonal)
def md_unemployment_rate(seasonal: bool = False) -> str:
return laus_state(24, "rate", seasonal)
def va_unemployment_rate(seasonal: bool = False) -> str:
return laus_state(51, "rate", seasonal)
def state_unemployment(state_fips: int, seasonal: bool = False) -> dict:
"""All four LAUS measures for a state."""
return {
"rate": laus_state(state_fips, "rate", seasonal),
"unemployed": laus_state(state_fips, "unemployed", seasonal),
"employed": laus_state(state_fips, "employed", seasonal),
"laborforce": laus_state(state_fips, "laborforce", seasonal),
}
def dc_region_unemployment() -> dict:
"""Unemployment rates for DC, MD, VA states + DC Metro, Baltimore, Richmond MSAs."""
return {
"DC State": laus_state(11, "rate"),
"Maryland": laus_state(24, "rate"),
"Virginia": laus_state(51, "rate"),
"DC Metro MSA": laus_msa(11, 47900, "rate"),
"Baltimore MSA":laus_msa(24, 12580, "rate"),
"Richmond MSA": laus_msa(51, 40060, "rate"),
}
# ---------------------------------------------------------------------------
# JOLTS
# ---------------------------------------------------------------------------
def job_openings_level(seasonal: bool = True) -> str:
return jolts("job_openings", "L", seasonal=seasonal)
def job_openings_rate(seasonal: bool = True) -> str:
return jolts("job_openings", "R", seasonal=seasonal)
def quits_level(seasonal: bool = True) -> str:
return jolts("quits", "L", seasonal=seasonal)
def hires_level(seasonal: bool = True) -> str:
return jolts("hires", "L", seasonal=seasonal)
def layoffs_level(seasonal: bool = True) -> str:
return jolts("layoffs", "L", seasonal=seasonal)
def jolts_dashboard() -> dict:
"""All five JOLTS measures (levels, SA) as a labeled dict."""
return {
"Job Openings": job_openings_level(),
"Hires": hires_level(),
"Quits": quits_level(),
"Layoffs": layoffs_level(),
"Total Separations": jolts("total_separations", "L"),
}
# ---------------------------------------------------------------------------
# QCEW
# ---------------------------------------------------------------------------
def qcew_national_private() -> str:
"""QCEW — national, private sector, all industries, quarterly."""
return "ENU0000010510000"
def qcew_dc_private() -> str:
"""QCEW — DC, private sector, all industries."""
return "ENU1100010510000"
def qcew_state(state_fips: int, ownership: str = "5") -> str:
"""
QCEW state-level series.
Args:
state_fips: 2-digit FIPS
ownership: "0"=all, "5"=private, "1"=federal, "2"=state, "3"=local
"""
return f"ENU{state_fips:02d}0001{ownership}10000"

124
bls_client/queries/prices.py Normal file
View File

@ -0,0 +1,124 @@
"""
Pre-built price series IDs — CPI, PPI, Import/Export, Average Prices.
"""
from ..series import cpi, ppi_commodity
# ---------------------------------------------------------------------------
# CPI-U
# ---------------------------------------------------------------------------
def cpi_all_items(seasonal: bool = False) -> str:
"""CPI-U all items, US city average."""
return cpi("all_items", seasonal=seasonal)
def cpi_core(seasonal: bool = True) -> str:
"""CPI-U all items less food and energy (core inflation)."""
return cpi("core", seasonal=seasonal)
def cpi_food(seasonal: bool = False) -> str:
return cpi("food", seasonal=seasonal)
def cpi_energy(seasonal: bool = False) -> str:
return cpi("energy", seasonal=seasonal)
def cpi_shelter(seasonal: bool = False) -> str:
return cpi("shelter", seasonal=seasonal)
def cpi_gasoline(seasonal: bool = True) -> str:
return cpi("gasoline", seasonal=seasonal)
def cpi_medical(seasonal: bool = False) -> str:
return cpi("medical", seasonal=seasonal)
def cpi_dashboard(seasonal: bool = False) -> dict:
"""Key CPI components as a labeled dict."""
return {
"All Items": cpi_all_items(seasonal),
"Core (ex food/NRG)": cpi_core(True),
"Food": cpi_food(seasonal),
"Energy": cpi_energy(seasonal),
"Shelter": cpi_shelter(seasonal),
"Medical Care": cpi_medical(seasonal),
"Gasoline": cpi_gasoline(True),
}
# ---------------------------------------------------------------------------
# CPI-W
# ---------------------------------------------------------------------------
def cpi_w_all_items(seasonal: bool = False) -> str:
"""CPI-W all items (used for Social Security COLA)."""
return cpi("all_items", seasonal=seasonal, series="W")
# ---------------------------------------------------------------------------
# PPI
# ---------------------------------------------------------------------------
def ppi_all_commodities() -> str:
"""PPI — all commodities index."""
return ppi_commodity("00000000")
def ppi_finished_goods() -> str:
"""PPI — finished goods."""
return ppi_commodity("3")
def ppi_energy() -> str:
return ppi_commodity("05")
def ppi_food() -> str:
return ppi_commodity("02")
def ppi_dashboard() -> dict:
return {
"All Commodities": ppi_all_commodities(),
"Finished Goods": ppi_finished_goods(),
"Food": ppi_food(),
"Energy": ppi_energy(),
}
# ---------------------------------------------------------------------------
# Average Prices (AP)
# ---------------------------------------------------------------------------
def avg_price_electricity() -> str:
"""Average retail price of electricity per KWh."""
return "APU000072610"
def avg_price_gasoline_regular() -> str:
"""Average retail price of regular unleaded gasoline per gallon."""
return "APU00007471A"
def avg_price_eggs() -> str:
"""Average retail price of eggs per dozen."""
return "APU0000708111"
def avg_price_ground_beef() -> str:
"""Average retail price of ground beef per pound."""
return "APU0000703112"
# ---------------------------------------------------------------------------
# Import/Export Price Indexes (EI)
# ---------------------------------------------------------------------------
def import_price_index() -> str:
"""Import price index — all imports."""
return "EIUIR"
def export_price_index() -> str:
"""Export price index — all exports."""
return "EIUIQ"

36
bls_client/queries/productivity.py Normal file
View File

@ -0,0 +1,36 @@
"""
Pre-built productivity series IDs — Major Sector Productivity & Costs.
"""
from ..series import productivity
def business_output_per_hour(seasonal: bool = True) -> str:
"""Business sector output per hour worked (quarterly)."""
return productivity("business", "output_per_hour", seasonal)
def business_unit_labor_cost(seasonal: bool = True) -> str:
"""Business sector unit labor costs (quarterly)."""
return productivity("business", "unit_labor_cost", seasonal)
def business_real_comp_per_hour(seasonal: bool = True) -> str:
"""Business sector real compensation per hour."""
return productivity("business", "real_comp_per_hr", seasonal)
def nonfarm_output_per_hour(seasonal: bool = True) -> str:
return productivity("nonfarm_business", "output_per_hour", seasonal)
def manufacturing_output_per_hour(seasonal: bool = True) -> str:
return productivity("manufacturing", "output_per_hour", seasonal)
def productivity_dashboard() -> dict:
return {
"Business Output/Hr": business_output_per_hour(),
"Business Unit Labor Cost": business_unit_labor_cost(),
"Nonfarm Output/Hr": nonfarm_output_per_hour(),
"Manufacturing Output/Hr": manufacturing_output_per_hour(),
}

120
bls_client/queries/wages.py Normal file
View File

@ -0,0 +1,120 @@
"""
Pre-built wage and compensation series IDs — OES, ECI, ECEC, CPS earnings.
"""
from ..series import oes_national, eci
# ---------------------------------------------------------------------------
# OES — Occupational Employment & Wage Statistics
# ---------------------------------------------------------------------------
def all_occupations_employment() -> str:
"""National employment across all occupations, all industries."""
return oes_national("000000", "000000", "employment")
def occupation_annual_median_wage(soc_code: str) -> str:
"""
Annual median wage for a specific occupation (national, all industries).
Args:
soc_code: 6-digit SOC code (e.g. "151132" for software developers,
"291141" for registered nurses, "119021" for construction mgrs)
"""
return oes_national(soc_code, "000000", "annual_median")
def occupation_employment(soc_code: str) -> str:
"""Employment level for a specific occupation (national)."""
return oes_national(soc_code, "000000", "employment")
# Common occupation codes
SOC_CODES = {
"software_developers": "151132",
"registered_nurses": "291141",
"teachers_elementary": "252021",
"accountants": "132011",
"construction_managers": "119021",
"truck_drivers": "533032",
"janitors": "372011",
"retail_salespersons": "412031",
"first_line_supervisors_mfg":"511011",
"lawyers": "231011",
"physicians": "291229",
"police_officers": "333051",
"social_workers": "211029",
"financial_analysts": "132051",
"data_scientists": "152051",
}
# ---------------------------------------------------------------------------
# ECI — Employment Cost Index
# ---------------------------------------------------------------------------
def eci_total_compensation(seasonal: bool = True) -> str:
"""ECI — civilian workers, all industries, total compensation."""
return eci("10", "10", component="A", seasonal=seasonal)
def eci_wages(seasonal: bool = True) -> str:
"""ECI — civilian workers, wages and salaries only."""
return eci("10", "10", component="W", seasonal=seasonal)
def eci_benefits(seasonal: bool = True) -> str:
"""ECI — civilian workers, benefit costs only."""
return eci("10", "10", component="B", seasonal=seasonal)
def eci_private(seasonal: bool = True) -> str:
"""ECI — private sector, total compensation."""
return eci("20", "10", component="A", seasonal=seasonal)
def eci_state_local(seasonal: bool = True) -> str:
"""ECI — state and local government, total compensation."""
return eci("30", "10", component="A", seasonal=seasonal)
def eci_dashboard() -> dict:
return {
"Total Compensation (Civilian)": eci_total_compensation(),
"Wages & Salaries (Civilian)": eci_wages(),
"Benefits (Civilian)": eci_benefits(),
"Total Comp (Private)": eci_private(),
"Total Comp (State/Local Gov)": eci_state_local(),
}
# ---------------------------------------------------------------------------
# ECEC — Employer Costs for Employee Compensation
# ---------------------------------------------------------------------------
def ecec_total_compensation() -> str:
"""ECEC — civilian workers, total compensation cost per hour."""
return "CMU1010000000000D"
def ecec_health_insurance() -> str:
"""ECEC — health insurance cost per hour worked."""
return "CMU1010000000000H"
def ecec_retirement() -> str:
"""ECEC — retirement & savings cost per hour worked."""
return "CMU1010000000000R"
# ---------------------------------------------------------------------------
# CPS Earnings (LE)
# ---------------------------------------------------------------------------
def median_weekly_earnings(seasonal: bool = True) -> str:
"""Median usual weekly earnings — full-time wage and salary workers."""
return "LEU0252881600" if seasonal else "LEU0252881500"
def median_weekly_earnings_men() -> str:
return "LEU0252882800"
def median_weekly_earnings_women() -> str:
return "LEU0252882900"

368
bls_client/series.py Normal file
View File

@ -0,0 +1,368 @@
"""
Series ID builders for major BLS surveys.
These functions construct the coded series IDs used by the BLS API.
Pass the returned string directly to BLSClient.fetch().
Example:
from bls_client.series import laus_state, ces_national
client.fetch([laus_state(11, "rate"), ces_national("00000000", "01")], 2020, 2025)
"""
# ---------------------------------------------------------------------------
# LAUS — Local Area Unemployment Statistics
# Series format: LA[adj][area_type+area_code 15 chars][measure 2 chars]
# ---------------------------------------------------------------------------
_LAUS_MEASURE = {
"rate": "03",
"unemployed": "04",
"employed": "05",
"laborforce": "06",
"emp_pop": "07",
"lfpr": "08",
"pop": "09",
}
def laus_state(state_fips: int, measure: str = "rate", seasonal: bool = False) -> str:
"""
LAUS series for a state.
Args:
state_fips: 2-digit state FIPS (e.g. 11=DC, 24=MD, 51=VA)
measure: "rate" | "unemployed" | "employed" | "laborforce"
seasonal: True for seasonally adjusted
Examples:
laus_state(11, "rate") → "LAUST110000000000003"
laus_state(24, "employed") → "LAUST240000000000005"
"""
adj = "S" if seasonal else "U"
m = _LAUS_MEASURE.get(measure, measure)
# Area code = ST(2) + state_fips(2) + 11 zeros = 15 chars → total series = 20
return f"LA{adj}ST{state_fips:02d}{'0'*11}{m}"
def laus_county(state_fips: int, county_fips: int, measure: str = "rate", seasonal: bool = False) -> str:
"""
LAUS series for a county.
Args:
state_fips: 2-digit state FIPS
county_fips: 3-digit county FIPS
measure: "rate" | "unemployed" | "employed" | "laborforce"
seasonal: True for seasonally adjusted
Examples:
laus_county(11, 1, "rate") → "LAUCN110010000000003"
"""
adj = "S" if seasonal else "U"
m = _LAUS_MEASURE.get(measure, measure)
# Area code = CN(2) + state_fips(2) + county_fips(3) + 8 zeros = 15 chars → total = 20
return f"LA{adj}CN{state_fips:02d}{county_fips:03d}{'0'*8}{m}"
def laus_msa(state_fips: int, cbsa_code: int, measure: str = "rate") -> str:
"""
LAUS series for a Metropolitan Statistical Area.
The state_fips should be the primary state in the MSA.
Args:
state_fips: 2-digit FIPS of the primary state
cbsa_code: 5-digit CBSA code
measure: "rate" | "unemployed" | "employed" | "laborforce"
Examples:
laus_msa(11, 47900, "rate") → DC-Arlington MSA unemployment rate
laus_msa(24, 12580, "rate") → Baltimore MSA unemployment rate
laus_msa(51, 40060, "rate") → Richmond MSA unemployment rate
"""
m = _LAUS_MEASURE.get(measure, measure)
return f"LAUMT{state_fips:02d}{cbsa_code:05d}000000{m}"
# ---------------------------------------------------------------------------
# CES — Current Employment Statistics (National)
# Series format: CE[adj][industry 8 chars][data_type 2 chars]
# ---------------------------------------------------------------------------
_CES_SUPERSECTOR = {
"total_nonfarm": "00000000",
"total_private": "05000000",
"mining_logging": "10000000",
"construction": "20000000",
"manufacturing": "30000000",
"durable": "31000000",
"nondurable": "32000000",
"trade_trans_util":"40000000",
"wholesale": "41000000",
"retail": "42000000",
"information": "50000000",
"financial": "55000000",
"professional": "60000000",
"education_health":"65000000",
"leisure": "70000000",
"other_services": "80000000",
"government": "90000000",
}
_CES_DATATYPE = {
"employees": "01",
"avg_weekly_hours": "02",
"avg_hourly_earn": "03",
"prod_employees": "06",
"women_employees": "10",
"avg_weekly_earn": "11",
}
def ces_national(
industry: str = "total_nonfarm",
data_type: str = "employees",
seasonal: bool = True,
) -> str:
"""
CES national employment series.
Args:
industry: Supersector name (see _CES_SUPERSECTOR) or 8-digit code string
data_type: Measure name (see _CES_DATATYPE) or 2-digit code string
seasonal: True for seasonally adjusted
Examples:
ces_national() → "CES0000000001"
ces_national("manufacturing", "employees") → "CES3000000001"
ces_national("retail", "avg_hourly_earn") → "CES4200000003"
"""
adj = "S" if seasonal else "U"
ind = _CES_SUPERSECTOR.get(industry, industry)
dtype = _CES_DATATYPE.get(data_type, data_type)
return f"CE{adj}{ind}{dtype}"
def ces_state(
state_fips: int,
industry: str = "total_nonfarm",
data_type: str = "employees",
seasonal: bool = True,
area_code: str = "00000",
) -> str:
"""
State or metro employment series (SM prefix).
Args:
state_fips: 2-digit state FIPS
industry: Supersector name or 8-digit code
data_type: Measure name or 2-digit code
seasonal: True for SA
area_code: 5-digit metro area code; "00000" = statewide
Examples:
ces_state(11) → "SMS110000000000001" (DC total nonfarm)
ces_state(24, "retail") → "SMS240000042000001" (MD retail)
"""
adj = "S" if seasonal else "U"
ind = _CES_SUPERSECTOR.get(industry, industry)
dtype = _CES_DATATYPE.get(data_type, data_type)
return f"SM{adj}{state_fips:02d}{area_code}{ind}{dtype}"
# ---------------------------------------------------------------------------
# CPI — Consumer Price Index
# Series format: CU[adj][periodicity][area 4 chars][item 6 chars]
# ---------------------------------------------------------------------------
_CPI_ITEM = {
"all_items": "SA0",
"core": "SA0L1E", # all items less food & energy
"food": "SAF",
"food_at_home": "SAF1",
"energy": "SA0E",
"shelter": "SAH1",
"apparel": "SAA",
"transportation":"SAT",
"medical": "SAM",
"education": "SAE",
"recreation": "SAR",
"gasoline": "SS47014",
"electricity": "SEHF01",
"new_vehicles": "SETA01",
}
def cpi(
item: str = "all_items",
area: str = "0000",
seasonal: bool = False,
series: str = "U",
) -> str:
"""
CPI series ID.
Args:
item: Item code name (see _CPI_ITEM) or raw item code
area: 4-char area code ("0000" = US city average)
seasonal: True for seasonally adjusted
series: "U" = CPI-U, "W" = CPI-W, "S" = Chained CPI-U
Examples:
cpi() → "CUUR0000SA0"
cpi("core", seasonal=True) → "CUSR0000SA0L1E"
cpi("gasoline", seasonal=True) → "CUSR0000SS47014"
cpi("all_items", series="W") → "CWUR0000SA0"
"""
prefix_map = {"U": "CU", "W": "CW", "S": "SU"}
prefix = prefix_map.get(series, "CU")
adj = "S" if seasonal else "U"
itm = _CPI_ITEM.get(item, item)
return f"{prefix}{adj}R{area}{itm}"
# ---------------------------------------------------------------------------
# PPI — Producer Price Index (commodity-based)
# ---------------------------------------------------------------------------
def ppi_commodity(commodity_code: str = "00000000") -> str:
"""
PPI commodity series.
Args:
commodity_code: 8-char commodity code ("00000000" = all commodities)
Examples:
ppi_commodity() → "WPU00000000"
ppi_commodity("1012") → "WPU1012" (iron & steel scrap)
"""
return f"WPU{commodity_code}"
# ---------------------------------------------------------------------------
# OES — Occupational Employment & Wage Statistics
# ---------------------------------------------------------------------------
_OES_DATATYPE = {
"employment": "01",
"hourly_mean": "03",
"annual_mean": "04",
"hourly_10pct": "06",
"hourly_25pct": "07",
"hourly_median":"08",
"hourly_75pct": "09",
"hourly_90pct": "10",
"annual_median":"13",
}
def oes_national(
occupation_code: str = "000000",
industry_code: str = "000000",
data_type: str = "employment",
) -> str:
"""
OES national series.
Args:
occupation_code: 6-digit SOC code or "000000" for all occupations
industry_code: 6-digit NAICS or "000000" for cross-industry
data_type: Measure name (see _OES_DATATYPE) or 2-digit code
Examples:
oes_national() → all occupations, employment
oes_national("151132", data_type="annual_median") → software devs median wage
"""
dtype = _OES_DATATYPE.get(data_type, data_type)
# OE+U+N(area_type)+0000400(national area 7 chars)+industry(6)+occupation(6)+dtype(2) = 25
return f"OEUN0000400{industry_code:0<6}{occupation_code:0<6}{dtype}"
# ---------------------------------------------------------------------------
# JOLTS — Job Openings & Labor Turnover
# ---------------------------------------------------------------------------
_JOLTS_ELEMENT = {
"job_openings": "JO",
"hires": "HI",
"quits": "QU",
"layoffs": "LD",
"total_separations": "TS",
}
def jolts(
element: str = "job_openings",
rate_level: str = "L",
industry: str = "000000",
ownership: str = "00",
seasonal: bool = True,
) -> str:
"""
JOLTS series.
Args:
element: "job_openings" | "hires" | "quits" | "layoffs" | "total_separations"
rate_level: "L" = level (thousands) | "R" = rate
industry: 6-digit industry code; "000000" = total nonfarm
ownership: "00" = total | "10" = private | "20" = government
seasonal: True for SA
Examples:
jolts() → "JTS000000000000JOL" (openings level)
jolts("quits", rate_level="R") → "JTS000000000000QUR" (quits rate)
jolts("hires", seasonal=False) → "JTU000000000000HIL" (hires level, NSA)
"""
adj = "S" if seasonal else "U"
elem = _JOLTS_ELEMENT.get(element, element)
# Format: JT+adj+industry(6)+zeros(6)+elem(2)+rate_level(1) = 18 chars
ind = industry[:6].ljust(6, "0")
return f"JT{adj}{ind}{'0'*6}{elem}{rate_level}"
# ---------------------------------------------------------------------------
# ECI — Employment Cost Index
# ---------------------------------------------------------------------------
def eci(
worker_type: str = "10", # 10=civilian, 20=private, 30=state/local
occupation: str = "00", # 00=all, 10=mgmt/prof, 20=service, etc.
industry: str = "000000000",
component: str = "A", # A=total comp, W=wages, B=benefits
seasonal: bool = True,
) -> str:
"""
ECI series for employment cost changes.
Examples:
eci() → "CIU1010000000000A" (civilian, all workers, total compensation)
eci(component="W") → wages only
"""
adj = "S" if seasonal else "U"
return f"CI{adj}{worker_type}{occupation}{industry}{component}"
# ---------------------------------------------------------------------------
# Productivity (Major Sector)
# ---------------------------------------------------------------------------
_PR_SECTOR = {
"business": "85",
"nonfarm_business":"86",
"manufacturing": "88",
"durable_mfg": "89",
"nondurable_mfg": "90",
}
_PR_MEASURE = {
"output_per_hour": "092",
"output": "041",
"hours": "051",
"compensation": "061",
"real_comp_per_hr": "071",
"unit_labor_cost": "111",
"unit_nonlabor_pay":"112",
}
def productivity(
sector: str = "business",
measure: str = "output_per_hour",
seasonal: bool = True,
) -> str:
"""
Major Sector Productivity series.
Examples:
productivity() → "PRS85006092" (business output per hour, SA)
productivity("manufacturing", "unit_labor_cost")
"""
adj = "S" if seasonal else "U"
sec = _PR_SECTOR.get(sector, sector)
mea = _PR_MEASURE.get(measure, measure)
return f"PR{adj}{sec}06{mea}"