import os
import platform
from dataclasses import dataclass
from pathlib import Path

import matplotlib
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import sklearn
import xgboost
from IPython.display import display

pd.set_option("display.width", 160)
pd.set_option("display.max_columns", 40)

TARGET_YEAR = 2026                          # l'année à prévoir
LAST_OBSERVED_YEAR = TARGET_YEAR - 1        # l'extrait s'arrête au 2025-12-31
REQUIRED_BACKTEST_YEARS = (2023, 2024, 2025)  # exigés par la consigne
PERCENT = 100.0                             # fraction -> pourcentage
DAYS_PER_YEAR = 365.25                      # annualisation des intervalles entre baux

# Les CSV d'origine se placent à côté du notebook, comme pour le starter.
# Variable d'environnement facultative si on préfère les garder ailleurs.
DATA_DIR = Path(os.environ.get("EQUINOXE_DATA_DIR", "."))
SOURCE_FILES = {"listings": "equinoxe_listings.csv", "leases": "equinoxe_lease_history.csv",
                "concessions": "equinoxe_concessions.csv", "asking": "equinoxe_asking_history.csv"}
missing_files = [file_name for file_name in SOURCE_FILES.values() if not (DATA_DIR / file_name).is_file()]
if missing_files:
    raise FileNotFoundError(f"Placer les CSV d'origine à côté du notebook (ou définir EQUINOXE_DATA_DIR). Manquants : {missing_files}")

source_tables = {table: pd.read_csv(DATA_DIR / file_name) for table, file_name in SOURCE_FILES.items()}
listings, leases = source_tables["listings"], source_tables["leases"]
concessions, asking = source_tables["concessions"], source_tables["asking"]

for name, table in source_tables.items():
    print(f"{name:<12} {len(table):>6,} lignes   {table.shape[1]:>2} colonnes")
print({"python": platform.python_version(), "pandas": pd.__version__, "numpy": np.__version__,
       "matplotlib": matplotlib.__version__, "scikit-learn": sklearn.__version__, "xgboost": xgboost.__version__})

listings      1,061 lignes   29 colonnes
leases        4,302 lignes   35 colonnes
concessions   3,560 lignes   18 colonnes
asking        3,857 lignes   14 colonnes
{'python': '3.12.7', 'pandas': '2.2.3', 'numpy': '2.3.5', 'matplotlib': '3.11.2', 'scikit-learn': '1.9.1', 'xgboost': '2.1.4'}

LEASE_DATE_COLUMNS = ["sLeaseFrom", "sLeaseTo", "sSignDate", "sAvailable"]
for column in LEASE_DATE_COLUMNS:
    if column in leases.columns:
        leases[column] = pd.to_datetime(leases[column], errors="coerce")

leases["year"] = leases.sLeaseFrom.dt.year
asking["date"] = pd.to_datetime(asking.sMonth + "-01")

lease_column_profile = pd.DataFrame({"type": leases.dtypes.astype(str),
                                     "valeurs_manquantes": leases.isna().sum(),
                                     "valeurs_distinctes": leases.nunique()})
lease_column_profile

UNIT_KEY = ["sPropCode", "sUnitCode"]       # clé d'unité sûre (jamais sSite + sUnitCode)
PROMO_CODE = "PromoPay"

COLUMN_RENAMES = {"sPropCode": "property_code", "sUnitCode": "unit_code", "sState": "province", "sBeds": "bedrooms",
                  "sSignDate": "sign_date", "sLeaseFrom": "lease_start", "sLeaseTo": "lease_end", "sTermSeq": "term_seq",
                  "sRent": "face_rent", "sRentEffective": "effective_rent"}


def normalise_unit_key(table):
    # Copie avec la clé d'unité nettoyée : espaces retirés, code de propriété en minuscules.
    keyed = table.copy()
    keyed["sPropCode"] = keyed["sPropCode"].astype(str).str.strip().str.lower()
    keyed["sUnitCode"] = keyed["sUnitCode"].astype(str).str.strip()
    return keyed


promo_entries = normalise_unit_key(concessions[concessions.sChargeCode.eq(PROMO_CODE)])
unit_listed_rent = normalise_unit_key(listings)[UNIT_KEY + ["sRent"]]
promo_entries = promo_entries.merge(unit_listed_rent, on=UNIT_KEY, how="left", validate="many_to_one")
promo_check = pd.Series({
    "écritures PromoPay": len(promo_entries),
    "part avec sMonths = 1": (promo_entries.sMonths == 1).mean(),
    "part avec sAmount négatif": (promo_entries.sAmount < 0).mean(),
    "crédit médian / loyer de l'unité": (-promo_entries.sAmount / promo_entries.sRent).median(),
})
display(promo_check.round(3))

COLUMN_MEANINGS = {
    "sPropCode": "code de propriété (une phase d'immeuble); moitié de la clé d'unité",
    "sUnitCode": "numéro d'unité dans la propriété; autre moitié de la clé d'unité",
    "sState": "province : Quebec (TAL) ou Ontario (ligne directrice)",
    "sBeds": "nombre de chambres",
    "sSignDate": "date de signature : sert à savoir ce qui était connu à une date donnée",
    "sLeaseFrom": "début du bail : date de l'événement mesuré",
    "sLeaseTo": "fin contractuelle du bail : sert à anticiper les baux qui expirent en 2026",
    "sTermSeq": "rang du bail dans l'historique de l'unité : définit le bail précédent",
    "sRent": "loyer contractuel mensuel, celui écrit au bail",
    "sRentEffective": "loyer effectif mensuel, après concessions amorties sur le bail : ce qui est encaissé",
}
column_dictionary = pd.DataFrame({"nouveau nom": COLUMN_RENAMES, "ce que la colonne représente": COLUMN_MEANINGS})
column_dictionary.index.name = "colonne d'origine"
column_dictionary

écritures PromoPay                  1494.000
part avec sMonths = 1                  0.882
part avec sAmount négatif              1.000
crédit médian / loyer de l'unité       0.711
dtype: float64

