arm-core/api/src/domain/elimination_packing.py

"""Упаковка встречных строк сводной ОСВ внутри одного правила элиминации (§5).

Счета уже отобраны пресетом. Функция не читает БД и не меняет сумму элиминации: она только группирует строки.
"""

import uuid
from dataclasses import dataclass
from decimal import Decimal
from fractions import Fraction
from typing import TypeAlias

RowKey: TypeAlias = tuple[uuid.UUID, ...]
MAX_SIDE_ROWS = 5
MIN_GROUP_SIZE = 2
MAX_DEVIATION_PERCENT = Decimal("5")
ZERO = Decimal("0")


@dataclass(frozen=True)
class PackRow:
    """Одна неделимая строка свода на дебетовой или кредитовой стороне."""

    key: RowKey
    owner_id: uuid.UUID
    counterparty_id: uuid.UUID
    amount: Decimal  # сумма стороны со знаком; в упаковку идут только строки с суммой > 0


@dataclass(frozen=True)
class KnapsackChoice:
    """Лучшие подмножества заданного размера; при ничьей возвращаем ключи всех затронутых строк."""

    ratio: Fraction  # расхождение к меньшей сумме, точной дробью: ничьи по проценту не зависят от округления
    unique_rows: tuple[PackRow, ...] | None
    involved_keys: frozenset[RowKey]


@dataclass(frozen=True)
class PackedGroup:
    debit_rows: tuple[PackRow, ...]
    credit_rows: tuple[PackRow, ...]
    debit_amount: Decimal
    credit_amount: Decimal
    deviation_amount: Decimal
    deviation_percent: Decimal


@dataclass(frozen=True)
class PackingResult:
    groups: tuple[PackedGroup, ...]
    used_keys: frozenset[RowKey]
    ambiguous_keys: frozenset[RowKey]


@dataclass(frozen=True)
class RuleTotals:
    """Итог тройки до упаковки строк; упаковка его не меняет."""

    amount_a: Decimal
    amount_b: Decimal
    deviation_amount: Decimal
    deviation_percent: Decimal | None
    eliminated_amount: Decimal
    exceeds_threshold: bool


def calculate_rule_totals(amount_a: Decimal, amount_b: Decimal, *, manual: bool) -> RuleTotals:
    """Правило 5 % для всей тройки, независимо от сопоставления строк; меньшая сторона ≤ 0 — элиминации нет."""
    deviation = abs(amount_a - amount_b)
    base = min(amount_a, amount_b)
    percent = deviation / base * Decimal("100") if base > ZERO else None
    exceeds_threshold = deviation > ZERO if percent is None else percent > MAX_DEVIATION_PERCENT
    eliminated = base if base > ZERO and not manual and not exceeds_threshold else ZERO
    return RuleTotals(amount_a, amount_b, deviation, percent, eliminated, exceeds_threshold)


def _ratio(first: Decimal, second: Decimal) -> Fraction:
    """Расхождение к меньшей из двух положительных сумм."""
    return Fraction(abs(first - second)) / Fraction(min(first, second))


@dataclass(frozen=True)
class _Candidate:
    ratio: Fraction
    involved_keys: frozenset[RowKey]
    group: PackedGroup | None  # None — несколько одинаково лучших подмножеств у одного якоря


def best_subsets(rows: list[PackRow], *, size: int, capacity: Decimal) -> KnapsackChoice | None:
    """Перебрать группы заданного размера; битовая маска отмечает выбранные строки и ничьи."""
    if not 1 <= size <= MAX_SIDE_ROWS:
        raise ValueError("размер стороны должен быть от 1 до 5")
    candidates = sorted((row for row in rows if row.amount > ZERO), key=lambda row: row.amount)
    if capacity <= ZERO or len(candidates) < size:
        return None
    hundred_percent = Decimal("100")
    limit = hundred_percent + MAX_DEVIATION_PERCENT
    upper = capacity * limit / hundred_percent
    combinations: list[tuple[int, Decimal]] = []  # (битовая маска выбранных строк, точная сумма)

    def search(start: int, selected: int, total: Decimal, mask: int) -> None:
        remaining = size - selected
        if remaining == 0:
            deviation = abs(capacity - total)
            if deviation * hundred_percent <= MAX_DEVIATION_PERCENT * min(capacity, total):
                combinations.append((mask, total))
            return

        if len(candidates) - start < remaining:
            return

        # Промежуточную сумму сравниваем с нижней границей лишь с учётом оставшихся строк.
        largest_total = total + sum((row.amount for row in candidates[-remaining:]), ZERO)
        if largest_total * limit < capacity * hundred_percent:
            return
        smallest_total = total + sum((row.amount for row in candidates[start : start + remaining]), ZERO)
        if smallest_total > upper:
            return

        for index in range(start, len(candidates) - remaining + 1):
            next_total = total + candidates[index].amount
            if next_total > upper:
                break  # суммы отсортированы: следующие строки только больше
            search(index + 1, selected + 1, next_total, mask | (1 << index))

    search(0, 0, ZERO, 0)
    if not combinations:
        return None
    # точные первыми, затем по возрастанию процента расхождения к меньшей сумме
    ratios = [(mask, _ratio(capacity, total)) for mask, total in combinations]
    best_ratio = min(ratio for _, ratio in ratios)
    best_masks = [mask for mask, ratio in ratios if ratio == best_ratio]
    involved_mask = 0
    for mask in best_masks:
        involved_mask |= mask
    involved_keys = frozenset(row.key for index, row in enumerate(candidates) if involved_mask & (1 << index))
    unique_rows = (
        tuple(row for index, row in enumerate(candidates) if best_masks[0] & (1 << index))
        if len(best_masks) == 1
        else None
    )
    return KnapsackChoice(best_ratio, unique_rows, involved_keys)


