1"""H5 validation of 'listing exit' as a sale proxy (no transaction data): cross-source2consistency of exit timing for the same VIN (dealer website vs otogo portal), share of3exits followed by a relisting of the same VIN at another dealer within 30 days4(lower bound on non-sale exits), and dealer-to-dealer transfers.5"""6from __future__ import annotations78import json910import numpy as np11import polars as pl1213from . import config as C1415OUT = C.RESULTS / "validation"16RELIABLE = {"direct_dealer", "full_portal", "banner"}171819def run() -> dict:20 OUT.mkdir(parents=True, exist_ok=True)21 L = pl.read_parquet(C.INTERIM / "listings_full.parquet").filter(pl.col("vin_ok") & pl.col("source_class").is_in(list(RELIABLE)) & pl.col("f_source_reliable") & (pl.col("kind") == "auto"))22 out = {"n_reliable_vin_listings": L.height}23 # --- 1. same VIN on a dealer website (or banner) AND on otogo ---------------------------24 direct = L.filter(pl.col("source_class").is_in(["direct_dealer", "banner"])).select(["vin_hash", "listing_id", "dealer_id", "entry_ts", "end_ts", "exit_confirmed", "active", "last_seen_ts"]).rename({c: c + "_d" for c in ["listing_id", "dealer_id", "entry_ts", "end_ts", "exit_confirmed", "active", "last_seen_ts"]})25 otogo = L.filter(pl.col("source") == "otogo").select(["vin_hash", "listing_id", "dealer_id", "entry_ts", "end_ts", "exit_confirmed", "active", "last_seen_ts"]).rename({c: c + "_o" for c in ["listing_id", "dealer_id", "entry_ts", "end_ts", "exit_confirmed", "active", "last_seen_ts"]})26 P = direct.join(otogo, on="vin_hash", how="inner")27 # overlapping spells only28 P = P.filter((pl.col("entry_ts_o") < pl.col("end_ts_d")) & (pl.col("entry_ts_d") < pl.col("end_ts_o")))29 both = P.filter(pl.col("exit_confirmed_d") & pl.col("exit_confirmed_o")).with_columns(((pl.col("end_ts_d") - pl.col("end_ts_o")) / 86400).alias("dd"))30 out["pairs_overlapping"] = P.height31 out["pairs_both_exited"] = both.height32 for k in (1, 3, 7):33 out[f"both_exited_within_{k}d"] = float((both["dd"].abs() <= k).mean()) if both.height else None34 out["median_abs_exit_gap_days"] = float(both["dd"].abs().median()) if both.height else None35 # direct exited (confirmed) while otogo copy still active >= 7 days later36 dx = P.filter(pl.col("exit_confirmed_d"))37 still = dx.filter((pl.col("end_ts_o") - pl.col("end_ts_d")) >= 7 * 86400)38 out["direct_exits_with_pair"] = dx.height39 out["direct_exit_but_otogo_still_listed_7d"] = float(still.height / dx.height) if dx.height else None40 ox = P.filter(pl.col("exit_confirmed_o"))41 still2 = ox.filter((pl.col("end_ts_d") - pl.col("end_ts_o")) >= 7 * 86400)42 out["otogo_exits_with_pair"] = ox.height43 out["otogo_exit_but_direct_still_listed_7d"] = float(still2.height / ox.height) if ox.height else None44 # --- 2. relisting of the same VIN after an exit (any reliable source, other listing) ------45 # "other dealer" = new listing located > 5 km away (dealer_id differs across sources for the same physical dealer)46 X = L.filter(pl.col("exit_confirmed")).select(["vin_hash", "listing_id", "dealer_id", "end_ts", "s2_dynamic", "lat", "lng"])47 A = L.select(["vin_hash", "listing_id", "dealer_id", "entry_ts", "lat", "lng"]).rename({"listing_id": "lid2", "dealer_id": "dealer2", "entry_ts": "entry2", "lat": "lat2", "lng": "lng2"})48 J = X.join(A, on="vin_hash", how="inner").filter(pl.col("lid2") != pl.col("listing_id")).filter(pl.col("entry2") > pl.col("end_ts") - 86400)49 dist = (((pl.col("lat") - pl.col("lat2")) * 111.0) ** 2 + ((pl.col("lng") - pl.col("lng2")) * 111.0 * (pl.col("lat") * 3.14159 / 180).cos()) ** 2).sqrt()50 J = J.with_columns(((pl.col("entry2") - pl.col("end_ts")) / 86400).alias("gap"), (dist > 5.0).fill_null(False).alias("other_dealer"))51 rel = J.filter(pl.col("gap") <= 30).group_by("listing_id").agg(pl.col("other_dealer").max().alias("relisted_other"), pl.lit(1).alias("relisted_any"))52 Xr = X.join(rel, on="listing_id", how="left").with_columns(pl.col("relisted_any").fill_null(0), pl.col("relisted_other").fill_null(False))53 for name, sub in [("all_exits", Xr), ("S2_exits", Xr.filter(pl.col("s2_dynamic")))]:54 out[f"{name}_n"] = sub.height55 out[f"{name}_relisted_any_30d"] = float(sub["relisted_any"].mean())56 out[f"{name}_relisted_other_dealer_30d"] = float(sub["relisted_other"].cast(pl.Int8).mean())57 # exits observable for 30 days only (end_ts <= snapshot - 30 d)58 obs = Xr.filter(pl.col("end_ts") <= C.SNAPSHOT_TS - 30 * 86400)59 out["exits_observable_30d_n"] = obs.height60 out["exits_observable_30d_relisted_any"] = float(obs["relisted_any"].mean())61 out["exits_observable_30d_relisted_other_dealer"] = float(obs["relisted_other"].cast(pl.Int8).mean())62 # gap distribution for relistings at another dealer (dealer-to-dealer transfers)63 g = J.filter(pl.col("other_dealer") & (pl.col("gap") <= 30))["gap"]64 out["transfer_gap_days_median"] = float(g.median()) if g.len() else None65 out["transfer_gap_days_q75"] = float(g.quantile(0.75)) if g.len() else None66 # VINs seen at >= 2 distinct dealers over the window (any time)67 md = L.filter(pl.col("lat").is_not_null()).group_by("vin_hash").agg([pl.col("lat").max().alias("la1"), pl.col("lat").min().alias("la0"), pl.col("lng").max().alias("lo1"), pl.col("lng").min().alias("lo0")])68 md = md.with_columns((((pl.col("la1") - pl.col("la0")) * 111.0) ** 2 + ((pl.col("lo1") - pl.col("lo0")) * 78.0) ** 2).sqrt().alias("span_km"))69 out["vins_total"] = md.height; out["vins_at_2plus_locations_5km"] = int((md["span_km"] > 5).sum())70 json.dump(out, open(OUT / "exit_validation.json", "w"), indent=2)71 return out727374if __name__ == "__main__":75 print(json.dumps(run(), indent=2))76