# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca
"""Fundamental-value triangulation: index chaining, resale arithmetic, yields, user-cost identity, lease cleaning."""
from __future__ import annotations

import sqlite3

import numpy as np
import pandas as pd
import pytest

from agentic_valuation.fundamental.repeat_sales import (describe_resale_sample, index_change, monthly_index, prior_targets,
                                                        resale_pairs, resale_targets)
from agentic_valuation.fundamental.rents import (annual_mortgage_rate, gross_yields, implied_value_ratio, load_leases,
                                                 median_rents, rooms_bucket, user_cost)


def toy_panel(growth: float = 0.01, months: int = 12, n_big: int = 20, n_small: int = 2) -> pd.DataFrame:
    """Two municipalities of one region, same class, log price per m² growing by `growth` per month.
    `big` has n_big sales per month (always ≥ 5), `small` has n_small (< 5, must fall back to the region)."""
    rows = []
    rng = np.random.default_rng(0)
    for m in range(months):
        date = pd.Timestamp(2021, 1, 1) + pd.DateOffset(months=m)
        for mun, n, base in (("big", n_big, 8.0), ("small", n_small, 7.5)):
            for k in range(n):
                area = 100.0 + 10 * (k % 3)
                lp = base + growth * m
                rows.append({"tx_id": f"{mun}-{m}-{k}", "parcel_id": f"{mun}-p{m}-{k}", "mun_code": mun, "region_group": "R",
                             "property_class": "house", "date": date + pd.Timedelta(days=int(rng.integers(0, 27))),
                             "floor_area_m2": area, "price": float(np.exp(lp) * area)})
    return pd.DataFrame(rows)


def test_index_chaining_recovers_constant_growth_and_falls_back_to_region():
    panel = toy_panel(growth=0.01, months=12)
    idx = monthly_index(panel, min_n=5, window=3)
    big = idx[(idx["level"] == "mun") & (idx["mun_code"] == "big")].sort_values("month_id")
    small = idx[(idx["level"] == "mun") & (idx["mun_code"] == "small")].sort_values("month_id")
    assert len(big) == 12 and big["log_index"].iloc[0] == 0.0
    # deterministic growth: every interior monthly change of the chained log index is 0.01, from the municipality
    # source; the first and last months of the grid carry the edge effect of the centred window (two-point median)
    d = np.diff(big["log_index"].to_numpy())
    np.testing.assert_allclose(d[1:-1], 0.01, atol=1e-9)
    np.testing.assert_allclose([d[0], d[-1]], [0.005, 0.005], atol=1e-9)
    assert set(big["source"].iloc[1:]) == {"mun"}
    # thin municipality: every change comes from the region series (22 sales/month in the region), no level jump
    assert set(small["source"].iloc[1:]) == {"region"}
    np.testing.assert_allclose(small["log_index"].to_numpy(), big["log_index"].to_numpy(), atol=1e-9)
    # region-level series are present for the fallback of index_change
    assert (idx["level"] == "region").sum() == 12
    ch = index_change(idx, ["big", "small", "absent"], ["house"] * 3, [pd.Timestamp(2021, 2, 15)] * 3, [pd.Timestamp(2021, 8, 3)] * 3,
                      region_group=["R", "R", "R"])
    np.testing.assert_allclose(ch, [0.06, 0.06, 0.06], atol=1e-9)
    ch_nofallback = index_change(idx, ["absent"], ["house"], [pd.Timestamp(2021, 2, 15)], [pd.Timestamp(2021, 8, 3)])
    assert np.isnan(ch_nofallback[0])


def test_index_switch_between_sources_has_no_level_jump():
    panel = toy_panel(growth=0.0, months=6, n_big=20)
    # municipality `mid` with 8 sales in months 0-2 and 2 sales in months 3-5, at a level very different from the region
    rows = []
    for m in range(6):
        date = pd.Timestamp(2021, 1, 1) + pd.DateOffset(months=m)
        for k in range(8 if m < 3 else 2):
            rows.append({"tx_id": f"mid-{m}-{k}", "parcel_id": f"mid-{m}-{k}", "mun_code": "mid", "region_group": "R",
                         "property_class": "house", "date": date, "floor_area_m2": 100.0, "price": float(np.exp(9.0) * 100)})
    panel = pd.concat([panel, pd.DataFrame(rows)], ignore_index=True)
    idx = monthly_index(panel, min_n=5, window=3)
    mid = idx[(idx["level"] == "mun") & (idx["mun_code"] == "mid")].sort_values("month_id")
    assert list(mid["source"]) == ["base", "mun", "mun", "region", "region", "region"]
    # flat markets everywhere → the chained index stays flat despite the switch of source (no jump of 9 − 8)
    np.testing.assert_allclose(mid["log_index"].to_numpy(), 0.0, atol=1e-9)


