316 lines
15 KiB
Python
316 lines
15 KiB
Python
"""Сквозной пример §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: все проверки прошли")
|