statspai.survey¶
survey ¶
Survey design and weighted estimation — StatsPAI's answer to R's survey
package and Stata's svy: prefix.
Supports stratified, clustered, and weighted survey designs with design-corrected standard errors for means, totals, and regression.
import numpy as np import pandas as pd import statspai as sp rng = np.random.default_rng(0) n = 200 df = pd.DataFrame({'stratum': rng.integers(0, 4, n), ... 'education': rng.integers(8, 18, n), ... 'age': rng.integers(20, 65, n), ... 'pw': rng.uniform(1, 3, n)}) df['psu'] = df['stratum'] * 10 + rng.integers(0, 5, n) df['income'] = (10 + 2 * df['education'] + 0.3 * df['age'] ... + rng.normal(0, 5, n)) design = sp.svydesign(data=df, weights='pw', strata='stratum', ... cluster='psu') # doctest: +SKIP design.mean('income') # doctest: +SKIP design.total('income') # doctest: +SKIP design.glm('income ~ education + age') # doctest: +SKIP
SurveyDesign ¶
Declare a complex survey design.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
DataFrame
|
Survey microdata. |
required |
weights
|
str or array - like
|
Sampling weights (inverse probability). If str, column name in data. |
required |
strata
|
str or None
|
Stratification variable (column name). |
None
|
cluster
|
str or None
|
Primary sampling unit (PSU) variable (column name). |
None
|
fpc
|
str or None
|
Finite population correction, as in R |
None
|
nest
|
bool
|
If True, PSU ids are relabelled to be unique within strata. PSUs
are always identified within their stratum for variance and
degrees-of-freedom purposes (Stata |
False
|
lonely_psu
|
('remove', 'certainty', 'adjust', 'average', 'fail')
|
How a stratum with a single sampled PSU enters the variance, with
the semantics of R's
Unless |
"remove"
|
Examples:
>>> import numpy as np
>>> import pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> n = 300
>>> df = pd.DataFrame({
... "stratum": rng.integers(0, 3, size=n),
... "psu": rng.integers(0, 30, size=n),
... "wt": rng.uniform(0.5, 2.0, size=n),
... "income": rng.normal(50, 10, size=n),
... })
>>> design = sp.SurveyDesign(df, weights="wt", strata="stratum",
... cluster="psu", nest=True)
>>> design.n
300
>>> m = design.mean("income")
>>> type(m).__name__
'SurveyResult'
mean ¶
Design-corrected weighted mean(s); kwargs go to sp.svymean.
total ¶
Design-corrected weighted total(s); kwargs go to sp.svytotal.
glm ¶
Survey-weighted GLM; kwargs go to sp.svyglm.
calibrate ¶
calibrate(*, margins: Optional[Dict[str, Dict]] = None, totals: Optional[Dict[str, float]] = None, variance: str = 'greg', max_iter: int = 100, tol: float = 1e-10) -> 'SurveyDesign'
Calibrated design whose standard errors account for calibration.
Raking to margins (as :func:sp.rake) or linear GREG
calibration to totals (as :func:sp.linear_calibration).
Unlike feeding the calibrated weights to a new design -- which
treats them as fixed -- the returned design keeps the auxiliary
variables, and every estimator's linearisation scores are replaced
by their residuals from the weighted regression on them before the
design variance is formed. Estimates of the calibration totals
themselves then have zero variance, as they should.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
margins
|
dict
|
|
None
|
totals
|
dict
|
|
None
|
variance
|
('greg', 'stata')
|
Weights of the residualising regression. |
"greg"
|
max_iter
|
int
|
Raking controls (see :func: |
100
|
tol
|
int
|
Raking controls (see :func: |
100
|
Returns:
| Type | Description |
|---|---|
SurveyDesign
|
|
Examples:
>>> import numpy as np, pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> n = 400
>>> df = pd.DataFrame({
... "psu": rng.integers(0, 40, n), "stratum": rng.integers(0, 4, n),
... "d": rng.uniform(20, 40, n), "sex": rng.choice(["F", "M"], n),
... })
>>> df["y"] = 5 + (df["sex"] == "M") + rng.normal(size=n)
>>> des = sp.svydesign(df, weights="d", strata="stratum",
... cluster="psu", nest=True)
>>> cal = des.calibrate(margins={"sex": {"F": 6000, "M": 5000}})
>>> round(float(cal.weights.sum()))
11000
>>> bool(cal.mean("y").std_error.iloc[0] < des.mean("y").std_error.iloc[0])
True
PoststratResult
dataclass
¶
Bases: ResultProtocolMixin
Result of :func:poststratify.
Attributes:
| Name | Type | Description |
|---|---|---|
estimate |
float
|
Population-weighted average of the cell predictions (posterior mean when draws are available). |
sd |
float
|
Posterior standard deviation; |
lower, upper |
float
|
Central posterior interval. |
table |
DataFrame
|
The estimate overall (and by group with |
draws |
DataFrame or None
|
Posterior draws of each row of |
cells |
DataFrame
|
The poststratification table with each cell's predicted mean. |
Examples:
>>> import numpy as np, pandas as pd
>>> import statspai as sp
>>> cells = pd.DataFrame({"g": ["a", "b"], "N": [300, 700]})
>>> out = sp.poststratify([0.2, 0.6], cells, count="N")
>>> round(out.estimate, 2)
0.48
ReplicateDesign ¶
A survey design described by replicate weights.
Build it with :func:sp.svrepdesign (replicate weights shipped with
the data, e.g. ACS, MEPS, NHANES-style files) or
:func:sp.as_svrepdesign (jackknife / BRR weights built from a
:class:SurveyDesign).
Attributes:
| Name | Type | Description |
|---|---|---|
data |
DataFrame
|
|
weights |
(ndarray, shape(n))
|
Full-sample weights. |
repweights |
(ndarray, shape(n, R))
|
Replicate analysis weights (sampling weights included). |
type |
str
|
R's name of the replicate type. |
scale |
float
|
|
rscales |
(ndarray, shape(R))
|
|
mse |
bool
|
Centre the squared deviations at the full-sample estimate (True) or at the mean of the replicates (False). |
df |
float
|
Design degrees of freedom. |
Examples:
>>> import numpy as np, pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> df = pd.DataFrame({"psu": np.repeat(np.arange(10), 20),
... "w": rng.uniform(1, 3, 200),
... "y": rng.normal(size=200)})
>>> rep = sp.as_svrepdesign(sp.svydesign(df, weights="w", cluster="psu"))
>>> rep.n_replicates
10
>>> round(float(rep.mean("y").estimate["y"]), 6) == round(
... float(np.average(df["y"], weights=df["w"])), 6)
True
variance ¶
Replicate covariance of a statistic.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
theta_full
|
(p,) array
|
|
required |
theta_reps
|
(R, p) array
|
The statistic recomputed with each replicate's weights. |
required |
mean ¶
Replicate-weight :func:sp.svymean.
total ¶
Replicate-weight :func:sp.svytotal.
glm ¶
Replicate-weight :func:sp.svyglm.
linear_calibration ¶
linear_calibration(data: DataFrame, totals: Dict[str, float], weight: Optional[str] = None) -> CalibrationResult
Deville-Särndal (1992) linear calibration.
Find calibrated weights w_i = g_i * d_i minimising the chi-squared
distance Σ (w_i - d_i)² / d_i = Σ d_i (g_i - 1)² subject to
Σ w_i x_{ik} = T_k for each auxiliary variable k with known total
T_k. Closed form g = 1 + X (X' D X)^{-1} (T - X' d); identical to R
survey::calibrate(design, ~ 0 + x1 + ..., population, calfun =
"linear") (unbounded). No intercept is added: include a column of
ones with total N for one (a dummy column for a category count).
Returns weights only; for calibration-aware standard errors use
design.calibrate(totals=...) on a :class:SurveyDesign.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
totals
|
dict
|
|
required |
weight
|
str
|
Design weight column. If |
None
|
Examples:
>>> import statspai as sp
>>> import numpy as np
>>> import pandas as pd
>>> rng = np.random.default_rng(0)
>>> n = 200
>>> df = pd.DataFrame({
... "income": rng.normal(50.0, 10.0, size=n),
... "age": rng.normal(40.0, 12.0, size=n),
... })
>>> totals = {"income": df["income"].sum() * 1.05,
... "age": df["age"].sum() * 0.98}
>>> res = sp.linear_calibration(df, totals=totals)
>>> cal_total = float((res.calibrated_weights * df["income"]).sum())
>>> bool(abs(cal_total - totals["income"]) < 1e-4)
True
rake ¶
rake(data: DataFrame, margins: Dict[str, Dict], weight: Optional[str] = None, max_iter: int = 100, tol: float = 1e-10) -> CalibrationResult
Raking (iterative proportional fitting).
Starting from the design weights, cycles through the margins and
rescales the weights within each category so its weighted share equals
the target, until every margin holds. The fixed point is the
raking-ratio (multiplicative-distance) calibration: R
survey::rake run to convergence and
survey::calibrate(calfun = "raking") give the same weights.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
DataFrame
|
|
required |
margins
|
dict
|
|
required |
weight
|
str
|
Existing design weight column. If |
None
|
max_iter
|
int
|
Maximum number of full sweeps over the margins. |
100
|
tol
|
float
|
Convergence: after a sweep, every weighted category share is within
|
1e-10
|
Returns:
| Type | Description |
|---|---|
CalibrationResult
|
|
Notes
Up to 1.28.0 convergence was declared when the largest absolute
change in the sum-to-one weights fell below tol = 1e-6. Those
weights are O(1/n), so for large samples the loop stopped after one or
two sweeps with margins still off (4e-4 relative at n = 100 000).
The calibrated weights are only weights: passing them to
sp.svydesign treats them as fixed, which is not R's
calibration-adjusted linearisation variance (residuals of y on the
calibration variables). For standard errors that account for the
calibration use design.calibrate(margins=...) on a
:class:SurveyDesign (test_survey_calibrated_design_parity.py).
Examples:
>>> import statspai as sp
>>> import numpy as np
>>> import pandas as pd
>>> rng = np.random.default_rng(0)
>>> n = 200
>>> df = pd.DataFrame({
... "sex": rng.choice(["M", "F"], size=n),
... "age_group": rng.choice(["18-34", "35-54", "55+"], size=n),
... })
>>> margins = {
... "sex": {"M": 0.49, "F": 0.51},
... "age_group": {"18-34": 0.30, "35-54": 0.40, "55+": 0.30},
... }
>>> res = sp.rake(df, margins=margins)
>>> bool(res.converged)
True
svydesign ¶
svydesign(data: DataFrame, weights: Union[str, ndarray], strata: Optional[str] = None, cluster: Optional[str] = None, fpc: Optional[str] = None, nest: bool = False, lonely_psu: Optional[str] = None) -> SurveyDesign
Create a survey design object — functional interface.
Parameters are identical to :class:SurveyDesign.
Examples:
>>> import numpy as np
>>> import pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> n = 300
>>> df = pd.DataFrame({
... "region": rng.integers(0, 3, size=n),
... "psu_id": rng.integers(0, 30, size=n),
... "pw": rng.uniform(0.5, 2.0, size=n),
... "income": rng.normal(50, 10, size=n),
... "age": rng.normal(40, 12, size=n),
... })
>>> design = sp.svydesign(data=df, weights='pw', strata='region',
... cluster='psu_id', nest=True)
>>> type(design).__name__
'SurveyDesign'
>>> m = design.mean('income')
>>> g = design.glm('income ~ age')
svyglm ¶
svyglm(formula: str, design: 'SurveyDesign', family: str = 'gaussian', alpha: float = 0.05, dof: str = 'design', subpop: Any = None, subpop_df: str = 'stata') -> SurveyResult
Survey-weighted generalised linear model.
Fits WLS (for gaussian family) or weighted IRLS (for binomial/poisson) and computes design-corrected standard errors via the sandwich estimator.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
formula
|
str
|
|
required |
design
|
SurveyDesign or ReplicateDesign
|
The design from :func: |
required |
family
|
str
|
|
'gaussian'
|
alpha
|
float
|
|
0.05
|
dof
|
('design', 'residual')
|
Degrees of freedom for t statistics, p-values and CIs.
|
"design"
|
subpop
|
str or bool array
|
Domain (subpopulation) estimation, as Stata |
None
|
subpop_df
|
('stata', 'r')
|
Design df for a domain: strata without domain members are dropped;
Stata then counts every PSU of the remaining strata, R
|
"stata"
|
Returns:
| Type | Description |
|---|---|
SurveyResult with regression coefficient estimates.
|
|
Examples:
>>> import numpy as np, pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> n = 200
>>> df = pd.DataFrame({
... "income": rng.normal(50, 10, n),
... "age": rng.integers(20, 65, n),
... "region": rng.integers(0, 4, n), # strata
... "psu_id": rng.integers(0, 20, n), # clusters
... "pw": rng.uniform(1.0, 3.0, n), # sampling weights
... })
>>> design = sp.svydesign(data=df, weights="pw", strata="region",
... cluster="psu_id", nest=True)
>>> res = sp.svyglm("income ~ age", design)
>>> list(res.estimate.index)
['Intercept', 'age']
svymean ¶
svymean(variables: Union[str, List[str]], design: 'SurveyDesign', alpha: float = 0.05, deff: str = 'wor', subpop: Any = None, subpop_df: str = 'stata') -> SurveyResult
Survey-weighted mean with design-corrected standard errors.
Uses Taylor-series linearisation; matches R survey::svymean output.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
variables
|
str or list of str
|
Column name(s) in the design's data. |
required |
design
|
SurveyDesign or ReplicateDesign
|
The design from :func: |
required |
alpha
|
float
|
CIs use the t distribution with the design df (#PSU - #strata), as
Stata |
0.05
|
deff
|
('wor', 'replace')
|
Denominator of the design effect: SRS without replacement with
N = sum of weights (R |
"wor"
|
subpop
|
str or bool array
|
Domain (subpopulation) estimation, as Stata |
None
|
subpop_df
|
('stata', 'r')
|
Design df for a domain: strata without domain members are dropped;
Stata then counts every PSU of the remaining strata, R
|
"stata"
|
Returns:
| Type | Description |
|---|---|
SurveyResult
|
|
Examples:
>>> import numpy as np, pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> n = 200
>>> df = pd.DataFrame({
... "income": rng.normal(50, 10, n),
... "region": rng.integers(0, 4, n), # strata
... "psu_id": rng.integers(0, 20, n), # clusters
... "pw": rng.uniform(1.0, 3.0, n), # sampling weights
... })
>>> design = sp.svydesign(data=df, weights="pw", strata="region",
... cluster="psu_id", nest=True)
>>> res = sp.svymean("income", design)
>>> list(res.estimate.index)
['income']
svytotal ¶
svytotal(variables: Union[str, List[str]], design: 'SurveyDesign', alpha: float = 0.05, deff: str = 'wor', subpop: Any = None, subpop_df: str = 'stata') -> SurveyResult
Survey-weighted total with design-corrected standard errors.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
variables
|
str or list of str
|
|
required |
design
|
SurveyDesign or ReplicateDesign
|
The design from :func: |
required |
alpha
|
float
|
CIs use the t distribution with the design df (#PSU - #strata), as
Stata |
0.05
|
deff
|
('wor', 'replace')
|
Denominator of the design effect: SRS without replacement with
N = sum of weights (R |
"wor"
|
subpop
|
str or bool array
|
Domain (subpopulation) estimation, as Stata |
None
|
subpop_df
|
('stata', 'r')
|
Design df for a domain: strata without domain members are dropped;
Stata then counts every PSU of the remaining strata, R
|
"stata"
|
Returns:
| Type | Description |
|---|---|
SurveyResult
|
|
Examples:
>>> import numpy as np, pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> n = 200
>>> df = pd.DataFrame({
... "income": rng.normal(50, 10, n),
... "region": rng.integers(0, 4, n), # strata
... "psu_id": rng.integers(0, 20, n), # clusters
... "pw": rng.uniform(1.0, 3.0, n), # sampling weights
... })
>>> design = sp.svydesign(data=df, weights="pw", strata="region",
... cluster="psu_id", nest=True)
>>> res = sp.svytotal("income", design)
>>> list(res.estimate.index)
['income']
poststratify ¶
poststratify(fit: Any, cells: DataFrame, count: str = 'N', by: Optional[Union[str, Sequence[str]]] = None, level: float = 0.95) -> PoststratResult
Average model predictions over the cells of a known population.
A sample that over- or under-represents some groups gives a biased
mean. Fit a regression of the outcome on the variables that define
the groups, predict the mean of every population cell, and weight
the predictions by the population cell counts:
sum_j N_j theta_j / sum_j N_j. With a Bayesian fit the same
average is taken draw by draw, so the estimate carries the
uncertainty of the regression.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fit
|
fitted model, array of predictions, or matrix of draws
|
A model with |
required |
cells
|
DataFrame
|
The poststratification table: one row per population cell with the columns the model's formula needs and the cell count. |
required |
count
|
str
|
Column of |
'N'
|
by
|
str or list of str
|
Also report the estimate within the levels of these columns of
|
None
|
level
|
float
|
Mass of the posterior interval. |
0.95
|
Returns:
| Type | Description |
|---|---|
PoststratResult
|
|
Raises:
| Type | Description |
|---|---|
MethodIncompatibility
|
(a |
Notes
The posterior describes uncertainty in the cell means given the model. It takes the population counts as known and assumes that, within a cell, respondents resemble the people they stand for; when the counts are estimates or nonresponse within cells is selective, the true uncertainty is larger.
Examples:
>>> import numpy as np, pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> poll = pd.DataFrame({"pid": rng.choice(["R", "D", "I"], 600,
... p=[0.5, 0.3, 0.2])})
>>> share = poll["pid"].map({"R": 0.8, "D": 0.1, "I": 0.4})
>>> poll["vote"] = (rng.uniform(size=600) < share).astype(int)
>>> fit = sp.bayes_regress("vote ~ C(pid)", poll, model="logit",
... draws=4000, burnin=1000, seed=1)
>>> cells = pd.DataFrame({"pid": ["R", "D", "I"], "N": [0.33, 0.36, 0.31]})
>>> out = sp.poststratify(fit, cells, count="N")
>>> bool(0.35 < out.estimate < 0.50)
True
References
gelman2020regression
as_svrepdesign ¶
as_svrepdesign(design: Any, type: str = 'auto', rho: float = 0.0, mse: Optional[bool] = None) -> ReplicateDesign
Replicate weights from a strata / PSU design (R as.svrepdesign).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
design
|
SurveyDesign
|
Built with :func: |
required |
type
|
('auto', 'JK1', 'JKn', 'BRR', 'Fay')
|
Bootstrap weights are random and are not built here: pass weights
produced elsewhere to :func: |
'auto'
|
rho
|
float
|
Fay coefficient for |
0
|
mse
|
bool
|
As :func: |
None
|
Returns:
| Type | Description |
|---|---|
ReplicateDesign
|
|
Examples:
>>> import numpy as np, pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(1)
>>> df = pd.DataFrame({"h": np.repeat(np.arange(4), 30),
... "psu": np.repeat(np.arange(12), 10),
... "w": rng.uniform(1, 2, 120),
... "y": rng.normal(size=120)})
>>> des = sp.svydesign(df, weights="w", strata="h", cluster="psu")
>>> sp.as_svrepdesign(des, type="JKn").n_replicates
12
svrepdesign ¶
svrepdesign(data: DataFrame, repweights: Any, weights: Union[str, ndarray, None] = None, type: str = 'JK1', rho: Optional[float] = None, scale: Optional[float] = None, rscales: Optional[Sequence[float]] = None, mse: Optional[bool] = None, combined_weights: bool = True, df: Optional[float] = None) -> ReplicateDesign
Declare a survey design by its replicate weights.
R survey::svrepdesign / Stata svyset [pw=w], brrweight(...),
jkrweight(...), bsrweight(...), sdrweight(...).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
DataFrame
|
Survey microdata. |
required |
repweights
|
list of str, str or (n, R) array
|
The replicate-weight columns, a regular expression matching their
names ( |
required |
weights
|
str or array
|
Full-sample sampling weights. Without it they are the mean of the replicate weights (with a warning). |
None
|
type
|
('BRR', 'Fay', 'JK1', 'JKn', 'JK2', 'bootstrap', 'ACS', 'successive-difference', 'other')
|
Replicate method; sets |
'BRR'
|
rho
|
float
|
Fay coefficient ( |
None
|
scale
|
optional
|
Overall and per-replicate multipliers of the squared deviations (ignored, with a warning, for the types that fix them). |
None
|
rscales
|
optional
|
Overall and per-replicate multipliers of the squared deviations (ignored, with a warning, for the types that fix them). |
None
|
mse
|
bool
|
True: deviations from the full-sample estimate (Stata |
None
|
combined_weights
|
bool
|
The replicate columns already include the sampling weights (the
usual case). False: they are multipliers of |
True
|
df
|
float
|
Design degrees of freedom; default the numerical rank of the
replicate weights minus one (R |
None
|
Returns:
| Type | Description |
|---|---|
ReplicateDesign
|
Usable wherever |
Examples:
>>> import numpy as np, pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> df = pd.DataFrame({"y": rng.normal(size=100), "w": 1.0})
>>> for r in range(10):
... df[f"rw{r}"] = np.where(np.arange(100) // 10 == r, 0.0, 10 / 9)
>>> rep = sp.svrepdesign(df, repweights="^rw", weights="w", type="JK1")
>>> rep.n_replicates, rep.df
(10, 9.0)