def test_resale_target_arithmetic():
    panel = toy_panel(growth=0.01, months=12)
    idx = monthly_index(panel, min_n=5, window=3)
    # add a resale of parcel big-p1-0: first sale 2021-02-10 at 100 000, resale 2021-08-10 at 130 000 (interior months)
    first = panel[panel["parcel_id"] == "big-p1-0"].iloc[0]
    panel.loc[panel["parcel_id"] == "big-p1-0", "price"] = 100_000.0
    panel.loc[panel["parcel_id"] == "big-p1-0", "date"] = pd.Timestamp(2021, 2, 10)
    resale = first.copy()
    resale["tx_id"], resale["date"], resale["price"] = "big-resale", pd.Timestamp(2021, 8, 10), 130_000.0
    full = pd.concat([panel, resale.to_frame().T], ignore_index=True)
    full["price"] = full["price"].astype(float)
    pairs = resale_pairs(full)
    assert len(pairs) == 1 and pairs["tx_id"].iloc[0] == "big-1-0" and pairs["resale_tx_id"].iloc[0] == "big-resale"
    t = resale_targets(panel, full, idx)
    assert len(t) == 1
    r = t.iloc[0]
    assert r["resale_tx_id"] == "big-resale" and r["resale_price"] == 130_000.0
    assert abs(r["months_held"] - 181 / (365.25 / 12)) < 1e-9
    np.testing.assert_allclose(r["index_change"], 0.06, atol=1e-9)                      # Feb → Aug = 6 × 0.01
    np.testing.assert_allclose(r["resale_implied_log_value"], np.log(130_000.0) - 0.06, atol=1e-9)
    np.testing.assert_allclose(r["gap"], np.log(1.3) - 0.06, atol=1e-9)
    np.testing.assert_allclose(r["annualised_log_return"], np.log(1.3) * 12 / r["months_held"], atol=1e-9)
    assert bool(r["short_hold"]) is True and bool(r["resale_flagged"]) is False
    # backward direction: the resale row is not in `panel` (analytic) so prior_targets is empty; add it and check
    analytic2 = full.copy()
    p = prior_targets(analytic2, full, idx)
    assert len(p) == 1
    np.testing.assert_allclose(p["prior_implied_log_value"].iloc[0], np.log(100_000.0) + 0.06, atol=1e-9)
    desc = describe_resale_sample(t, panel)
    assert desc["overall"]["n"] == 1 and desc["by_class"]["house"]["share_short_hold"] == 1.0


def test_resale_pairs_drop_structural_exclusions_and_same_date():
    full = pd.DataFrame({
        "tx_id": ["a", "b", "c", "d"], "parcel_id": ["p", "p", "p", "p"],
        "date": pd.to_datetime(["2021-01-01", "2021-01-01", "2022-01-01", "2023-01-01"]),
        "price": [100.0, 100.0, 120.0, 150.0], "exclude_reason": [None, "exact_duplicate", None, "assessment_ratio_outlier"],
        "in_headline_sample": [True, False, True, False],
    })
    pairs = resale_pairs(full)
    assert list(pairs["tx_id"]) == ["a", "c"] and list(pairs["resale_tx_id"]) == ["c", "d"]
    assert pairs["resale_exclude_reason"].iloc[1] == "assessment_ratio_outlier"


def test_gross_yield_and_user_cost_identity():
    sales = pd.DataFrame({
        "tx_id": ["s1", "s2", "s3"], "date": pd.to_datetime(["2023-05-01", "2023-06-01", "2023-06-01"]),
        "mun_code": ["66023", "66023", "66023"], "mun_name": ["Montréal"] * 3, "region_group": ["Montréal CMA"] * 3,
        "property_class": ["condo", "plex", "house"], "price": [300_000.0, 900_000.0, 500_000.0], "n_units": [1, 3, 1],
    })
    rents = pd.DataFrame({"mun_code": ["66023"], "year": [2023], "rooms_bucket": ["all"], "n_leases": [50], "rent_median": [1250.0]})
    gy = gross_yields(sales, rents, min_leases=10)
    assert list(gy["tx_id"]) == ["s1", "s2"]                      # houses are not in the yield classes
    np.testing.assert_allclose(gy["gross_yield"].to_numpy(), [12 * 1250 / 300_000, 12 * 1250 * 3 / 900_000])
    assert gross_yields(sales, rents, min_leases=60).empty         # cell too thin
    u = user_cost(0.05, 0.01, 0.025, 0.02, 0.02)
    np.testing.assert_allclose(u, 0.085)
    # identity: when R/P = u the rent-implied value equals the price
    np.testing.assert_allclose(implied_value_ratio(0.085, u), 1.0)
    np.testing.assert_allclose(implied_value_ratio(np.array([0.085, 0.17]), u), [1.0, 2.0])
    np.testing.assert_allclose(user_cost(np.array([0.04, 0.06]), 0.01, 0.025, 0.02, np.array([0.0, 0.04])), [0.095, 0.075])


