Algorithm Search⌘K

Amir-Lewenstein-Porat algorithm in Python

Solves k-mismatch matching in O(n (k log k)^(1/2)) time by treating frequent and rare pattern symbols differently.

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

"""The Amir-Lewenstein-Porat algorithm for the k-mismatch problem.

Edit model: Hamming distance (substitutions only). Indices are 0-based.
The result is the sorted list of start positions i in 0..len(text)-len(pattern)
such that pattern and text[i:i+len(pattern)] differ in at most k places.

Symbols that occur at least f = ceil(sqrt(k * ceil(log2 m))) times in the
pattern are frequent. With many frequent symbols, 2k marks of the pattern
filter the alignments and a kangaroo check with longest common extensions
verifies the few that remain. With few frequent symbols, one convolution per
frequent symbol and direct counting for the other symbols give every
Hamming distance. Convolutions use a number-theoretic transform modulo the
prime 998244353, exact because every count is at most m. All loops are
iterative.
"""

from math import isqrt

MOD = 998244353
ROOT = 3  # a primitive root modulo MOD


def ntt(a: list[int], invert: bool) -> None:
    """Transform a in place; len(a) is a power of two that divides 2**23."""
    n = len(a)
    j = 0
    for i in range(1, n):
        bit = n >> 1
        while j & bit:
            j ^= bit
            bit >>= 1
        j |= bit
        if i < j:
            a[i], a[j] = a[j], a[i]
    length = 2
    while length <= n:
        w = pow(ROOT, (MOD - 1) // length, MOD)
        if invert:
            w = pow(w, MOD - 2, MOD)
        half = length >> 1
        for start in range(0, n, length):
            wn = 1
            for k in range(start, start + half):
                u = a[k]
                v = a[k + half] * wn % MOD
                a[k] = (u + v) % MOD
                a[k + half] = (u - v) % MOD
                wn = wn * w % MOD
        length <<= 1
    if invert:
        inv_n = pow(n, MOD - 2, MOD)
        for i in range(n):
            a[i] = a[i] * inv_n % MOD


class LCE:
    """Longest common extension of a suffix of a and a suffix of b."""

    def __init__(self, a: str, b: str) -> None:
        syms = sorted(set(a) | set(b))
        code = {c: r + 2 for r, c in enumerate(syms)}
        s = [code[c] for c in a] + [1] + [code[c] for c in b] + [0]
        self.off = len(a) + 1
        n = len(s)
        rank = s[:]
        sa = list(range(n))
        h = 1
        while True:  # prefix doubling: sort by (rank[i], rank[i + h])
            key = [rank[i] * (n + 2) + (rank[i + h] + 1 if i + h < n else 0) for i in range(n)]
            sa.sort(key=key.__getitem__)
            new = [0] * n
            for t in range(1, n):
                new[sa[t]] = new[sa[t - 1]] + (key[sa[t]] != key[sa[t - 1]])
            rank = new
            if rank[sa[-1]] == n - 1:
                break
            h *= 2
        self.rank = rank
        lcp = [0] * n  # lcp[t] = lcp of suffixes sa[t - 1] and sa[t]
        q = 0
        for i in range(n):
            if rank[i] == 0:
                q = 0
                continue
            j = sa[rank[i] - 1]
            while i + q < n and j + q < n and s[i + q] == s[j + q]:
                q += 1
            lcp[rank[i]] = q
            if q:
                q -= 1
        self.table = [lcp]
        w = 1
        while 2 * w <= n:
            prev = self.table[-1]
            self.table.append([min(prev[t], prev[t + w]) for t in range(n - 2 * w + 1)])
            w *= 2

    def query(self, i: int, j: int) -> int:
        """Length of the longest common prefix of a[i:] and b[j:]."""
        x, y = self.rank[i], self.rank[self.off + j]
        if x > y:
            x, y = y, x
        x += 1  # minimum of lcp[x..y]
        lev = (y - x + 1).bit_length() - 1
        row = self.table[lev]
        return min(row[x], row[y - (1 << lev) + 1])


def parameters(m: int, k: int) -> tuple[int, int]:
    """Return (f, g): the frequency threshold and the number of mark symbols."""
    lam = (m - 1).bit_length() if m > 1 else 0  # ceil(log2 m)
    x = k * lam
    f = isqrt(x)
    if f * f < x:
        f += 1  # f = ceil(sqrt(k * lam))
    f = max(1, f)
    return f, -(-2 * k // f)  # g = ceil(2k / f)


def kangaroo(lce: LCE, m: int, i: int, k: int) -> bool:
    """Return True when the window at i has at most k mismatches."""
    q, e = 0, 0
    while q < m:
        q += lce.query(q, i + q)  # jump over a run of equal symbols
        if q < m:
            e += 1
            if e > k:
                return False
            q += 1
    return True


def correlate(pattern: str, text: str, c: str, out: list[int]) -> None:
    """Add to out[i] the number of q with pattern[q] = text[i + q] = c."""
    m, n = len(pattern), len(text)
    size = 1
    while size < 3 * m - 2:
        size <<= 1
    x = [0] * size
    for q in range(m):
        if pattern[m - 1 - q] == c:
            x[q] = 1
    ntt(x, False)
    for b in range(0, n - m + 1, m):  # text blocks of length 2m - 1
        y = [0] * size
        for j, t in enumerate(text[b:b + 2 * m - 1]):
            if t == c:
                y[j] = 1
        ntt(y, False)
        z = [u * v % MOD for u, v in zip(x, y)]
        ntt(z, True)
        for t in range(min(m, n - m + 1 - b)):
            out[b + t] += z[t + m - 1]


def alp_k_mismatch(pattern: str, text: str, k: int) -> list[int]:
    """Return the start positions with at most k mismatches.

    Precondition: k >= 0. Cost: O(n + n * sqrt(k log m)) time after the
    longest common extension structure, O(n + m) extra space.
    """
    if k < 0:
        raise ValueError("k must be at least 0")
    m, n = len(pattern), len(text)
    if m > n:
        return []
    if k >= m:  # @line L1
        return list(range(n - m + 1))
    f, g = parameters(m, k)  # @line L2
    occ: dict[str, list[int]] = {}
    for q, c in enumerate(pattern):
        occ.setdefault(c, []).append(q)
    freq = sorted(c for c in occ if len(occ[c]) >= f)
    if len(freq) >= g:  # @line L3
        marks = {c: occ[c][:f] for c in freq[:g]}  # @line L4
        count = [0] * (n - m + 1)
        for j, t in enumerate(text):  # @line L5
            for q in marks.get(t, ()):
                if 0 <= j - q <= n - m:
                    count[j - q] += 1
        lce = LCE(pattern, text)
        need = g * f - k
        return [i for i in range(n - m + 1) if count[i] >= need and kangaroo(lce, m, i, k)]  # @line L6
    match = [0] * (n - m + 1)  # @line L7
    for c in freq:  # @line L8
        correlate(pattern, text, c, match)
    frequent = set(freq)
    for j, t in enumerate(text):  # @line L9
        if t not in frequent:
            for q in occ.get(t, ()):
                if 0 <= j - q <= n - m:
                    match[j - q] += 1
    return [i for i in range(n - m + 1) if m - match[i] <= k]  # @line L10


def case_of(pattern: str, k: int) -> str:
    """Return "A" (marks and kangaroo) or "B" (convolutions), for tests."""
    m = len(pattern)
    f, g = parameters(m, k)
    counts: dict[str, int] = {}
    for c in pattern:
        counts[c] = counts.get(c, 0) + 1
    return "A" if sum(1 for v in counts.values() if v >= f) >= g else "B"
test_amir_lewenstein_porat_algorithm.py 71 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/amir-lewenstein-porat-algorithm/. The authors are listed in prov/EGHY1F.jsonl.

import random

import pytest
from amir_lewenstein_porat_algorithm import alp_k_mismatch, case_of, parameters


def brute(pattern: str, text: str, k: int) -> list[int]:
    m = len(pattern)
    return [
        i for i in range(len(text) - m + 1)
        if sum(pattern[q] != text[i + q] for q in range(m)) <= k
    ]


def test_example() -> None:
    assert alp_k_mismatch("ACGTA", "ACGAAACGTTACCTA", 1) == [0, 5, 10]
    assert alp_k_mismatch("abc", "abxabcbc", 1) == [0, 3, 5]
    assert alp_k_mismatch("abc", "abxabcbc", 0) == [3]


def test_edges() -> None:
    assert alp_k_mismatch("", "", 0) == [0]
    assert alp_k_mismatch("", "ab", 0) == [0, 1, 2]
    assert alp_k_mismatch("a", "", 0) == []
    assert alp_k_mismatch("ab", "ab", 5) == [0]
    assert alp_k_mismatch("aaaa", "aaaaaa", 0) == [0, 1, 2]
    with pytest.raises(ValueError):
        alp_k_mismatch("a", "a", -1)


def test_parameters() -> None:
    assert parameters(8, 1) == (2, 1)
    assert parameters(8, 4) == (4, 2)
    assert parameters(1, 0) == (1, 0)
    for m in range(1, 200):
        for k in range(0, m):
            f, g = parameters(m, k)
            assert f >= 1 and g * f >= 2 * k


@pytest.mark.parametrize("alphabet", ["ab", "acgt"])
def test_random_against_brute_force(alphabet: str) -> None:
    rng = random.Random(20241106 + len(alphabet))
    seen = {"A": 0, "B": 0}
    for _ in range(1500):
        n = rng.randint(0, 40)
        m = rng.randint(0, 14)
        text = "".join(rng.choice(alphabet) for _ in range(n))
        pattern = "".join(rng.choice(alphabet) for _ in range(m))
        k = rng.randint(0, m + 1)
        if m <= n and k < m:
            seen[case_of(pattern, k)] += 1
        assert alp_k_mismatch(pattern, text, k) == brute(pattern, text, k)
    assert seen["A"] > 50 and seen["B"] > 50


def test_case_b_long_patterns() -> None:
    rng = random.Random(7)
    for _ in range(300):
        m = rng.randint(10, 40)
        k = rng.randint(m // 2, m - 1)
        sigma = rng.choice(["ab", "acgt", "abcdefgh"])
        pattern = "".join(rng.choice(sigma) for _ in range(m))
        text = "".join(rng.choice(sigma) for _ in range(rng.randint(m, 3 * m + 5)))
        if rng.random() < 0.5:
            i = rng.randint(0, len(text) - m)
            text = text[:i] + pattern + text[i + m:]
        assert alp_k_mismatch(pattern, text, k) == brute(pattern, text, k)

What the next level needs

  • the tests did not run in this build

Read the full page about Amir-Lewenstein-Porat algorithm