`panel_tools`

Panel data loading and processing utilities for Databricks DAME project.

This module provides utility functions for loading, standardizing, and combining raw panel data from multiple vendors, as well as helper functions for partition discovery and log enrichment.

Requires: - pyspark - dbutils (Databricks utilities) - spark (Databricks SparkSession) - access to configured S3-backed data sources

Author: Gino F. Fazzi gino.franco.fazzi@audienceproject.com

`panel_tools.load_yougov_raw_files(yyyy=None, mm=None, rename_cols=True)`

Load raw YouGov panel files for a given month.

Loads the latest available YouGov raw files, or the files matching the provided year and month, derives country information from file names, optionally standardizes column names, and excludes opt-out records.

Parameters:

Name	Type	Description	Default
`yyyy`	`int or str`	Target year to load. If not provided, the latest available year is inferred from the source files.	`None`
`mm`	`int or str`	Target month to load. If not provided, the latest available month is inferred from the source files.	`None`
`rename_cols`	`bool`	Whether to rename vendor-specific columns to the shared schema. Defaults to True.	`True`

Returns:

Type	Description
	pyspark.sql.DataFrame: Deduplicated YouGov panel data.

Raises:

Type	Description
`IndexError`	If no matching files are found for the requested period.
`Exception`	If the source files or opt-out files cannot be read.

Note

Reads from s3://ext-yougov-0/
Excludes files containing optouts from the primary panel load
Derives country from the source filename
Removes opt-out records using a left anti join

Example

df = load_yougov_raw_files(yyyy=2025, mm=8, rename_cols=True)

`panel_tools.load_norstat_raw_files(yyyy=None, mm=None, load_only_latest=False, rename_cols=False)`

Load raw Norstat panel files for a given month or latest available day.

Reads Norstat panelist files and their associated deletion files, removes deleted panelists, and optionally standardizes columns to the shared schema.

Parameters:

Name	Type	Description	Default
`yyyy`	`int or str`	Target year to load. Defaults to the current year when not provided.	`None`
`mm`	`int or str`	Target month to load. Defaults to the current month when not provided.	`None`
`load_only_latest`	`bool`	Whether to search backwards for the latest available daily snapshot within the requested month. Defaults to False.	`False`
`rename_cols`	`bool`	Whether to rename vendor-specific columns to the shared schema. Defaults to False.	`False`

Returns:

Type	Description
	pyspark.sql.DataFrame: Deduplicated Norstat panel data with deletions
	removed.

Raises:

Type	Description
`Exception`	If the source files cannot be loaded.

Note

Reads from s3://ext-norstat-0/
Removes deleted panelists using data_deletion.csv
Converts year_of_birth to age when rename_cols=True
When load_only_latest=True, searches backwards from the target date

Example

df = load_norstat_raw_files( yyyy=2025, mm=8, load_only_latest=True, rename_cols=True )

`panel_tools.load_cint_raw_files(yyyy=None, mm=None, load_only_latest=False, rename_cols=False, rename_countries=True, get_vendor_id=False)`

Load raw Cint panel files for a given month or latest available day.

Reads both Lucid respondent files and Cobra panelist files, unions them into a single DataFrame, optionally standardizes columns, normalizes country values, and can preserve vendor source identifiers.

Parameters:

Name	Type	Description	Default
`yyyy`	`int or str`	Target year to load. Defaults to the current year when not provided.	`None`
`mm`	`int or str`	Target month to load. Defaults to the current month when not provided.	`None`
`load_only_latest`	`bool`	Whether to search backwards for the latest available daily snapshot within the requested month. Defaults to False.	`False`
`rename_cols`	`bool`	Whether to rename vendor-specific columns to the shared schema. Defaults to False.	`False`
`rename_countries`	`bool`	Whether to normalize country values to two-letter country codes. Defaults to True.	`True`
`get_vendor_id`	`bool`	Whether to add a `vendor_key` column before unioning the source files. Defaults to False.	`False`

Returns:

Type	Description
	pyspark.sql.DataFrame: Deduplicated Cint panel data.

Raises:

Type	Description
`Exception`	If the source files cannot be loaded.

Note

Reads from s3://cint-panel-ingestion/panel-data/
Combines Lucid and Cobra source files
Explodes pipe-delimited email and device ID values into separate rows
Normalizes selected three-letter country codes to two-letter codes

Example

df = load_cint_raw_files( yyyy=2025, mm=8, load_only_latest=True, rename_cols=True )

