tegwick
f8d1c542d5
Implement RESOURCE-WP-0005: entity register, V0.1 terms parameters, entity association on inventory and planning records, transfer-price and credit-state arithmetic, monthly settlement, and entity views on the portfolio report. Live close emits nothing until delivered cost is known. Handoffs are FIN-WP-0006 and RAILIANCE-WP-0017.
349 lines
14 KiB
Python
349 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Delivered-cost transfer prices, credit state, and monthly settlement."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from calendar import monthrange
|
|
from datetime import date, timedelta
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
|
|
from entities import RAILIANCE, association_ok, load_register, load_terms, require_entity
|
|
from financial_exchange import money, money_text
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def transfer_price(delivered_cost: Decimal | None, consuming_entity: str, terms: dict) -> Decimal | None:
|
|
if delivered_cost is None:
|
|
return None
|
|
require_entity(consuming_entity)
|
|
rate = terms["railiance_self_markup_rate"] if consuming_entity == RAILIANCE else terms["markup_rate"]
|
|
return money(delivered_cost * (Decimal("1") + rate))
|
|
|
|
|
|
def max_new_delivered_cost(allowance: Decimal | None, consuming_entity: str, terms: dict) -> Decimal | None:
|
|
if allowance is None:
|
|
return None
|
|
if consuming_entity == RAILIANCE or allowance == Decimal("0.00"):
|
|
return money(allowance) if consuming_entity == RAILIANCE else money(0)
|
|
divisor = Decimal("1") + terms["markup_rate"]
|
|
return money(allowance / divisor)
|
|
|
|
|
|
def due_date(statement_date: date, terms: dict) -> date:
|
|
return statement_date + timedelta(days=terms["payment_term_days"])
|
|
|
|
|
|
def monthly_interest(overdue: Decimal, terms: dict) -> Decimal:
|
|
if overdue <= 0:
|
|
return money(0)
|
|
return money(overdue * terms["interest_rate_per_year"] / Decimal(12))
|
|
|
|
|
|
def apply_payment(interest_due: Decimal, principal: Decimal, payment: Decimal) -> dict:
|
|
"""Interest first, then principal (Terms OQ-7)."""
|
|
interest_due = money(interest_due)
|
|
principal = money(principal)
|
|
remaining = money(payment)
|
|
interest_paid = money(min(interest_due, remaining))
|
|
remaining = money(remaining - interest_paid)
|
|
principal_paid = money(min(principal, remaining))
|
|
remaining = money(remaining - principal_paid)
|
|
return {
|
|
"interest_paid": interest_paid,
|
|
"principal_paid": principal_paid,
|
|
"interest_remaining": money(interest_due - interest_paid),
|
|
"principal_remaining": money(principal - principal_paid),
|
|
"unapplied": remaining,
|
|
}
|
|
|
|
|
|
def credit_limit_for(entity_id: str, terms: dict, register: dict[str, dict]) -> Decimal | None:
|
|
require_entity(entity_id, register)
|
|
if entity_id == RAILIANCE:
|
|
return None
|
|
raw = register[entity_id].get("credit_limit_eur")
|
|
if raw is None:
|
|
return terms["default_credit_limit_eur"]
|
|
return money(raw)
|
|
|
|
|
|
def evaluate_credit(outstanding: Decimal, terms: dict, *, entity_id: str, register: dict[str, dict], overdue: bool) -> dict:
|
|
outstanding = money(outstanding)
|
|
limit = credit_limit_for(entity_id, terms, register)
|
|
interest = monthly_interest(outstanding, terms) if overdue else money(0)
|
|
if limit is None:
|
|
return {
|
|
"consumption_mode": "open",
|
|
"credit_limit_eur": None,
|
|
"credit_headroom_eur": None,
|
|
"interest_this_month_eur": money_text(interest),
|
|
"new_transfer_charges_allowed_eur": None,
|
|
"max_new_delivered_cost_eur": None,
|
|
}
|
|
restricted = outstanding >= limit
|
|
if restricted:
|
|
allowance = money(max(Decimal("0.00"), terms["restricted_monthly_consumption_eur"] - interest))
|
|
else:
|
|
allowance = None
|
|
headroom = money(max(Decimal("0.00"), limit - outstanding))
|
|
return {
|
|
"consumption_mode": "restricted" if restricted else "open",
|
|
"credit_limit_eur": money_text(limit),
|
|
"credit_headroom_eur": money_text(headroom),
|
|
"interest_this_month_eur": money_text(interest),
|
|
"new_transfer_charges_allowed_eur": None if allowance is None else money_text(allowance),
|
|
"max_new_delivered_cost_eur": (
|
|
None if allowance is None else money_text(max_new_delivered_cost(allowance, entity_id, terms))
|
|
),
|
|
}
|
|
|
|
|
|
def statement_date_for(period: str) -> date:
|
|
year, month = (int(part) for part in period.split("-"))
|
|
if month == 12:
|
|
return date(year + 1, 1, 1)
|
|
return date(year, month + 1, 1)
|
|
|
|
|
|
def _line_item(charge: dict, terms: dict) -> dict:
|
|
entity_id = charge["financial_entity_id"]
|
|
delivered = None if charge.get("delivered_cost") is None else money(charge["delivered_cost"])
|
|
price = transfer_price(delivered, entity_id, terms)
|
|
markup = None
|
|
if delivered is not None and price is not None:
|
|
markup = money(price - delivered)
|
|
components = charge.get("components") or {}
|
|
return {
|
|
"resource_id": charge["resource_id"],
|
|
"usage_summary": charge.get("usage_summary") or "",
|
|
"delivered_cost_eur": None if delivered is None else money_text(delivered),
|
|
"components": {
|
|
"infrastructure": components.get("infrastructure"),
|
|
"internal_labor": components.get("internal_labor"),
|
|
"external_services": components.get("external_services"),
|
|
"setup": components.get("setup"),
|
|
"other": components.get("other"),
|
|
},
|
|
"markup_eur": None if markup is None else money_text(markup),
|
|
"transfer_price_eur": None if price is None else money_text(price),
|
|
"unknown_remainder": charge.get("unknown_remainder"),
|
|
}
|
|
|
|
|
|
def build_statement(
|
|
*,
|
|
entity_id: str,
|
|
period: str,
|
|
terms: dict,
|
|
register: dict[str, dict],
|
|
statement_date: date,
|
|
lines: list[dict],
|
|
prior_outstanding: Decimal,
|
|
recognized_payments: list[dict],
|
|
as_of: date | None = None,
|
|
payment_status: str = "unknown",
|
|
) -> dict | None:
|
|
require_entity(entity_id, register)
|
|
as_of = as_of or statement_date
|
|
due = due_date(statement_date, terms)
|
|
prior = money(prior_outstanding)
|
|
paid = money(0)
|
|
for payment in recognized_payments:
|
|
if payment.get("financial_entity_id") == entity_id:
|
|
paid = money(paid + money(payment["amount"]))
|
|
payment_status = "recognized"
|
|
known_cost = money(0)
|
|
known_transfer = money(0)
|
|
unknown = False
|
|
for line in lines:
|
|
if line["delivered_cost_eur"] is None or line["transfer_price_eur"] is None:
|
|
unknown = True
|
|
else:
|
|
known_cost = money(known_cost + money(line["delivered_cost_eur"]))
|
|
known_transfer = money(known_transfer + money(line["transfer_price_eur"]))
|
|
prior_due = due_date(statement_date_for(_previous_period(period)), terms) if prior > 0 else due
|
|
# Interest is quantified when prior outstanding is past its due date at as_of.
|
|
overdue = prior > 0 and as_of > prior_due
|
|
if prior > 0 and as_of == statement_date and statement_date <= due:
|
|
# Brand-new statement of this period: prior from last period is overdue
|
|
# only if its own due date has passed.
|
|
overdue = as_of > prior_due
|
|
interest = monthly_interest(prior, terms) if overdue else money(0)
|
|
applied = apply_payment(interest, prior, paid)
|
|
balance_before_interest = money(applied["principal_remaining"] + known_transfer)
|
|
outstanding = money(balance_before_interest + applied["interest_remaining"])
|
|
credit = evaluate_credit(
|
|
balance_before_interest, terms, entity_id=entity_id, register=register, overdue=overdue,
|
|
)
|
|
has_activity = bool(lines) or prior > 0 or paid > 0 or interest > 0
|
|
if not has_activity:
|
|
return None
|
|
entity = register[entity_id]
|
|
provider = register[RAILIANCE]
|
|
payment_instruction = None
|
|
if entity_id != RAILIANCE and known_transfer > 0:
|
|
payment_instruction = {
|
|
"pay_from_account_ref": entity.get("account_ref"),
|
|
"pay_to_account_ref": provider.get("account_ref"),
|
|
}
|
|
return {
|
|
"schema_version": "0.1",
|
|
"record_type": "settlement_statement",
|
|
"terms_version": terms["terms_version"],
|
|
"financial_entity_id": entity_id,
|
|
"procuring_entity_id": RAILIANCE,
|
|
"period": period,
|
|
"statement_date": statement_date.isoformat(),
|
|
"due_date": due.isoformat(),
|
|
"currency": terms["currency"],
|
|
"line_items": lines,
|
|
"known_delivered_cost_eur": money_text(known_cost),
|
|
"known_transfer_price_eur": money_text(known_transfer),
|
|
"unknown_cost_remainder": unknown,
|
|
"prior_outstanding_eur": money_text(prior),
|
|
"recognized_payments_eur": None if payment_status == "unknown" and paid == 0 else money_text(paid),
|
|
"payment_recognition": payment_status if paid > 0 else "unknown",
|
|
"interest_eur": money_text(interest),
|
|
"new_transfer_charges_eur": money_text(known_transfer),
|
|
"outstanding_eur": money_text(outstanding),
|
|
"credit_limit_eur": credit["credit_limit_eur"],
|
|
"credit_headroom_eur": credit["credit_headroom_eur"],
|
|
"consumption_mode": credit["consumption_mode"],
|
|
"next_month_allowance_eur": credit["new_transfer_charges_allowed_eur"],
|
|
"payment_instruction": payment_instruction,
|
|
}
|
|
|
|
|
|
def _previous_period(period: str) -> str:
|
|
year, month = (int(part) for part in period.split("-"))
|
|
if month == 1:
|
|
return f"{year - 1}-12"
|
|
return f"{year}-{month - 1:02d}"
|
|
|
|
|
|
def close_fixture(payload: dict, terms: dict, register: dict[str, dict]) -> list[dict]:
|
|
statement_date = date.fromisoformat(payload["statement_date"])
|
|
as_of = date.fromisoformat(payload["as_of"]) if payload.get("as_of") else statement_date
|
|
by_entity: dict[str, list[dict]] = {}
|
|
for charge in payload.get("charges", []):
|
|
association_ok({
|
|
"financial_entity_id": charge["financial_entity_id"],
|
|
"procuring_entity_id": charge.get("procuring_entity_id", RAILIANCE),
|
|
"entity_gap": None,
|
|
})
|
|
by_entity.setdefault(charge["financial_entity_id"], []).append(_line_item(charge, terms))
|
|
prior = {key: money(value) for key, value in (payload.get("prior_outstanding") or {}).items()}
|
|
entities = sorted(set(by_entity) | set(prior) | {
|
|
payment["financial_entity_id"] for payment in payload.get("recognized_payments") or []
|
|
})
|
|
statements = []
|
|
for entity_id in entities:
|
|
statement = build_statement(
|
|
entity_id=entity_id,
|
|
period=payload["period"],
|
|
terms=terms,
|
|
register=register,
|
|
statement_date=statement_date,
|
|
lines=by_entity.get(entity_id, []),
|
|
prior_outstanding=prior.get(entity_id, money(0)),
|
|
recognized_payments=payload.get("recognized_payments") or [],
|
|
as_of=as_of,
|
|
payment_status=payload.get("payment_recognition", "unknown"),
|
|
)
|
|
if statement:
|
|
statements.append(statement)
|
|
return statements
|
|
|
|
|
|
def collect_live_charges(root: Path, period: str) -> list[dict]:
|
|
"""Live close only emits a charge when delivered cost is known. Unknown is not zero."""
|
|
charges = []
|
|
year, month = (int(part) for part in period.split("-"))
|
|
period_end = date(year, month, monthrange(year, month)[1])
|
|
for path in sorted((root / "data" / "resources").glob("*.json")):
|
|
resource = json.loads(path.read_text())
|
|
if resource.get("status") in {"rejected", "retired"}:
|
|
continue
|
|
commissioned = resource.get("lifecycle", {}).get("commissioned_on")
|
|
if commissioned and date.fromisoformat(commissioned) > period_end:
|
|
continue
|
|
proposed = resource.get("lifecycle", {}).get("proposed_on")
|
|
if resource.get("status") == "proposed" and proposed and date.fromisoformat(proposed) > period_end:
|
|
continue
|
|
# No booked delivered cost lives on inventory records today.
|
|
# A control-cycle actual with known infrastructure would be the source;
|
|
# until then there are no live charges.
|
|
del path
|
|
control_dir = root / "data" / "control-cycle"
|
|
if control_dir.exists():
|
|
for path in sorted(control_dir.glob("*.json")):
|
|
record = json.loads(path.read_text())
|
|
if record.get("period") != period:
|
|
continue
|
|
costs = record.get("costs") or {}
|
|
parts = [costs.get("infrastructure"), costs.get("internal_labor"), costs.get("external_labor")]
|
|
if any(part is None for part in parts):
|
|
continue
|
|
entity_id = record.get("financial_entity_id")
|
|
if not entity_id:
|
|
continue
|
|
charges.append({
|
|
"financial_entity_id": entity_id,
|
|
"procuring_entity_id": record.get("procuring_entity_id", RAILIANCE),
|
|
"resource_id": record["resource_id"],
|
|
"delivered_cost": money_text(sum(parts)),
|
|
"usage_summary": f"control-cycle {record['record_type']} {record['record_id']}",
|
|
"components": {
|
|
"infrastructure": None if costs.get("infrastructure") is None else money_text(costs["infrastructure"]),
|
|
"internal_labor": None if costs.get("internal_labor") is None else money_text(costs["internal_labor"]),
|
|
"external_services": None if costs.get("external_labor") is None else money_text(costs["external_labor"]),
|
|
"setup": None,
|
|
"other": None,
|
|
},
|
|
})
|
|
return charges
|
|
|
|
|
|
def close_live(root: Path, period: str) -> list[dict]:
|
|
terms = load_terms(root)
|
|
_, register = load_register(root)
|
|
payload = {
|
|
"period": period,
|
|
"statement_date": statement_date_for(period).isoformat(),
|
|
"charges": collect_live_charges(root, period),
|
|
"prior_outstanding": {},
|
|
"recognized_payments": [],
|
|
"payment_recognition": "unknown",
|
|
}
|
|
return close_fixture(payload, terms, register)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Monthly settlement close")
|
|
parser.add_argument("command", choices=["close"])
|
|
parser.add_argument("--period", help="YYYY-MM")
|
|
parser.add_argument("--fixture", help="path to a settlement fixture")
|
|
parser.add_argument("--root", default=str(ROOT))
|
|
args = parser.parse_args()
|
|
root = Path(args.root)
|
|
terms = load_terms(root)
|
|
_, register = load_register(root)
|
|
if args.fixture:
|
|
payload = json.loads(Path(args.fixture).read_text())
|
|
statements = close_fixture(payload, terms, register)
|
|
else:
|
|
if not args.period:
|
|
print("PERIOD=YYYY-MM is required without --fixture", file=sys.stderr)
|
|
return 2
|
|
statements = close_live(root, args.period)
|
|
print(json.dumps(statements, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|