API Reference

This page contains the complete API reference for fewlab.

Main Functions

Fewlab: Optimal item selection for efficient labeling and survey sampling.

Main API functions: - items_to_label: Deterministic A-optimal selection - pi_aopt_for_budget: A-optimal inclusion probabilities - balanced_fixed_size: Balanced sampling with fixed size - scale_pi_to_budget: The inclusion probabilities that sampler delivers - row_se_min_labels: Row-wise SE minimization - calibrate_weights: GREG-style weight calibration - core_plus_tail: Hybrid deterministic core + balanced tail - adaptive_core_tail: Data-driven hybrid selection

class fewlab.CoreTailResult(selected, probabilities, core, tail, ht_weights, mixed_weights, diagnostics)

Bases: object

Structured result for hybrid core+tail selection methods.

Parameters:

• selected (Index)

• probabilities (Series)

• core (Index)

• tail (Index)

• ht_weights (Series)

• mixed_weights (Series)

• diagnostics (dict[str, Any])

selected

All selected item identifiers (core + tail).

Type:

pandas.Index

probabilities

A-optimal inclusion probabilities for all items.

Type:

pandas.Series

core

Deterministic core items (highest influence).

Type:

pandas.Index

tail

Probabilistic tail items (balanced sampling).

Type:

pandas.Index

ht_weights

Standard Horvitz-Thompson weights for the selected items.

Type:

pandas.Series

mixed_weights

Mixed weights (1/pi for core, 1.0 for tail) for variance reduction.

Type:

pandas.Series

diagnostics

Additional metadata such as budget splits and tail fraction.

Type:

dict[str, Any]

Properties:

budget_used: Total number of items selected. budget_core: Number of items in the deterministic core. budget_tail: Number of items in the probabilistic tail. tail_frac: Fraction of the budget allocated to the tail.

Examples

>>> import numpy as np
>>> import pandas as pd
>>> from fewlab import Design
>>> rng = np.random.default_rng(0)
>>> counts = pd.DataFrame(rng.poisson(5, (1000, 100)))
>>> X = pd.DataFrame(rng.standard_normal((1000, 3)))
>>> design = Design(counts, X)
>>> result = design.sample(
...     budget=50, method="core_plus_tail", tail_frac=0.2
... )
>>> len(result.selected), len(result.core), len(result.tail)
(50, 40, 10)

selected: Index

probabilities: Series

core: Index

tail: Index

ht_weights: Series

mixed_weights: Series

diagnostics: dict[str, Any]

property budget_used: int

Total number of items selected.

property budget_core: int

Number of items in deterministic core.

property budget_tail: int

Number of items in probabilistic tail.

property tail_frac: float

Fraction of budget allocated to tail.

property probability_sum: float

Sum of inclusion probabilities.

class fewlab.Design(counts, X, *, ridge='auto', ensure_full_rank=True)

Bases: object

Primary interface for optimal experimental design with cached computations.

The class stores processed data, cached influence matrices, and diagnostics so that repeated operations such as selection, sampling, and calibration can reuse expensive intermediate results.

Examples

>>> import pandas as pd
>>> import numpy as np
>>> from fewlab import Design
>>>
>>> rng = np.random.default_rng(0)
>>> counts = pd.DataFrame(rng.poisson(5, (1000, 100)))
>>> X = pd.DataFrame(rng.standard_normal((1000, 3)))
>>> design = Design(counts, X)
>>> len(design.select(budget=20).selected)
20

Parameters:

• counts (DataFrame)

• X (DataFrame)

• ridge (float | Literal['auto'])

• ensure_full_rank (bool)

property n_units: int

Number of units (rows) after preprocessing.

property n_items: int

Number of items (columns) after preprocessing.

property influence_weights: Series

A-optimal influence weights w_j for each item.

property diagnostics: dict[str, Any]

Comprehensive diagnostic information about the design.

select(budget, method='deterministic')

Select items using deterministic algorithms.

Parameters:

• budget (int) – Number of items to select.

• method (Literal['deterministic', 'greedy']) – Selection algorithm: “deterministic” (batch) or “greedy” (sequential).

Returns:

Selection result with items, influence weights, and diagnostics.

Raises:

ValidationError – If the method name is unknown.

Return type:

SelectionResult

inclusion_probabilities(budget, *, pi_min=0.0001, method='aopt', **kwargs)

Compute inclusion probabilities for a given budget.

Parameters:

