dodil/erp-brick-tax
DODIL ERP brick: tax
src/brick_tax/domain/vat_return.py
"""The VAT return (TAX-28 .. TAX-33): nine boxes that are each a SUM of entries, a lifecycle draft -> locked -> submitting ->
submitted, and the arithmetic HMRC checks. Pure: the use case hands in the entries, the adjustments and the thresholds.
box 1 VAT due on sales box 6 total sales ex VAT } whole pounds: pence are dropped (HMRC's rule)
box 2 VAT due on NI acquisitions box 7 total purchases ex VAT }
box 3 box 1 + box 2 box 8 goods supplied to the EU
box 4 VAT reclaimed on purchases box 9 acquisitions from the EU
box 5 |box 3 - box 4|
A document's tax enters ONE return (the entry's key); a submitted return never changes: a correction is an adjustment on a
later return or a VAT652 notice (the thresholds are dated tax_statutory rows).
"""
from dataclasses import asdict, dataclass, fields
from datetime import date, datetime
from decimal import ROUND_DOWN, Decimal
from typing import Optional
from brick_tax.domain.errors import RuleBroken, unmet
DRAFT, LOCKED, SUBMITTING, SUBMITTED = "draft", "locked", "submitting", "submitted"
Z = Decimal(0)
BOX_FIELDS = ("box1_vat_due_sales", "box2_vat_due_acquisitions", "box3_total_vat_due", "box4_vat_reclaimed", "box5_net_vat_due",
"box6_total_sales_ex_vat", "box7_total_purchases_ex_vat", "box8_goods_supplied_ex_vat", "box9_acquisitions_ex_vat")
@dataclass
class VatReturn:
vrn: str
period_key: str
lifecycle: str = DRAFT
accounting_basis: str = "invoice"
flat_rate_pct: Optional[Decimal] = None
box1_vat_due_sales: Decimal = Z
box2_vat_due_acquisitions: Decimal = Z
box3_total_vat_due: Decimal = Z
box4_vat_reclaimed: Decimal = Z
box5_net_vat_due: Decimal = Z
box6_total_sales_ex_vat: Decimal = Z
box7_total_purchases_ex_vat: Decimal = Z
box8_goods_supplied_ex_vat: Decimal = Z
box9_acquisitions_ex_vat: Decimal = Z
prepared_by: str = ""
locked_by: str = ""
locked_by_sub: str = ""
locked_at: Optional[datetime] = None
submitted_by: str = ""
submitted_at: Optional[datetime] = None
declaration_text: str = ""
processing_date: str = ""
form_bundle_number: str = ""
payment_indicator: str = ""
charge_ref_number: str = ""
fraud_headers_json: str = ""
settle_journal_id: str = ""
hmrc_error: str = ""
@classmethod
def from_row(cls, r: dict) -> "VatReturn":
out = {}
for f in fields(cls):
v = r.get(f.name)
if f.name in BOX_FIELDS:
v = Decimal(str(v)) if v is not None else Z
elif f.name == "flat_rate_pct":
v = Decimal(str(v)) if v is not None else None
elif f.type is str or f.name in ("vrn", "period_key"):
v = v or ""
out[f.name] = v
return cls(**out)
def to_row(self) -> dict:
return {k: (v or None if k in ("locked_at", "submitted_at") else v) for k, v in asdict(self).items()}
@property
def boxes(self) -> dict:
return {i + 1: getattr(self, name) for i, name in enumerate(BOX_FIELDS)}
def editable(self) -> None:
if self.lifecycle != DRAFT:
raise RuleBroken(f"return {self.period_key} is {self.lifecycle}: only a draft changes"
+ (" (a correction after submission is an adjustment on a later return, or a VAT652 notice)"
if self.lifecycle == SUBMITTED else ""))
def set_boxes(self, entries: list, adjustments: list) -> None:
"""Every box is the SUM of its entries (and of the adjustments in box 1 and box 4)."""
self.editable()
b1 = sum((Decimal(str(e["box1_amount"] or 0)) for e in entries), Z) + sum(
(Decimal(str(a["amount"])) for a in adjustments if a["box"] == 1), Z)
b4 = sum((Decimal(str(e["box4_amount"] or 0)) for e in entries), Z) + sum(
(Decimal(str(a["amount"])) for a in adjustments if a["box"] == 4), Z)
def pounds(box: str) -> Decimal:
total = sum((Decimal(str(e["net_amount"] or 0)) for e in entries if box in (e["box_value"] or "").split(",")), Z)
return total.quantize(Decimal(1), ROUND_DOWN).quantize(Decimal("0.01")) # pence dropped, toward zero
b2 = sum((Decimal(str(e.get("box2_amount") or 0)) for e in entries), Z)
self.box1_vat_due_sales, self.box2_vat_due_acquisitions = b1, b2
self.box3_total_vat_due = b1 + self.box2_vat_due_acquisitions
self.box4_vat_reclaimed = b4
self.box5_net_vat_due = abs(self.box3_total_vat_due - b4)
self.box6_total_sales_ex_vat, self.box7_total_purchases_ex_vat = pounds("6"), pounds("7")
self.box8_goods_supplied_ex_vat, self.box9_acquisitions_ex_vat = pounds("8"), pounds("9")
for name in BOX_FIELDS:
setattr(self, name, getattr(self, name).quantize(Decimal("0.01")))
def lock(self, *, by: str, by_sub: str, now: datetime) -> None:
if self.lifecycle == LOCKED:
return
self.editable()
self.lifecycle, self.locked_by, self.locked_by_sub, self.locked_at = LOCKED, by, by_sub, now
def reopen(self) -> None:
if self.lifecycle not in (LOCKED,):
raise RuleBroken(f"return {self.period_key} is {self.lifecycle}: only a locked return is reopened (never a submitted one)")
self.lifecycle, self.locked_by, self.locked_by_sub, self.locked_at = DRAFT, "", "", None
def hmrc_body(self) -> dict:
"""The body of POST /organisations/vat/{vrn}/returns (HMRC's field names)."""
return {"periodKey": self.period_key, "vatDueSales": self.box1_vat_due_sales, "vatDueAcquisitions": self.box2_vat_due_acquisitions,
"totalVatDue": self.box3_total_vat_due, "vatReclaimedCurrPeriod": self.box4_vat_reclaimed,
"netVatDue": self.box5_net_vat_due, "totalValueSalesExVAT": int(self.box6_total_sales_ex_vat),
"totalValuePurchasesExVAT": int(self.box7_total_purchases_ex_vat),
"totalValueGoodsSuppliedExVAT": int(self.box8_goods_supplied_ex_vat),
"totalAcquisitionsExVAT": int(self.box9_acquisitions_ex_vat), "finalised": True}
def add_months(d: date, n: int) -> date:
y, m = divmod(d.month - 1 + n, 12)
year, month = d.year + y, m + 1
last = [31, 29 if year % 4 == 0 and (year % 100 or year % 400 == 0) else 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31][month - 1]
return date(year, month, min(d.day, last))
def money(v) -> str:
return f"{Decimal(str(v)):,.2f}"
def check_error_threshold(net: Decimal, box6: Decimal, small: Decimal, large: Decimal, pct: Decimal) -> None:
"""Net errors within `small` are corrected on the next return; so are those within `large` AND within `pct`% of box 6 (Notice 700/45)."""
e = abs(net)
if e <= small or (e <= large and e <= box6 * pct / Decimal(100)):
return
raise unmet("within_error_threshold", f"net errors of {money(e)} exceed what may be corrected on a return (Notice 700/45)",
"notify HMRC on form VAT652 and record the notice")
def check_four_years(period_end: date, today: date) -> None:
if period_end < add_months(today, -48):
raise unmet("within_four_years", f"the error is in a period that ended {period_end}, more than four years ago: it cannot be corrected",
"HMRC's four-year time limit applies; contact HMRC")