Survey

Survey and polling analysis: crosstabs, weighting, significance testing, wave management, rendering.

siege_utilities.survey – Survey/crosstab report engine.

Modelled on the Quantipy Stack → Cluster → Chain → View hierarchy, built fresh for Python 3.12 on pandas + weightipy.

Install the optional dependency:: pip install siege-utilities[survey]

Hierarchy

View – one cell statistic (count, pct, CI)
Chain – one crosstab (row_var × break_vars)
Cluster – a named group of Chains = one report section
Stack – a complete report (all Clusters + shared WeightScheme)

Waves (longitudinal surveys)

Wave – the same questionnaire fielded at a single point in time
(carries df and optionally a per-wave Stack)
WaveSet – an ordered set of Waves; compare_chain produces a
LONGITUDINAL Chain aligned across waves

A Stack describes one fielding. A WaveSet composes many fieldings into change detection / trend output. Use a Stack when you have a single snapshot; use a WaveSet to track the same question across time.

Quick start:

from siege_utilities.survey import build_chain, chain_to_argument
from siege_utilities.reporting.pages.page_models import TableType

chain = build_chain(df, row_var="party", break_vars=["county"],
                    table_type=TableType.SINGLE_RESPONSE, geo_column="county")
argument = chain_to_argument(chain, headline="Party ID by County",
                             narrative="Democrats lead in metro counties.")

Longitudinal (multi-wave) example:

from datetime import date
from siege_utilities.survey import Wave, WaveSet

waveset = WaveSet(name="2024 tracker", waves=[
    Wave(id="W1", date=date(2024, 3, 1), df=df_march),
    Wave(id="W2", date=date(2024, 6, 1), df=df_june),
    Wave(id="W3", date=date(2024, 9, 1), df=df_sept),
])
chain = waveset.compare_chain(row_var="party")
# chain.views keyed by wave id; delta column appended automatically.

Submodules

Chain builder – converts a respondent/donor-level DataFrame into a Chain.

Delegates chi-square to siege_utilities.data.cross_tabulation (not reimplemented here).

exception siege_utilities.survey.crosstab.CrosstabInputError[source]

Bases: ValueError

Raised when build_chain receives an empty or schema-incompatible DataFrame.

Distinct from a successful Chain that happens to contain no rows (e.g., every category was filtered out downstream). This signals the caller fed in something the builder cannot work with at all – typically an upstream filter that returned nothing, or a column name typo.

siege_utilities.survey.crosstab.build_chain(df, row_var, break_vars, metric='value', weight_var=None, table_type=TableType.CROSS_TAB, top_n=None, geo_column=None)[source]

Build a Chain from a respondent/donor-level DataFrame.

Parameters:

df (pandas.DataFrame) – Input DataFrame; one row per respondent.
row_var (str) – Column whose distinct values become row categories.
break_vars (List[str]) – Columns to cross-tab against (become column groups).
metric (str) – Numeric column to aggregate (default “value”; use column count if absent).
weight_var (str | None) – Optional weight column name.
table_type (TableType) – Drives cell computation, chart type, and base note.
top_n (int | None) – If set, keep only the top N row categories by total volume.
geo_column (str | None) – If set, attached to Chain to signal map generation.

Return type:

Chain

Survey hierarchy dataclasses: View → Chain → Cluster → Stack.

Stack = complete report (all sections + shared WeightScheme) Cluster = one named section (→ one deck section) Chain = one cross-tab slide (question × break variables) View = one cell statistic (count, pct, sig flag)

class siege_utilities.survey.models.Chain[source]

Bases: object

One cross-tabulation: row_var × break_vars.

table_type drives chart selection, significance testing approach, and whether percent bases are respondents (MULTIPLE_RESPONSE) or column totals.

Fields populated by later pipeline stages (significance testing, weighting) are declared here with None defaults so dataclasses.fields() / dataclasses.asdict() see them – the previous pattern of attaching them dynamically via setattr broke introspection.

row_var: str

break_vars: List[str]

views: Dict[str, List[View]]

table_type: TableType

base_note: str = ''

geo_column: str | None = None

delta_column: str | None = None

chi_square_significant: bool | None = None

chi_square_p: float | None = None

to_dataframe()[source]

Render the chain as a display-ready wide DataFrame.

Columns = break variable values; rows = row variable categories. For MULTIPLE_RESPONSE, cells are percents (may exceed 100% column-sum).

Return type:: pandas.DataFrame

__init__(row_var, break_vars, views, table_type, base_note='', geo_column=None, delta_column=None, chi_square_significant=None, chi_square_p=None)

Parameters:

