build(stage1): create primary outcome response layer

This commit is contained in:
Jinotech
2026-09-20 09:11:16 +12:00
parent 2bc0062aef
commit 89c20f32c7
7 changed files with 435 additions and 7 deletions
@@ -430,7 +430,7 @@ Generated from the direct-source registry and audits by `build_primary_outcome_p
- source conflicts or evidence gaps: {conflicts}
- human reviewers recorded: 0
The package does not satisfy Gate 1 by itself. Two actual human reviewers must independently sign every primary-outcome item; `responses_long.parquet` is not yet built; NSDUH youth attempt is absent from local public-use microdata; and several design and administered-language fields remain provisional.
The package does not satisfy Gate 1 by itself. `responses_long.parquet` has passed structural validation, but two actual human reviewers must independently sign every primary-outcome item; NSDUH youth attempt is absent from local public-use microdata; and several design and administered-language fields remain provisional.
"""
(STAGE1 / "primary_outcome_package.md").write_text(text, encoding="utf-8")
+241
View File
@@ -0,0 +1,241 @@
"""Build the Gate-1 long response table from source-verified primary outcomes."""
from __future__ import annotations
import csv
import hashlib
import json
from collections import Counter
from pathlib import Path
from typing import Any
import duckdb
import pyarrow as pa
import pyarrow.parquet as pq
ROOT = Path(__file__).resolve().parents[2]
STAGE1 = ROOT / "research" / "stage1"
GSHS = ROOT / "Dataset" / "GSHS-全球学生健康调查数据" / "GSHS" / "01_data" / "GSHS.csv"
YRBS_DIR = ROOT / "Dataset" / "可直接分析数据包_NSDUH_YRBS" / "YRBS" / "full_by_year"
NSDUH = ROOT / "Dataset" / "可直接分析数据包_NSDUH_YRBS" / "NSDUH" / "nsduh_2021_2024_full.parquet"
OUTPUT = ROOT / "Dataset" / "可直接分析数据包_NSDUH_YRBS" / "harmonization" / "responses_long.parquet"
MANIFEST = STAGE1 / "responses_long_manifest.json"
REPORT = STAGE1 / "responses_long_audit.md"
SCHEMA = pa.schema([
("respondent_id", pa.string()),
("survey_id", pa.string()),
("component_id", pa.string()),
("survey_year", pa.int16()),
("item_version_id", pa.string()),
("raw_response", pa.string()),
("canonical_response", pa.string()),
("eligibility", pa.string()),
("observed_status", pa.string()),
("missing_type", pa.string()),
("source_row", pa.int64()),
])
def load_csv(path: Path) -> list[dict[str, str]]:
with path.open(encoding="utf-8-sig", newline="") as handle:
return list(csv.DictReader(handle))
def normalized_code(value: Any) -> str:
if value is None:
return "<missing>"
if isinstance(value, float) and value.is_integer():
return str(int(value))
text = str(value).strip()
if not text:
return "<missing>"
if text.endswith(".0"):
try:
return str(int(float(text)))
except ValueError:
pass
return text
def sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for block in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(block)
return digest.hexdigest()
class LongWriter:
def __init__(self, path: Path, response_map: dict[tuple[str, str], dict[str, str]]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
self.writer = pq.ParquetWriter(path, SCHEMA, compression="zstd")
self.response_map = response_map
self.buffer: list[dict[str, Any]] = []
self.rows_by_survey: Counter[str] = Counter()
self.rows_by_item: Counter[str] = Counter()
self.observed_status: Counter[str] = Counter()
self.missing_types: Counter[str] = Counter()
def add(self, *, respondent_id: str, survey_id: str, component_id: str, year: int,
item_version_id: str, raw_value: Any, eligibility: str, source_row: int) -> None:
raw_code = normalized_code(raw_value)
key = (item_version_id, raw_code)
if key not in self.response_map:
raise ValueError(f"No response dictionary entry for {key}")
mapping = self.response_map[key]
observed = mapping["is_observed_response"] == "true"
missing_type = mapping["missing_type"]
status = "observed" if observed else (missing_type or "missing")
survey = item_version_id.split("::", 1)[0]
self.buffer.append({
"respondent_id": respondent_id,
"survey_id": survey_id,
"component_id": component_id,
"survey_year": year,
"item_version_id": item_version_id,
"raw_response": None if raw_code == "<missing>" else raw_code,
"canonical_response": mapping["canonical_code"] or None,
"eligibility": eligibility,
"observed_status": status,
"missing_type": missing_type or None,
"source_row": source_row,
})
self.rows_by_survey[survey] += 1
self.rows_by_item[item_version_id] += 1
self.observed_status[status] += 1
if missing_type:
self.missing_types[missing_type] += 1
if len(self.buffer) >= 50_000:
self.flush()
def flush(self) -> None:
if self.buffer:
self.writer.write_table(pa.Table.from_pylist(self.buffer, schema=SCHEMA))
self.buffer.clear()
def close(self) -> None:
self.flush()
self.writer.close()
def main() -> None:
item_rows = load_csv(STAGE1 / "item_bank.csv")
dictionary_rows = load_csv(STAGE1 / "response_dictionary.csv")
item_index = {(r["survey"], r["component_id"], r["source_variable"].lower()): r for r in item_rows}
response_map = {(r["item_version_id"], r["raw_code"]): r for r in dictionary_rows}
writer = LongWriter(OUTPUT, response_map)
expected_rows: dict[str, int] = {}
# GSHS: stream the large CSV, retaining only the seven source-verified components.
gshs_components = {r["component_id"] for r in item_rows if r["survey"] == "GSHS"}
gshs_vars = ["raw_mh_considersui", "raw_mh_plansui", "raw_mh_attemptsui"]
with GSHS.open(encoding="utf-8-sig", newline="") as handle:
reader = csv.DictReader(handle)
for source_row, row in enumerate(reader, start=1):
component = row["dataset_id"]
if component not in gshs_components:
continue
year = int(str(row["survey_year"])[:4])
survey_id = next(r["survey_id"] for r in item_rows if r["survey"] == "GSHS" and r["component_id"] == component)
respondent = f"{component}::{row['id']}"
for variable in gshs_vars:
item = item_index[("GSHS", component, variable)]
writer.add(respondent_id=respondent, survey_id=survey_id, component_id=component,
year=year, item_version_id=item["item_version_id"], raw_value=row[variable],
eligibility="eligible_administered_component", source_row=source_row)
gshs_audit = json.loads((STAGE1 / "gshs_primary_outcome_component_audit.json").read_text(encoding="utf-8"))
expected_rows["GSHS"] = sum(c["observed_rows"] for c in gshs_audit["components"] if c["primary_outcomes_observed"]) * 3
# YRBS: all 17 annual full tables have passed source/layout and value-domain checks.
yrbs_items = [r for r in item_rows if r["survey"] == "YRBS"]
yrbs_respondents = 0
for year in sorted({int(r["year"]) for r in yrbs_items}):
annual_items = [r for r in yrbs_items if int(r["year"]) == year]
columns = ["source_row", *[r["source_variable"].lower() for r in annual_items]]
table = pq.read_table(YRBS_DIR / f"yrbs_{year}_full.parquet", columns=columns).to_pydict()
yrbs_respondents += len(table["source_row"])
for offset, source_row in enumerate(table["source_row"]):
respondent = f"YRBS::{year}::{source_row}"
for item in annual_items:
variable = item["source_variable"].lower()
writer.add(respondent_id=respondent, survey_id=item["survey_id"], component_id=item["component_id"],
year=year, item_version_id=item["item_version_id"], raw_value=table[variable][offset],
eligibility="eligible_sampled_student", source_row=int(source_row))
expected_rows["YRBS"] = yrbs_respondents * 3
# NSDUH: restrict to CATAG2=1 (12-17); attempt is documented but absent from public-use data.
nsduh_items = [r for r in item_rows if r["survey"] == "NSDUH" and r["source_variable"] in {"YUSUITHK", "YUSUIPLN"}]
nsduh_table = pq.read_table(NSDUH, columns=["YEAR", "QUESTID2", "CATAG2", "YUSUITHK", "YUSUIPLN"]).to_pydict()
nsduh_youth = 0
for offset, age_group in enumerate(nsduh_table["CATAG2"]):
if normalized_code(age_group) != "1":
continue
nsduh_youth += 1
year = int(nsduh_table["YEAR"][offset])
annual_items = [r for r in nsduh_items if int(r["year"]) == year]
respondent = f"NSDUH::{year}::{nsduh_table['QUESTID2'][offset]}"
for item in annual_items:
variable = item["source_variable"]
writer.add(respondent_id=respondent, survey_id=item["survey_id"], component_id=item["component_id"],
year=year, item_version_id=item["item_version_id"], raw_value=nsduh_table[variable][offset],
eligibility="eligible_age_12_17", source_row=offset + 1)
expected_rows["NSDUH"] = nsduh_youth * 2
writer.close()
con = duckdb.connect()
path_sql = str(OUTPUT).replace("'", "''")
totals = con.execute(f"SELECT count(*), count(DISTINCT respondent_id || '|' || item_version_id), count(DISTINCT item_version_id) FROM read_parquet('{path_sql}')").fetchone()
null_checks = con.execute(f"SELECT count(*) FILTER (WHERE respondent_id IS NULL OR item_version_id IS NULL OR eligibility IS NULL OR observed_status IS NULL), count(*) FILTER (WHERE observed_status='observed' AND canonical_response IS NULL) FROM read_parquet('{path_sql}')").fetchone()
bad_codes = con.execute(f"SELECT count(*) FROM read_parquet('{path_sql}') WHERE observed_status='observed' AND raw_response IS NULL").fetchone()[0]
con.close()
if totals[0] != totals[1]:
raise AssertionError("respondent_id × item_version_id keys are not unique")
if null_checks != (0, 0) or bad_codes:
raise AssertionError(f"Long-table required-field/canonical checks failed: {null_checks=} {bad_codes=}")
if totals[2] != 80:
raise AssertionError(f"Expected responses for 80 locally analyzable items, found {totals[2]}")
if dict(writer.rows_by_survey) != expected_rows:
raise AssertionError(f"Source-derived row totals do not match output: {expected_rows=} actual={dict(writer.rows_by_survey)}")
manifest = {
"schema_version": "1.0", "generated_on": "2026-09-20", "status": "provisional",
"output_file": str(OUTPUT.relative_to(ROOT)).replace("\\", "/"),
"sha256": sha256(OUTPUT), "bytes": OUTPUT.stat().st_size,
"rows": totals[0], "unique_response_keys": totals[1], "item_instances_with_rows": totals[2],
"rows_by_survey": dict(sorted(writer.rows_by_survey.items())),
"source_derived_expected_rows": dict(sorted(expected_rows.items())),
"rows_by_item": dict(sorted(writer.rows_by_item.items())),
"observed_status_counts": dict(sorted(writer.observed_status.items())),
"missing_type_counts": dict(sorted(writer.missing_types.items())),
"required_field_nulls": null_checks[0], "observed_rows_without_canonical_response": null_checks[1],
"eligibility_rules": {
"GSHS": "Rows only from the seven components where the primary items were administered.",
"YRBS": "All respondents in each verified national annual full table; blank source responses remain unknown missing.",
"NSDUH": "CATAG2=1 (official label: 12-17 years old); only YUSUITHK and YUSUIPLN are locally available.",
},
"excluded_item_instances": [
r["item_version_id"] for r in item_rows
if r["survey"] == "NSDUH" and r["source_variable"] == "YUSUICTRY"
],
"gate_boundary": "No human reviewer fields are inferred. Four NSDUH attempt items have no response rows because the fields are absent from local public-use microdata.",
}
MANIFEST.write_text(json.dumps(manifest, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
REPORT.write_text(
"# Primary-outcome response long-table audit\n\n"
"Status: **provisional; structural validation passed**.\n\n"
f"- rows: {manifest['rows']:,}\n"
f"- unique respondent × item keys: {manifest['unique_response_keys']:,}\n"
f"- item instances with local response data: {manifest['item_instances_with_rows']} of 84\n"
f"- rows by survey: `{manifest['rows_by_survey']}`\n"
f"- observed/missing statuses: `{manifest['observed_status_counts']}`\n"
f"- SHA-256: `{manifest['sha256']}`\n\n"
"The output row totals equal counts independently derived from the seven GSHS components, all 17 YRBS annual full tables, and NSDUH respondents with CATAG2=1. Required identifiers/status fields are complete, respondent × item keys are unique, and every observed response has a canonical response. The four NSDUH attempt instances remain in the item bank without response rows because the public-use microdata omit those fields. Human double review is still required.\n",
encoding="utf-8",
)
print(json.dumps({k: manifest[k] for k in ["rows", "unique_response_keys", "item_instances_with_rows", "rows_by_survey", "observed_status_counts", "missing_type_counts", "sha256"]}, indent=2))
if __name__ == "__main__":
main()
+1 -1
View File
@@ -10,4 +10,4 @@ Generated from the direct-source registry and audits by `build_primary_outcome_p
- source conflicts or evidence gaps: 32
- human reviewers recorded: 0
The package does not satisfy Gate 1 by itself. Two actual human reviewers must independently sign every primary-outcome item; `responses_long.parquet` is not yet built; NSDUH youth attempt is absent from local public-use microdata; and several design and administered-language fields remain provisional.
The package does not satisfy Gate 1 by itself. `responses_long.parquet` has passed structural validation, but two actual human reviewers must independently sign every primary-outcome item; NSDUH youth attempt is absent from local public-use microdata; and several design and administered-language fields remain provisional.
+12
View File
@@ -0,0 +1,12 @@
# Primary-outcome response long-table audit
Status: **provisional; structural validation passed**.
- rows: 919,863
- unique respondent × item keys: 919,863
- item instances with local response data: 80 of 84
- rows by survey: `{'GSHS': 64599, 'NSDUH': 91236, 'YRBS': 764028}`
- observed/missing statuses: `{'data_error': 177, 'dont_know': 458, 'observed': 867995, 'ordinary_missing': 172, 'refused': 6544, 'unknown': 44517}`
- SHA-256: `0260386444fd2d4a9b0183ef79b171b4fa47ab1d7c1706b05619c14cc8f8a197`
The output row totals equal counts independently derived from the seven GSHS components, all 17 YRBS annual full tables, and NSDUH respondents with CATAG2=1. Required identifiers/status fields are complete, respondent × item keys are unique, and every observed response has a canonical response. The four NSDUH attempt instances remain in the item bank without response rows because the public-use microdata omit those fields. Human double review is still required.
@@ -0,0 +1,132 @@
{
"schema_version": "1.0",
"generated_on": "2026-09-20",
"status": "provisional",
"output_file": "Dataset/可直接分析数据包_NSDUH_YRBS/harmonization/responses_long.parquet",
"sha256": "0260386444fd2d4a9b0183ef79b171b4fa47ab1d7c1706b05619c14cc8f8a197",
"bytes": 4405969,
"rows": 919863,
"unique_response_keys": 919863,
"item_instances_with_rows": 80,
"rows_by_survey": {
"GSHS": 64599,
"NSDUH": 91236,
"YRBS": 764028
},
"source_derived_expected_rows": {
"GSHS": 64599,
"NSDUH": 91236,
"YRBS": 764028
},
"rows_by_item": {
"GSHS::2023_MAR_GSHS_v01::mar_fez2023::raw_mh_attemptsui": 4172,
"GSHS::2023_MAR_GSHS_v01::mar_fez2023::raw_mh_considersui": 4172,
"GSHS::2023_MAR_GSHS_v01::mar_fez2023::raw_mh_plansui": 4172,
"GSHS::GHA_2022_GSHS_v01::gha_sek_tak2022::raw_mh_attemptsui": 3796,
"GSHS::GHA_2022_GSHS_v01::gha_sek_tak2022::raw_mh_considersui": 3796,
"GSHS::GHA_2022_GSHS_v01::gha_sek_tak2022::raw_mh_plansui": 3796,
"GSHS::IND_2022_GSHS_v01::ind_jaipur_2022::raw_mh_attemptsui": 4299,
"GSHS::IND_2022_GSHS_v01::ind_jaipur_2022::raw_mh_considersui": 4299,
"GSHS::IND_2022_GSHS_v01::ind_jaipur_2022::raw_mh_plansui": 4299,
"GSHS::JAM_2023_GSHS_v01::jam_st_cath_2023::raw_mh_attemptsui": 3041,
"GSHS::JAM_2023_GSHS_v01::jam_st_cath_2023::raw_mh_considersui": 3041,
"GSHS::JAM_2023_GSHS_v01::jam_st_cath_2023::raw_mh_plansui": 3041,
"GSHS::MNG_2023_GSHS_v01::mng2023::raw_mh_attemptsui": 3311,
"GSHS::MNG_2023_GSHS_v01::mng2023::raw_mh_considersui": 3311,
"GSHS::MNG_2023_GSHS_v01::mng2023::raw_mh_plansui": 3311,
"GSHS::SLB_2023_GSHS_v01::slb_2023::raw_mh_attemptsui": 1995,
"GSHS::SLB_2023_GSHS_v01::slb_2023::raw_mh_considersui": 1995,
"GSHS::SLB_2023_GSHS_v01::slb_2023::raw_mh_plansui": 1995,
"GSHS::WLF_2023_GSHS_v01::wlf_2023::raw_mh_attemptsui": 919,
"GSHS::WLF_2023_GSHS_v01::wlf_2023::raw_mh_considersui": 919,
"GSHS::WLF_2023_GSHS_v01::wlf_2023::raw_mh_plansui": 919,
"NSDUH::2021::YUSUIPLN": 10743,
"NSDUH::2021::YUSUITHK": 10743,
"NSDUH::2022::YUSUIPLN": 11969,
"NSDUH::2022::YUSUITHK": 11969,
"NSDUH::2023::YUSUIPLN": 11572,
"NSDUH::2023::YUSUITHK": 11572,
"NSDUH::2024::YUSUIPLN": 11334,
"NSDUH::2024::YUSUITHK": 11334,
"YRBS::1991::Q19": 12272,
"YRBS::1991::Q20": 12272,
"YRBS::1991::Q21": 12272,
"YRBS::1993::Q24": 16296,
"YRBS::1993::Q25": 16296,
"YRBS::1993::Q26": 16296,
"YRBS::1995::Q22": 10904,
"YRBS::1995::Q23": 10904,
"YRBS::1995::Q24": 10904,
"YRBS::1997::Q22": 16263,
"YRBS::1997::Q23": 16263,
"YRBS::1997::Q24": 16263,
"YRBS::1999::Q23": 15349,
"YRBS::1999::Q24": 15349,
"YRBS::1999::Q25": 15349,
"YRBS::2001::Q24": 13601,
"YRBS::2001::Q25": 13601,
"YRBS::2001::Q26": 13601,
"YRBS::2003::Q24": 15214,
"YRBS::2003::Q25": 15214,
"YRBS::2003::Q26": 15214,
"YRBS::2005::Q24": 13917,
"YRBS::2005::Q25": 13917,
"YRBS::2005::Q26": 13917,
"YRBS::2007::Q24": 14041,
"YRBS::2007::Q25": 14041,
"YRBS::2007::Q26": 14041,
"YRBS::2009::Q24": 16410,
"YRBS::2009::Q25": 16410,
"YRBS::2009::Q26": 16410,
"YRBS::2011::Q25": 15425,
"YRBS::2011::Q26": 15425,
"YRBS::2011::Q27": 15425,
"YRBS::2013::Q27": 13583,
"YRBS::2013::Q28": 13583,
"YRBS::2013::Q29": 13583,
"YRBS::2015::Q27": 15624,
"YRBS::2015::Q28": 15624,
"YRBS::2015::Q29": 15624,
"YRBS::2017::Q26": 14765,
"YRBS::2017::Q27": 14765,
"YRBS::2017::Q28": 14765,
"YRBS::2019::Q26": 13677,
"YRBS::2019::Q27": 13677,
"YRBS::2019::Q28": 13677,
"YRBS::2021::Q26": 17232,
"YRBS::2021::Q27": 17232,
"YRBS::2021::Q28": 17232,
"YRBS::2023::Q27": 20103,
"YRBS::2023::Q28": 20103,
"YRBS::2023::Q29": 20103
},
"observed_status_counts": {
"data_error": 177,
"dont_know": 458,
"observed": 867995,
"ordinary_missing": 172,
"refused": 6544,
"unknown": 44517
},
"missing_type_counts": {
"data_error": 177,
"dont_know": 458,
"ordinary_missing": 172,
"refused": 6544,
"unknown": 44517
},
"required_field_nulls": 0,
"observed_rows_without_canonical_response": 0,
"eligibility_rules": {
"GSHS": "Rows only from the seven components where the primary items were administered.",
"YRBS": "All respondents in each verified national annual full table; blank source responses remain unknown missing.",
"NSDUH": "CATAG2=1 (official label: 12-17 years old); only YUSUITHK and YUSUIPLN are locally available."
},
"excluded_item_instances": [
"NSDUH::2021::YUSUICTRY",
"NSDUH::2022::YUSUICTRY",
"NSDUH::2023::YUSUICTRY",
"NSDUH::2024::YUSUICTRY"
],
"gate_boundary": "No human reviewer fields are inferred. Four NSDUH attempt items have no response rows because the fields are absent from local public-use microdata."
}