dodil/erp-brick-tax
DODIL ERP brick: tax
src/brick_tax/application/declarations.py
"""Trade declarations and import VAT (v4: TAX-42 .. TAX-45): the EC sales list and Intrastat for Northern Ireland goods trade with
the EU, the import entries purchasing records, and the monthly postponed-import-VAT statement reconciled to them.
EC sales list one line per customer VAT number: the month's goods net in whole pounds, indicator 0; the total must tie to box 8
Intrastat one line per (commodity code, partner country, nature of transaction, country of origin), read from ar's invoice lines
and product's commodity code and net weight; arrivals only once their threshold has passed
import entry keyed by MRN: postponed VAT accounting (PVA: output and input in boxes 1 and 4, value in box 7) or import VAT paid
(reclaimed only once the monthly import VAT certificate lists the entry)
PVA statement one per month, each line matched by MRN, every difference explained, then the return uses HMRC's figures
HMRC offers no API for either declaration: the file is built for a person to upload, who records the submission reference.
"""
import json
from datetime import date, timedelta
from decimal import ROUND_DOWN, Decimal
from typing import Literal, Optional
from erp_platform.bricks import Refused
from erp_platform.repository import Conflict
from erp_platform.types import text
from pydantic import BaseModel, ConfigDict
from sqlalchemy.exc import IntegrityError
from brick_tax.application import schemes
from brick_tax.application import use_cases as uc
from brick_tax.application.inputs import Money, NonNegMoney, Reason, day
from brick_tax.application.use_cases import (
IMPORT_RECORD,
IMPORT_WRITE,
TAX,
TRADE_WRITE,
)
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
Z = Decimal(0)
WHOLE = Decimal(1)
def _period_dates(period: str) -> tuple:
"""'2026-11' -> the month; '2026-Q4' -> the calendar quarter."""
try:
if "-Q" in period:
y, q = period.split("-Q")
m = (int(q) - 1) * 3 + 1
return date(int(y), m, 1), add_months(date(int(y), m, 1), 3) - timedelta(days=1)
y, m = (int(x) for x in period.split("-"))
return date(y, m, 1), add_months(date(y, m, 1), 1) - timedelta(days=1)
except (ValueError, IndexError):
raise Refused(422, "a declaration period is a month such as 2026-11 or a quarter such as 2026-Q4") from None
def _eu_sales(repo, frm: date, to: date) -> list:
"""The frozen sale documents in a range that carry an eu_goods code: [(doc, {code: net in sterling})]."""
eu = {c.tax_code for c in repo.codes.all() if c.eu_goods and c.direction == "sale"}
out = []
for ref, (d, _) in sorted(repo.documents.get_many(repo.documents.refs()).items()):
if d.status != "frozen" or d.direction != "sale" or not d.registered or d.tax_point is None or not (frm <= d.tax_point <= to):
continue
c = d.computed()
net = {r.code: r.net * (Decimal(1) if d.currency == "GBP" else d.fx_rate) for r in c.summary if r.code in eu}
if net:
out.append((d, net))
return out
def _vat_number(repo, doc) -> str:
p = repo.parties.effective(doc.bp_id, doc.tax_point) if doc.bp_id is not None else None
return ((p or {}).get("vat_number") or "").replace(" ", "").upper()
def _view(repo, d: dict) -> dict:
lines = repo.trade_lines.where(kind=d["kind"], period=d["period"], version=d["version"])
return {"kind": d["kind"], "period": d["period"], "version": d["version"], "lifecycle": d["lifecycle"], "total": d["total"],
"lines": [{k: v for k, v in ln.items() if k not in ("kind", "period", "version", "created_at", "updated_at")} for ln in lines],
"file": d["file_text"], "gates": json.loads(d["gates_json"] or "[]"), "submission_ref": d["submission_ref"] or None,
"filed_on": d["filed_on"]}
def _box8(repo, frm: date, to: date) -> Decimal:
total = sum((Decimal(str(e["net_amount"] or 0)) for e in repo.entries.where() if frm <= e["tax_point"] <= to
and "8" in (e["box_value"] or "").split(",")), Z)
return total.quantize(WHOLE, ROUND_DOWN)
def _gates_ec(repo, total: Decimal, frm: date, to: date, quarterly: bool, goods_by_quarter: dict) -> list:
box8 = _box8(repo, frm, to)
gates = [Gate("ec_sales_ties_box8", total == box8, f"the list totals {total:,.0f} and box 8 of the returns for {frm} to {to} holds {box8:,.0f}"
if total != box8 else f"the list ties to box 8 ({box8:,.0f})",
"" if total == box8 else "prepare or refresh the VAT return for those months, then prepare the list again").as_dict()]
if quarterly:
limit = schemes.stat(repo, "ec_sales_quarterly_limit", to)
over = {q: v for q, v in goods_by_quarter.items() if v > limit}
gates.append(Gate("quarterly_allowed", not over, f"goods stay below {limit:,.0f} a quarter for the current and four previous quarters"
if not over else f"goods exceeded {limit:,.0f} in {', '.join(sorted(over))}: file monthly",
"" if not over else "prepare a monthly list").as_dict())
return gates
@TAX.use_case(kind="command", route="POST /trade-declarations/prepare", permission=TRADE_WRITE)
def prepare_trade_declaration(caller: dict, kind: Literal["ec_sales_list", "intrastat_dispatches"], period: text(10, min_length=7)) -> dict:
"""Prepare the EC sales list or the Intrastat dispatches declaration of a month (2026-11) — the EC list also of a quarter
(2026-Q4) while goods stay below 35,000 a quarter for the current and four previous quarters. Needs an XI (ni_goods)
registration. EC list: one line per customer VAT number, the goods net in whole pounds, indicator 0, tied to box 8. Intrastat:
the dispatches threshold must be passed; lines come from ar's invoice lines and product's commodity code, net weight and
origin; a product with no commodity code is a gate. The result carries the upload file; a repeat replaces a draft in place and
a filed declaration starts a new version (an amendment)."""
def unit(repo):
frm, to = _period_dates(period)
reg = registration.status_on(repo.registrations.rows(), to)
if reg is None or not reg.ni_goods:
raise unmet("ni_goods", "trade declarations are for a Northern Ireland (XI) registration trading goods with the EU",
"Tax settings > Registration > goods with the EU")
lines, gates = [], []
if kind == "ec_sales_list":
by: dict = {}
for doc, net in _eu_sales(repo, frm, to):
number = _vat_number(repo, doc)
by[number] = by.get(number, Z) + sum(net.values(), Z)
for i, (number, total) in enumerate(sorted(by.items()), 1):
lines.append({"line_no": i, "partner_vat_number": number, "partner_country": number[:2], "value": total.quantize(WHOLE, ROUND_DOWN),
"indicator": "0"})
total = sum((ln["value"] for ln in lines), Z)
quarterly = "-Q" in period
qs = {}
if quarterly:
for back in range(5):
qf = add_months(frm, -3 * back)
qs[f"{qf.year}-Q{(qf.month - 1) // 3 + 1}"] = sum((sum(n.values(), Z) for _, n in _eu_sales(repo, qf, add_months(qf, 3) - timedelta(days=1))), Z)
gates = _gates_ec(repo, total, frm, to, quarterly, qs)
file_text = "VAT number,Value,Indicator\n" + "".join(f"{ln['partner_vat_number']},{ln['value']:.0f},{ln['indicator']}\n" for ln in lines)
else:
year = sum((sum(n.values(), Z) for _, n in _eu_sales(repo, add_months(to, -12) + timedelta(days=1), to)), Z)
limit = schemes.stat(repo, "intrastat_dispatches", to)
if year <= limit:
raise unmet("intrastat_threshold", f"dispatches to the EU were {year:,.2f} in the 12 months to {to}: Intrastat is declared above {limit:,.2f}",
"no declaration is due")
groups: dict = {}
for doc, _net in _eu_sales(repo, frm, to):
inv = seams.receivables.invoice(doc.document_ref)
if inv is None:
gates.append(Gate("ar_invoice", False, f"ar could not be asked for invoice {doc.document_ref}", "check the connection to ar").as_dict())
continue
eu_codes = {c.tax_code for c in repo.codes.all() if c.eu_goods}
for ln in inv.get("lines") or []:
if ln.get("tax_code") not in eu_codes:
continue
p = seams.products.product(ln["product_id"], doc.tax_point) or {}
unit_code, qty = ln.get("unit_code"), Decimal(str(ln.get("quantity") or 0))
measures = next((u.get("measures") for u in p.get("units") or [] if u.get("unit_code") == unit_code), None) or {}
weight = Decimal(str(measures.get("net_weight") or 0)) / (Decimal(1000) if measures.get("weight_unit_code") == "g" else Decimal(1))
code = p.get("commodity_code") or ""
if not code:
gates.append(Gate("commodity_code", False, f"product {ln['product_id']} has no commodity code", "Product > the product > Trade").as_dict())
number = _vat_number(repo, doc)
g = groups.setdefault((code, number[:2], "11", p.get("country_of_origin") or ""), {
"partner_vat_number": number, "value": Z, "net_mass_kg": Z, "supplementary_units": Z})
g["value"] += Decimal(str(ln["net_amount"]))
g["net_mass_kg"] += qty * weight
g["supplementary_units"] += qty
for i, ((code, country, nature, origin), g) in enumerate(sorted(groups.items()), 1):
lines.append({"line_no": i, "commodity_code": code, "partner_country": country, "nature": nature, "origin_country": origin,
"partner_vat_number": g["partner_vat_number"], "value": g["value"].quantize(WHOLE, ROUND_DOWN),
"net_mass_kg": g["net_mass_kg"], "supplementary_units": g["supplementary_units"], "indicator": ""})
total = sum((ln["value"] for ln in lines), Z)
file_text = "Commodity code,Country,Nature,Origin,Value,Net mass kg,Supplementary units,Partner VAT number\n" + "".join(
f"{ln['commodity_code']},{ln['partner_country']},{ln['nature']},{ln['origin_country']},{ln['value']:.0f},"
f"{ln['net_mass_kg']},{ln['supplementary_units']},{ln['partner_vat_number']}\n" for ln in lines)
have = [r for r in repo.trade_declarations.where(kind=kind, period=period)]
latest = have[-1] if have else None
version = (latest["version"] + (1 if latest["lifecycle"] == "filed" else 0)) if latest else 1
head = {"kind": kind, "period": period, "version": version, "lifecycle": "draft", "total": total, "file_text": file_text,
"prepared_by": uc._who(caller), "gates_json": json.dumps(gates)}
if repo.trade_declarations.get(kind=kind, period=period, version=version) is not None:
repo.trade_declarations.put(head) # this draft, refreshed in place
else:
try: # a new version: check, then a plain INSERT, so a racer collides -> 409
repo.conn.execute(repo.trade_declarations.table.insert().values(**head))
except IntegrityError:
raise Conflict("tax_trade_declaration", {"kind": kind, "period": period, "version": version}) from None
for ln in lines: # upsert in place; a line the draft no longer has is deleted (distinct keys)
repo.trade_lines.put({"kind": kind, "period": period, "version": version, "line_no": ln["line_no"],
**{k: v for k, v in ln.items() if k != "line_no"}})
for old in repo.trade_lines.where(kind=kind, period=period, version=version):
if old["line_no"] > len(lines):
repo.trade_lines.remove(kind=kind, period=period, version=version, line_no=old["line_no"])
return _view(repo, repo.trade_declarations.get(kind=kind, period=period, version=version))
return uc._atomic(unit)
@TAX.use_case(kind="command", route="POST /trade-declarations/file", permission=TRADE_WRITE)
def record_trade_declaration_filed(caller: dict, kind: Literal["ec_sales_list", "intrastat_dispatches"], period: text(10, min_length=7),
submission_ref: text(100, min_length=1), filed_on: Optional[text(10)] = None) -> dict:
"""Record that the person uploaded the declaration to HMRC, with the submission reference and date. Refused unless the EC sales
list ties to box 8 for its months and every Intrastat line has a commodity code. An amendment is a new version (prepare again)."""
def unit(repo):
rows = repo.trade_declarations.where(kind=kind, period=period)
if not rows:
raise Refused(404, f"no {kind} declaration for {period}: prepare it first")
d = rows[-1]
if d["lifecycle"] == "filed":
return _view(repo, d)
frm, to = _period_dates(period)
bad = []
if kind == "ec_sales_list":
total = Decimal(str(d["total"]))
bad = [g for g in _gates_ec(repo, total, frm, to, False, {}) if not g["met"]]
else:
if any(not ln["commodity_code"] for ln in repo.trade_lines.where(kind=kind, period=period, version=d["version"])):
bad = [Gate("commodity_code", False, "a line has no commodity code", "Product > the product > Trade").as_dict()]
if bad:
raise GateUnmet([Gate(**g) for g in bad])
repo.trade_declarations.put({**{k: v for k, v in d.items() if k not in ("created_at", "updated_at")}, "lifecycle": "filed",
"submission_ref": submission_ref, "filed_on": day(filed_on) or uc._today(repo), "filed_by": uc._who(caller)})
return _view(repo, repo.trade_declarations.get(kind=kind, period=period, version=d["version"]))
return uc._atomic(unit)
@TAX.use_case(kind="query", route="GET /trade-declarations/{kind}/{period}", permission=TRADE_WRITE)
def get_trade_declaration(caller: dict, kind: Literal["ec_sales_list", "intrastat_dispatches"], period: text(10, min_length=7)) -> dict:
"""The latest version of a trade declaration: its lines, total, upload file, gates and (once filed) the submission reference."""
with uc.repository() as repo:
rows = repo.trade_declarations.where(kind=kind, period=period)
if not rows:
raise Refused(404, f"no {kind} declaration for {period}")
return _view(repo, rows[-1])
# ═══ import entries (TAX-44) ════════════════════════════════════════════════════════════════════
def _import_view(r: dict, replay: bool) -> dict:
return {"mrn": r["mrn"], "tax_code": r["tax_code"], "box_vat_output": r["box_vat_output"], "box_vat_input": r["box_vat_input"],
"box_value": r["box_value"], "certificate_listed": bool(r["certificate_listed"]), "replay": replay}
@TAX.use_case(kind="command", route="POST /imports", permission=IMPORT_RECORD, idempotent=True)
def record_import(caller: dict, mrn: text(40, min_length=5), entry_date: text(10, min_length=10), bill_ref: text(200, min_length=1), bp_id: int,
customs_value: Money, import_vat: NonNegMoney, method: Literal["postponed", "paid"]) -> dict:
"""Purchasing records an import entry with how its VAT was accounted. Postponed VAT accounting (code PVA): the VAT is output in
box 1 and input in box 4 (it nets to 0.00 for a fully taxable business), the value in box 7 never box 6. Import VAT paid (code
IMPORT_PAID) is reclaimed only once the monthly import VAT certificate lists the entry (confirm_import_certificate). Keyed by
MRN: recording it again answers the same entry."""
def unit(repo):
have = repo.imports.get(mrn=mrn)
if have is not None:
return _import_view(have, True)
pva = method == "postponed"
row = {"mrn": mrn, "entry_date": uc._date(entry_date, "entry_date"), "bill_ref": bill_ref, "bp_id": int(bp_id), "customs_value": customs_value,
"import_vat": import_vat, "method": method, "tax_code": "PVA" if pva else "IMPORT_PAID", "box_vat_output": import_vat if pva else Z,
"box_vat_input": import_vat, "box_value": customs_value, "certificate_listed": pva, "recorded_by": uc._who(caller)}
try:
repo.conn.execute(repo.imports.table.insert().values(**row))
except IntegrityError:
raise Conflict("tax_import_entry", {"mrn": mrn}) from None
return _import_view(row, False)
return uc._atomic(unit)
@TAX.use_case(kind="command", route="POST /imports/{mrn}/certificate-listed", permission=IMPORT_WRITE, idempotent=True)
def confirm_import_certificate(caller: dict, mrn: text(40, min_length=5)) -> dict:
"""The monthly import VAT certificate (C79, or its CDS successor) lists this import: its VAT paid becomes reclaimable on the
next return prepared."""
def unit(repo):
row = repo.imports.get(mrn=mrn)
if row is None:
raise Refused(404, f"no import entry {mrn}")
repo.imports.put({**{k: v for k, v in row.items() if k not in ("created_at", "updated_at")}, "certificate_listed": True})
return _import_view(repo.imports.get(mrn=mrn), False)
return uc._atomic(unit)
# ═══ the postponed import VAT statement (TAX-45) ════════════════════════════════════════════════
class PvaLineIn(BaseModel):
model_config = ConfigDict(extra="forbid")
mrn: text(40, min_length=5)
import_vat: NonNegMoney
def _statement_view(repo, st: dict) -> dict:
lines = repo.pva_lines.where(month=st["month"])
return {"month": st["month"], "lifecycle": st["lifecycle"], "file": st["file_text"],
"lines": [{"mrn": ln["mrn"], "statement_vat": ln["statement_vat"], "our_vat": ln["our_vat"], "status": ln["status"],
"difference": ln["difference"], "explanation": ln["explanation"] or None,
"use_hmrc_figure": bool(ln["use_hmrc_figure"])} for ln in lines]}
@TAX.use_case(kind="command", route="POST /pva-statements", permission=IMPORT_WRITE)
def import_pva_statement(caller: dict, month: text(7, min_length=7), lines: list[PvaLineIn], file_text: Optional[text(100000)] = None) -> dict:
"""Upload a month's postponed import VAT statement (one per month): each line is matched by MRN against our import entries —
matched, differs (both amounts shown) or missing (no import entry). The statement file is kept read-only."""
def unit(repo):
_period_dates(month)
if repo.pva_statements.get(month=month) is not None:
raise unmet("one_statement_per_month", f"the statement for {month} is already imported: a statement is never replaced",
"explain its differences instead")
try: # check (above), then a plain INSERT: a racer collides -> 409, never an overwrite
repo.conn.execute(repo.pva_statements.table.insert().values(month=month, lifecycle="imported", file_text=file_text or "",
imported_by=uc._who(caller)))
except IntegrityError:
raise Conflict("tax_pva_statement", {"month": month}) from None
for ln in lines:
ours = repo.imports.get(mrn=ln.mrn)
our = Decimal(str(ours["import_vat"])) if ours is not None and ours["tax_code"] == "PVA" else None
status = "missing" if our is None else ("matched" if our == ln.import_vat else "differs")
repo.pva_lines.put({"month": month, "mrn": ln.mrn, "statement_vat": ln.import_vat, "our_vat": our, "status": status,
"difference": ln.import_vat - (our or Z), "use_hmrc_figure": False})
return _statement_view(repo, repo.pva_statements.get(month=month))
return uc._atomic(unit)
@TAX.use_case(kind="command", route="POST /pva-statements/{month}/explain", permission=IMPORT_WRITE)
def explain_pva_difference(caller: dict, month: text(7, min_length=7), mrn: text(40, min_length=5), reason: Reason,
use_hmrc_figure: bool = True) -> dict:
"""A person explains a statement line that differs or has no import entry. `use_hmrc_figure` (the default) makes the return use
HMRC's figure: the difference becomes an adjustment row naming the MRN in box 1 and box 4."""
if not (reason or "").strip():
raise Refused(422, "explaining a difference needs a reason")
def unit(repo):
st = repo.pva_statements.get(month=month)
ln = repo.pva_lines.get(month=month, mrn=mrn)
if st is None or ln is None:
raise Refused(404, f"no line {mrn} on the statement for {month}")
if st["lifecycle"] == "reconciled":
raise Refused(409, f"the statement for {month} is reconciled")
repo.pva_lines.put({**{k: v for k, v in ln.items() if k not in ("created_at", "updated_at")}, "status": "explained",
"explanation": reason, "use_hmrc_figure": bool(use_hmrc_figure), "explained_by": uc._who(caller)})
return _statement_view(repo, st)
return uc._atomic(unit)
@TAX.use_case(kind="command", route="POST /pva-statements/{month}/reconcile", permission=IMPORT_WRITE, idempotent=True)
def reconcile_pva_statement(caller: dict, month: text(7, min_length=7)) -> dict:
"""Reconcile the statement once every line is matched or explained by a person. The month's return then uses HMRC's figures."""
def unit(repo):
st = repo.pva_statements.get(month=month)
if st is None:
raise Refused(404, f"no statement for {month}: upload it first")
if st["lifecycle"] == "reconciled":
return _statement_view(repo, st)
open_ = [ln["mrn"] for ln in repo.pva_lines.where(month=month) if ln["status"] not in ("matched", "explained")]
if open_:
raise unmet("pva_lines_resolved", f"{len(open_)} statement line(s) are neither matched nor explained: {', '.join(open_)}",
"Import VAT > explain each difference")
repo.pva_statements.put({**{k: v for k, v in st.items() if k not in ("created_at", "updated_at")}, "lifecycle": "reconciled",
"reconciled_by": uc._who(caller)})
return _statement_view(repo, repo.pva_statements.get(month=month))
return uc._atomic(unit)
@TAX.use_case(kind="query", route="GET /pva-statements/{month}", permission=IMPORT_WRITE)
def get_pva_statement(caller: dict, month: text(7, min_length=7)) -> dict:
"""A month's postponed import VAT statement with each line's match."""
with uc.repository() as repo:
st = repo.pva_statements.get(month=month)
if st is None:
raise Refused(404, f"no statement for {month}")
return _statement_view(repo, st)