def test_rooms_bucket_and_annual_rate():
    b = rooms_bucket(pd.Series([0, 1, 2, 3, 4, 7, np.nan]))
    assert list(b[:6]) == ["0-1", "0-1", "2", "3", "4+", "4+"] and pd.isna(b.iloc[6])
    weekly = pd.DataFrame({"date": pd.to_datetime(["2023-01-04", "2023-06-07", "2024-02-14"]), "rate_pct": [6.0, 7.0, 6.5]})
    r = annual_mortgage_rate(weekly, [2023, 2024, 2025])
    np.testing.assert_allclose(r.loc[2023], 0.065)
    np.testing.assert_allclose(r.loc[2024], 0.065)
    assert np.isnan(r.loc[2025])                                   # never filled
    assert annual_mortgage_rate(pd.DataFrame(columns=["date", "rate_pct"]), [2023]).isna().all()


def test_load_leases_cleaning_funnel(tmp_path):
    db = tmp_path / "rdl.db"
    con = sqlite3.connect(db)
    con.execute("CREATE TABLE rdl_housings (id INTEGER PRIMARY KEY, full_address TEXT, street_number TEXT, apartment_number TEXT, "
                "street_name TEXT, city TEXT, zip TEXT, lat REAL, lng REAL, price REAL, rooms INTEGER, year INTEGER, start_date TEXT, "
                "type_of_accomodation TEXT, heating_included INTEGER, electricity_included INTEGER, furnishing_included INTEGER, "
                "parking_included INTEGER, animal_allowed INTEGER, address_slug TEXT, updated_at TEXT, fetched_at TEXT)")
    rows = [
        (1, "1 rue A", "1", "", "rue A", "Montréal", "H1H1H1", 45.5, -73.6, 1200, 2, 2023, "2023-07-01", "apartement", 0, 0, 0, 0, 0, "1-rue-a", "", ""),
        (2, "1 rue A", "1", "", "rue A", "Montréal", "H1H1H1", 45.5, -73.6, 1200, 2, 2023, "", "apartement", 0, 0, 0, 0, 0, "1-rue-a", "", ""),  # duplicate
        (3, "2 rue B", "2", "", "rue B", "Montréal", "H1H1H1", None, None, 900, 1, 2023, "", "apartement", 0, 0, 0, 0, 0, "2-rue-b", "", ""),   # no coords
        (4, "3 rue C", "3", "", "rue C", "Montréal", "H1H1H1", 45.5, -73.6, 5, 1, 2023, "", "apartement", 0, 0, 0, 0, 0, "3-rue-c", "", ""),      # rent 5 $
        (5, "4 rue D", "4", "", "rue D", "Montréal", "H1H1H1", 45.5, -73.6, 1500, 3, 2923, "", "apartement", 0, 0, 0, 0, 0, "4-rue-d", "", ""),  # year typo
        (6, "5 rue E", "5", "", "rue E", "Montréal", "H1H1H1", 45.5, -73.6, 1500, None, 2024, "", "apartement", 0, 0, 0, 0, 0, "5-rue-e", "", ""),  # rooms null
    ]
    con.executemany("INSERT INTO rdl_housings VALUES (" + ",".join("?" * 22) + ")", rows)
    con.commit()
    con.close()
    df = load_leases(db)
    assert len(df) == 1 and df["rooms_bucket"].iloc[0] == "2" and bool(df["start_date_present"].iloc[0])
    assert df.attrs["funnel"] == {"raw": 6, "coordinates_present": 5, "rent_in_range": 4, "year_in_range": 3, "rooms_recorded": 2, "deduplicated": 1}
    df["mun_code"] = "66023"
    r = median_rents(df)
    assert set(r["rooms_bucket"]) == {"2", "all"} and (r["rent_median"] == 1200).all()


@pytest.mark.real
def test_real_resale_targets_cover_prior_sale_information():
    """On the real data every analytic row with `prior_sale_price` must appear as the later leg of a pair."""
    import polars as pl

    from agentic_valuation.config import PROCESSED, RAW

    a = pl.read_parquet(PROCESSED / "analytic.parquet")
    full = pl.read_parquet(RAW / "wp22_clean_transactions.parquet")
    idx = monthly_index(a)
    p = prior_targets(a, full, idx)
    with_prior = a.filter(pl.col("prior_sale_price").is_not_null())["tx_id"].to_list()
    assert len(set(with_prior) - set(p["tx_id"])) / max(len(with_prior), 1) < 0.02