spb/uqo-chat

1"""Deterministic financial calculations: six functions of a dollar and friends."""23from __future__ import annotations45import math6from typing import Any, Literal78from pydantic import BaseModel, Field910from app.llm.schemas import ToolResult11from app.tools.registry import ToolContext, registry1213Function = Literal[14    "fv_lump", "pv_lump", "fv_annuity", "pv_annuity", "sinking_fund", "capital_recovery",15    "payment", "irr", "npv", "age_life", "unit_cost", "market_extraction",16]171819class FinancialArgs(BaseModel):20    function: Function21    params: dict[str, Any] = Field(default_factory=dict)222324def _rate_per_period(p: dict[str, Any]) -> tuple[float, int]:25    rate = float(p.get("rate", 0.0))26    if rate > 1:  # given as percent27        rate = rate / 100.028    periods_per_year = int(p.get("periods_per_year", 1))29    n = int(p.get("periods", p.get("years", 0) * periods_per_year if "years" in p else 0))30    return rate / periods_per_year, n313233def compute(function: str, p: dict[str, Any]) -> dict[str, Any]:34    if function == "age_life":35        ae = float(p["effective_age"])36        dve = float(p["economic_life"])37        cn = float(p.get("cost_new", 1.0))38        ratio = ae / dve39        return {"formula": "D = (A_e / DVE) × C_N", "depreciation_ratio": ratio,40                "depreciation": ratio * cn, "depreciated_cost": cn - ratio * cn,41                "remaining_life": dve - ae}42    if function == "unit_cost":43        area = float(p["area"])44        unit = str(p.get("unit", "m2"))45        cost_per_unit = float(p["cost_per_unit"])46        factors = [float(x) for x in p.get("factors", [])]47        total = area * cost_per_unit48        for f in factors:49            total *= f50        return {"formula": "C_N = S × c_u × Π facteurs", "area": area, "unit": unit,51                "cost_new": total,52                "area_other_unit": area * 10.7639 if unit == "m2" else area / 10.7639}53    if function == "market_extraction":54        price = float(p["sale_price"])55        land = float(p["land_value"])56        cn = float(p["cost_new"])57        age = float(p.get("age", 0)) or None58        dep = cn - (price - land)59        out = {"formula": "D = C_N − (Prix − V_T)", "depreciation": dep, "ratio": dep / cn}60        if age:61            out["annual_rate"] = dep / cn / age62        return out6364    i, n = _rate_per_period(p)65    if function == "fv_lump":66        f = (1 + i) ** n67        return {"formula": "VF = VA × (1+i)^n", "factor": f,68                "value": float(p.get("amount", 1.0)) * f}69    if function == "pv_lump":70        f = (1 + i) ** -n71        return {"formula": "VA = VF × (1+i)^-n", "factor": f,72                "value": float(p.get("amount", 1.0)) * f}73    if function == "fv_annuity":74        f = n if i == 0 else ((1 + i) ** n - 1) / i75        return {"formula": "VF_annuité = PMT × [((1+i)^n − 1)/i]", "factor": f,76                "value": float(p.get("amount", 1.0)) * f}77    if function == "pv_annuity":78        f = n if i == 0 else (1 - (1 + i) ** -n) / i79        return {"formula": "VA_annuité = PMT × [(1 − (1+i)^-n)/i]", "factor": f,80                "value": float(p.get("amount", 1.0)) * f}81    if function == "sinking_fund":82        f = 1 / n if i == 0 else i / ((1 + i) ** n - 1)83        return {"formula": "FA = i / ((1+i)^n − 1)", "factor": f,84                "value": float(p.get("amount", 1.0)) * f}85    if function in {"capital_recovery", "payment"}:86        f = 1 / n if i == 0 else i / (1 - (1 + i) ** -n)87        return {"formula": "FRC = i / (1 − (1+i)^-n)", "factor": f,88                "value": float(p.get("amount", 1.0)) * f}89    if function == "npv":90        flows = [float(x) for x in p["cash_flows"]]91        rate = float(p.get("rate", 0.0))92        rate = rate / 100 if rate > 1 else rate93        npv = sum(cf / (1 + rate) ** t for t, cf in enumerate(flows))94        return {"formula": "VAN = Σ CF_t / (1+r)^t", "value": npv}95    if function == "irr":96        flows = [float(x) for x in p["cash_flows"]]97        lo, hi = -0.99, 10.09899        def f(r: float) -> float:100            return sum(cf / (1 + r) ** t for t, cf in enumerate(flows))101102        if f(lo) * f(hi) > 0:103            return {"formula": "TRI : VAN(r) = 0", "value": None,104                    "note": "Pas de TRI unique dans l'intervalle."}105        for _ in range(200):106            mid = (lo + hi) / 2107            if f(lo) * f(mid) <= 0:108                hi = mid109            else:110                lo = mid111        return {"formula": "TRI : VAN(r) = 0", "value": (lo + hi) / 2}112    raise ValueError(f"unknown function {function}")113114115def _fmt(v: Any) -> str:116    if isinstance(v, float):117        if math.isnan(v):118            return "nan"119        return f"{v:,.6f}".rstrip("0").rstrip(".") if abs(v) < 1e6 else f"{v:,.2f}"120    return str(v)121122123async def run(args: dict[str, Any], ctx: ToolContext) -> ToolResult:124    fn = args["function"]125    params = args["params"]126    try:127        out = compute(fn, params)128    except (KeyError, ValueError, ZeroDivisionError) as exc:129        return ToolResult(content=f"Paramètre manquant ou invalide pour {fn} : {exc}. "130                          "Paramètres attendus : rate (décimal ou %), periods ou years, "131                          "periods_per_year, amount ; ou effective_age/economic_life/cost_new ; "132                          "ou area/cost_per_unit/unit/factors ; ou sale_price/land_value/"133                          "cost_new/age ; ou cash_flows/rate.", error=True)134    lines = [f"{k} = {_fmt(v)}" for k, v in out.items()]135    content = f"Résultat {fn} (paramètres : {params}) :\n" + "\n".join(lines)136    return ToolResult(content=content, payload={"function": fn, "params": params, "result": out},137                      meta={"summary": f"{fn} : {out.get('formula', '')}"})138139140registry.register("financial_calc", run, FinancialArgs)141