"""Сквозной пример §8 «КФО» (docs #206): мини-движок по design.md §4.1, числа для §8 и examples контракта. Запуск: python3 -I kfo_example.py [--json out.json] """ import json import sys from decimal import ROUND_HALF_UP, Decimal as D K = D("1000") EXPENSE = {"2120", "2210", "2220", "2330", "2350"} COMPANIES = ["A", "B", "C"] NAMES = {"A": "ООО «Альфа»", "B": "ООО «Бета»", "C": "ООО «Гамма»"} INN = {"A": "7701000001", "B": "7701000002", "C": "7701000003"} # Формы, тыс. руб. (как в форме; знак расходов у «Беты» — «без знака») BB = { "A": {"1170": 70000, "1230": 520000, "1100": 70000, "1200": 520000, "1600": 590000, "1310": 100000, "1350": 0, "1370": 280000, "1300": 380000, "1410": 50000, "1400": 50000, "1520": 160000, "1500": 160000, "1700": 590000}, "B": {"1170": 0, "1230": 130000, "1100": 0, "1200": 130000, "1600": 130000, "1310": 40000, "1350": 0, "1370": -20000, "1300": 20000, "1410": 0, "1400": 0, "1520": 110000, "1500": 110000, "1700": 130000}, "C": {"1170": 0, "1230": 48000, "1100": 0, "1200": 48000, "1600": 48000, "1310": 10000, "1350": 3000, "1370": 5000, "1300": 18000, "1410": 0, "1400": 0, "1520": 30000, "1500": 30000, "1700": 48000}, } OFR = { "A": {"2110": 200000, "2120": -120000, "2100": 80000, "2210": -20000, "2220": -10000, "2200": 50000}, "B": {"2110": 80000, "2120": 60000, "2100": 20000, "2210": -5000, "2220": 0, "2200": 15000}, # «Гамма» — формы ОФР нет: FORM_MISSING } BB_LINES = ["1170", "GOODWILL", "1100", "1230", "1200", "1600", "1310", "1350", "1370", "1300", "1410", "1400", "1520", "1500", "1700"] OFR_LINES = ["2110", "2120", "2100", "2210", "2220", "2200"] TOTALS = { # итог ← слагаемые (с учётом знака, значения уже со знаком) "1100": ["1170", "GOODWILL"], "1200": ["1230"], "1600": ["1100", "1200"], "1300": ["1310", "1350", "1370"], "1400": ["1410"], "1500": ["1520"], "1700": ["1300", "1400", "1500"], "2100": ["2110", "2120"], "2200": ["2100", "2210", "2220"], } # Агрегаты таблицы корректировок (рубли). rule: (mode, article_1, article_2) RULES = {2: ("min_or_full", "1170", "1310/1350/1370"), 6: ("min", "1230", "1520"), "ofr1": ("full_side1_and_side2", "2110", "2120")} AGGS = [ # (rule, c1, c2, amount_1, amount_2, correction) (6, "A", "B", D("102000000.00"), D("100000400.50"), None), (6, "B", "A", D("10000000.00"), D("50000000.00"), {"debit": [("1230", D("10000000.00"))], "credit": [("1520", D("10000000.00"))]}), (6, "A", "C", D("30000000.00"), D("29100000.00"), {"debit": [("1230", D("20000000.00")), ("1170", D("5000000.00"))], "credit": [("1520", D("22500000.00"))]}), (2, "A", "B", D("50000000.00"), D("40000000.00"), None), (2, "A", "C", D("15000000.00"), D("0.00"), None), ("ofr1", "A", "B", D("10000000.00"), D("10000000.00"), None), # S без НДС (правила 1999) ] GOODWILL_PAIRS = [ # ГО, ДО, ФВ, УК (руб.), доля %, отмечена ("A", "B", D("50000000.00"), D("40000000.00"), D("80.00"), True), ] THRESHOLD = D("5") def r2(x): return x.quantize(D("0.01"), ROUND_HALF_UP) def k_int(rub): # рубли → целые тысячи, половина от нуля return (rub / K).quantize(D("1"), ROUND_HALF_UP) def split_vgo(auto_rub, manual_rub, sign): """Ячейка компания × строка: vgo — округлённая Σ исключений; manual округляется так же; auto = vgo − manual.""" vgo = sign * k_int(auto_rub + manual_rub) man = sign * k_int(manual_rub) return vgo, vgo - man, man def auto_amount(mode, a1, a2): """Автосумма агрегата по elimination_mode (без порога для min — порог проверяется отдельно).""" if mode == "min": return min(a1, a2) if a1 > 0 and a2 > 0 else D(0) if mode == "min_or_full": if a1 <= 0: return D(0) return min(a1, a2) if a2 > 0 else a1 return D(0) def agg_deviation(mode, a1, a2): """Расхождение агрегата для порога: только режим min, в том числе односторонний.""" return abs(a1 - a2) if mode == "min" else D(0) def share(cleaned, bank): if not bank: return None x = r2(cleaned / bank * 100) return None if abs(x) >= 1000 else x def total_client(form_c, line, parts): """Итог «по данным клиента»: из формы, а если строки итога в форме нет — Σ слагаемых.""" if line in form_c: return D(form_c[line]) return sum((D(form_c.get(p_, 0)) for p_ in parts), D(0)) def client(form, c, line): v = D(form.get(c, {}).get(line, 0)) return -abs(v) if line in EXPENSE else v def compute(goodwill_enabled=True): excl = {} # (c, line) -> рубли auto, manual = {}, {} dev = {} log = [] def add(c, line, amt, kind): excl[(c, line)] = excl.get((c, line), D(0)) + amt (auto if kind == "auto" else manual)[(c, line)] = (auto if kind == "auto" else manual).get((c, line), D(0)) + amt log.append((c, line, amt, kind)) def available(c, line): return max(client(BB, c, line) * K - excl.get((c, line), D(0)), D(0)) def cascade(c, lines, amt, kind): codes = lines.split("/") for i, code in enumerate(codes): take = amt if i == len(codes) - 1 else min(amt, available(c, code)) if take > 0: add(c, code, take, kind) amt -= take if amt <= 0: break gw_applied = [] g_total = D(0) for go, do, fv, uk, d, marked in GOODWILL_PAIRS: p = r2(uk * d / 100) g = fv - p if goodwill_enabled and marked and g > 0: gw_applied.append((go, do)) add(go, "1170", fv, "auto") add(do, "1310", p, "auto") g_total += g order = sorted(AGGS, key=lambda a: (str(a[0]), INN[a[1]], INN[a[2]])) for rule, c1, c2, a1, a2, corr in order: mode, art1, art2 = RULES[rule] dv = agg_deviation(mode, a1, a2) if dv: line_dev = (c1, art1) if a1 > a2 else (c2, art2) dev[line_dev] = dev.get(line_dev, D(0)) + dv if rule == 2 and (c1, c2) in gw_applied: continue # гудвилл-пара подавляет агрегат правила 2 if corr: dt = sum((x for _, x in corr["debit"]), D(0)) kt = sum((x for _, x in corr["credit"]), D(0)) lim = min(dt, kt) if dt > 0 and kt > 0 else None for side, c in (("debit", c1), ("credit", c2)): rest = lim for line, x in corr[side]: take = x if rest is None else min(x, rest) if take > 0: add(c, line, take, "manual") if rest is not None: rest -= take continue if mode == "min": pct = abs(a1 - a2) / min(a1, a2) * 100 if a1 > 0 and a2 > 0 else None s = auto_amount(mode, a1, a2) if pct is not None and pct <= THRESHOLD else D(0) if s: add(c1, art1, s, "auto"); add(c2, art2, s, "auto") elif mode == "min_or_full": s = auto_amount(mode, a1, a2) if s: add(c1, art1, s, "auto"); cascade(c2, art2, s, "auto") elif mode == "full_side1_and_side2": add(c1, art1, a1, "auto"); add(c2, art2, a2, "auto") def build(form, lines, stype): out = {} for line in lines: if line == "GOODWILL": ga = k_int(g_total) out[line] = {"companies": None, "sum": D(0), "vgo": D(0), "auto": D(0), "manual": D(0), "goodwill_adjust": ga, "bank": ga, "dev": D(0)} continue comps = {} if line in TOTALS: for c in COMPANIES: cl = total_client(form[c], line, TOTALS[line]) if c in form else D(0) vg = sum((out[s]["companies"][c]["vgo"] for s in TOTALS[line] if out[s]["companies"]), D(0)) comps[c] = {"client": cl, "vgo": vg, "cleaned": cl - vg} ga = sum((out[s]["goodwill_adjust"] for s in TOTALS[line]), D(0)) au = sum((out[s]["auto"] for s in TOTALS[line]), D(0)) ma = sum((out[s]["manual"] for s in TOTALS[line]), D(0)) dv = None # у итоговых строк расхождений нет else: ga = D(0) au = ma = D(0) for c in COMPANIES: cl = client(form, c, line) if c in form else D(0) sign = -1 if line in EXPENSE else 1 vg, a_, m_ = split_vgo(auto.get((c, line), D(0)), manual.get((c, line), D(0)), sign) comps[c] = {"client": cl, "vgo": vg, "cleaned": cl - vg} au += a_ ma += m_ dv = k_int(sum((v for (c, l), v in dev.items() if l == line), D(0))) dv = D(0) if stype == "income" else dv s = sum((x["client"] for x in comps.values()), D(0)) vgo = sum((x["vgo"] for x in comps.values()), D(0)) bank = s - vgo + ga for x in comps.values(): x["share"] = share(x["cleaned"], bank) out[line] = {"companies": comps, "sum": s, "vgo": vgo, "auto": au, "manual": ma, "goodwill_adjust": ga, "bank": bank, "dev": dv} return out bb = build(BB, BB_LINES, "balance") ofr = build(OFR, OFR_LINES, "income") # Расхождения: только строки бланка, не итоги dev_bb = sum((v["dev"] for k_, v in bb.items() if k_ not in TOTALS), D(0)) na, vb = bb["1300"]["bank"], bb["1600"]["bank"] if na >= vb * D("0.10"): base_bb, lim_pct_bb, base_kind_bb = na, D("10"), "NET_ASSETS" else: base_bb, lim_pct_bb, base_kind_bb = vb, D("1"), "BALANCE_TOTAL" lim_bb = r2(base_bb * lim_pct_bb / 100) rev, sp = ofr["2110"]["bank"], ofr["2200"]["bank"] dev_ofr = D(0) if rev != 0 and sp >= 0 and sp / rev >= D("0.10"): base_ofr, lim_pct_ofr, base_kind_ofr = sp, D("10"), "SALES_PROFIT" else: base_ofr, lim_pct_ofr, base_kind_ofr = rev, D("1"), "REVENUE" lim_ofr = r2(base_ofr * lim_pct_ofr / 100) summary = { "balance": {"deviation_total": dev_bb, "base_kind": base_kind_bb, "base_amount": base_bb, "limit_percent": lim_pct_bb, "limit_amount": lim_bb, "deviation_percent": r2(dev_bb / base_bb * 100) if base_bb else None, "exceeded": dev_bb > lim_bb}, "income": {"deviation_total": dev_ofr, "base_kind": base_kind_ofr, "base_amount": base_ofr, "limit_percent": lim_pct_ofr, "limit_amount": lim_ofr, "deviation_percent": r2(dev_ofr / base_ofr * 100) if base_ofr else None, "exceeded": dev_ofr > lim_ofr}, } # Проверки (Review Focus) assert bb["1600"]["bank"] == bb["1700"]["bank"], (bb["1600"]["bank"], bb["1700"]["bank"]) assert ofr["2100"]["vgo"] == 0, ofr["2100"]["vgo"] assert ofr["2100"]["bank"] == ofr["2100"]["sum"] # агрегат с корректировкой: B→A правило 6 — только 10 000 тыс., автосуммы нет assert bb["1230"]["companies"]["B"]["vgo"] == D("10000") # каскад правила 2 (A→C): Σ у «Гаммы» по 1310/1350/1370 = 15 000 = исключению из 1170 по этой паре c_r2 = sum((amt for c, l, amt, _ in log if c == "C" and l in ("1310", "1350", "1370")), D(0)) assert c_r2 == D("15000000.00"), c_r2 assert k_int(excl[("C", "1370")]) == D("2000") # меньший итог сторон корректировки A→C: 25 000 vs 22 500 → 22 500 с каждой стороны a_from_ac = sum((amt for c, l, amt, kind in log if c == "A" and kind == "manual" and l in ("1230", "1170")), D(0)) assert a_from_ac == D("22500000.00"), a_from_ac if goodwill_enabled: assert bb["GOODWILL"]["bank"] == D("18000") assert bb["1100"]["goodwill_adjust"] == D("18000") and bb["1600"]["goodwill_adjust"] == D("18000") else: assert bb["GOODWILL"]["bank"] == 0 return bb, ofr, summary, g_total def show(title, rows, lines): print(f"\n## {title}") print("| Строка | " + " | ".join(f"{c}: клиент / взаимные / очищ. / доля" for c in COMPANIES) + " | Σ клиент | взаимные | гудвилл | КФО банка | расхожд. |") for line in lines: r = rows[line] cells = [] for c in COMPANIES: x = (r["companies"] or {}).get(c) cells.append("—" if x is None else f"{x['client']} / {x['vgo']} / {x['cleaned']} / {x['share']}") print(f"| {line} | " + " | ".join(cells) + f" | {r['sum']} | {r['vgo']} | {r['goodwill_adjust']} | {r['bank']} | {r['dev']} |") def self_test(): """Проверки правил, которые сквозной пример не задевает (ревью ветки I-2, I-3, I-4, I-6).""" # I-2: auto и manual — из одной округлённой ячейки: manual округляется, auto = vgo − manual assert split_vgo(D("500.50"), D("500.50"), 1) == (D("1"), D("0"), D("1")) assert split_vgo(D("0"), D("700.00"), -1) == (D("-1"), D("0"), D("-1")) # I-3: автосумма min_or_full при отрицательной стороне 2 — как при пустой; при amount_1 ≤ 0 — 0 assert auto_amount("min_or_full", D("100"), D("-5")) == D("100") assert auto_amount("min_or_full", D("-1"), D("50")) == D("0") # I-3: расхождение агрегата min — и у одностороннего (|a1 − a2|), у min_or_full — нет assert agg_deviation("min", D("1000"), D("0")) == D("1000") assert agg_deviation("min", D("100"), D("-50")) == D("150") assert agg_deviation("min_or_full", D("1000"), D("0")) == D("0") # I-4: доля по модулю ≥ 1000 % — null assert share(D("280000"), D("1")) is None assert share(D("400000"), D("568000")) == D("70.42") # I-6: итога нет в форме (упрощённая) — client итога = Σ client слагаемых assert total_client({"1170": D("10"), "1150": D("5")}, "1100", ["1170", "1150"]) == D("15") assert total_client({"1100": D("20"), "1170": D("10")}, "1100", ["1170"]) == D("20") if __name__ == "__main__": self_test() res = {} for enabled in (True, False): bb, ofr, summary, g = compute(enabled) tag = "гудвилл включён" if enabled else "гудвилл выключен" show(f"ББ, {tag}", bb, BB_LINES) show(f"ОФР, {tag}", ofr, OFR_LINES) print("\nСводка:", json.dumps(summary, default=str, ensure_ascii=False)) res["enabled" if enabled else "disabled"] = {"balance": bb, "income": ofr, "summary": summary} if "--json" in sys.argv: with open(sys.argv[sys.argv.index("--json") + 1], "w") as f: json.dump(res, f, default=str, ensure_ascii=False, indent=1) print("\nOK: все проверки прошли")