получение разнесение осв по правилам автоматическое + наложение ручных корректировок сверху

api/alembic/versions/0011_vgo_assignment.py

@@ -0,0 +1,60 @@
+"""Хранение назначения строк ОСВ правилам по версии консолидации."""
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+from alembic import op
+from src.models.base import SCHEMA
+
+revision = "0011_vgo_assignment"
+down_revision = "0010_dbml_v5_23"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "vgo_assignment",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
+        sa.Column("consolidation_version_id", postgresql.UUID(as_uuid=True), nullable=False),
+        sa.Column(
+            "statement_type",
+            postgresql.ENUM("balance", "income", name="statement_type", schema=SCHEMA, create_type=False),
+            nullable=False,
+        ),
+        sa.Column("osv_entry_id", postgresql.UUID(as_uuid=True), nullable=False),
+        sa.Column("rule_id", postgresql.UUID(as_uuid=True)),
+        sa.Column("action", sa.String(), nullable=False),
+        sa.Column("is_exclude", sa.Boolean(), server_default=sa.text("false"), nullable=False),
+        sa.Column("created_by", postgresql.UUID(as_uuid=True)),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated_by", postgresql.UUID(as_uuid=True)),
+        sa.Column("updated_at", sa.DateTime(timezone=True)),
+        sa.ForeignKeyConstraint(
+            ["consolidation_version_id"], [f"{SCHEMA}.consolidation_version.id"], name="fk_vgo_assignment_version"
+        ),
+        sa.ForeignKeyConstraint(["osv_entry_id"], [f"{SCHEMA}.osv_entry.id"], name="fk_vgo_assignment_entry"),
+        sa.ForeignKeyConstraint(["rule_id"], [f"{SCHEMA}.elimination_rule.id"], name="fk_vgo_assignment_rule"),
+        sa.ForeignKeyConstraint(["created_by"], [f"{SCHEMA}.app_user.id"], name="fk_vgo_assignment_created_by"),
+        sa.ForeignKeyConstraint(["updated_by"], [f"{SCHEMA}.app_user.id"], name="fk_vgo_assignment_updated_by"),
+        sa.CheckConstraint("action IN ('ASSIGN', 'UNASSIGNED')", name="ck_vgo_assignment_action_values"),
+        sa.CheckConstraint(
+            "(action = 'ASSIGN' AND rule_id IS NOT NULL) OR (action = 'UNASSIGNED' AND rule_id IS NULL)",
+            name="ck_vgo_assignment_action_rule_consistency",
+        ),
+        schema=SCHEMA,
+    )
+    op.create_index(
+        "uq_vgo_assignment_version_type_entry",
+        "vgo_assignment",
+        ["consolidation_version_id", "statement_type", "osv_entry_id"],
+        unique=True,
+        schema=SCHEMA,
+    )
+    op.create_index("ix_vgo_assignment_osv_entry_id", "vgo_assignment", ["osv_entry_id"], schema=SCHEMA)
+
+
+def downgrade() -> None:
+    op.drop_index("ix_vgo_assignment_osv_entry_id", table_name="vgo_assignment", schema=SCHEMA)
+    op.drop_index("uq_vgo_assignment_version_type_entry", table_name="vgo_assignment", schema=SCHEMA)
+    op.drop_table("vgo_assignment", schema=SCHEMA)

api/src/api/dependencies/consolidations.py

@@ -16,6 +16,7 @@ from src.services.consolidations import ConsolidationService
 from src.services.counterparty_mapping import CounterpartyMappingService
 from src.services.elimination_rules import EliminationRuleService
 from src.services.intercompany_matrix import IntercompanyMatrixService
