# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca
"""Rent-based proxies: lease matching, median rents, crude gross rental yields and the user-cost diagnostic.

Data (docs/fundamental_value.md §3). `data/raw/external/rdl.db`, table `rdl_housings`, is a sample of residential
leases (monthly rent, rooms, lease year, optional start date, coordinates) copied from the Immo-Ka platform. Limits
that this module documents but cannot repair: (i) it is a self-selected sample of registered leases, not the rental
stock; (ii) no floor area, so a rooms-to-area bridge to the sales table does NOT exist; (iii) the `rooms` field is an
integer whose meaning the registry does not document (its distribution — mode at 2, 90 % in 1..3 — is consistent with
bedrooms); (iv) `year` is the lease year and `start_date` is missing for about 40 % of rows; (v) the raw table also
contains Ontario cities and implausible rents, which the cleaning below removes.

Economics. Let R be the annual rent a comparable unit commands and P the price. The one-period user-cost condition
(Poterba 1984; Himmelberg, Mayer & Sinai 2005), in the form of theory.tex eq. (usercost) with the Canadian tax treatment,
    R / P = u = i + τ_p + δ + γ − π^e,
so the fundamental value implied by a rent is V^F = R / u and its ratio to the price is (R/P) / u = gross_yield / u.
Here gross_yield is a *crude* ratio: the median rent of leases with a similar rooms count in the same municipality and
year, annualised, divided by the price of the condo (one unit) or the plex (rent × number of units). It ignores vacancy,
operating costs, quality differences between the rental and owner-occupied stock, and rent regulation; it is a
distribution diagnostic, not a valuation.
"""
from __future__ import annotations

import json
import sqlite3
import urllib.request
from pathlib import Path
from typing import Any

import numpy as np
import pandas as pd

from ..config import RAW
from ..evaluation._common import to_pandas

RDL_DB = RAW / "external" / "rdl.db"
QUARTIER_DB = RAW / "external" / "quartier.db"
BOC_SERIES = "V80691335"   # Bank of Canada Valet: "Conventional mortgage, 5-year", weekly, percent (verified 2026-10-27)
BOC_URL = f"https://www.bankofcanada.ca/valet/observations/{BOC_SERIES}/json?start_date=2015-01-01"
BOC_CACHE = RAW / "external" / f"boc_{BOC_SERIES}.json"

LEASE_RENT_MIN, LEASE_RENT_MAX = 200.0, 15_000.0     # monthly CAD; outside = data-entry error
LEASE_YEAR_MIN, LEASE_YEAR_MAX = 2015, 2026
MATCH_MAX_KM = 2.0                                     # lease → nearest analytic sale (municipality attribution)
ROOMS_BUCKETS = ["0-1", "2", "3", "4+", "all"]
YIELD_CLASSES = ("condo", "plex")
MIN_LEASES_PER_CELL = 10

# user-cost parameters (annual rates). ASSUMPTIONS, documented in docs/fundamental_value.md §4; the mortgage rate is
# the only externally measured input (Bank of Canada series above, annual mean of the weekly observations).
USER_COST_PARAMS: dict[str, dict[str, Any]] = {
    "tau_p": {"value": 0.010, "label": "property tax rate (municipal + school, share of market value)", "source": "assumption; Quebec effective rates cluster around 0.7–1.2 % of market value"},
    "delta": {"value": 0.025, "label": "maintenance and depreciation", "source": "assumption; Himmelberg, Mayer and Sinai (2005) use 2.5 %"},
    "gamma": {"value": 0.020, "label": "risk and liquidity premium", "source": "assumption; Himmelberg, Mayer and Sinai (2005) use 2.0 %"},
    "g_scenarios": {"value": [0.00, 0.02, 0.04], "label": "expected nominal appreciation π^e (three scenarios)", "source": "assumption; bracket of long-run nominal house-price growth"},
}


# ------------------------------------------------------------------------------------------------ leases