def _candidate(anchor: PackRow, choice: KnapsackChoice, *, anchor_is_debit: bool) -> _Candidate:
    keys = choice.involved_keys | {anchor.key}
    if choice.unique_rows is None:
        return _Candidate(choice.ratio, keys, None)
    debit = (anchor,) if anchor_is_debit else choice.unique_rows
    credit = choice.unique_rows if anchor_is_debit else (anchor,)
    debit_amount = sum((row.amount for row in debit), ZERO)
    credit_amount = sum((row.amount for row in credit), ZERO)
    group = PackedGroup(
        debit_rows=debit,
        credit_rows=credit,
        debit_amount=debit_amount,
        credit_amount=credit_amount,
        deviation_amount=abs(debit_amount - credit_amount),
        deviation_percent=abs(debit_amount - credit_amount) / min(debit_amount, credit_amount) * 100,
    )
    return _Candidate(choice.ratio, keys, group)


def _candidates(
    debit_rows: list[PackRow], credit_rows: list[PackRow], unavailable: set[RowKey], total_size: int
) -> list[_Candidate]:
    """Входы уже ограничены одной тройкой: правило, компания А, компания Б; строки с суммой ≤ 0 не участвуют."""
    available_debit = [row for row in debit_rows if row.key not in unavailable and row.amount > ZERO]
    available_credit = [row for row in credit_rows if row.key not in unavailable and row.amount > ZERO]
    candidates: list[_Candidate] = []
    for anchor in available_debit:
        choice = best_subsets(available_credit, size=total_size - 1, capacity=anchor.amount)
        if choice is not None:
            candidates.append(_candidate(anchor, choice, anchor_is_debit=True))
    if total_size > MIN_GROUP_SIZE:
        for anchor in available_credit:
            choice = best_subsets(available_debit, size=total_size - 1, capacity=anchor.amount)
            if choice is not None:
                candidates.append(_candidate(anchor, choice, anchor_is_debit=False))
    return candidates


def pack_rule_rows(debit_rows: list[PackRow], credit_rows: list[PackRow]) -> PackingResult:
    """Жадно упаковать 1:1, 1:N, N:1; ничьи по размеру и проценту расхождения отправить на ручной разбор."""
    used: set[RowKey] = set()
    ambiguous: set[RowKey] = set()
    groups: list[PackedGroup] = []
    for total_size in range(MIN_GROUP_SIZE, MAX_SIDE_ROWS + MIN_GROUP_SIZE):
        while candidates := _candidates(debit_rows, credit_rows, used | ambiguous, total_size):
            best_ratio = min(candidate.ratio for candidate in candidates)
            best = [candidate for candidate in candidates if candidate.ratio == best_ratio]
            conflicted = {index for index, candidate in enumerate(best) if candidate.group is None}
            for left in range(len(best)):
                for right in range(left + 1, len(best)):
                    if best[left].involved_keys & best[right].involved_keys:
                        conflicted.update((left, right))
            for index, candidate in enumerate(best):
                if index in conflicted:
                    ambiguous.update(candidate.involved_keys)
                else:
                    assert candidate.group is not None
                    groups.append(candidate.group)
                    used.update(candidate.involved_keys)
    return PackingResult(tuple(groups), frozenset(used), frozenset(ambiguous))