173 lines
6.8 KiB
Python
173 lines
6.8 KiB
Python
import math
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
|
||
import pytest
|
||
|
||
from synthgen import dictionary
|
||
from synthgen.config import Config
|
||
from synthgen.counterparties import plan as make_plan
|
||
from synthgen.economy import last_day, months
|
||
from synthgen.group import build_group
|
||
from synthgen.intragroup import build, pair_balance, plan_pairs
|
||
from synthgen.periods import book_start, build_periods
|
||
|
||
D = Decimal
|
||
|
||
|
||
def _make(**kw):
|
||
mean = kw.pop("mean", 10)
|
||
cfg = Config(group_size=kw.pop("group_size", 5), periods=kw.pop("periods", 6), osv_counterparties_mean=mean, **kw)
|
||
g = build_group(cfg, dictionary.load())
|
||
ps = build_periods(cfg.start, cfg.periods)
|
||
books, pairs = build(cfg, g, ps, make_plan(cfg, g))
|
||
return cfg, g, ps, books, pairs
|
||
|
||
|
||
@pytest.fixture(scope="module")
|
||
def built():
|
||
return _make(vgo_mismatch_share=D("0.3"))
|
||
|
||
|
||
def test_plan_pairs_deterministic_and_loans():
|
||
cfg = Config(group_size=6)
|
||
g = build_group(cfg, dictionary.load())
|
||
p1, p2 = plan_pairs(cfg, g), plan_pairs(cfg, g)
|
||
assert p1 == p2
|
||
loans = [p for p in p1 if p.kind == "loan"]
|
||
assert [p.b for p in loans] == [m.code for m in g.members[1:]]
|
||
assert all(p.mismatch is None for p in p1) and all(p.a == "C01" for p in loans)
|
||
|
||
|
||
def test_pairs_mirror(built):
|
||
cfg, g, ps, books, pairs = built
|
||
clean = [p for p in pairs if not p.mismatches]
|
||
assert clean
|
||
for pair in clean:
|
||
for per in ps:
|
||
ba, bb = pair_balance(books, pair, per.end)
|
||
assert ba == -bb, (pair, per.key)
|
||
assert any(pair_balance(books, p, ps[0].end)[0] != 0 for p in clean if p.kind == "trade")
|
||
assert any(pair_balance(books, p, ps[0].end)[0] != 0 for p in clean if p.kind == "loan")
|
||
|
||
|
||
def _years(ps):
|
||
return sorted({p.year for p in ps})
|
||
|
||
|
||
def test_mismatch_share():
|
||
cfg, g, ps, books, pairs = _make(group_size=10, vgo_mismatch_share=D("0.1"), mean=5)
|
||
n = len(pairs)
|
||
for y in _years(ps):
|
||
got = sum(1 for p in pairs for m in p.mismatches if m.year == y)
|
||
assert got == max(1, math.floor(D("0.1") * n)), y
|
||
cfg0, g0, ps0, books0, pairs0 = _make(group_size=10, vgo_mismatch_share=D("0"), mean=5, periods=2)
|
||
assert all(not p.mismatches and p.mismatch is None for p in pairs0)
|
||
|
||
|
||
def test_mismatch_mix(built):
|
||
cfg, g, ps, books, pairs = built
|
||
for y in _years(ps):
|
||
mms = [m for p in pairs for m in p.mismatches if m.year == y]
|
||
assert len(mms) >= 2
|
||
small = [m for m in mms if D(1) <= m.pct <= D(4)]
|
||
large = [m for m in mms if D(8) <= m.pct <= D(25)]
|
||
assert len(small) == len(mms) // 2
|
||
assert len(large) == len(mms) - len(mms) // 2
|
||
for m in mms:
|
||
assert m.above_threshold == (m.pct > 5)
|
||
|
||
|
||
def test_mismatch_lifecycle(built):
|
||
cfg, g, ps, books, pairs = built
|
||
checked_removal = 0
|
||
for pair in pairs:
|
||
for m in pair.mismatches:
|
||
for per in ps:
|
||
if per.year != m.year or per.key < m.period_key:
|
||
continue
|
||
ba, bb = pair_balance(books, pair, per.end)
|
||
diff = (ba + bb) # зеркальные стороны: сумма = разница по модулю со знаком стороны
|
||
assert abs(diff) == m.amount, (pair, per.key)
|
||
# у стороны с проводкой модуль сальдо больше
|
||
with_entry = ba if m.side == pair.a else bb
|
||
other = bb if m.side == pair.a else ba
|
||
assert abs(with_entry) - abs(other) == m.amount
|
||
# pct — от меньшей стороны на конец периода расхождения (в следующих периодах сальдо торговли иное)
|
||
if per.key == m.period_key:
|
||
assert abs(m.pct - (m.amount / abs(other) * 100)) < D("0.01")
|
||
jan = date(m.year + 1, 1, 31)
|
||
if jan <= ps[-1].end:
|
||
ba, bb = pair_balance(books, pair, jan)
|
||
assert ba == -bb
|
||
checked_removal += 1
|
||
assert checked_removal > 0
|
||
|
||
|
||
def test_dividends_no_balance(built):
|
||
cfg, g, ps, books, pairs = built
|
||
codes = {m.code for m in g.members}
|
||
for lg in books.values():
|
||
for per in ps:
|
||
assert lg.balance(per.end, "76.03") == {}
|
||
head = books["C01"]
|
||
t = head.turnover(date(2025, 4, 1), date(2025, 6, 30), "76.03")
|
||
assert t and all(dt > 0 and ct > 0 for dt, ct in t.values())
|
||
for code in codes - {"C01"}:
|
||
t = books[code].turnover(date(2025, 4, 1), date(2025, 6, 30), "76.03")
|
||
assert t and any(ct > 0 for dt, ct in t.values())
|
||
|
||
|
||
def test_shares(built):
|
||
cfg, g, ps, books, pairs = built
|
||
start = book_start(ps)
|
||
head = books["C01"]
|
||
h58 = {an.cp: v for (a, an), v in head.balance(start, "58.01").items()}
|
||
assert set(h58) == {m.code for m in g.members[1:]}
|
||
for m in g.members[1:]:
|
||
c80 = {an.cp: v for (a, an), v in books[m.code].balance(start, "80.09").items()}
|
||
assert c80[head.company] == -h58[m.code]
|
||
if m.minority is None:
|
||
assert set(c80) == {"C01"} and m.head_share == D("1.00")
|
||
else:
|
||
assert set(c80) == {"C01", m.minority.inn}
|
||
assert c80[m.minority.inn] < 0
|
||
# миноритарий не попал к головной
|
||
assert not any(an.cp in {m.minority.inn for m in g.members if m.minority} for (_, an) in head.balance(start, "58") if an)
|
||
assert any(m.minority for m in g.members)
|
||
|
||
|
||
def test_books_still_balanced(built):
|
||
cfg, g, ps, books, pairs = built
|
||
start = book_start(ps)
|
||
for lg in books.values():
|
||
for day in (start, date(2025, 3, 31), date(2025, 12, 31), ps[-1].end):
|
||
assert sum(lg.balance(day).values()) == 0
|
||
assert lg.balance(day, "000") == {}
|
||
assert not [p for p in lg.postings if "000" in (p.dt, p.ct) and p.day > start]
|
||
for per in ps:
|
||
for acct in ("90", "91"):
|
||
assert lg.balance(per.end, acct) == {}
|
||
for y, mth in months(ps):
|
||
assert lg.balance(last_day(y, mth), "51").get(("51", None), D(0)) >= 0, (lg.company, y, mth)
|
||
|
||
|
||
def test_group_analytics_not_mixed(built):
|
||
cfg, g, ps, books, pairs = built
|
||
codes = {m.code for m in g.members}
|
||
ext = {o.inn for o in g.externals}
|
||
for lg in books.values():
|
||
for (acct, an) in lg.balance(ps[-1].end):
|
||
if an is not None and an.cp in codes:
|
||
assert acct.split(".")[0] in {"58", "60", "62", "66", "67", "76", "80", "91"}
|
||
if acct.startswith(("60", "62")) and an is not None and an.cp in ext:
|
||
assert an.cp not in codes
|
||
|
||
|
||
def test_edge_group_two():
|
||
cfg, g, ps, books, pairs = _make(group_size=2, periods=4, vgo_mismatch_share=D("0.5"))
|
||
assert [p.kind for p in pairs if p.kind == "loan"] == ["loan"]
|
||
for lg in books.values():
|
||
assert sum(lg.balance(ps[-1].end).values()) == 0
|
||
assert any(p.mismatches for p in pairs)
|