def rooms_bucket(rooms: pd.Series) -> pd.Series:
    r = pd.to_numeric(rooms, errors="coerce")
    out = pd.Series(pd.NA, index=r.index, dtype="string")
    out[r <= 1] = "0-1"
    out[r == 2] = "2"
    out[r == 3] = "3"
    out[r >= 4] = "4+"
    return out


def load_leases(path: Path = RDL_DB) -> pd.DataFrame:
    """Read `rdl_housings`, clean, and record the cleaning funnel in `df.attrs['funnel']`.

    Kept: coordinates present, rent in [200, 15 000] CAD/month, lease year in [2015, 2026], rooms recorded (≥ 0),
    exact duplicates (address, apartment, year, rent, rooms) collapsed to one row.
    """
    con = sqlite3.connect(path)
    df = pd.read_sql_query(
        "SELECT id, full_address, apartment_number, city, zip, lat, lng, price AS rent, rooms, year, start_date, "
        "type_of_accomodation, heating_included, electricity_included, furnishing_included, parking_included, address_slug "
        "FROM rdl_housings", con)
    con.close()
    funnel = {"raw": int(len(df))}
    df = df.dropna(subset=["lat", "lng"])
    funnel["coordinates_present"] = int(len(df))
    df["rent"] = pd.to_numeric(df["rent"], errors="coerce")
    df = df[(df["rent"] >= LEASE_RENT_MIN) & (df["rent"] <= LEASE_RENT_MAX)]
    funnel["rent_in_range"] = int(len(df))
    df["year"] = pd.to_numeric(df["year"], errors="coerce")
    df = df[(df["year"] >= LEASE_YEAR_MIN) & (df["year"] <= LEASE_YEAR_MAX)]
    df["year"] = df["year"].astype(int)
    funnel["year_in_range"] = int(len(df))
    df["rooms"] = pd.to_numeric(df["rooms"], errors="coerce")
    df = df[df["rooms"].notna() & (df["rooms"] >= 0)]
    funnel["rooms_recorded"] = int(len(df))
    df = df.drop_duplicates(subset=["address_slug", "apartment_number", "year", "rent", "rooms"])
    funnel["deduplicated"] = int(len(df))
    df["rooms_bucket"] = rooms_bucket(df["rooms"])
    df["start_date"] = pd.to_datetime(df["start_date"].replace("", pd.NA), errors="coerce")
    df["start_date_present"] = df["start_date"].notna()
    df = df.reset_index(drop=True)
    df.attrs["funnel"] = funnel
    return df


def _to_32198(lat: np.ndarray, lng: np.ndarray) -> tuple[np.ndarray, np.ndarray]:
    from pyproj import Transformer
    tr = Transformer.from_crs("EPSG:4326", "EPSG:32198", always_xy=True)
    x, y = tr.transform(np.asarray(lng, dtype=float), np.asarray(lat, dtype=float))
    return np.asarray(x), np.asarray(y)


def match_leases(leases: pd.DataFrame, analytic: Any, *, max_km: float = MATCH_MAX_KM, quartier_db: Path | None = QUARTIER_DB) -> pd.DataFrame:
    """Attribute each lease to a municipality (nearest analytic sale within `max_km`, Quebec Lambert EPSG:32198) and,
    when `quartier.db` is available, to a 2021 dissemination area (point in polygon). Leases outside Quebec or farther
    than `max_km` from any sale are left unmatched (`mun_code` NaN). Adds: mun_code, mun_name, region_group,
    match_dist_km, dauid."""
    from scipy.spatial import cKDTree

    a = to_pandas(analytic)
    cols = ["mun_code", "mun_name", "region_group", "lat", "lng"]
    ref = a[cols].dropna(subset=["lat", "lng"]).drop_duplicates(subset=["lat", "lng"])
    rx, ry = _to_32198(ref["lat"].to_numpy(), ref["lng"].to_numpy())
    tree = cKDTree(np.column_stack([rx, ry]))
    lx, ly = _to_32198(leases["lat"].to_numpy(), leases["lng"].to_numpy())
    dist, idx = tree.query(np.column_stack([lx, ly]), k=1, distance_upper_bound=max_km * 1000.0)
    ok = np.isfinite(dist)
    out = leases.copy()
    out["match_dist_km"] = np.where(ok, dist / 1000.0, np.nan)
    for c in ("mun_code", "mun_name", "region_group"):
        vals = ref[c].to_numpy()
        out[c] = pd.Series(np.where(ok, vals[np.where(ok, idx, 0)], None), index=out.index, dtype="object")
    out["dauid"] = pd.NA
    if quartier_db is not None and Path(quartier_db).exists():
        out.loc[ok, "dauid"] = _point_in_da(out.loc[ok, "lat"].to_numpy(), out.loc[ok, "lng"].to_numpy(), Path(quartier_db))
    out["matched"] = ok
    return out