row_var (str)
break_vars (List[str])
views (Dict[str, List[View]])
table_type (TableType)
base_note (str)
geo_column (Optional[str])
delta_column (Optional[str])
chi_square_significant (Optional[bool])
chi_square_p (Optional[float])

Return type:

None

class siege_utilities.survey.models.Cluster[source]

Bases: object

A named group of Chains forming one report section.

name: str

chains: List[Chain]

add_chain(chain)[source]

Parameters:: chain (Chain)
Return type:: Cluster

__init__(name, chains=<factory>)

Parameters:

name (str)
chains (List[Chain])

Return type:

None

class siege_utilities.survey.models.Stack[source]

Bases: object

A complete survey report: all Clusters + shared WeightScheme.

name: str

clusters: List[Cluster]

weight_scheme: WeightScheme | None = None

add_cluster(cluster)[source]

Parameters:: cluster (Cluster)
Return type:: Stack

property all_chains: List[Chain]

__init__(name, clusters=<factory>, weight_scheme=None)

Parameters:

name (str)
clusters (List[Cluster])
weight_scheme (WeightScheme | None)

Return type:

None

class siege_utilities.survey.models.View[source]

Bases: object

A single statistic in one cell of a cross-tabulation.

metric: str

base: float

count: float

pct: float | None = None

sig_flag: str | None = None

ci: float | None = None

__init__(metric, base, count, pct=None, sig_flag=None, ci=None)

Parameters:

metric (str)
base (float)
count (float)
pct (float | None)
sig_flag (str | None)
ci (float | None)

Return type:

None

class siege_utilities.survey.models.Wave[source]

Bases: object

One survey wave: the same questionnaire fielded at a point in time.

df is the respondent-level DataFrame for this wave and is what WaveSet.compare_chain() consumes. stack is optional: callers that have already built a per-wave Stack (with its own weight scheme) can attach it for later retrieval; longitudinal comparison does not require it.

id: str

date: date

df: pandas.DataFrame | None = None

stack: Stack | None = None

weight_scheme: WeightScheme | None = None

__init__(id, date, df=None, stack=None, weight_scheme=None)

Parameters:

id (str)
date (date)
df (pandas.DataFrame | None)
stack (Stack | None)
weight_scheme (WeightScheme | None)

Return type:

None

class siege_utilities.survey.models.WaveSet[source]

Bases: object

An ordered set of Waves from the same survey instrument.

Waves are sorted by date on access so longitudinal output is always chronological regardless of insertion order.

name: str

waves: List[Wave]

client_id: str | None = None

add_wave(wave)[source]

Parameters:: wave (Wave)
Return type:: WaveSet

property ordered: List[Wave]

compare_chain(row_var, break_vars=None, *, metric='value', weight_var=None, top_n=None)[source]

Return a LONGITUDINAL Chain aligned across waves.

Columns are wave_id keys (ordered by date); rows are row_var categories. Delegates to siege_utilities.survey.waves so the primitive lives in one place.

Parameters:

row_var (str)
break_vars (List[str] | None)
metric (str)
weight_var (str | None)
top_n (int | None)

Return type:

Chain

__init__(name, waves=<factory>, client_id=None)

Parameters:

name (str)
waves (List[Wave])
client_id (str | None)

Return type:

None

class siege_utilities.survey.models.WeightScheme[source]

Bases: object

Target marginals for RIM (raking) weighting.

Example:

WeightScheme(targets={
    "age_group": {"18-34": 0.25, "35-54": 0.40, "55+": 0.35},
    "gender":    {"M": 0.48, "F": 0.52},
})

targets: Dict[str, Dict[Any, float]]

weight_col: str = 'weight'

max_iterations: int = 1000

convergence: float = 1e-06

validate()[source]

Return type:: None

__init__(targets, weight_col='weight', max_iterations=1000, convergence=1e-06)

Parameters:

targets (Dict[str, Dict[Any, float]])
weight_col (str)
max_iterations (int)
convergence (float)

Return type:

None

Client ↔ Survey registry.

Associates WaveSet instances (each one representing a survey instrument with its waves) to the client that owns them. Supports many surveys per client and fast lookup without scanning a flat list.

Survey identity within a client is the WaveSet’s name – two surveys for the same client must have distinct names. Names are not required to be globally unique; ("Acme", "Tracker Q2") and ("Beacon", "Tracker Q2") coexist fine.

Client branding / display info continues to live in siege_utilities.reporting.client_branding. The registry only answers “which surveys belong to this client” – not “how do we render this client’s report”.

exception siege_utilities.survey.registry.ClientSurveyError[source]

Bases: KeyError

Raised for registry-level lookup / duplication problems.

