確率的データ構造入門：Bloom FilterとHyperLogLogのPython実装

import hashlib
import math


class BloomFilter:
    """Bloom Filter のスクラッチ実装"""

    def __init__(self, expected_items: int, fp_rate: float = 0.01):
        """
        :param expected_items: 挿入予定の要素数
        :param fp_rate: 許容する偽陽性率（デフォルト 1%）
        """
        # 最適なビット配列サイズ: m = -n*ln(p) / (ln2)^2
        self.size = self._optimal_size(expected_items, fp_rate)
        # 最適なハッシュ関数の個数: k = (m/n) * ln2
        self.hash_count = self._optimal_hash_count(self.size, expected_items)
        self.bit_array = [0] * self.size
        self.item_count = 0

    @staticmethod
    def _optimal_size(n: int, p: float) -> int:
        """最適なビット配列サイズを計算"""
        m = -n * math.log(p) / (math.log(2) ** 2)
        return int(math.ceil(m))

    @staticmethod
    def _optimal_hash_count(m: int, n: int) -> int:
        """最適なハッシュ関数の個数を計算"""
        k = (m / n) * math.log(2)
        return int(math.ceil(k))

    def _hashes(self, item: str) -> list[int]:
        """k 個のハッシュ値を生成（double hashing 方式）"""
        h1 = int(hashlib.md5(item.encode()).hexdigest(), 16)
        h2 = int(hashlib.sha256(item.encode()).hexdigest(), 16)
        return [(h1 + i * h2) % self.size for i in range(self.hash_count)]

    def add(self, item: str) -> None:
        """要素を追加"""
        for pos in self._hashes(item):
            self.bit_array[pos] = 1
        self.item_count += 1

    def contains(self, item: str) -> bool:
        """要素が含まれるか判定（偽陽性あり、偽陰性なし）"""
        return all(self.bit_array[pos] == 1 for pos in self._hashes(item))

    def false_positive_rate(self) -> float:
        """現在の理論上の偽陽性率を計算"""
        n = self.item_count
        m = self.size
        k = self.hash_count
        return (1 - math.exp(-k * n / m)) ** k

    def memory_bytes(self) -> int:
        """おおよそのメモリ使用量（バイト）"""
        return self.size // 8 + 1


# --- 使用例 ---
if __name__ == "__main__":
    bf = BloomFilter(expected_items=10000, fp_rate=0.01)
    print(f"ビット配列サイズ: {bf.size:,} bits ({bf.memory_bytes():,} bytes)")
    print(f"ハッシュ関数の個数: {bf.hash_count}")

    # 単語リストを追加
    words = [f"word_{i}" for i in range(10000)]
    for w in words:
        bf.add(w)

    # 追加済み要素の検索（偽陰性は発生しない）
    fn_count = sum(1 for w in words if not bf.contains(w))
    print(f"\n偽陰性数: {fn_count} (常に 0)")

    # 未追加要素の検索（偽陽性の測定）
    test_words = [f"test_{i}" for i in range(10000)]
    fp_count = sum(1 for w in test_words if bf.contains(w))
    print(f"偽陽性数: {fp_count} / {len(test_words)}")
    print(f"実測偽陽性率: {fp_count / len(test_words):.4f}")
    print(f"理論偽陽性率: {bf.false_positive_rate():.4f}")

    # メモリ比較
    actual_set = set(words)
    import sys
    set_size = sys.getsizeof(actual_set)
    print(f"\n--- メモリ比較 ---")
    print(f"Bloom Filter: {bf.memory_bytes():,} bytes")
    print(f"Python set:   {set_size:,} bytes")
    print(f"削減率: {(1 - bf.memory_bytes() / set_size) * 100:.1f}%")

ビット配列サイズ: 95,851 bits (11,982 bytes)
ハッシュ関数の個数: 7
偽陰性数: 0 (常に 0)
偽陽性数: 107 / 10000
実測偽陽性率: 0.0107
理論偽陽性率: 0.0101

--- メモリ比較 ---
Bloom Filter: 11,982 bytes
Python set:   524,504 bytes
削減率: 97.7%

import hashlib
import math


