Skip to content

[PULL REQUEST] Align employment estimates with SANDAG categories #283

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
GregorSchroeder wants to merge 16 commits into
base: main
Choose a base branch
from
Open

[PULL REQUEST] Align employment estimates with SANDAG categories #283

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
eba0ef4
#281 - add ownership to SQL table
Jul 30, 2026
10fbc0b
#281 - adjust LODES sql query to use SANDAG employment categories
Jul 30, 2026
675435f
#281 - adjust Military sql query to use SANDAG employment categories
Jul 30, 2026
fa2ef5e
#281 - update BLS QCEW to use SANDAG employment; minor formatting fixes
Jul 30, 2026
c4e48f1
#281 - remove self employment SQL
Jul 30, 2026
e43c4e9
#281 - remove self employment python
Jul 30, 2026
6efc388
#281 - update QCEW sql to use SANDAG employment category function
Jul 31, 2026
da2bd84
#281 - update block to mgra xref to use sandag employment categories
Jul 31, 2026
f3c2480
#281 - minor SQL adjustments
Jul 31, 2026
94d8927
#281 - update tests to handle employment category tuple
Jul 31, 2026
16f4446
#281 - update employment python to handle ownership title
Jul 31, 2026
249a0e9
Potential fix for pull request finding
GregorSchroeder Jul 31, 2026
7f78735
Potential fix for pull request finding
GregorSchroeder Jul 31, 2026
d77f64d
Potential fix for pull request finding
GregorSchroeder Jul 31, 2026
82a8928
#283 - CoPilot PR suggestion
Jul 31, 2026
6635d07
Merge branch '281-feature-sandag-employment' of https://github.com/SA…
Jul 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,7 @@ direction TB
inputs_controls_jobs {
run_id INT UK, FK
year INT UK
ownership_title NVARCHAR(50) UK
industry_code NVARCHAR(5) UK
metric NVARCHAR(4) UK
value INT
Expand Down Expand Up @@ -209,6 +210,7 @@ direction TB
run_id INT UK, FK
year INT UK
mgra INT UK, FK
ownership_title NVARCHAR(50) UK
industry_code NVARCHAR(5) UK
value INT
}
Expand Down
260 changes: 73 additions & 187 deletions python/employment.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,7 @@ def _get_lodes_data(year: int) -> pd.DataFrame:
),
],
ignore_index=True,
)[["year", "block", "industry_code", "jobs"]]
)[["year", "block", "ownership_title", "industry_code", "jobs"]]

return combined_data