REQUIRED_LEASE_COLUMNS = {"sPropCode", "sUnitCode", "sState", "sBeds", "sRent", "sRentEffective",
                          "sSignDate", "sLeaseFrom", "sLeaseTo", "sTermSeq", "sRenewal"}
KNOWN_PROVINCES = ["Quebec", "Ontario"]
LEASE_NUMERIC_COLUMNS = ["sRent", "sRentEffective", "sTermSeq", "sRenewal", "sBeds"]


class DataContractError(ValueError):
    # Une entrée ne respecte pas une hypothèse dont le modèle dépend.
    pass


def validate_leases(lease_table):
    # Copie validée des baux : dates et nombres convertis, clé nettoyée, règles vérifiées.
    if not isinstance(lease_table, pd.DataFrame) or lease_table.empty:
        raise DataContractError("leases doit être un DataFrame non vide de l'historique des baux")
    missing_columns = REQUIRED_LEASE_COLUMNS - set(lease_table.columns)
    if missing_columns:
        raise DataContractError(f"colonnes manquantes : {sorted(missing_columns)}")
    checked = lease_table.copy()
    for column in ["sSignDate", "sLeaseFrom", "sLeaseTo"]:
        checked[column] = pd.to_datetime(checked[column], errors="coerce")
    for column in LEASE_NUMERIC_COLUMNS:
        checked[column] = pd.to_numeric(checked[column], errors="coerce")
    if checked[list(REQUIRED_LEASE_COLUMNS)].isna().any().any():
        raise DataContractError("valeurs manquantes ou illisibles dans les colonnes requises")
    checked = normalise_unit_key(checked)
    if checked[UNIT_KEY].eq("").any().any() or not checked["sState"].isin(KNOWN_PROVINCES).all():
        raise DataContractError("clé d'unité vide ou province inconnue")
    if not np.isfinite(checked[LEASE_NUMERIC_COLUMNS].to_numpy()).all():
        raise DataContractError("valeur numérique non finie")
    if not checked["sRenewal"].isin([0, 1]).all():
        raise DataContractError("sRenewal doit valoir 0 ou 1")
    if ((checked["sTermSeq"] < 1) | (checked["sTermSeq"] % 1 != 0)).any():
        raise DataContractError("sTermSeq doit être un entier positif")
    if checked.duplicated(UNIT_KEY + ["sTermSeq"]).any():
        raise DataContractError("séquence de bail dupliquée pour une même unité")
    if (checked[["sRent", "sRentEffective"]] <= 0).any().any():
        raise DataContractError("les loyers doivent être positifs")
    if (checked["sRentEffective"] > checked["sRent"]).any():
        raise DataContractError("loyer effectif supérieur au loyer contractuel")
    if ((checked["sLeaseTo"] < checked["sLeaseFrom"]) | (checked["sSignDate"] > checked["sLeaseFrom"])).any():
        raise DataContractError("dates de bail ou de signature incohérentes")
    if checked.groupby(UNIT_KEY)["sState"].nunique().gt(1).any():
        raise DataContractError("une unité ne peut pas changer de province")
    return checked


validated_leases = validate_leases(leases)
print(f"Contrat de données respecté : {len(validated_leases):,} baux, {validated_leases.groupby(UNIT_KEY).ngroups:,} unités.")

Contrat de données respecté : 4,302 baux, 1,061 unités.

print(listings.groupby(["sState", "sCity", "sBuilding"]).size().to_string())
print()
print(listings.groupby(["sBeds", "sBaths"])
      .agg(units=("sRent", "size"),
           median_rent=("sRent", "median"),
           median_sqft=("sSqft", "median"))
      .to_string())

sState   sCity          sBuilding     
Ontario  Ottawa         The Met           124
Quebec   Laval          Daniel-Johnson    134
                        Levesque           84
                        Saint-Elzear      361
         Mont-Royal     Le Carlyle        192
         Pointe-Claire  Westpark          166

              units  median_rent  median_sqft
sBeds sBaths                                 
0     1.0        49       2565.0        550.0
1     1.0       350       1380.0        600.0
2     1.0       358       2440.0       1150.0
      2.0       281       2605.0       1265.0
3     1.0         2       2687.5       1597.5
      2.0        20       3237.5       1545.0
      3.0         1       4115.0       1905.0

FIRST_YEAR_SHOWN = 2019

naive = (leases[leases.year.between(FIRST_YEAR_SHOWN, LAST_OBSERVED_YEAR)]
         .groupby("year")
         .agg(leases=("sRent", "size"), median_rent=("sRent", "median")))
naive["yoy_pct"] = (naive.median_rent.pct_change() * PERCENT).round(2)
naive

TWO_BEDROOMS = 2

window = leases[leases.year.between(FIRST_YEAR_SHOWN, LAST_OBSERVED_YEAR)]
mix = window.groupby("year").agg(
    leases=("sRent", "size"),
    median_sqft=("sSqft", "median"),
    pct_2bed_plus=("sBeds", lambda beds: (beds >= TWO_BEDROOMS).mean() * PERCENT),
    buildings=("sBuilding", "nunique"))
mix["rent_per_sqft"] = (window.assign(psf=window.sRent / window.sSqft)
                        .groupby("year").psf.median())
mix.round(2)

SNAPSHOT_YEAR = 2024   # première année où les trois nouveaux immeubles louent tous

# À partir de quand chaque immeuble a-t-il commencé à générer des baux?
print(window.groupby("sBuilding").year.min().sort_values().to_string())
print()
# Et à quel niveau de loyer? (baux de SNAPSHOT_YEAR)
print(window[window.year == SNAPSHOT_YEAR].groupby("sBuilding")
      .agg(leases=("sRent", "size"), median_rent=("sRent", "median"))
      .sort_values("median_rent").to_string())

sBuilding
Daniel-Johnson    2019
Levesque          2019
Saint-Elzear      2019
Le Carlyle        2023
The Met           2023
Westpark          2024

                leases  median_rent