Subclasses KeyError so existing except KeyError handlers at call sites still catch it; use except ClientSurveyError when you need the more specific signal.

class siege_utilities.survey.registry.ClientSurveyRegistry[source]

Bases: object

In-memory map of client_id → {survey name → WaveSet}.

Deliberately in-memory and dependency-free. Persistence (YAML, SQL, wherever) is a caller concern – serialize _by_client directly or build higher-level storage on top.

register(waveset)[source]

Register a WaveSet whose client_id is already set.

Raises ClientSurveyError if client_id is unset or if a WaveSet with the same name is already registered for that client.

Parameters:: waveset (WaveSet)
Return type:: WaveSet

register_for(client_id, waveset)[source]

Set waveset.client_id = client_id and register it.

Raises ClientSurveyError if a WaveSet with the same name is already registered for that client.

Parameters:

client_id (str)
waveset (WaveSet)

Return type:

WaveSet

unregister(client_id, name)[source]

Remove and return the named survey for client_id.

Raises ClientSurveyError if the client or the named survey is not registered.

Parameters:

client_id (str)
name (str)

Return type:

WaveSet

clients()[source]

Return all registered client ids.

Return type:: List[str]

surveys_for(client_id)[source]

Return all WaveSets registered to client_id (empty list if none).

Parameters:: client_id (str)
Return type:: List[WaveSet]

get_survey(client_id, name)[source]

Return the named WaveSet for a client, or None if missing.

Parameters:

client_id (str)
name (str)

Return type:

WaveSet | None

require_survey(client_id, name)[source]

Like get_survey() but raises ClientSurveyError on miss.

Parameters:

client_id (str)
name (str)

Return type:

WaveSet

client_of(waveset)[source]

Return the client id a WaveSet is registered to, if any.

Prefers waveset.client_id but falls back to reverse lookup so callers stay correct even if the field was cleared after registration.

Parameters:: waveset (WaveSet)
Return type:: str | None

__init__(_by_client=<factory>)

Parameters:: _by_client (Dict[str, Dict[str, WaveSet]])
Return type:: None

Chain → Argument renderer.

chain_to_argument: converts one Chain to an Argument with chart and optional map. stack_to_arguments: walks all Chains in a Stack.

Chart type selection by TableType:: SINGLE_RESPONSE → horizontal bar MULTIPLE_RESPONSE → grouped bar (NOT stacked – stacked implies mutual exclusivity) CROSS_TAB → grouped bar (≤5 categories) or heatmap (>5) LONGITUDINAL → line chart RANKING → horizontal bar sorted descending MEAN_SCALE → bar with error bars BANNER → small-multiple bars

exception siege_utilities.survey.render.RenderError[source]

Bases: RuntimeError

Raised when Chain-to-Argument rendering hits a configuration or data issue.

class siege_utilities.survey.render.ArgumentCluster[source]

Bases: object

A named group of rendered Argument objects.

This is the output of stack_to_arguments() – a typed companion to models.Cluster (which holds Chains). Separate type prevents callers from accidentally feeding Argument-bearing objects into code that expects Chain-bearing Clusters.

name: str

arguments: List[Argument]

__init__(name, arguments=<factory>)

Parameters:

name (str)
arguments (List[Argument])

Return type:

None

siege_utilities.survey.render.chain_to_argument(chain, headline, narrative, *, chart_generator=None, map_generator=None)[source]

Convert a Chain to an Argument with chart and optional map.

Parameters:

chain (Chain) – Source Chain (output of build_chain).
headline (str) – Slide headline / section title.
narrative (str) – One-paragraph narrative text for the argument.
chart_generator (Optional[Any]) – Optional callable for chart rendering, invoked as chart_generator(df, chart_type, headline) -> Figure | None. When None, the library’s default matplotlib path is used (no branding required for sensible output).
map_generator (Optional[Any]) – Optional object exposing create_choropleth_map(df, geo_column, value_column). When None, a default ChartGenerator instance is used.
work (These kwargs make three cases)
None (1. Ad hoc -- both)
configured (no branding)
generator (2. Override on a one-off -- caller passes a pre-configured) – without mutating global branding state.
object (3. Alternative backend -- caller passes a Plotly-based callable /) – that matches the interface.

Returns:

chart populated when matplotlib is available; map_figure populated only when chain.geo_column is set and equals chain.row_var.

Return type:

Argument

Raises:

RenderError – On plotting failure or geo_column/row_var mismatch.

siege_utilities.survey.render.stack_to_arguments(stack, headlines=None, narratives=None)[source]

Walk all Chains in a Stack, converting each to an Argument.

Returns a list of ArgumentCluster (not Cluster) so the result type is explicit: these hold Arguments, not Chains.

