feat: шаг 7 — посчитанные соотношения §7, abs_terms, допуск и округление, упрощённый набор, подстроки
Добавлены evaluation.py (EvalLine/EvalValue/is_check_line) и checks.py (form_variant, subline_parents, evaluate_relations, checks_from). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W
This commit is contained in:
Andrey Epifancev
parent
0c4dc6d221
commit
3df179b7a3
184
arm_statements/checks.py
Normal file
184
arm_statements/checks.py
Normal file
@ -0,0 +1,184 @@
|
||||
"""Шаг 7: самопроверки §7 и вариант формы."""
|
||||
|
||||
from collections.abc import Iterable, Mapping, Sequence
|
||||
from dataclasses import dataclass
|
||||
from decimal import Decimal
|
||||
|
||||
from arm_statements.codes import document_type_of, section_of
|
||||
from arm_statements.evaluation import EvalLine, is_check_line
|
||||
from arm_statements.loader import RelationRule, Rules
|
||||
from arm_statements.model import DocumentType, FailedRelation, FormVariant, StatementChecks
|
||||
|
||||
_ZERO = Decimal("0")
|
||||
_HUNDRED = Decimal("100")
|
||||
_PARENT_LEN = 4
|
||||
_SUBLINE_LEN = 5
|
||||
_GROUP_LEN = 3 # первые цифры, общие у строки и её подстрок (1230 → 12301…12314)
|
||||
|
||||
|
||||
def form_variant(document_type: DocumentType, ok_codes: Iterable[str]) -> FormVariant | None:
|
||||
"""Вариант формы по структуре кодов (спека «Шаг 6»); ОКУД признаком не служит."""
|
||||
codes = set(ok_codes)
|
||||
if document_type is DocumentType.BALANCE:
|
||||
simplified = "1600" in codes and not {"1100", "1200"} & codes
|
||||
elif document_type is DocumentType.INCOME:
|
||||
simplified = bool({"2400", "2110"} & codes) and not {"2100", "2200", "2300"} & codes
|
||||
else:
|
||||
return None
|
||||
return FormVariant.SIMPLIFIED if simplified else FormVariant.FULL
|
||||
|
||||
|
||||
def subline_parents(codes: Sequence[str]) -> list[str | None]:
|
||||
"""Родитель каждой позиции: у пятизначного кода — ближайший четырёхзначный выше с теми же тремя цифрами,
|
||||
иначе первые четыре цифры; у прочих — `None`."""
|
||||
result: list[str | None] = []
|
||||
last_parent: str | None = None
|
||||
for code in codes:
|
||||
if len(code) == _PARENT_LEN:
|
||||
last_parent = code
|
||||
result.append(None)
|
||||
elif len(code) == _SUBLINE_LEN:
|
||||
if last_parent is not None and last_parent[:_GROUP_LEN] == code[:_GROUP_LEN]:
|
||||
result.append(last_parent)
|
||||
else:
|
||||
result.append(code[:_PARENT_LEN])
|
||||
else:
|
||||
result.append(None)
|
||||
return result
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class RelationOutcome:
|
||||
column: str
|
||||
rule: RelationRule
|
||||
codes: tuple[str, ...]
|
||||
computed: bool
|
||||
expected: Decimal | None # None — цель пуста
|
||||
actual: Decimal | None # None — пусты все члены
|
||||
within_tolerance: bool
|
||||
within_rounding: bool
|
||||
|
||||
@property
|
||||
def hard_failed(self) -> bool:
|
||||
return self.rule.strict and self.computed and not self.within_tolerance and not self.within_rounding
|
||||
|
||||
|
||||
def _column_sums(lines: Sequence[EvalLine], column: str) -> tuple[dict[str, Decimal], set[str]]:
|
||||
"""Код → сумма `amount` по строкам `is_check_line` в колонке; второй результат — коды с непустой суммой."""
|
||||
sums: dict[str, Decimal] = {}
|
||||
for item in lines:
|
||||
if not is_check_line(item) or item.line_code is None:
|
||||
continue
|
||||
for value in item.values:
|
||||
if value.column == column and value.amount is not None:
|
||||
sums[item.line_code] = sums.get(item.line_code, _ZERO) + value.amount
|
||||
return sums, set(sums)
|
||||
|
||||
|
||||
def _rhs(sums: Mapping[str, Decimal], rule: RelationRule, *, absolute: bool) -> Decimal:
|
||||
total = _ZERO
|
||||
for sign, code in rule.terms:
|
||||
value = sums.get(code, _ZERO)
|
||||
if absolute and code in rule.abs_terms:
|
||||
value = abs(value)
|
||||
total += sign * value
|
||||
return total
|
||||
|
||||
|
||||
def _actual(sums: Mapping[str, Decimal], rule: RelationRule, expected: Decimal, tolerance: Decimal) -> Decimal:
|
||||
"""Правая часть: сошедшийся вариант (как записано, затем по модулю), иначе ближайший к цели."""
|
||||
plain = _rhs(sums, rule, absolute=False)
|
||||
if abs(expected - plain) <= tolerance or not rule.abs_terms:
|
||||
return plain
|
||||
with_abs = _rhs(sums, rule, absolute=True)
|
||||
if abs(expected - with_abs) <= tolerance or abs(expected - with_abs) < abs(expected - plain):
|
||||
return with_abs
|
||||
return plain
|
||||
|
||||
|
||||
def _outcome(
|
||||
column: str, rule: RelationRule, sums: Mapping[str, Decimal], filled: set[str], rules: Rules
|
||||
) -> RelationOutcome:
|
||||
codes = (rule.target, *(code for _, code in rule.terms))
|
||||
target_filled = rule.target in filled
|
||||
members_filled = any(code in filled for _, code in rule.terms)
|
||||
if not target_filled and not members_filled:
|
||||
return RelationOutcome(column, rule, codes, False, None, None, True, True)
|
||||
expected = sums[rule.target] if target_filled else None
|
||||
if expected is None or not members_filled:
|
||||
actual = _rhs(sums, rule, absolute=False) if members_filled else None
|
||||
return RelationOutcome(column, rule, codes, True, expected, actual, False, False)
|
||||
actual = _actual(sums, rule, expected, rules.checks.tolerance)
|
||||
diff = abs(expected - actual)
|
||||
scale = max(abs(expected), *(abs(sums.get(code, _ZERO)) for _, code in rule.terms))
|
||||
within_rounding = diff <= scale * rules.checks.rounding_tolerance_percent / _HUNDRED
|
||||
return RelationOutcome(column, rule, codes, True, expected, actual, diff <= rules.checks.tolerance, within_rounding)
|
||||
|
||||
|
||||
def _subline_rules(
|
||||
lines: Sequence[EvalLine], document_type: DocumentType, variant: FormVariant, rules: Rules
|
||||
) -> list[RelationRule]:
|
||||
"""«родитель = Σ подстрок» для родителей из отрезка формы (мягкие), в порядке строк документа."""
|
||||
codes = [item.line_code for item in lines if is_check_line(item) and item.line_code is not None]
|
||||
children: dict[str, list[str]] = {}
|
||||
for code, parent in zip(codes, subline_parents(codes), strict=True):
|
||||
if parent is None or code in children.get(parent, []):
|
||||
continue
|
||||
section = section_of(parent, rules)
|
||||
if section is None or document_type_of(section) is not document_type:
|
||||
continue
|
||||
children.setdefault(parent, []).append(code)
|
||||
return [
|
||||
RelationRule(
|
||||
form=document_type,
|
||||
variant=variant,
|
||||
relation=f"{parent} = {' + '.join(kids)}",
|
||||
target=parent,
|
||||
terms=tuple((1, kid) for kid in kids),
|
||||
strict=False,
|
||||
abs_terms=frozenset(),
|
||||
)
|
||||
for parent, kids in children.items()
|
||||
]
|
||||
|
||||
|
||||
def evaluate_relations(
|
||||
lines: Sequence[EvalLine],
|
||||
columns: Sequence[str],
|
||||
document_type: DocumentType,
|
||||
variant: FormVariant | None,
|
||||
rules: Rules,
|
||||
) -> list[RelationOutcome]:
|
||||
"""Все соотношения набора формы и подстрок по каждой колонке. `lines` — в порядке документа."""
|
||||
effective = variant or FormVariant.FULL
|
||||
static = list(rules.checks.relations_for(document_type, effective))
|
||||
generated = _subline_rules(lines, document_type, effective, rules) if rules.checks.subline_check else []
|
||||
outcomes: list[RelationOutcome] = []
|
||||
for column in columns:
|
||||
sums, filled = _column_sums(lines, column)
|
||||
outcomes.extend(_outcome(column, rule, sums, filled, rules) for rule in static)
|
||||
# подстроки — только если хоть одна подстрока в колонке непуста (AURORA-1527)
|
||||
outcomes.extend(
|
||||
_outcome(column, rule, sums, filled, rules)
|
||||
for rule in generated
|
||||
if any(code in filled for _, code in rule.terms)
|
||||
)
|
||||
return outcomes
|
||||
|
||||
|
||||
def checks_from(outcomes: Sequence[RelationOutcome], column_index_of: Mapping[str, int | None]) -> StatementChecks:
|
||||
hard: list[FailedRelation] = []
|
||||
soft: list[FailedRelation] = []
|
||||
for item in outcomes:
|
||||
if not item.computed or item.within_tolerance:
|
||||
continue
|
||||
failure = FailedRelation(
|
||||
column_index=column_index_of.get(item.column),
|
||||
column_key=item.column,
|
||||
relation=item.rule.relation,
|
||||
codes=item.codes,
|
||||
expected=item.expected,
|
||||
actual=item.actual,
|
||||
)
|
||||
(hard if item.hard_failed else soft).append(failure)
|
||||
return StatementChecks(relations_failed=tuple(hard), soft_relations_failed=tuple(soft))
|
||||
36
arm_statements/evaluation.py
Normal file
36
arm_statements/evaluation.py
Normal file
@ -0,0 +1,36 @@
|
||||
"""Общий вид строк шагов 6–8 у сервиса и ядра."""
|
||||
|
||||
from dataclasses import dataclass
|
||||
from decimal import Decimal
|
||||
|
||||
from arm_statements.model import LineStatus
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class EvalValue:
|
||||
column: str # ключ колонки: str(column_index) у сервиса, ключ хоста у ядра
|
||||
amount: Decimal | None
|
||||
text: str | None
|
||||
confidence: float | None
|
||||
is_recognized: bool
|
||||
fixed: bool = False # правлено или подтверждено аналитиком
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class EvalLine:
|
||||
key: str # ref у сервиса, ключ хоста у ядра
|
||||
order: int | None # row_index; у ядра None — строка правки
|
||||
line_status: LineStatus
|
||||
line_code: str | None
|
||||
code_confidence: float | None
|
||||
code_fixed: bool
|
||||
values: tuple[EvalValue, ...]
|
||||
|
||||
|
||||
def is_check_line(line: EvalLine) -> bool:
|
||||
"""Строка участвует в §7: `ok` с кодом; строка правки (`order is None`) — только с суммой (P1-8)."""
|
||||
if line.line_status is not LineStatus.OK or line.line_code is None:
|
||||
return False
|
||||
if line.order is None:
|
||||
return any(value.amount is not None for value in line.values)
|
||||
return True
|
||||
107
arm_statements/tests/test_checks.py
Normal file
107
arm_statements/tests/test_checks.py
Normal file
@ -0,0 +1,107 @@
|
||||
from decimal import Decimal as D
|
||||
|
||||
from arm_statements.checks import checks_from, evaluate_relations, form_variant, subline_parents
|
||||
from arm_statements.evaluation import EvalLine, EvalValue, is_check_line
|
||||
from arm_statements.loader import Rules
|
||||
from arm_statements.model import DocumentType, FormVariant, LineStatus, StatementChecks
|
||||
|
||||
BAL, INC = DocumentType.BALANCE, DocumentType.INCOME
|
||||
|
||||
|
||||
def line(code: str | None, *amounts: str | None, status: LineStatus = LineStatus.OK, order: int = 0) -> EvalLine:
|
||||
values = tuple(EvalValue(str(i), None if a is None else D(a), a, 100.0, True) for i, a in enumerate(amounts))
|
||||
return EvalLine(f"k{order}{code}", order, status, code, 100.0, False, values)
|
||||
|
||||
|
||||
def failed(lines: list[EvalLine], doc: DocumentType, variant: FormVariant | None, rules: Rules) -> StatementChecks:
|
||||
outcomes = evaluate_relations(lines, ["0"], doc, variant, rules)
|
||||
return checks_from(outcomes, {"0": 0})
|
||||
|
||||
|
||||
def test_form_variant() -> None:
|
||||
assert form_variant(BAL, ["1150", "1210", "1600", "1310", "1700"]) is FormVariant.SIMPLIFIED
|
||||
assert form_variant(BAL, ["1100", "1600"]) is FormVariant.FULL
|
||||
assert form_variant(INC, ["2110", "2400"]) is FormVariant.SIMPLIFIED
|
||||
assert form_variant(INC, ["2110", "2100", "2400"]) is FormVariant.FULL
|
||||
assert form_variant(DocumentType.DISCLOSURE, ["5101"]) is None
|
||||
|
||||
|
||||
def test_empty_target_is_violation(rules: Rules) -> None:
|
||||
lines = [
|
||||
line("1110", "30"),
|
||||
line("1100", "30"),
|
||||
line("1210", "70"),
|
||||
line("1200", "70"),
|
||||
line(None, "100", status=LineStatus.DETAIL),
|
||||
]
|
||||
checks = failed(lines, BAL, FormVariant.FULL, rules)
|
||||
assert [f.relation for f in checks.relations_failed] == ["1600 = 1100 + 1200"]
|
||||
assert (checks.relations_failed[0].expected, checks.relations_failed[0].actual) == (None, D("100"))
|
||||
|
||||
|
||||
def test_target_without_members_is_violation(rules: Rules) -> None:
|
||||
lines = [line("1110", "30"), line("1100", "30"), line("1210", "70"), line("1200", "70"), line("1600", "100")]
|
||||
checks = failed(lines, BAL, FormVariant.FULL, rules)
|
||||
assert [f.relation for f in checks.relations_failed] == ["1600 = 1700"]
|
||||
assert checks.relations_failed[0].codes == ("1600", "1700")
|
||||
assert (checks.relations_failed[0].expected, checks.relations_failed[0].actual) == (D("100"), None)
|
||||
|
||||
|
||||
def test_zero_target_with_empty_members_is_violation(rules: Rules) -> None:
|
||||
lines = [line("1600", "0"), line("1110", "0"), line("1100", "0")]
|
||||
hard = failed(lines, BAL, FormVariant.FULL, rules).relations_failed
|
||||
assert [(f.relation, f.expected, f.actual) for f in hard] == [("1600 = 1700", D("0"), None)]
|
||||
|
||||
|
||||
def test_duplicate_code_is_summed(rules: Rules) -> None:
|
||||
lines = [line("2300", "849"), line("2410", "100"), line("2410", "649"), line("2400", "100")]
|
||||
outcomes = evaluate_relations(lines, ["0"], INC, FormVariant.FULL, rules)
|
||||
target = next(o for o in outcomes if o.rule.relation == "2400 = 2300 - 2410 + 2420")
|
||||
assert target.within_tolerance
|
||||
|
||||
|
||||
def test_simplified_income_soft(rules: Rules) -> None:
|
||||
lines = [line("2110", "1000"), line("2120", "-600"), line("2410", "50"), line("2400", "500")]
|
||||
checks = failed(lines, INC, FormVariant.SIMPLIFIED, rules)
|
||||
assert checks.relations_failed == ()
|
||||
assert [f.relation for f in checks.soft_relations_failed] == ["2400 = 2110 - 2120 - 2330 + 2340 - 2350 - 2410"]
|
||||
|
||||
|
||||
def test_rounding_goes_soft(rules: Rules) -> None:
|
||||
lines = [
|
||||
line("1600", "17000000009.02"),
|
||||
line("1700", "17000000000"),
|
||||
line("1100", "17000000009.02"),
|
||||
line("1300", "17000000000"),
|
||||
]
|
||||
checks = failed(lines, BAL, FormVariant.FULL, rules)
|
||||
assert "1600 = 1700" in [f.relation for f in checks.soft_relations_failed]
|
||||
assert "1600 = 1700" not in [f.relation for f in checks.relations_failed]
|
||||
|
||||
|
||||
def test_abs_terms_hold_with_minus(rules: Rules) -> None:
|
||||
lines = [line("2110", "100"), line("2120", "-60"), line("2100", "40")]
|
||||
hard = failed(lines, INC, FormVariant.FULL, rules).relations_failed
|
||||
assert "2100 = 2110 - 2120" not in [f.relation for f in hard]
|
||||
|
||||
|
||||
def test_sublines(rules: Rules) -> None:
|
||||
assert subline_parents(["1230", "12301", "12302", "1240", "11101"]) == [None, "1230", "1230", None, "1110"]
|
||||
# расхождение 2 (>1): в плане было 10 против 4 + 5 = 9, это в пределах допуска ±1 — спека «Шаг 7» (T13.1)
|
||||
lines = [line("1230", "10"), line("12301", "4"), line("12302", "4")]
|
||||
soft = failed(lines, BAL, FormVariant.FULL, rules).soft_relations_failed
|
||||
assert [f.relation for f in soft] == ["1230 = 12301 + 12302"]
|
||||
within = [line("1230", "10"), line("12301", "4"), line("12302", "5")]
|
||||
assert failed(within, BAL, FormVariant.FULL, rules).soft_relations_failed == ()
|
||||
empty = [line("1230", "10"), line("12301", None), line("12302", None)]
|
||||
assert failed(empty, BAL, FormVariant.FULL, rules).soft_relations_failed == ()
|
||||
|
||||
|
||||
def test_is_check_line() -> None:
|
||||
assert is_check_line(line("1600", "1"))
|
||||
assert not is_check_line(line("1600", "1", status=LineStatus.DETAIL))
|
||||
assert not is_check_line(line(None, "1"))
|
||||
correction = EvalLine("c", None, LineStatus.OK, "1600", 100.0, False, line("1600", None).values)
|
||||
assert not is_check_line(correction)
|
||||
with_amount = EvalLine("c", None, LineStatus.OK, "1600", 100.0, False, line("1600", "5").values)
|
||||
assert is_check_line(with_amount)
|
||||
Loading…
x
Reference in New Issue
Block a user