sBuilding                          
Levesque            84       1672.5
Daniel-Johnson     136       1870.0
Saint-Elzear       215       2145.0
Westpark           144       2202.5
Le Carlyle         195       2350.0
The Met            128       2752.5

building_mix = pd.crosstab(validated_leases.sLeaseFrom.dt.year, validated_leases.sBuilding, normalize="index")
ax = building_mix.plot.area(stacked=True, figsize=(9, 4))
ax.set(title="Composition des baux par immeuble", ylabel="Part des baux de l'année", xlabel="Année de début du bail")
ax.legend(loc="center left", bbox_to_anchor=(1, 0.5))
plt.tight_layout()
plt.show()

COLLISION_EXAMPLE_SITE, COLLISION_EXAMPLE_UNIT = "saint-elzear", "0203"   # exemple donné par le starter

site_key_collisions = listings[listings.duplicated(["sSite", "sUnitCode"], keep=False)]
print(f"{len(site_key_collisions):,} unités sur {len(listings):,} entrent en collision sur sSite + sUnitCode")
example = listings[(listings.sSite == COLLISION_EXAMPLE_SITE) & (listings.sUnitCode == COLLISION_EXAMPLE_UNIT)]
print(f"Exemple {COLLISION_EXAMPLE_SITE} / {COLLISION_EXAMPLE_UNIT} : {example.hUnit.nunique()} appartements distincts (hUnit), "
      f"{example.sPropCode.nunique()} codes de propriété différents")
print(f"Doublons sur sPropCode + sUnitCode dans les listings : {listings.duplicated(UNIT_KEY).sum()}")

248 unités sur 1,061 entrent en collision sur sSite + sUnitCode
Exemple saint-elzear / 0203 : 3 appartements distincts (hUnit), 3 codes de propriété différents
Doublons sur sPropCode + sUnitCode dans les listings : 0

GAP_MIN_YEARS, GAP_MAX_YEARS = 0.5, 2.5   # plus court : souvent le même bail enregistré deux fois; plus long : trop ancien
FIRST_PAIR_YEAR_SHOWN = 2021


def same_unit_growth(df, price_col="sRent", date_col="sLeaseFrom", min_gap=GAP_MIN_YEARS, max_gap=GAP_MAX_YEARS):
    # Version du starter : variation annualisée du loyer, chaque unité comparée à son propre bail précédent.
    d = df.sort_values(UNIT_KEY + [date_col]).copy()
    d["prev_price"] = d.groupby(UNIT_KEY)[price_col].shift(1)
    d["prev_date"] = d.groupby(UNIT_KEY)[date_col].shift(1)
    d = d.dropna(subset=["prev_price", "prev_date"])
    d["gap_years"] = (d[date_col] - d.prev_date).dt.days / DAYS_PER_YEAR
    d = d[d.gap_years.between(min_gap, max_gap, inclusive="neither")]
    d = d[(d.prev_price > 0) & (d[price_col] > 0)]
    d["growth_pct"] = ((d[price_col] / d.prev_price) ** (1 / d.gap_years) - 1) * PERCENT
    return d


pairs = same_unit_growth(leases)
print(f"{len(pairs):,} paires utilisables sur {pairs.groupby(UNIT_KEY).ngroups:,} unités")

honest = (pairs[pairs.year.between(FIRST_PAIR_YEAR_SHOWN, LAST_OBSERVED_YEAR)]
          .groupby("year")
          .growth_pct.agg(pairs="size", median="median").round(2))
honest

3,179 paires utilisables sur 890 unités

fig, ax = plt.subplots(figsize=(9, 4.5))
years_shown = range(FIRST_PAIR_YEAR_SHOWN, TARGET_YEAR)
ax.plot(years_shown, naive.loc[years_shown, "yoy_pct"], "o--", label="naïf : médiane de tous les baux", lw=2)
ax.plot(years_shown, honest.loc[years_shown, "median"], "o-", label="même unité comparée à elle-même", lw=2)
ax.axhline(0, color="grey", lw=0.8)
ax.set_ylabel("croissance annuelle des loyers (%)")
ax.set_title("La méthode naïve mesure surtout la composition")
ax.set_xticks(list(years_shown))
ax.legend()
ax.grid(alpha=0.3)
plt.tight_layout()
plt.show()

@dataclass(frozen=True)
class ForecastSpec:
    # Paramètres figés avant les backtests de cette révision.
    origin_month: int = 12                         # origine de chaque prévision : 31 décembre de l'année précédente
    origin_day: int = 31
    gap_min_years: float = GAP_MIN_YEARS           # intervalle admissible entre deux débuts de bail
    gap_max_years: float = GAP_MAX_YEARS
    growth_clip: tuple = (-0.30, 0.60)             # bornes de robustesse, moyennes d'entraînement seulement
    gap_clip: tuple = (0.0, 0.75)                  # bornes de l'écart de concession prévu
    shrink_k_cell: float = 10.0                    # rétrécissement immeuble × événement vers province × événement
    shrink_k_mix: float = 20.0                     # rétrécissement de la part de renouvellements vers la province
    trailing_years_mix: int = 3                    # fenêtre de la part de renouvellements
    components: tuple = ("DIRECT", "BRIDGE", "ANCHOR")
    component_weights: tuple = (1 / 3, 1 / 3, 1 / 3)
    face_persistence_weight: float = 0.5           # contractuel : moitié dernière année, moitié historique
    evaluation_years: tuple = (2020, 2021, 2022, 2023, 2024, 2025)


SPEC = ForecastSpec()
PAIR_COLUMNS = ["property_code", "unit_code", "province", "bedrooms", "term_seq", "prior_seq", "sign_date", "lease_start",
                "lease_end", "prior_start", "prior_sign", "event_year", "event_type", "prior_face_rent", "prior_effective_rent",
                "face_rent", "effective_rent", "interval_years", "face_growth", "effective_growth", "prior_gap", "new_gap", "eligible"]