• budget (int) – Expected total budget (sum of inclusion probabilities).

• pi_min (float) – Minimum inclusion probability per item.

• method (Literal['aopt', 'row_se']) – Probability computation strategy, “aopt” or “row_se”.

• **kwargs (Any) – Additional method-specific arguments (e.g. eps2 for “row_se”).

Returns:

Probability result with inclusion probabilities and diagnostics.

Raises:

ValidationError – If the method name is unknown.

Return type:

ProbabilityResult

sample(budget, method='balanced', *, random_state=None, **kwargs)

Generate probabilistic samples using various methods.

Parameters:

• budget (int) – Number of items to sample.

• method (Literal['balanced', 'core_plus_tail', 'adaptive']) – Sampling method (“balanced”, “core_plus_tail”, or “adaptive”).

• random_state (int | Generator | None) – Random state for reproducible sampling. Accepts None, an int seed, or a numpy Generator.

• **kwargs (Any) – Method-specific parameters (e.g. tail_frac, pi_min, tolerances).

Returns:

Sampled item identifiers.

Raises:

ValidationError – If the method name is unknown.

Return type:

SamplingResult | CoreTailResult

calibrate_weights(selected, pop_totals=None, *, distance='chi2', ridge=1e-08, nonneg=True)

Compute calibrated weights for selected items.

Parameters:

• selected (Index | list[str]) – Identifiers of sampled items.

• pop_totals (ndarray | None) – Optional population totals; defaults to sums of the g matrix.

• distance (str) – Calibration distance measure (e.g., “chi2”).

• ridge (float) – Ridge regularization parameter.

• nonneg (bool) – Whether to enforce non-negative calibrated weights.

Returns:

Calibrated weights indexed by the selected items.

Return type:

Series

estimate(selected, labels, weights=None, *, normalize_by_total=True)

Compute calibrated Horvitz-Thompson estimates for row shares.

Parameters:

• selected (Index | list[str]) – Identifiers of sampled items.

• labels (Series) – Observed labels for the selected items.

• weights (Series | None) – Optional calibrated weights; if omitted they are computed internally.

• normalize_by_total (bool) – Whether to divide by row totals to produce shares.

Returns:

Estimation result with estimates, weights, and diagnostics.

Return type:

EstimationResult

class fewlab.EstimationResult(estimates, weights, selected, diagnostics)

Bases: object

Structured result for estimation methods.

Parameters:

• estimates (Series)

• weights (Series)

• selected (Index)

• diagnostics (dict[str, Any])

estimates

Row-wise estimates.

Type:

pandas.Series

weights

Calibrated weights used for estimation.

Type:

pandas.Series

selected

Items used for estimation.

Type:

pandas.Index

diagnostics

Estimation diagnostics.

Type:

dict[str, Any]

Examples

>>> import numpy as np
>>> import pandas as pd
>>> from fewlab import Design
>>> rng = np.random.default_rng(0)
>>> counts = pd.DataFrame(rng.poisson(5, (1000, 100)))
>>> X = pd.DataFrame(rng.standard_normal((1000, 3)))
>>> design = Design(counts, X)
>>> selected = design.select(budget=30).selected
>>> labels = pd.Series(rng.random(len(selected)), index=selected)
>>> result = design.estimate(selected, labels)
>>> len(result.estimates)
1000

estimates: Series

weights: Series

selected: Index

diagnostics: dict[str, Any]

class fewlab.ProbabilityResult(probabilities, influence_projections, diagnostics)

Bases: object

Structured result for probability computation methods.

Provides access to computed probabilities, influence projections, and computation diagnostics.

Parameters:

• probabilities (Series)

• influence_projections (ndarray)

• diagnostics (dict[str, Any])

probabilities

Inclusion probabilities indexed by item identifiers.

Type:

pandas.Series

influence_projections

Regression projections g_j = X^T v_j for all items (shape (p, m)). Used for balanced sampling and weight calibration.

Type:

numpy.ndarray

diagnostics

Computation diagnostics and metadata.

Type:

dict[str, Any]

Properties:

budget_used: Sum of inclusion probabilities.

Examples

>>> import numpy as np
>>> import pandas as pd
>>> from fewlab import Design
>>> rng = np.random.default_rng(0)
>>> counts = pd.DataFrame(rng.poisson(5, (1000, 100)))
>>> X = pd.DataFrame(rng.standard_normal((1000, 3)))
>>> design = Design(counts, X)
>>> result = design.inclusion_probabilities(budget=50, method="aopt")
>>> round(result.budget_used, 1)
50.0
>>> # influence_projections then feeds balanced sampling
>>> from fewlab import balanced_fixed_size
>>> selected = balanced_fixed_size(
...     result.probabilities, result.influence_projections, 50
... )
>>> len(selected)
50