Expand All @@ -97,18 +97,18 @@ def _aggregate_lodes_to_mgra(
This function allocates jobs from Census blocks to MGRAs using distributions from
the California Employment Development Department (EDD) point-level dataset. Blocks
with no EDD data available use a simple land area intersection to allocate jobs to
MGRAs. The allocation first attempts to allocate within industry codes using EDD
data, then falls back to using EDD data without considering industry codes, and
finally falls back to using the land area intersection.
MGRAs. The allocation first attempts to allocate within SANDAG employment categories
using EDD data, then falls back to using EDD data without considering categories,
and finally falls back to using the land area intersection.

Args:
combined_data: LODES data with columns: year, block, industry_code, jobs
xref: Crosswalk with columns: block, mgra, pct_industry, pct_edd, pct_area, flag
xref: Crosswalk with columns: block, mgra, pct_edd_category, pct_edd, pct_area, flag
year: The year for which to aggregate data

Returns:
Aggregated data at MGRA level with columns: run_id, year, mgra,
industry_code, value
ownership_title, industry_code, value
"""
# Get MGRA data from SQL
with utils.ESTIMATES_ENGINE.connect() as con:
Expand All @@ -123,118 +123,44 @@ def _aggregate_lodes_to_mgra(
params={"run_id": utils.RUN_ID},
)

# Get unique industry codes and cross join with MGRA data
unique_industries = combined_data["industry_code"].unique()
jobs = (
mgra_data.merge(pd.DataFrame({"industry_code": unique_industries}), how="cross")
# Get unique SANDAG employment categories and cross join with MGRA data
mgra_data.merge(
combined_data[["ownership_title", "industry_code"]]
.drop_duplicates()
.reset_index(drop=True),
how="cross",
)
.assign(year=year)
# Get the LODES data and allocated to MGRAs using the crosswalk and allocation percentages
.merge(
combined_data.merge(xref, on=["block", "industry_code"], how="inner")
combined_data.merge(
xref, on=["block", "ownership_title", "industry_code"], how="inner"
)
.assign(
value=lambda df: df["jobs"]
* np.where(
df["flag"] == "pct_industry",
df["pct_industry"],
df["flag"] == "pct_edd_category",
df["pct_edd_category"],
np.where(df["flag"] == "pct_edd", df["pct_edd"], df["pct_area"]),
)
)
.groupby(["year", "mgra", "industry_code"], as_index=False)["value"]
.groupby(
["year", "mgra", "ownership_title", "industry_code"], as_index=False
)["value"]
.sum(),
on=["year", "mgra", "industry_code"],
on=["year", "mgra", "ownership_title", "industry_code"],
how="left",
)
.fillna({"value": 0})
.assign(run_id=utils.RUN_ID)[
["run_id", "year", "mgra", "industry_code", "value"]
["run_id", "year", "mgra", "ownership_title", "industry_code", "value"]
]
)

return jobs


def _distribute_self_emp_to_mgra(
b24080: pd.DataFrame, xref: pd.DataFrame
) -> pd.DataFrame:
"""Distribute subregional self-employment counts to MGRA level.

This function allocates self-employment counts from block groups (for years
post 2012) or tracts (for 2010-2012) to MGRAs using allocation percentages.
The allocation percentages are based on the distribution of persons aged
18-64, total population, or an equal split across intersection MGRAs
depending on data availability within the block group or tract.

Args:
b24080: DataFrame containing subregional self-employment counts.
Must include columns (year, geography, industry_code, value).
xref: Geography crosswalk DataFrame with columns
(geography, mgra, flag, pct_18_64, pct_pop, pct_split)

Returns:
Self employment counts at the MGRA level
"""
# Check that required columns are present
required_b24080_cols = {"year", "geography", "industry_code", "value"}
required_xref_cols = {
"geography",
"mgra",
"flag",
"pct_18_64",
"pct_pop",
"pct_split",
}
if not required_b24080_cols.issubset(b24080.columns):
raise ValueError(
f"B24080 DataFrame is missing required columns: {required_b24080_cols - set(b24080.columns)}"
)
if not required_xref_cols.issubset(xref.columns):
raise ValueError(
f"xref DataFrame is missing required columns: {required_xref_cols - set(xref.columns)}"
)

# Check that flag column only contains expected values
expected_flags = {"pct_18_64", "pct_pop", "pct_split"}
if not set(xref["flag"].unique()).issubset(expected_flags):
raise ValueError(
f"xref 'flag' column contains unexpected values: {set(xref['flag'].unique()) - expected_flags}"
)

# Merge subregional self employment counts with MGRA crosswalk
merged = b24080.merge(xref, on="geography", how="inner")

# Calculate weighted value based on flag
merged = merged.assign(
weighted_value=np.select(
[
merged["flag"] == "pct_18_64",
merged["flag"] == "pct_pop",
merged["flag"] == "pct_split",
],
[
merged["value"] * merged["pct_18_64"],
merged["value"] * merged["pct_pop"],
merged["value"] * merged["pct_split"],
],
default=np.nan,
)
)

if merged["weighted_value"].isna().any():
raise ValueError(
"Unexpected allocation flag found; expected one of {'pct_18_64', 'pct_pop', 'pct_split'}"
)

# Sum weighted values to the MGRA level
merged = (
merged.groupby(["year", "mgra", "industry_code"])["weighted_value"]
.sum()
.reset_index()
.assign(run_id=utils.RUN_ID)
.rename(columns={"weighted_value": "value"})
)[["run_id", "year", "mgra", "industry_code", "value"]]

return merged


def _get_jobs_inputs(year: int) -> dict[str, pd.DataFrame]:
"""Get input data related to jobs for a specified year.

Expand All @@ -257,51 +183,13 @@ def _get_jobs_inputs(year: int) -> dict[str, pd.DataFrame]:
params={
"year": year,
},
)

# Get self-employed totals and append to control_totals
with open(utils.SQL_FOLDER / "employment/get_region_self_emp.sql") as file:
self_emp_control = utils.read_sql_query_fallback(
sql=sql.text(file.read()),
con=con,
params={
"year": year,
},
)

jobs_inputs["control_totals"] = pd.concat(
[jobs_inputs["control_totals"], self_emp_control],
ignore_index=True,
)

jobs_inputs["control_totals"]["run_id"] = utils.RUN_ID

# Get self-employed block group data
with open(utils.SQL_FOLDER / "employment/get_B24080.sql") as file:
jobs_inputs["B24080"] = utils.read_sql_query_fallback(
sql=sql.text(file.read()),
con=con,
params={
"year": year,
},
)

# Get census block group or tract to MGRA crosswalk
with open(utils.SQL_FOLDER / "employment/xref_se_to_mgra.sql") as file:
jobs_inputs["xref_se_to_mgra"] = pd.read_sql_query(
sql=sql.text(file.read()),
con=con,
params={
"run_id": utils.RUN_ID,
"year": year,
},
)
).assign(run_id=utils.RUN_ID)