def build_pairs(lease_table):
    # Paires de baux consécutifs d'une même unité, croissances annualisées et écarts de concession.
    ordered = validate_leases(lease_table).sort_values(UNIT_KEY + ["sTermSeq", "sLeaseFrom"])
    by_unit = ordered.groupby(UNIT_KEY, sort=False)
    paired = ordered.assign(prior_start=by_unit["sLeaseFrom"].shift(1), prior_sign=by_unit["sSignDate"].shift(1),
                            prior_face_rent=by_unit["sRent"].shift(1), prior_effective_rent=by_unit["sRentEffective"].shift(1),
                            prior_seq=by_unit["sTermSeq"].shift(1))
    paired = paired[paired["prior_start"].notna()].copy()
    paired["interval_years"] = (paired["sLeaseFrom"] - paired["prior_start"]).dt.days / DAYS_PER_YEAR
    paired["face_growth"] = (paired["sRent"] / paired["prior_face_rent"]) ** (1.0 / paired["interval_years"]) - 1.0
    paired["effective_growth"] = (paired["sRentEffective"] / paired["prior_effective_rent"]) ** (1.0 / paired["interval_years"]) - 1.0
    paired["prior_gap"] = 1.0 - paired["prior_effective_rent"] / paired["prior_face_rent"]
    paired["new_gap"] = 1.0 - paired["sRentEffective"] / paired["sRent"]
    rents_positive = (paired[["sRent", "prior_face_rent", "sRentEffective", "prior_effective_rent"]] > 0).all(axis=1)
    consecutive = paired["sTermSeq"].sub(paired["prior_seq"]).eq(1)
    paired["eligible"] = (paired["interval_years"].gt(SPEC.gap_min_years) & paired["interval_years"].lt(SPEC.gap_max_years)
                          & rents_positive & consecutive)
    paired["event_year"] = paired["sLeaseFrom"].dt.year
    paired["event_type"] = np.where(paired["sRenewal"].eq(1), "renewal", "turnover")
    return paired.rename(columns=COLUMN_RENAMES)[PAIR_COLUMNS].reset_index(drop=True)


def exact_bridge(face_growth, prior_gap, new_gap, interval_years=1.0):
    # Identité exacte : (1 + g_contractuel) × ((1 − écart_nouveau) / (1 − écart_précédent)) ** (1/t) − 1.
    inputs = (interval_years, prior_gap, new_gap, face_growth)
    if (any(not np.isfinite(np.asarray(value)).all() for value in inputs)
            or np.any(np.asarray(interval_years) <= 0) or np.any(np.asarray(prior_gap) >= 1)
            or np.any(np.asarray(prior_gap) < 0) or np.any(np.asarray(new_gap) < 0)
            or np.any(np.asarray(new_gap) >= 1) or np.any(np.asarray(face_growth) <= -1)):
        raise DataContractError("le pont exige des durées positives et des ratios de loyer valides")
    return (1.0 + face_growth) * ((1.0 - new_gap) / (1.0 - prior_gap)) ** (1.0 / interval_years) - 1.0


unit_pairs = build_pairs(leases)
eligible_pairs = unit_pairs[unit_pairs["eligible"]]
exclusion_reasons = pd.Series({
    "intervalle ≤ 0,5 an": int((unit_pairs.interval_years <= SPEC.gap_min_years).sum()),
    "intervalle ≥ 2,5 ans": int((unit_pairs.interval_years >= SPEC.gap_max_years).sum()),
    "séquences non consécutives": int(unit_pairs.term_seq.sub(unit_pairs.prior_seq).ne(1).sum()),
}, name="paires (un cas peut cumuler deux raisons)")
display(pd.DataFrame([{"paires_candidates": len(unit_pairs), "admissibles": len(eligible_pairs),
                       "exclues": len(unit_pairs) - len(eligible_pairs)}]))
display(exclusion_reasons)

assert eligible_pairs["term_seq"].sub(eligible_pairs["prior_seq"]).eq(1).all()
np.testing.assert_allclose(exact_bridge(eligible_pairs.face_growth, eligible_pairs.prior_gap,
                                        eligible_pairs.new_gap, eligible_pairs.interval_years),
                           eligible_pairs.effective_growth, atol=1e-12)
print("Identité contractuel / effectif vérifiée sur toutes les paires admissibles.")

intervalle ≤ 0,5 an           62
intervalle ≥ 2,5 ans           0
séquences non consécutives     0
Name: paires (un cas peut cumuler deux raisons), dtype: int64

Identité contractuel / effectif vérifiée sur toutes les paires admissibles.

annual_growth = eligible_pairs.groupby("event_year").agg(
    pairs=("effective_growth", "size"),
    face_pct=("face_growth", lambda growth: PERCENT * growth.mean()),
    effective_pct=("effective_growth", lambda growth: PERCENT * growth.mean()),
    median_effective_pct=("effective_growth", lambda growth: PERCENT * growth.median()),
    renewal_share=("event_type", lambda event: event.eq("renewal").mean())).reset_index()
naive_median_change = validated_leases.groupby(validated_leases.sLeaseFrom.dt.year).sRent.median().pct_change() * PERCENT
annual_growth["naive_median_face_change_pct"] = annual_growth.event_year.map(naive_median_change)
display(annual_growth.round(3))

ax = annual_growth.set_index("event_year")[["naive_median_face_change_pct", "face_pct", "effective_pct"]].rename(columns={
    "naive_median_face_change_pct": "Médiane naïve (tous les baux)", "face_pct": "Même unité, contractuel",
    "effective_pct": "Même unité, effectif"}).plot(marker="o", figsize=(9, 4))
ax.set(title="Médiane naïve vs croissance à unité identique", ylabel="% annualisé / variation annuelle", xlabel="Année de début du nouveau bail")
ax.axhline(0, color="grey", lw=0.8)
plt.tight_layout()
plt.show()