probabilities: Series

influence_projections: ndarray

diagnostics: dict[str, Any]

property budget_used: float

Sum of inclusion probabilities.

class fewlab.RowSEResult(probabilities, max_violation, tolerance, iterations, best_iteration, feasible)

Bases: object

Result container for row_se_min_labels.

Parameters:

• probabilities (Series)

• max_violation (float)

• tolerance (float)

• iterations (int)

• best_iteration (int)

• feasible (bool)

probabilities

Inclusion probabilities indexed by item identifiers.

Type:

pandas.Series

max_violation

Maximum constraint violation encountered.

Type:

float

tolerance

Target violation tolerance.

Type:

float

iterations

Number of iterations executed.

Type:

int

best_iteration

Iteration index where the best solution was recorded.

Type:

int

feasible

Whether the best solution satisfies the tolerance.

Type:

bool

probabilities: Series

max_violation: float

tolerance: float

iterations: int

best_iteration: int

feasible: bool

to_series()

Return a copy of the probabilities as a Series.

Return type:

Series

to_dict()

Return diagnostic information as a dict.

Return type:

dict[str, Any]

class fewlab.SamplingResult(sample, probabilities, weights, diagnostics)

Bases: object

Structured result for probabilistic sampling methods.

Parameters:

• sample (Index)

• probabilities (Series)

• weights (Series)

• diagnostics (dict[str, Any])

sample

Sampled item identifiers.

Type:

pandas.Index

probabilities

Inclusion probabilities used for sampling.

Type:

pandas.Series

weights

Suggested sampling weights for the sampled items.

Type:

pandas.Series

diagnostics

Sampling diagnostics and metadata.

Type:

dict[str, Any]

Properties:

sample_size: Number of sampled items.

Examples

>>> import numpy as np
>>> import pandas as pd
>>> from fewlab import Design
>>> rng = np.random.default_rng(0)
>>> counts = pd.DataFrame(rng.poisson(5, (1000, 100)))
>>> X = pd.DataFrame(rng.standard_normal((1000, 3)))
>>> design = Design(counts, X)
>>> result = design.sample(budget=30, method="balanced")
>>> result.sample_size
30

sample: Index

probabilities: Series

weights: Series

diagnostics: dict[str, Any]

property sample_size: int

Number of sampled items.

property probability_sum: float

Sum of inclusion probabilities.

class fewlab.SelectionResult(selected, influence_weights, diagnostics)

Bases: object

Structured result for deterministic selection methods.

Parameters:

• selected (Index)

• influence_weights (Series)

• diagnostics (dict[str, Any])

selected

Selected item identifiers ordered by influence.

Type:

pandas.Index

influence_weights

A-optimal influence weights used for selection.

Type:

pandas.Series

diagnostics

Selection diagnostics and metadata.

Type:

dict[str, Any]

Properties:

budget_used: Number of items selected.

Examples

>>> import numpy as np
>>> import pandas as pd
>>> from fewlab import Design
>>> rng = np.random.default_rng(0)
>>> counts = pd.DataFrame(rng.poisson(5, (1000, 100)))
>>> X = pd.DataFrame(rng.standard_normal((1000, 3)))
>>> design = Design(counts, X)
>>> result = design.select(budget=30, method="deterministic")
>>> len(result.selected)
30

selected: Index

influence_weights: Series

diagnostics: dict[str, Any]

property budget_used: int

Number of items selected.

fewlab.adaptive_core_tail(counts, X, budget, *, min_tail_frac=0.1, max_tail_frac=0.4, condition_threshold=1000000.0, random_state=None)

Adaptive core+tail selection with a data-driven tail fraction.

The routine increases the tail fraction when X^T X is poorly conditioned and decreases it when influence weights are highly concentrated.

Parameters:

• counts (DataFrame) – Count matrix.

• X (DataFrame) – Feature matrix.

• budget (int) – Total number of items to select.

• min_tail_frac (float) – Minimum allowable tail fraction.

• max_tail_frac (float) – Maximum allowable tail fraction.

• condition_threshold (float) – Baseline condition number scale.

• random_state (int | Generator | None) – Random state for the balanced sampling step. Accepts None, an int seed, or a numpy Generator.