+from src.services.vgo import VgoService
 
 
 def get_consolidation_service(
@@ -97,3 +98,13 @@ def get_intercompany_matrix_service(
 
 
 IntercompanyMatrixServiceDep = Annotated[IntercompanyMatrixService, Depends(get_intercompany_matrix_service)]
+
+
+def get_vgo_service(
+    session: SessionDep, osv_service: ConsolidatedOsvServiceDep, matrix_service: IntercompanyMatrixServiceDep
+) -> VgoService:
+    """Сервис подготовки строк ВГО на основе полной сводной ОСВ версии."""
+    return VgoService(session, osv_service, matrix_service)
+
+
+VgoServiceDep = Annotated[VgoService, Depends(get_vgo_service)]

api/src/api/endpoints/consolidations/__init__.py

@@ -14,6 +14,7 @@ from src.api.endpoints.consolidations import (
     periods,
     status,
     versions,
+    vgo_assignment,
 )
 
 router = APIRouter(tags=["consolidations"])
@@ -28,3 +29,4 @@ router.include_router(export.router)
 router.include_router(mapping.router)
 router.include_router(elimination_rule.router)
 router.include_router(intercompany_matrix.router)
+router.include_router(vgo_assignment.router)

api/src/api/endpoints/consolidations/vgo_assignment.py

@@ -0,0 +1,22 @@
+"""Подготовка распределения строк ОСВ по статьям КФО."""
+
+import uuid
+
+from fastapi import APIRouter, Depends
+
+from src.api.dependencies.auth import require
+from src.api.dependencies.consolidations import VgoServiceDep
+from src.schemas.consolidations import VgoAssignmentCandidatesOut
+
+router = APIRouter(prefix="/consolidations/versions")
+
+
+@router.post(
+    "/{version_id}/vgo-assignments/distribute",
+    dependencies=[Depends(require("CONSOLIDATION_EDIT"))],
+    response_model=VgoAssignmentCandidatesOut,
+    summary="Агрегаты ВГО по правилам и компаниям с ручными назначениями",
+)
+async def distribute_articles(version_id: uuid.UUID, service: VgoServiceDep) -> VgoAssignmentCandidatesOut:
+    """Возвращает агрегаты и ручные назначения по группам, а также строки без назначения."""
+    return await service.assignment_candidates(version_id)

api/src/models/kfo.py

@@ -8,7 +8,7 @@ import datetime
 import decimal
 import uuid
 
-from sqlalchemy import Boolean, DateTime, ForeignKey, Index, Integer, Numeric, String, Text, text
+from sqlalchemy import Boolean, CheckConstraint, DateTime, ForeignKey, Index, Integer, Numeric, String, Text, text
 from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.orm import Mapped, mapped_column
 
@@ -16,6 +16,50 @@ from .base import Base, UUIDPrimaryMixin
 from .enums import CorrectionStatus, StatementType, pg_enum
 
 
+class VgoAssignment(UUIDPrimaryMixin, Base):
+    """Выбор правила для строки ОСВ в рамках версии и типа отчётности."""
+
+    __tablename__ = "vgo_assignment"
+    __table_args__ = (
+        Index(
+            "uq_vgo_assignment_version_type_entry",
+            "consolidation_version_id",
+            "statement_type",
+            "osv_entry_id",
+            unique=True,
+        ),
+        Index("ix_vgo_assignment_osv_entry_id", "osv_entry_id"),
+        CheckConstraint("action IN ('ASSIGN', 'UNASSIGNED')", name="action_values"),
+        CheckConstraint(
+            "(action = 'ASSIGN' AND rule_id IS NOT NULL) OR (action = 'UNASSIGNED' AND rule_id IS NULL)",
+            name="action_rule_consistency",
+        ),
+    )
+
+    consolidation_version_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("consolidation_version.id", name="fk_vgo_assignment_version"), nullable=False
+    )
+    statement_type: Mapped[StatementType] = mapped_column(pg_enum("statement_type"), nullable=False)
+    osv_entry_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("osv_entry.id", name="fk_vgo_assignment_entry"), nullable=False
+    )
+    rule_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("elimination_rule.id", name="fk_vgo_assignment_rule")
+    )
+    action: Mapped[str] = mapped_column(String, nullable=False)
+    is_exclude: Mapped[bool] = mapped_column(Boolean, nullable=False, server_default=text("false"))
+    created_by: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("app_user.id", name="fk_vgo_assignment_created_by")
+    )
+    created_at: Mapped[datetime.datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=text("now()")
+    )
+    updated_by: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True), ForeignKey("app_user.id", name="fk_vgo_assignment_updated_by")
+    )
+    updated_at: Mapped[datetime.datetime | None] = mapped_column(DateTime(timezone=True))
+
+
 class Correction(UUIDPrimaryMixin, Base):
     """Корректировка ВГО: итог = свод первички − Σ применённых корректировок; пара — по (rule_id, компании)."""
 

api/src/repositories/vgo_assignments.py

@@ -0,0 +1,24 @@
+"""Ручные назначения строк ОСВ в версии консолидации."""
+
+import uuid
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.models.enums import StatementType
+from src.models.kfo import VgoAssignment
+
+
+async def list_assignments(
+    session: AsyncSession, version_id: uuid.UUID, statement_type: StatementType
+) -> list[VgoAssignment]:
+    """Все ручные решения для версии и типа отчётности."""
+    stmt = (
+        select(VgoAssignment)
+        .where(
+            VgoAssignment.consolidation_version_id == version_id,
+            VgoAssignment.statement_type == statement_type,
+        )
+        .order_by(VgoAssignment.created_at, VgoAssignment.id)
+    )
+    return list((await session.scalars(stmt)).all())

api/src/schemas/consolidations.py

@@ -425,6 +425,47 @@ class EliminationPairsOut(BaseModel):
         return None if value is None else value.quantize(CENT, rounding=ROUND_HALF_UP)
 
 
+class VgoAssignmentRowOut(ConsolidatedRowOut):
+    """Строка свода с данными ручного решения, если оно есть."""
+
+    osv_entry_id: uuid.UUID | None = None
+    action: Literal["ASSIGN", "UNASSIGNED"] | None = None
+    is_exclude: bool | None = None
+
+
+class VgoAssignmentGroupOut(BaseModel):
+    """Принятые агрегаты и ручные строки одного правила и направленной пары компаний."""
+
+    rule_id: uuid.UUID
+    rule_no: int
+    article_1_line: str
+    article_2_line: str
+    measure: Literal["closing", "turnover"]
+    company_a_id: uuid.UUID
+    company_a_name: str | None
+    company_b_id: uuid.UUID
+    company_b_name: str | None
+    amount_a: Decimal
+    amount_b: Decimal
+    deviation_amount: Decimal
+    deviation_percent: Decimal | None
+    matched: list[EliminationMatchedPairOut] = Field(default_factory=list)
+    with_discrepancy: list[EliminationMatchedPairOut] = Field(default_factory=list)
+    manual: list[VgoAssignmentRowOut] = Field(default_factory=list, exclude_if=lambda rows: not rows)
+
+    @field_validator("amount_a", "amount_b", "deviation_amount", "deviation_percent")
+    @classmethod
+    def _two_decimals(cls, value: Decimal | None) -> Decimal | None:
+        return None if value is None else value.quantize(CENT, rounding=ROUND_HALF_UP)
+
+
+class VgoAssignmentCandidatesOut(BaseModel):
+    """Принятые агрегаты и общий список строк свода, оставшихся без назначения."""
+
+    groups: list[VgoAssignmentGroupOut]
+    unassigned: list[VgoAssignmentRowOut]
+
+
 EliminationTotalStatus = Literal["AUTO", "DISCREPANCY", "ONE_SIDED", "MANUAL"]
 
 

api/src/services/intercompany_matrix.py

@@ -43,7 +43,7 @@ from src.schemas.consolidations import (
     IntercompanyResultTableOut,
     UnpairedReason,
 )