conc = (leases[leases.year.between(FIRST_PAIR_YEAR_SHOWN, LAST_OBSERVED_YEAR)]
        .groupby("year")
        .agg(leases=("sRent", "size"),
             pct_with_concession=("sConcession", lambda flag: flag.mean() * PERCENT),
             median_contracted=("sRent", "median"),
             median_effective=("sRentEffective", "median")))
conc["gap_pct"] = ((conc.median_effective / conc.median_contracted - 1) * PERCENT)
conc.round(2)

# La concession change-t-elle l'évolution, ou seulement le niveau?
eff = same_unit_growth(leases, price_col="sRentEffective")
compare = pd.DataFrame({
    "contracted": honest["median"],
    "effective": eff[eff.year.between(FIRST_PAIR_YEAR_SHOWN, LAST_OBSERVED_YEAR)].groupby("year").growth_pct.median(),
}).round(2)
compare

RENT_CREDIT_CODES = ("promopay", "freerent")
AMENITY_CODES = ("freepark", "freelock", "freeappl")


def match_concessions_to_leases(lease_table, concession_table):
    # Rattache chaque écriture au bail de la même unité qui contient sa date de début.
    lease_rows = validate_leases(lease_table).reset_index(drop=True).reset_index(names="lease_id")
    ledger = normalise_unit_key(concession_table).reset_index(names="concession_id")
    for column in ("sDateFrom", "sDateTo"):
        ledger[column] = pd.to_datetime(ledger[column])
    matched = ledger.merge(lease_rows[["lease_id", *UNIT_KEY, "sLeaseFrom", "sLeaseTo"]], on=UNIT_KEY, how="left")
    matched = matched[matched.sDateFrom.between(matched.sLeaseFrom, matched.sLeaseTo)].copy()
    if matched.groupby("concession_id").lease_id.nunique().gt(1).any():
        raise DataContractError("une écriture de concession correspond à plusieurs baux")
    charge_code = matched.sChargeCode.str.lower()
    matched["code"] = charge_code
    matched["category"] = np.select([charge_code.isin(RENT_CREDIT_CODES), charge_code.isin(AMENITY_CODES)],
                                    ["rent", "amenity"], default="other")
    matched["total_credit"] = -matched.sAmount * matched.sMonths
    if (matched.total_credit < 0).any():
        raise DataContractError("une écriture de concession est une charge positive")
    return lease_rows, matched


lease_rows, matched_concessions = match_concessions_to_leases(leases, concessions)
rent_credit_by_lease = matched_concessions[matched_concessions.category.eq("rent")].groupby("lease_id").total_credit.sum()
entries_per_lease = matched_concessions.groupby("lease_id").size()
lease_rows["matched_entries"] = lease_rows.lease_id.map(entries_per_lease).fillna(0)
lease_rows["rent_only"] = lease_rows.sRent - lease_rows.lease_id.map(rent_credit_by_lease).fillna(0) / lease_rows.sTermMonths
lease_rows["flag_consistent"] = ~(lease_rows.sConcession.eq(1) & lease_rows.matched_entries.eq(0))
by_unit_sequence = lease_rows.sort_values(UNIT_KEY + ["sTermSeq"]).groupby(UNIT_KEY)
lease_rows["prior_rent_only"] = by_unit_sequence.rent_only.shift()
lease_rows["prior_flag_consistent"] = by_unit_sequence.flag_consistent.shift(fill_value=False)

concession_reconciliation = pd.DataFrame([{
    "écritures": len(concessions),
    "rattachées à un bail": matched_concessions.concession_id.nunique(),
    "non rattachées": len(concessions) - matched_concessions.concession_id.nunique(),
    "baux marqués sans écriture": int((lease_rows.sConcession.eq(1) & lease_rows.matched_entries.eq(0)).sum()),
    "baux non marqués avec écriture": int((lease_rows.sConcession.eq(0) & lease_rows.matched_entries.gt(0)).sum())}])
display(concession_reconciliation)

pairs_with_reconstruction = eligible_pairs.merge(
    lease_rows[[*UNIT_KEY, "sTermSeq", "rent_only", "prior_rent_only", "flag_consistent", "prior_flag_consistent"]],
    left_on=["property_code", "unit_code", "term_seq"], right_on=[*UNIT_KEY, "sTermSeq"], validate="one_to_one")
common_sample = pairs_with_reconstruction[pairs_with_reconstruction.flag_consistent & pairs_with_reconstruction.prior_flag_consistent
                                          & pairs_with_reconstruction.rent_only.gt(0)
                                          & pairs_with_reconstruction.prior_rent_only.gt(0)].copy()
common_sample["rent_only_pct"] = PERCENT * ((common_sample.rent_only / common_sample.prior_rent_only) ** (1 / common_sample.interval_years) - 1)
common_sample_by_year = common_sample.groupby("event_year").agg(
    n=("effective_growth", "size"), face_pct=("face_growth", lambda growth: PERCENT * growth.mean()),
    rent_only_pct=("rent_only_pct", "mean"),
    supplied_effective_pct=("effective_growth", lambda growth: PERCENT * growth.mean())).reset_index()
display(common_sample_by_year.round(3))
ax = common_sample_by_year.set_index("event_year")[["face_pct", "rent_only_pct", "supplied_effective_pct"]].rename(columns={
    "face_pct": "Contractuel", "rent_only_pct": "Contractuel − crédits locatifs du registre",
    "supplied_effective_pct": "Effectif fourni"}).plot(marker="o", figsize=(9, 4))
ax.set(title="Même échantillon : effet de la définition du loyer", ylabel="Croissance annualisée (%)", xlabel="Année de début du nouveau bail")
plt.tight_layout()
plt.show()