Returns:

Selection result identical to core_plus_tail, with adaptive metadata in info.

Return type:

CoreTailResult

fewlab.balanced_fixed_size(pi, g, budget, *, random_state=None)

Fixed-size balanced sampling with exact inclusion probabilities.

Draws exactly budget items by the cube method, so that three things hold at once: item j is included with probability exactly pi_delivered[j], the sample size never varies, and the calibration residual sum_S (I_j/pi_j - 1) g_j is driven to zero as far as the first two allow. The balance is what reduces the variance of Horvitz-Thompson estimators; the exact probabilities are what make them unbiased in the first place.

The delivered probabilities are scale_pi_to_budget(pi, budget) rather than pi itself, because sum(pi) is the expected sample size and a fixed-size design of budget items is incoherent unless the probabilities sum to budget. When they already do – as they do for pi_aopt_for_budget(…, budget) – the rescaling is a no-op. Weight by the delivered probabilities, not by the input, or call Design.sample, which reports them.

Parameters:

• pi (Series) – Inclusion probabilities for items. Index contains item identifiers.

• g (ndarray) – Regression projections g_j = X^T v_j for each item j (shape (p, m)).

• budget (int) – Fixed sample size (number of items to select).

• random_state (int | Generator | None) – Random state for reproducible sampling. Accepts None, an int seed, or a numpy Generator.

Returns:

Index of selected items. Length equals budget.

Raises:

ValidationError – If pi, g, or budget fail validation checks.

Return type:

Index

See also

scale_pi_to_budget: The probabilities this sampler actually delivers. pi_aopt_for_budget: Compute optimal inclusion probabilities. core_plus_tail: Hybrid deterministic + balanced sampling. calibrate_weights: Post-stratification weight adjustment.

Examples

>>> import numpy as np
>>> import pandas as pd
>>> from fewlab import balanced_fixed_size, pi_aopt_for_budget
>>> rng = np.random.default_rng(0)
>>> counts = pd.DataFrame(rng.poisson(5, (1000, 100)))
>>> X = pd.DataFrame(rng.standard_normal((1000, 3)))
>>> probs = pi_aopt_for_budget(counts, X, budget=30)
>>> selected = balanced_fixed_size(
...     probs.probabilities, probs.influence_projections, 30, random_state=42
... )
>>> len(selected)
30

Notes

Earlier releases drew the sample proportional to pi without replacement and then swapped items greedily to improve balance. Neither step preserves pi: on the package’s own test fixture, items with pi == 1 were included 75% of the time before the swaps and 45% after, which biased every estimator weighted by 1 / pi.

fewlab.calibrate_weights(pi, g, selected, pop_totals=None, *, distance='chi2', ridge=1e-08, nonneg=True)

Compute calibrated weights via GREG/Deville-Särndal calibration.

Parameters:

• pi (Series) – Inclusion probabilities for all items (index = item names).

• g (ndarray) – Regression projections g_j = X^T v_j for all items (shape (p, m)).

• selected (Sequence[str] | Index) – Item identifiers drawn in the sample.

• pop_totals (ndarray | None) – Known population totals (shape (p,)); defaults to g.sum(axis=1).

• distance (str) – Calibration distance measure; currently only “chi2” is supported.

• ridge (float) – Ridge regularization parameter for numerical stability.

• nonneg (bool) – Whether to enforce non-negative calibrated weights.

Returns:

Calibrated weights indexed by the selected items.

Raises:

• NotImplementedError – If distance is not “chi2”.

• ValueError – If pop_totals has the wrong shape.

Return type:

Series

Notes

The closed-form solution for chi-square distance is w* = d_S + G_S^T (G_S G_S^T + ridge I)^{-1} (t - G_S d_S), where d_S are the base weights.

References

Deville, J.-C., & Särndal, C.-E. (1992). Calibration estimators in survey sampling. Journal of the American Statistical Association, 87(418), 376-382.

fewlab.calibrated_ht_estimator(counts, labels, weights, *, normalize_by_total=True)

Compute calibrated Horvitz-Thompson estimator for row shares.

Parameters:

• counts (DataFrame) – Count matrix with rows as units and columns as items.

• labels (Series) – Item labels for the selected items.

• weights (Series) – Calibrated weights for the selected items.

• normalize_by_total (bool) – Whether to divide by row totals to obtain shares.

Returns:

Estimated row shares (or totals if normalize_by_total is False).

