Probabilistic Sketches & Compact Filters¶

import hashlib


class CuckooFilter:
    """Fingerprint table. Lookup is exact for stored fingerprints, probabilistic for keys."""

    def __init__(self, num_buckets: int = 32, slots: int = 4, fp_bits: int = 8, max_kicks: int = 32):
        self.n = num_buckets
        self.slots = slots
        self.fp_mask = (1 << fp_bits) - 1
        self.max_kicks = max_kicks
        self.table: list[list[int | None]] = [[None] * slots for _ in range(num_buckets)]

    def _fp(self, item: str) -> int:
        h = int(hashlib.md5(item.encode()).hexdigest(), 16)
        return (h & self.fp_mask) or 1  # fingerprint 0 reserved for empty

    def _h1(self, item: str) -> int:
        h = int(hashlib.md5(item.encode()).hexdigest(), 16)
        return (h >> 16) % self.n

    def _h2(self, bucket: int, fp: int) -> int:
        return (bucket ^ (fp * 0x5bd1e995)) % self.n

    def insert(self, item: str) -> bool:
        fp = self._fp(item)
        i1 = self._h1(item)
        i2 = self._h2(i1, fp)
        for i in (i1, i2):
            for s, occ in enumerate(self.table[i]):
                if occ is None:
                    self.table[i][s] = fp
                    return True
        i = i1
        for _ in range(self.max_kicks):
            s = 0
            fp, self.table[i][s] = self.table[i][s], fp
            i = self._h2(i, fp)
            for s, occ in enumerate(self.table[i]):
                if occ is None:
                    self.table[i][s] = fp
                    return True
        return False  # table too full — unlike Bloom, insert can fail

    def might_contain(self, item: str) -> bool:
        fp = self._fp(item)
        i1 = self._h1(item)
        i2 = self._h2(i1, fp)
        return fp in self.table[i1] or fp in self.table[i2]

    def delete(self, item: str) -> bool:
        """Safe only for an item you previously inserted (same as counting Bloom)."""
        fp = self._fp(item)
        i1 = self._h1(item)
        i2 = self._h2(i1, fp)
        for i in (i1, i2):
            for s, occ in enumerate(self.table[i]):
                if occ == fp:
                    self.table[i][s] = None
                    return True
        return False
        # Time: O(1) typical; insert worst case O(max_kicks)

import hashlib
from bisect import bisect_left, insort


class QuotientFilter:
    """Quotienting: bucket = high bits, store only the remainder. Soft FPs, deleteable."""

    def __init__(self, q_bits: int = 10, r_bits: int = 8):
        self.n_buckets = 1 << q_bits
        self.r_bits = r_bits
        self.r_mask = (1 << r_bits) - 1
        self.runs: list[list[int]] = [[] for _ in range(self.n_buckets)]

    def _qr(self, item: str) -> tuple[int, int]:
        h = int(hashlib.md5(item.encode()).hexdigest(), 16)
        remainder = h & self.r_mask
        quotient = (h >> self.r_bits) % self.n_buckets
        return quotient, remainder

    def insert(self, item: str) -> None:
        q, r = self._qr(item)
        run = self.runs[q]
        i = bisect_left(run, r)
        if i == len(run) or run[i] != r:
            insort(run, r)
        # Time: O(run length) — runs stay short at reasonable load

    def might_contain(self, item: str) -> bool:
        q, r = self._qr(item)
        run = self.runs[q]
        i = bisect_left(run, r)
        return i < len(run) and run[i] == r

    def delete(self, item: str) -> bool:
        q, r = self._qr(item)
        run = self.runs[q]
        i = bisect_left(run, r)
        if i < len(run) and run[i] == r:
            run.pop(i)
            return True
        return False

import hashlib


def _u64(item: str, seed: int) -> int:
    digest = hashlib.sha1(f"{seed}:{item}".encode()).digest()
    return int.from_bytes(digest[:8], "little")


class XorFilter:
    def __init__(self, keys: list[str], seed: int = 1):
        keys = list(dict.fromkeys(keys))  # unique, stable
        self.B: list[int] = []
        self.n_part = 0
        self.seed = seed
        if not keys:
            return
        n = len(keys)
        m = int(1.23 * n) + 33
        while m % 3:
            m += 1
        part = m // 3
        for attempt in range(16):
            s = seed + attempt
            triples = []
            for k in keys:
                h = _u64(k, s)
                triples.append((
                    h % part,
                    part + ((h >> 21) % part),
                    2 * part + ((h >> 42) % part),
                    (h & 0xFF) or 1,
                ))
            placed = self._peel(m, triples)
            if placed is not None:
                self.B = placed
                self.n_part = part
                self.seed = s
                return
        raise RuntimeError("XOR filter: peeling failed — grow m or change seed")

    def _peel(self, m: int, triples: list[tuple[int, int, int, int]]) -> list[int] | None:
        count = [0] * m
        at: list[list[int]] = [[] for _ in range(m)]
        for i, (a, b, c, _) in enumerate(triples):
            for x in (a, b, c):
                count[x] += 1
                at[x].append(i)
        queue = [i for i in range(m) if count[i] == 1]
        order: list[tuple[int, int]] = []  # (key index, cell we peel it from)
        used = [False] * len(triples)
        qi = 0
        while qi < len(queue):
            cell = queue[qi]
            qi += 1
            if count[cell] != 1:
                continue
            kid = next((i for i in at[cell] if not used[i]), None)
            if kid is None:
                continue
            used[kid] = True
            order.append((kid, cell))
            for x in triples[kid][:3]:
                count[x] -= 1
                if count[x] == 1:
                    queue.append(x)
        if len(order) != len(triples):
            return None
        B = [0] * m
        for kid, cell in reversed(order):
            a, b, c, fp = triples[kid]
            B[cell] = fp ^ B[a] ^ B[b] ^ B[c]  # cell is still 0, so this sets it
        return B

    def might_contain(self, item: str) -> bool:
        if not self.B:
            return False
        h = _u64(item, self.seed)
        p = self.n_part
        a, b, c = h % p, p + ((h >> 21) % p), 2 * p + ((h >> 42) % p)
        fp = (h & 0xFF) or 1
        return (self.B[a] ^ self.B[b] ^ self.B[c]) == fp
        # Time: O(1). Space: ~1.23 fingerprints per key, built once.