-from src.services.consolidated_osv import ConsolidatedOsvService
+from src.services.consolidated_osv import ConsolidatedOsvBuildResult, ConsolidatedOsvService
 
 Measure = Literal["closing", "turnover"]
 Side = Literal["D", "K"]
@@ -553,11 +553,25 @@ class IntercompanyMatrixService:
         self._scope = scope
         self._osv_service = service_osv
 
-    async def build_pairs(
+    async def build_pairs(  # noqa: PLR0913 - четыре идентификатора маршрута и необязательный готовый свод со строками
-        self, version_id: uuid.UUID, rule_id: uuid.UUID, company_a_id: uuid.UUID, company_b_id: uuid.UUID
+        self,
+        version_id: uuid.UUID,
+        rule_id: uuid.UUID,
+        company_a_id: uuid.UUID,
+        company_b_id: uuid.UUID,
+        rows: Sequence[ConsolidatedRow] | None = None,
+        *,
+        osv_result: ConsolidatedOsvBuildResult | None = None,
     ) -> EliminationPairsOut:
-        """Счета правила ББ и строки одной пары компаний без сохранения в БД (§4–§5)."""
+        """Счета правила ББ и строки одной пары компаний без сохранения в БД (§4–§5).
+
+        `rows` ограничивает исходные строки для поиска пар; без него используется весь свод. Готовый `osv_result`
+        позволяет повторно использовать уже собранный свод при расчёте нескольких правил и пар.
+        """
+        if osv_result is None:
             osv_result = await self._osv_service.build_with_members(version_id)
+        elif osv_result.version.id != version_id:
+            raise ValueError("Передан свод другой версии консолидации")
         version = osv_result.version
         member_ids = set(osv_result.members)
         if company_a_id == company_b_id or company_a_id not in member_ids or company_b_id not in member_ids:
@@ -572,7 +586,7 @@ class IntercompanyMatrixService:
         # в результаты разных правил. Глобальное исключение повторного использования потребует изменения модели
         # данных: нужно хранить/передавать принадлежность исходной строки конкретной паре между правилами.
         packing = _pack_preset_rows(
-            osv_result.consolidated.rows,
+            osv_result.consolidated.rows if rows is None else rows,
             selection,
             PairContext(rule, company_a_id, company_b_id),
         )

api/src/services/vgo.py

@@ -0,0 +1,271 @@
+"""Подготовка строк сводной ОСВ для распределения ВГО по правилам."""
+
+import uuid
+from collections import defaultdict
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass, field
+from decimal import Decimal
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.common.exceptions import DomainValidationError
+from src.domain.consolidated_osv import ConsolidatedRow
+from src.domain.elimination_packing import calculate_rule_totals
+from src.models.accounting import EliminationRule
+from src.models.enums import StatementType
+from src.models.kfo import VgoAssignment
+from src.repositories import elimination_rules as elimination_rules_repo
+from src.repositories import vgo_assignments as vgo_assignments_repo
+from src.schemas.consolidations import (
+    EliminationMatchedPairOut,
+    VgoAssignmentCandidatesOut,
+    VgoAssignmentGroupOut,
+    VgoAssignmentRowOut,
+)
+from src.services.consolidated_osv import ConsolidatedOsvBuildResult, ConsolidatedOsvService
+from src.services.intercompany_matrix import IntercompanyMatrixService
+
+ZERO = Decimal("0")
+GroupKey = tuple[uuid.UUID, uuid.UUID, uuid.UUID]
+ManualItem = tuple[VgoAssignmentRowOut, Decimal, Decimal]
+
+
+def assignment_candidates(rows: Sequence[ConsolidatedRow]) -> list[ConsolidatedRow]:
+    """Оставить строки с контрагентом и ненулевым оборотом или конечным остатком.
+
+    Все четыре суммы проверяются вместе; `None` равнозначен нулю. Начальный остаток не учитывается.
+    """
+    candidates = []
+    for row in rows:
+        if row.counterparty_id is None:
+            continue
+        amounts = (row.turnover_debit, row.turnover_credit, row.closing_debit, row.closing_credit)
+        for amount in amounts:
+            if amount not in (None, ZERO):
+                candidates.append(row)
+                break
+    return candidates
+
+
+def _ids_from_aggregate(aggregate: EliminationMatchedPairOut) -> set[uuid.UUID]:
+    return set(aggregate.osv_entry_ids_a) | set(aggregate.osv_entry_ids_b)
+
+
+def _assignment_row(row: ConsolidatedRow, assignment: VgoAssignment | None = None) -> VgoAssignmentRowOut:
+    output = VgoAssignmentRowOut.model_validate(row, from_attributes=True)
+    if assignment is not None:
+        output.osv_entry_id = assignment.osv_entry_id if len(row.entry_ids) == 1 else None
+        output.action = "ASSIGN" if assignment.action == "ASSIGN" else "UNASSIGNED"
+        output.is_exclude = assignment.is_exclude
+    return output
+
+
+def _manual_key_amount(row: ConsolidatedRow, rule: EliminationRule) -> tuple[GroupKey, Decimal, Decimal]:
+    """Определить сторону ручной строки по сумме и стороне A правила."""
+    if rule.measure == "turnover":
+        debit, credit = row.turnover_debit or ZERO, row.turnover_credit or ZERO
+    else:
+        debit, credit = row.closing_debit or ZERO, row.closing_credit or ZERO
+    first, second = (debit, credit) if rule.side_1 == "D" else (credit, debit)
+    assert row.counterparty_id is not None
+    if first != ZERO or second == ZERO:
+        return (rule.id, row.legal_entity_id, row.counterparty_id), first, ZERO
+    return (rule.id, row.counterparty_id, row.legal_entity_id), ZERO, second
+
+
+@dataclass
+class _ManualResult:
+    """Ручные решения по строкам свода и группам правил."""
+
+    groups: dict[GroupKey, list[ManualItem]]
+    by_row: dict[tuple[uuid.UUID, ...], VgoAssignment]
+    assigned_ids: set[uuid.UUID]
+
+
+def _manual_assignments(
+    rows: Sequence[ConsolidatedRow], assignments: Sequence[VgoAssignment], rules: Sequence[EliminationRule]
+) -> _ManualResult:
+    by_entry = {assignment.osv_entry_id: assignment for assignment in assignments}
+    rules_by_id = {rule.id: rule for rule in rules}
+    groups: defaultdict[GroupKey, list[ManualItem]] = defaultdict(list)
+    by_row: dict[tuple[uuid.UUID, ...], VgoAssignment] = {}
+    assigned_ids: set[uuid.UUID] = set()
+    for row in rows:
+        found = [by_entry[entry_id] for entry_id in row.entry_ids if entry_id in by_entry]
+        if not found:
+            continue
+        if len(found) != len(row.entry_ids):
+            raise DomainValidationError(
+                "AMBIGUOUS_VGO_ASSIGNMENT",
+                "Для части исходных строк сводной ОСВ есть ручное назначение, а для остальных нет",
+            )
+        assignment = found[0]
+        if any(
+            (item.action, item.rule_id, item.is_exclude)
+            != (assignment.action, assignment.rule_id, assignment.is_exclude)
+            for item in found[1:]
+        ):
+            raise DomainValidationError(
+                "CONFLICTING_VGO_ASSIGNMENT",
+                "У исходных строк одной сводной ОСВ разные ручные назначения",
+            )
+        by_row[row.entry_ids] = assignment
+        rule = rules_by_id.get(assignment.rule_id) if assignment.rule_id is not None else None
+        if assignment.action == "ASSIGN" and rule is not None and row.counterparty_id is not None:
+            key, amount_a, amount_b = _manual_key_amount(row, rule)
+            groups[key].append((_assignment_row(row, assignment), amount_a, amount_b))
+            assigned_ids.update(row.entry_ids)
+    return _ManualResult(groups, by_row, assigned_ids)
+
+
+def _unassigned_rows(
+    all_rows: Sequence[ConsolidatedRow],
+    candidates: Sequence[ConsolidatedRow],
+    manual: _ManualResult,
+    used_ids: set[uuid.UUID],
+) -> list[VgoAssignmentRowOut]:
+    rows = [*candidates]
+    candidate_keys = {row.entry_ids for row in candidates}
+    rows.extend(row for row in all_rows if row.entry_ids not in candidate_keys and row.entry_ids in manual.by_row)
+    return [
+        _assignment_row(row, manual.by_row.get(row.entry_ids))
+        for row in rows
+        if used_ids.isdisjoint(row.entry_ids) and manual.assigned_ids.isdisjoint(row.entry_ids)
+    ]
+
+
+@dataclass
+class _AggregateSearchResult:
+    """Принятые агрегаты, кандидаты с расхождением и занятые строки."""
+
+    matched_aggregates: defaultdict[GroupKey, list[EliminationMatchedPairOut]] = field(
+        default_factory=lambda: defaultdict(list)
+    )
+    diff_aggregates: list[EliminationMatchedPairOut] = field(default_factory=list)
+    used_ids: set[uuid.UUID] = field(default_factory=set)
+
+
+def _select_diff_aggregates(search: _AggregateSearchResult) -> dict[GroupKey, list[EliminationMatchedPairOut]]:
+    """Принять только расхождения с единственным вариантом для набора исходных строк."""
+    by_ids: defaultdict[tuple[uuid.UUID, ...], list[EliminationMatchedPairOut]] = defaultdict(list)
+    for aggregate in search.diff_aggregates:
+        ids = _ids_from_aggregate(aggregate)
+        if ids and search.used_ids.isdisjoint(ids):
+            by_ids[tuple(sorted(ids))].append(aggregate)
+
+    diff_aggregates: defaultdict[GroupKey, list[EliminationMatchedPairOut]] = defaultdict(list)
+    for options in by_ids.values():
+        if len(options) == 1:
+            aggregate = options[0]
+            key = (aggregate.rule_id, aggregate.company_a_id, aggregate.company_b_id)
+            diff_aggregates[key].append(aggregate)
+            search.used_ids.update(_ids_from_aggregate(aggregate))
+    return diff_aggregates
+
+
+def _build_groups(
+    search: _AggregateSearchResult,
+    diff_aggregates: dict[GroupKey, list[EliminationMatchedPairOut]],
+    rules: Sequence[EliminationRule],
+    manual: dict[GroupKey, list[ManualItem]],
+    names: Mapping[uuid.UUID, str],
+) -> list[VgoAssignmentGroupOut]:
+    """Сгруппировать принятые агрегаты и посчитать суммы по правилу и компаниям."""
+    rule_by_id = {rule.id: rule for rule in rules}
+    keys = list(dict.fromkeys([*search.matched_aggregates, *diff_aggregates, *manual]))
+    groups = []
+    for rule_id, company_a_id, company_b_id in keys:
+        key = (rule_id, company_a_id, company_b_id)
+        matched = search.matched_aggregates.get(key, [])
+        with_discrepancy = diff_aggregates.get(key, [])
+        aggregates = [*matched, *with_discrepancy]
+        manual_rows = manual.get(key, [])
+        amount_a = sum((aggregate.amount_a for aggregate in aggregates), ZERO)
+        amount_b = sum((aggregate.amount_b for aggregate in aggregates), ZERO)
+        amount_a += sum((amount_a for row, amount_a, _ in manual_rows if not row.is_exclude), ZERO)
+        amount_b += sum((amount_b for row, _, amount_b in manual_rows if not row.is_exclude), ZERO)
+        totals = calculate_rule_totals(amount_a, amount_b, manual=False)
+        groups.append(
+            VgoAssignmentGroupOut(
+                rule_id=rule_id,
+                rule_no=rule_by_id[rule_id].rule_no,
+                article_1_line=rule_by_id[rule_id].article_1_line,
+                article_2_line=rule_by_id[rule_id].article_2_line,
+                measure=rule_by_id[rule_id].measure,
+                company_a_id=company_a_id,
+                company_a_name=names.get(company_a_id),
+                company_b_id=company_b_id,
+                company_b_name=names.get(company_b_id),
+                amount_a=amount_a,
+                amount_b=amount_b,
+                deviation_amount=totals.deviation_amount,
+                deviation_percent=totals.deviation_percent,
+                matched=matched,
+                with_discrepancy=with_discrepancy,
+                manual=[row for row, _, _ in manual_rows],
+            )
+        )
+    return groups
+
+
+class VgoService:
+    """Операции ВГО над сводной ОСВ версии консолидации."""
+
+    def __init__(
+        self, session: AsyncSession, osv_service: ConsolidatedOsvService, matrix_service: IntercompanyMatrixService
+    ) -> None:
+        self._osv_service = osv_service
+        self._matrix_service = matrix_service
+        self.session = session
+
+    async def _find_aggregates(
+        self,
+        version_id: uuid.UUID,
+        osv_result: ConsolidatedOsvBuildResult,
+        rules: Sequence[EliminationRule],
+        rows: Sequence[ConsolidatedRow],
+    ) -> _AggregateSearchResult:
+        """Перебрать правила и компании, сразу закрепляя только точные агрегаты."""
+        search = _AggregateSearchResult()
+        for company_a in osv_result.members:
+            for company_b in osv_result.members:
+                if company_a == company_b:
+                    continue
+                for rule in rules:
+                    free_rows = [row for row in rows if search.used_ids.isdisjoint(row.entry_ids)]
+                    result = await self._matrix_service.build_pairs(
+                        version_id, rule.id, company_a, company_b, free_rows, osv_result=osv_result
+                    )
+                    for aggregate in result.matched:
+                        ids = _ids_from_aggregate(aggregate)
+                        if search.used_ids.isdisjoint(ids):
+                            search.used_ids.update(ids)
+                            key = (aggregate.rule_id, aggregate.company_a_id, aggregate.company_b_id)
+                            search.matched_aggregates[key].append(aggregate)
+                    search.diff_aggregates.extend(result.with_discrepancy)
+        return search
+
+    async def assignment_candidates(self, version_id: uuid.UUID) -> VgoAssignmentCandidatesOut:
+        """Найти агрегаты и общие для всех правил строки без назначения, не записывая их в БД."""
+        osv_result = await self._osv_service.build_with_members(version_id)
+        all_rules = await elimination_rules_repo.list_rules(self.session, osv_result.version.elimination_rules_version)
+        rules = [
+            rule for rule in all_rules if rule.statement_type == StatementType.balance and rule.measure == "closing"
+        ]
+        rows = assignment_candidates(osv_result.consolidated.rows)
+        assignments = await vgo_assignments_repo.list_assignments(self.session, version_id, StatementType.balance)
+        manual = _manual_assignments(osv_result.consolidated.rows, assignments, rules)
+        free_rows = [row for row in rows if row.entry_ids not in manual.by_row]
+        search = await self._find_aggregates(version_id, osv_result, rules, free_rows)
+        diff_aggregates = _select_diff_aggregates(search)
+        
+        names = {company_id: entity.name for company_id, entity in osv_result.members.items()}
+        for row in osv_result.consolidated.rows:
+            if row.counterparty_id is not None:
+                name = row.counterparty_name or row.counterparty_entity_name
+                if name is not None:
+                    names.setdefault(row.counterparty_id, name)
+                    
+        groups = _build_groups(search, diff_aggregates, rules, manual.groups, names)
+        unassigned = _unassigned_rows(osv_result.consolidated.rows, rows, manual, search.used_ids)
+        return VgoAssignmentCandidatesOut(groups=groups, unassigned=unassigned)

api/tests/test_domain_consolidated_osv.py

@@ -27,6 +27,7 @@ from src.domain.consolidated_osv import (
 )
 from src.domain.osv_mapping import Resolution
 from src.models.enums import CounterpartyMatchKind, PeriodQuarter
+from src.services.vgo import assignment_candidates
 
 Q1, Q2, Q3 = PeriodQuarter.Q1, PeriodQuarter.Q2, PeriodQuarter.Q3
 INN, NAME, ALIAS, MANUAL, PROPAGATED = (
@@ -142,6 +143,22 @@ def _sums(row: ConsolidatedRow) -> tuple[Decimal | None, ...]:
     )
 
 
+def test_assignment_candidates_require_counterparty_and_any_nonzero_turnover_or_closing() -> None:
+    document = _doc(cumulative=True)
+    base = _build([document], [_row(document, counterparty_id=SCH.id)]).rows[0]
+    zero = dataclasses.replace(base, opening_debit=Decimal("100"))
+    empty = dataclasses.replace(
+        base, turnover_debit=None, turnover_credit=None, closing_debit=None, closing_credit=None
+    )
+    without_counterparty = dataclasses.replace(base, counterparty_id=None, turnover_debit=Decimal("1"))
+    nonzero = [
+        dataclasses.replace(base, **{field: Decimal("1")})
+        for field in ("turnover_debit", "turnover_credit", "closing_debit", "closing_credit")
+    ]
+
+    assert assignment_candidates([zero, empty, without_counterparty, *nonzero]) == nonzero
+
+
 # --------------------------------------------------------------------------- приведение периодов
 
 

api/tests/test_intercompany_matrix.py

@@ -4,7 +4,7 @@ import datetime
 import uuid
 from decimal import Decimal
 from types import SimpleNamespace
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, Mock
 
 import pytest
 from fastapi import status as http_status
@@ -89,6 +89,35 @@ def test_routes_are_separate_and_matrix_expands_manual_accounts() -> None:
     ]
 
 
