Skip to content

Code Reference

Generated API documentation for the modules that implement harmonization, data loading, and bootstrap analysis. The Streamlit UI layer (src/ui/) is presentation glue over these modules and is not documented here.

Harmonization

src.data.harmonizer

Data harmonization functions for multi-cycle CCHS analysis.

build_crosswalk 

build_crosswalk(cycles, descriptions, cutoff=0.6)

Build a crosswalk mapping reference-cycle variable names to each cycle's matching variable name, by fuzzy-matching variable descriptions.

The last entry in cycles is treated as the reference cycle: every other cycle's variable is matched against each reference variable's description via difflib.get_close_matches (single best match). This is a heuristic textual match, not an authoritative concordance - unmatched or ambiguous descriptions are recorded as None and should be reviewed.

Parameters:

Name Type Description Default
cycles list

Cycle years in order, with the reference cycle last

 required
descriptions dict

Dict mapping cycle -> {variable_name: description}

 required
cutoff float

difflib similarity cutoff (0-1) for a match to count

 0.6

Returns:

Type Description
dict

Dict mapping reference_var -> {cycle: cycle_specific_var_or_None}
Source code in src/data/harmonizer.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def build_crosswalk(cycles: list, descriptions: dict, cutoff: float = 0.6) -> dict:
    """
    Build a crosswalk mapping reference-cycle variable names to each cycle's
    matching variable name, by fuzzy-matching variable descriptions.

    The last entry in `cycles` is treated as the reference cycle: every
    other cycle's variable is matched against each reference variable's
    description via difflib.get_close_matches (single best match). This is
    a heuristic textual match, not an authoritative concordance - unmatched
    or ambiguous descriptions are recorded as None and should be reviewed.

    Args:
        cycles: Cycle years in order, with the reference cycle last
        descriptions: Dict mapping cycle -> {variable_name: description}
        cutoff: difflib similarity cutoff (0-1) for a match to count

    Returns:
        Dict mapping reference_var -> {cycle: cycle_specific_var_or_None}
    """
    reference_cycle = cycles[-1]
    reference_vars = descriptions[reference_cycle]

    crosswalk = {}
    for ref_var, ref_desc in reference_vars.items():
        crosswalk[ref_var] = {reference_cycle: ref_var}
        for cycle in cycles[:-1]:
            candidates = descriptions[cycle]
            best_match = difflib.get_close_matches(ref_desc, candidates.values(), n=1, cutoff=cutoff)
            if best_match:
                for var, desc in candidates.items():
                    if desc == best_match[0]:
                        crosswalk[ref_var][cycle] = var
                        break
            else:
                crosswalk[ref_var][cycle] = None

    return crosswalk

auto_harmonize 

auto_harmonize(label)

Normalize a raw codebook category label into a shared harmonized category.

Order matters: "not stated", "valid skip", "don't know", and "female" each contain "no" or "male" as a substring (e.g. "not stated", "female"), so the more specific phrases must be checked before the shorter "no"/"male" rules or they get misclassified.

Source code in src/data/harmonizer.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def auto_harmonize(label: str) -> str:
    """
    Normalize a raw codebook category label into a shared harmonized category.

    Order matters: "not stated", "valid skip", "don't know", and "female"
    each contain "no" or "male" as a substring (e.g. "**no**t stated",
    "fe**male**"), so the more specific phrases must be checked before the
    shorter "no"/"male" rules or they get misclassified.
    """
    l = label.lower()
    if "not stated" in l:
        return "Not stated"
    if "valid skip" in l:
        return "Valid skip"
    if "don’t know" in l or "don't know" in l:
        return "Don't know"
    if "female" in l:
        return "Female"
    if "male" in l:
        return "Male"
    if "yes" in l:
        return "Yes"
    if "no" in l:
        return "No"
    return label.strip()

get_common_harmonized_vars 

get_common_harmonized_vars(cycles, crosswalk, data_dict)

Get harmonized variables that exist in all specified cycles.

Parameters:

Name Type Description Default
cycles list

List of cycle years

 required
crosswalk dict

Crosswalk dictionary

 required
data_dict dict

Dictionary mapping cycle -> DataFrame (with cycle-specific column names)

 required

Returns:

Type Description
list

List of harmonized variable names available in all cycles
Source code in src/data/harmonizer.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def get_common_harmonized_vars(cycles: list, crosswalk: dict, data_dict: dict) -> list:
    """
    Get harmonized variables that exist in all specified cycles.

    Args:
        cycles: List of cycle years
        crosswalk: Crosswalk dictionary
        data_dict: Dictionary mapping cycle -> DataFrame (with cycle-specific column names)

    Returns:
        List of harmonized variable names available in all cycles
    """
    common_vars = []

    for harmonized_var, cycle_mapping in crosswalk.items():
        available_in_all = True
        for cycle in cycles:
            cycle_specific_var = cycle_mapping.get(cycle)
            if not cycle_specific_var or cycle_specific_var not in data_dict[cycle].columns:
                available_in_all = False
                break

        if available_in_all:
            common_vars.append(harmonized_var)

    return common_vars

src.data.codebook_extractor

Parsing logic for Statistics Canada CCHS Data Dictionary/Freqs PDF codebooks.

parse_codebook_lines 

parse_codebook_lines(lines)

Parse codebook text lines into a variable -> {description, categories} dict.

Expects the Statistics Canada CCHS Data Dictionary/Freqs layout: each variable starts with a "Variable Name:" line, has a "Concept:" line for its description, and an "Answer Categories" section listing "label code frequency frequency%" rows until a blank line or a Note:/Source:/ Universe:/Total line ends the section.

Parameters:

Name Type Description Default
lines Iterable[str]

An iterable of text lines, in document order (may span multiple PDF pages - state carries across page boundaries the same way it does across lines within a page).

 required

Returns:

Type Description
dict

Dict mapping variable_name -> {"description": str, "categories": {code: label}}
Source code in src/data/codebook_extractor.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def parse_codebook_lines(lines: Iterable[str]) -> dict:
    """
    Parse codebook text lines into a variable -> {description, categories} dict.

    Expects the Statistics Canada CCHS Data Dictionary/Freqs layout: each
    variable starts with a "Variable Name:" line, has a "Concept:" line for
    its description, and an "Answer Categories" section listing "label code
    frequency frequency%" rows until a blank line or a Note:/Source:/
    Universe:/Total line ends the section.

    Args:
        lines: An iterable of text lines, in document order (may span
            multiple PDF pages - state carries across page boundaries the
            same way it does across lines within a page).

    Returns:
        Dict mapping variable_name -> {"description": str, "categories": {code: label}}
    """
    variables = {}
    current_variable = None
    current_description = None
    codes = {}
    in_answer_categories = False

    for line in lines:
        if "Variable Name:" in line:
            if current_variable:
                variables[current_variable] = {
                    "description": current_description,
                    "categories": codes.copy() if codes else {}
                }
            current_variable = line.split(":")[1].strip().replace(" Length", "")
            current_description = None
            codes = {}
            in_answer_categories = False
        elif "Concept:" in line:
            current_description = line.split(":")[1].strip()
        elif "Answer Categories" in line:
            in_answer_categories = True
            continue
        elif in_answer_categories:
            match = CATEGORY_ROW.match(line)
            if match:
                meaning = match.group(1).strip()
                code = match.group(2).strip()
                codes[code] = meaning
            elif line.strip() == "" or SECTION_END.match(line):
                in_answer_categories = False
                continue
        elif "Note:" in line or "Source:" in line or "Universe:" in line:
            in_answer_categories = False
            continue

    if current_variable:
        variables[current_variable] = {
            "description": current_description,
            "categories": codes.copy() if codes else {}
        }

    return variables

extract_variables_with_categories 

extract_variables_with_categories(pdf_path)

Extract variable/category metadata from a CCHS codebook PDF.

Source code in src/data/codebook_extractor.py 
73
74
75
76
77
78
79
80
81
82
83
def extract_variables_with_categories(pdf_path) -> dict:
    """Extract variable/category metadata from a CCHS codebook PDF."""
    from pypdf import PdfReader

    lines = []
    with open(pdf_path, "rb") as file:
        reader = PdfReader(file)
        for page in reader.pages:
            lines.extend(page.extract_text().splitlines())

    return parse_codebook_lines(lines)

Data loading and precomputation

src.data.loader

Data loading functions for CCHS analysis.

build_harmonization_mapping 

build_harmonization_mapping(crosswalk, cycle, available_columns)

Build a safe rename mapping for one cycle.

If multiple harmonized variables point at the same source column, prefer the identity mapping (e.g. GEODVHR4 -> GEODVHR4) and otherwise keep the first mapping encountered. This prevents exact-match geography columns from being renamed away by alias entries later in the crosswalk.

Source code in src/data/loader.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def build_harmonization_mapping(crosswalk: dict, cycle: str, available_columns) -> tuple[dict, list]:
    """
    Build a safe rename mapping for one cycle.

    If multiple harmonized variables point at the same source column, prefer the
    identity mapping (e.g. GEODVHR4 -> GEODVHR4) and otherwise keep the first
    mapping encountered. This prevents exact-match geography columns from being
    renamed away by alias entries later in the crosswalk.
    """
    rename_dict = {}
    available_vars = []
    available_column_set = set(available_columns)

    for harmonized_var, cycle_mapping in crosswalk.items():
        cycle_specific_var = cycle_mapping.get(cycle)
        if not cycle_specific_var or cycle_specific_var == "Not Available":
            continue
        if cycle_specific_var not in available_column_set:
            continue

        existing_target = rename_dict.get(cycle_specific_var)
        if existing_target is None:
            rename_dict[cycle_specific_var] = harmonized_var
            available_vars.append(harmonized_var)
            continue

        if existing_target == cycle_specific_var:
            continue

        if harmonized_var == cycle_specific_var:
            rename_dict[cycle_specific_var] = harmonized_var
            if existing_target in available_vars:
                available_vars.remove(existing_target)
            if harmonized_var not in available_vars:
                available_vars.append(harmonized_var)

    return rename_dict, available_vars

restore_geography_aliases 

restore_geography_aliases(df)

Restore expected geography columns when an older harmonization pass renamed them to alias fields.