import hashlib


class CountMinSketch:
    def __init__(self, depth: int = 4, width: int = 64):
        self.d = depth
        self.w = width
        self.table = [[0] * width for _ in range(depth)]

    def _idx(self, item: str, row: int) -> int:
        h = int(hashlib.md5(f"{row}:{item}".encode()).hexdigest(), 16)
        return h % self.w

    def add(self, item: str, count: int = 1) -> None:
        for r in range(self.d):
            self.table[r][self._idx(item, r)] += count
        # Time: O(d)

    def estimate(self, item: str) -> int:
        return min(self.table[r][self._idx(item, r)] for r in range(self.d))
        # Time: O(d)
        # Always ≥ true count (non-negative updates). Never a false "zero"
        # if the item was added — the analogue of Bloom's no false negatives,
        # for counts.

import math


class LogUniformHistogram:
    """Log10-uniform buckets (constant ratio between adjacent buckets), not a real HDR Histogram. Values ≥ 1."""

    def __init__(self, highest: int = 10**7, sig_digits: int = 2):
        self.scale = 10 ** sig_digits          # 100 buckets per decade at sig_digits=2
        self.counts = [0] * (int(math.log10(highest) * self.scale) + 2)
        self.n = 0

    def _idx(self, v: int) -> int:
        v = max(1, v)
        return min(int(math.log10(v) * self.scale), len(self.counts) - 1)

    def _value_at(self, idx: int) -> int:
        return max(1, int(round(10 ** (idx / self.scale))))

    def record(self, value: int) -> None:
        self.counts[self._idx(int(value))] += 1
        self.n += 1
        # Time: O(1)

    def merge(self, other: "LogUniformHistogram") -> None:
        assert len(other.counts) == len(self.counts)
        for i, c in enumerate(other.counts):
            self.counts[i] += c
        self.n += other.n

    def percentile(self, q: float) -> int:
        if self.n == 0:
            return 0
        target = q * self.n
        seen = 0
        for i, c in enumerate(self.counts):
            seen += c
            if seen >= target:
                return self._value_at(i)
        return self._value_at(len(self.counts) - 1)
        # Time for percentile: O(buckets) ≪ O(samples)

class TDigest:
    """Merging centroids with a scale function: more capacity in the tails than the body.

    Production: use a battle-tested t-digest. This is the algorithm, not the paper's
    full interpolation / combined-buffer variant.
    """

    def __init__(self, compression: float = 50.0):
        self.delta = compression
        self.centroids: list[list[float]] = []  # [mean, count]

    def add(self, x: float, w: int = 1) -> None:
        self.centroids.append([float(x), float(w)])
        if len(self.centroids) > self.delta * 4:
            self.compress()

    def _cap(self, q: float, n: float) -> float:
        # Smaller cap near q=0 and q=1 → tails stay unmerged.
        return max(1.0, self.delta * 4 * q * (1 - q))

    def compress(self) -> None:
        if not self.centroids:
            return
        self.centroids.sort()
        n = sum(c[1] for c in self.centroids)
        out: list[list[float]] = []
        seen = 0.0
        for mean, w in self.centroids:
            q = (seen + w / 2) / n
            if out and out[-1][1] + w <= self._cap(q, n):
                m0, w0 = out[-1]
                out[-1] = [(m0 * w0 + mean * w) / (w0 + w), w0 + w]
            else:
                out.append([mean, w])
            seen += w
        self.centroids = out

    def quantile(self, q: float) -> float:
        self.compress()
        if not self.centroids:
            return float("nan")
        n = sum(c[1] for c in self.centroids)
        target = q * n
        seen = 0.0
        for mean, w in self.centroids:
            seen += w
            if seen >= target:
                return mean
        return self.centroids[-1][0]

import hashlib


def _h(item: str) -> int:
    return int(hashlib.sha1(item.encode()).hexdigest(), 16)


def minhash(items: set[str], k: int = 128) -> tuple[int, ...]:
    """k independent hashes approximated by (h + i * h2) — same trick as Bloom."""
    sig = [2**64] * k
    for item in items:
        a, b = _h(item + ":a"), _h(item + ":b")
        for i in range(k):
            sig[i] = min(sig[i], (a + i * b) % (2**64))
    return tuple(sig)


def estimated_jaccard(sig_a: tuple[int, ...], sig_b: tuple[int, ...]) -> float:
    return sum(x == y for x, y in zip(sig_a, sig_b)) / len(sig_a)
    # Time to build: O(|set| · k). Compare two sets: O(k), independent of set size.