SPB Git forge

spb/uqo-chat

Public
14commits 1branches 0releases
1.4 MBsize
maindefault branch
17 days agolast push
Python 64.6% TypeScript 33.7% CSS 0.8%
16.4 KB · 344 lines python
Raw Blame History
1"""appraisal_calc — deterministic appraisal calculators (cost, land, depreciation, income, grids)."""23from __future__ import annotations45import statistics6from typing import Any, Literal78from pydantic import BaseModel, Field910from app.llm.schemas import ToolResult11from app.tools.coerce import boolish, num, pct12from app.tools.registry import ToolContext, registry1314Function = Literal[15    "cost_approach", "breakdown_depreciation", "indexed_cost", "unit_cost_estimate",16    "land_extraction", "land_allocation", "land_residual", "land_subdivision",17    "direct_capitalization", "gross_income_multiplier", "adjust_comparables", "effective_age_market",18]1920QUALITY = ["très mauvais", "mauvais", "passable", "moyen", "bon", "très bon", "excellent", "neuf"]212223class Args(BaseModel):24    function: Function25    params: dict[str, Any] = Field(default_factory=dict)262728def _n(p: dict[str, Any], key: str, default: float | None = None, *, required: bool = False) -> float:29    v = num(p.get(key), None)30    if v is None:31        if required:32            raise KeyError(key)33        return float(default or 0.0)34    return v353637def _p(p: dict[str, Any], key: str, default: float = 0.0) -> float:38    v = pct(p.get(key), None)39    return float(default) if v is None else v404142def _rank(v: Any, scale: list[str] | None) -> float:43    n = num(v, None)44    if n is not None and not isinstance(v, bool):45        return n46    b = boolish(v)47    words = [w.lower() for w in (scale or QUALITY)]48    if isinstance(v, str) and v.strip().lower() in words:49        return float(words.index(v.strip().lower()))50    if b is not None:51        return 1.0 if b else 0.052    if isinstance(v, str):53        for i, w in enumerate(words):54            if w in v.lower():55                return float(i)56    return 0.0575859# ------------------------------------------------------------------ calculators60def cost_approach(p: dict[str, Any]) -> dict[str, Any]:61    land = _n(p, "land_value", required=True)62    cn = _n(p, "cost_new", required=True)63    phys = _n(p, "physical_depreciation")64    if not phys and p.get("effective_age") and p.get("economic_life"):65        phys = cn * _n(p, "effective_age") / _n(p, "economic_life")66    func = _n(p, "functional_depreciation")67    ext = _n(p, "external_depreciation")68    site = _n(p, "site_improvements")69    total = phys + func + ext70    value = land + (cn - total) + site71    rows = [["Valeur du terrain", land], ["Coût neuf des améliorations", cn],72            ["− Dépréciation physique", -phys], ["− Dépréciation fonctionnelle", -func],73            ["− Dépréciation économique", -ext], ["= Coût déprécié", cn - total],74            ["+ Améliorations du site", site], ["= VALEUR INDIQUÉE", value]]75    return {"formula": "V = V_T + (C_N − D_phys − D_fonct − D_écon) + améliorations du site",76            "value": value, "total_depreciation": total, "depreciation_ratio": total / cn if cn else 0,77            "table": rows}787980def breakdown_depreciation(p: dict[str, Any]) -> dict[str, Any]:81    cn = _n(p, "cost_new", required=True)82    rows: list[list[Any]] = []83    curable = 0.084    for it in p.get("curable_physical") or []:85        c = num(it.get("cost_to_cure"), 0.0) or 0.086        curable += c87        rows.append(["Physique récupérable", str(it.get("item", "")), c])88    short_total = 0.089    short_cn = 0.090    for it in p.get("short_lived") or []:91        icn = num(it.get("cost_new"), 0.0) or 0.092        age = num(it.get("effective_age"), 0.0) or 0.093        life = num(it.get("life"), 1.0) or 1.094        d = icn * min(1.0, age / life)95        short_total += d96        short_cn += icn97        rows.append(["Physique non récup. — courte vie", f"{it.get('item', '')} ({age:g}/{life:g} ans)", d])98    long_ = p.get("long_lived") or {}99    long_age = num(long_.get("effective_age"), 0.0) or 0.0100    long_life = num(long_.get("economic_life") or long_.get("life"), 1.0) or 1.0101    long_base = cn - curable - short_cn102    long_dep = max(0.0, long_base) * min(1.0, long_age / long_life)103    rows.append(["Physique non récup. — longue vie", f"base {long_base:,.0f} × {long_age:g}/{long_life:g}", long_dep])104    functional = 0.0105    for it in p.get("functional") or []:106        amt = num(it.get("amount"), None)107        if amt is None and it.get("cost_to_cure") is not None:108            amt = (num(it.get("cost_to_cure"), 0.0) or 0.0) - (num(it.get("cost_if_new"), 0.0) or 0.0)109        if amt is None and it.get("rent_loss_annual") is not None:110            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)111        functional += amt or 0.0112        rows.append(["Fonctionnelle", f"{it.get('item', '')} ({it.get('type', 'déficience')})", amt or 0.0])113    ext = p.get("external") or {}114    external = 0.0115    if isinstance(ext, dict):116        if ext.get("amount") is not None:117            external = num(ext.get("amount"), 0.0) or 0.0118        elif ext.get("rent_loss_annual") is not None:119            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)120            share = pct(ext.get("building_share"), None)121            if share is not None:122                external *= share123    else:124        external = num(ext, 0.0) or 0.0125    if external:126        rows.append(["Économique (externe)", "", external])127    total = curable + short_total + long_dep + functional + external128    return {"formula": "D = récupérable + courte vie + longue vie (sur le résidu) + fonctionnelle + externe",129            "total_depreciation": total, "depreciated_cost": cn - total, "ratio": total / cn if cn else 0,130            "components": {"curable_physical": curable, "short_lived": short_total, "long_lived": long_dep,131                           "functional": functional, "external": external},132            "table": rows, "note": "Anti-double-comptage : la base longue vie exclut les éléments déjà déduits."}133134135def indexed_cost(p: dict[str, Any]) -> dict[str, Any]:136    hist = _n(p, "historical_cost", required=True)137    i0 = _n(p, "index_then", required=True)138    i1 = _n(p, "index_now", required=True)139    regional = _n(p, "regional_factor", 1.0) or 1.0140    size = _n(p, "size_factor", 1.0) or 1.0141    v = hist * (i1 / i0) * regional * size142    return {"formula": "C_N = C_hist × (I_actuel / I_hist) × facteur régional × facteur de taille",143            "value": v, "index_ratio": i1 / i0}144145146def unit_cost_estimate(p: dict[str, Any]) -> dict[str, Any]:147    area = _n(p, "area", required=True)148    unit = str(p.get("unit", "m2"))149    rate = _n(p, "cost_per_unit", required=True)150    direct = area * rate151    for f in p.get("factors") or []:152        direct *= num(f, 1.0) or 1.0153    extras = sum((num(x.get("amount"), 0.0) or 0.0) for x in (p.get("extras") or []))154    indirect = direct * _p(p, "indirect_pct")155    profit = (direct + extras + indirect) * _p(p, "profit_pct")156    total = direct + extras + indirect + profit157    return {"formula": "C_N = (S × c_u × facteurs + extras) × (1 + indirects) × (1 + profit)",158            "direct_costs": direct, "extras": extras, "indirect_costs": indirect, "entrepreneur_profit": profit,159            "cost_new": total, "cost_per_unit_all_in": total / area if area else 0, "unit": unit,160            "table": [["Coûts directs", direct], ["Extras", extras], ["Coûts indirects", indirect],161                      ["Profit de l'entrepreneur", profit], ["COÛT NEUF", total]]}162163164def land_extraction(p: dict[str, Any]) -> dict[str, Any]:165    price = _n(p, "sale_price", required=True)166    imp = _n(p, "improvements_depreciated_cost", None)167    if imp is None or (imp == 0 and p.get("cost_new")):168        imp = _n(p, "cost_new") - _n(p, "depreciation")169    v = price - imp170    return {"formula": "V_T = Prix de vente − coût déprécié des améliorations", "land_value": v,171            "land_ratio": v / price if price else 0}172173174def land_allocation(p: dict[str, Any]) -> dict[str, Any]:175    total = _n(p, "total_value", required=True)176    ratio = _p(p, "land_ratio", 0.25)177    return {"formula": "V_T = Valeur totale × ratio terrain", "land_value": total * ratio, "ratio": ratio}178179180def land_residual(p: dict[str, Any]) -> dict[str, Any]:181    noi = _n(p, "noi", required=True)182    bv = _n(p, "building_value", required=True)183    rb = _p(p, "building_rate", 0.09)184    rl = _p(p, "land_rate", 0.07)185    income_b = bv * rb186    residual = noi - income_b187    return {"formula": "V_T = (RNE − V_B × r_B) / r_T", "building_income": income_b, "residual_income": residual,188            "land_value": residual / rl if rl else 0}189190191def land_subdivision(p: dict[str, Any]) -> dict[str, Any]:192    lots = int(_n(p, "lots", required=True))193    price = _n(p, "price_per_lot", required=True)194    gross = lots * price195    costs = _n(p, "development_costs") + _n(p, "selling_costs") + _n(p, "carrying_costs")196    profit = gross * _p(p, "profit_pct", 0.15)197    net = gross - costs - profit198    years = _n(p, "absorption_years", 1.0) or 1.0199    r = _p(p, "discount_rate", 0.10)200    # cash flow spread evenly over absorption period, discounted at mid-year201    n = max(1, int(round(years)))202    pv = sum((net / n) / (1 + r) ** (t + 0.5) for t in range(n))203    return {"formula": "V_T = VA[(recettes − coûts − profit) étalés sur l'absorption]", "gross_sales": gross,204            "costs": costs, "profit": profit, "net_undiscounted": net, "land_value_pv": pv,205            "per_lot": pv / lots if lots else 0}206207208def direct_capitalization(p: dict[str, Any]) -> dict[str, Any]:209    noi = _n(p, "noi") if p.get("noi") is not None else None210    if noi is None:211        pgi = _n(p, "potential_gross_income", required=True)212        vac = pgi * _p(p, "vacancy_pct", 0.05)213        egi = pgi - vac + _n(p, "other_income")214        exp = _n(p, "operating_expenses")215        if not exp and p.get("expense_ratio") is not None:216            exp = egi * _p(p, "expense_ratio")217        noi = egi - exp218    else:219        pgi = vac = egi = exp = None220    cap = _p(p, "cap_rate", 0.07)221    return {"formula": "V = RNE / TGA", "noi": noi, "cap_rate": cap, "value": noi / cap if cap else 0,222            "table": [x for x in [["Revenu brut potentiel", pgi], ["− Vacances et mauvaises créances", -vac if vac else None],223                                  ["= Revenu brut effectif", egi], ["− Dépenses d'exploitation", -exp if exp else None],224                                  ["= RNE", noi], [f"÷ TGA {cap:.2%}", None], ["= VALEUR", noi / cap if cap else 0]]225                      if x[1] is not None or x[0].startswith("÷")]}226227228def gross_income_multiplier(p: dict[str, Any]) -> dict[str, Any]:229    if p.get("sale_price") is not None and p.get("gross_income") is not None:230        gim = _n(p, "sale_price") / _n(p, "gross_income")231        return {"formula": "MRB = Prix / Revenu brut", "gim": gim}232    gim = _n(p, "gim", required=True)233    gi = _n(p, "gross_income", required=True)234    return {"formula": "V = Revenu brut × MRB", "value": gim * gi, "gim": gim}235236237def adjust_comparables(p: dict[str, Any]) -> dict[str, Any]:238    subject = {k.lower(): v for k, v in (p.get("subject") or p.get("sujet") or {}).items()}239    rates = {k.lower(): num(v, 0.0) or 0.0 for k, v in (p.get("rates") or p.get("taux") or {}).items()}240    scales = {k.lower(): v for k, v in (p.get("scales") or p.get("echelles") or {}).items()}241    monthly = pct(p.get("monthly_trend") or p.get("taux_temps_mensuel"), 0.0) or 0.0242    comps = p.get("comparables") or []243    if not comps:244        raise KeyError("comparables")245    out_rows = []246    adjusted = []247    gross_pcts = []248    for c in comps:249        c = {k.lower(): v for k, v in c.items()}250        price = num(c.get("price") or c.get("prix"), 0.0) or 0.0251        months = num(c.get("months") or c.get("mois"), 0.0) or 0.0252        t_pct = pct(c.get("time_pct") or c.get("temps"), None)253        if t_pct is None:254            t_pct = months * monthly255        pat = price * (1 + t_pct)256        adjustments: dict[str, float] = {}257        for k, rate in rates.items():258            if k in subject or k in c:259                adj = (_rank(subject.get(k), scales.get(k)) - _rank(c.get(k), scales.get(k))) * rate260                adjustments[k] = adj261        for k, v in (c.get("adjustments") or {}).items():262            adjustments[k.lower()] = num(v, 0.0) or 0.0263        net = sum(adjustments.values())264        gross = sum(abs(v) for v in adjustments.values()) + abs(pat - price)265        final = pat + net266        adjusted.append(final)267        gp = gross / price if price else 0268        gross_pcts.append(gp)269        out_rows.append({"name": c.get("address") or c.get("adresse") or c.get("name") or "comparable",270                         "price": price, "time_pct": t_pct, "time_adjusted": pat, "adjustments": adjustments,271                         "net": net, "adjusted_price": final, "net_pct": net / price if price else 0, "gross_pct": gp,272                         "reliable": gp <= 0.25 and abs(net / price if price else 0) <= 0.15})273    weights = [1 / (1 + g) for g in gross_pcts]274    weighted = sum(a * w for a, w in zip(adjusted, weights, strict=False)) / sum(weights)275    best = min(range(len(comps)), key=lambda i: gross_pcts[i])276    return {"formula": "Prix ajusté = Prix × (1 + temps) + Σ (sujet − comparable) × taux",277            "comparables": out_rows,278            "stats": {"min": min(adjusted), "max": max(adjusted), "mean": statistics.fmean(adjusted),279                      "median": statistics.median(adjusted), "weighted": weighted,280                      "least_adjusted": out_rows[best]["name"], "least_adjusted_price": adjusted[best]},281            "note": "Repères du cours : ajustements bruts ≤ 25 %, nets ≤ 15 %. La pondération n'est pas une moyenne."}282283284def effective_age_market(p: dict[str, Any]) -> dict[str, Any]:285    price = _n(p, "sale_price", required=True)286    land = _n(p, "land_value", required=True)287    cn = _n(p, "cost_new", required=True)288    life = _n(p, "economic_life", required=True)289    dep = cn - (price - land)290    ratio = dep / cn if cn else 0291    return {"formula": "A_e = (D / C_N) × DVE, avec D = C_N − (Prix − V_T)", "depreciation": dep, "ratio": ratio,292            "effective_age": ratio * life, "annual_rate": ratio / _n(p, "actual_age", 1.0) if p.get("actual_age") else None}293294295FUNCS = {296    "cost_approach": cost_approach, "breakdown_depreciation": breakdown_depreciation, "indexed_cost": indexed_cost,297    "unit_cost_estimate": unit_cost_estimate, "land_extraction": land_extraction, "land_allocation": land_allocation,298    "land_residual": land_residual, "land_subdivision": land_subdivision,299    "direct_capitalization": direct_capitalization, "gross_income_multiplier": gross_income_multiplier,300    "adjust_comparables": adjust_comparables, "effective_age_market": effective_age_market,301}302303304def _fmt(v: Any) -> str:305    if isinstance(v, float):306        return f"{v:,.4f}".rstrip("0").rstrip(".") if abs(v) < 10 else f"{v:,.2f}"307    return str(v)308309310def _render(out: dict[str, Any], depth: int = 0) -> list[str]:311    lines = []312    for k, v in out.items():313        if k == "table" and isinstance(v, list):314            lines.append("table:")315            for row in v:316                lines.append("  " + " | ".join(_fmt(x) if not isinstance(x, str) else x for x in row))317        elif k == "comparables" and isinstance(v, list):318            for c in v:319                adj = ", ".join(f"{a}: {_fmt(b)}" for a, b in c["adjustments"].items())320                lines.append(f"  {c['name']}: prix {_fmt(c['price'])} → ajusté temps {_fmt(c['time_adjusted'])} ; "321                             f"ajust. [{adj}] net {_fmt(c['net'])} → PRIX AJUSTÉ {_fmt(c['adjusted_price'])} "322                             f"(net {c['net_pct']:.1%}, brut {c['gross_pct']:.1%}, {'fiable' if c['reliable'] else 'à pondérer faiblement'})")323        elif isinstance(v, dict):324            lines.append(f"{k}:")325            lines += ["  " + line for line in _render(v, depth + 1)]326        else:327            lines.append(f"{k} = {_fmt(v) if v is not None else '—'}")328    return lines329330331async def run(args: dict[str, Any], ctx: ToolContext) -> ToolResult:332    fn = args["function"]333    try:334        out = FUNCS[fn](args["params"])335    except (KeyError, ValueError, ZeroDivisionError, TypeError) as exc:336        return ToolResult(content=f"Paramètre manquant ou invalide pour {fn} : {exc}. Voir la description de "337                          "l'outil pour les paramètres attendus.", error=True)338    return ToolResult(content=f"Résultat {fn} :\n" + "\n".join(_render(out)),339                      payload={"function": fn, "params": args["params"], "result": out},340                      meta={"summary": f"{fn} : {out.get('formula', '')}"})341342343registry.register("appraisal_calc", run, Args)344