Source code in src/data/loader.py 
49
50
51
52
53
54
55
56
57
58
59
def restore_geography_aliases(df: pd.DataFrame) -> pd.DataFrame:
    """
    Restore expected geography columns when an older harmonization pass renamed
    them to alias fields.
    """
    restored = df.copy()

    if 'GEODVHR4' not in restored.columns and 'GEODVOHR' in restored.columns:
        restored['GEODVHR4'] = restored['GEODVOHR']

    return restored

load_cycle_data 

load_cycle_data(cycle)

Load main data and bootstrap data for a specific cycle.

Source code in src/data/loader.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
@st.cache_data
def load_cycle_data(cycle: str) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Load main data and bootstrap data for a specific cycle."""
    data_file = os.path.join(DATA_PATH, f"hs{cycle}_on_distr.parquet")
    bootstrap_file = os.path.join(DATA_PATH, f"hs{cycle}_on_bootwt.parquet")

    if os.path.exists(data_file) and os.path.exists(bootstrap_file):
        data = pd.read_parquet(data_file)
        bootstrap_data = pd.read_parquet(bootstrap_file)
        return data, bootstrap_data
    else:
        st.error(f"Data files for cycle {cycle} are missing. Please check the 'data/' directory.")
        return None, None

load_variable_descriptions 

load_variable_descriptions(cycle)

Load variable descriptions from harmonized JSON file for a specific cycle.

Source code in src/data/loader.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
@st.cache_data
def load_variable_descriptions(cycle: str) -> tuple[pd.DataFrame, dict]:
    """Load variable descriptions from harmonized JSON file for a specific cycle."""
    try:
        # Load from harmonized JSON file first
        json_file = os.path.join("harmonization", f"CCHS_{cycle}.json")
        if os.path.exists(json_file):
            with open(json_file, "r", encoding="utf-8") as f:
                var_dict = json.load(f)

            # Extract descriptions from JSON structure
            descriptions_data = []
            desc_dict = {}

            for var_name, var_info in var_dict.items():
                description = var_info.get("description", "")
                if description:
                    descriptions_data.append({
                        'Variable': var_name,
                        'Description': description
                    })
                    desc_dict[var_name] = description

            # Ensure all variables are included, even if description is missing
            for var_name in var_dict.keys():
                if var_name not in desc_dict:
                    descriptions_data.append({
                        'Variable': var_name,
                        'Description': "No description available"
                    })
                    desc_dict[var_name] = "No description available"

            # Create DataFrame from extracted data
            if descriptions_data:
                desc_df = pd.DataFrame(descriptions_data)
                return desc_df, desc_dict

        # Fallback to CSV file if JSON doesn't exist or is empty
        desc_file = os.path.join(DATA_PATH, f"CCHS_{cycle}_Recoded_Variables.csv")
        if os.path.exists(desc_file):
            descriptions = pd.read_csv(desc_file)
            desc_dict = dict(zip(descriptions['Variable'], descriptions['Description']))
            return descriptions, desc_dict

        # If neither exists, return empty
        return None, {}

    except Exception as e:
        st.error(f"Error loading variable descriptions for cycle {cycle}: {e}")
        return None, {}

load_json_variable_descriptions 

load_json_variable_descriptions(cycle)

Load JSON variable descriptions for a specific cycle (legacy function for compatibility).

Source code in src/data/loader.py 
129
130
131
132
133
134
135
136
137
@st.cache_data
def load_json_variable_descriptions(cycle: str) -> dict:
    """Load JSON variable descriptions for a specific cycle (legacy function for compatibility)."""
    json_file = os.path.join("harmonization", f"CCHS_{cycle}.json")
    if os.path.exists(json_file):
        with open(json_file, "r", encoding="utf-8") as f:
            var_dict = json.load(f)
        return {k: v.get("description", "") for k, v in var_dict.items()}
    return {}

load_cycle_variable_info 

load_cycle_variable_info(cycle)

Load complete variable information (descriptions AND categories) for a specific cycle.

Source code in src/data/loader.py 
140
141
142
143
144
145
146
147
@st.cache_data
def load_cycle_variable_info(cycle: str) -> dict:
    """Load complete variable information (descriptions AND categories) for a specific cycle."""
    json_file = os.path.join("harmonization", f"CCHS_{cycle}.json")
    if os.path.exists(json_file):
        with open(json_file, "r", encoding="utf-8") as f:
            return json.load(f)
    return {}

load_crosswalk 

load_crosswalk()

Load harmonization crosswalk.

Source code in src/data/loader.py 
150
151
152
153
154
155
156
157
@st.cache_data
def load_crosswalk() -> dict:
    """Load harmonization crosswalk."""
    crosswalk_file = os.path.join("harmonization", "crosswalk.json")
    if os.path.exists(crosswalk_file):
        with open(crosswalk_file, "r", encoding="utf-8") as f:
            return json.load(f)
    return {}

load_categories 

load_categories()

Load harmonization categories.

Source code in src/data/loader.py 
160
161
162
163
164
165
166
167
@st.cache_data
def load_categories() -> dict:
    """Load harmonization categories."""
    categories_file = os.path.join("harmonization", "categories.json")
    if os.path.exists(categories_file):
        with open(categories_file, "r", encoding="utf-8") as f:
            return json.load(f)
    return {}

load_ontario_csd_lookup 

load_ontario_csd_lookup()

Load the Ontario census subdivision code-to-name lookup.

Source code in src/data/loader.py 
170
171
172
173
174
175
176
177
@st.cache_data
def load_ontario_csd_lookup() -> dict:
    """Load the Ontario census subdivision code-to-name lookup."""
    lookup_file = os.path.join("harmonization", "ontario_csd_lookup.json")
    if os.path.exists(lookup_file):
        with open(lookup_file, "r", encoding="utf-8") as f:
            return json.load(f)
    return {}

load_ontario_official_municipalities 

load_ontario_official_municipalities()

Load the official Ontario municipalities lookup keyed by CSD code.

Source code in src/data/loader.py 
180
181
182
183
184
185
186
187
@st.cache_data
def load_ontario_official_municipalities() -> dict:
    """Load the official Ontario municipalities lookup keyed by CSD code."""
    lookup_file = os.path.join("harmonization", "ontario_official_municipalities.json")
    if os.path.exists(lookup_file):
        with open(lookup_file, "r", encoding="utf-8") as f:
            return json.load(f)
    return {}

merge_data 

merge_data(filtered_data, bootstrap_data)

Merge survey records with bootstrap weights without pooling cycles.

Multi-cycle trend analysis keeps each survey cycle independent. When a CYCLE column is present, it must be present on both frames and becomes part of the join key. The one-to-one validation prevents repeated IDs from silently multiplying records and biasing estimates.

Source code in src/data/loader.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
@st.cache_data
def merge_data(filtered_data, bootstrap_data):
    """Merge survey records with bootstrap weights without pooling cycles.

    Multi-cycle trend analysis keeps each survey cycle independent. When a
    ``CYCLE`` column is present, it must be present on both frames and becomes
    part of the join key. The one-to-one validation prevents repeated IDs from
    silently multiplying records and biasing estimates.
    """
    survey_has_cycle = 'CYCLE' in filtered_data.columns
    bootstrap_has_cycle = 'CYCLE' in bootstrap_data.columns

    if survey_has_cycle != bootstrap_has_cycle:
        raise ValueError(
            "Cycle-aware analysis requires a CYCLE column on both survey and "
            "bootstrap data. Regenerate precomputed files if necessary."
        )

    join_columns = ['ONT_ID']
    if survey_has_cycle:
        join_columns.insert(0, 'CYCLE')

    return pd.merge(
        filtered_data,
        bootstrap_data,
        on=join_columns,
        how='left',
        validate='one_to_one',
    )

load_multi_cycle_data 

load_multi_cycle_data(cycles, crosswalk, categories)

Load and harmonize data from multiple cycles using pre-computed crosswalk. This uses simple column renaming (no value transformation) for performance.

Parameters:

Name Type Description Default
cycles list

List of cycle years to load (e.g., ["2021", "2022", "2023"])

 required
crosswalk dict

Crosswalk dictionary mapping harmonized_var -> {cycle: cycle_specific_var}

 required
categories dict

Categories dictionary (not used, kept for compatibility)

 required

Returns:

Type Description

Tuple of (combined_data, combined_bootstrap_data) or (None, None) if error
Source code in src/data/loader.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
@st.cache_data
def load_multi_cycle_data(cycles: list, crosswalk: dict, categories: dict):
    """
    Load and harmonize data from multiple cycles using pre-computed crosswalk.
    This uses simple column renaming (no value transformation) for performance.

    Args:
        cycles: List of cycle years to load (e.g., ["2021", "2022", "2023"])
        crosswalk: Crosswalk dictionary mapping harmonized_var -> {cycle: cycle_specific_var}
        categories: Categories dictionary (not used, kept for compatibility)

    Returns:
        Tuple of (combined_data, combined_bootstrap_data) or (None, None) if error
    """
    if not cycles:
        st.error("No cycles specified for multi-cycle loading.")
        return None, None

    if not crosswalk:
        st.warning("No crosswalk provided. Loading cycles without harmonization.")

    combined_data_list = []
    combined_bootstrap_list = []

    for cycle in cycles:
        # Load raw data for this cycle
        data, bootstrap_data = load_cycle_data(cycle)
        if data is None or bootstrap_data is None:
            st.warning(f"Skipping cycle {cycle} due to missing data files.")
            continue

        rename_dict = {}
        if crosswalk:
            rename_dict, _ = build_harmonization_mapping(crosswalk, cycle, data.columns)

        # Apply harmonization - just column rename, no value transformation (fast!)
        harmonized_data = data.rename(columns=rename_dict) if rename_dict else data.copy()
        harmonized_data = restore_geography_aliases(harmonized_data)
        harmonized_data['CYCLE'] = cycle

        cycle_bootstrap = bootstrap_data.copy()
        cycle_bootstrap['CYCLE'] = cycle

        combined_data_list.append(harmonized_data)
        combined_bootstrap_list.append(cycle_bootstrap)

    if not combined_data_list:
        st.error("No valid cycles could be loaded.")
        return None, None

    # Before the concat line, check for and remove duplicate columns
    for i, df in enumerate(combined_data_list):
        duplicates = df.columns[df.columns.duplicated()].unique()
        if len(duplicates) > 0:
            import streamlit as st
            st.warning(f"Cycle {cycles[i]} has duplicate columns: {duplicates.tolist()}")
            print(f"Duplicate columns in cycle {cycles[i]}: {duplicates.tolist()}")
            # Remove duplicate columns, keeping first occurrence
            combined_data_list[i] = df.loc[:, ~df.columns.duplicated()]

    # Same check for bootstrap data if it exists
    if combined_bootstrap_list:
        for i, df in enumerate(combined_bootstrap_list):
            duplicates = df.columns[df.columns.duplicated()].unique()
            if len(duplicates) > 0:
                import streamlit as st
                st.warning(f"Cycle {cycles[i]} bootstrap data has duplicate columns: {duplicates.tolist()}")
                combined_bootstrap_list[i] = df.loc[:, ~df.columns.duplicated()]

    # Combine all cycles (fast concat operation)
    combined_data = pd.concat(combined_data_list, ignore_index=True)
    combined_bootstrap = pd.concat(combined_bootstrap_list, ignore_index=True)

    return combined_data, combined_bootstrap

load_data 

load_data()

Legacy function - loads 2021 data by default for backward compatibility.

Source code in src/data/loader.py 
298
299
300
301
@st.cache_data
def load_data():
    """Legacy function - loads 2021 data by default for backward compatibility."""
    return load_cycle_data("2021")

src.data.precompute

Precompute harmonized datasets for multi-cycle analysis.

check_precompute_status 

check_precompute_status(cycles, save_dir=PRECOMPUTE_DIR)

Check which cycles have been precomputed.

Parameters:

Name Type Description Default
cycles List[str]

List of cycle years (e.g., ["2021", "2022", "2023"])

 required
save_dir Path

Directory where precomputed data is stored

 PRECOMPUTE_DIR

Returns:

Type Description
Dict[str, bool]

Dictionary mapping cycle -> True/False (precomputed status)
Source code in src/data/precompute.py 
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def check_precompute_status(cycles: List[str], save_dir: Path = PRECOMPUTE_DIR) -> Dict[str, bool]:
    """
    Check which cycles have been precomputed.

    Args:
        cycles: List of cycle years (e.g., ["2021", "2022", "2023"])
        save_dir: Directory where precomputed data is stored

    Returns:
        Dictionary mapping cycle -> True/False (precomputed status)
    """
    status = {}

    for cycle in cycles:
        data_path = save_dir / f"harmonized_data_{cycle}.parquet"
        bootstrap_path = save_dir / f"harmonized_bootstrap_{cycle}.parquet"
        meta_path = save_dir / f"metadata_{cycle}.json"
        status[cycle] = (
            data_path.exists()
            and bootstrap_path.exists()
            and meta_path.exists()
        )

    return status

load_precomputed_data 

load_precomputed_data(cycle, save_dir=PRECOMPUTE_DIR)

Load precomputed harmonized data for a cycle.

Parameters:

Name Type Description Default
cycle str

Cycle year (e.g., "2021")

 required
save_dir Path

Directory where precomputed data is stored

 PRECOMPUTE_DIR

Returns:

Type Description
DataFrame

DataFrame with harmonized variables

Raises:

Type Description
FileNotFoundError

If precomputed data doesn't exist
Source code in src/data/precompute.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def load_precomputed_data(cycle: str, save_dir: Path = PRECOMPUTE_DIR) -> pd.DataFrame:
    """
    Load precomputed harmonized data for a cycle.

    Args:
        cycle: Cycle year (e.g., "2021")
        save_dir: Directory where precomputed data is stored

    Returns:
        DataFrame with harmonized variables

    Raises:
        FileNotFoundError: If precomputed data doesn't exist
    """
    path = save_dir / f"harmonized_data_{cycle}.parquet"

    if not path.exists():
        raise FileNotFoundError(f"Precomputed data not found: {path}")

    return pd.read_parquet(path)

load_precomputed_bootstrap 

load_precomputed_bootstrap(cycle, save_dir=PRECOMPUTE_DIR)

Load precomputed bootstrap weights for a cycle.

Parameters:

Name Type Description Default
cycle str

Cycle year (e.g., "2021")

 required
save_dir Path

Directory where precomputed data is stored

 PRECOMPUTE_DIR

Returns:

Type Description
DataFrame

DataFrame with bootstrap weights (ONT_ID and BSW columns)

Raises:

Type Description
FileNotFoundError

If precomputed data doesn't exist
Source code in src/data/precompute.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def load_precomputed_bootstrap(cycle: str, save_dir: Path = PRECOMPUTE_DIR) -> pd.DataFrame:
    """
    Load precomputed bootstrap weights for a cycle.

    Args:
        cycle: Cycle year (e.g., "2021")
        save_dir: Directory where precomputed data is stored

    Returns:
        DataFrame with bootstrap weights (ONT_ID and BSW columns)

    Raises:
        FileNotFoundError: If precomputed data doesn't exist
    """
    path = save_dir / f"harmonized_bootstrap_{cycle}.parquet"

    if not path.exists():
        raise FileNotFoundError(f"Precomputed bootstrap data not found: {path}")

    return pd.read_parquet(path)

load_precomputed_metadata 

load_precomputed_metadata(cycle, save_dir=PRECOMPUTE_DIR)

Load metadata for precomputed cycle.

Parameters:

Name Type Description Default
cycle str

Cycle year (e.g., "2021")

 required
save_dir Path

Directory where precomputed data is stored

 PRECOMPUTE_DIR

Returns:

Type Description
Dict

Dictionary with metadata (available_vars, record_count, etc.)

Raises:

Type Description
FileNotFoundError

If metadata doesn't exist
Source code in src/data/precompute.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def load_precomputed_metadata(cycle: str, save_dir: Path = PRECOMPUTE_DIR) -> Dict:
    """
    Load metadata for precomputed cycle.

    Args:
        cycle: Cycle year (e.g., "2021")
        save_dir: Directory where precomputed data is stored

    Returns:
        Dictionary with metadata (available_vars, record_count, etc.)

    Raises:
        FileNotFoundError: If metadata doesn't exist
    """
    path = save_dir / f"metadata_{cycle}.json"

    if not path.exists():
        raise FileNotFoundError(f"Metadata not found: {path}")

    with open(path, 'r', encoding='utf-8') as f:
        return json.load(f)

get_common_variables 

get_common_variables(cycles, save_dir=PRECOMPUTE_DIR)

Get variables available across all selected cycles using precomputed metadata.

Parameters:

Name Type Description Default
cycles List[str]

List of cycle years

 required
save_dir Path

Directory where precomputed data is stored

 PRECOMPUTE_DIR

Returns:

Type Description
List[str]

Sorted list of common harmonized variable names
Source code in src/data/precompute.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def get_common_variables(cycles: List[str], save_dir: Path = PRECOMPUTE_DIR) -> List[str]:
    """
    Get variables available across all selected cycles using precomputed metadata.

    Args:
        cycles: List of cycle years
        save_dir: Directory where precomputed data is stored

    Returns:
        Sorted list of common harmonized variable names
    """
    if not cycles:
        return []

    # Load metadata for each cycle
    all_vars = []
    for cycle in cycles:
        try:
            meta = load_precomputed_metadata(cycle, save_dir)
            all_vars.append(set(meta['available_vars']))
        except FileNotFoundError:
            return []  # Missing precomputed data

    # Find intersection
    common = set.intersection(*all_vars) if all_vars else set()
    return sorted(list(common))

precompute_cycle_data 

precompute_cycle_data(cycle, crosswalk, data_path='data', save_dir=PRECOMPUTE_DIR)

Precompute harmonized data for a single cycle.

Creates: - harmonized_data_{cycle}.parquet: Harmonized survey data - harmonized_bootstrap_{cycle}.parquet: Bootstrap weights with ONT_ID - metadata_{cycle}.json: Variable metadata and availability

Category value labels are not baked in here - they're resolved at display time from categories.json via get_value_label().

Parameters:

Name Type Description Default
cycle str

Cycle year (e.g., "2021")

 required
crosswalk Dict

Crosswalk dictionary for variable harmonization

 required
data_path str

Path to raw data files

 'data'
save_dir Path

Directory to save precomputed files

 PRECOMPUTE_DIR

Returns:

Type Description
Tuple[DataFrame, DataFrame, Dict]

Tuple of (harmonized_data, harmonized_bootstrap, metadata)
Source code in src/data/precompute.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
@st.cache_data
def precompute_cycle_data(
    cycle: str,
    crosswalk: Dict,
    data_path: str = "data",
    save_dir: Path = PRECOMPUTE_DIR
) -> Tuple[pd.DataFrame, pd.DataFrame, Dict]:
    """
    Precompute harmonized data for a single cycle.

    Creates:
    - harmonized_data_{cycle}.parquet: Harmonized survey data
    - harmonized_bootstrap_{cycle}.parquet: Bootstrap weights with ONT_ID
    - metadata_{cycle}.json: Variable metadata and availability

    Category value labels are not baked in here - they're resolved at
    display time from categories.json via get_value_label().

    Args:
        cycle: Cycle year (e.g., "2021")
        crosswalk: Crosswalk dictionary for variable harmonization
        data_path: Path to raw data files
        save_dir: Directory to save precomputed files

    Returns:
        Tuple of (harmonized_data, harmonized_bootstrap, metadata)
    """
    import os
    print(f"Precomputing {cycle}...")

    # Create output directory if needed
    save_dir.mkdir(parents=True, exist_ok=True)

    # Load raw data
    data_file = os.path.join(data_path, f"hs{cycle}_on_distr.parquet")
    bootstrap_file = os.path.join(data_path, f"hs{cycle}_on_bootwt.parquet")

    if not os.path.exists(data_file) or not os.path.exists(bootstrap_file):
        raise FileNotFoundError(f"Raw data files not found for cycle {cycle}")

    data = pd.read_parquet(data_file)
    bootstrap_data = pd.read_parquet(bootstrap_file)

    # Create harmonization mapping for this cycle
    rename_dict, available_vars = build_harmonization_mapping(
        crosswalk, cycle, data.columns
    )

    # Apply harmonization (rename columns)
    harmonized_data = data.rename(columns=rename_dict)
    harmonized_data = restore_geography_aliases(harmonized_data)

    # Handle duplicate columns (keep first occurrence only)
    # This can happen when multiple cycle-specific variables map to same harmonized name
    if harmonized_data.columns.duplicated().any():
        duplicate_cols = harmonized_data.columns[harmonized_data.columns.duplicated()].unique().tolist()
        print(f"   ⚠️  Warning: Removing {len(duplicate_cols)} duplicate columns: {duplicate_cols[:10]}{'...' if len(duplicate_cols) > 10 else ''}")
        harmonized_data = harmonized_data.loc[:, ~harmonized_data.columns.duplicated(keep='first')]

    # Add CYCLE column for multi-cycle identification
    harmonized_data['CYCLE'] = cycle

    # Keep only harmonized columns plus core variables
    core_vars = ['ONT_ID', 'WTS_S', 'GEODVHR4', 'GEODVCSD']
    # Add age column (varies by cycle)
    if 'DHH_AGE' in harmonized_data.columns:
        core_vars.append('DHH_AGE')
    if 'AWCAGE' in harmonized_data.columns:
        core_vars.append('AWCAGE')

    # Keep harmonized vars + core vars that exist
    cols_to_keep = list(set(available_vars + core_vars + ['CYCLE']) & set(harmonized_data.columns))
    harmonized_data = harmonized_data[cols_to_keep]

    # Cycle-qualified bootstrap data keeps trend comparisons independent and
    # prevents repeated ONT_ID values in different years from cross-joining.
    harmonized_bootstrap = bootstrap_data.copy()
    harmonized_bootstrap['CYCLE'] = cycle

    # Save harmonized data
    output_path = save_dir / f"harmonized_data_{cycle}.parquet"
    harmonized_data.to_parquet(output_path, index=False)

    # Save harmonized bootstrap
    bootstrap_output_path = save_dir / f"harmonized_bootstrap_{cycle}.parquet"
    harmonized_bootstrap.to_parquet(bootstrap_output_path, index=False)

    # Save metadata
    metadata = {
        'cycle': cycle,
        'available_vars': available_vars,
        'record_count': len(harmonized_data),
        'core_vars': core_vars,
        'precompute_date': pd.Timestamp.now().isoformat(),
        'harmonization_mapping': rename_dict
    }

    metadata_path = save_dir / f"metadata_{cycle}.json"
    with open(metadata_path, 'w', encoding='utf-8') as f:
        json.dump(metadata, f, indent=2)

    print(f"✅ {cycle}: {len(available_vars)} variables, {len(harmonized_data):,} records")
    print(f"   Saved to: {output_path}")

    return harmonized_data, harmonized_bootstrap, metadata

precompute_all_cycles 

precompute_all_cycles(cycles, crosswalk, data_path='data', save_dir=PRECOMPUTE_DIR)

Precompute data for all specified cycles.

Parameters:

Name Type Description Default
cycles List[str]

List of cycle years to precompute

 required
crosswalk Dict

Crosswalk dictionary

 required
data_path str

Path to raw data files

 'data'
save_dir Path

Directory to save precomputed files

 PRECOMPUTE_DIR

Returns:

Type Description
Dict[str, Dict]

Dictionary mapping cycle -> {'data': DataFrame, 'bootstrap': DataFrame, 'metadata': Dict}
Source code in src/data/precompute.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
def precompute_all_cycles(
    cycles: List[str],
    crosswalk: Dict,
    data_path: str = "data",
    save_dir: Path = PRECOMPUTE_DIR
) -> Dict[str, Dict]:
    """
    Precompute data for all specified cycles.

    Args:
        cycles: List of cycle years to precompute
        crosswalk: Crosswalk dictionary
        data_path: Path to raw data files
        save_dir: Directory to save precomputed files

    Returns:
        Dictionary mapping cycle -> {'data': DataFrame, 'bootstrap': DataFrame, 'metadata': Dict}
    """
    results = {}

    for cycle in cycles:
        try:
            data, bootstrap, meta = precompute_cycle_data(
                cycle, crosswalk, data_path, save_dir
            )
            results[cycle] = {
                'data': data,
                'bootstrap': bootstrap,
                'metadata': meta
            }
        except Exception as e:
            print(f"❌ Failed to precompute {cycle}: {e}")
            import traceback
            traceback.print_exc()

    return results

run_precompute_workflow 

run_precompute_workflow(cycles, crosswalk, data_path='data', save_dir=PRECOMPUTE_DIR)

Run the full precompute workflow and return a compact status summary.

Parameters:

Name Type Description Default
cycles List[str]

List of cycle years to precompute

 required
crosswalk Dict

Crosswalk dictionary

 required
data_path str

Path to raw data files

 'data'
save_dir Path

Directory to save precomputed files

 PRECOMPUTE_DIR

Returns:

Type Description
Dict

Dictionary with success flag, validation results, and per-cycle summary.
Source code in src/data/precompute.py 
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
def run_precompute_workflow(
    cycles: List[str],
    crosswalk: Dict,
    data_path: str = "data",
    save_dir: Path = PRECOMPUTE_DIR
) -> Dict:
    """
    Run the full precompute workflow and return a compact status summary.

    Args:
        cycles: List of cycle years to precompute
        crosswalk: Crosswalk dictionary
        data_path: Path to raw data files
        save_dir: Directory to save precomputed files

    Returns:
        Dictionary with success flag, validation results, and per-cycle summary.
    """
    results = precompute_all_cycles(
        cycles=cycles,
        crosswalk=crosswalk,
        data_path=data_path,
        save_dir=save_dir,
    )

    if not results:
        return {
            "success": False,
            "results": {},
            "validation": {},
            "message": "No cycles were successfully precomputed.",
        }

    validation = validate_precomputed_data(list(results.keys()), save_dir)

    try:
        create_variable_availability_index(
            cycles=list(results.keys()),
            save_dir=save_dir,
        )
    except Exception as exc:
        print(f"⚠️ Failed to create availability index: {exc}")

    all_valid = all(validation.values()) if validation else False
    return {
        "success": all_valid,
        "results": {
            cycle: {
                "record_count": result["metadata"]["record_count"],
                "available_vars": len(result["metadata"]["available_vars"]),
            }
            for cycle, result in results.items()
        },
        "validation": validation,
        "message": "Precompute completed successfully." if all_valid else "Precompute completed, but validation failed for one or more cycles.",
    }

load_variable_availability_index 

load_variable_availability_index(save_dir=PRECOMPUTE_DIR)

Load the variable availability index (which variables exist in which cycles).

Parameters:

Name Type Description Default
save_dir Path

Directory where precomputed data is stored

 PRECOMPUTE_DIR

Returns:

Type Description
Dict[str, List[str]]

Dictionary mapping variable_name -> [list of cycles where it exists]
Source code in src/data/precompute.py 
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
def load_variable_availability_index(save_dir: Path = PRECOMPUTE_DIR) -> Dict[str, List[str]]:
    """
    Load the variable availability index (which variables exist in which cycles).

    Args:
        save_dir: Directory where precomputed data is stored

    Returns:
        Dictionary mapping variable_name -> [list of cycles where it exists]
    """
    index_path = save_dir / "variable_availability.json"

    if not index_path.exists():
        raise FileNotFoundError(f"Variable availability index not found: {index_path}")

    with open(index_path, 'r') as f:
        return json.load(f)

create_variable_availability_index 

create_variable_availability_index(cycles, save_dir=PRECOMPUTE_DIR)

Create a variable availability index from precomputed metadata.

Parameters:

Name Type Description Default
cycles List[str]

List of cycles to include in index

 required
save_dir Path

Directory where precomputed data is stored

 PRECOMPUTE_DIR

Returns:

Type Description
Dict[str, List[str]]

Dictionary mapping variable_name -> [list of cycles where it exists]
Source code in src/data/precompute.py 
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
def create_variable_availability_index(
    cycles: List[str],
    save_dir: Path = PRECOMPUTE_DIR
) -> Dict[str, List[str]]:
    """
    Create a variable availability index from precomputed metadata.

    Args:
        cycles: List of cycles to include in index
        save_dir: Directory where precomputed data is stored

    Returns:
        Dictionary mapping variable_name -> [list of cycles where it exists]
    """
    availability_index = {}

    for cycle in cycles:
        try:
            meta = load_precomputed_metadata(cycle, save_dir)
            for var in meta['available_vars']:
                if var not in availability_index:
                    availability_index[var] = []
                availability_index[var].append(cycle)
        except FileNotFoundError:
            print(f"⚠️ Metadata not found for {cycle}, skipping...")
            continue

    # Save index
    index_path = save_dir / "variable_availability.json"
    with open(index_path, 'w') as f:
        json.dump(availability_index, f, indent=2)

    print(f"✅ Variable availability index created: {len(availability_index)} variables")
    print(f"   Saved to: {index_path}")

    return availability_index

validate_precomputed_data 

validate_precomputed_data(cycles, save_dir=PRECOMPUTE_DIR)

Validate that precomputed data exists and is loadable for specified cycles.

Parameters:

Name Type Description Default
cycles List[str]

List of cycles to validate

 required
save_dir Path

Directory where precomputed data is stored

 PRECOMPUTE_DIR

Returns:

Type Description
Dict[str, bool]

Dictionary mapping cycle -> True (valid) or False (invalid/missing)
Source code in src/data/precompute.py 
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
def validate_precomputed_data(cycles: List[str], save_dir: Path = PRECOMPUTE_DIR) -> Dict[str, bool]:
    """
    Validate that precomputed data exists and is loadable for specified cycles.

    Args:
        cycles: List of cycles to validate
        save_dir: Directory where precomputed data is stored

    Returns:
        Dictionary mapping cycle -> True (valid) or False (invalid/missing)
    """
    validation = {}

    for cycle in cycles:
        try:
            # Try loading data
            data = load_precomputed_data(cycle, save_dir)
            bootstrap = load_precomputed_bootstrap(cycle, save_dir)
            metadata = load_precomputed_metadata(cycle, save_dir)

            # Basic validation checks
            checks = [
                len(data) > 0,
                'CYCLE' in data.columns,
                'ONT_ID' in data.columns,
                len(bootstrap) > 0,
                'ONT_ID' in bootstrap.columns,
                'CYCLE' in bootstrap.columns,
                len(metadata.get('available_vars', [])) > 0,
                metadata.get('cycle') == cycle
            ]

            validation[cycle] = all(checks)

            if validation[cycle]:
                print(f"✅ {cycle}: Valid ({len(data):,} records, {len(metadata['available_vars'])} vars)")
            else:
                print(f"❌ {cycle}: Failed validation checks")

        except Exception as e:
            print(f"❌ {cycle}: {e}")
            validation[cycle] = False

    return validation

src.data.smart_loader

Smart data loader that uses precomputed data when available, falls back to real-time.

smart_load_cycle 

smart_load_cycle(cycle, crosswalk=None, use_precompute=True)

Smart loader that uses precomputed data when available, falls back to real-time.

Parameters:

Name Type Description Default
cycle str

Cycle name (e.g., "2021")

 required
crosswalk Dict

Crosswalk dictionary (needed for real-time harmonization)

 None
use_precompute bool

Whether to attempt using precomputed data

 True

Returns:

Type Description
Tuple[DataFrame, DataFrame, Dict, bool]

Tuple of (harmonized_data, bootstrap_data, metadata, is_precomputed)
Source code in src/data/smart_loader.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
def smart_load_cycle(
    cycle: str,
    crosswalk: Dict = None,
    use_precompute: bool = True
) -> Tuple[pd.DataFrame, pd.DataFrame, Dict, bool]:
    """
    Smart loader that uses precomputed data when available, falls back to real-time.

    Args:
        cycle: Cycle name (e.g., "2021")
        crosswalk: Crosswalk dictionary (needed for real-time harmonization)
        use_precompute: Whether to attempt using precomputed data

    Returns:
        Tuple of (harmonized_data, bootstrap_data, metadata, is_precomputed)
    """
    is_precomputed = False

    # Try precomputed first if enabled
    if use_precompute:
        try:
            status = check_precompute_status([cycle])
            if status.get(cycle, False):
                data = load_precomputed_data(cycle)
                data = restore_geography_aliases(data)
                bootstrap = load_precomputed_bootstrap(cycle)
                if 'CYCLE' not in bootstrap.columns:
                    raise ValueError(
                        f"Precomputed bootstrap data for {cycle} is missing CYCLE. "
                        "Regenerate precomputed files."
                    )
                metadata = load_precomputed_metadata(cycle)
                is_precomputed = True
                return data, bootstrap, metadata, is_precomputed
        except Exception as e:
            print(f"⚠️ Failed to load precomputed data for {cycle}: {e}")
            print("Falling back to real-time processing...")

    # Fall back to real-time processing
    if crosswalk is None:
        raise ValueError("Crosswalk required for real-time processing")

    # Import here to avoid circular dependency
    from src.data.loader import load_cycle_data

    # Load raw data
    raw_data, bootstrap_data = load_cycle_data(cycle)

    # Create harmonization mapping
    rename_dict, available_vars = build_harmonization_mapping(
        crosswalk, cycle, raw_data.columns
    )

    # Apply harmonization
    harmonized_data = raw_data.rename(columns=rename_dict)
    harmonized_data = restore_geography_aliases(harmonized_data)

    # Add CYCLE column
    harmonized_data['CYCLE'] = cycle
    bootstrap_data = bootstrap_data.copy()
    bootstrap_data['CYCLE'] = cycle

    # Create metadata
    metadata = {
        'cycle': cycle,
        'available_vars': available_vars,
        'record_count': len(harmonized_data),
        'harmonization_mapping': rename_dict,
        'realtime': True
    }

    return harmonized_data, bootstrap_data, metadata, is_precomputed

smart_load_multiple_cycles 

smart_load_multiple_cycles(cycles, crosswalk=None, use_precompute=True)

Load multiple cycles using smart loading.

Parameters:

Name Type Description Default
cycles List[str]

List of cycle years to load

 required
crosswalk Dict

Crosswalk dictionary (needed for real-time fallback)

 None
use_precompute bool

Whether to attempt using precomputed data

 True

Returns:

Type Description
Dict[str, Dict]

Dict mapping cycle -> { 'data': DataFrame, 'bootstrap': DataFrame, 'metadata': Dict, 'precomputed': bool
Dict[str, Dict]

}
Source code in src/data/smart_loader.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
def smart_load_multiple_cycles(
    cycles: List[str],
    crosswalk: Dict = None,
    use_precompute: bool = True
) -> Dict[str, Dict]:
    """
    Load multiple cycles using smart loading.

    Args:
        cycles: List of cycle years to load
        crosswalk: Crosswalk dictionary (needed for real-time fallback)
        use_precompute: Whether to attempt using precomputed data

    Returns:
        Dict mapping cycle -> {
            'data': DataFrame,
            'bootstrap': DataFrame,
            'metadata': Dict,
            'precomputed': bool
        }
    """
    results = {}

    for cycle in cycles:
        try:
            data, bootstrap, metadata, is_precomputed = smart_load_cycle(
                cycle, crosswalk, use_precompute
            )
            results[cycle] = {
                'data': data,
                'bootstrap': bootstrap,
                'metadata': metadata,
                'precomputed': is_precomputed
            }
        except Exception as e:
            print(f"❌ Error loading {cycle}: {e}")
            import traceback
            traceback.print_exc()
            continue

    return results

get_common_vars_smart 

get_common_vars_smart(cycles, crosswalk=None, use_precompute=True)

Get common variables across cycles using smart approach. Uses precomputed metadata when available for speed.

Parameters:

Name Type Description Default
cycles List[str]

List of cycle years

 required
crosswalk Dict

Crosswalk dictionary (needed for real-time fallback)

 None
use_precompute bool

Whether to attempt using precomputed data

 True

Returns:

Type Description
List[str]

Sorted list of common harmonized variable names
Source code in src/data/smart_loader.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def get_common_vars_smart(
    cycles: List[str],
    crosswalk: Dict = None,
    use_precompute: bool = True
) -> List[str]:
    """
    Get common variables across cycles using smart approach.
    Uses precomputed metadata when available for speed.

    Args:
        cycles: List of cycle years
        crosswalk: Crosswalk dictionary (needed for real-time fallback)
        use_precompute: Whether to attempt using precomputed data

    Returns:
        Sorted list of common harmonized variable names
    """
    # Check if all cycles are precomputed
    if use_precompute:
        status = check_precompute_status(cycles)
        all_precomputed = all(status.get(c, False) for c in cycles)

        if all_precomputed:
            # Fast path - use precomputed metadata
            try:
                return get_common_variables(cycles)
            except Exception as e:
                print(f"⚠️ Failed to get common vars from precomputed: {e}")

    # Slow path - load and check manually
    cycle_data = smart_load_multiple_cycles(cycles, crosswalk, use_precompute)

    if not cycle_data:
        return []

    # Get intersection of available vars
    all_vars = [set(result['metadata']['available_vars']) for result in cycle_data.values()]
    common = set.intersection(*all_vars) if all_vars else set()
    return sorted(list(common))

load_and_combine_cycles_smart 

load_and_combine_cycles_smart(cycles, crosswalk=None, use_precompute=True)

Load multiple cycles and combine into single DataFrames. Smart loader that uses precomputed data when available.

Parameters:

Name Type Description Default
cycles List[str]

List of cycle years to load and combine

 required
crosswalk Dict

Crosswalk dictionary (needed for real-time fallback)

 None
use_precompute bool

Whether to attempt using precomputed data

 True

Returns:

Type Description
Tuple[DataFrame, DataFrame]

Tuple of (combined_data, combined_bootstrap)
Source code in src/data/smart_loader.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
def load_and_combine_cycles_smart(
    cycles: List[str],
    crosswalk: Dict = None,
    use_precompute: bool = True
) -> Tuple[pd.DataFrame, pd.DataFrame]:
    """
    Load multiple cycles and combine into single DataFrames.
    Smart loader that uses precomputed data when available.

    Args:
        cycles: List of cycle years to load and combine
        crosswalk: Crosswalk dictionary (needed for real-time fallback)
        use_precompute: Whether to attempt using precomputed data

    Returns:
        Tuple of (combined_data, combined_bootstrap)
    """
    # Load all cycles
    cycle_data = smart_load_multiple_cycles(cycles, crosswalk, use_precompute)

    if not cycle_data:
        raise ValueError("No cycles were successfully loaded")

    # Combine data
    combined_data_list = []
    combined_bootstrap_list = []

    for cycle, result in cycle_data.items():
        combined_data_list.append(result['data'])
        combined_bootstrap_list.append(result['bootstrap'])

    # Concatenate
    combined_data = pd.concat(combined_data_list, ignore_index=True)
    combined_bootstrap = pd.concat(combined_bootstrap_list, ignore_index=True)

    return combined_data, combined_bootstrap

get_precompute_summary 

get_precompute_summary(cycles)

Get summary of precompute status for cycles.

Parameters:

Name Type Description Default
cycles List[str]

List of cycle years to check

 required

Returns:

Type Description
Dict

Dictionary with summary information
Source code in src/data/smart_loader.py 
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def get_precompute_summary(cycles: List[str]) -> Dict:
    """
    Get summary of precompute status for cycles.

    Args:
        cycles: List of cycle years to check

    Returns:
        Dictionary with summary information
    """
    status = check_precompute_status(cycles)

    precomputed = [c for c, s in status.items() if s]
    missing = [c for c, s in status.items() if not s]

    summary = {
        'total_cycles': len(cycles),
        'precomputed_cycles': precomputed,
        'missing_cycles': missing,
        'all_precomputed': len(missing) == 0,
        'none_precomputed': len(precomputed) == 0,
        'partial_precomputed': 0 < len(precomputed) < len(cycles)
    }

    # Try to get common variables if all precomputed
    if summary['all_precomputed']:
        try:
            common_vars = get_common_variables(cycles)
            summary['common_variables_count'] = len(common_vars)
        except Exception:
            summary['common_variables_count'] = None

    return summary

src.data.processor

Data processing functions for CCHS analysis.

create_age_groups 

create_age_groups(df, age_column=None, age_bins=None, age_labels=None)

Create age groups from the age column with flexible bin configuration. Automatically detects the correct age column for each cycle: - 2021: DHH_AGE - 2022/2023: AWCAGE - Multi-cycle harmonized: AWCAGE

Parameters:

Name Type Description Default
df

DataFrame containing age data

 required
age_column

Optional specific age column name. If None, auto-detects.

 None
age_bins

List of bin edges (e.g., [0, 15, 25, 45, 65, 120])

 None
age_labels

List of labels for bins (e.g., ['0-14', '15-24', '25-44', '45-64', '65+'])

 None

Returns:

Type Description

DataFrame with AgeGroup column added
Source code in src/data/processor.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
def create_age_groups(df, age_column=None, age_bins=None, age_labels=None):
    """
    Create age groups from the age column with flexible bin configuration.
    Automatically detects the correct age column for each cycle:
    - 2021: DHH_AGE
    - 2022/2023: AWCAGE
    - Multi-cycle harmonized: AWCAGE

    Args:
        df: DataFrame containing age data
        age_column: Optional specific age column name. If None, auto-detects.
        age_bins: List of bin edges (e.g., [0, 15, 25, 45, 65, 120])
        age_labels: List of labels for bins (e.g., ['0-14', '15-24', '25-44', '45-64', '65+'])

    Returns:
        DataFrame with AgeGroup column added
    """
    # Use default bins/labels if not provided
    if age_bins is None:
        from config.settings import DEFAULT_AGE_BINS
        age_bins = DEFAULT_AGE_BINS

    if age_labels is None:
        from config.settings import DEFAULT_AGE_LABELS
        age_labels = DEFAULT_AGE_LABELS

    # Auto-detect age column if not specified
    if age_column is None:
        if 'AWCAGE' in df.columns:
            age_column = 'AWCAGE'
        elif 'DHH_AGE' in df.columns:
            age_column = 'DHH_AGE'
        else:
            print("Warning: No age column found (DHH_AGE or AWCAGE). No age groups created.")
            result_df = df.copy()
            result_df['AgeGroup'] = None
            return result_df

    # Check if specified column exists
    if age_column not in df.columns:
        print(f"Warning: Column '{age_column}' not found. No age groups created.")
        result_df = df.copy()
        result_df['AgeGroup'] = None
        return result_df

    # Validate bins and labels
    if len(age_labels) != len(age_bins) - 1:
        st.error(f"Error: Number of labels ({len(age_labels)}) must be one less than bins ({len(age_bins)})")
        result_df = df.copy()
        result_df['AgeGroup'] = None
        return result_df

    # Create a new DataFrame instead of modifying a copy
    result_df = df.copy()

    # Create age groups using pd.cut for flexible binning
    try:
        result_df['AgeGroup'] = pd.cut(
            result_df[age_column],
            bins=age_bins,
            labels=age_labels,
            right=False,
            include_lowest=True
        )

        print(f"Age groups created using column: {age_column}")
        print(f"Bins: {age_bins}")
        print(f"Labels: {age_labels}")
    except Exception as e:
        st.error(f"Error creating age groups: {str(e)}")
        result_df['AgeGroup'] = None

    return result_df

apply_region_filter 

apply_region_filter(data, district_codes=None, health_region_codes=None)

Apply geographic filters using optional district and health region code lists.

Source code in src/data/processor.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def apply_region_filter(data, district_codes=None, health_region_codes=None):
    """
    Apply geographic filters using optional district and health region code lists.
    """
    filtered = data.copy()
    applied_filters = []

    if health_region_codes and 'GEODVHR4' in filtered.columns:
        health_region_set = {int(code) for code in health_region_codes}
        health_region_values = _normalize_geo_code_series(filtered['GEODVHR4'])
        filtered = filtered[health_region_values.isin(health_region_set)]
        applied_filters.append(
            f"GEODVHR4 in {', '.join(_format_health_region_labels(health_region_set))}"
        )

    if district_codes and 'GEODVCSD' in filtered.columns:
        district_code_set = {int(code) for code in district_codes.keys()}
        district_values = _normalize_geo_code_series(filtered['GEODVCSD'])
        filtered = filtered[district_values.isin(district_code_set)]
        applied_filters.append(
            f"GEODVCSD in {', '.join(_format_district_labels(district_code_set))}"
        )

    if applied_filters:
        st.write(f"Applied geographic filters: {', '.join(applied_filters)}")
    else:
        st.write("No region filter applied; using the entire dataset.")

    return filtered

apply_inclusion_flag_filters 

apply_inclusion_flag_filters(data, selected_flags)

Apply inclusion flag filters to the dataset. Works for all cycles (2021, 2022, 2023).

Parameters:

Name Type Description Default
data DataFrame

DataFrame to filter

 required
selected_flags dict

Dictionary mapping flag_name -> True/False

 required

Returns:

Type Description
DataFrame

Filtered DataFrame
Source code in src/data/processor.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def apply_inclusion_flag_filters(data: pd.DataFrame, selected_flags: dict) -> pd.DataFrame:
    """
    Apply inclusion flag filters to the dataset.
    Works for all cycles (2021, 2022, 2023).

    Args:
        data: DataFrame to filter
        selected_flags: Dictionary mapping flag_name -> True/False

    Returns:
        Filtered DataFrame
    """
    filtered = data.copy()
    applied_filters = []

    # Apply each selected inclusion flag filter
    for flag, should_filter in selected_flags.items():
        if should_filter and flag in filtered.columns:
            # Filter to only include rows where flag == 1
            initial_count = len(filtered)
            filtered = filtered[filtered[flag] == 1]
            final_count = len(filtered)
            applied_filters.append(f"{flag} ({initial_count:,} → {final_count:,} records)")

    if applied_filters:
        st.write(f"Applied inclusion flag filters: {', '.join(applied_filters)}")
    else:
        st.write("No inclusion flag filters applied")

    return filtered

Bootstrap analysis and quality

src.analysis.bootstrap

Bootstrap analysis functions for CCHS data.

run_bootstrap_analysis_for_all_values 

run_bootstrap_analysis_for_all_values(merged_data, variable_col, weight_col=DEFAULT_WEIGHT_COLUMN, standards_cycle=None)

Perform bootstrap analysis using vectorized groupby operations. Computes weighted prevalences and bootstrap variances.

Source code in src/analysis/bootstrap.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def run_bootstrap_analysis_for_all_values(
    merged_data,
    variable_col,
    weight_col=DEFAULT_WEIGHT_COLUMN,
    standards_cycle=None,
):
    """
    Perform bootstrap analysis using vectorized groupby operations.
    Computes weighted prevalences and bootstrap variances.
    """
    # Identify bootstrap weight columns (those starting with 'BSW')
    bootstrap_cols = [col for col in merged_data.columns if col.startswith(BOOTSTRAP_PREFIX)]

    # Precompute total weights for base and bootstrap replicates
    total_weight_base = merged_data[weight_col].sum()
    total_weights_boot = {col: merged_data[col].sum() for col in bootstrap_cols}

    # Compute weighted sums for the base weight grouped by the selected variable
    base_numerators = merged_data.groupby(variable_col)[weight_col].sum()
    base_prevalence = (base_numerators / total_weight_base) * 100
    unweighted_numerators = merged_data.groupby(variable_col).size()
    unweighted_denominator = merged_data[variable_col].notna().sum()

    # Weighted population for each group (sum of weights)
    weighted_population = base_numerators

    # OPTIMIZED: Compute all bootstrap replicates at once using vectorized operations
    # Group by variable and sum all bootstrap columns simultaneously
    bootstrap_sums = merged_data.groupby(variable_col)[bootstrap_cols].sum()

    # Convert total weights to Series for vectorized division
    total_weights_series = pd.Series(total_weights_boot, name='total_weights')

    # Vectorized calculation: divide each bootstrap sum by its corresponding total weight
    replicate_prevalence_df = bootstrap_sums.div(total_weights_series, axis=1) * 100

    # Compute variance, standard deviation, confidence intervals, etc.
    variance = ((replicate_prevalence_df.sub(base_prevalence, axis=0))**2).mean(axis=1)
    std_dev = np.sqrt(variance)
    ci_lower = base_prevalence - CONFIDENCE_Z * std_dev
    ci_upper = base_prevalence + CONFIDENCE_Z * std_dev
    cv = (std_dev / base_prevalence) * 100

    result_df = pd.DataFrame({
        'Value': base_prevalence.index,
        'Prevalence': base_prevalence.values,
        'Unweighted Numerator': unweighted_numerators.reindex(base_prevalence.index).values,
        'Unweighted Denominator': unweighted_denominator,
        'Weighted Population': weighted_population.values,
        'Variance': variance.values,
        'Standard Deviation': std_dev.values,
        'CI Lower': ci_lower.values,
        'CI Upper': ci_upper.values,
        'CV (%)': cv.values,
        'Error': CONFIDENCE_Z * std_dev.values  # for error bars in plots
    }).reset_index(drop=True)

    if str(standards_cycle) in QUALITY_FLAG_CYCLES:
        result_df = apply_cchs_quality_flags(result_df)

    return result_df

src.analysis.quality

Helpers for applying CCHS 2022+ data quality reporting standards.

calculate_effective_sample_size 

calculate_effective_sample_size(prevalence_pct, cv_pct)

Calculate effective sample size for a proportion estimate.

Formula from the CCHS 2022+ standards: (1 - p) / (p * CV^2) where p and CV are expressed as proportions, not percentages.

Source code in src/analysis/quality.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def calculate_effective_sample_size(prevalence_pct: float, cv_pct: float) -> Optional[float]:
    """
    Calculate effective sample size for a proportion estimate.

    Formula from the CCHS 2022+ standards:
    (1 - p) / (p * CV^2)
    where p and CV are expressed as proportions, not percentages.
    """
    if pd.isna(prevalence_pct) or pd.isna(cv_pct):
        return None

    p_hat = prevalence_pct / 100.0
    cv = cv_pct / 100.0

    if p_hat <= 0 or p_hat >= 1:
        return None

    if cv < 0:
        return None

    if math.isclose(cv, 0.0):
        return math.inf

    denominator = p_hat * (cv ** 2)
    if math.isclose(denominator, 0.0):
        return math.inf

    return (1 - p_hat) / denominator

classify_proportion_release 

classify_proportion_release(prevalence_pct, cv_pct, numerator_n, denominator_n, ci_lower=None, ci_upper=None)

Classify a proportion estimate using the CCHS 2022+ A/E/F rules.

Source code in src/analysis/quality.py 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def classify_proportion_release(
    prevalence_pct: float,
    cv_pct: float,
    numerator_n: Optional[float],
    denominator_n: Optional[float],
    ci_lower: Optional[float] = None,
    ci_upper: Optional[float] = None,
) -> dict:
    """
    Classify a proportion estimate using the CCHS 2022+ A/E/F rules.
    """
    effective_n = calculate_effective_sample_size(prevalence_pct, cv_pct)

    result = {
        "Release Category": "F",
        "Release Action": "Suppress",
        "Effective Sample Size": effective_n,
        "Release Reason": "Insufficient information to classify",
    }

    if pd.isna(prevalence_pct) or pd.isna(cv_pct):
        result["Release Reason"] = "Missing prevalence or coefficient of variation"
        return result

    if prevalence_pct <= 0 or prevalence_pct >= 100:
        result["Release Reason"] = "Estimates of 0% or 100% should never be released"
        return result

    if ci_lower is not None and ci_upper is not None:
        if not pd.isna(ci_lower) and not pd.isna(ci_upper):
            if math.isclose(ci_lower, ci_upper):
                result["Release Reason"] = "Confidence interval has zero length"
                return result
            if ci_lower < 0 or ci_upper > 100:
                result["Release Reason"] = "Confidence interval bounds are implausible"
                return result

    n1 = float(numerator_n) if numerator_n is not None and not pd.isna(numerator_n) else None
    n2 = float(denominator_n) if denominator_n is not None and not pd.isna(denominator_n) else None

    if n1 is None or n2 is None or effective_n is None:
        result["Release Reason"] = "Missing numerator, denominator, or effective sample size"
        return result

    if n2 < 50:
        result["Release Reason"] = "Denominator unweighted count is below 50"
        return result

    if n1 < 10:
        result["Release Reason"] = "Numerator unweighted count is below 10"
        return result

    if effective_n < 30:
        result["Release Reason"] = "Effective sample size is below 30"
        return result

    if n2 >= 100 and effective_n >= 60:
        result["Release Category"] = "A"
        result["Release Action"] = "Release with no warning"
        result["Release Reason"] = "Denominator >= 100 and effective sample size >= 60"
        return result

    result["Release Category"] = "E"
    result["Release Action"] = "Release with caution warning"
    result["Release Reason"] = "Releasable, but does not meet the no-warning threshold"
    return result

apply_cchs_quality_flags 

apply_cchs_quality_flags(result_df)

Append CCHS 2022+ release-quality fields to a result dataframe.

Source code in src/analysis/quality.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def apply_cchs_quality_flags(result_df: pd.DataFrame) -> pd.DataFrame:
    """Append CCHS 2022+ release-quality fields to a result dataframe."""
    if result_df.empty:
        return result_df

    classified = result_df.apply(
        lambda row: classify_proportion_release(
            prevalence_pct=row.get("Prevalence"),
            cv_pct=row.get("CV (%)"),
            numerator_n=row.get("Unweighted Numerator"),
            denominator_n=row.get("Unweighted Denominator"),
            ci_lower=row.get("CI Lower"),
            ci_upper=row.get("CI Upper"),
        ),
        axis=1,
        result_type="expand",
    )

    return pd.concat([result_df, classified], axis=1)

src.analysis.comparison

Comparative analysis functions for multi-cycle CCHS data.

compare_cycles 

compare_cycles(results_df)

Pivot results to show cycles side-by-side for comparison.

Parameters:

Name Type Description Default
results_df DataFrame

DataFrame with columns including 'Variable', 'Value', 'CYCLE', 'Prevalence', etc.

 required

Returns:

Type Description
DataFrame

Pivoted DataFrame with cycles as columns
Source code in src/analysis/comparison.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def compare_cycles(results_df: pd.DataFrame) -> pd.DataFrame:
    """
    Pivot results to show cycles side-by-side for comparison.

    Args:
        results_df: DataFrame with columns including 'Variable', 'Value', 'CYCLE', 'Prevalence', etc.

    Returns:
        Pivoted DataFrame with cycles as columns
    """
    if 'CYCLE' not in results_df.columns:
        return results_df

    pivot_df = results_df.pivot_table(
        index=['Variable', 'Value'],
        columns='CYCLE',
        values='Prevalence',
        aggfunc='first'
    ).reset_index()

    return pivot_df

calculate_change 

calculate_change(cycle1_val, cycle2_val)

Calculate percentage point change between two cycles.

Parameters:

Name Type Description Default
cycle1_val float

Prevalence value from first cycle

 required
cycle2_val float

Prevalence value from second cycle

 required

Returns:

Type Description
float

Percentage point change (cycle2 - cycle1)
Source code in src/analysis/comparison.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def calculate_change(cycle1_val: float, cycle2_val: float) -> float:
    """
    Calculate percentage point change between two cycles.

    Args:
        cycle1_val: Prevalence value from first cycle
        cycle2_val: Prevalence value from second cycle

    Returns:
        Percentage point change (cycle2 - cycle1)
    """
    if pd.isna(cycle1_val) or pd.isna(cycle2_val):
        return np.nan
    return cycle2_val - cycle1_val

calculate_percent_change 

calculate_percent_change(cycle1_val, cycle2_val)

Calculate percentage change between two cycles.

Parameters:

Name Type Description Default
cycle1_val float

Prevalence value from first cycle

 required
cycle2_val float

Prevalence value from second cycle

 required

Returns:

Type Description
float

Percentage change ((cycle2 - cycle1) / cycle1 * 100)
Source code in src/analysis/comparison.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
def calculate_percent_change(cycle1_val: float, cycle2_val: float) -> float:
    """
    Calculate percentage change between two cycles.

    Args:
        cycle1_val: Prevalence value from first cycle
        cycle2_val: Prevalence value from second cycle

    Returns:
        Percentage change ((cycle2 - cycle1) / cycle1 * 100)
    """
    if pd.isna(cycle1_val) or pd.isna(cycle2_val) or cycle1_val == 0:
        return np.nan
    return ((cycle2_val - cycle1_val) / cycle1_val) * 100

calculate_trend 

calculate_trend(results_df)

Calculate trend direction (increasing/decreasing/stable) across cycles.

Parameters:

Name Type Description Default
results_df DataFrame

DataFrame with 'CYCLE', 'Variable', 'Value', 'Prevalence' columns

 required

Returns:

Type Description
DataFrame

DataFrame with added 'Trend' column indicating direction
Source code in src/analysis/comparison.py 
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def calculate_trend(results_df: pd.DataFrame) -> pd.DataFrame:
    """
    Calculate trend direction (increasing/decreasing/stable) across cycles.

    Args:
        results_df: DataFrame with 'CYCLE', 'Variable', 'Value', 'Prevalence' columns

    Returns:
        DataFrame with added 'Trend' column indicating direction
    """
    if 'CYCLE' not in results_df.columns:
        return results_df

    result_df = results_df.copy()
    result_df['Trend'] = None

    for (variable, value), group in result_df.groupby(['Variable', 'Value']):
        if len(group) < 2:
            continue

        sorted_group = group.sort_values('CYCLE')
        prevalences = sorted_group['Prevalence'].values

        if len(prevalences) == 2:
            if prevalences[1] > prevalences[0] * 1.05:
                trend = 'Increasing'
            elif prevalences[1] < prevalences[0] * 0.95:
                trend = 'Decreasing'
            else:
                trend = 'Stable'
        else:
            slope = np.polyfit(range(len(prevalences)), prevalences, 1)[0]
            if slope > 0.1:
                trend = 'Increasing'
            elif slope < -0.1:
                trend = 'Decreasing'
            else:
                trend = 'Stable'

        result_df.loc[group.index, 'Trend'] = trend

    return result_df

test_significance 

test_significance(cycle1_results, cycle2_results, variable, value=None)

Test statistical significance of difference between two cycles.

Uses overlapping confidence intervals as a simple test. More sophisticated tests could be added later.

Parameters:

Name Type Description Default
cycle1_results DataFrame

Results DataFrame for first cycle

 required
cycle2_results DataFrame

Results DataFrame for second cycle

 required
variable str

Variable name to test

 required
value Optional[str]

Optional specific value to test

 None

Returns:

Type Description
dict

Dictionary with test results including 'significant' boolean
Source code in src/analysis/comparison.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def test_significance(cycle1_results: pd.DataFrame, cycle2_results: pd.DataFrame, 
                      variable: str, value: Optional[str] = None) -> dict:
    """
    Test statistical significance of difference between two cycles.

    Uses overlapping confidence intervals as a simple test.
    More sophisticated tests could be added later.

    Args:
        cycle1_results: Results DataFrame for first cycle
        cycle2_results: Results DataFrame for second cycle
        variable: Variable name to test
        value: Optional specific value to test

    Returns:
        Dictionary with test results including 'significant' boolean
    """
    filter1 = cycle1_results['Variable'] == variable
    filter2 = cycle2_results['Variable'] == variable

    if value is not None:
        filter1 = filter1 & (cycle1_results['Value'] == value)
        filter2 = filter2 & (cycle2_results['Value'] == value)

    result1 = cycle1_results[filter1].iloc[0] if len(cycle1_results[filter1]) > 0 else None
    result2 = cycle2_results[filter2].iloc[0] if len(cycle2_results[filter2]) > 0 else None

    if result1 is None or result2 is None:
        return {'significant': False, 'reason': 'Missing data'}

    ci1_lower = result1.get('CI Lower', np.nan)
    ci1_upper = result1.get('CI Upper', np.nan)
    ci2_lower = result2.get('CI Lower', np.nan)
    ci2_upper = result2.get('CI Upper', np.nan)

    if pd.isna(ci1_lower) or pd.isna(ci1_upper) or pd.isna(ci2_lower) or pd.isna(ci2_upper):
        return {'significant': False, 'reason': 'Missing confidence intervals'}

    overlap = not (ci1_upper < ci2_lower or ci2_upper < ci1_lower)

    return {
        'significant': not overlap,
        'overlap': overlap,
        'cycle1_ci': (ci1_lower, ci1_upper),
        'cycle2_ci': (ci2_lower, ci2_upper),
        'cycle1_prevalence': result1.get('Prevalence', np.nan),
        'cycle2_prevalence': result2.get('Prevalence', np.nan)
    }

create_comparison_summary 

create_comparison_summary(results_df)

Create a summary table comparing cycles.

Parameters:

Name Type Description Default
results_df DataFrame

DataFrame with cycle comparison results

 required

Returns:

Type Description
DataFrame

Summary DataFrame with comparison statistics
Source code in src/analysis/comparison.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
def create_comparison_summary(results_df: pd.DataFrame) -> pd.DataFrame:
    """
    Create a summary table comparing cycles.

    Args:
        results_df: DataFrame with cycle comparison results

    Returns:
        Summary DataFrame with comparison statistics
    """
    if 'CYCLE' not in results_df.columns:
        return pd.DataFrame()

    summary_data = []

    for (variable, value), group in results_df.groupby(['Variable', 'Value']):
        if len(group) < 2:
            continue

        sorted_group = group.sort_values('CYCLE')
        cycles = sorted_group['CYCLE'].tolist()
        prevalences = sorted_group['Prevalence'].tolist()

        first_cycle = cycles[0]
        last_cycle = cycles[-1]
        first_prev = prevalences[0]
        last_prev = prevalences[-1]

        change_pp = calculate_change(first_prev, last_prev)
        change_pct = calculate_percent_change(first_prev, last_prev)

        trend_df = calculate_trend(group)
        trend = trend_df['Trend'].iloc[0] if 'Trend' in trend_df.columns else None

        summary_data.append({
            'Variable': variable,
            'Value': value,
            'First Cycle': first_cycle,
            'Last Cycle': last_cycle,
            'First Prevalence': first_prev,
            'Last Prevalence': last_prev,
            'Change (pp)': change_pp,
            'Change (%)': change_pct,
            'Trend': trend
        })

    return pd.DataFrame(summary_data)

Helpers

src.utils.helpers

Utility helper functions for the CCHS application.

format_number 

format_number(num, format_type='comma')

Format numbers with different styles.

Source code in src/utils/helpers.py 
 8
 9
10
11
12
13
14
15
16
17
def format_number(num, format_type="comma"):
    """Format numbers with different styles."""
    if format_type == "comma":
        return f"{num:,}"
    elif format_type == "percentage":
        return f"{num:.2f}%"
    elif format_type == "decimal":
        return f"{num:.3f}"
    else:
        return str(num)

create_excel_download 

create_excel_download(data, sheet_name='Analysis Results')

Create Excel file in memory for download.

Source code in src/utils/helpers.py 
20
21
22
23
24
25
def create_excel_download(data: pd.DataFrame, sheet_name: str = "Analysis Results") -> bytes:
    """Create Excel file in memory for download."""
    excel_buffer = BytesIO()
    with pd.ExcelWriter(excel_buffer, engine='openpyxl') as writer:
        data.to_excel(writer, sheet_name=sheet_name, index=False)
    return excel_buffer.getvalue()

validate_data_columns 

validate_data_columns(data, required_columns)

Validate that required columns exist in the dataset.

Source code in src/utils/helpers.py 
28
29
30
31
32
33
34
def validate_data_columns(data: pd.DataFrame, required_columns: list) -> bool:
    """Validate that required columns exist in the dataset."""
    missing_columns = [col for col in required_columns if col not in data.columns]
    if missing_columns:
        st.error(f"Missing required columns: {', '.join(missing_columns)}")
        return False
    return True

safe_division 

safe_division(numerator, denominator, default=0)

Safely divide two numbers, returning default if denominator is zero.

Source code in src/utils/helpers.py 
37
38
39
40
41
42
def safe_division(numerator, denominator, default=0):
    """Safely divide two numbers, returning default if denominator is zero."""
    try:
        return numerator / denominator if denominator != 0 else default
    except (TypeError, ZeroDivisionError):
        return default

filter_dataframe_by_values 

filter_dataframe_by_values(df, column, values)

Filter dataframe by specific values in a column.

Source code in src/utils/helpers.py 
45
46
47
48
49
def filter_dataframe_by_values(df, column, values):
    """Filter dataframe by specific values in a column."""
    if column in df.columns and values:
        return df[df[column].isin(values)]
    return df

get_memory_usage_mb 

get_memory_usage_mb(df)

Get memory usage of dataframe in MB.

Source code in src/utils/helpers.py 
52
53
54
def get_memory_usage_mb(df):
    """Get memory usage of dataframe in MB."""
    return df.memory_usage(deep=True).sum() / 1024**2

get_cycle_varname 

get_cycle_varname(harmonized_var, cycle, crosswalk)

Helper to get cycle-specific variable name from harmonization crosswalk.

Source code in src/utils/helpers.py 
57
58
59
60
def get_cycle_varname(harmonized_var: str, cycle: str, crosswalk: dict) -> str:
    """Helper to get cycle-specific variable name from harmonization crosswalk."""
    mapping = crosswalk.get(harmonized_var, {})
    return mapping.get(cycle, harmonized_var)

get_value_label 

get_value_label(harmonized_var, value, cycle, categories)

Helper to get value label for cycle from harmonization categories.

Source code in src/utils/helpers.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def get_value_label(harmonized_var: str, value, cycle: str, categories: dict) -> str:
    """Helper to get value label for cycle from harmonization categories."""
    cat = categories.get(harmonized_var, {})
    mappings = cat.get("mappings", {})
    year_map = mappings.get(str(cycle), {})

    # Convert value to string, but if it's a float and is_integer, cast to int first
    if isinstance(value, float) and value.is_integer():
        value_str = str(int(value))
    else:
        value_str = str(value)

    label = year_map.get(value_str, None)
    if label is None:
        return value_str
    return label

get_cycle_value_label 

get_cycle_value_label(varname, value, cycle_var_info)

Get value label from cycle-specific JSON (CCHS_YYYY.json). Used for single-cycle analysis to show proper category labels.

Parameters:

Name Type Description Default
varname str

Variable name (cycle-specific, not harmonized)

 required
value

The value to get label for

 required
cycle_var_info dict

Full cycle variable info from CCHS_YYYY.json

 required

Returns:

Type Description
str

Label string or original value if not found
Source code in src/utils/helpers.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def get_cycle_value_label(varname: str, value, cycle_var_info: dict) -> str:
    """
    Get value label from cycle-specific JSON (CCHS_YYYY.json).
    Used for single-cycle analysis to show proper category labels.

    Args:
        varname: Variable name (cycle-specific, not harmonized)
        value: The value to get label for
        cycle_var_info: Full cycle variable info from CCHS_YYYY.json

    Returns:
        Label string or original value if not found
    """
    var_info = cycle_var_info.get(varname, {})
    categories = var_info.get("categories", {})

    # Convert value to string (handle floats that are actually integers)
    if isinstance(value, float) and value.is_integer():
        value_str = str(int(value))
    else:
        value_str = str(value)

    # Return label if found, otherwise return the value as string
    return categories.get(value_str, value_str)

get_available_harmonized_vars 

get_available_harmonized_vars(crosswalk, cycle, merged_data)

Helper to get available harmonized variables for the selected cycle and data.

Source code in src/utils/helpers.py 
107
108
109
110
111
112
113
114
def get_available_harmonized_vars(crosswalk: dict, cycle: str, merged_data: pd.DataFrame) -> list:
    """Helper to get available harmonized variables for the selected cycle and data."""
    available = []
    for harmonized_var, mapping in crosswalk.items():
        varname = mapping.get(cycle)
        if varname and varname in merged_data.columns:
            available.append(harmonized_var)
    return available

merge_descriptions 

merge_descriptions(json_desc_dict, csv_desc_dict)

Merge CSV and JSON descriptions, preferring CSV if present.

Source code in src/utils/helpers.py 
117
118
119
def merge_descriptions(json_desc_dict: dict, csv_desc_dict: dict) -> dict:
    """Merge CSV and JSON descriptions, preferring CSV if present."""
    return {**json_desc_dict, **csv_desc_dict}

create_multi_cycle_excel 

create_multi_cycle_excel(results_df, cycles)

Create Excel file with multiple sheets for multi-cycle results.

Parameters:

Name Type Description Default
results_df DataFrame

DataFrame with multi-cycle results (must contain 'CYCLE' column)

 required
cycles list

List of cycles included in the results

 required

Returns:

Type Description
bytes

Excel file as bytes
Source code in src/utils/helpers.py 
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
def create_multi_cycle_excel(results_df: pd.DataFrame, cycles: list) -> bytes:
    """
    Create Excel file with multiple sheets for multi-cycle results.

    Args:
        results_df: DataFrame with multi-cycle results (must contain 'CYCLE' column)
        cycles: List of cycles included in the results

    Returns:
        Excel file as bytes
    """
    excel_buffer = BytesIO()

    with pd.ExcelWriter(excel_buffer, engine='openpyxl') as writer:
        results_df.to_excel(writer, sheet_name="Combined Results", index=False)

        for cycle in cycles:
            cycle_data = results_df[results_df['CYCLE'] == cycle].copy()
            if not cycle_data.empty:
                cycle_data = cycle_data.drop(columns=['CYCLE'])
                cycle_data.to_excel(writer, sheet_name=f"Cycle {cycle}", index=False)

        from src.analysis.comparison import create_comparison_summary
        summary_df = create_comparison_summary(results_df)
        if not summary_df.empty:
            summary_df.to_excel(writer, sheet_name="Comparison Summary", index=False)

    return excel_buffer.getvalue()

get_inclusion_flags 

get_inclusion_flags(data, desc_dict)

Detect inclusion flag variables in the dataset. Inclusion flags are identified by having "Inclusion Flag" in their description. Works for all cycles (2021, 2022, 2023).

Parameters:

Name Type Description Default
data DataFrame

DataFrame with columns

 required
desc_dict dict

Dictionary mapping variable names to descriptions (required)

 required

Returns:

Type Description
dict

Dictionary mapping flag_name -> description
Source code in src/utils/helpers.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def get_inclusion_flags(data: pd.DataFrame, desc_dict: dict) -> dict:
    """
    Detect inclusion flag variables in the dataset.
    Inclusion flags are identified by having "Inclusion Flag" in their description.
    Works for all cycles (2021, 2022, 2023).

    Args:
        data: DataFrame with columns
        desc_dict: Dictionary mapping variable names to descriptions (required)

    Returns:
        Dictionary mapping flag_name -> description
    """
    inclusion_flags = {}

    if desc_dict is None or not desc_dict:
        return inclusion_flags

    # Check all variables in the dataset
    for col in data.columns:
        # Skip bootstrap weights and other non-variable columns
        if col.startswith('BSW') or col in ['ONT_ID', 'CYCLE', 'WTS_S', 'AgeGroup']:
            continue

        # Check if description contains "Inclusion Flag"
        if col in desc_dict:
            desc = desc_dict[col]
            if 'Inclusion Flag' in desc or 'inclusion flag' in desc.lower():
                inclusion_flags[col] = desc

    return inclusion_flags