# Author: Simon-Pierre Boucher — contact@spboucher.ai
"""IAAO ratio-study statistics.

Implements the standard diagnostics of the IAAO (2013) *Standard on Ratio
Studies*: the median assessment ratio, the coefficient of dispersion (COD),
the price-related differential (PRD), and the coefficient of price-related
bias (PRB), together with nonparametric bootstrap confidence intervals.

Notation: r_i = AV_i / SP_i is the assessment ratio of sale i.
"""
from __future__ import annotations

import numpy as np
import pandas as pd

from . import config


# ---------------------------------------------------------------- point stats
def cod(ratio: np.ndarray) -> float:
    """Coefficient of dispersion: 100 × mean |r − med| / med (horizontal equity)."""
    med = np.median(ratio)
    return 100.0 * np.mean(np.abs(ratio - med)) / med


def prd(av: np.ndarray, sp: np.ndarray) -> float:
    """Price-related differential: mean ratio ÷ sale-weighted mean ratio.

    PRD > 1 indicates regressivity (low-priced properties assessed at higher
    ratios); the IAAO acceptable range is [0.98, 1.03].
    """
    r = av / sp
    return float(np.mean(r) / (np.sum(av) / np.sum(sp)))


def prb(av: np.ndarray, sp: np.ndarray) -> tuple[float, float]:
    """Coefficient of price-related bias (IAAO 2013, Appendix on PRB).

    Regress the proportional deviation of the ratio from its median on the
    log (base 2) of a value proxy that blends the sale price and the
    median-deflated assessment:

        (r_i − med) / med = α + PRB · ln2( 0.5·SP_i + 0.5·AV_i/med ) + u_i

    Returns (PRB, robust standard error). PRB = −0.03 means ratios fall by
    3% of the median with every doubling of value: regressive if PRB < 0.
    """
    r = av / sp
    med = np.median(r)
    y = (r - med) / med
    proxy = 0.5 * sp + 0.5 * av / med
    x = np.log2(proxy)
    X = np.column_stack([np.ones_like(x), x])
    beta, *_ = np.linalg.lstsq(X, y, rcond=None)
    resid = y - X @ beta
    # HC1 robust standard error of the slope
    XtX_inv = np.linalg.inv(X.T @ X)
    meat = (X * (resid ** 2)[:, None]).T @ X
    k = len(y) / (len(y) - 2)
    se = float(np.sqrt(k * (XtX_inv @ meat @ XtX_inv)[1, 1]))
    return float(beta[1]), se


# ---------------------------------------------------------------- bootstrap
def bootstrap_ci(av: np.ndarray, sp: np.ndarray,
                 n_boot: int = config.N_BOOT,
                 seed: int = config.SEED_BOOT) -> dict:
    """Percentile bootstrap 95% CIs for the median ratio, COD, PRD and PRB."""
    rng = np.random.default_rng(seed)
    n = len(av)
    stats = {"median": [], "cod": [], "prd": [], "prb": []}
    for _ in range(n_boot):
        idx = rng.integers(0, n, n)
        a, s = av[idx], sp[idx]
        r = a / s
        stats["median"].append(np.median(r))
        stats["cod"].append(cod(r))
        stats["prd"].append(prd(a, s))
        stats["prb"].append(prb(a, s)[0])
    return {k: (float(np.percentile(v, 2.5)), float(np.percentile(v, 97.5)))
            for k, v in stats.items()}


# ---------------------------------------------------------------- group table
def group_metrics(df: pd.DataFrame, by: str | list[str],
                  min_n: int = 50, ci: bool = False) -> pd.DataFrame:
    """IAAO statistics computed within each group of ``by``.

    Returns one row per group with n, median ratio, COD, PRD, PRB (and its
    SE); optionally percentile-bootstrap CIs (slow — reserve for headline
    rows).
    """
    rows = []
    for key, g in df.groupby(by):
        if len(g) < min_n:
            continue
        av = g["role_valeur_immeuble"].to_numpy(float)
        sp = g["amount"].to_numpy(float)
        r = av / sp
        b, se = prb(av, sp)
        row = {"group": key if isinstance(key, str) else "_".join(map(str, key)),
               "n": len(g), "median_ratio": float(np.median(r)),
               "cod": cod(r), "prd": prd(av, sp), "prb": b, "prb_se": se}
        if ci:
            cis = bootstrap_ci(av, sp)
            for stat, (lo, hi) in cis.items():
                row[f"{stat}_lo"], row[f"{stat}_hi"] = lo, hi
        rows.append(row)
    return pd.DataFrame(rows)