176 lines
7.9 KiB
Python
176 lines
7.9 KiB
Python
"""Упаковка встречных строк сводной ОСВ внутри одного правила элиминации (§5).
|
||
|
||
Счета уже отобраны пресетом. Функция не читает БД и не меняет сумму элиминации: она только группирует строки.
|
||
"""
|
||
|
||
import uuid
|
||
from dataclasses import dataclass
|
||
from decimal import Decimal
|
||
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 # модуль суммы соответствующей стороны
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class KnapsackChoice:
|
||
"""Лучшие подмножества заданного размера; при ничьей возвращаем ключи всех затронутых строк."""
|
||
|
||
deviation: Decimal
|
||
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 _Candidate:
|
||
deviation: Decimal
|
||
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
|
||
combinations.sort(key=lambda item: abs(capacity - item[1])) # точные первыми, затем меньшие расхождения
|
||
best_deviation = abs(capacity - combinations[0][1])
|
||
best_masks = [mask for mask, total in combinations if abs(capacity - total) == best_deviation]
|
||
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_deviation, 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.deviation, 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=choice.deviation,
|
||
deviation_percent=choice.deviation / min(debit_amount, credit_amount) * 100,
|
||
)
|
||
return _Candidate(choice.deviation, keys, group)
|
||
|
||
|
||
def _candidates(
|
||
debit_rows: list[PackRow], credit_rows: list[PackRow], unavailable: set[RowKey], total_size: int
|
||
) -> list[_Candidate]:
|
||
"""Входы уже ограничены одной тройкой: правило, компания А, компания Б."""
|
||
available_debit = [row for row in debit_rows if row.key not in unavailable]
|
||
available_credit = [row for row in credit_rows if row.key not in unavailable]
|
||
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_deviation = min(candidate.deviation for candidate in candidates)
|
||
best = [candidate for candidate in candidates if candidate.deviation == best_deviation]
|
||
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))
|