Return type:

Series

fewlab.core_plus_tail(counts, X, budget, *, tail_frac=0.2, random_state=None, ensure_full_rank=True, ridge=None)

Hybrid sampler combining a deterministic core with a balanced probabilistic tail.

Strategy:

1. Select budget_core = (1 - tail_frac) * budget items deterministically (largest w_j).

2. Compute A-optimal inclusion probabilities for the full budget.

3. Draw the remaining budget_tail items using balanced sampling.

Parameters:

• counts (DataFrame) – Count matrix with units as rows and candidate items as columns.

• X (DataFrame) – Feature matrix aligned with counts.index.

• budget (int) – Total number of items to select.

• tail_frac (float) – Fraction of the budget allocated to the probabilistic tail.

• random_state (int | Generator | None) – Random state for balanced tail selection. Accepts None, an int seed, or a numpy Generator.

• ensure_full_rank (bool) – Whether to regularize X^T X if it is rank-deficient.

• ridge (float | None) – Optional ridge penalty added to X^T X.

Returns:

Selection result containing the chosen items, inclusion probabilities, and metadata.

Raises:

ValidationError – If inputs fail validation or the core/tail split is infeasible.

Return type:

CoreTailResult

Examples

>>> import pandas as pd
>>> import numpy as np
>>> from fewlab import core_plus_tail
>>>
>>> counts = pd.DataFrame(np.random.poisson(10, (1000, 200)))
>>> X = pd.DataFrame(np.random.randn(1000, 5))
>>> result = core_plus_tail(counts, X, budget=50, tail_frac=0.2)
>>> result.selected.shape
(50,)

fewlab.greedy_aopt_selection(counts, X, budget, *, ensure_full_rank=True, ridge=None)

Select items using greedy A-optimal sequential selection.

The algorithm iteratively chooses the item that maximally reduces the trace of the covariance matrix using Sherman-Morrison updates.

Parameters:

• counts (DataFrame) – Count matrix with non-negative entries.

• X (DataFrame) – Feature matrix aligned with counts.index.

• budget (int) – Number of items to select sequentially.

• ensure_full_rank (bool) – Whether to add a ridge if the information matrix becomes singular.

• ridge (float | None) – Optional explicit ridge parameter.

Returns:

Selection result with items, influence weights, and diagnostics.

Return type:

SelectionResult

See also

items_to_label: Batch A-optimal selection (faster, different results). pi_aopt_for_budget: Compute inclusion probabilities for A-optimal design.

Examples

>>> import pandas as pd
>>> import numpy as np
>>> from fewlab import greedy_aopt_selection
>>>
>>> counts = pd.DataFrame(np.random.poisson(5, (1000, 100)))
>>> X = pd.DataFrame(np.random.randn(1000, 3))
>>> result = greedy_aopt_selection(counts, X, budget=20)
>>> len(result.selected)
20

fewlab.items_to_label(counts, X, budget, *, ensure_full_rank=True, ridge=None)

Select items to label using deterministic A-optimal design.

Influence weights are computed as w_j = g_j^T (X^T X)^{-1} g_j, and the top entries are returned.

Parameters:

• counts (DataFrame) – Count matrix with units as rows and items as columns.

• X (DataFrame) – Feature matrix aligned with counts.index.

• budget (int) – Number of items to select.

• ensure_full_rank (bool) – Whether to add a ridge term when X^T X is ill-conditioned.

• ridge (float | None) – Optional ridge parameter overriding the automatic heuristic.

Returns:

Selection result with items, influence weights, and diagnostics.

Return type:

SelectionResult

See also

pi_aopt_for_budget: Compute inclusion probabilities for the same design. greedy_aopt_selection: Greedy sequential variant. core_plus_tail: Hybrid deterministic and probabilistic selection.

Examples

>>> import pandas as pd
>>> import numpy as np
>>> from fewlab import items_to_label
>>>
>>> counts = pd.DataFrame(np.random.poisson(5, (1000, 200)))
>>> X = pd.DataFrame(np.random.randn(1000, 3))
>>> result = items_to_label(counts, X, budget=50)
>>> len(result.selected)
50

fewlab.pi_aopt_for_budget(counts, X, budget, *, pi_min=0.0001, ensure_full_rank=True, ridge=None)

Compute A-optimal first-order inclusion probabilities for a target budget.

The probabilities follow the square-root rule pi_j = clip(c * sqrt(w_j), [pi_min, 1]), with c chosen so that sum(pi) = budget.