Parameters:

stack (Stack) – Source Stack.
headlines (Optional[Dict]) – Optional dict mapping (cluster_name, row_var) → headline string.
narratives (Optional[Dict]) – Optional dict mapping (cluster_name, row_var) → narrative string.

Returns:

One per input Cluster; preserves cluster name.

Return type:

List[ArgumentCluster]

Significance testing for survey chains.

column_proportion_test: two-proportion z-test across all column pairs. chi_square_flag: wraps data.cross_tabulation.chi_square_test.

exception siege_utilities.survey.significance.SignificanceError[source]

Bases: RuntimeError

Raised when significance testing cannot complete on the given chain.

siege_utilities.survey.significance.column_proportion_test(chain, alpha=0.05)[source]

Add significance markers (letters A/B/C…) to Chain Views.

Tests each column proportion against every other column using a two-proportion z-test. When column j is significantly higher than column i (at alpha), appends column i’s letter to column j’s sig_flag for that row category.

Mutates chain in place and returns it.

Parameters:

chain (Chain)
alpha (float)

Return type:

Chain

siege_utilities.survey.significance.chi_square_flag(chain, alpha=0.05)[source]

Add an overall chi-square flag to chain.

Delegates to siege_utilities.data.statistics.cross_tabulation.chi_square_test. Sets chain.chi_square_significant = True/False and chain.chi_square_p.

Mutates chain in place and returns it.

Parameters:

chain (Chain)
alpha (float)

Return type:

Chain

Longitudinal analysis over a WaveSet.

Builds per-wave Chains via survey.crosstab.build_chain() and merges them into a single LONGITUDINAL Chain whose columns are wave ids (ordered by date). A delta column (last − first) is appended when ≥2 waves exist, matching the single-DataFrame LONGITUDINAL builder.

exception siege_utilities.survey.waves.WavesError[source]

Bases: ValueError

Raised when a WaveSet operation is given incomplete or mismatched data.

siege_utilities.survey.waves.compare_waves(waveset, row_var, break_vars=None, *, metric='value', weight_var=None, top_n=None)[source]

Merge per-wave crosstabs into one LONGITUDINAL Chain.

Parameters:

waveset (WaveSet) – The WaveSet to compare. Every Wave must have a non-None df.
row_var (str) – Column whose categories become rows (same across all waves).
break_vars (Optional[List[str]]) – Optional secondary breaks applied within each wave (rare for wave comparison but supported). If omitted or empty, each wave contributes a single column keyed by wave id.
metric (str) – Passed through to build_chain() for each wave.
weight_var (Optional[str]) – Passed through to build_chain() for each wave.
top_n (Optional[int]) – Passed through to build_chain() for each wave.

Returns:

table_type == TableType.LONGITUDINAL; views keyed by wave id (or "{wave_id}|{break_key}" when break_vars are used); delta_column set when ≥2 waves exist.

Return type:

Chain

Raises:

WavesError – If waveset has no waves, or any wave lacks a DataFrame.

RIM (raking) weighting wrapper around weightipy.

Install: pip install siege-utilities[survey]

weightipy’s public API uses weightipy.scheme_from_dict() + weightipy.weight_dataframe(); pass iteration / convergence parameters via the weightipy.internal.rim.Rim constructor’s rim_params dict.

exception siege_utilities.survey.weights.WeightingConvergenceError[source]

Bases: RuntimeError

Raised when weightipy fails to converge within max_iterations.

siege_utilities.survey.weights.apply_rim_weights(df, targets, *, weight_col='weight', max_iterations=1000, convergence=1e-06, verbose=False)[source]

Apply RIM (raking / iterative proportional fitting) weights.

Adds weight_col to a copy of df and returns the result. Does NOT mutate the input.

Parameters:

df (pd.DataFrame) – Respondent-level DataFrame; each row is one respondent.

targets (dict) –

Mapping of column name → {category: proportion}. Proportions per variable must sum to 1.0. Example:

{
    "age_group": {"18-34": 0.25, "35-54": 0.40, "55+": 0.35},
    "gender":    {"M": 0.48, "F": 0.52},
}

weight_col (str, keyword-only, default "weight") – Column name for the computed weight values.
max_iterations (int, keyword-only, default 1000) – Upper bound on raking iterations.
convergence (float, keyword-only, default 1e-6) – Convergence threshold (convcrit in weightipy terms).
verbose (bool, keyword-only, default False) – If True, weightipy prints per-iteration progress.

Returns:

Copy of df with weight_col populated.

Return type:

pd.DataFrame

Raises:

ImportError – If weightipy is not installed.
WeightingConvergenceError – If weightipy fails to converge.