with utils.GIS_ENGINE.connect() as con:
# Get crosswalk from Census blocks to MGRAs
with open(utils.SQL_FOLDER / "employment/xref_block_to_mgra.sql") as file:
jobs_inputs["xref_block_to_mgra"] = utils.read_sql_query_fallback(
max_lookback=2,
max_lookback=1,
sql=sql.text(file.read()),
con=con,
params={
Expand All @@ -324,11 +212,12 @@ def _get_jobs_inputs(year: int) -> dict[str, pd.DataFrame]:

military_control_totals = (
jobs_inputs["military_emp"]
.groupby(["run_id", "year", "industry_code", "metric"], as_index=False)[
"value"
]
.groupby(
["run_id", "year", "ownership_title", "industry_code", "metric"],
as_index=False,
)["value"]
.sum()
)[["run_id", "year", "industry_code", "metric", "value"]]
)[["run_id", "year", "ownership_title", "industry_code", "metric", "value"]]

jobs_inputs["control_totals"] = pd.concat(
[jobs_inputs["control_totals"], military_control_totals],
Expand All @@ -348,28 +237,13 @@ def _validate_jobs_inputs(jobs_inputs: dict[str, pd.DataFrame]) -> None:
negative={},
null={},
)
# Self Employed only includes block groups with self-employed individuals therefore
# no row count validation performed
tests.validate_data(
"Self-employed block group data",
jobs_inputs["B24080"],
negative={},
null={},
)
# No row count validation performed as xref is many-to-many
# NULLs are allowed in the result set
tests.validate_data(
"xref_block_to_mgra",
jobs_inputs["xref_block_to_mgra"],
negative={},
)
# No row count validation performed as xref is many-to-many
tests.validate_data(
"xref_se_to_mgra",
jobs_inputs["xref_se_to_mgra"],
negative={},
null={},
)
tests.validate_data(
"Military employment data",
jobs_inputs["military_emp"],
Expand All @@ -380,7 +254,7 @@ def _validate_jobs_inputs(jobs_inputs: dict[str, pd.DataFrame]) -> None:
tests.validate_data(
"Jobs control totals",
jobs_inputs["control_totals"],
row_count={"key_columns": {"industry_code"}},
row_count={"key_columns": {("ownership_title", "industry_code")}},
negative={},
null={},
)
Expand All @@ -397,52 +271,64 @@ def _create_jobs_output(
Returns:
Controlled employment data.
"""
# Create MGRA level jobs data by combining LODES and self-employment data
# Create MGRA level jobs data by combining LODES and military data
mgra_jobs = pd.concat(
[
# Aggregate LODES jobs to MGRA level
_aggregate_lodes_to_mgra(
jobs_inputs["lodes_data"], jobs_inputs["xref_block_to_mgra"], year
),
# Distribute self-employment data to MGRA level
_distribute_self_emp_to_mgra(
jobs_inputs["B24080"], jobs_inputs["xref_se_to_mgra"]
),
# Include military employment at MGRA level
jobs_inputs["military_emp"][
["run_id", "year", "mgra", "industry_code", "value"]
["run_id", "year", "mgra", "ownership_title", "industry_code", "value"]
],
],
ignore_index=True,
).sort_values(by=["mgra", "industry_code"])
).sort_values(by=["mgra", "ownership_title", "industry_code"])

# Create list to store controlled values for each industry
results = []

# Apply integerize_1d to each industry_code
for industry_code in mgra_jobs["industry_code"].unique():
# Filter for this industry_code
naics_mask = mgra_jobs.loc[mgra_jobs["industry_code"] == industry_code]

# Get control value and apply integerize_1d
control_value = (
jobs_inputs["control_totals"]
.loc[
jobs_inputs["control_totals"]["industry_code"] == industry_code, "value"
# Apply integerize_1d to each SANDAG employment category
for ownership_title in mgra_jobs["ownership_title"].unique():
for industry_code in mgra_jobs["industry_code"].unique():
# Filter for this ownership_title and industry_code
mask = mgra_jobs.loc[
(mgra_jobs["ownership_title"] == ownership_title)
& (mgra_jobs["industry_code"] == industry_code)
]
.iloc[0]
)

results.append(
naics_mask.assign(
value=utils.integerize_1d(
data=naics_mask["value"],
control=control_value,
methodology="weighted_random",
generator=generator,
# If no records are returned, skip to next iteration
if mask.empty:
continue
else:
# Get control value and apply integerize_1d
control_value = (
jobs_inputs["control_totals"]
.loc[
(
jobs_inputs["control_totals"]["ownership_title"]
== ownership_title
)
& (
jobs_inputs["control_totals"]["industry_code"]
== industry_code
),
"value",
]
.iloc[0]
)
Comment thread
GregorSchroeder marked this conversation as resolved.

results.append(
mask.assign(
value=utils.integerize_1d(
data=mask["value"],
control=control_value,
methodology="weighted_random",
generator=generator,
)
)
)
)
)

return {"results": pd.concat(results, ignore_index=True)}

Expand All @@ -452,7 +338,7 @@ def _validate_jobs_outputs(jobs_outputs: dict[str, pd.DataFrame]) -> None:
tests.validate_data(
"Controlled jobs data",
jobs_outputs["results"],
row_count={"key_columns": {"mgra", "industry_code"}},
row_count={"key_columns": {"mgra", ("ownership_title", "industry_code")}},
negative={},
null={},
)
Expand Down
Loading