417 lines
20 KiB
Python
417 lines
20 KiB
Python
"""Экономика компаний группы с внешними контрагентами (спека §5.2–5.3).
|
||
|
||
Источник образца потоков и закрытия месяца: synthetic-group/gen/economy.py (#35); контрагенты здесь — из плана
|
||
`counterparties.plan`, а не фиксированные SUP_A/BUY_G.
|
||
|
||
Порядок построения (на него опирается Task 4, операции внутри группы):
|
||
open_books(...) входящие остатки на book_start (31.12 перед первым периодом);
|
||
для каждого месяца: post_month(...) внешние операции месяца,
|
||
hook(...) операции внутри группы (extra),
|
||
close_month(...) закрытие 26/44 → 90, 90/91 → 99, налог, в декабре 99 → 84.
|
||
`build_ledgers` — эта последовательность целиком.
|
||
|
||
Аналитика контрагента: `An(cp=<ИНН Org>)` для внешних, `An(cp=<код члена>)` для компаний группы (последние — Task 4).
|
||
Модель компании (веса, классы контрагентов, профиль) строится один раз в `open_books` и лежит в `Ledger.meta["model"]`.
|
||
"""
|
||
|
||
from collections.abc import Callable
|
||
from dataclasses import dataclass, field
|
||
from datetime import date, timedelta
|
||
from decimal import ROUND_HALF_UP, Decimal
|
||
import math
|
||
|
||
from synthgen.config import Config
|
||
from synthgen.counterparties import CounterpartyPlan
|
||
from synthgen.group import Group, Member
|
||
from synthgen.ledger import CENT, An, Ledger, opening
|
||
from synthgen.periods import Period, book_end, book_start
|
||
from synthgen.rng import rng_for
|
||
|
||
D = Decimal
|
||
VAT = D("0.2")
|
||
PROFIT_TAX = D("0.2")
|
||
SOCIAL = D("0.3")
|
||
GROWTH = D("0.07") # рост выручки в год
|
||
FNS = An("FNS")
|
||
PARETO_ALPHA = 1.2
|
||
|
||
# kind словаря → (доля закупок товара к выручке, доля расходов на продажу, доля зарплаты, ОС к годовой выручке)
|
||
KIND_PROFILE: dict[str, tuple[str, str, str, str]] = {
|
||
"агро": ("0.55", "0.05", "0.08", "0.5"),
|
||
"торговля": ("0.65", "0.07", "0.06", "0.3"),
|
||
"производство": ("0.50", "0.06", "0.10", "0.5"),
|
||
"строительство": ("0.50", "0.05", "0.12", "0.5"),
|
||
"логистика": ("0.35", "0.08", "0.12", "0.5"),
|
||
"услуги": ("0.15", "0.06", "0.20", "0.3"),
|
||
}
|
||
HEAD_PROFILE = ("0.10", "0.05", "0.25", "1.5") # головная — холдинг: малая выручка, большие активы
|
||
MEDIAN_REVENUE = 1_000_000_000
|
||
HEAD_REVENUE = 50_000_000
|
||
FLOW_ACCOUNTS = ("62", "60", "76") # с потоками по контрагентам (продажи, закупки, прочие поставщики)
|
||
|
||
|
||
def q(x: Decimal) -> Decimal:
|
||
return x.quantize(CENT, rounding=ROUND_HALF_UP)
|
||
|
||
|
||
def vat_of(gross: Decimal) -> Decimal:
|
||
return q(gross * 20 / 120)
|
||
|
||
|
||
def last_day(year: int, month: int) -> date:
|
||
return date(year + (month == 12), month % 12 + 1, 1) - timedelta(days=1)
|
||
|
||
|
||
def _prev(year: int, month: int) -> tuple[int, int]:
|
||
return (year - 1, 12) if month == 1 else (year, month - 1)
|
||
|
||
|
||
@dataclass
|
||
class FileModel:
|
||
"""Контрагенты одного файла (компания × счёт): аналитика, веса Парето, класс поведения."""
|
||
|
||
ans: list[An | None]
|
||
w: list[Decimal] # по убыванию
|
||
cls: list[str] # normal (оплата в следующем месяце) / turn (оплата в месяце) / bal (только сальдо) / rev (аванс .02)
|
||
w_all: Decimal = D(0)
|
||
w_active: Decimal = D(0)
|
||
|
||
def __post_init__(self) -> None:
|
||
self.w_all = sum(self.w, D(0))
|
||
self.w_active = sum((w for w, c in zip(self.w, self.cls) if c in ("normal", "turn")), D(0))
|
||
|
||
|
||
@dataclass
|
||
class CompanyModel:
|
||
code: str
|
||
seed: int
|
||
first_year: int
|
||
rev_year: Decimal
|
||
cogs: Decimal
|
||
sell: Decimal
|
||
salary: Decimal
|
||
ppe_ratio: Decimal
|
||
files: dict[str, FileModel] = field(default_factory=dict)
|
||
banks: dict[str, FileModel] = field(default_factory=dict) # "66"/"67" → банки и веса долей долга
|
||
|
||
|
||
def _weights(rng, n: int) -> list[Decimal]:
|
||
"""Веса контрагентов по кривой Лоренца Парето: доля верхних p контрагентов = p^(1 − 1/α); α = 1.2 даёт
|
||
~68 % оборота у верхних 10 %. Детерминированная кривая вместо выборки Парето (иначе доля верхних 10 % прыгает
|
||
от 23 % до 97 % по зерну), небольшой шум — чтобы веса разных файлов не совпадали."""
|
||
expo = 1 - 1 / PARETO_ALPHA
|
||
shares = [((k / n) ** expo - ((k - 1) / n) ** expo) * rng.uniform(0.85, 1.15) for k in range(1, n + 1)]
|
||
return [D(max(1, round(w * 1e9))) for w in sorted(shares, reverse=True)]
|
||
|
||
|
||
def _classes(rng, acct: str, n: int) -> list[str]:
|
||
out = []
|
||
for i in range(n):
|
||
u = rng.random()
|
||
if i < 3: # крупнейшие контрагенты — обычные: оплата в следующем месяце
|
||
out.append("normal")
|
||
elif u < 0.03:
|
||
out.append("rev" if acct in ("60", "62") else "bal")
|
||
elif u < 0.08:
|
||
out.append("bal")
|
||
elif u < 0.28:
|
||
out.append("turn")
|
||
else:
|
||
out.append("normal")
|
||
if acct in ("60", "62") and n >= 8 and "rev" not in out:
|
||
out[n // 2] = "rev" # хотя бы один контрагент с обратным знаком сальдо (аванс/переплата, §5.3)
|
||
return out
|
||
|
||
|
||
def _build_file(cfg: Config, code: str, acct: str, orgs: list) -> FileModel:
|
||
n = len(orgs)
|
||
if n == 0: # нет контрагентов в плане (mean = 0): операции идут без аналитики контрагента
|
||
return FileModel([None], [D(1)], ["normal"])
|
||
ans = [An(cp=o.inn) for o in orgs]
|
||
return FileModel(ans, _weights(rng_for(cfg.seed, "cp-weights", code, acct), n), _classes(rng_for(cfg.seed, "cp-class", code, acct), acct, n))
|
||
|
||
|
||
def _build_model(cfg: Config, plan: CounterpartyPlan, member: Member, first_year: int) -> CompanyModel:
|
||
code = member.code
|
||
scale = rng_for(cfg.seed, "scale", code)
|
||
if member.is_head:
|
||
rev = HEAD_REVENUE * math.exp(scale.gauss(0, 0.5))
|
||
prof = HEAD_PROFILE
|
||
else:
|
||
rev = MEDIAN_REVENUE * math.exp(scale.gauss(0, math.log(10) / 2)) # ×10 в обе стороны на ±2σ
|
||
prof = KIND_PROFILE[member.org.kind]
|
||
rev = min(max(rev, 5e7), 2e10)
|
||
model = CompanyModel(code, cfg.seed, first_year, D(int(rev)), D(prof[0]), D(prof[1]), D(prof[2]), D(prof[3]))
|
||
for acct in ("58", *FLOW_ACCOUNTS):
|
||
model.files[acct] = _build_file(cfg, code, acct, plan.by_file.get((code, acct), []))
|
||
for acct in ("66", "67"):
|
||
orgs = plan.by_file.get((code, acct), [])
|
||
if orgs:
|
||
model.banks[acct] = _build_file(cfg, code, acct, orgs)
|
||
return model
|
||
|
||
|
||
# ---------- суммы месяца ----------
|
||
|
||
def _noise(m: CompanyModel, tag: str, year: int, month: int) -> Decimal:
|
||
return D(rng_for(m.seed, "noise", m.code, tag, year, month).randint(92, 108)) / 100
|
||
|
||
|
||
def _rev_month(m: CompanyModel, year: int, month: int) -> Decimal:
|
||
"""Чистая выручка месяца (без НДС)."""
|
||
growth = 1 + GROWTH * (year - m.first_year)
|
||
return q(m.rev_year / 12 * growth * _noise(m, "rev", year, month))
|
||
|
||
|
||
def _total(m: CompanyModel, acct: str, year: int, month: int) -> Decimal:
|
||
"""Брутто-оборот месяца по счёту расчётов: 62 — продажи, 60 — закупки товара, 76 — прочие поставщики."""
|
||
rev = _rev_month(m, year, month)
|
||
share = {"62": D(1), "60": m.cogs, "76": m.sell}[acct]
|
||
return q(rev * share * (1 + VAT))
|
||
|
||
|
||
def _split(fm: FileModel, total: Decimal, classes: tuple[str, ...]) -> list[tuple[int, Decimal]]:
|
||
"""Раскладка оборота по контрагентам указанных классов с весами Парето; нулевые суммы отбрасываются."""
|
||
w_sum = sum((w for w, c in zip(fm.w, fm.cls) if c in classes), D(0))
|
||
if w_sum == 0 or total <= 0:
|
||
return []
|
||
out = []
|
||
for i, (w, c) in enumerate(zip(fm.w, fm.cls)):
|
||
if c in classes:
|
||
a = q(total * w / w_sum)
|
||
if a > 0:
|
||
out.append((i, a))
|
||
return out
|
||
|
||
|
||
def _side(fm: FileModel, ref: Decimal, i: int, factor: str) -> Decimal:
|
||
"""Сальдо «без оборотов» контрагента: доля месячного оборота по весу среди всех контрагентов файла."""
|
||
return q(ref * fm.w[i] / fm.w_all * D(factor))
|
||
|
||
|
||
# ---------- входящие остатки ----------
|
||
|
||
def opening_lines(m: CompanyModel) -> list[tuple[str, An | None, Decimal]]:
|
||
py, pm = m.first_year - 1, 12
|
||
ppe = q(m.rev_year * m.ppe_ratio)
|
||
lines: list[tuple[str, An | None, Decimal]] = [
|
||
("01.01", None, ppe),
|
||
("02", None, -q(ppe * D("0.3"))),
|
||
("51", None, q(m.rev_year * D("0.06"))),
|
||
]
|
||
if m.cogs:
|
||
lines.append(("41", None, q(m.rev_year * m.cogs * D("0.15"))))
|
||
# 62: дебиторка — оплата прошлого месяца; bal — старая задолженность; rev — аванс (62.02, кредит)
|
||
for acct, sign, sub, sub_rev in (("62", 1, "62.01", "62.02"), ("60", -1, "60.01", "60.02"), ("76", -1, "76.05", None)):
|
||
fm = m.files[acct]
|
||
total = _total(m, acct, py, pm)
|
||
if acct != "62" and total == 0:
|
||
continue
|
||
ref = total if total else q(_rev_month(m, py, pm))
|
||
for i, a in _split(fm, total, ("normal", "turn")):
|
||
if fm.cls[i] == "normal": # оплата в следующем месяце; turn закрыт в своём месяце
|
||
lines.append((sub, fm.ans[i], sign * a))
|
||
for i, c in enumerate(fm.cls):
|
||
if c == "bal":
|
||
lines.append((sub, fm.ans[i], sign * _side(fm, ref, i, "0.5")))
|
||
elif c == "rev" and sub_rev:
|
||
lines.append((sub_rev, fm.ans[i], -sign * _side(fm, ref, i, "0.3")))
|
||
fm58 = m.files["58"]
|
||
ref58 = q(_rev_month(m, py, pm))
|
||
for i, c in enumerate(fm58.cls):
|
||
if c in ("normal", "bal") and fm58.ans[i] is not None:
|
||
lines.append(("58.03", fm58.ans[i], _side(fm58, ref58, i, "3")))
|
||
for acct, share in (("66", "0.12"), ("67", "0.08")):
|
||
fm = m.banks.get(acct)
|
||
if fm:
|
||
for i, a in enumerate(_split_all(fm, m.rev_year * D(share))):
|
||
if a > 0:
|
||
lines.append((f"{acct}.01", fm.ans[i], -a))
|
||
lines.append(("80", None, -q(m.rev_year * D("0.02"))))
|
||
return lines
|
||
|
||
|
||
def _split_all(fm: FileModel, total: Decimal) -> list[Decimal]:
|
||
return [q(total * w / fm.w_all) for w in fm.w]
|
||
|
||
|
||
def open_books(cfg: Config, group: Group, ps: list[Period], plan: CounterpartyPlan) -> dict[str, Ledger]:
|
||
"""Книги членов группы с входящими остатками на `book_start(ps)`; 84 — балансирующая строка."""
|
||
day = book_start(ps)
|
||
ledgers: dict[str, Ledger] = {}
|
||
for member in group.members:
|
||
model = _build_model(cfg, plan, member, ps[0].year)
|
||
lg = Ledger(member.code)
|
||
lg.meta["model"] = model
|
||
lines = opening_lines(model)
|
||
total = sum((v for _, _, v in lines), D(0))
|
||
if total <= 0:
|
||
raise ValueError(f"{member.code}: активы не больше обязательств, 84 стал бы дебетовым")
|
||
for acct, an, v in lines:
|
||
opening(lg, day, acct, an, v)
|
||
opening(lg, day, "84", None, -total)
|
||
ledgers[member.code] = lg
|
||
return ledgers
|
||
|
||
|
||
# ---------- операции месяца ----------
|
||
|
||
def _flow(lg: Ledger, m: CompanyModel, acct: str, year: int, month: int) -> None:
|
||
"""Продажи (62), закупки товара (60) или прочие расходы (76) месяца и оплата контрагентам."""
|
||
fm = m.files[acct]
|
||
py, pm = _prev(year, month)
|
||
d = lambda n: date(year, month, n) # noqa: E731
|
||
cur = _split(fm, _total(m, acct, year, month), ("normal", "turn"))
|
||
prev = {i: a for i, a in _split(fm, _total(m, acct, py, pm), ("normal", "turn")) if fm.cls[i] == "normal"}
|
||
vat_total = D(0)
|
||
net_total = D(0)
|
||
sub = {"62": "62.01", "60": "60.01", "76": "76.05"}[acct]
|
||
for i, gross in cur:
|
||
an = fm.ans[i]
|
||
vat = vat_of(gross)
|
||
net = gross - vat
|
||
vat_total += vat
|
||
net_total += net
|
||
if acct == "62":
|
||
lg.post(d(10), sub, an, "90.01", None, gross, "реализация")
|
||
else:
|
||
exp = "41" if acct == "60" else "44"
|
||
lg.post(d(5), exp, None, sub, an, net, "закупка")
|
||
if vat > 0:
|
||
lg.post(d(5), "19", None, sub, an, vat, "НДС по закупке")
|
||
if vat_total > 0:
|
||
if acct == "62":
|
||
lg.post(d(10), "90.03", None, "68", FNS, vat_total, "НДС с реализации")
|
||
else:
|
||
lg.post(d(5), "68", FNS, "19", None, vat_total, "НДС к вычету")
|
||
if acct == "60" and net_total > 0:
|
||
lg.post(d(5), "90.02", None, "41", None, net_total, "себестоимость")
|
||
pay_day = d(20) if acct == "62" else d(15)
|
||
for i, gross in cur:
|
||
if fm.cls[i] == "turn":
|
||
_settle(lg, acct, sub, fm.ans[i], gross, pay_day)
|
||
for i, amount in prev.items():
|
||
_settle(lg, acct, sub, fm.ans[i], amount, pay_day)
|
||
|
||
|
||
def _settle(lg: Ledger, acct: str, sub: str, an: An | None, amount: Decimal, day: date) -> None:
|
||
if acct == "62":
|
||
lg.post(day, "51", None, sub, an, amount, "поступление от покупателя")
|
||
else:
|
||
lg.post(day, sub, an, "51", None, amount, "оплата поставщику")
|
||
|
||
|
||
def _loans(lg: Ledger, m: CompanyModel, acct: str, year: int, month: int) -> None:
|
||
"""Проценты банку по кредитам: начислены и уплачены в конце месяца (остаток по .02 не копится)."""
|
||
fm = m.banks.get(acct)
|
||
if not fm:
|
||
return
|
||
share = D("0.12") if acct == "66" else D("0.08")
|
||
day = date(year, month, 28)
|
||
for i, principal in enumerate(_split_all(fm, m.rev_year * share)):
|
||
interest = q(principal * D("0.12") / 12)
|
||
if interest > 0:
|
||
an = fm.ans[i]
|
||
lg.post(day, "91.02", An(an.cp, item="Проценты"), f"{acct}.02", an, interest, "проценты по кредиту")
|
||
lg.post(day, f"{acct}.02", an, "51", None, interest, "уплата процентов")
|
||
|
||
|
||
def _issued_loans(lg: Ledger, m: CompanyModel, year: int, month: int) -> None:
|
||
"""58.03: часть контрагентов — займ выдан и возвращён в месяце (только обороты); остальные — сальдо с открытия."""
|
||
fm = m.files["58"]
|
||
amounts = _split(fm, q(_rev_month(m, year, month) * D("0.05")), ("turn",))
|
||
for i, a in amounts:
|
||
lg.post(date(year, month, 12), "58.03", fm.ans[i], "51", None, a, "выдан займ")
|
||
lg.post(date(year, month, 22), "51", None, "58.03", fm.ans[i], a, "возврат займа")
|
||
|
||
|
||
def post_month(ledgers: dict[str, Ledger], cfg: Config, group: Group, plan: CounterpartyPlan, year: int, month: int) -> None:
|
||
"""Внешние операции месяца всех членов группы (до операций внутри группы и закрытия)."""
|
||
for code in sorted(ledgers):
|
||
lg = ledgers[code]
|
||
m: CompanyModel = lg.meta["model"]
|
||
d = lambda n: date(year, month, n) # noqa: E731
|
||
for acct in ("60", "76", "62"):
|
||
if _total(m, acct, year, month) > 0:
|
||
_flow(lg, m, acct, year, month)
|
||
_issued_loans(lg, m, year, month)
|
||
sal = q(_rev_month(m, year, month) * m.salary)
|
||
if sal > 0:
|
||
soc = q(sal * SOCIAL)
|
||
lg.post(d(25), "26", None, "70", None, sal, "начислена зарплата")
|
||
lg.post(d(25), "26", None, "69", None, soc, "страховые взносы")
|
||
lg.post(d(26), "70", None, "51", None, sal, "выплата зарплаты")
|
||
lg.post(d(26), "69", None, "51", None, soc, "уплата взносов")
|
||
lg.post(d(28), "26", None, "02", None, q(m.rev_year * m.ppe_ratio / 120), "амортизация")
|
||
_loans(lg, m, "66", year, month)
|
||
_loans(lg, m, "67", year, month)
|
||
|
||
|
||
def _close_all(lg: Ledger, day: date, prefix: str) -> Decimal:
|
||
"""Закрывает все субсчета prefix на 99; возвращает итог (Кт > 0 — прибыль)."""
|
||
result = D(0)
|
||
for (acct, an), bal in sorted(lg.balance(day, prefix).items(), key=lambda kv: (kv[0][0], str(kv[0][1]))):
|
||
if bal > 0:
|
||
lg.post(day, "99", None, acct, an, bal, "закрытие на 99")
|
||
else:
|
||
lg.post(day, acct, an, "99", None, -bal, "закрытие на 99")
|
||
result -= bal
|
||
return result
|
||
|
||
|
||
def close_month(ledgers: dict[str, Ledger], year: int, month: int) -> None:
|
||
day = last_day(year, month)
|
||
for code in sorted(ledgers):
|
||
lg = ledgers[code]
|
||
vat_due = -lg.balance(day, "68").get(("68", FNS), D(0))
|
||
if vat_due > 0:
|
||
lg.post(day, "68", FNS, "51", None, vat_due, "уплата НДС")
|
||
for cost, target in (("26", "90.08"), ("44", "90.07")):
|
||
for (acct, an), bal in lg.balance(day, cost).items():
|
||
if bal > 0:
|
||
lg.post(day, target, None, acct, an, bal, "закрытие затрат")
|
||
result = _close_all(lg, day, "90") + _close_all(lg, day, "91")
|
||
tax = q(result * PROFIT_TAX)
|
||
if tax > 0:
|
||
lg.post(day, "99", None, "68.04", FNS, tax, "налог на прибыль")
|
||
elif tax < 0:
|
||
lg.post(day, "68.04", FNS, "99", None, -tax, "корректировка налога на прибыль")
|
||
if month % 3 == 0:
|
||
due = -lg.balance(day, "68.04").get(("68.04", FNS), D(0))
|
||
if due > 0:
|
||
lg.post(day, "68.04", FNS, "51", None, due, "уплата налога на прибыль")
|
||
if month == 12:
|
||
profit = -lg.balance(day, "99").get(("99", None), D(0))
|
||
if profit > 0:
|
||
lg.post(day, "99", None, "84", None, profit, "реформация баланса")
|
||
elif profit < 0:
|
||
lg.post(day, "84", None, "99", None, -profit, "реформация баланса")
|
||
|
||
|
||
def months(ps: list[Period]) -> list[tuple[int, int]]:
|
||
"""Месяцы книги: с января года первого периода до конца последнего периода."""
|
||
end = book_end(ps)
|
||
out = []
|
||
y, mth = ps[0].year, 1
|
||
while (y, mth) <= (end.year, end.month):
|
||
out.append((y, mth))
|
||
y, mth = (y + 1, 1) if mth == 12 else (y, mth + 1)
|
||
return out
|
||
|
||
|
||
def build_ledgers(
|
||
cfg: Config,
|
||
group: Group,
|
||
ps: list[Period],
|
||
plan: CounterpartyPlan,
|
||
extra: Callable[[dict[str, Ledger], int, int], None] | None = None,
|
||
) -> dict[str, Ledger]:
|
||
"""Книги всех членов: входящие остатки + помесячно (внешние операции, `extra`, закрытие)."""
|
||
ledgers = open_books(cfg, group, ps, plan)
|
||
for year, month in months(ps):
|
||
post_month(ledgers, cfg, group, plan, year, month)
|
||
if extra is not None:
|
||
extra(ledgers, year, month)
|
||
close_month(ledgers, year, month)
|
||
return ledgers
|