dodil/erp-brick-tax
DODIL ERP brick: tax
src/brick_tax/application/schemes.py
"""What a VAT return holds under the business's SCHEME (v4: TAX-36 cash accounting, TAX-37 flat rate, TAX-48 partial exemption,
TAX-49 capital goods, TAX-44/45 import VAT): `plan` turns the frozen document tax, the neighbours' payments and the standing rows
into the CANDIDATE entries of one period. It writes nothing: the preparation claims them (check, then a plain INSERT).
invoice basis a sale by its tax point within the period; a purchase by the period it is held in (as TAX-28)
cash basis the VAT follows the money: payments' receipt allocations and purchasing's confirmed supplier payments, each taking
the document's tax by the share paid — cumulative, so the penny rounding telescopes and the periods sum exactly
flat rate box 1 = the flat rate x the turnover including VAT; box 4 holds only capital goods of 2,000 or more
partial exemp. input VAT by its code's attribution: taxable 100%, exempt only while de minimis holds, residual at last year's %
Every figure the law sets is a dated tax_statutory row read on the date it applies to.
"""
import json
from datetime import date, timedelta
from decimal import ROUND_CEILING, ROUND_HALF_UP, Decimal
from brick_tax.application.inputs import day
from brick_tax.domain import registration
from brick_tax.domain.errors import Gate, GateUnmet, unmet
from brick_tax.domain.vat_return import add_months
from brick_tax.infrastructure import seams
P2 = Decimal("0.01")
Z = Decimal(0)
HUNDRED = Decimal(100)
FIELDS = ("net_amount", "vat_amount", "vat_reclaimable", "box1_amount", "box2_amount", "box4_amount")
PURCHASE_SOURCES = ("document", "cash_basis")
def r2(v) -> Decimal:
return Decimal(str(v)).quantize(P2, ROUND_HALF_UP)
def gate(name: str, met: bool, reason: str, fix: str = "") -> dict:
return {"gate": name, "met": met, "reason": reason, "fix": fix}
def months_of(start: date, end: date) -> int:
return max(1, (end.year - start.year) * 12 + end.month - start.month + 1)
def stat(repo, key: str, on: date) -> Decimal:
from brick_tax.application import returns as R
return R._stat(repo, key, on)
def held_elsewhere(repo, row: dict, vrn: str, period_key: str) -> bool:
"""True when ANOTHER return already holds this entry's key (the claim would be refused: its tax is in that return)."""
have = repo.entries.get(document_ref=row["document_ref"], tax_code=row["tax_code"], entry_kind=row["entry_kind"])
return have is not None and (have["vrn"], have["period_key"]) != (vrn, period_key)
def year_end_for(mmdd: str, on: date) -> date:
"""The partial-exemption year end (MM-DD) on or after a date."""
m, d = (int(x) for x in mmdd.split("-"))
cand = date(on.year, m, d)
return cand if cand >= on else date(on.year + 1, m, d)
def _factor(doc) -> Decimal:
return Decimal(1) if doc.currency == "GBP" else doc.fx_rate
def excluded_from_cash(repo, doc, on: date) -> bool:
"""A document stays on the invoice basis when it is payable more than six months after its tax point (a figure the law sets)."""
return bool(doc.due_date and doc.due_date > add_months(doc.tax_point, int(stat(repo, "cash_exclude_months", on))))
def _turnover(docs, direction: str, frm: date, to: date, field: str) -> Decimal:
total = Z
for d in docs:
if d.direction == direction and d.tax_point is not None and frm <= d.tax_point <= to and d.computed() is not None:
total += getattr(d.computed(), field) * _factor(d)
return total
# ═══ the plan of one period ═════════════════════════════════════════════════════════════════════
def plan(repo, vrn: str, ob: dict) -> dict:
from brick_tax.application import returns as R
start, end, pk = ob["period_start"], ob["period_end"], ob["period_key"]
reg = registration.status_on(repo.registrations.rows(), start)
basis = "cash" if reg and reg.accounting_basis == "cash" else "invoice"
flat = bool(reg and reg.scheme == "flat_rate")
out = {"rows": [], "gates": [], "extras": {}, "basis": basis, "flat_rate_pct": None}
all_docs, docs = {}, []
for ref, (doc, _) in sorted(repo.documents.get_many(repo.documents.refs()).items()):
if doc.status != "frozen" or not doc.registered or doc.tax_point is None:
continue
all_docs[ref] = doc
if doc.tax_point <= end:
docs.append(doc)
cap = _capital_vat(docs, repo, end) if flat else {}
for doc in docs:
if doc.direction == "sale":
if doc.tax_point < start or flat:
continue
if basis == "cash" and not excluded_from_cash(repo, doc, end):
continue
elif basis == "cash":
continue
rows = R._doc_entries(doc, vrn, pk)
if flat and doc.direction == "purchase":
for e in rows:
if e["box1_amount"] == 0 and e["box2_amount"] == 0:
e["box4_amount"] = r2(cap.get((doc.document_ref, e["tax_code"]), Z))
out["rows"] += rows
if flat:
_flat_rate(repo, vrn, ob, reg, docs, out)
before = registration.status_on(repo.registrations.rows(), start - timedelta(days=1))
if basis == "cash":
_cash(repo, vrn, ob, all_docs, docs, out)
elif before is not None and before.accounting_basis == "cash":
_cash(repo, vrn, ob, all_docs, docs, out, voluntary_exit=True) # the first period back on the invoice basis
_imports(repo, vrn, ob, out)
_pva_adjustments(repo, vrn, ob, out)
if reg is not None and reg.partial_exemption_year_end and not flat:
_partial_exemption(repo, vrn, ob, reg, out)
_annual_rows(repo, vrn, ob, out)
for row in out["rows"]:
row.setdefault("vrn", vrn)
row.setdefault("period_key", pk)
for f in FIELDS:
row.setdefault(f, None if f == "vat_reclaimable" else Z)
row.setdefault("gross_amount", None)
row.setdefault("attribution", None)
row.setdefault("detail", None)
return out
# ═══ cash accounting (TAX-36) ═══════════════════════════════════════════════════════════════════
def _cash(repo, vrn, ob, all_docs, docs, out, voluntary_exit: bool = False) -> None:
"""The cash basis of a period. `voluntary_exit`: the business moved back to the invoice basis from this period, so the VAT of
what was supplied under the cash basis and is still outstanding comes in now, like the turnover-limit exit."""
from brick_tax.application import returns as R
start, end, pk = ob["period_start"], ob["period_end"], ob["period_key"]
reported = {r["document_ref"] for r in out["rows"] if r.get("entry_kind") == "document"}
def on_invoice_basis(ref: str) -> bool:
"""Already taxed on the invoice basis (before joining the scheme, or payable more than six months out): never taxed again."""
return ref in reported or any(e["entry_kind"] == "document" for e in repo.entries.where(document_ref=ref))
def in_scope(doc) -> bool:
return not voluntary_exit or doc.tax_point < start
earliest = min([o["period_start"] for o in repo.obligations.where(vrn=vrn)] or [start])
allocs = seams.payments.cash_basis(earliest, end)
pays = seams.purchasing.supplier_payments(earliest, end)
if allocs is None or pays is None:
raise unmet("cash_basis_source", "the cash basis is read from payments' receipts and purchasing's supplier payments, and one of "
"them could not be asked", "check the connection to payments and purchasing, then prepare again")
events = []
for a in allocs:
if a.get("source_kind") == "receipt":
events.append((day(a["received_on"]), "sale", f"{a['payment_id']}:{a['seq']}", a["invoice_id"], Decimal(str(a["amount"]))))
for p in pays:
if p.get("status", "confirmed") == "confirmed":
for b in p.get("bills") or []:
events.append((day(p["paid_on"]), "purchase", p["payment_id"], b["bill_id"], Decimal(str(b["amount"]))))
events.sort(key=lambda e: (e[0], e[2], e[3]))
state: dict = {}
def prior(ref: str, code: str) -> dict:
if (ref, code) not in state:
s = dict.fromkeys(FIELDS + ("gross_amount",), Z)
for r in repo.entries.where(document_ref=ref, tax_code=code):
if r["entry_kind"].startswith("cash") and (r["vrn"], r["period_key"]) != (vrn, pk):
for f in s:
s[f] += Decimal(str(r.get(f) or 0))
state[(ref, code)] = s
return state[(ref, code)]
def take(doc, entries, total_after_of, kind, when, detail) -> None:
for e in entries:
g_c = e["_gross"]
if g_c == 0:
continue
p = prior(doc.document_ref, e["tax_code"])
after = total_after_of(e)
row = {**e, "entry_kind": kind, "tax_point": when, "source": "cash_basis", "detail": detail}
for f in FIELDS:
full = Decimal(str(e[f])) if e[f] is not None else None
if full is None:
row[f] = None
continue
row[f] = r2(full * after / g_c) - p[f]
p[f] += row[f]
row["gross_amount"] = after - p["gross_amount"]
p["gross_amount"] = after
if any(row[f] for f in FIELDS if row[f] is not None) or row["gross_amount"]:
out["rows"].append(row)
for when, side, key, ref, amount in events:
doc = all_docs.get(ref)
if doc is None or doc.direction != side or on_invoice_basis(ref) or not in_scope(doc):
continue
if any((r["vrn"], r["period_key"]) != (vrn, pk) for r in repo.entries.where(document_ref=ref, entry_kind=f"cash:{key}")):
continue # this receipt or payment is in another return already
entries = R._doc_entries(doc, vrn, pk)
big_g = sum((e["_gross"] for e in entries), Z)
if big_g == 0:
continue
before = sum((prior(ref, e["tax_code"])["gross_amount"] for e in entries), Z)
d_after = min(before + amount, big_g) if big_g > 0 else max(before + amount, big_g)
take(doc, entries, lambda e: r2(d_after * e["_gross"] / big_g), f"cash:{key}", when,
f"cash basis: {amount} {'received' if side == 'sale' else 'paid'} on {when}")
# the period a business LEAVES the scheme (turnover above the limit): every outstanding VAT comes in
limit = stat(repo, "cash_leave_limit", end)
year = _turnover(list(all_docs.values()), "sale", add_months(end, -12) + timedelta(days=1), end, "net")
leaving = voluntary_exit or year > limit
if not voluntary_exit:
out["gates"].append(gate("cash_accounting_limit", not leaving,
f"turnover {year:,.2f} in the 12 months to {end} is {'above' if leaving else 'within'} the cash accounting "
f"limit {limit:,.2f}", "leave the scheme: Tax settings > Scheme" if leaving else ""))
if leaving:
for doc in all_docs.values():
if doc.tax_point > end or excluded_from_cash(repo, doc, end) or on_invoice_basis(doc.document_ref) or not in_scope(doc):
continue
entries = R._doc_entries(doc, vrn, pk)
take(doc, entries, lambda e: e["_gross"], "cash_exit", end, f"leaving the cash accounting scheme: the VAT still outstanding at {end}")
out["extras"]["cash_basis"] = {"leaving": leaving, "turnover_12_months": year, "limit": limit}
# ═══ flat rate (TAX-37) ═════════════════════════════════════════════════════════════════════════
def _capital_vat(docs, repo, on: date) -> dict:
"""(document, code) -> the reclaimable VAT of goods lines costing at least the capital minimum including VAT."""
minimum = stat(repo, "flat_capital_min", on)
out: dict = {}
for d in docs:
c = d.computed()
if d.direction != "purchase" or c is None:
continue
for ln in d.lines:
res = c.lines.get(ln.line_no)
if ln.supply == "goods" and res is not None and res.gross * _factor(d) >= minimum:
k = (d.document_ref, ln.tax_code)
out[k] = out.get(k, Z) + res.vat * _factor(d) * (HUNDRED if ln.reclaimable_pct is None else ln.reclaimable_pct) / HUNDRED
return out
def _flat_rate(repo, vrn, ob, reg, docs, out) -> None:
start, end, pk = ob["period_start"], ob["period_end"], ob["period_key"]
turnover = r2(_turnover(docs, "sale", start, end, "gross"))
minimum = stat(repo, "flat_capital_min", end)
goods = Z
for d in docs:
c = d.computed()
if d.direction == "purchase" and c is not None and start <= d.tax_point <= end:
for ln in d.lines:
res = c.lines.get(ln.line_no)
if ln.supply == "goods" and res is not None and res.gross * _factor(d) < minimum:
goods += res.gross * _factor(d)
goods = r2(goods)
share, per_year = stat(repo, "flat_lct_pct", end), stat(repo, "flat_lct_min_annual", end)
days = (end - start).days + 1
lct = goods < turnover * share / HUNDRED or goods < per_year * days / Decimal(365)
if not lct and reg.flat_rate_pct is None:
raise unmet("flat_rate_pct", "this flat rate registration has no percentage for its sector",
"Tax settings > Scheme: record the flat rate percentage from the next VAT period")
rate = stat(repo, "flat_lct_rate", end) if lct else reg.flat_rate_pct
discount = Z
if reg.first_year_until is not None and end <= reg.first_year_until:
discount = stat(repo, "flat_first_year_discount", end)
rate -= discount
vat = r2(turnover * rate / HUNDRED)
out["gates"].append(gate("limited_cost_trader", lct,
f"goods {goods:,.2f} are {'below' if lct else 'not below'} {share}% of turnover {turnover:,.2f} ({r2(turnover * share / HUNDRED):,.2f})"
+ (f", so the limited cost trader rate {stat(repo, 'flat_lct_rate', end)}% replaces the sector's {reg.flat_rate_pct}%" if lct else ""),
""))
leave = stat(repo, "flat_leave_limit", end)
year = r2(_turnover(docs, "sale", add_months(end, -12) + timedelta(days=1), end, "gross"))
out["gates"].append(gate("flat_rate_eligible", year <= leave,
f"turnover including VAT {year:,.2f} in the 12 months to {end} is {'within' if year <= leave else 'above'} the limit {leave:,.2f}",
"" if year <= leave else "leave the flat rate scheme: Tax settings > Scheme"))
out["flat_rate_pct"] = rate
out["rows"].append({"document_ref": f"FLAT:{pk}", "tax_code": "FRS", "entry_kind": "flat_rate", "direction": "sale", "tax_point": end,
"net_amount": turnover, "vat_amount": vat, "box1_amount": vat, "box_value": "6", "source": "flat_rate",
"detail": f"flat rate {rate}% of {turnover}"})
out["extras"]["flat_rate"] = {"turnover_incl_vat": turnover, "goods_cost": goods, "limited_cost_trader": lct, "rate_pct": rate,
"sector_rate_pct": reg.flat_rate_pct, "first_year_discount_pct": discount, "box1": vat,
"turnover_12_months": year, "eligible": year <= leave}
# ═══ partial exemption (TAX-48) ═════════════════════════════════════════════════════════════════
def _partial_exemption(repo, vrn, ob, reg, out) -> None:
start, end = ob["period_start"], ob["period_end"]
ye = year_end_for(reg.partial_exemption_year_end, end)
row = repo.pe_years.get(vrn=vrn, year_end=ye)
if row is None or row["provisional_pct"] is None:
raise unmet("pe_percentage", f"the partial exemption year ending {ye} has no recovery percentage: this year starts from last year's",
"Partial exemption > record the year's provisional percentage")
pct = Decimal(str(row["provisional_pct"]))
buckets = {"taxable": Z, "exempt": Z, "residual": Z}
mine = [r for r in out["rows"] if r["direction"] == "purchase" and r["source"] in PURCHASE_SOURCES
and not held_elsewhere(repo, r, vrn, ob["period_key"])] # only what THIS return claims
for r in mine:
buckets[r.get("attribution") or "taxable"] += Decimal(str(r["vat_reclaimable"] or 0))
months = months_of(start, end)
monthly, half = stat(repo, "pe_de_minimis_monthly", end), stat(repo, "pe_de_minimis_pct", end) / HUNDRED
total = sum(buckets.values(), Z)
tests = [
gate("de_minimis_1", buckets["exempt"] <= monthly * months and buckets["exempt"] <= total * half,
f"exempt input VAT {buckets['exempt']:,.2f} against {monthly * months:,.2f} ({monthly}/month x {months}) and {half * 100}% of {total:,.2f}"),
gate("de_minimis_2", buckets["exempt"] + buckets["residual"] <= monthly * months and buckets["exempt"] + buckets["residual"] <= total * half,
f"exempt plus residual input VAT {buckets['exempt'] + buckets['residual']:,.2f} against {monthly * months:,.2f} and {half * 100}% of {total:,.2f}"),
gate("de_minimis_3", total <= monthly * months, f"all input VAT {total:,.2f} against {monthly * months:,.2f}"),
]
holds = any(t["met"] for t in tests)
factor = {"taxable": Decimal(1), "exempt": Decimal(1) if holds else Z, "residual": pct / HUNDRED}
got = dict.fromkeys(buckets, Z)
for r in mine:
a = r.get("attribution") or "taxable"
r["box4_amount"] = r2(Decimal(str(r["box4_amount"])) * factor[a])
got[a] += r["box4_amount"]
out["gates"] += tests
out["extras"]["partial_exemption"] = {
"year_end": ye, "provisional_pct": pct, "de_minimis_holds": holds,
"input_vat": buckets, "reclaimed": got,
"taxable_and_residual_reclaimed": got["taxable"] + got["residual"], "exempt_reclaimed": got["exempt"]}
def annual_percentage(repo, taxable: Decimal, exempt: Decimal, residual_input: Decimal, months: int, on: date) -> Decimal:
"""The year's recovery percentage: taxable / all supplies, rounded UP to a whole percent while residual input tax averages
at most the cap a month (VAT Notice 706 para 11); above it the exact percentage stands (2 places)."""
all_ = taxable + exempt
if all_ <= 0:
return HUNDRED
raw = taxable * HUNDRED / all_
if residual_input / months <= stat(repo, "pe_residual_cap_monthly", on):
return raw.quantize(Decimal(1), ROUND_CEILING)
return raw.quantize(P2, ROUND_HALF_UP)
# ═══ import VAT (TAX-44, TAX-45), annual adjustments (TAX-48, TAX-49) ═══════════════════════════
def _imports(repo, vrn, ob, out) -> None:
for e in repo.imports.where():
d = e["entry_date"]
paid = e["tax_code"] == "IMPORT_PAID"
if paid and not e["certificate_listed"]:
continue # reclaimed only once the import VAT certificate lists the entry
if not (ob["period_start"] <= d <= ob["period_end"]) and not (paid and d < ob["period_start"]):
continue # (a paid entry listed after its own return was locked: the next return prepared)
row = {"document_ref": e["mrn"], "tax_code": e["tax_code"], "entry_kind": "import", "direction": "purchase",
"tax_point": d, "net_amount": e["customs_value"], "vat_amount": e["import_vat"],
"vat_reclaimable": e["box_vat_input"], "box1_amount": e["box_vat_output"], "box4_amount": e["box_vat_input"],
"box_value": "7", "source": "pva", "detail": f"import {e['mrn']} ({e['method']})"}
if d < ob["period_start"] and held_elsewhere(repo, row | {"vrn": vrn, "period_key": ob["period_key"]}, vrn, ob["period_key"]):
continue # a return that already holds it
out["rows"].append(row)
def _pva_adjustments(repo, vrn, ob, out) -> None:
"""A reconciled statement line explained as HMRC's figure: the difference is an adjustment naming the MRN (box 1 and box 4)."""
for st in repo.pva_statements.where(lifecycle="reconciled"):
y, m = (int(x) for x in st["month"].split("-"))
if not (ob["period_start"] <= date(y, m, 1) <= ob["period_end"]):
continue
for ln in repo.pva_lines.where(month=st["month"]):
if ln["use_hmrc_figure"] and ln["difference"] not in (None, 0):
diff = Decimal(str(ln["difference"]))
out["rows"].append({"document_ref": ln["mrn"], "tax_code": "PVA", "entry_kind": "pva_adjustment", "direction": "purchase",
"tax_point": date(y, m, 1), "net_amount": Z, "vat_amount": diff, "vat_reclaimable": diff,
"box1_amount": diff, "box4_amount": diff, "box_value": None, "source": "pva",
"detail": f"HMRC's statement figure for {ln['mrn']}: {ln['explanation']}"})
def _annual_rows(repo, vrn, ob, out) -> None:
"""The first return after a partial-exemption year end carries its annual adjustment and each capital goods interval."""
start = ob["period_start"]
for y in repo.pe_years.where(vrn=vrn):
amt = y["adjustment_amount"]
if y["final_pct"] is None or not amt or y["year_end"] >= start or y["adjusted_in_period_key"] not in (None, "", ob["period_key"]):
continue
out["rows"].append({"document_ref": f"PE-ANNUAL:{y['year_end']}", "tax_code": "PE", "entry_kind": "partial_exemption",
"direction": "purchase", "tax_point": y["year_end"], "net_amount": Z, "vat_amount": amt, "vat_reclaimable": amt,
"box4_amount": amt, "box_value": None, "source": "partial_exemption",
"detail": f"annual adjustment for the year ending {y['year_end']}: {y['provisional_pct']}% -> {y['final_pct']}%",
"_mark": ("pe", y["year_end"])})
for item in repo.capital_items.where():
for a in json.loads(item["adjustments_json"] or "[]"):
if day(a["year_end"]) >= start or a.get("period_key") not in (None, "", ob["period_key"]):
continue
out["rows"].append({"document_ref": item["item_ref"], "tax_code": f"I{a['interval']}", "entry_kind": "capital_goods",
"direction": "purchase", "tax_point": day(a["year_end"]), "net_amount": Z, "vat_amount": Decimal(a["amount"]),
"vat_reclaimable": Decimal(a["amount"]), "box4_amount": Decimal(a["amount"]), "box_value": None,
"source": "adjustment", "detail": f"capital goods item {item['item_ref']}, interval {a['interval']}: "
f"{a['pct']}% against {item['first_pct']}%", "_mark": ("cap", item["item_ref"], a["interval"])})
def mark(repo, tag: tuple, period_key: str) -> None:
"""After the claim succeeded: remember which return carries the annual adjustment / capital goods interval."""
if tag[0] == "pe":
repo.pe_years.conn.execute(repo.pe_years.table.update().where(repo.pe_years.table.c.year_end == tag[1]).values(adjusted_in_period_key=period_key))
else:
item = repo.capital_items.get(item_ref=tag[1])
adj = json.loads(item["adjustments_json"] or "[]")
for a in adj:
if a["interval"] == tag[2]:
a["period_key"] = period_key
repo.capital_items.put({**item, "adjustments_json": json.dumps(adj)})
# ═══ the lock gates (TAX-45, TAX-37) ═══════════════════════════════════════════════════════════
def lock_gates(repo, vrn: str, ob: dict) -> None:
"""The period's postponed-import statements are reconciled; flat-rate eligibility is answered (an ineligible business must have
recorded its way out of the scheme from a later period)."""
unmet_: list = []
months = set()
for e in repo.imports.where():
if e["tax_code"] == "PVA" and ob["period_start"] <= e["entry_date"] <= ob["period_end"]:
months.add(f"{e['entry_date'].year:04d}-{e['entry_date'].month:02d}")
for m in sorted(months):
st = repo.pva_statements.get(month=m)
if st is None or st["lifecycle"] != "reconciled":
unmet_.append(Gate("pva_reconciled", False, f"the postponed import VAT statement for {m} is "
+ ("not imported" if st is None else "not reconciled") + ": box 1 and box 4 must carry HMRC's figures",
"Import VAT > upload the statement and reconcile it"))
regs = repo.registrations.rows()
reg = registration.status_on(regs, ob["period_start"])
if reg is not None and reg.scheme == "flat_rate":
docs = [d for d in (repo.documents.get(r)[0] for r in repo.documents.refs()) if d.status == "frozen" and d.registered and d.tax_point]
year = r2(_turnover(docs, "sale", add_months(ob["period_end"], -12) + timedelta(days=1), ob["period_end"], "gross"))
limit = stat(repo, "flat_leave_limit", ob["period_end"])
left = any(r.effective_from > ob["period_end"] and r.scheme != "flat_rate" and r.change_state != "withdrawn" for r in regs)
if year > limit and not left:
unmet_.append(Gate("flat_rate_eligible", False, f"turnover including VAT {year:,.2f} in the 12 months to {ob['period_end']} is above "
f"the flat rate limit {limit:,.2f}", "Tax settings > Scheme: change the scheme from the next VAT period"))
if unmet_:
raise GateUnmet(unmet_)
def unclaimed_rows(repo, vrn: str, ob: dict) -> list:
"""The entries a fresh plan of this period would claim that NO return holds yet: a document frozen (or a certificate listed)
after the last prepare. A lock that froze the boxes without them would leave them in no return."""
out = []
for row in plan(repo, vrn, ob)["rows"]:
if repo.entries.get(document_ref=row["document_ref"], tax_code=row["tax_code"], entry_kind=row["entry_kind"]) is None:
out.append(row)
return out