def _point_in_da(lat: np.ndarray, lng: np.ndarray, quartier_db: Path) -> np.ndarray:
    from shapely import STRtree
    from shapely.geometry import MultiPolygon, Point, Polygon

    con = sqlite3.connect(quartier_db)
    geoms, ids = [], []
    for dauid, poly in con.execute("SELECT dauid, poly FROM da_poly"):
        rings = json.loads(poly)
        try:
            if rings and isinstance(rings[0][0][0], (int, float)):
                g = Polygon(rings[0], rings[1:])
            else:
                g = MultiPolygon([Polygon(p[0], p[1:]) for p in rings])
        except (IndexError, TypeError, ValueError):
            continue
        if not g.is_valid:
            g = g.buffer(0)
        geoms.append(g)
        ids.append(str(dauid))
    con.close()
    tree = STRtree(geoms)
    pts = [Point(float(x), float(y)) for x, y in zip(lng, lat)]
    res = np.full(len(pts), None, dtype=object)
    q = tree.query(pts, predicate="within")
    # several polygons can claim a boundary point: keep the first
    seen: set[int] = set()
    for pi, gi in zip(q[0], q[1]):
        if pi not in seen:
            res[pi] = ids[gi]
            seen.add(int(pi))
    return res


def median_rents(leases: pd.DataFrame, *, by: tuple[str, ...] = ("mun_code", "year"), min_n: int = 1) -> pd.DataFrame:
    """Median monthly rent by `by` × rooms bucket, plus an "all" bucket. Columns: *by, rooms_bucket, n_leases,
    rent_median, rent_p25, rent_p75. Only matched leases (mun_code present) are used."""
    df = leases.dropna(subset=["mun_code"]).copy()
    df["mun_code"] = df["mun_code"].astype(str)
    keys = list(by)
    frames = []
    for bucket_df, label in ((df, None), (df.assign(rooms_bucket="all"), "all")):
        g = bucket_df.groupby(keys + ["rooms_bucket"], observed=True)["rent"]
        s = g.agg(n_leases="size", rent_median="median", rent_p25=lambda v: v.quantile(0.25), rent_p75=lambda v: v.quantile(0.75)).reset_index()
        frames.append(s)
    out = pd.concat(frames, ignore_index=True)
    out = out[out["n_leases"] >= min_n]
    out["rooms_bucket"] = out["rooms_bucket"].astype(str)
    return out.sort_values(keys + ["rooms_bucket"]).reset_index(drop=True)


# ------------------------------------------------------------------------------------------------ yields