+async def test_build_pairs_uses_supplied_rows_and_prebuilt_osv(monkeypatch: pytest.MonkeyPatch) -> None:
+    version_id, rule_id, company_a, company_b = (uuid.uuid4() for _ in range(4))
+    full_rows = [object()]
+    selected_rows: list[object] = []
+    osv_result = SimpleNamespace(
+        version=SimpleNamespace(id=version_id),
+        members={company_a: object(), company_b: object()},
+        consolidated=SimpleNamespace(rows=full_rows),
+    )
+    osv_service = SimpleNamespace(build_with_members=AsyncMock(side_effect=AssertionError("свод собран повторно")))
+    service = matrix.IntercompanyMatrixService(AsyncMock(), None, osv_service)
+    rule = SimpleNamespace(id=rule_id, rule_no=1, version_no=2, measure="closing")
+    service._resolve_preset_selection = AsyncMock(  # type: ignore[method-assign]
+        return_value=(matrix.MatrixSelection(frozenset(), frozenset(), "closing"), rule)
+    )
+    monkeypatch.setattr(matrix.directory_repo, "list_accounts", AsyncMock(return_value=[]))
+    totals = matrix.RuleTotals(Decimal("0"), Decimal("0"), Decimal("0"), None, Decimal("0"), False)
+    pack_rows = Mock(return_value=matrix.RulePacking([], [], [], totals))
+    monkeypatch.setattr(matrix, "_pack_preset_rows", pack_rows)
+
+    response = await service.build_pairs(
+        version_id, rule_id, company_a, company_b, selected_rows, osv_result=osv_result
+    )
+
+    assert response.rule_id == rule_id
+    assert pack_rows.call_args.args[0] is selected_rows
+    osv_service.build_with_members.assert_not_awaited()
+
+
 async def test_preset_applies_exclusions_to_its_own_side_and_rule_measure(monkeypatch: pytest.MonkeyPatch) -> None:
     parent = _account(AccountKind.AP)
     child = _account(AccountKind.AP, parent.id)

