127 lines
7.1 KiB
Python
127 lines
7.1 KiB
Python
"""Private, deterministic aggregation for comprehensive settlement previews."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import date
|
|
|
|
from .shared import CASHFLOW_KIND_SQL
|
|
|
|
_AGGREGATION_VERSION = "comprehensive-v1"
|
|
_SNAPSHOT_SCHEMA_VERSION = 1
|
|
_ALLOWED_KINDS = {"income", "expense", "excluded"}
|
|
|
|
|
|
def _normalize_cashflow_policy(payload: dict) -> dict:
|
|
if not isinstance(payload, dict) or set(payload) - {"savings_categories", "category_decisions"}:
|
|
raise ValueError("cashflow policy must contain only supported keys")
|
|
savings = payload.get("savings_categories", [])
|
|
decisions = payload.get("category_decisions", [])
|
|
if not isinstance(savings, list) or not isinstance(decisions, list):
|
|
raise ValueError("policy categories and decisions must be lists")
|
|
def pair(row, required):
|
|
if not isinstance(row, dict) or set(row) != required:
|
|
raise ValueError("invalid category policy entry")
|
|
if any(not isinstance(row[k], str) for k in ("category", "subcategory")):
|
|
raise ValueError("category fields must be strings")
|
|
return row["category"], row["subcategory"]
|
|
normalized_savings = [pair(x, {"category", "subcategory"}) for x in savings]
|
|
if len(set(normalized_savings)) != len(normalized_savings):
|
|
raise ValueError("duplicate savings category")
|
|
normalized_decisions = []
|
|
for row in decisions:
|
|
cat, sub = pair(row, {"category", "subcategory", "cashflow_kind", "reason"})
|
|
kind, reason = row["cashflow_kind"], row["reason"]
|
|
if not isinstance(kind, str) or kind not in _ALLOWED_KINDS:
|
|
raise ValueError("invalid cashflow decision kind")
|
|
if not isinstance(reason, str) or not reason.strip():
|
|
raise ValueError("decision reason must be nonempty")
|
|
normalized_decisions.append((cat, sub, kind, reason))
|
|
keys = [(x[0], x[1]) for x in normalized_decisions]
|
|
if len(set(keys)) != len(keys):
|
|
raise ValueError("duplicate category decision")
|
|
return {"savings_categories": [{"category": c, "subcategory": s} for c, s in sorted(normalized_savings)],
|
|
"category_decisions": [{"category": c, "subcategory": s, "cashflow_kind": k, "reason": r}
|
|
for c, s, k, r in sorted(normalized_decisions)]}
|
|
|
|
|
|
def _review_cashflow(conn, month: str, policy: dict) -> dict:
|
|
policy = _normalize_cashflow_policy(policy)
|
|
start = date.fromisoformat(month + "-01")
|
|
end = date(start.year + (start.month == 12), 1 if start.month == 12 else start.month + 1, 1)
|
|
rows = conn.execute(f"""SELECT t.id,t.transaction_date,t.type,t.amount,t.currency,t.category,t.subcategory,
|
|
{CASHFLOW_KIND_SQL} AS originalcashflow_kind, lc.cashflow_kind AS category_cashflow_kind,
|
|
lc.active AS category_active
|
|
FROM transactions t LEFT JOIN ledger_categories lc
|
|
ON lc.category=t.category AND lc.subcategory=t.subcategory AND lc.active=1
|
|
WHERE t.status='active' AND t.transaction_date>=? AND t.transaction_date<?
|
|
ORDER BY t.transaction_date,t.id""", (start.isoformat(), end.isoformat())).fetchall()
|
|
savings = {(x["category"], x["subcategory"]) for x in policy["savings_categories"]}
|
|
decisions = {(x["category"], x["subcategory"]): x for x in policy["category_decisions"]}
|
|
grouped, source, blockers = {}, [], []
|
|
income = expense = 0
|
|
for row in rows:
|
|
t = dict(row)
|
|
original = t["originalcashflow_kind"]
|
|
pair = (t["category"], t["subcategory"])
|
|
decision = decisions.get(pair)
|
|
typ, catkind = t["type"], original
|
|
amount = t["amount"]
|
|
raw_valid = isinstance(amount, int) and not isinstance(amount, bool)
|
|
reason = None
|
|
# Explicit movement classes are immutable exclusions; investment is only selected by pair.
|
|
general_transfer = catkind == "transfer" or (typ == "transfer" and catkind != "investment")
|
|
initial_adjustment = typ == "adjustment"
|
|
explicit_exclusion = decision and decision["cashflow_kind"] == "excluded"
|
|
kind = "excluded"
|
|
applied = amount if raw_valid else None
|
|
if general_transfer or initial_adjustment or catkind in {"exclude", "transfer"} or explicit_exclusion:
|
|
kind = "excluded"
|
|
elif typ == "refund":
|
|
if catkind in {"expense", "debt_payment"} or pair in savings or (decision and decision["cashflow_kind"] == "expense"):
|
|
kind, applied = "expense", -amount if raw_valid else None
|
|
else:
|
|
reason = "refund requires an expense classification or explicit expense decision"
|
|
elif catkind == "investment":
|
|
if pair in savings:
|
|
kind = "expense"
|
|
else:
|
|
kind = "excluded"
|
|
elif catkind == "debt_payment":
|
|
kind = "expense"
|
|
elif catkind in {"income", "expense"}:
|
|
kind = catkind
|
|
elif decision and decision["cashflow_kind"] in {"income", "expense"}:
|
|
kind = decision["cashflow_kind"]
|
|
else:
|
|
reason = "unknown cashflow classification requires an explicit decision"
|
|
if not raw_valid and kind != "excluded":
|
|
reason = "transaction amount is not an integer"
|
|
if reason:
|
|
blockers.append({"transaction_id": t["id"], "reason": reason})
|
|
kind, applied = "unknown", None
|
|
if kind != "excluded" and t["currency"].upper() != "KRW":
|
|
blockers.append({"transaction_id": t["id"], "reason": "non-KRW currency requires FX review"})
|
|
applied = None
|
|
provenance = "category" if t["category_active"] else "fallback"
|
|
evidence = {k: t[k] for k in ("id", "transaction_date", "type", "amount", "currency", "category", "subcategory")}
|
|
evidence.update(originalcashflow_kind=original, cashflow_kind=kind, applied_amount=applied,
|
|
decision=decision, sourceprovenance=provenance,
|
|
category_cashflow_kind=t["category_cashflow_kind"], category_active=t["category_active"])
|
|
source.append(evidence)
|
|
if applied is None or kind not in {"income", "expense"}:
|
|
if kind == "excluded" and applied is not None:
|
|
key = (kind, pair[0], pair[1], t["currency"].upper())
|
|
grp = grouped.setdefault(key, [0, 0]); grp[0] += applied; grp[1] += 1
|
|
continue
|
|
if t["currency"].upper() != "KRW":
|
|
continue
|
|
key = (kind, pair[0], pair[1], t["currency"].upper())
|
|
grp = grouped.setdefault(key, [0, 0]); grp[0] += applied; grp[1] += 1
|
|
if kind == "income": income += applied
|
|
else: expense += applied
|
|
items = [{"cashflow_kind": k, "category": c, "subcategory": s, "amount": v[0], "transaction_count": v[1], "currency": cur}
|
|
for (k, c, s, cur), v in sorted(grouped.items())]
|
|
return {"cashflow_policy": policy, "aggregation_version": _AGGREGATION_VERSION,
|
|
"snapshot_schema_version": _SNAPSHOT_SCHEMA_VERSION, "income_total": income,
|
|
"expense_total": expense, "net_cashflow": income-expense, "cashflow_items": items,
|
|
"cashflow_blockers": blockers, "cashflow_warnings": [], "source_snapshot": {"transactions": source}}
|