def gross_yields(analytic: Any, rents: pd.DataFrame, *, classes: tuple[str, ...] = YIELD_CLASSES, condo_bucket: str = "all",
                 plex_bucket: str = "all", min_leases: int = MIN_LEASES_PER_CELL) -> pd.DataFrame:
    """Crude gross rental yield for condo and plex sales: 12 × median rent (same municipality, same year, chosen rooms
    bucket) × units / price, where units = 1 for condos and `n_units` for plexes (null → 2). No rooms-to-area bridge
    exists, so the rent is NOT adjusted for the size of the sold unit. Returns one row per matched sale."""
    a = to_pandas(analytic)
    a = a[a["property_class"].isin(list(classes))].copy()
    a["tx_id"] = a["tx_id"].astype(str)
    a["mun_code"] = a["mun_code"].astype(str)
    a["year"] = pd.to_datetime(a["date"]).dt.year.astype(int)
    a["rooms_bucket"] = np.where(a["property_class"] == "condo", condo_bucket, plex_bucket)
    r = rents[rents["n_leases"] >= min_leases][["mun_code", "year", "rooms_bucket", "n_leases", "rent_median"]].copy()
    r["mun_code"] = r["mun_code"].astype(str)
    r["year"] = r["year"].astype(int)
    m = a.merge(r, on=["mun_code", "year", "rooms_bucket"], how="inner", validate="many_to_one")
    units = np.where(m["property_class"] == "plex", pd.to_numeric(m.get("n_units"), errors="coerce").fillna(2).clip(lower=1), 1.0)
    m["units"] = units
    m["annual_rent"] = 12.0 * m["rent_median"] * units
    m["gross_yield"] = m["annual_rent"] / m["price"].astype(float)
    keep = ["tx_id", "date", "year", "mun_code", "mun_name", "region_group", "property_class", "price", "units", "n_leases",
            "rent_median", "annual_rent", "gross_yield"]
    return m[[c for c in keep if c in m.columns]].reset_index(drop=True)


# ------------------------------------------------------------------------------------------------ user cost

def user_cost(i: Any, tau_p: float, delta: float, gamma: float, g: Any) -> Any:
    """u = i + τ_p + δ + γ − π^e (annual rates; `i` and `g` may be arrays)."""
    return np.asarray(i, dtype=float) + tau_p + delta + gamma - np.asarray(g, dtype=float)


def implied_value_ratio(gross_yield: Any, u: Any) -> Any:
    """(R / u) / P = gross_yield / u: the rent-implied fundamental value relative to the price."""
    return np.asarray(gross_yield, dtype=float) / np.asarray(u, dtype=float)


def fetch_boc_mortgage_rate(cache: Path = BOC_CACHE, *, download: bool = True, timeout: float = 20.0) -> pd.DataFrame:
    """Weekly 5-year conventional mortgage rate (percent) from the Bank of Canada Valet API, cached as JSON.
    Returns an empty frame (never an invented series) when neither the download nor the cache is available."""
    payload: dict[str, Any] | None = None
    if download:
        try:
            with urllib.request.urlopen(BOC_URL, timeout=timeout) as resp:  # noqa: S310 (fixed https URL)
                payload = json.loads(resp.read().decode("utf-8"))
            cache.parent.mkdir(parents=True, exist_ok=True)
            cache.write_text(json.dumps(payload), encoding="utf-8")
        except Exception:  # network failure → cache
            payload = None
    if payload is None and cache.exists():
        payload = json.loads(cache.read_text(encoding="utf-8"))
    if not payload or "observations" not in payload:
        return pd.DataFrame(columns=["date", "rate_pct"])
    rows = [(o["d"], float(o[BOC_SERIES]["v"])) for o in payload["observations"] if o.get(BOC_SERIES, {}).get("v") not in (None, "")]
    df = pd.DataFrame(rows, columns=["date", "rate_pct"])
    df["date"] = pd.to_datetime(df["date"])
    df.attrs["label"] = payload.get("seriesDetail", {}).get(BOC_SERIES, {}).get("label", "")
    return df.sort_values("date").reset_index(drop=True)


def annual_mortgage_rate(weekly: pd.DataFrame, years: list[int]) -> pd.Series:
    """Annual mean of the weekly rate as a decimal; NaN for years without observations (never filled)."""
    if weekly.empty:
        return pd.Series(np.nan, index=years, name="i")
    w = weekly.assign(year=weekly["date"].dt.year)
    m = w.groupby("year")["rate_pct"].mean() / 100.0
    return pd.Series([m.get(y, np.nan) for y in years], index=years, name="i")