def concession_divergence(pairs_table):
    # Par année : croissance contractuelle vs effective, et variation de l'écart = incidence + profondeur.
    rows = []
    for year, year_pairs in pairs_table[pairs_table["eligible"]].groupby("event_year"):
        incidence_prior, incidence_new = float((year_pairs["prior_gap"] > 0).mean()), float((year_pairs["new_gap"] > 0).mean())
        depth_prior = float(year_pairs.loc[year_pairs["prior_gap"] > 0, "prior_gap"].mean()) if incidence_prior else 0.0
        depth_new = float(year_pairs.loc[year_pairs["new_gap"] > 0, "new_gap"].mean()) if incidence_new else 0.0
        gap_prior, gap_new = float(year_pairs["prior_gap"].mean()), float(year_pairs["new_gap"].mean())
        incidence_part = (incidence_new - incidence_prior) * (depth_prior + depth_new) / 2.0
        depth_part = (depth_new - depth_prior) * (incidence_prior + incidence_new) / 2.0
        rows.append({
            "event_year": int(year), "pairs": len(year_pairs),
            "face_pct": PERCENT * year_pairs["face_growth"].mean(), "effective_pct": PERCENT * year_pairs["effective_growth"].mean(),
            "face_minus_effective_pp": PERCENT * (year_pairs["face_growth"].mean() - year_pairs["effective_growth"].mean()),
            "prior_gap_pct": PERCENT * gap_prior, "new_gap_pct": PERCENT * gap_new, "gap_change_pp": PERCENT * (gap_new - gap_prior),
            "incidence_prior": incidence_prior, "incidence_new": incidence_new,
            "depth_prior_pct": PERCENT * depth_prior, "depth_new_pct": PERCENT * depth_new,
            "gap_change_from_incidence_pp": PERCENT * incidence_part, "gap_change_from_depth_pp": PERCENT * depth_part})
    return pd.DataFrame(rows)


def concession_code_contribution(lease_rows_table, matched_table):
    # Concession amortie moyenne en % du loyer contractuel, sur tous les baux, par année et par code.
    credit_by_code = matched_table.groupby(["lease_id", "code"])["total_credit"].sum().unstack(fill_value=0.0)
    codes = list(credit_by_code.columns)
    per_lease = lease_rows_table[["lease_id", "sLeaseFrom", "sTermMonths", "sRent", "sRentEffective"]].join(credit_by_code, on="lease_id")
    per_lease[codes] = per_lease[codes].fillna(0.0).div(per_lease["sTermMonths"], axis=0).div(per_lease["sRent"], axis=0)
    start_year = per_lease["sLeaseFrom"].dt.year
    contribution = per_lease.groupby(start_year)[codes].mean().mul(PERCENT)
    contribution["total_matched_pct"] = contribution[codes].sum(axis=1)
    contribution["supplied_gap_pct"] = PERCENT * (1 - per_lease["sRentEffective"] / per_lease["sRent"]).groupby(start_year).mean()
    contribution["unexplained_by_ledger_pct"] = contribution["supplied_gap_pct"] - contribution["total_matched_pct"]
    contribution.index.name = "lease_start_year"
    return contribution.reset_index()


divergence = concession_divergence(unit_pairs)
display(divergence.round(2))
fig, axes = plt.subplots(1, 2, figsize=(13, 4))
divergence.set_index("event_year")[["face_pct", "effective_pct"]].rename(
    columns={"face_pct": "Contractuel", "effective_pct": "Effectif"}).plot(ax=axes[0], marker="o", title="Contractuel vs effectif, même unité")
axes[0].set(ylabel="% annualisé", xlabel="Année de début du nouveau bail")
divergence.set_index("event_year")[["gap_change_from_incidence_pp", "gap_change_from_depth_pp"]].rename(
    columns={"gap_change_from_incidence_pp": "Incidence", "gap_change_from_depth_pp": "Profondeur"}).plot.bar(
    ax=axes[1], stacked=True, title="Variation de l'écart de concession : incidence + profondeur")
axes[1].set(ylabel="points de pourcentage", xlabel="Année de début du nouveau bail")
plt.tight_layout()
plt.show()
display(concession_code_contribution(lease_rows, matched_concessions).round(2))

FIRST_SPLIT_YEAR = 2022

split = (pairs[pairs.year.between(FIRST_SPLIT_YEAR, LAST_OBSERVED_YEAR)]
         .groupby(["year", "sRenewal"])
         .growth_pct.agg(pairs="size", median="median").round(2)
         .unstack("sRenewal"))
split.columns = [f"{stat}_{'renewal' if is_renewal else 'turnover'}" for stat, is_renewal in split.columns]
split

segments = eligible_pairs.groupby(["event_year", "province", "event_type"]).agg(
    n=("effective_growth", "size"),
    face_pct=("face_growth", lambda growth: PERCENT * growth.mean()),
    effective_pct=("effective_growth", lambda growth: PERCENT * growth.mean())).reset_index()
display(segments[segments.event_year.ge(REQUIRED_BACKTEST_YEARS[0])].round(3))

asking_known = asking.copy()
asking_known["sMonth"] = pd.to_datetime(asking_known.sMonth)
asking_known = asking_known[asking_known.sMonth.le(pd.Timestamp(LAST_OBSERVED_YEAR, 12, 31))]
latest_asking = asking_known.sort_values("sMonth").drop_duplicates(UNIT_KEY, keep="last")
asking_vs_listing = latest_asking.merge(listings[UNIT_KEY + ["sRent"]], on=UNIT_KEY, validate="one_to_one")
display(pd.DataFrame([{"origine": f"{LAST_OBSERVED_YEAR}-12-31", "unités avec loyer affiché": len(latest_asking),
                       "appariées aux listings": len(asking_vs_listing),
                       "ratio médian affiché / listing": float((asking_vs_listing.sAskingRent / asking_vs_listing.sRent).median())}]).round(4))

external = pd.read_csv("external_context.csv")
external_post_origin = pd.read_csv("external_postorigin_2026.csv")
EXTERNAL_REQUIRED_COLUMNS = {"target_year", "release_date", "source_url", "metric", "value"}


def origin_date(target_year):
    # Origine de la prévision : le 31 décembre de l'année précédant la cible.
    return pd.Timestamp(target_year - 1, SPEC.origin_month, SPEC.origin_day)


