"""appraisal_calc — deterministic appraisal calculators (cost, land, depreciation, income, grids).""" from __future__ import annotations import statistics from typing import Any, Literal from pydantic import BaseModel, Field from app.llm.schemas import ToolResult from app.tools.coerce import boolish, num, pct from app.tools.registry import ToolContext, registry Function = Literal[ "cost_approach", "breakdown_depreciation", "indexed_cost", "unit_cost_estimate", "land_extraction", "land_allocation", "land_residual", "land_subdivision", "direct_capitalization", "gross_income_multiplier", "adjust_comparables", "effective_age_market", ] QUALITY = ["très mauvais", "mauvais", "passable", "moyen", "bon", "très bon", "excellent", "neuf"] class Args(BaseModel): function: Function params: dict[str, Any] = Field(default_factory=dict) def _n(p: dict[str, Any], key: str, default: float | None = None, *, required: bool = False) -> float: v = num(p.get(key), None) if v is None: if required: raise KeyError(key) return float(default or 0.0) return v def _p(p: dict[str, Any], key: str, default: float = 0.0) -> float: v = pct(p.get(key), None) return float(default) if v is None else v def _rank(v: Any, scale: list[str] | None) -> float: n = num(v, None) if n is not None and not isinstance(v, bool): return n b = boolish(v) words = [w.lower() for w in (scale or QUALITY)] if isinstance(v, str) and v.strip().lower() in words: return float(words.index(v.strip().lower())) if b is not None: return 1.0 if b else 0.0 if isinstance(v, str): for i, w in enumerate(words): if w in v.lower(): return float(i) return 0.0 # ------------------------------------------------------------------ calculators def cost_approach(p: dict[str, Any]) -> dict[str, Any]: land = _n(p, "land_value", required=True) cn = _n(p, "cost_new", required=True) phys = _n(p, "physical_depreciation") if not phys and p.get("effective_age") and p.get("economic_life"): phys = cn * _n(p, "effective_age") / _n(p, "economic_life") func = _n(p, "functional_depreciation") ext = _n(p, "external_depreciation") site = _n(p, "site_improvements") total = phys + func + ext value = land + (cn - total) + site rows = [["Valeur du terrain", land], ["Coût neuf des améliorations", cn], ["− Dépréciation physique", -phys], ["− Dépréciation fonctionnelle", -func], ["− Dépréciation économique", -ext], ["= Coût déprécié", cn - total], ["+ Améliorations du site", site], ["= VALEUR INDIQUÉE", value]] return {"formula": "V = V_T + (C_N − D_phys − D_fonct − D_écon) + améliorations du site", "value": value, "total_depreciation": total, "depreciation_ratio": total / cn if cn else 0, "table": rows} def breakdown_depreciation(p: dict[str, Any]) -> dict[str, Any]: cn = _n(p, "cost_new", required=True) rows: list[list[Any]] = [] curable = 0.0 for it in p.get("curable_physical") or []: c = num(it.get("cost_to_cure"), 0.0) or 0.0 curable += c rows.append(["Physique récupérable", str(it.get("item", "")), c]) short_total = 0.0 short_cn = 0.0 for it in p.get("short_lived") or []: icn = num(it.get("cost_new"), 0.0) or 0.0 age = num(it.get("effective_age"), 0.0) or 0.0 life = num(it.get("life"), 1.0) or 1.0 d = icn * min(1.0, age / life) short_total += d short_cn += icn rows.append(["Physique non récup. — courte vie", f"{it.get('item', '')} ({age:g}/{life:g} ans)", d]) long_ = p.get("long_lived") or {} long_age = num(long_.get("effective_age"), 0.0) or 0.0 long_life = num(long_.get("economic_life") or long_.get("life"), 1.0) or 1.0 long_base = cn - curable - short_cn long_dep = max(0.0, long_base) * min(1.0, long_age / long_life) rows.append(["Physique non récup. — longue vie", f"base {long_base:,.0f} × {long_age:g}/{long_life:g}", long_dep]) functional = 0.0 for it in p.get("functional") or []: amt = num(it.get("amount"), None) if amt is None and it.get("cost_to_cure") is not None: amt = (num(it.get("cost_to_cure"), 0.0) or 0.0) - (num(it.get("cost_if_new"), 0.0) or 0.0) if amt is None and it.get("rent_loss_annual") is not None: amt = (num(it.get("rent_loss_annual"), 0.0) or 0.0) / max(1e-9, pct(it.get("cap_rate"), 0.08) or 0.08) functional += amt or 0.0 rows.append(["Fonctionnelle", f"{it.get('item', '')} ({it.get('type', 'déficience')})", amt or 0.0]) ext = p.get("external") or {} external = 0.0 if isinstance(ext, dict): if ext.get("amount") is not None: external = num(ext.get("amount"), 0.0) or 0.0 elif ext.get("rent_loss_annual") is not None: external = (num(ext.get("rent_loss_annual"), 0.0) or 0.0) / max(1e-9, pct(ext.get("cap_rate"), 0.08) or 0.08) share = pct(ext.get("building_share"), None) if share is not None: external *= share else: external = num(ext, 0.0) or 0.0 if external: rows.append(["Économique (externe)", "", external]) total = curable + short_total + long_dep + functional + external return {"formula": "D = récupérable + courte vie + longue vie (sur le résidu) + fonctionnelle + externe", "total_depreciation": total, "depreciated_cost": cn - total, "ratio": total / cn if cn else 0, "components": {"curable_physical": curable, "short_lived": short_total, "long_lived": long_dep, "functional": functional, "external": external}, "table": rows, "note": "Anti-double-comptage : la base longue vie exclut les éléments déjà déduits."} def indexed_cost(p: dict[str, Any]) -> dict[str, Any]: hist = _n(p, "historical_cost", required=True) i0 = _n(p, "index_then", required=True) i1 = _n(p, "index_now", required=True) regional = _n(p, "regional_factor", 1.0) or 1.0 size = _n(p, "size_factor", 1.0) or 1.0 v = hist * (i1 / i0) * regional * size return {"formula": "C_N = C_hist × (I_actuel / I_hist) × facteur régional × facteur de taille", "value": v, "index_ratio": i1 / i0} def unit_cost_estimate(p: dict[str, Any]) -> dict[str, Any]: area = _n(p, "area", required=True) unit = str(p.get("unit", "m2")) rate = _n(p, "cost_per_unit", required=True) direct = area * rate for f in p.get("factors") or []: direct *= num(f, 1.0) or 1.0 extras = sum((num(x.get("amount"), 0.0) or 0.0) for x in (p.get("extras") or [])) indirect = direct * _p(p, "indirect_pct") profit = (direct + extras + indirect) * _p(p, "profit_pct") total = direct + extras + indirect + profit return {"formula": "C_N = (S × c_u × facteurs + extras) × (1 + indirects) × (1 + profit)", "direct_costs": direct, "extras": extras, "indirect_costs": indirect, "entrepreneur_profit": profit, "cost_new": total, "cost_per_unit_all_in": total / area if area else 0, "unit": unit, "table": [["Coûts directs", direct], ["Extras", extras], ["Coûts indirects", indirect], ["Profit de l'entrepreneur", profit], ["COÛT NEUF", total]]} def land_extraction(p: dict[str, Any]) -> dict[str, Any]: price = _n(p, "sale_price", required=True) imp = _n(p, "improvements_depreciated_cost", None) if imp is None or (imp == 0 and p.get("cost_new")): imp = _n(p, "cost_new") - _n(p, "depreciation") v = price - imp return {"formula": "V_T = Prix de vente − coût déprécié des améliorations", "land_value": v, "land_ratio": v / price if price else 0} def land_allocation(p: dict[str, Any]) -> dict[str, Any]: total = _n(p, "total_value", required=True) ratio = _p(p, "land_ratio", 0.25) return {"formula": "V_T = Valeur totale × ratio terrain", "land_value": total * ratio, "ratio": ratio} def land_residual(p: dict[str, Any]) -> dict[str, Any]: noi = _n(p, "noi", required=True) bv = _n(p, "building_value", required=True) rb = _p(p, "building_rate", 0.09) rl = _p(p, "land_rate", 0.07) income_b = bv * rb residual = noi - income_b return {"formula": "V_T = (RNE − V_B × r_B) / r_T", "building_income": income_b, "residual_income": residual, "land_value": residual / rl if rl else 0} def land_subdivision(p: dict[str, Any]) -> dict[str, Any]: lots = int(_n(p, "lots", required=True)) price = _n(p, "price_per_lot", required=True) gross = lots * price costs = _n(p, "development_costs") + _n(p, "selling_costs") + _n(p, "carrying_costs") profit = gross * _p(p, "profit_pct", 0.15) net = gross - costs - profit years = _n(p, "absorption_years", 1.0) or 1.0 r = _p(p, "discount_rate", 0.10) # cash flow spread evenly over absorption period, discounted at mid-year n = max(1, int(round(years))) pv = sum((net / n) / (1 + r) ** (t + 0.5) for t in range(n)) return {"formula": "V_T = VA[(recettes − coûts − profit) étalés sur l'absorption]", "gross_sales": gross, "costs": costs, "profit": profit, "net_undiscounted": net, "land_value_pv": pv, "per_lot": pv / lots if lots else 0} def direct_capitalization(p: dict[str, Any]) -> dict[str, Any]: noi = _n(p, "noi") if p.get("noi") is not None else None if noi is None: pgi = _n(p, "potential_gross_income", required=True) vac = pgi * _p(p, "vacancy_pct", 0.05) egi = pgi - vac + _n(p, "other_income") exp = _n(p, "operating_expenses") if not exp and p.get("expense_ratio") is not None: exp = egi * _p(p, "expense_ratio") noi = egi - exp else: pgi = vac = egi = exp = None cap = _p(p, "cap_rate", 0.07) return {"formula": "V = RNE / TGA", "noi": noi, "cap_rate": cap, "value": noi / cap if cap else 0, "table": [x for x in [["Revenu brut potentiel", pgi], ["− Vacances et mauvaises créances", -vac if vac else None], ["= Revenu brut effectif", egi], ["− Dépenses d'exploitation", -exp if exp else None], ["= RNE", noi], [f"÷ TGA {cap:.2%}", None], ["= VALEUR", noi / cap if cap else 0]] if x[1] is not None or x[0].startswith("÷")]} def gross_income_multiplier(p: dict[str, Any]) -> dict[str, Any]: if p.get("sale_price") is not None and p.get("gross_income") is not None: gim = _n(p, "sale_price") / _n(p, "gross_income") return {"formula": "MRB = Prix / Revenu brut", "gim": gim} gim = _n(p, "gim", required=True) gi = _n(p, "gross_income", required=True) return {"formula": "V = Revenu brut × MRB", "value": gim * gi, "gim": gim} def adjust_comparables(p: dict[str, Any]) -> dict[str, Any]: subject = {k.lower(): v for k, v in (p.get("subject") or p.get("sujet") or {}).items()} rates = {k.lower(): num(v, 0.0) or 0.0 for k, v in (p.get("rates") or p.get("taux") or {}).items()} scales = {k.lower(): v for k, v in (p.get("scales") or p.get("echelles") or {}).items()} monthly = pct(p.get("monthly_trend") or p.get("taux_temps_mensuel"), 0.0) or 0.0 comps = p.get("comparables") or [] if not comps: raise KeyError("comparables") out_rows = [] adjusted = [] gross_pcts = [] for c in comps: c = {k.lower(): v for k, v in c.items()} price = num(c.get("price") or c.get("prix"), 0.0) or 0.0 months = num(c.get("months") or c.get("mois"), 0.0) or 0.0 t_pct = pct(c.get("time_pct") or c.get("temps"), None) if t_pct is None: t_pct = months * monthly pat = price * (1 + t_pct) adjustments: dict[str, float] = {} for k, rate in rates.items(): if k in subject or k in c: adj = (_rank(subject.get(k), scales.get(k)) - _rank(c.get(k), scales.get(k))) * rate adjustments[k] = adj for k, v in (c.get("adjustments") or {}).items(): adjustments[k.lower()] = num(v, 0.0) or 0.0 net = sum(adjustments.values()) gross = sum(abs(v) for v in adjustments.values()) + abs(pat - price) final = pat + net adjusted.append(final) gp = gross / price if price else 0 gross_pcts.append(gp) out_rows.append({"name": c.get("address") or c.get("adresse") or c.get("name") or "comparable", "price": price, "time_pct": t_pct, "time_adjusted": pat, "adjustments": adjustments, "net": net, "adjusted_price": final, "net_pct": net / price if price else 0, "gross_pct": gp, "reliable": gp <= 0.25 and abs(net / price if price else 0) <= 0.15}) weights = [1 / (1 + g) for g in gross_pcts] weighted = sum(a * w for a, w in zip(adjusted, weights, strict=False)) / sum(weights) best = min(range(len(comps)), key=lambda i: gross_pcts[i]) return {"formula": "Prix ajusté = Prix × (1 + temps) + Σ (sujet − comparable) × taux", "comparables": out_rows, "stats": {"min": min(adjusted), "max": max(adjusted), "mean": statistics.fmean(adjusted), "median": statistics.median(adjusted), "weighted": weighted, "least_adjusted": out_rows[best]["name"], "least_adjusted_price": adjusted[best]}, "note": "Repères du cours : ajustements bruts ≤ 25 %, nets ≤ 15 %. La pondération n'est pas une moyenne."} def effective_age_market(p: dict[str, Any]) -> dict[str, Any]: price = _n(p, "sale_price", required=True) land = _n(p, "land_value", required=True) cn = _n(p, "cost_new", required=True) life = _n(p, "economic_life", required=True) dep = cn - (price - land) ratio = dep / cn if cn else 0 return {"formula": "A_e = (D / C_N) × DVE, avec D = C_N − (Prix − V_T)", "depreciation": dep, "ratio": ratio, "effective_age": ratio * life, "annual_rate": ratio / _n(p, "actual_age", 1.0) if p.get("actual_age") else None} FUNCS = { "cost_approach": cost_approach, "breakdown_depreciation": breakdown_depreciation, "indexed_cost": indexed_cost, "unit_cost_estimate": unit_cost_estimate, "land_extraction": land_extraction, "land_allocation": land_allocation, "land_residual": land_residual, "land_subdivision": land_subdivision, "direct_capitalization": direct_capitalization, "gross_income_multiplier": gross_income_multiplier, "adjust_comparables": adjust_comparables, "effective_age_market": effective_age_market, } def _fmt(v: Any) -> str: if isinstance(v, float): return f"{v:,.4f}".rstrip("0").rstrip(".") if abs(v) < 10 else f"{v:,.2f}" return str(v) def _render(out: dict[str, Any], depth: int = 0) -> list[str]: lines = [] for k, v in out.items(): if k == "table" and isinstance(v, list): lines.append("table:") for row in v: lines.append(" " + " | ".join(_fmt(x) if not isinstance(x, str) else x for x in row)) elif k == "comparables" and isinstance(v, list): for c in v: adj = ", ".join(f"{a}: {_fmt(b)}" for a, b in c["adjustments"].items()) lines.append(f" {c['name']}: prix {_fmt(c['price'])} → ajusté temps {_fmt(c['time_adjusted'])} ; " f"ajust. [{adj}] net {_fmt(c['net'])} → PRIX AJUSTÉ {_fmt(c['adjusted_price'])} " f"(net {c['net_pct']:.1%}, brut {c['gross_pct']:.1%}, {'fiable' if c['reliable'] else 'à pondérer faiblement'})") elif isinstance(v, dict): lines.append(f"{k}:") lines += [" " + line for line in _render(v, depth + 1)] else: lines.append(f"{k} = {_fmt(v) if v is not None else '—'}") return lines async def run(args: dict[str, Any], ctx: ToolContext) -> ToolResult: fn = args["function"] try: out = FUNCS[fn](args["params"]) except (KeyError, ValueError, ZeroDivisionError, TypeError) as exc: return ToolResult(content=f"Paramètre manquant ou invalide pour {fn} : {exc}. Voir la description de " "l'outil pour les paramètres attendus.", error=True) return ToolResult(content=f"Résultat {fn} :\n" + "\n".join(_render(out)), payload={"function": fn, "params": args["params"], "result": out}, meta={"summary": f"{fn} : {out.get('formula', '')}"}) registry.register("appraisal_calc", run, Args)