Гудвилл: доля до двух знаков, итоги по ГО и УК с бэка

participation() — единственное место формулы доли участия; totals с УК, group_totals
по ГО, GoodwillTableOut.groups; экспорт берёт «Всего» и «Итого по листу» из ответа таблицы;
доля 0–100 не больше двух знаков (DOC-259025 §5.2), ответ — два знака.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev 2026-11-06 18:22:26 +04:00
parent 3e07276a96
commit a086c39a78
8 changed files with 126 additions and 35 deletions

View File

@ -1,4 +1,4 @@
"""Чистый расчёт направленных пар гудвилла; суммы в рублях.""" """Чистый расчёт направленных пар гудвилла (§8 «Гудвилл», DOC-259025 §§3, 6); суммы в рублях."""
import uuid import uuid
from collections import defaultdict from collections import defaultdict
@ -48,6 +48,11 @@ class Row:
) )
def participation(capital: Decimal, share: Decimal) -> Decimal:
"""Доля участия P = УК × доля / 100, до копеек ROUND_HALF_UP (DOC-259025 §6)."""
return (capital * share / 100).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
def build( def build(
investments: Sequence[Investment], investments: Sequence[Investment],
decisions: Mapping[Pair, Decision], decisions: Mapping[Pair, Decision],
@ -73,7 +78,7 @@ def build(
if active and uk is None: if active and uk is None:
warnings.append("CAPITAL_SOURCE_MISSING") warnings.append("CAPITAL_SOURCE_MISSING")
if active and decision.include_in_kfo and uk is not None and decision.ownership_share is not None: if active and decision.include_in_kfo and uk is not None and decision.ownership_share is not None:
p = (uk * decision.ownership_share / 100).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) p = participation(uk, decision.ownership_share)
g = fv - p g = fv - p
if active and shares[subsidiary] > MAX_SHARE: if active and shares[subsidiary] > MAX_SHARE:
warnings.append("OWNERSHIP_TOTAL_EXCEEDED") warnings.append("OWNERSHIP_TOTAL_EXCEEDED")
@ -83,11 +88,21 @@ def build(
def totals(rows: Sequence[Row], enabled: bool) -> dict[str, Decimal]: def totals(rows: Sequence[Row], enabled: bool) -> dict[str, Decimal]:
"""ФВ и УК — все действующие строки; P и G — действующие отмеченные; в КФО — применимые положительные G."""
active = [r for r in rows if r.has_investments] active = [r for r in rows if r.has_investments]
selected = [r for r in active if r.decision.include_in_kfo] selected = [r for r in active if r.decision.include_in_kfo]
return { return {
"financial_investment_amount": sum((r.financial_investment_amount or ZERO for r in active), ZERO), "financial_investment_amount": sum((r.financial_investment_amount or ZERO for r in active), ZERO),
"charter_capital_amount": sum((r.charter_capital_amount or ZERO for r in active), ZERO),
"participation_amount": sum((r.participation_amount or ZERO for r in selected), ZERO), "participation_amount": sum((r.participation_amount or ZERO for r in selected), ZERO),
"goodwill_amount": sum((r.goodwill_amount or ZERO for r in selected), ZERO), "goodwill_amount": sum((r.goodwill_amount or ZERO for r in selected), ZERO),
"goodwill_in_kfo_amount": sum((r.goodwill_amount or ZERO for r in active if enabled and r.applicable), ZERO), "goodwill_in_kfo_amount": sum((r.goodwill_amount or ZERO for r in active if enabled and r.applicable), ZERO),
} }
def group_totals(rows: Sequence[Row], enabled: bool) -> dict[uuid.UUID, dict[str, Decimal]]:
"""Итоги «Всего» по каждой ГО — те же правила, что у `totals`."""
groups: dict[uuid.UUID, list[Row]] = defaultdict(list)
for row in rows:
groups[row.head_id].append(row)
return {head: totals(items, enabled) for head, items in groups.items()}

View File

@ -10,7 +10,7 @@ from pydantic import BaseModel, ConfigDict, Field, PlainSerializer
from src.schemas.consolidations import RefOut from src.schemas.consolidations import RefOut
from src.schemas.corrections import Money from src.schemas.corrections import Money
Share = Annotated[Decimal, PlainSerializer(lambda v: format(v, ".4f"), return_type=str, when_used="json")] Share = Annotated[Decimal, PlainSerializer(lambda v: format(v, ".2f"), return_type=str, when_used="json")]
GoodwillSign = Literal["all", "positive", "negative", "zero"] GoodwillSign = Literal["all", "positive", "negative", "zero"]
@ -40,17 +40,26 @@ class GoodwillRowOut(BaseModel):
class GoodwillTotalsOut(BaseModel): class GoodwillTotalsOut(BaseModel):
financial_investment_amount: Money financial_investment_amount: Money
charter_capital_amount: Money
participation_amount: Money participation_amount: Money
goodwill_amount: Money goodwill_amount: Money
goodwill_in_kfo_amount: Money goodwill_in_kfo_amount: Money
class GoodwillGroupOut(BaseModel):
"""Итог «Всего» блока одной ГО."""
head_legal_entity: RefOut
totals: GoodwillTotalsOut
class GoodwillTableOut(BaseModel): class GoodwillTableOut(BaseModel):
version_id: UUID version_id: UUID
goodwill_enabled: bool goodwill_enabled: bool
editable: bool editable: bool
read_only_reason: Literal["VERSION_NOT_EDITABLE", "LOCK_REQUIRED", "LOCKED_BY_OTHER"] | None read_only_reason: Literal["VERSION_NOT_EDITABLE", "LOCK_REQUIRED", "LOCKED_BY_OTHER"] | None
rows: list[GoodwillRowOut] rows: list[GoodwillRowOut]
groups: list[GoodwillGroupOut]
totals: GoodwillTotalsOut totals: GoodwillTotalsOut

View File

@ -1,7 +1,7 @@
"""Таблица и построчные решения гудвилла в транзакции запроса.""" """Таблица и построчные решения гудвилла в транзакции запроса."""
import re import re
from decimal import ROUND_HALF_UP, Decimal from decimal import Decimal
from uuid import UUID, uuid4 from uuid import UUID, uuid4
from fastapi.concurrency import run_in_threadpool from fastapi.concurrency import run_in_threadpool
@ -22,6 +22,7 @@ from src.repositories import consolidations as versions
from src.repositories import goodwill as repo from src.repositories import goodwill as repo
from src.schemas.consolidations import RefOut from src.schemas.consolidations import RefOut
from src.schemas.goodwill import ( from src.schemas.goodwill import (
GoodwillGroupOut,
GoodwillRowIn, GoodwillRowIn,
GoodwillRowOut, GoodwillRowOut,
GoodwillSign, GoodwillSign,
@ -46,8 +47,8 @@ def validate_decision(data: GoodwillRowIn) -> domain.Decision:
value = data.ownership_share.replace(",", ".") if data.ownership_share is not None else None value = data.ownership_share.replace(",", ".") if data.ownership_share is not None else None
if value is None: if value is None:
share = None share = None
elif not re.fullmatch(r"\d+(?:\.\d{1,4})?", value) or Decimal(value) > MAX_SHARE: elif not re.fullmatch(r"\d{1,3}(?:\.\d{1,2})?", value) or Decimal(value) > MAX_SHARE:
raise DomainValidationError("INVALID_SHARE", "Доля должна быть от 0 до 100, до четырёх знаков") raise DomainValidationError("INVALID_SHARE", "Доля — число от 0 до 100, не больше двух знаков после запятой")
else: else:
share = Decimal(value) share = Decimal(value)
if data.include_in_kfo and share is None: if data.include_in_kfo and share is None:
@ -131,12 +132,21 @@ class GoodwillService:
r.subsidiary_legal_entity.id, r.subsidiary_legal_entity.id,
) )
) )
groups = domain.group_totals(rows, ctx.version.goodwill_enabled)
heads = sorted(groups, key=lambda head: (entities[head].name, head))
return GoodwillTableOut( return GoodwillTableOut(
version_id=version_id, version_id=version_id,
goodwill_enabled=ctx.version.goodwill_enabled, goodwill_enabled=ctx.version.goodwill_enabled,
editable=reason is None, editable=reason is None,
read_only_reason=reason, read_only_reason=reason,
rows=output, rows=output,
groups=[
GoodwillGroupOut(
head_legal_entity=RefOut(id=head, name=entities[head].name),
totals=GoodwillTotalsOut(**groups[head]),
)
for head in heads
],
totals=GoodwillTotalsOut(**domain.totals(rows, ctx.version.goodwill_enabled)), totals=GoodwillTotalsOut(**domain.totals(rows, ctx.version.goodwill_enabled)),
) )
@ -189,9 +199,7 @@ class GoodwillService:
if row.has_investments: if row.has_investments:
assert row.financial_investment_amount is not None assert row.financial_investment_amount is not None
p = ( p = (
(row.charter_capital_amount * decision.ownership_share / 100).quantize( domain.participation(row.charter_capital_amount, decision.ownership_share)
Decimal("0.01"), rounding=ROUND_HALF_UP
)
if decision.include_in_kfo if decision.include_in_kfo
and row.charter_capital_amount is not None and row.charter_capital_amount is not None
and decision.ownership_share is not None and decision.ownership_share is not None

View File

@ -1,4 +1,4 @@
"""XLSX с колонками и группировкой экранной таблицы гудвилла.""" """XLSX с колонками и группировкой экранной таблицы гудвилла; итоги — из ответа таблицы (`groups`, `totals`)."""
from collections import defaultdict from collections import defaultdict
from decimal import Decimal from decimal import Decimal
@ -6,7 +6,7 @@ from io import BytesIO
from openpyxl import Workbook from openpyxl import Workbook
from src.schemas.goodwill import GoodwillRowOut, GoodwillSign, GoodwillTableOut from src.schemas.goodwill import GoodwillRowOut, GoodwillSign, GoodwillTableOut, GoodwillTotalsOut
from src.services.osv_export import _put_money, _put_text from src.services.osv_export import _put_money, _put_text
COLUMNS = { COLUMNS = {
@ -48,12 +48,13 @@ def build_workbook(table: GoodwillTableOut, sign: GoodwillSign = "all") -> bytes
else: else:
_put_text(sheet, row, col, value) _put_text(sheet, row, col, value)
if key == "ownership_share": if key == "ownership_share":
sheet.cell(row, col).number_format = "0.0000" sheet.cell(row, col).number_format = "0.00"
groups: dict = defaultdict(list) groups: dict = defaultdict(list)
for item in table.rows: for item in table.rows:
groups[item.head_legal_entity.id].append(item) groups[item.head_legal_entity.id].append(item)
for items in groups.values(): group_totals = {group.head_legal_entity.id: group.totals for group in table.groups}
for head, items in groups.items():
visible = [item for item in items if matches_sign(item, sign)] visible = [item for item in items if matches_sign(item, sign)]
if not visible: if not visible:
continue continue
@ -72,21 +73,12 @@ def build_workbook(table: GoodwillTableOut, sign: GoodwillSign = "all") -> bytes
comment=item.comment, comment=item.comment,
) )
append(values) append(values)
append({"name": "Всего", **_totals(items)}) append({"name": "Всего", **_money(group_totals[head])})
totals = table.totals append({"name": "Итого по листу", **_money(table.totals)})
values = dict(
name="Итого по листу",
ownership_share=None,
financial_investment_amount=totals.financial_investment_amount,
participation_amount=totals.participation_amount,
goodwill_amount=totals.goodwill_amount,
)
append(values)
stream = BytesIO() stream = BytesIO()
book.save(stream) book.save(stream)
return stream.getvalue() return stream.getvalue()
def _totals(rows: list[GoodwillRowOut]) -> dict[str, Decimal]: def _money(totals: GoodwillTotalsOut) -> dict[str, Decimal]:
active = [r for r in rows if r.has_investments] return {field: getattr(totals, field) for field in MONEY}
return {field: sum((getattr(r, field) or Decimal(0) for r in active), Decimal(0)) for field in MONEY}

View File

@ -97,6 +97,20 @@ async def test_table_no_write_and_visibility(db_session, auth_client, fixtures_d
assert row["charter_capital_amount"] == "100000000.00" assert row["charter_capital_amount"] == "100000000.00"
assert row["goodwill_amount"] is None assert row["goodwill_amount"] is None
assert row["has_investments"] is True assert row["has_investments"] is True
table = response.json()
assert table["totals"]["charter_capital_amount"] == "100000000.00"
assert table["groups"] == [
{
"head_legal_entity": {"id": str(w.a.id), "name": w.a.name},
"totals": {
"financial_investment_amount": "269997300.00",
"charter_capital_amount": "100000000.00",
"participation_amount": "0.00",
"goodwill_amount": "0.00",
"goodwill_in_kfo_amount": "0.00",
},
}
]
assert (await db_session.scalar(select(func.count()).select_from(Goodwill))) == 0 assert (await db_session.scalar(select(func.count()).select_from(Goodwill))) == 0
foreign = await auth_client(FOREIGN) foreign = await auth_client(FOREIGN)
assert (await foreign.get(base + "/goodwill")).status_code == HTTP_NOT_FOUND assert (await foreign.get(base + "/goodwill")).status_code == HTTP_NOT_FOUND

View File

@ -19,14 +19,16 @@ from tests.test_api_consolidated_osv import KA_RF
from tests.test_api_goodwill import goodwill_world from tests.test_api_goodwill import goodwill_world
@pytest.mark.parametrize("value", ["-1", "100.0001", "0.00001", "NaN", "Infinity", "abc", ""]) @pytest.mark.parametrize(
"value", ["-1", "100.01", "100.0001", "80.123", "80,123", "12.3456", "1e2", "NaN", "Infinity", "abc", ""]
)
def test_invalid_share(value): def test_invalid_share(value):
with pytest.raises(DomainValidationError) as error: with pytest.raises(DomainValidationError) as error:
validate_decision(GoodwillRowIn(ownership_share=value, include_in_kfo=False)) validate_decision(GoodwillRowIn(ownership_share=value, include_in_kfo=False))
assert error.value.code == "INVALID_SHARE" assert error.value.code == "INVALID_SHARE"
@pytest.mark.parametrize("value", ["0", "100", "12.3456", "12,3456"]) @pytest.mark.parametrize("value", ["0", "100", "7.5", "80.12", "80,12"])
def test_valid_share(value): def test_valid_share(value):
result = validate_decision(GoodwillRowIn(ownership_share=value, include_in_kfo=True, comment=" Обоснование ")) result = validate_decision(GoodwillRowIn(ownership_share=value, include_in_kfo=True, comment=" Обоснование "))
assert str(result.ownership_share) == value.replace(",", ".") assert str(result.ownership_share) == value.replace(",", ".")
@ -51,7 +53,7 @@ async def test_save_settings_and_disappearing_pair(db_session, auth_client, fixt
client = await auth_client(KA_RF) client = await auth_client(KA_RF)
base = f"/consolidations/versions/{w.version.id}" base = f"/consolidations/versions/{w.version.id}"
path = base + f"/goodwill/{w.a.id}/{w.b.id}" path = base + f"/goodwill/{w.a.id}/{w.b.id}"
body = dict(ownership_share="80.1234", include_in_kfo=True, comment="Подтверждено") body = dict(ownership_share="80.12", include_in_kfo=True, comment="Подтверждено")
response = await client.put(path, json=body) response = await client.put(path, json=body)
assert response.status_code == HTTPStatus.CONFLICT and response.json()["detail"]["code"] == "LOCK_REQUIRED" assert response.status_code == HTTPStatus.CONFLICT and response.json()["detail"]["code"] == "LOCK_REQUIRED"
assert (await client.post(base + "/lock")).status_code == HTTPStatus.CREATED assert (await client.post(base + "/lock")).status_code == HTTPStatus.CREATED
@ -62,15 +64,15 @@ async def test_save_settings_and_disappearing_pair(db_session, auth_client, fixt
response = await client.put(path, json=body) response = await client.put(path, json=body)
assert response.status_code == HTTPStatus.OK, response.text assert response.status_code == HTTPStatus.OK, response.text
row = response.json() row = response.json()
assert row["ownership_share"] == "80.1234" assert row["ownership_share"] == "80.12"
assert row["participation_amount"] == "80123400.00" assert row["participation_amount"] == "80120000.00"
assert row["goodwill_amount"] == "189873900.00" assert row["goodwill_amount"] == "189877300.00"
again = await client.put(path, json=body) again = await client.put(path, json=body)
assert again.status_code == HTTPStatus.OK assert again.status_code == HTTPStatus.OK
assert await db_session.scalar(select(func.count()).select_from(Goodwill)) == 1 assert await db_session.scalar(select(func.count()).select_from(Goodwill)) == 1
on = await client.put(base + "/goodwill-settings", json=dict(goodwill_enabled=True)) on = await client.put(base + "/goodwill-settings", json=dict(goodwill_enabled=True))
assert on.status_code == HTTPStatus.OK assert on.status_code == HTTPStatus.OK
assert on.json()["totals"]["goodwill_in_kfo_amount"] == "189873900.00" assert on.json()["totals"]["goodwill_in_kfo_amount"] == "189877300.00"
audit_count = await db_session.scalar( audit_count = await db_session.scalar(
select(func.count()).select_from(AuditEvent).where(AuditEvent.action == AuditAction.GOODWILL_EDIT) select(func.count()).select_from(AuditEvent).where(AuditEvent.action == AuditAction.GOODWILL_EDIT)
) )
@ -82,7 +84,7 @@ async def test_save_settings_and_disappearing_pair(db_session, auth_client, fixt
) )
off = await client.put(base + "/goodwill-settings", json=dict(goodwill_enabled=False)) off = await client.put(base + "/goodwill-settings", json=dict(goodwill_enabled=False))
assert off.json()["totals"]["goodwill_in_kfo_amount"] == "0.00" assert off.json()["totals"]["goodwill_in_kfo_amount"] == "0.00"
assert off.json()["rows"][0]["ownership_share"] == "80.1234" assert off.json()["rows"][0]["ownership_share"] == "80.12"
body["include_in_kfo"] = False body["include_in_kfo"] = False
assert (await client.put(path, json=body)).status_code == HTTPStatus.OK assert (await client.put(path, json=body)).status_code == HTTPStatus.OK
assert (await client.get(base + "/goodwill")).json()["totals"]["goodwill_amount"] == "0.00" assert (await client.get(base + "/goodwill")).json()["totals"]["goodwill_amount"] == "0.00"

View File

@ -6,7 +6,7 @@ from uuid import UUID
import pytest import pytest
from src.domain.goodwill import Decision, Investment, build, totals from src.domain.goodwill import Decision, Investment, build, group_totals, participation, totals
from tests.test_goodwill_golden import investment_rows from tests.test_goodwill_golden import investment_rows
pytestmark = pytest.mark.fixtures pytestmark = pytest.mark.fixtures
@ -50,3 +50,42 @@ def test_rounding_and_unselected(fixtures_dir: Path) -> None:
assert selected.participation_amount == Decimal("0.01") assert selected.participation_amount == Decimal("0.01")
assert totals([row], True)["goodwill_amount"] == 0 assert totals([row], True)["goodwill_amount"] == 0
assert totals([row], True)["financial_investment_amount"] == fv assert totals([row], True)["financial_investment_amount"] == fv
def test_participation_rounds_half_up() -> None:
assert participation(Decimal("371.17"), Decimal("99.99")) == Decimal("371.13")
assert participation(Decimal("0.05"), Decimal("50")) == Decimal("0.03")
assert participation(Decimal("100000"), Decimal("0")) == Decimal("0.00")
def test_doc_example_totals() -> None:
"""DOC-259025 §11.1: Альфа → Бета (+100 000), Гамма (−30 000), Дельта без доли; плюс пара «Нет вложений»."""
alpha, beta, gamma, delta, gone = (UUID(int=i) for i in (10, 11, 12, 13, 14))
source = [
Investment(alpha, beta, Decimal("180000.00")),
Investment(alpha, gamma, Decimal("50000.00")),
Investment(alpha, delta, Decimal("60000.00")),
]
saved = {
(alpha, beta): Decision(Decimal("80"), True, "Устав"),
(alpha, gamma): Decision(Decimal("80"), True, "Устав"),
(alpha, gone): Decision(Decimal("50"), True, "Было"),
}
capital = {beta: Decimal("100000.00"), gamma: Decimal("100000.00"), delta: Decimal("60000.00")}
rows = {r.subsidiary_id: r for r in build(source, saved, capital)}
assert rows[beta].participation_amount == Decimal("80000.00")
assert rows[beta].goodwill_amount == Decimal("100000.00")
assert rows[gamma].participation_amount == Decimal("80000.00")
assert rows[gamma].goodwill_amount == Decimal("-30000.00")
assert rows[delta].participation_amount is None and rows[delta].goodwill_amount is None
assert not rows[gone].has_investments
expected = {
"financial_investment_amount": Decimal("290000.00"),
"charter_capital_amount": Decimal("260000.00"),
"participation_amount": Decimal("160000.00"),
"goodwill_amount": Decimal("70000.00"),
"goodwill_in_kfo_amount": Decimal("100000.00"),
}
assert totals(list(rows.values()), True) == expected
assert totals(list(rows.values()), False)["goodwill_in_kfo_amount"] == 0
assert group_totals(list(rows.values()), True) == {alpha: expected}

View File

@ -76,3 +76,15 @@ async def test_export_matches_server_table(db_session, auth_client, fixtures_dir
assert sheet.cell(3, 7).value is None assert sheet.cell(3, 7).value is None
assert sheet.cell(5, 4).value == 0 assert sheet.cell(5, 4).value == 0
assert sheet.cell(5, 7).value == 0 assert sheet.cell(5, 7).value == 0
async def test_export_totals_come_from_table(db_session, auth_client, fixtures_dir):
world = await goodwill_world(db_session, fixtures_dir)
client = await auth_client(KA_RF)
base = f"/consolidations/versions/{world.version.id}"
table = (await client.get(base + "/goodwill")).json()
sheet = load_workbook(BytesIO((await client.get(base + "/goodwill/export")).content)).active
assert sheet.cell(4, 2).value == "Всего"
assert sheet.cell(4, 6).value == float(table["groups"][0]["totals"]["charter_capital_amount"])
assert sheet.cell(5, 2).value == "Итого по листу"
assert sheet.cell(5, 6).value == float(table["totals"]["charter_capital_amount"])