dodil/erp-brick-fixed-assets
DODIL ERP brick: fixed assets
src/brick_fixed_assets/application/uc_allowances.py
"""The UK capital allowances (FA-21, FA-22): the TAX AREA. Pools (main, special rate, single asset, structures and buildings),
the annual investment allowance, full expensing and first-year allowances, writing-down allowances, balancing allowances and
charges — worked out from the register's tax-area transactions, per company and chargeable period. The tax area NEVER
posts a journal. The rates and the AIA limit are dated owner-controlled rows (fa_ca_rate), not code.
A computation is prepared by the tax preparer, approved by somebody else (email AND sign-in) and then FILED with the return's
reference, which freezes it: a transaction dated inside a filed period lands in the NEXT period as an adjustment, named. The
closing pools of one period are the opening pools of the next (derived)."""
import json
from calendar import monthrange
from datetime import date, timedelta
from decimal import Decimal
from typing import Optional
from erp_platform.bricks import Refused
from brick_fixed_assets.application import services
from brick_fixed_assets.application import use_cases as hub
from brick_fixed_assets.application.inputs import ClaimIn, Code, IsoDate, OpeningPoolIn, Reason, day
from brick_fixed_assets.domain import capital_allowances as ca
from brick_fixed_assets.domain.errors import RuleBroken
from brick_fixed_assets.domain.values import ZERO, same_person
from brick_fixed_assets.infrastructure import tables as T
FA = hub.FA
def _months(start: date, end: date) -> int:
if start.day != 1 or end.day != monthrange(end.year, end.month)[1] or end < start:
raise RuleBroken("a chargeable period runs from the first of a month to the end of a month")
m = (end.year - start.year) * 12 + end.month - start.month + 1
if m > 18:
raise RuleBroken("a chargeable period is at most 18 months")
return m
def _load(repo, company: str, start: date, end: date) -> dict:
got = repo.load(T.CA, company_code=company, period_start=start, period_end=end)
if got is None:
raise Refused(404, f"no capital allowances computation for {company} {start} to {end}")
root = got["fa_ca_computation"][0]
return {"root": root, "claims": got["fa_ca_claim"], "version": root["version"]}
def _save(repo, comp: dict) -> int:
return T.CA.save(repo.conn, {"fa_ca_computation": [comp["root"]], "fa_ca_claim": comp["claims"]},
expected_version=comp["version"],
snapshot=repo.spend(T.CA, comp["root"]["company_code"], comp["root"]["period_start"], comp["root"]["period_end"]))
def _comps(repo, company: str) -> list:
from sqlalchemy import select
rows = repo.conn.execute(select(T.fa_ca_computation).where(T.fa_ca_computation.c.company_code == company)
.order_by(T.fa_ca_computation.c.period_start))
return [dict(r._mapping) for r in rows]
def _tax_areas(repo, today: date) -> list:
return [a["area_code"] for a in repo.areas(today) if a["purpose"] == "tax"]
def _included_elsewhere(repo, company: str, except_start: Optional[date]) -> set:
out = set()
for c in _comps(repo, company):
if c["status"] == "filed" and c["period_start"] != except_start:
out |= set(json.loads(c["result"] or "{}").get("included", []))
return out
def _effective(value_date: date, filed: list) -> tuple:
"""(date the transaction counts from, the filed period it was moved out of): inside a filed period it lands in the
next period as an adjustment."""
eff, moved = value_date, None
for c in sorted(filed, key=lambda x: x["period_start"]):
if c["period_start"] <= eff <= c["period_end"]:
moved = f"{c['period_start']} to {c['period_end']}"
eff = c["period_end"] + timedelta(days=1)
return eff, moved
def _derive(repo, company: str, start: date, end: date) -> tuple:
"""(additions {asset_id: cost}, disposals [(asset_id, proceeds, cost)], included tx ids, adjustments) from the TAX area
of the company's posted transactions that fall in the period (adjustments from a filed period included)."""
today = repo.today()
codes = _tax_areas(repo, today)
if not codes:
raise RuleBroken("there is no tax area: define the area with purpose 'tax' (Settings: areas)")
comps = _comps(repo, company)
filed = [c for c in comps if c["status"] == "filed" and c["period_start"] != start]
already = _included_elsewhere(repo, company, start)
additions: dict = {}
disposals, included, adjustments = [], [], []
book_area = next((a["area_code"] for a in repo.posting_areas(today)), None)
for row in repo.area_values_rows(company_code=company, area_code=codes[0], statuses=("posted",)):
if row["transaction_id"] in already or row["kind"] == "reversal":
continue
eff, moved = _effective(row["value_date"], filed)
if not (start <= eff <= end):
continue
included.append(row["transaction_id"])
if moved:
adjustments.append({"transaction_id": row["transaction_id"], "from_period": moved})
cost = row["cost"] or ZERO
if row["kind"] in ("acquisition", "grant", "settlement", "takeover") or (row["kind"] == "transfer" and cost > 0):
additions[row["asset_id"]] = additions.get(row["asset_id"], ZERO) + cost
elif row["kind"] == "retirement" or (row["kind"] == "transfer" and cost < 0):
proceeds = row["proceeds"] or ZERO
if row["kind"] == "transfer":
nbv = ZERO
if book_area:
b = next((x for x in repo.area_values_rows(row["asset_id"], book_area, statuses=("posted",))
if x["transaction_id"] == row["transaction_id"]), None)
if b:
nbv = abs(b["cost"] or ZERO) - abs(b["accumulated_depreciation"] or ZERO)
proceeds = nbv
disposals.append((row["asset_id"], proceeds, abs(cost)))
return {k: v for k, v in additions.items() if v != 0}, disposals, included, adjustments
def _allocation(repo, company: str, today: date, rates: dict) -> Decimal:
limit = rates.get("aia_limit", ZERO)
spec = repo.setting("aia_group_allocation", today)
if spec and "=" in spec:
parts = dict(x.split("=", 1) for x in spec.split(",") if "=" in x)
return Decimal(parts.get(company, "0"))
first = sorted({r["company_code"] for r in repo.asset_roots()})[:1]
return limit if (not first or first[0] == company) else ZERO
def _view(repo, comp: dict) -> dict:
r = comp["root"]
result = json.loads(r["result"] or "{}")
return {"company_code": r["company_code"], "period_start": r["period_start"].isoformat(),
"period_end": r["period_end"].isoformat(), "status": r["status"], "months": r["months"],
"aia_limit_allocated": str(r["aia_limit_allocated"]), "opening": json.loads(r["opening"] or "{}"),
"claims": [{"asset_id": c["asset_id"], "claim": c["claim"], "pool": c["pool"], "amount": str(c["amount"]),
"disposal_value": str(c["disposal_value"]) if c["disposal_value"] is not None else None}
for c in sorted(comp["claims"], key=lambda c: c["asset_id"])],
"additions": result.get("additions", {}), "disposals": result.get("disposals", []),
"adjustments": result.get("adjustments", []), "pools": result.get("pools", {}),
"totals": result.get("totals", {}), "prepared_by_email": r["prepared_by_email"],
"approved_by_email": r["approved_by_email"], "return_reference": r["return_reference"],
"returned_reason": r["returned_reason"], "version": comp["version"]}
@FA.use_case(kind="command", route="POST /ca-computations", permission=hub.CA_WRITE, idempotent=True)
def draft_ca_computation(caller: dict, company_code: Code, period_start: IsoDate, period_end: IsoDate,
opening_pools: Optional[list[OpeningPoolIn]] = None) -> dict:
"""Open the capital allowances computation of a company for a chargeable period (FA-21). The pools open from the previous
period's closing pools when it was FILED; the first period needs the opening pools entered (main, special_rate, sba,
single_asset:<asset>). The additions and disposals are DERIVED from the tax area's transactions in the period, a
transaction dated inside a filed period arriving as a named adjustment. The AIA limit is the company's share
(owner's allocation), pro-rated by months/12."""
with hub.repository() as repo, hub.decide():
start, end = day(period_start), day(period_end)
months = _months(start, end)
today = hub.today(repo, True)
existing = [c for c in _comps(repo, company_code) if c["period_start"] == start and c["period_end"] == end]
if existing:
return _view(repo, _load(repo, company_code, start, end))
prev = [c for c in _comps(repo, company_code) if c["period_end"] + timedelta(days=1) == start]
if prev:
if prev[-1]["status"] != "filed":
raise RuleBroken("the previous period is not filed (previous_filed): file it first, or enter opening pools")
opening = {k: Decimal(v["closing"]) for k, v in json.loads(prev[-1]["result"])["pools"].items()
if Decimal(v["closing"]) != 0}
elif opening_pools:
opening = {p.pool: p.balance for p in opening_pools}
else:
raise RuleBroken("enter the opening pools (previous_filed): there is no earlier computation for this company")
rates = repo.ca_rates(start)
if "aia_limit" not in rates:
raise RuleBroken("the owner has not set the capital allowance rates in force (Settings: capital allowance rates)")
additions, disposals, included, adjustments = _derive(repo, company_code, start, end)
result = {"additions": {k: str(v) for k, v in sorted(additions.items())},
"disposals": [{"asset_id": a, "proceeds": str(p), "cost": str(c)} for a, p, c in disposals],
"included": included, "adjustments": adjustments}
root = {"company_code": company_code, "period_start": start, "period_end": end, "status": "draft", "months": months,
"aia_limit_allocated": _allocation(repo, company_code, today, rates),
"opening": json.dumps({k: str(v) for k, v in opening.items()}), "result": json.dumps(result),
"prepared_by_email": None, "prepared_by_sub": None, "approved_by_email": None, "approved_by_sub": None,
"return_reference": None, "returned_reason": None}
comp = {"root": root, "claims": [], "version": 0}
comp["version"] = _save(repo, comp)
return _view(repo, comp)
def _addition(repo, asset_id: str, cost: Decimal, claim: str, amount: Optional[Decimal]) -> ca.Addition:
a = repo.asset(asset_id)
if a is None:
raise RuleBroken(f"no asset {asset_id}")
pool = a.root["ca_pool"] or "main"
if claim == "single_asset":
pool = "single_asset"
elif claim == "sba":
pool = "sba"
elif pool == "none":
pool = "main"
taken = amount if amount is not None else (cost if claim == "aia" else ZERO)
return ca.Addition(asset_id, cost, claim, pool, taken, bool(a.root["is_car"]), bool(a.root["new_and_unused"]))
@FA.use_case(kind="command", route="POST /ca-computations/{company_code}/{period_start}/{period_end}/claims",
permission=hub.CA_WRITE)
def choose_allowance(caller: dict, company_code: Code, period_start: IsoDate, period_end: IsoDate,
claims: list[ClaimIn]) -> dict:
"""Choose, per addition, which allowance is claimed: aia (the amount, default the whole cost), full_expensing (new and
unused main-pool plant only), fya_50, pool, single_asset, sba or none. A claim that does not fit the asset is refused
naming the rule (claim_allowed): a car cannot take AIA. Only while the computation is a draft."""
with hub.repository() as repo, hub.decide():
start, end = day(period_start), day(period_end)
comp = _load(repo, company_code, start, end)
if comp["root"]["status"] != "draft":
raise RuleBroken(f"the computation is {comp['root']['status']}: send it back to change a claim")
result = json.loads(comp["root"]["result"] or "{}")
for c in claims:
cost = Decimal(result.get("additions", {}).get(c.asset_id, "0"))
if cost <= 0:
raise RuleBroken(f"{c.asset_id} has no addition in this period to claim on")
add = _addition(repo, c.asset_id, cost, c.claim, c.amount)
ca.check_claim(add, repo.ca_rates(start))
rows = [x for x in comp["claims"] if x["asset_id"] != c.asset_id]
rows.append({"company_code": company_code, "period_start": start, "period_end": end, "asset_id": c.asset_id,
"claim": c.claim, "pool": add.pool, "amount": add.amount, "disposal_value": None})
comp["claims"] = rows
comp["version"] = _save(repo, comp)
return _view(repo, comp)
@FA.use_case(kind="command", route="POST /ca-computations/{company_code}/{period_start}/{period_end}/prepare",
permission=hub.CA_WRITE, idempotent=True)
def prepare_ca_computation(caller: dict, company_code: Code, period_start: IsoDate, period_end: IsoDate,
ends_pools: Optional[list[Code]] = None) -> dict:
"""Prepare the computation: every addition needs a claim (claims_complete), AIA within the company's share of the limit for
the period (aia_within_limit), then writing-down allowances (pool x rate x months/12), balancing allowances and charges and
the closing pools are computed and kept. A disposal value is the lower of proceeds and cost; a pool at or under the small
pools limit may be written off whole; `ends_pools` names pools that end (a balancing allowance or charge)."""
with hub.repository() as repo, hub.decide():
start, end = day(period_start), day(period_end)
comp = _load(repo, company_code, start, end)
r = comp["root"]
if r["status"] not in ("draft",):
if r["status"] == "prepared":
return _view(repo, comp)
raise RuleBroken(f"the computation is {r['status']}: send it back to prepare it again")
result = json.loads(r["result"] or "{}")
claimed = {c["asset_id"] for c in comp["claims"]}
missing = [a for a in result.get("additions", {}) if a not in claimed]
if missing:
raise RuleBroken("every addition needs a claim (claims_complete): " + ", ".join(sorted(missing)))
rates = repo.ca_rates(start)
by_claim = {c["asset_id"]: c for c in comp["claims"]}
additions = []
for asset_id, cost in sorted(result.get("additions", {}).items()):
c = by_claim[asset_id]
additions.append(_addition(repo, asset_id, Decimal(cost), c["claim"], c["amount"]))
claimed_aia, allowed = ca.aia_within_limit(additions, r["aia_limit_allocated"], r["months"])
if claimed_aia > allowed:
raise RuleBroken(f"AIA claimed {claimed_aia} is above the company's share of the limit for this period "
f"({allowed}) (aia_within_limit): move the excess to the pool, or change the group allocation")
prior = {}
for c in _comps(repo, company_code):
for cl in T_claims(repo, company_code, c):
prior[cl["asset_id"]] = cl
disposals = []
sold = repo.assets(d["asset_id"] for d in result.get("disposals", []))
for d in result.get("disposals", []):
was = prior.get(d["asset_id"])
a = sold.get(d["asset_id"])
pool = (was["pool"] if was else (a.root["ca_pool"] if a else "main")) or "main"
disposals.append(ca.Disposal(d["asset_id"], "main" if pool == "none" else pool, Decimal(d["proceeds"]),
Decimal(d["cost"]), was["claim"] if was else "pool"))
opening = {k: Decimal(v) for k, v in json.loads(r["opening"] or "{}").items()}
out = ca.compute(r["months"], opening, additions, disposals, rates, tuple(ends_pools or ()))
for d, row in zip(disposals, result.get("disposals", [])):
row["disposal_value"] = str(d.value)
result["pools"] = {k: {kk: str(vv) for kk, vv in v.items()} for k, v in sorted(out["pools"].items())}
result["totals"] = {k: str(v) for k, v in out.items() if k not in ("pools", "months")}
result["rates"] = {k: str(v) for k, v in rates.items()}
r["result"] = json.dumps(result)
r["status"] = "prepared"
r["prepared_by_email"], r["prepared_by_sub"] = services.email_of(caller), services.sub_of(caller)
comp["version"] = _save(repo, comp)
return _view(repo, comp)
def T_claims(repo, company: str, c: dict) -> list:
from sqlalchemy import and_, select
rows = repo.conn.execute(select(T.fa_ca_claim).where(and_(T.fa_ca_claim.c.company_code == company,
T.fa_ca_claim.c.period_start == c["period_start"],
T.fa_ca_claim.c.period_end == c["period_end"])))
return [dict(x._mapping) for x in rows]
@FA.use_case(kind="command", route="POST /ca-computations/{company_code}/{period_start}/{period_end}/approve",
permission=hub.CA_APPROVE, idempotent=True)
def approve_ca_computation(caller: dict, company_code: Code, period_start: IsoDate, period_end: IsoDate) -> dict:
"""Approve a prepared computation before it goes into the tax return (FA-22). The preparer cannot approve their own: the same
email under another sign-in, or the same sign-in under another email, is the same person (second_person). Records who
and when."""
with hub.repository() as repo, hub.decide():
comp = _load(repo, company_code, day(period_start), day(period_end))
r = comp["root"]
if r["status"] in ("approved", "filed"):
return _view(repo, comp)
if r["status"] != "prepared":
raise RuleBroken(f"the computation is {r['status']}: only a prepared one is approved")
if same_person(services.email_of(caller), services.sub_of(caller), r["prepared_by_email"], r["prepared_by_sub"]):
raise RuleBroken("you prepared this computation (second_person: same email or same sign-in): another approver "
"approves it")
r["status"] = "approved"
r["approved_by_email"], r["approved_by_sub"] = services.email_of(caller), services.sub_of(caller)
comp["version"] = _save(repo, comp)
return _view(repo, comp)
@FA.use_case(kind="command", route="POST /ca-computations/{company_code}/{period_start}/{period_end}/return",
permission=hub.CA_APPROVE)
def return_ca_computation(caller: dict, company_code: Code, period_start: IsoDate, period_end: IsoDate, reason: Reason) -> dict:
"""Send a prepared or approved computation back to draft with a reason. A filed one never moves."""
with hub.repository() as repo, hub.decide():
comp = _load(repo, company_code, day(period_start), day(period_end))
r = comp["root"]
if r["status"] not in ("prepared", "approved"):
raise RuleBroken(f"the computation is {r['status']}: only a prepared or approved one is sent back")
r["status"] = "draft"
r["returned_reason"] = reason
r["approved_by_email"] = r["approved_by_sub"] = None
comp["version"] = _save(repo, comp)
return _view(repo, comp)
@FA.use_case(kind="command", route="POST /ca-computations/{company_code}/{period_start}/{period_end}/file",
permission=hub.CA_APPROVE, idempotent=True)
def file_ca_computation(caller: dict, company_code: Code, period_start: IsoDate, period_end: IsoDate,
return_reference: Code) -> dict:
"""Mark an approved computation FILED with the return's reference. Filed freezes it: an asset transaction dated in that
chargeable period lands in the next period as a named adjustment, and its closing pools open the next period."""
with hub.repository() as repo, hub.decide():
comp = _load(repo, company_code, day(period_start), day(period_end))
r = comp["root"]
if r["status"] == "filed":
return _view(repo, comp)
if r["status"] != "approved":
raise RuleBroken(f"the computation is {r['status']}: only an approved one is filed")
r["status"] = "filed"
r["return_reference"] = return_reference
comp["version"] = _save(repo, comp)
return _view(repo, comp)
@FA.use_case(kind="query", route="GET /ca-computations/{company_code}/{period_start}/{period_end}", permission=hub.CA_READ)
def show_ca_computation(caller: dict, company_code: Code, period_start: IsoDate, period_end: IsoDate) -> dict:
"""A computation: opening pools, additions with the claim chosen, disposals, allowances, balancing charges and the closing
pools, its state, who prepared and approved it, and the return reference once filed."""
with hub.repository() as repo:
return _view(repo, _load(repo, company_code, day(period_start), day(period_end)))
@FA.use_case(kind="query", route="GET /ca-computations", permission=hub.CA_READ)
def list_ca_computations(caller: dict, company_code: Optional[Code] = None) -> dict:
"""The capital allowances computations, oldest first, with their state and totals."""
with hub.repository() as repo:
from sqlalchemy import select
q = select(T.fa_ca_computation).order_by(T.fa_ca_computation.c.company_code, T.fa_ca_computation.c.period_start)
if company_code:
q = q.where(T.fa_ca_computation.c.company_code == company_code)
rows = [dict(r._mapping) for r in repo.conn.execute(q)]
return {"computations": [{"company_code": r["company_code"], "period_start": r["period_start"].isoformat(),
"period_end": r["period_end"].isoformat(), "status": r["status"],
"totals": json.loads(r["result"] or "{}").get("totals", {})} for r in rows]}
@FA.use_case(kind="query", route="GET /ca-computations/book-tax-difference", permission=hub.CA_READ)
def show_book_tax_difference(caller: dict, company_code: Code) -> dict:
"""The difference between book net book value and the tax written-down value (the closing pools of the latest approved or
filed computation) per company: the figure the deferred tax calculation starts from (FRS 102 section 29, IAS 12). Deferred
tax is shown, not posted: the accountant posts the journal."""
with hub.repository() as repo:
today = hub.today(repo)
book = next((a["area_code"] for a in repo.posting_areas(today)), None)
nbv = ZERO
if book:
for root in repo.asset_roots(company_code):
nbv += services.area_values(repo, root["asset_id"], book).nbv
latest = next((c for c in reversed(_comps(repo, company_code)) if c["status"] in ("approved", "filed")), None)
wdv = ZERO
if latest:
wdv = sum((Decimal(v["closing"]) for v in json.loads(latest["result"])["pools"].values()), ZERO)
return {"company_code": company_code, "book_area": book, "book_nbv": str(nbv),
"tax_written_down_value": str(wdv), "difference": str(nbv - wdv),
"computation": f"{latest['period_start']} to {latest['period_end']}" if latest else None}