wsample

Weighted sampling (with and without replacement) and reservoir sampling over streams, in pure Python with zero dependencies.

What is weighted sampling?

Standard random sampling picks items with equal probability. Weighted sampling lets you assign an importance (weight) to each item, so high-weight items are drawn more often.

Without replacement -- each item can appear in the result at most once. Use this when you need a diverse but importance-aware subset: top-k document candidates by relevance score, stratified data samples, sketch algorithms over data streams.

With replacement -- an item can be drawn multiple times. Use this when each draw is independent: Monte Carlo simulation, bootstrapping, generating random sequences from a fixed distribution. The multinomial function summarizes the same process as a count vector.

Why this library?

• Pure Python, zero dependencies. No NumPy, no SciPy. One import, works everywhere.
• Streaming. weighted_reservoir and reservoir consume any iterable in a single pass without materializing the whole dataset in memory.
• Correct algorithms. A-Res and A-ExpJ are the accepted standard for weighted reservoir sampling (Efraimidis and Spirakis 2006). Algorithm L is the standard for unweighted reservoir sampling (Li 1994).
• Reproducible. The caller passes an explicit rng=random.Random(seed); no global state is touched.

Install

pip install wsample

Note: wsample has not yet been published to PyPI. Install from source: pip install git+https://github.com/amaar-mc/wsample.git

Quick start

import random
from wsample import (
    weighted_sample_no_replacement,
    weighted_reservoir,
    reservoir,
    alias_sampler,
    sample_with_replacement,
    multinomial,
)

rng = random.Random(42)

# Weighted sample without replacement from a list (A-Res)
items = ["apple", "banana", "cherry", "date", "elderberry"]
weights = [1.0, 2.0, 5.0, 3.0, 1.0]
result = weighted_sample_no_replacement(items, weights, k=3, rng=rng)
# -> high-weight items like "cherry" appear more often; each item at most once

# Weighted sampling with replacement -- k i.i.d. draws, duplicates allowed
rng5 = random.Random(42)
indices = sample_with_replacement([1.0, 2.0, 5.0, 3.0, 1.0], k=10, rng=rng5)
# -> list of 10 indices in [0, 4]; index 2 (weight 5) appears most often

# Multinomial count vector -- same i.i.d. draws summarized as counts
rng6 = random.Random(42)
counts = multinomial([1.0, 2.0, 5.0, 3.0, 1.0], trials=1000, rng=rng6)
# -> list of 5 counts summing to 1000; counts[2] is largest

# Streaming weighted reservoir (A-ExpJ) -- same distribution, works on any iterable
def generate_documents():
    for doc_id, score in enumerate([0.9, 0.1, 0.7, 0.4, 0.8]):
        yield doc_id, score

rng2 = random.Random(0)
top_docs = weighted_reservoir(
    generate_documents(),
    weight_fn=lambda pair: pair[1],
    k=2,
    rng=rng2,
)

# Unweighted reservoir (Algorithm L) -- O(k(1+log(n/k))) RNG calls
rng3 = random.Random(0)
sample = reservoir(range(1_000_000), k=100, rng=rng3)

# Alias method for fast with-replacement draws from a fixed distribution
rng4 = random.Random(0)
draw = alias_sampler([1.0, 2.0, 3.0], rng=rng4)
index = draw()  # -> 0, 1, or 2 proportionally

API

All sampling functions take rng as a keyword-only argument with no default. Pass random.Random(seed) for reproducibility.

weighted_sample_no_replacement(items, weights, *, k, rng) -> list

Efraimidis-Spirakis A-Res. Returns up to k items from items, sampled without replacement proportional to weights. Items with weight 0 are never selected. k is clamped to the number of positive-weight items.

Raises ValueError if k < 0, len(weights) != len(items), or any weight is negative.

weighted_reservoir(stream, weight_fn, *, k, rng) -> list

A-ExpJ streaming algorithm. Consumes stream in a single pass. weight_fn(item) must return a strictly positive float. Equivalent in distribution to weighted_sample_no_replacement on the materialized list.

Raises ValueError if k < 0 or weight_fn returns a non-positive value.

reservoir(stream, *, k, rng) -> list

Unweighted Algorithm L. Single pass, O(k(1 + log(n/k))) RNG calls. Returns exactly min(k, n) items chosen uniformly at random without replacement.

Raises ValueError if k < 0.

alias_sampler(weights, *, rng) -> Callable[[], int]

Builds a Vose alias table in O(n) time. Returns a callable that draws an index in [0, len(weights)) in O(1) time per call, with replacement, proportional to weights.

Raises ValueError if weights is empty, any weight is negative, or all weights are zero.

sample_with_replacement(weights, *, k, rng) -> list[int]

Draws k indices independently with probability proportional to weight. Each draw is i.i.d. so the same index can appear multiple times. Builds a Vose alias table in O(n) then performs k O(1) draws, for O(n + k) total. k must be >= 1.

Raises ValueError if weights is empty, any weight is negative, all weights are zero, or k < 1.

multinomial(weights, *, trials, rng) -> list[int]

Returns the count vector for trials i.i.d. weighted draws. counts[i] is the number of times index i was drawn; len(counts) == len(weights) and sum(counts) == trials. Implemented by tallying sample_with_replacement, so both functions produce identical output under the same seed. trials must be >= 1.

Raises ValueError if weights is empty, any weight is negative, all weights are zero, or trials < 1.

With replacement vs without replacement: sample_with_replacement and multinomial allow repeated draws from the same index. weighted_sample_no_replacement and weighted_reservoir guarantee each index appears at most once in the result. Choose without-replacement when you need a diverse subset; choose with-replacement when draws are independent (bootstrapping, simulation, Monte Carlo).

See also

examples/top_k_stream.py for a worked example of streaming top-k weighted selection.

Contributing

See CONTRIBUTING.md.

License

MIT. See LICENSE.