`panel_tools.find_uids_in_logs(start_date, uri_stem='/sync/put/yougovredir', vendor_key='yougovid', retrieve=['uid'])`

Find vendor panelist identifiers and selected fields in sync logs.

Filters Aqueduct logs for matching sync requests after the specified start date, extracts the vendor panelist identifier from the query string, and returns distinct records with country and selected output columns.

Parameters:

Name	Type	Description	Default
`start_date`	`str`	Start date in "YYYY-MM-DD" format. Only logs with log_date >= start_date will be included.	required
`uri_stem`	`str`	Request path to filter on. Defaults to "/sync/put/yougovredir".	`'/sync/put/yougovredir'`
`vendor_key`	`str`	Query-string parameter name containing the vendor panelist ID. Defaults to "yougovid".	`'yougovid'`
`retrieve`	`list[str]`	Additional columns to return from the log table. Defaults to ["uid"].	`['uid']`

Returns:

Type	Description
	pyspark.sql.DataFrame: Distinct log records with `country`,
	`vendor_panelist_id`, and the requested columns.

Note

Reads from main.aqueduct.pure_logs
Filters out rows with null request_query
Extracts vendor_panelist_id via regular expression

Example

df = find_uids_in_logs( start_date="2025-08-01", vendor_key="yougovid", retrieve=["uid", "request_id"] )

`panel_tools.calculate_number_missing_bg_vars(df, bg_vars_cols)`

Count missing background-variable values per row.

Adds a new column named n_missing_vars containing the number of null values across the specified background-variable columns.

Parameters:

Name	Type	Description	Default
`df`	`DataFrame`	Input DataFrame to enrich.	required
`bg_vars_cols`	`list[str]`	Columns to evaluate for missing values.	required

Returns:

Type	Description
	pyspark.sql.DataFrame: The input DataFrame with an added
	`n_missing_vars` column.

Example

df = calculate_number_missing_bg_vars( df, ["gender", "age", "income", "education"] )

`panel_tools.load_vendors_raw_files(yyyy=None, mm=None, drop_extra_cols=True, load_latest=True, get_vendor_id=False)`

Load and combine raw panel files across supported vendors.

Orchestrates the loading of Cint, Norstat, and YouGov raw files, optionally trims vendor-specific extra columns, and returns a unified DataFrame with vendor metadata attached.

Parameters:

Name	Type	Description	Default
`yyyy`	`int or str`	Target year to load. Passed through to the vendor-specific loaders.	`None`
`mm`	`int or str`	Target month to load. Passed through to the vendor-specific loaders.	`None`
`drop_extra_cols`	`bool`	Whether to drop selected vendor- specific columns that are not part of the common working schema. Defaults to True.	`True`
`load_latest`	`bool`	Whether vendor loaders that support daily backtracking should load only the latest available snapshot. Defaults to True.	`True`
`get_vendor_id`	`bool`	Whether to add a `vendor_key` column for vendor source tracking where supported. Defaults to False.	`False`

Returns:

Type	Description
	pyspark.sql.DataFrame: Unified vendor panel data including `vendor` and
	`vendor_panelist_id`.

Note

Loads from Cint, Norstat, and YouGov sources
Renames columns to the shared schema before unioning
Prints reminders about known source limitations

Example

df = load_vendors_raw_files( yyyy=2025, mm=8, drop_extra_cols=True, load_latest=True )

`panel_tools.find_vendor_latest_partition_date(vendor, cutoff_date=None, base=BASE)`

Find the latest available partition date for a vendor.

Parameters:

Name	Type	Description	Default
`vendor`	`str`	Vendor key to inspect.	required
`cutoff_date`	`str \| None`	Inclusive cutoff date in `YYYY-MM-DD` format. When provided, the returned partition date is the latest available date on or before the cutoff. Defaults to None.	`None`
`base`	`str`	Base path containing `vendor_key=...` partitions. Defaults to `BASE`.	`BASE`

Returns:

Name	Type	Description
`datetime`	`datetime`	Latest matching partition date.

Raises:

Type	Description
`ValueError`	If no matching partition date is found.

`panel_tools.load_ap_prod_panel_latest_per_vendor(base=BASE, vendor_keys=None, date=None, add_partition_cols=True, explode_uids=False, explode_mobile_ids=False, verbose=False)`

Load AP production panel partitions for each vendor.

Selects each vendor's latest available partition, optionally constrained by an inclusive date cutoff, then reads only that vendor/date partition.

Parameters:

Name	Type	Description	Default
`base`	`str`	Base partitioned path to inspect. Defaults to `BASE`.	`BASE`
`vendor_keys`	`list[str] \| None`	Vendor keys to load. If not provided, all vendor partitions under `base` are discovered.	`None`
`date`	`str \| None`	Inclusive partition cutoff in `YYYY-MM-DD` format. If provided, each vendor's latest partition on or before this date is loaded.	`None`
`add_partition_cols`	`bool`	Whether to append `vendor_key`, `yyyy`, `mm`, and `dd` columns. Defaults to True.	`True`
`explode_uids`	`bool`	Whether to explode the `uids` column into separate rows. Defaults to False.	`False`
`explode_mobile_ids`	`bool`	Whether to explode the `mobile_ids` column into separate rows. Defaults to False.	`False`
`verbose`	`bool`	Whether to print discovery and load progress. Defaults to False.	`False`

Returns:

Type	Description
	pyspark.sql.DataFrame: Unified AP production panel data.

Raises:

Type	Description
`Exception`	If no partitions are found under `base`, or if an explode was requested for a column the panel data does not contain.

Note

Expects partitioned vendor paths under base
Lowercases the vendor column before returning
Prints informational messages when array columns are not exploded

Example

df = load_ap_prod_panel_latest_per_vendor( vendor_keys=["cint", "norstat"], explode_uids=True )

`panel_tools.mbm_model_preprocessing(df)`

Prepare age and missing-variable features for the MBM model.

Parameters:

Name	Type	Description	Default
`df`		Spark DataFrame containing `age` and `n_missing_vars` columns.	required

Returns:

Type	Description
	Spark DataFrame with integer `age`, a three-band `age_cat` column, and
	`n_missing_vars` mapped to `None`, `Some`, or `All`.

Raises:

Type	Description
`Exception`	If either required input column is missing.

Notes

Ages outside 18–120 are removed. Age bands are 18–34, 35–54, and 55+; missing-variable bands are 0, 1–5, and greater than 5.

`panel_tools.execute_joiner_checker(display_table=False, lookback_days=42, ignore_vendors=[])`

Simulate Engine's joiner checker to see whether it would pass or fail.

Compares each vendor's most recent daily panel row count against its recent history, mirroring the check Engine performs before accepting new panel data.

Parameters:

Name	Type	Description	Default
`display_table`	`bool`	Display the per-vendor comparison table in the notebook.	`False`
`lookback_days`	`int`	Size of the history window, in days, used as the comparison baseline.	`42`
`ignore_vendors`	`list`	Vendor keys excluded from the check, mirroring Engine's `vendorsToIgnore`.	`[]`

Returns:

Type	Description
`bool`	`True` if the simulated check would pass, `False` if it would fail.

Notes

Reads the full s3://ap-prod-panels-v2/ panel and prints the outcome as a side effect. Vendors are discovered from the vendor_key= partitions, so a vendor with no partition is not evaluated. A vendor whose partition exists but has no data inside the lookback window fails the check (stale data), matching Engine's tooOldData condition.

`panel_tools.check_date_range_validity(start_yy, start_mm, start_dd, end_yy, end_mm, end_dd, campaign=False)`

Validate a date range used to select panel data.

Parameters:

Name	Description	Default
`start_yy`	Start year.	required
`start_mm`	Start month.	required
`start_dd`	Start day.	required
`end_yy`	End year.	required
`end_mm`	End month.	required
`end_dd`	End day.	required
`campaign`	If True, allow the end date to be in the future for an ongoing campaign. Otherwise, both dates must not be later than today.	`False`

Raises:

Type	Description
`Exception`	If a date component is not an integer, either date is invalid, a date is on or before January 1, 2000, or the range is reversed.

`panel_tools.get_folder_names(base_path, start_yy, start_mm, start_dd, end_yy, end_mm, end_dd, sort_result=True)`

Find date-partitioned folders within an inclusive range.

The folder tree is expected to use yyyy=.../mm=.../dd=... partitions. Paths are discovered with dbutils.fs without first loading the dataset.

Parameters:

Name	Description	Default
`base_path`	Root path containing the year partitions.	required
`start_yy`	Start year.	required
`start_mm`	Start month.	required
`start_dd`	Start day.	required
`end_yy`	End year.	required
`end_mm`	End month.	required
`end_dd`	End day.	required
`sort_result`	Whether to sort the result chronologically.	`True`

Returns:

Type	Description
	A pair containing the original folder paths and normalized paths whose
	month and day components are zero-padded for comparisons.

Raises:

Type	Description
`Exception`	If the requested date range is invalid.

`panel_tools.get_most_recent_panel_date(start_yy, start_mm, start_dd, panel_type='online', get_folder_path=False, load_earliest_available=False)`

Find the latest available panel at or before a reference date.

Searches the preceding 31 days in the storage location associated with the requested panel type.

Parameters:

Name	Description	Default
`start_yy`	Reference year.	required
`start_mm`	Reference month.	required
`start_dd`	Reference day.	required
`panel_type`	Panel storage family, such as `online`, `human`, `tv`, `household`, `deterministic_mobile`, `probabilistic_mobile`, or `amazon`.	`'online'`
`get_folder_path`	Return the matched storage path instead of its date.	`False`
`load_earliest_available`	For Amazon requests before October 6, 2023, return that earliest date instead of raising an exception.	`False`

Returns:

Type	Description
	The matched `[year, month, day]` list, or its folder path when
	`get_folder_path` is True.

Raises:

Type	Description
`Exception`	If `panel_type` is unsupported or an Amazon date predates the earliest available snapshot without fallback enabled.

`panel_tools.check_country_code_validity(country_code)`

Validate that a country code is supported by the online panel.

Parameters:

Name	Type	Description	Default
`country_code`		Uppercase ISO-style alpha-2 country code.	required

Raises:

Type	Description
`Exception`	If the country code is not in the supported panel list.

`panel_tools.load_online_panel(start_yy, start_mm, start_dd, end_yy, end_mm, end_dd, country_code=None, panel_source='user_report')`

Load online panel partitions for an inclusive date range.

Parameters:

Name	Description	Default
`start_yy`	Start year.	required
`start_mm`	Start month.	required
`start_dd`	Start day.	required
`end_yy`	End year.	required
`end_mm`	End month.	required
`end_dd`	End day.	required
`country_code`	Optional country used to select country partitions.	`None`
`panel_source`	Panel variant. Supported values are `user_report`, `all`, `all_raw`, and `all_sandbox`.	`'user_report'`

Returns:

Type	Description
	Spark DataFrame containing the requested panel partitions.

Raises:

Type	Description
`Exception`	If the dates, country code, or panel source are unsupported.

Notes

Loading several snapshots can introduce duplicate mask indexes because each panel generation resets that field.

`panel_tools.generate_churn_table(panel_source, country_code, N_weeks=6, dates=None, include_latest=True)`

Calculate weekly panel churn and survival metrics.

Parameters:

Name	Description	Default
`panel_source`	Retained for API compatibility. The current loader reads `user_report` panel data.	required
`country_code`	Country code passed to `load_online_panel`.	required
`N_weeks`	Number of recent panel snapshots to compare.	`6`
`dates`	Optional `[year, month, day]` reference date. Defaults to today.	`None`
`include_latest`	If True, include the latest panel at or before the reference date.	`True`

Returns:

Type	Description
	A pair of Spark DataFrames. The first contains `Date`,
	`Survival %`, and `Churn %`; the second also contains panelist
	and absolute churn counts.

Notes

Survival compares each snapshot with the oldest loaded panel using uid membership. Churn compares each snapshot with the preceding snapshot.

`panel_tools.get_N_most_recent_panel_dates(start_yy, start_mm, start_dd, N, panel_type='online', include_latest=True)`

Return the most recent available panel dates in reverse chronology.

Parameters:

Name	Description	Default
`start_yy`	Reference year.	required
`start_mm`	Reference month.	required
`start_dd`	Reference day.	required
`N`	Number of available panel dates to return.	required
`panel_type`	Panel storage family passed to `get_most_recent_panel_date`.	`'online'`
`include_latest`	If True, allow the panel on the reference date; otherwise begin strictly before it.	`True`

Returns:

Type	Description
	List of `[year, month, day]` date lists, newest first.

`panel_tools.convert_date_list_to_string(date_list)`

Format a three-element date sequence as YYYY-MM-DD.

Parameters:

Name	Type	Description	Default
`date_list`		Sequence containing year, month, and day.	required

Returns:

Type	Description
	Zero-padded date string.

`panel_tools.convert_date_string_to_list(date_string)`

Parse a hyphen-delimited date into integer components.

Parameters:

Name	Type	Description	Default
`date_string`		Date string such as `"2026-07-22"`.	required

Returns:

Type	Description
	`[year, month, day]` as integers.