class HyperLogLog:
    """HyperLogLog のスクラッチ実装"""

    def __init__(self, precision: int = 14):
        """
        :param precision: レジスタ数を決定する精度パラメータ p
                          レジスタ数 m = 2^p（デフォルト p=14 → 16384 レジスタ）
        """
        self.p = precision
        self.m = 1 << precision  # 2^p
        self.registers = [0] * self.m
        self.alpha = self._compute_alpha(self.m)

    @staticmethod
    def _compute_alpha(m: int) -> float:
        """バイアス補正係数を計算"""
        if m == 16:
            return 0.673
        elif m == 32:
            return 0.697
        elif m == 64:
            return 0.709
        else:
            return 0.7213 / (1 + 1.079 / m)

    def _hash(self, item: str) -> int:
        """要素を 64 ビットハッシュ値に変換"""
        h = hashlib.sha256(item.encode()).hexdigest()
        return int(h[:16], 16)  # 先頭 64 ビットを使用

    @staticmethod
    def _leading_zeros(value: int, max_bits: int) -> int:
        """ハッシュ値の先頭ゼロビット数 + 1 を返す"""
        if value == 0:
            return max_bits + 1
        count = 1
        for i in range(max_bits - 1, -1, -1):
            if value & (1 << i):
                break
            count += 1
        return count

    def add(self, item: str) -> None:
        """要素を追加"""
        h = self._hash(item)
        # 先頭 p ビットでレジスタのインデックスを決定
        j = h >> (64 - self.p)
        # 残りのビットで先頭ゼロ数を計算
        remaining = h & ((1 << (64 - self.p)) - 1)
        rank = self._leading_zeros(remaining, 64 - self.p)
        self.registers[j] = max(self.registers[j], rank)

    def count(self) -> int:
        """カーディナリティ（ユニーク要素数）を推定"""
        # 調和平均による推定
        indicator = sum(2.0 ** (-r) for r in self.registers)
        estimate = self.alpha * self.m * self.m / indicator

        # 小さな値の補正（Linear Counting）
        if estimate <= 2.5 * self.m:
            zeros = self.registers.count(0)
            if zeros > 0:
                estimate = self.m * math.log(self.m / zeros)

        # 大きな値の補正
        if estimate > (1 << 32) / 30.0:
            estimate = -(1 << 32) * math.log(1 - estimate / (1 << 32))

        return int(estimate)

    def merge(self, other: "HyperLogLog") -> "HyperLogLog":
        """2 つの HyperLogLog を統合"""
        if self.p != other.p:
            raise ValueError("精度パラメータが一致しません")
        merged = HyperLogLog(self.p)
        merged.registers = [
            max(a, b) for a, b in zip(self.registers, other.registers)
        ]
        return merged

    def standard_error(self) -> float:
        """標準誤差を返す"""
        return 1.04 / math.sqrt(self.m)

    def memory_bytes(self) -> int:
        """おおよそのメモリ使用量（バイト）"""
        # 各レジスタは最大 6 ビットで十分（値は最大 64-p+1）
        return self.m * 6 // 8


# --- 使用例 ---
if __name__ == "__main__":
    hll = HyperLogLog(precision=14)
    true_count = 1_000_000

    for i in range(true_count):
        hll.add(f"user_{i}")

    estimated = hll.count()
    error = abs(estimated - true_count) / true_count * 100

    print(f"真のユニーク数: {true_count:,}")
    print(f"推定ユニーク数: {estimated:,}")
    print(f"誤差: {error:.2f}%")
    print(f"理論標準誤差: {hll.standard_error() * 100:.2f}%")

    # メモリ比較
    import sys
    actual_set = {f"user_{i}" for i in range(true_count)}
    set_size = sys.getsizeof(actual_set)
    hll_size = hll.memory_bytes()
    print(f"\n--- メモリ比較 ---")
    print(f"HyperLogLog: {hll_size:,} bytes")
    print(f"Python set:  {set_size:,} bytes")
    print(f"削減率: {(1 - hll_size / set_size) * 100:.1f}%")

    # merge のデモ
    hll_a = HyperLogLog(precision=14)
    hll_b = HyperLogLog(precision=14)
    for i in range(500_000):
        hll_a.add(f"user_{i}")
    for i in range(300_000, 800_000):
        hll_b.add(f"user_{i}")
    merged = hll_a.merge(hll_b)
    print(f"\n--- merge デモ ---")
    print(f"HLL_A のユニーク数: {hll_a.count():,} (真値: 500,000)")
    print(f"HLL_B のユニーク数: {hll_b.count():,} (真値: 500,000)")
    print(f"merge 後のユニーク数: {merged.count():,} (真値: 800,000)")

真のユニーク数: 1,000,000
推定ユニーク数: 1,007,822
誤差: 0.78%
理論標準誤差: 0.81%

--- メモリ比較 ---
HyperLogLog: 12,288 bytes
Python set:  33,554,656 bytes
削減率: 99.96%

--- merge デモ ---
HLL_A のユニーク数: 498,653 (真値: 500,000)
HLL_B のユニーク数: 501,247 (真値: 500,000)
merge 後のユニーク数: 802,134 (真値: 800,000)