dodil/erp-brick-tax
DODIL ERP brick: tax
src/brick_tax/application/exemption.py
"""Partial exemption and the capital goods scheme (v4: TAX-48, TAX-49).
A business making exempt supplies reclaims input VAT by what it is for: taxable (all), exempt (only while de minimis holds) and
residual (at last year's recovery percentage). When the year ends the annual adjustment recomputes the year's percentage from the
supplies and puts the difference on the FIRST RETURN AFTER the year end. Expensive capital items (land and buildings of 250,000 or
more, computers of 50,000 or more) are adjusted once per interval after the first for any change in the recovery percentage.
"""
import json
from datetime import date
from decimal import Decimal
from typing import Literal, Optional
from erp_platform.bricks import Refused
from erp_platform.repository import Conflict
from erp_platform.types import IsoDate, text
from sqlalchemy.exc import IntegrityError
from brick_tax.application import returns as R
from brick_tax.application import schemes
from brick_tax.application import use_cases as uc
from brick_tax.application.inputs import Money, NonNegMoney, Pct
from brick_tax.application.use_cases import PE_WRITE, TAX
from brick_tax.domain import registration
from brick_tax.domain.errors import unmet
from brick_tax.domain.vat_return import add_months
Z = Decimal(0)
KINDS = ("land_building", "computer", "ship_aircraft")
def _year_view(r: dict, extra: Optional[dict] = None) -> dict:
return {"vrn": r["vrn"], "year_end": r["year_end"], "provisional_pct": r["provisional_pct"], "final_pct": r["final_pct"],
"taxable_supplies": r["taxable_supplies"], "exempt_supplies": r["exempt_supplies"], "exempt_input_vat": r["exempt_input_vat"],
"residual_input_vat": r["residual_input_vat"], "de_minimis_met": r["de_minimis_met"],
"adjustment_amount": r["adjustment_amount"], "adjusted_in_period_key": r["adjusted_in_period_key"] or None} | (extra or {})
@TAX.use_case(kind="command", route="POST /partial-exemption/years", permission=PE_WRITE)
def record_partial_exemption_year(caller: dict, year_end: IsoDate, provisional_pct: Pct) -> dict:
"""Record the partial exemption year ending on a date with last year's recovery percentage (the provisional one each return of
the year applies to residual input VAT). Once the year's annual adjustment has run the year is closed."""
def unit(repo):
vrn, ye = R._vrn(repo), uc._date(year_end, "year_end")
have = repo.pe_years.get(vrn=vrn, year_end=ye)
if have is not None and have["final_pct"] is not None:
raise Refused(409, f"the year ending {ye} is closed: its annual adjustment has run")
repo.pe_years.put({**(have or {}), "vrn": vrn, "year_end": ye, "provisional_pct": provisional_pct})
return _year_view(repo.pe_years.get(vrn=vrn, year_end=ye))
return uc._atomic(unit)
@TAX.use_case(kind="query", route="GET /partial-exemption/years/{year_end}", permission=PE_WRITE)
def get_partial_exemption_year(caller: dict, year_end: IsoDate) -> dict:
"""The partial exemption year: its provisional and final percentage, the figures behind the annual adjustment and where it went."""
with uc.repository() as repo:
row = repo.pe_years.get(vrn=R._vrn(repo), year_end=uc._date(year_end, "year_end"))
if row is None:
raise Refused(404, f"no partial exemption year ending {year_end}")
return _year_view(row)
@TAX.use_case(kind="command", route="POST /partial-exemption/years/{year_end}/adjust", permission=PE_WRITE, idempotent=True)
def run_partial_exemption_adjustment(caller: dict, year_end: IsoDate, taxable_supplies: Optional[Money] = None,
exempt_supplies: Optional[Money] = None) -> dict:
"""Run the annual adjustment once the year has ended: the year's recovery percentage = taxable supplies / all supplies, rounded
UP to a whole percent while residual input tax averages at most 400,000 a month (VAT Notice 706); the difference to the
provisional percentage on the year's residual input VAT goes onto the first return after the year end, and each capital goods
item still in its adjustment period is adjusted for the interval. The supplies are summed from the year's return entries unless
given (supplies the returns do not hold). Running it twice answers the same."""
def unit(repo):
vrn, ye, today = R._vrn(repo), uc._date(year_end, "year_end"), uc._today(repo)
if ye >= today:
raise unmet("year_ended", f"the partial exemption year ends on {ye}: the annual adjustment runs after it", f"run it from {ye}+1")
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 year ending {ye} has no recorded provisional percentage", "Partial exemption > record the year")
if row["final_pct"] is not None:
return _year_view(row, {"replay": True, "capital_goods": _capital_run(repo, ye, Decimal(str(row["final_pct"])), True)})
start = add_months(ye, -12) + (date(2000, 1, 2) - date(2000, 1, 1))
treatment = {c.tax_code: c.treatment for c in repo.codes.all()}
taxable = exempt = residual = exempt_in = total_in = Z
for e in repo.entries.where(vrn=vrn):
if not (start <= e["tax_point"] <= ye) or e["source"] not in schemes.PURCHASE_SOURCES + ("flat_rate",):
continue
net = Decimal(str(e["net_amount"] or 0))
if e["direction"] == "sale":
t = treatment.get(e["tax_code"], "")
if t == "exempt":
exempt += net
elif t in ("standard", "reduced", "zero", "reverse_charge"):
taxable += net
else:
v = Decimal(str(e["vat_reclaimable"] or 0))
total_in += v
if e["attribution"] == "residual":
residual += v
elif e["attribution"] == "exempt":
exempt_in += v
taxable = taxable if taxable_supplies is None else taxable_supplies
exempt = exempt if exempt_supplies is None else exempt_supplies
pct = schemes.annual_percentage(repo, taxable, exempt, residual, 12, ye)
adjustment = schemes.r2(residual * (pct - Decimal(str(row["provisional_pct"]))) / Decimal(100))
monthly = schemes.stat(repo, "pe_de_minimis_monthly", ye)
met = exempt_in <= monthly * 12 and exempt_in <= total_in * schemes.stat(repo, "pe_de_minimis_pct", ye) / Decimal(100)
repo.pe_years.put({**row, "final_pct": pct, "taxable_supplies": taxable, "exempt_supplies": exempt, "exempt_input_vat": exempt_in,
"residual_input_vat": residual, "de_minimis_met": met, "adjustment_amount": adjustment})
return _year_view(repo.pe_years.get(vrn=vrn, year_end=ye), {"replay": False, "capital_goods": _capital_run(repo, ye, pct, False)})
return uc._atomic(unit)
# ═══ capital goods (TAX-49) ═════════════════════════════════════════════════════════════════════
def _item_view(r: dict) -> dict:
return {"item_ref": r["item_ref"], "kind": r["kind"], "bought_on": r["bought_on"], "net_value": r["net_value"], "input_vat": r["input_vat"],
"intervals": r["intervals"], "first_pct": r["first_pct"], "adjustments": json.loads(r["adjustments_json"] or "[]"),
"disposed_on": r.get("disposed_on")}
@TAX.use_case(kind="command", route="POST /capital-goods", permission=PE_WRITE, idempotent=True)
def record_capital_goods_item(caller: dict, item_ref: text(200, min_length=1), kind: Literal["land_building", "computer", "ship_aircraft"],
bought_on: IsoDate, net_value: Money, input_vat: NonNegMoney, first_pct: Pct) -> dict:
"""Record a capital item once (keyed item_ref) with its input VAT, its adjustment intervals (10 for land and buildings of
250,000 or more, 5 for computers of 50,000 or more: dated statutory rows) and the percentage recovered in the first interval.
An item below the threshold is not in the scheme. Recording it again answers the same item."""
def unit(repo):
have = repo.capital_items.get(item_ref=item_ref)
if have is not None:
return _item_view(have) | {"replay": True}
when = uc._date(bought_on, "bought_on")
if kind == "ship_aircraft":
raise Refused(422, "ships and aircraft are not carried by this version: record the adjustments as a VAT adjustment")
key = "land_building" if kind == "land_building" else "computer"
minimum = schemes.stat(repo, f"capital_min_{key}", when)
if net_value < minimum:
raise unmet("capital_item_threshold", f"{net_value:,.2f} is below {minimum:,.2f}: a {kind.replace('_', ' ')} that cheap is not in the "
"capital goods scheme", "reclaim its VAT in the ordinary way")
intervals = int(schemes.stat(repo, f"capital_interval_{key}", when))
row = {"item_ref": item_ref, "kind": kind, "bought_on": when, "net_value": net_value, "input_vat": input_vat, "intervals": intervals,
"first_pct": first_pct, "adjustments_json": "[]", "recorded_by": uc._who(caller)}
try:
repo.conn.execute(repo.capital_items.table.insert().values(**row))
except IntegrityError:
raise Conflict("tax_capital_goods", {"item_ref": item_ref}) from None
return _item_view(row) | {"replay": False}
return uc._atomic(unit)
def _capital_run(repo, ye: date, pct: Decimal, replay: bool) -> list:
"""Each item still in its adjustment period gets the adjustment for THIS year's interval, once."""
out = []
mmdd = ye.strftime("%m-%d")
for item in repo.capital_items.where():
if item["disposed_on"] is not None and item["disposed_on"] <= ye:
continue
first_ye = schemes.year_end_for(mmdd, item["bought_on"])
n = ye.year - first_ye.year + 1
if not (2 <= n <= item["intervals"]):
continue
adj = json.loads(item["adjustments_json"] or "[]")
done = next((a for a in adj if a["interval"] == n), None)
if done is None and not replay:
amount = schemes.r2(Decimal(str(item["input_vat"])) / item["intervals"] * (pct - Decimal(str(item["first_pct"]))) / Decimal(100))
done = {"interval": n, "year_end": ye.isoformat(), "pct": str(pct), "amount": str(amount), "period_key": None}
adj.append(done)
repo.capital_items.put({**item, "adjustments_json": json.dumps(adj)})
if done is not None:
out.append({"item_ref": item["item_ref"], "interval": n, "pct": done["pct"], "amount": done["amount"]})
return out
@TAX.use_case(kind="command", route="POST /capital-goods/{item_ref}/dispose", permission=PE_WRITE, idempotent=True)
def record_capital_goods_disposal(caller: dict, item_ref: text(200, min_length=1), disposed_on: IsoDate,
disposal_pct: Optional[Pct] = None) -> dict:
"""A capital item is sold. Within its adjustment period this answers {gate: disposal_adjustment} until the disposal adjustment is
given as `disposal_pct`: 100 for a taxable sale, 0 for an exempt one. The remaining intervals are then adjusted in one go onto the
first return after the year of the sale."""
def unit(repo):
item = repo.capital_items.get(item_ref=item_ref)
if item is None:
raise Refused(404, f"no capital goods item {item_ref}")
if item["disposed_on"] is not None:
return _item_view(item) | {"replay": True}
when = uc._date(disposed_on, "disposed_on")
in_period = when < add_months(item["bought_on"], 12 * item["intervals"])
if in_period and disposal_pct is None:
raise unmet("disposal_adjustment", f"{item_ref} was sold on {when}, inside its {item['intervals']}-interval adjustment period "
f"(from {item['bought_on']})", "give the disposal adjustment: disposal_pct 100 for a taxable sale, 0 for an exempt one")
adj = json.loads(item["adjustments_json"] or "[]")
if in_period:
rows = [r for r in registration.live(repo.registrations.rows()) if r.partial_exemption_year_end]
mmdd = rows[-1].partial_exemption_year_end if rows else "03-31"
ye = schemes.year_end_for(mmdd, when)
first_ye = schemes.year_end_for(mmdd, item["bought_on"])
n = ye.year - first_ye.year + 1
remaining = max(0, item["intervals"] - n + 1)
amount = schemes.r2(Decimal(str(item["input_vat"])) / item["intervals"] * remaining * (disposal_pct - Decimal(str(item["first_pct"]))) / Decimal(100))
adj.append({"interval": f"{n}D", "year_end": ye.isoformat(), "pct": str(disposal_pct), "amount": str(amount), "period_key": None})
repo.capital_items.put({**item, "disposed_on": when, "adjustments_json": json.dumps(adj)})
return _item_view(repo.capital_items.get(item_ref=item_ref)) | {"replay": False}
return uc._atomic(unit)
@TAX.use_case(kind="query", route="GET /capital-goods/{item_ref}", permission=PE_WRITE)
def get_capital_goods_item(caller: dict, item_ref: text(200, min_length=1)) -> dict:
"""A capital item with its VAT, intervals, first-interval percentage and each interval adjusted so far."""
with uc.repository() as repo:
item = repo.capital_items.get(item_ref=item_ref)
if item is None:
raise Refused(404, f"no capital goods item {item_ref}")
return _item_view(item)