Parameters:

• counts (DataFrame) – Count matrix with non-negative values.

• X (DataFrame) – Feature matrix aligned with counts.index.

• budget (int) – Expected total budget (sum of inclusion probabilities).

• pi_min (float) – Minimum allowed inclusion probability.

• ensure_full_rank (bool) – Whether to add a small ridge term when X^T X is ill-conditioned.

• ridge (float | None) – Explicit ridge parameter overriding the automatic heuristic.

Returns:

Probability result with inclusion probabilities and computation diagnostics.

Return type:

ProbabilityResult

Note

If budget < m * pi_min (where m is the number of items), the budget constraint cannot be satisfied. The function then returns every probability as pi_min, so sum(pi) = m * pi_min > budget, and issues a warning. The violation details land in the result’s diagnostics under budget_violation.

See also

items_to_label: Deterministic selection using the same influence weights. balanced_fixed_size: Fixed-size balanced sampling using these probabilities.

Examples

>>> import pandas as pd
>>> import numpy as np
>>> from fewlab import pi_aopt_for_budget
>>>
>>> counts = pd.DataFrame(np.random.poisson(5, (1000, 200)))
>>> X = pd.DataFrame(np.random.randn(1000, 3))
>>> result = pi_aopt_for_budget(counts, X, budget=50)
>>> round(result.budget_used, 1)
50.0

fewlab.row_se_min_labels(counts, eps2, *, pi_min=0.0001, max_iter=8000, tol=1e-06, random_state=None, return_result=False, raise_on_failure=False)

Minimize expected labels subject to row-wise SE limits.

The routine solves:

` minimize   sum_j pi_j subject to sum_j q_ij / pi_j <= eps2_i + sum_j q_ij,  q_ij = (c_ij / T_i)^2 `

Parameters:

• counts (DataFrame) – Non-negative count matrix with units as rows and items as columns.

• eps2 (ndarray | Series) – Row-wise squared standard-error tolerance; scalar applies to every row.

• pi_min (float) – Minimum allowable inclusion probability.

• max_iter (int) – Maximum optimization iterations.

• tol (float) – Convergence tolerance for constraint violations.

• random_state (int | Generator | None) – Random state for the stochastic subgradient steps. Accepts None, an int seed, or a numpy Generator.

• return_result (bool) – If True, return a RowSEResult with diagnostics.

• raise_on_failure (bool) – If True, raise a ValidationError when constraints remain violated.

Returns:

Probability series if return_result is False (default) or a RowSEResult with diagnostics when return_result is True.

Raises:

ValidationError – If inputs are invalid or raise_on_failure is True and the constraints remain violated after optimization.

Return type:

RowSEResult | Series

See also

pi_aopt_for_budget: A-optimal probabilities for a fixed budget. items_to_label: Deterministic selection without SE constraints.

Examples

>>> import numpy as np
>>> import pandas as pd
>>> from fewlab import row_se_min_labels
>>> rng = np.random.default_rng(0)
>>> counts = pd.DataFrame(rng.poisson(10, (200, 20)))
>>> pi = row_se_min_labels(counts, eps2=0.3**2, max_iter=20_000)
>>> bool(0 < pi.sum() <= counts.shape[1])
True

fewlab.scale_pi_to_budget(pi, budget, *, max_iter=1000)

Rescale inclusion probabilities to sum to budget, respecting the cap.

sum(pi) is the expected sample size, so a fixed-size design of budget items is only coherent when the probabilities sum to budget. Scaling by a constant would push large entries above one, which is not a probability, so entries that would exceed one are pinned there and the remaining budget is redistributed over the rest. Repeating that to a fixed point is the standard construction, and it terminates because each pass pins at least one entry.

Parameters:

• pi (ndarray) – Non-negative target probabilities.

• budget (int) – Desired sample size, at most len(pi).

• max_iter (int) – Guard on the redistribution loop.

Returns:

Probabilities in [0, 1] summing to budget.

Return type:

np.ndarray

Raises:

ValueError – If budget is negative, exceeds the number of items, or exceeds the number of items with positive probability.

Examples: >>> import numpy as np >>> scaled = scale_pi_to_budget(np.array([0.9, 0.2, 0.2, 0.2]), 2) >>> float(scaled.sum()) 2.0 >>> bool((scaled <= 1.0).all()) True

fewlab.topk(arr, k, *, index=None)

Return indices of the top-k entries of arr in descending order.

