# -*- coding: utf-8 -*-
"""01 — Jeu de données : transactions unifamiliales Vrai Prix ⋈ rôle, nettoyage journalisé,
reventes, géographie, périodes. Sorties : sales.parquet, exclusions.csv, dataset_stats.json."""
from common import *

con = connect()
log(f"lecture {DB}")
sql = """
select t.id, t.date, t.amount, t.lat, t.lng, t.city, t.property_type, t.year_built, t.floor_area,
       t.building_type, t.id_provinc, t.valeur_role as role_vente, t.land_area,
       u.code_mun, u.municipalite, u.type_prop as u_type, u.annee_construction, u.aire_etages_m2,
       u.superficie_terrain_m2, u.front_terrain_m, u.nb_etages, u.nb_logements, u.lien_physique,
       u.genre_construction, u.unite_voisinage, u.dat_cond_marche, u.valeur_terrain, u.valeur_batiment,
       u.valeur_role as role_actuel, u.valeur_anterieure as role_anterieur, u.cubf
from transactions t left join units u on u.id_provinc = t.id_provinc
where t.property_type = 'unifamilial'
"""
df = pd.read_sql(sql, con)
log(f"{len(df):,} transactions unifamiliales brutes")
n0 = len(df)
excl = []  # (règle, n)
def drop(mask, rule):
    global df
    n = int(mask.sum()); excl.append((rule, n)); df = df[~mask].copy(); log(f"  {rule}: -{n:,} → {len(df):,}")

df["date"] = pd.to_datetime(df["date"])
df = df.sort_values(["id_provinc", "date"]).reset_index(drop=True)
drop(df["id_provinc"].isna() | df["code_mun"].isna(), "R0 sans jumelage au rôle (id_provinc / code_mun)")
drop(df.duplicated(["id_provinc", "date"]), "R1 doublons (même unité, même date)")
drop(~df["u_type"].isin(["unifamilial"]) | (df["nb_logements"].fillna(1) > 1), "R2 unité du rôle non unifamiliale (chalet, plex, terrain…) ou > 1 logement")
drop(df["lat"].isna() | df["lng"].isna() | (df["lat"] < 44.5) | (df["lat"] > 63) | (df["lng"] > -56) | (df["lng"] < -80), "R3 coordonnées absentes ou hors Québec")
fa = df["floor_area"].fillna(df["aire_etages_m2"])
df["floor_area"] = fa
drop(fa.isna() | (fa < 40) | (fa > 600), "R4 aire habitable absente ou hors [40, 600] m²")
la = df["land_area"].fillna(df["superficie_terrain_m2"])
df["land_area"] = la
drop(la.isna() | (la < 80) | (la > 20000), "R5 superficie de terrain absente ou hors [80, 20 000] m²")
yb = df["year_built"].fillna(df["annee_construction"])
df["year_built"] = yb
drop(yb.isna() | (yb < 1850) | (yb > df["date"].dt.year), "R6 année de construction absente, < 1850 ou postérieure à la vente (vente de terrain)")
drop((df["year_built"] == df["date"].dt.year), "R6b vente l'année même de la construction (prix du terrain ou d'une construction inachevée : médiane 346 k$ et ratio au rôle 1,15 contre 450-480 k$ et 1,3 aux âges 1-4)")
drop((df["amount"] < 50000) | (df["amount"] > 3_000_000), "R7 prix hors [50 k$, 3 M$]")
ratio = df["amount"] / df["role_vente"]
drop(df["role_vente"].isna() | (df["role_vente"] <= 0) | (ratio < 0.33) | (ratio > 3.0), "R8 rôle absent ou ratio prix/rôle hors [0,33 ; 3] (transactions présumées non représentatives)")
# reventes rapides avec forte variation (flips / rénovations : les caractéristiques enregistrées ne décrivent plus le bien)
df = df.sort_values(["id_provinc", "date"]).reset_index(drop=True)
g = df.groupby("id_provinc")
prev_date = g["date"].shift(1); prev_amt = g["amount"].shift(1)
dd = (df["date"] - prev_date).dt.days
chg = df["amount"] / prev_amt - 1
flip = (dd <= 180) & (chg.abs() > 0.40)
drop(flip.fillna(False), "R9 revente ≤ 180 jours avec variation > 40 % (flip)")
drop(df["lien_physique"].isna() | df["genre_construction"].isna(), "R10 lien physique ou genre de construction absent au rôle")

# ----------------------------------------------------------------- variables dérivées
rep = load_repertoire()
df = df.merge(rep, on="code_mun", how="left")
drop(df["region"].isna(), "R11 municipalité absente du répertoire MAMH")
df["sale_year"] = df["date"].dt.year
df["month"] = month_index(df["date"])
df["day"] = (df["date"] - pd.Timestamp("2021-01-01")).dt.days.astype(int)
df["age"] = (df["sale_year"] - df["year_built"]).clip(lower=0)
df["ln_price"] = np.log(df["amount"])
df["ln_floor"] = np.log(df["floor_area"])
df["ln_land"] = np.log(df["land_area"])
df["lien"] = df["lien_physique"].map(LIEN).fillna(0).astype(int)
df["genre"] = df["genre_construction"].map(GENRE).fillna(0).astype(int)
df["btype"] = df["building_type"].map(BTYPE).fillna(-1).astype(int)
df["etages"] = df["nb_etages"].fillna(1).clip(1, 3)
df["x_km"], df["y_km"] = to_xy_km(df["lat"], df["lng"])
# millésime du rôle au moment de la vente (rôle courant / antérieur / autre)
df["role_vintage"] = np.where(np.isclose(df["role_vente"], df["role_actuel"].fillna(-1)), 2,
                      np.where(np.isclose(df["role_vente"], df["role_anterieur"].fillna(-1)), 1, 0))
df["ratio_role"] = df["amount"] / df["role_vente"]
df["part_terrain"] = (df["valeur_terrain"] / df["role_actuel"]).clip(0.02, 0.95).fillna(0.2)
df["neuf"] = (df["age"] <= 3).astype(int)
df["period"] = np.select([df["date"] <= TRAIN_END, df["date"] <= VAL_END], ["train", "val"], "test")
df = df.sort_values("date").reset_index(drop=True)
df["idx"] = np.arange(len(df), dtype=np.int32)

keep = ["idx", "id", "id_provinc", "date", "day", "month", "sale_year", "period", "amount", "ln_price",
        "lat", "lng", "x_km", "y_km", "code_mun", "mun_nom", "mrc", "mrc_nom", "region", "region_nom",
        "unite_voisinage", "floor_area", "ln_floor", "land_area", "ln_land", "year_built", "age", "neuf",
        "lien", "genre", "btype", "etages", "role_vente", "role_vintage", "ratio_role", "part_terrain"]
df = df[keep]
save_parquet(df, "sales.parquet")
pd.DataFrame(excl, columns=["regle", "n"]).to_csv(OUT / "exclusions.csv", index=False)
stats = {
    "brut": n0, "propre": len(df), "exclusions": excl,
    "par_periode": df["period"].value_counts().to_dict(),
    "par_annee": df["sale_year"].value_counts().sort_index().to_dict(),
    "par_region": df.groupby("region_nom").size().sort_values(ascending=False).to_dict(),
    "prix": {k: float(v) for k, v in df["amount"].describe().items()},
    "dates": [str(df["date"].min().date()), str(df["date"].max().date())],
}
save_json(stats, "dataset_stats.json")
log(f"jeu propre : {len(df):,} ventes ({stats['par_periode']})")