Algorithm Search⌘K

Multi-vantage-point tree in Python

A vantage-point tree that uses several vantage points per node to reduce distance computations.

File
mvp_tree.py, 155 lines
Level
I0 not checked yet
Toolchain
Python 3.14, pytest 9
Test framework
pytest
Run the tests
python3 -m pytest
mvp_tree.pyRawGitHub
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/mvp-tree/. The authors are listed in prov/WR0RGE.jsonl.

import random
from dataclasses import dataclass, field
from typing import Callable, Generic, Optional, Sequence, TypeVar

T = TypeVar("T")
Metric = Callable[[T, T], float]


@dataclass
class Entry(Generic[T]):
    obj: T
    path: list[float]  # distances to the first p vantage points above it
    d1: float = 0.0
    d2: float = 0.0


@dataclass
class Leaf(Generic[T]):
    v1: Optional[T]
    v2: Optional[T]
    entries: list[Entry[T]]


@dataclass
class Internal(Generic[T]):
    v1: T
    v2: Optional[T]
    # children[i][j] with the interval of d(., v1) of group i and of d(., v2) of subgroup j
    ring1: list[tuple[float, float]]
    ring2: list[list[tuple[float, float]]]
    children: list[list[object]] = field(default_factory=list)


def _split(entries: list[Entry[T]], key: Callable[[Entry[T]], float], m: int) -> list[list[Entry[T]]]:
    """Split by rank of key into m groups of sizes as equal as possible (some may be empty)."""
    s = sorted(entries, key=key)
    n = len(s)
    return [s[(i * n) // m : ((i + 1) * n) // m] for i in range(m)]


def _interval(group: list[Entry[T]], key: Callable[[Entry[T]], float]) -> tuple[float, float]:
    if not group:
        return (1.0, 0.0)  # empty interval, never met
    vals = [key(e) for e in group]
    return (min(vals), max(vals))


class MVPTree(Generic[T]):
    """A multi-vantage-point tree with fan-out m*m, leaf capacity k and p path distances per object.

    choose(n) gives the index of the first vantage point among n candidates; the default is random.
    The second vantage point is the object farthest from the first. evaluations counts metric calls.
    """

    def __init__(self, points: Sequence[T], d: Metric, m: int = 2, k: int = 4, p: int = 4, seed: int = 0,
                 choose: Optional[Callable[[int], int]] = None) -> None:
        if m < 2 or k < 1 or p < 0:
            raise ValueError("need m >= 2, k >= 1, p >= 0")
        self._d, self.m, self.k, self.p = d, m, k, p
        self.evaluations = 0
        rng = random.Random(seed)
        self._choose = choose or rng.randrange
        self.root = self._build([Entry(x, []) for x in points])

    def dist(self, a: T, b: T) -> float:
        self.evaluations += 1
        return self._d(a, b)

    def _push(self, e: Entry[T], value: float) -> None:
        if len(e.path) < self.p:
            e.path.append(value)

    def _vantage_points(self, s: list[Entry[T]]) -> tuple[Optional[Entry[T]], Optional[Entry[T]], list[Entry[T]]]:
        """v1 at random; d1 for the others; v2 the farthest from v1; d2 for the rest."""
        if not s:
            return None, None, []
        v1 = s.pop(self._choose(len(s)))
        for e in s:
            e.d1 = self.dist(e.obj, v1.obj)
        if not s:
            return v1, None, []
        j = max(range(len(s)), key=lambda i: s[i].d1)
        v2 = s.pop(j)
        for e in s:
            e.d2 = self.dist(e.obj, v2.obj)
        return v1, v2, s

    def _build(self, s: list[Entry[T]]):
        if len(s) <= self.k + 2:  # @line L1
            v1, v2, rest = self._vantage_points(s)  # @line L2
            return Leaf(v1.obj if v1 else None, v2.obj if v2 else None, rest)
        v1 = s.pop(self._choose(len(s)))  # @line L3
        for e in s:
            e.d1 = self.dist(e.obj, v1.obj)
            self._push(e, e.d1)
        groups = _split(s, lambda e: e.d1, self.m)  # @line L4
        ring1 = [_interval(g, lambda e: e.d1) for g in groups]
        gi = max(range(self.m), key=lambda i: len(groups[i]))
        j = max(range(len(groups[gi])), key=lambda i: groups[gi][i].d1)
        v2 = groups[gi].pop(j)  # @line L5
        ring2: list[list[tuple[float, float]]] = []
        children: list[list[object]] = []
        for g in groups:
            for e in g:
                e.d2 = self.dist(e.obj, v2.obj)
                self._push(e, e.d2)
            sub = _split(g, lambda e: e.d2, self.m)  # @line L6
            ring2.append([_interval(h, lambda e: e.d2) for h in sub])
            children.append([self._build(h) for h in sub])  # @line L7
        return Internal(v1.obj, v2.obj, ring1, ring2, children)

    def range(self, q: T, r: float) -> list[T]:
        """All stored objects x with d(q, x) <= r."""
        out: list[T] = []
        self._range(self.root, q, r, [], out)
        return out

    def _range(self, node, q: T, r: float, qpath: list[float], out: list[T]) -> None:
        if isinstance(node, Leaf):  # @line L8
            if node.v1 is None:
                return
            t1 = self.dist(q, node.v1)  # @line L9
            if t1 <= r:
                out.append(node.v1)
            if node.v2 is None:
                return
            t2 = self.dist(q, node.v2)
            if t2 <= r:
                out.append(node.v2)
            for e in node.entries:  # @line L10
                if abs(t1 - e.d1) > r or abs(t2 - e.d2) > r:  # @line L11
                    continue
                if any(abs(a - b) > r for a, b in zip(qpath, e.path)):  # @line L12
                    continue
                if self.dist(q, e.obj) <= r:  # @line L13
                    out.append(e.obj)
            return
        t1 = self.dist(q, node.v1)  # @line L14
        t2 = self.dist(q, node.v2)
        for t, v in ((t1, node.v1), (t2, node.v2)):
            if t <= r:
                out.append(v)
        path = (qpath + [t1, t2])[: self.p]  # @line L15
        for i in range(self.m):  # @line L16
            lo, hi = node.ring1[i]
            if lo > hi or t1 + r < lo or t1 - r > hi:
                continue
            for j in range(self.m):  # @line L17
                lo2, hi2 = node.ring2[i][j]
                if lo2 > hi2 or t2 + r < lo2 or t2 - r > hi2:
                    continue
                self._range(node.children[i][j], q, r, path, out)  # @line L18
test_mvp_tree.py 74 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/mvp-tree/. The authors are listed in prov/WR0RGE.jsonl.

import math
import random

from mvp_tree import Internal, Leaf, MVPTree

P = [(1, 1), (2, 6), (4, 4), (5, 1), (6, 7), (8, 3), (9, 8), (3, 9), (7, 5), (0, 4), (10, 2), (6, 2)]
N = dict(zip("abcdefghijkl", P))


def depth(node) -> int:
    if isinstance(node, Leaf):
        return 1
    return 1 + max(depth(c) for row in node.children for c in row)


def edit(a: str, b: str) -> int:
    prev = list(range(len(b) + 1))
    for i, ca in enumerate(a, 1):
        cur = [i]
        for j, cb in enumerate(b, 1):
            cur.append(min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + (ca != cb)))
        prev = cur
    return prev[-1]


def test_example() -> None:
    t = MVPTree(P, math.dist, m=2, k=2, p=2, choose=lambda n: 0)
    assert t.evaluations == 29
    root = t.root
    assert isinstance(root, Internal) and root.v1 == N["a"] and root.v2 == N["g"]
    leaf = root.children[0][1]
    assert {leaf.v1, leaf.v2} == {N["b"], N["d"]}
    (j,) = leaf.entries
    assert j.obj == N["j"] and [round(x, 4) for x in j.path] == [3.1623, 9.8489]
    t.evaluations = 0
    assert sorted(t.range((5, 5), 2)) == sorted([N["c"], N["i"]])
    assert t.evaluations == 8


def test_edges() -> None:
    assert MVPTree([], math.dist).range((0, 0), 1) == []
    assert MVPTree([(1, 1)], math.dist).range((0, 0), 2) == [(1, 1)]
    same = MVPTree([(2, 2)] * 20, math.dist, k=1, p=3)
    assert len(same.range((2, 2), 0)) == 20


def test_height() -> None:
    for n in [1, 5, 16, 64, 300]:
        for m in [2, 3]:
            t = MVPTree([(i, 0) for i in range(n)], math.dist, m=m, k=1, p=2, seed=n)
            assert depth(t.root) <= math.ceil(math.log(max(n, 2), m * m)) + 1
            assert t.evaluations <= 2 * n * depth(t.root)


def test_random_against_scan() -> None:
    rng = random.Random(1997)
    for trial in range(500):
        m, k, p = rng.choice([2, 3]), rng.randint(1, 5), rng.randint(0, 6)
        if trial % 2:
            dim = rng.choice([1, 2, 3])
            pts = [tuple(rng.randint(0, 9) for _ in range(dim)) for _ in range(rng.randint(0, 60))]
            d = math.dist
            q = tuple(rng.uniform(-1, 10) for _ in range(dim))
            r = rng.uniform(0, 5)
        else:
            pts = ["".join(rng.choice("abc") for _ in range(rng.randint(0, 5))) for _ in range(rng.randint(0, 40))]
            d = edit
            q = "".join(rng.choice("abc") for _ in range(rng.randint(0, 5)))
            r = rng.randint(0, 3)
        t = MVPTree(pts, d, m=m, k=k, p=p, seed=trial)
        assert sorted(t.range(q, r)) == sorted(x for x in pts if d(q, x) <= r)

What the next level needs

  • the tests did not run in this build

Read the full page about Multi-vantage-point tree