Parameters:

• arr (ndarray) – Array of scores to rank.

• k (int) – Number of entries to keep.

• index (Index | None) – Optional index to map positions back to labels.

Returns:

Index of the top-k entries ordered by decreasing value.

Return type:

Index

Core Module

A-optimal influence weights, inclusion probabilities, and item selection.

This is the entry point most callers use: it turns a counts matrix and a feature matrix into per-item influence weights, and from those into either a deterministic shortlist or a set of inclusion probabilities for a budget.

class fewlab.core.Influence(w, g, cols)

Bases: object

Influence data structure with memory-optimized slots.

Parameters:

• w (ndarray)

• g (ndarray)

• cols (list[str])

w: ndarray

g: ndarray

cols: list[str]

fewlab.core.pi_aopt_for_budget(counts, X, budget, *, pi_min=0.0001, ensure_full_rank=True, ridge=None)

Compute A-optimal first-order inclusion probabilities for a target budget.

The probabilities follow the square-root rule pi_j = clip(c * sqrt(w_j), [pi_min, 1]), with c chosen so that sum(pi) = budget.

Parameters:

• counts (DataFrame) – Count matrix with non-negative values.

• X (DataFrame) – Feature matrix aligned with counts.index.

• budget (int) – Expected total budget (sum of inclusion probabilities).

• pi_min (float) – Minimum allowed inclusion probability.

• ensure_full_rank (bool) – Whether to add a small ridge term when X^T X is ill-conditioned.

• ridge (float | None) – Explicit ridge parameter overriding the automatic heuristic.

Returns:

Probability result with inclusion probabilities and computation diagnostics.

Return type:

ProbabilityResult

Note

If budget < m * pi_min (where m is the number of items), the budget constraint cannot be satisfied. The function then returns every probability as pi_min, so sum(pi) = m * pi_min > budget, and issues a warning. The violation details land in the result’s diagnostics under budget_violation.

See also

items_to_label: Deterministic selection using the same influence weights. balanced_fixed_size: Fixed-size balanced sampling using these probabilities.

Examples

>>> import pandas as pd
>>> import numpy as np
>>> from fewlab import pi_aopt_for_budget
>>>
>>> counts = pd.DataFrame(np.random.poisson(5, (1000, 200)))
>>> X = pd.DataFrame(np.random.randn(1000, 3))
>>> result = pi_aopt_for_budget(counts, X, budget=50)
>>> round(result.budget_used, 1)
50.0

fewlab.core.items_to_label(counts, X, budget, *, ensure_full_rank=True, ridge=None)

Select items to label using deterministic A-optimal design.

Influence weights are computed as w_j = g_j^T (X^T X)^{-1} g_j, and the top entries are returned.

Parameters:

• counts (DataFrame) – Count matrix with units as rows and items as columns.

• X (DataFrame) – Feature matrix aligned with counts.index.

• budget (int) – Number of items to select.

• ensure_full_rank (bool) – Whether to add a ridge term when X^T X is ill-conditioned.

• ridge (float | None) – Optional ridge parameter overriding the automatic heuristic.

Returns:

Selection result with items, influence weights, and diagnostics.

Return type:

SelectionResult

See also

pi_aopt_for_budget: Compute inclusion probabilities for the same design. greedy_aopt_selection: Greedy sequential variant. core_plus_tail: Hybrid deterministic and probabilistic selection.

Examples

>>> import pandas as pd
>>> import numpy as np
>>> from fewlab import items_to_label
>>>
>>> counts = pd.DataFrame(np.random.poisson(5, (1000, 200)))
>>> X = pd.DataFrame(np.random.randn(1000, 3))
>>> result = items_to_label(counts, X, budget=50)
>>> len(result.selected)
50

Selection Module

Small selection helpers shared by the selection strategies.

fewlab.selection.topk(arr, k, *, index=None)

Return indices of the top-k entries of arr in descending order.

Parameters:

• arr (ndarray) – Array of scores to rank.

• k (int) – Number of entries to keep.

• index (Index | None) – Optional index to map positions back to labels.

Returns:

Index of the top-k entries ordered by decreasing value.

Return type:

Index

Balanced Sampling

Fixed-size balanced sampling via the cube method.

Draws a sample of exactly the requested size while honouring the given inclusion probabilities and keeping the regression projections balanced.

fewlab.balanced.balanced_fixed_size(pi, g, budget, *, random_state=None)

