Random hyperplane LSH in Python
LSH for angular distance that hashes each vector by the signs of its dot products with random hyperplanes.
- File
random_hyperplane_lsh.py, 67 lines- Level
- I0 not checked yet
- Toolchain
- Python 3.14, pytest 9
- Test framework
- pytest
- Run the tests
python3 -m pytest
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/random-hyperplane-lsh/. The authors are listed in prov/7MZBA9.jsonl.
import math
import random
from typing import Optional, Sequence
Vector = Sequence[float]
def dot(a: Vector, b: Vector) -> float:
return sum(x * y for x, y in zip(a, b))
def angle(a: Vector, b: Vector) -> float:
"""The angle between two nonzero vectors, in radians, in [0, pi]."""
c = dot(a, b) / (math.sqrt(dot(a, a)) * math.sqrt(dot(b, b)))
return math.acos(max(-1.0, min(1.0, c)))
def hash_key(x: Vector, hyperplanes: Sequence[Vector]) -> tuple[int, ...]:
"""The k-bit key of x: bit i is 1 when hyperplanes[i] . x >= 0. Cost O(dk)."""
return tuple(1 if dot(r, x) >= 0 else 0 for r in hyperplanes) # @line L1
def parameters(n: int, theta1: float, theta2: float) -> tuple[int, int]:
"""k = ceil(log_{1/p2} n) and L = ceil(n^rho / p1), for angles in radians."""
p1, p2 = 1 - theta1 / math.pi, 1 - theta2 / math.pi
rho = math.log(1 / p1) / math.log(1 / p2)
k = max(1, math.ceil(math.log(n) / math.log(1 / p2)))
return k, math.ceil(n**rho / p1)
class RandomHyperplaneLSH:
"""L hash tables, each keyed by the signs of k random Gaussian hyperplanes."""
def __init__(self, points: Sequence[Vector], k: int, num_tables: int, seed: int = 0) -> None:
if any(dot(p, p) == 0 for p in points):
raise ValueError("zero vectors have no angle")
self.points = [tuple(p) for p in points]
d = len(self.points[0]) if self.points else 0
rng = random.Random(seed)
self.planes: list[list[tuple[float, ...]]] = []
self.tables: list[dict[tuple[int, ...], list[int]]] = []
for _ in range(num_tables): # @line L2
planes = [tuple(rng.gauss(0, 1) for _ in range(d)) for _ in range(k)] # @line L3
table: dict[tuple[int, ...], list[int]] = {}
for i, x in enumerate(self.points): # @line L4
table.setdefault(hash_key(x, planes), []).append(i)
self.planes.append(planes)
self.tables.append(table)
def query(self, q: Vector, theta2: float) -> Optional[int]:
"""Return the index of a point within angle theta2 of q, or None.
Stops after 3L rejected candidates. Cost O(dkL) plus at most 3L + 1 angles.
"""
num_tables = len(self.tables)
c = 0 # @line L5
for planes, table in zip(self.planes, self.tables): # @line L6
for i in table.get(hash_key(q, planes), []): # @line L7
if angle(q, self.points[i]) <= theta2: # @line L8
return i
c += 1 # @line L9
if c >= 3 * num_tables: # @line L10
return None
return None # @line L11
test_random_hyperplane_lsh.py 79 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/random-hyperplane-lsh/. The authors are listed in prov/7MZBA9.jsonl.
import math
import random
from random_hyperplane_lsh import RandomHyperplaneLSH, angle, hash_key, parameters
R = [(0.5, 1), (-1, 0.3), (0.2, -1), (1, -0.4)]
def test_example_keys() -> None:
assert hash_key((1, 0), R) == (1, 0, 1, 1)
assert hash_key((1, 1), R) == (1, 0, 0, 1)
assert hash_key((0.9, 0.3), R) == (1, 0, 0, 1)
assert hash_key((-1, 0.2), R) == (0, 1, 0, 0)
assert abs(math.degrees(angle((1, 0), (1, 1))) - 45) < 1e-9
def test_example_parameters() -> None:
assert parameters(10**6, math.pi / 6, math.pi / 3) == (35, 599)
def test_scale_invariant() -> None:
rng = random.Random(3)
planes = [tuple(rng.gauss(0, 1) for _ in range(5)) for _ in range(16)]
x = tuple(rng.uniform(-1, 1) for _ in range(5))
assert hash_key(x, planes) == hash_key(tuple(2.5 * v for v in x), planes)
def test_collision_probability() -> None:
rng = random.Random(7)
for x, y in [((1, 0, 0), (1, 1, 0)), ((1, 0, 0), (-1, 0.2, 0)), ((1, 2, 3), (3, -1, 0.5))]:
trials = 20000
same = sum(
hash_key(x, [r]) == hash_key(y, [r])
for r in (tuple(rng.gauss(0, 1) for _ in range(3)) for _ in range(trials))
)
assert abs(same / trials - (1 - angle(x, y) / math.pi)) < 0.02
def test_edges() -> None:
empty = RandomHyperplaneLSH([], k=3, num_tables=2)
assert empty.query((1, 0), 1.0) is None
one = RandomHyperplaneLSH([(1, 0)], k=3, num_tables=2)
assert one.query((2, 0), 0.1) == 0
assert one.query((-1, 0), 1.0) is None
try:
RandomHyperplaneLSH([(0, 0)], k=1, num_tables=1)
raise AssertionError("zero vector accepted")
except ValueError:
pass
def test_random_against_brute_force() -> None:
rng = random.Random(2026)
theta1, theta2 = math.pi / 6, math.pi / 3
successes = trials = 0
for t in range(600):
n = rng.randint(1, 60)
d = rng.randint(2, 6)
pts = []
while len(pts) < n:
p = tuple(rng.gauss(0, 1) for _ in range(d))
if any(p):
pts.append(p)
k, num_tables = parameters(max(n, 2), theta1, theta2)
index = RandomHyperplaneLSH(pts, k, num_tables, seed=t)
q = tuple(rng.gauss(0, 1) for _ in range(d))
ans = index.query(q, theta2)
if ans is not None:
assert angle(q, pts[ans]) <= theta2
if all(angle(q, p) > theta2 for p in pts):
assert ans is None
if any(angle(q, p) <= theta1 for p in pts):
trials += 1
successes += ans is not None
assert trials > 50
assert successes / trials >= 1 - 1 / math.e - 1 / 3
What the next level needs
- the tests did not run in this build