api/tests/test_vgo.py

@@ -0,0 +1,279 @@
+"""Группировка найденных пар ВГО и единственность исходных строк во всей версии."""
+
+import uuid
+from decimal import Decimal
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from src.common.exceptions import DomainValidationError
+from src.domain.consolidated_osv import CounterpartyType
+from src.models.enums import StatementType
+from src.schemas.consolidations import EliminationMatchedPairOut
+from src.services import vgo
+
+
+def _pair(  # noqa: PLR0913 - фабрика тестовой пары с настраиваемой суммой
+    rule_id: uuid.UUID,
+    company_a: uuid.UUID,
+    company_b: uuid.UUID,
+    entry_a: uuid.UUID,
+    entry_b: uuid.UUID,
+    *,
+    amount_b: str = "10",
+) -> EliminationMatchedPairOut:
+    deviation = Decimal("10") - Decimal(amount_b)
+    return EliminationMatchedPairOut(
+        rule_id=rule_id,
+        company_a_id=company_a,
+        company_b_id=company_b,
+        amount_a=Decimal("10"),
+        amount_b=Decimal(amount_b),
+        deviation_amount=deviation,
+        deviation_percent=deviation / Decimal(amount_b) * 100,
+        osv_entry_ids_a=[entry_a],
+        osv_entry_ids_b=[entry_b],
+    )
+
+
+def _row(entry_id: uuid.UUID, company_a: uuid.UUID, company_b: uuid.UUID, index: int) -> SimpleNamespace:
+    return SimpleNamespace(
+        entry_ids=(entry_id,),
+        legal_entity_id=company_a,
+        legal_entity_name="A",
+        legal_entity_inn="1234567890",
+        account_code="76",
+        account_root="76",
+        period="2026 Q1",
+        is_cumulative=True,
+        row_number=index,
+        counterparty_name="B",
+        counterparty_inn="0987654321",
+        counterparty_id=company_b,
+        counterparty_entity_name="B",
+        counterparty_entity_inn="0987654321",
+        counterparty_type=CounterpartyType.GSK,
+        match_kind=None,
+        match_stored=True,
+        opening_debit=Decimal("0"),
+        opening_credit=Decimal("0"),
+        turnover_debit=Decimal("0"),
+        turnover_credit=Decimal("0"),
+        closing_debit=Decimal("1"),
+        closing_credit=Decimal("0"),
+        warnings=(),
+    )
+
+
+def _members(company_a: uuid.UUID, company_b: uuid.UUID) -> dict[uuid.UUID, SimpleNamespace]:
+    return {company_a: SimpleNamespace(name="A"), company_b: SimpleNamespace(name="B")}
+
+
+async def test_pairs_are_grouped_and_entry_ids_are_used_once(monkeypatch: pytest.MonkeyPatch) -> None:
+    version_id, company_a, company_b, rule_1, rule_2 = (uuid.uuid4() for _ in range(5))
+    entry_ids = [uuid.uuid4() for _ in range(11)]
+    rows = [_row(entry_id, company_a, company_b, index) for index, entry_id in enumerate(entry_ids)]
+    osv_result = SimpleNamespace(
+        version=SimpleNamespace(elimination_rules_version=2),
+        consolidated=SimpleNamespace(rows=rows),
+        members=_members(company_a, company_b),
+    )
+    rules = [
+        SimpleNamespace(
+            id=rule_1, rule_no=1, statement_type=StatementType.balance, measure="closing",
+            article_1_line="Статья A1", article_2_line="Статья B1",
+        ),
+        SimpleNamespace(
+            id=rule_2, rule_no=2, statement_type=StatementType.balance, measure="closing",
+            article_1_line="Статья A2", article_2_line="Статья B2",
+        ),
+    ]
+    monkeypatch.setattr(vgo.elimination_rules_repo, "list_rules", AsyncMock(return_value=rules))
+    monkeypatch.setattr(vgo.vgo_assignments_repo, "list_assignments", AsyncMock(return_value=[]))
+    exact_1 = _pair(rule_1, company_a, company_b, entry_ids[0], entry_ids[1])
+    exact_1_more = _pair(rule_1, company_a, company_b, entry_ids[9], entry_ids[10])
+    exact_conflict = _pair(rule_2, company_a, company_b, entry_ids[0], entry_ids[2])
+    exact_2 = _pair(rule_2, company_b, company_a, entry_ids[2], entry_ids[3])
+    discrepancy_1 = _pair(rule_1, company_a, company_b, entry_ids[4], entry_ids[5])
+    discrepancy_conflict = _pair(rule_2, company_b, company_a, entry_ids[5], entry_ids[4])
+    discrepancy_unique = _pair(rule_1, company_b, company_a, entry_ids[7], entry_ids[8], amount_b="9")
+
+    async def build_pairs(
+        _version: uuid.UUID, rule: uuid.UUID, a: uuid.UUID, b: uuid.UUID, available: list, **_kwargs: object
+    ) -> SimpleNamespace:
+        if (rule, a, b) == (rule_1, company_a, company_b):
+            return SimpleNamespace(rule_no=1, matched=[exact_1, exact_1_more], with_discrepancy=[discrepancy_1])
+        if (rule, a, b) == (rule_2, company_a, company_b):
+            assert entry_ids[0] not in {row.entry_ids[0] for row in available}
+            return SimpleNamespace(rule_no=2, matched=[exact_conflict], with_discrepancy=[])
+        if (rule, a, b) == (rule_1, company_b, company_a):
+            return SimpleNamespace(rule_no=1, matched=[], with_discrepancy=[discrepancy_unique])
+        if (rule, a, b) == (rule_2, company_b, company_a):
+            return SimpleNamespace(rule_no=2, matched=[exact_2], with_discrepancy=[discrepancy_conflict])
+        return SimpleNamespace(rule_no=1, matched=[], with_discrepancy=[])
+
+    service = vgo.VgoService(
+        AsyncMock(),
+        SimpleNamespace(build_with_members=AsyncMock(return_value=osv_result)),
+        SimpleNamespace(build_pairs=AsyncMock(side_effect=build_pairs)),
+    )
+    response = await service.assignment_candidates(version_id)
+    groups = response.groups
+
+    assert [(group.rule_id, group.company_a_id, group.company_b_id) for group in groups] == [
+        (rule_1, company_a, company_b),
+        (rule_2, company_b, company_a),
+        (rule_1, company_b, company_a),
+    ]
+    assert groups[0].matched == [exact_1, exact_1_more]
+    assert (groups[0].article_1_line, groups[0].article_2_line) == ("Статья A1", "Статья B1")
+    assert (groups[0].company_a_name, groups[0].company_b_name) == ("A", "B")
+    assert (groups[0].amount_a, groups[0].amount_b, groups[0].deviation_amount) == (
+        Decimal("20.00"),
+        Decimal("20.00"),
+        Decimal("0.00"),
+    )
+    assert groups[0].with_discrepancy == []
+    assert "manual" not in response.model_dump()["groups"][0]
+    assert groups[1].matched == [exact_2]
+    assert groups[1].with_discrepancy == []
+    assert groups[2].matched == []
+    assert groups[2].with_discrepancy == [discrepancy_unique]
+    assert (groups[2].amount_a, groups[2].amount_b, groups[2].deviation_amount, groups[2].deviation_percent) == (
+        Decimal("10.00"),
+        Decimal("9.00"),
+        Decimal("1.00"),
+        Decimal("11.11"),
+    )
+    assert [row.entry_ids for row in response.unassigned] == [
+        [entry_ids[4]],
+        [entry_ids[5]],
+        [entry_ids[6]],
+    ]
+
+
+async def test_manual_assignments_override_auto_and_keep_exclusion_flag(monkeypatch: pytest.MonkeyPatch) -> None:
+    version_id, company_a, company_b, rule_id = (uuid.uuid4() for _ in range(4))
+    entry_ids = [uuid.uuid4() for _ in range(4)]
+    rows = [_row(entry_id, company_a, company_b, index) for index, entry_id in enumerate(entry_ids)]
+    for row, amount in zip(rows, ("10", "20", "5", "1"), strict=True):
+        row.closing_debit = Decimal(amount)
+    osv_result = SimpleNamespace(
+        version=SimpleNamespace(elimination_rules_version=2),
+        consolidated=SimpleNamespace(rows=rows),
+        members=_members(company_a, company_b),
+    )
+    rule = SimpleNamespace(
+        id=rule_id, rule_no=1, statement_type=StatementType.balance, measure="closing", side_1="D",
+        article_1_line="Статья A", article_2_line="Статья B",
+    )
+    assignments = [
+        SimpleNamespace(osv_entry_id=entry_ids[0], rule_id=rule_id, action="ASSIGN", is_exclude=False),
+        SimpleNamespace(osv_entry_id=entry_ids[1], rule_id=rule_id, action="ASSIGN", is_exclude=True),
+        SimpleNamespace(osv_entry_id=entry_ids[2], rule_id=None, action="UNASSIGNED", is_exclude=True),
+    ]
+    monkeypatch.setattr(vgo.elimination_rules_repo, "list_rules", AsyncMock(return_value=[rule]))
+    monkeypatch.setattr(vgo.vgo_assignments_repo, "list_assignments", AsyncMock(return_value=assignments))
+
+    async def build_pairs(
+        _version: uuid.UUID, _rule: uuid.UUID, _a: uuid.UUID, _b: uuid.UUID, available: list, **_kwargs: object
+    ) -> SimpleNamespace:
+        assert [row.entry_ids[0] for row in available] == [entry_ids[3]]
+        return SimpleNamespace(matched=[], with_discrepancy=[])
+
+    service = vgo.VgoService(
+        AsyncMock(),
+        SimpleNamespace(build_with_members=AsyncMock(return_value=osv_result)),
+        SimpleNamespace(build_pairs=AsyncMock(side_effect=build_pairs)),
+    )
+    response = await service.assignment_candidates(version_id)
+
+    assert len(response.groups) == 1
+    assert (response.groups[0].rule_id, response.groups[0].company_a_id, response.groups[0].company_b_id) == (
+        rule_id,
+        company_a,
+        company_b,
+    )
+    assert (response.groups[0].amount_a, response.groups[0].amount_b) == (Decimal("10.00"), Decimal("0.00"))
+    assert [(row.osv_entry_id, row.is_exclude) for row in response.groups[0].manual] == [
+        (entry_ids[0], False),
+        (entry_ids[1], True),
+    ]
+    assert "manual" in response.model_dump()["groups"][0]
+    assert [(row.entry_ids, row.action, row.is_exclude) for row in response.unassigned] == [
+        ([entry_ids[2]], "UNASSIGNED", True),
+        ([entry_ids[3]], None, None),
+    ]
+
+
+async def test_glued_row_requires_assignments_for_all_source_entries(monkeypatch: pytest.MonkeyPatch) -> None:
+    version_id, company_a, company_b, rule_id, entry_a, entry_b = (uuid.uuid4() for _ in range(6))
+    row = _row(entry_a, company_a, company_b, 1)
+    row.entry_ids = (entry_a, entry_b)
+    osv_result = SimpleNamespace(
+        version=SimpleNamespace(elimination_rules_version=2),
+        consolidated=SimpleNamespace(rows=[row]),
+        members=_members(company_a, company_b),
+    )
+    assignment = SimpleNamespace(osv_entry_id=entry_a, rule_id=rule_id, action="ASSIGN", is_exclude=False)
+    monkeypatch.setattr(vgo.elimination_rules_repo, "list_rules", AsyncMock(return_value=[]))
+    monkeypatch.setattr(vgo.vgo_assignments_repo, "list_assignments", AsyncMock(return_value=[assignment]))
+    service = vgo.VgoService(
+        AsyncMock(),
+        SimpleNamespace(build_with_members=AsyncMock(return_value=osv_result)),
+        SimpleNamespace(build_pairs=AsyncMock()),
+    )
+
+    with pytest.raises(DomainValidationError, match="для остальных нет"):
+        await service.assignment_candidates(version_id)
+
+
+async def test_glued_row_with_same_assignments_is_counted_once(monkeypatch: pytest.MonkeyPatch) -> None:
+    version_id, company_a, company_b, rule_id, entry_a, entry_b = (uuid.uuid4() for _ in range(6))
+    row = _row(entry_a, company_a, company_b, 1)
+    row.entry_ids = (entry_a, entry_b)
+    row.closing_debit = Decimal("10")
+    osv_result = SimpleNamespace(
+        version=SimpleNamespace(elimination_rules_version=2),
+        consolidated=SimpleNamespace(rows=[row]),
+        members=_members(company_a, company_b),
+    )
+    rule = SimpleNamespace(
+        id=rule_id, rule_no=1, statement_type=StatementType.balance, measure="closing", side_1="D",
+        article_1_line="Статья A", article_2_line="Статья B",
+    )
+    assignments = [
+        SimpleNamespace(osv_entry_id=entry_id, rule_id=rule_id, action="ASSIGN", is_exclude=False)
+        for entry_id in (entry_a, entry_b)
+    ]
+    monkeypatch.setattr(vgo.elimination_rules_repo, "list_rules", AsyncMock(return_value=[rule]))
+    monkeypatch.setattr(vgo.vgo_assignments_repo, "list_assignments", AsyncMock(return_value=assignments))
+    empty_result = SimpleNamespace(matched=[], with_discrepancy=[])
+    matrix_service = SimpleNamespace(build_pairs=AsyncMock(return_value=empty_result))
+    service = vgo.VgoService(
+        AsyncMock(), SimpleNamespace(build_with_members=AsyncMock(return_value=osv_result)), matrix_service
+    )
+
+    response = await service.assignment_candidates(version_id)
+
+    assert len(response.groups) == 1
+    assert response.groups[0].amount_a == Decimal("10.00")
+    assert len(response.groups[0].manual) == 1
+    assert response.groups[0].manual[0].entry_ids == [entry_a, entry_b]
+    assert response.groups[0].manual[0].osv_entry_id is None
+    assert response.unassigned == []
+    assert all(call.args[4] == [] for call in matrix_service.build_pairs.await_args_list)
+
+
+def test_glued_row_rejects_conflicting_manual_assignments() -> None:
+    company_a, company_b, rule_id, entry_a, entry_b = (uuid.uuid4() for _ in range(5))
+    row = _row(entry_a, company_a, company_b, 1)
+    row.entry_ids = (entry_a, entry_b)
+    assignments = [
+        SimpleNamespace(osv_entry_id=entry_a, rule_id=rule_id, action="ASSIGN", is_exclude=False),
+        SimpleNamespace(osv_entry_id=entry_b, rule_id=None, action="UNASSIGNED", is_exclude=False),
+    ]
+
+    with pytest.raises(DomainValidationError, match="разные ручные назначения"):
+        vgo._manual_assignments([row], assignments, [])