Fixed-size balanced sampling with exact inclusion probabilities.

Draws exactly budget items by the cube method, so that three things hold at once: item j is included with probability exactly pi_delivered[j], the sample size never varies, and the calibration residual sum_S (I_j/pi_j - 1) g_j is driven to zero as far as the first two allow. The balance is what reduces the variance of Horvitz-Thompson estimators; the exact probabilities are what make them unbiased in the first place.

The delivered probabilities are scale_pi_to_budget(pi, budget) rather than pi itself, because sum(pi) is the expected sample size and a fixed-size design of budget items is incoherent unless the probabilities sum to budget. When they already do – as they do for pi_aopt_for_budget(…, budget) – the rescaling is a no-op. Weight by the delivered probabilities, not by the input, or call Design.sample, which reports them.

Parameters:

• pi (Series) – Inclusion probabilities for items. Index contains item identifiers.

• g (ndarray) – Regression projections g_j = X^T v_j for each item j (shape (p, m)).

• budget (int) – Fixed sample size (number of items to select).

• random_state (int | Generator | None) – Random state for reproducible sampling. Accepts None, an int seed, or a numpy Generator.

Returns:

Index of selected items. Length equals budget.

Raises:

ValidationError – If pi, g, or budget fail validation checks.

Return type:

Index

See also

scale_pi_to_budget: The probabilities this sampler actually delivers. pi_aopt_for_budget: Compute optimal inclusion probabilities. core_plus_tail: Hybrid deterministic + balanced sampling. calibrate_weights: Post-stratification weight adjustment.

Examples

>>> import numpy as np
>>> import pandas as pd
>>> from fewlab import balanced_fixed_size, pi_aopt_for_budget
>>> rng = np.random.default_rng(0)
>>> counts = pd.DataFrame(rng.poisson(5, (1000, 100)))
>>> X = pd.DataFrame(rng.standard_normal((1000, 3)))
>>> probs = pi_aopt_for_budget(counts, X, budget=30)
>>> selected = balanced_fixed_size(
...     probs.probabilities, probs.influence_projections, 30, random_state=42
... )
>>> len(selected)
30

Notes

Earlier releases drew the sample proportional to pi without replacement and then swapped items greedily to improve balance. Neither step preserves pi: on the package’s own test fixture, items with pi == 1 were included 75% of the time before the swaps and 45% after, which biased every estimator weighted by 1 / pi.

Row Standard Error Minimization

Inclusion probabilities under row-wise standard-error constraints.

Instead of fixing a budget and minimising variance, this module fixes a per-row variance ceiling and minimises the expected number of labels.

fewlab.rowse.row_se_min_labels(counts, eps2, *, pi_min=0.0001, max_iter=8000, tol=1e-06, random_state=None, return_result=False, raise_on_failure=False)

Minimize expected labels subject to row-wise SE limits.

The routine solves:

` minimize   sum_j pi_j subject to sum_j q_ij / pi_j <= eps2_i + sum_j q_ij,  q_ij = (c_ij / T_i)^2 `

Parameters:

• counts (DataFrame) – Non-negative count matrix with units as rows and items as columns.

• eps2 (ndarray | Series) – Row-wise squared standard-error tolerance; scalar applies to every row.

• pi_min (float) – Minimum allowable inclusion probability.

• max_iter (int) – Maximum optimization iterations.

• tol (float) – Convergence tolerance for constraint violations.

• random_state (int | Generator | None) – Random state for the stochastic subgradient steps. Accepts None, an int seed, or a numpy Generator.

• return_result (bool) – If True, return a RowSEResult with diagnostics.

• raise_on_failure (bool) – If True, raise a ValidationError when constraints remain violated.

Returns:

Probability series if return_result is False (default) or a RowSEResult with diagnostics when return_result is True.

Raises:

ValidationError – If inputs are invalid or raise_on_failure is True and the constraints remain violated after optimization.

Return type:

RowSEResult | Series

See also

pi_aopt_for_budget: A-optimal probabilities for a fixed budget. items_to_label: Deterministic selection without SE constraints.

Examples

>>> import numpy as np
>>> import pandas as pd
>>> from fewlab import row_se_min_labels
>>> rng = np.random.default_rng(0)
>>> counts = pd.DataFrame(rng.poisson(10, (200, 20)))
>>> pi = row_se_min_labels(counts, eps2=0.3**2, max_iter=20_000)
>>> bool(0 < pi.sum() <= counts.shape[1])
True