def external_asof(external_table, target_year):
    # Sources publiques disponibles à l'origine; refuse les publications postérieures ou sans provenance.
    if external_table is None:
        return pd.DataFrame()
    if not isinstance(external_table, pd.DataFrame):
        raise DataContractError("external doit être un DataFrame de sources publiques avec provenance")
    if EXTERNAL_REQUIRED_COLUMNS - set(external_table):
        raise DataContractError(f"external : colonnes manquantes {sorted(EXTERNAL_REQUIRED_COLUMNS - set(external_table))}")
    available = external_table[external_table.target_year.eq(target_year)].copy()
    release = pd.to_datetime(available.release_date, errors="coerce")
    if release.isna().any() or (release > origin_date(target_year)).any():
        raise DataContractError("publication manquante ou postérieure à l'origine de la prévision")
    if available.source_url.isna().any() or not available.source_url.astype(str).str.match(r"https?://\S+").all():
        raise DataContractError("provenance manquante")
    if not np.isfinite(pd.to_numeric(available.value, errors="coerce")).all():
        raise DataContractError("valeurs publiques non finies")
    return available


display(external_asof(external, TARGET_YEAR)[["geography", "metric", "reference_period", "release_date", "value", "unit"]])

@dataclass(frozen=True)
class OriginBundle:
    # Tout ce que la prévision a le droit de voir pour une année cible.
    target_year: int
    origin: pd.Timestamp
    pairs: pd.DataFrame     # paires admissibles signées et commencées avant l'origine
    roster: pd.DataFrame    # unités dont le dernier bail connu se termine pendant l'année cible

    @property
    def summary(self):
        return {"target_year": self.target_year, "origin": self.origin.date().isoformat(),
                "n_training_pairs": int(len(self.pairs)), "n_roster_units": int(len(self.roster)),
                "training_years": sorted(int(year) for year in self.pairs["event_year"].unique())}


ROSTER_COLUMNS = ["property_code", "unit_code", "province", "bedrooms", "term_seq", "lease_end", "face_rent", "effective_rent",
                  "prior_gap", "expected_years", "comparable"]


def build_roster(lease_table, target_year):
    # Unités dont le dernier bail connu à l'origine se termine pendant l'année cible.
    # Le filtre d'origine passe AVANT le choix du dernier bail : un extrait plus récent ne change pas le roster.
    checked = validate_leases(lease_table)
    origin = origin_date(target_year)
    known_at_origin = checked[(checked["sSignDate"] <= origin) & (checked["sLeaseFrom"] <= origin)]
    latest_lease = known_at_origin.sort_values(UNIT_KEY + ["sTermSeq", "sLeaseFrom"]).drop_duplicates(UNIT_KEY, keep="last")
    roster = latest_lease[latest_lease["sLeaseTo"].dt.year.eq(target_year)].rename(columns=COLUMN_RENAMES)
    roster["prior_gap"] = 1.0 - roster["effective_rent"] / roster["face_rent"]
    # Hypothèse : le prochain bail commence le lendemain de la fin du bail actuel.
    roster["expected_years"] = ((roster["lease_end"] + pd.Timedelta(days=1)) - roster["lease_start"]).dt.days / DAYS_PER_YEAR
    roster["comparable"] = roster["expected_years"].between(SPEC.gap_min_years, SPEC.gap_max_years, inclusive="neither")
    return roster[ROSTER_COLUMNS].reset_index(drop=True)


def make_bundle(lease_table, target_year):
    # Filtre signature ET début avant d'apparier : une séquence manquante ne peut pas être enjambée.
    checked = validate_leases(lease_table)
    origin = origin_date(target_year)
    observed = checked[(checked["sSignDate"] <= origin) & (checked["sLeaseFrom"] <= origin)]
    observed_pairs = build_pairs(observed)
    known_pairs = observed_pairs[observed_pairs["eligible"]].copy()
    roster = build_roster(checked, target_year)
    if known_pairs.empty:
        raise DataContractError(f"aucune paire admissible connue avant le {origin.date()}")
    if roster.empty or not roster["comparable"].any():
        raise DataContractError(f"aucune unité expirant en {target_year} n'est connue au {origin.date()}")
    return OriginBundle(target_year, origin, known_pairs, roster)


def clip_for_training(values, column):
    # Bornes de robustesse : écart de concession prévu, ou croissance.
    lower, upper = SPEC.gap_clip if column == "new_gap" else SPEC.growth_clip
    return values.clip(lower, upper)


def shrunk_cell_means(pairs_table, column, shrink_k):
    # Moyennes immeuble × événement rétrécies vers province × événement (puis événement, puis portefeuille).
    if pairs_table.empty:
        return {}
    bounded = pairs_table.assign(_value=clip_for_training(pairs_table[column], column))
    portfolio_mean = float(bounded["_value"].mean())
    event_means = bounded.groupby("event_type")["_value"].mean().to_dict()
    province_event_means = bounded.groupby(["province", "event_type"])["_value"].mean()
    province_of = bounded.groupby("property_code")["province"].first().to_dict()
    cell_means = {}
    for (property_code, event_type), cell in bounded.groupby(["property_code", "event_type"]):
        parent = province_event_means.get((province_of[property_code], event_type), event_means.get(event_type, portfolio_mean))
        n_pairs = len(cell)
        cell_means[(property_code, event_type)] = (n_pairs * float(cell["_value"].mean()) + shrink_k * parent) / (n_pairs + shrink_k)
    return cell_means


def cell_value(cell_means, pairs_table, column, property_code, event_type, province):
    # Valeur de la cellule, avec repli hiérarchique pour une cellule absente.
    if (property_code, event_type) in cell_means:
        return cell_means[(property_code, event_type)]
    bounded = pairs_table.assign(_value=clip_for_training(pairs_table[column], column))
    province_event = bounded[bounded["province"].eq(province) & bounded["event_type"].eq(event_type)]["_value"]
    if len(province_event):
        return float(province_event.mean())
    same_event = bounded[bounded["event_type"].eq(event_type)]["_value"]
    return float(same_event.mean()) if len(same_event) else float(bounded["_value"].mean())


def event_mix(bundle):
    # Part attendue de renouvellements / relocations par immeuble, rétrécie vers la province.
    roster = bundle.roster[bundle.roster["comparable"]]
    recent = bundle.pairs[bundle.pairs["event_year"].between(bundle.target_year - SPEC.trailing_years_mix, bundle.target_year - 1)]
    overall_rate = (float(recent["event_type"].eq("renewal").mean()) if len(recent)
                    else float(bundle.pairs["event_type"].eq("renewal").mean()))
    rows = []
    for property_code, units in roster.groupby("property_code"):
        province = units["province"].iloc[0]
        property_recent = recent[recent["property_code"].eq(property_code)]
        province_recent = recent[recent["province"].eq(province)]
        prior_rate = float(province_recent["event_type"].eq("renewal").mean()) if len(province_recent) else overall_rate
        renewal_rate = ((property_recent["event_type"].eq("renewal").sum() + SPEC.shrink_k_mix * prior_rate)
                        / (len(property_recent) + SPEC.shrink_k_mix))
        for event_type, share in (("renewal", renewal_rate), ("turnover", 1.0 - renewal_rate)):
            rows.append({"property_code": property_code, "province": province, "event_type": event_type,
                         "roster_units": len(units), "event_share": float(share), "expected_units": len(units) * float(share),
                         "roster_prior_gap": float(units["prior_gap"].mean()), "n_recent_pairs": int(len(property_recent))})
    mix_table = pd.DataFrame(rows)
    mix_table["weight"] = mix_table["expected_units"] / mix_table["expected_units"].sum()
    return mix_table


def roster_bridge(bundle, property_code, face_growth, new_gap):
    # Moyenne des identités unité par unité du roster d'un immeuble (prochain début = fin actuelle + 1 jour).
    units = bundle.roster[bundle.roster["property_code"].eq(property_code) & bundle.roster["comparable"]]
    return float(np.mean(exact_bridge(face_growth, units["prior_gap"], new_gap, units["expected_years"])))

def component_cells(bundle):
    # DIRECT, BRIDGE, ANCHOR et contractuel pour chaque immeuble × événement.
    known_pairs, target_year = bundle.pairs, bundle.target_year
    last_year = known_pairs[known_pairs["event_year"].eq(target_year - 1)]
    if last_year.empty:   # aucun bail commencé l'an dernier : année la plus récente disponible
        last_year = known_pairs[known_pairs["event_year"].eq(known_pairs["event_year"].max())]
    weights = event_mix(bundle)
    direct = shrunk_cell_means(last_year, "effective_growth", SPEC.shrink_k_cell)
    face_last = shrunk_cell_means(last_year, "face_growth", SPEC.shrink_k_cell)
    gap_last = shrunk_cell_means(last_year, "new_gap", SPEC.shrink_k_cell)
    anchor = shrunk_cell_means(known_pairs, "effective_growth", SPEC.shrink_k_cell)
    face_anchor = shrunk_cell_means(known_pairs, "face_growth", SPEC.shrink_k_cell)
    rows = []
    for _, cell in weights.iterrows():
        key = (cell["property_code"], cell["event_type"], cell["province"])
        face_persistence = cell_value(face_last, last_year, "face_growth", *key)
        new_gap_persistence = cell_value(gap_last, last_year, "new_gap", *key)
        rows.append({**cell.to_dict(),
                     "DIRECT": cell_value(direct, last_year, "effective_growth", *key),
                     "BRIDGE": roster_bridge(bundle, cell["property_code"], face_persistence, new_gap_persistence),
                     "ANCHOR": cell_value(anchor, known_pairs, "effective_growth", *key),
                     "face_persistence": face_persistence, "face_anchor": cell_value(face_anchor, known_pairs, "face_growth", *key),
                     "new_gap_persistence": new_gap_persistence,
                     "n_last_year_pairs": int((last_year["property_code"].eq(cell["property_code"])
                                               & last_year["event_type"].eq(cell["event_type"])).sum())})
    return pd.DataFrame(rows)


def combine(cells, components=None, weights=None):
    # Combinaison à poids fixés : effectif = moyenne des composantes; contractuel = mélange dernière année / historique.
    components = SPEC.components if components is None else components
    weights = SPEC.component_weights if weights is None else weights
    if (not components or len(components) != len(weights) or not np.isfinite(weights).all() or min(weights) < 0
            or abs(sum(weights) - 1.0) > 1e-9 or len(set(components)) != len(components)
            or not set(components).issubset(SPEC.components)):
        raise DataContractError("les poids doivent correspondre aux composantes et sommer à 1")
    combined = cells.copy()
    combined["effective"] = sum(weight * combined[name] for name, weight in zip(components, weights))
    combined["face"] = (SPEC.face_persistence_weight * combined["face_persistence"]
                        + (1 - SPEC.face_persistence_weight) * combined["face_anchor"])
    return combined


def aggregate(cells, column="effective"):
    # Moyenne pondérée par le poids attendu de chaque cellule.
    return float(np.average(cells[column], weights=cells["weight"]))


def forecast(bundle, components=None, weights=None):
    cells = combine(component_cells(bundle), components, weights)
    return {"target_year": bundle.target_year, "origin": bundle.origin.date().isoformat(),
            "effective": aggregate(cells, "effective"), "face": aggregate(cells, "face"),
            "components": {name: aggregate(cells, name) for name in SPEC.components},
            "renewal_share": float(cells.loc[cells["event_type"].eq("renewal"), "weight"].sum()),
            "roster_units": int(bundle.roster["comparable"].sum()),
            "roster_excluded": int((~bundle.roster["comparable"]).sum()),
            "cells": cells, "bundle": bundle.summary}