1"""Sustained asking-price paths (DECISIONS.md D-005).23Input: the platform's price_log (one row at first observation and one per price4change; NULL = price withdrawn). Output: price levels with transient A->B->A5reversions removed, plus event classification (entry / cut / rise).6"""7from __future__ import annotations89from dataclasses import dataclass1011import numpy as np12import polars as pl1314from . import config as C151617@dataclass18class PathSummary:19 levels: list[tuple[float, float]] # (ts, price) sustained levels20 n_transients: int21 n_null: int222324def sustain(ts: list[float], price: list[float | None], last_seen: float,25 transient_seconds: float = C.TRANSIENT_SECONDS) -> PathSummary:26 """Remove transient levels: a level k with duration < transient_seconds whose27 successor returns to the price of level k-1 is dropped (and k-1, k+1 merged).28 Repeated until stable. NULL prices are dropped (counted)."""29 pts = [(t, p) for t, p in zip(ts, price) if p is not None and np.isfinite(p)]30 n_null = len(ts) - len(pts)31 pts.sort()32 # merge consecutive equal prices33 lv: list[tuple[float, float]] = []34 for t, p in pts:35 if lv and lv[-1][1] == p:36 continue37 lv.append((t, p))38 n_tr = 039 changed = True40 while changed and len(lv) >= 3:41 changed = False42 for k in range(1, len(lv) - 1):43 dur = lv[k + 1][0] - lv[k][0]44 if dur < transient_seconds and lv[k + 1][1] == lv[k - 1][1]:45 # drop level k and the (now redundant) level k+146 del lv[k:k + 2]47 n_tr += 148 changed = True49 break50 return PathSummary(levels=lv, n_transients=n_tr, n_null=n_null)515253def classify(levels: list[tuple[float, float]], min_pct: float = C.MARKDOWN_MIN_PCT,54 min_abs: float = C.MARKDOWN_MIN_ABS) -> list[dict]:55 """Events from sustained levels: first level = entry; then cut / rise / minor."""56 ev = []57 for k, (t, p) in enumerate(levels):58 if k == 0:59 ev.append({"ts": t, "price": p, "prev_price": None, "change_pct": None, "kind": "entry"})60 continue61 pp = levels[k - 1][1]62 ch = (p - pp) / pp if pp else None63 if ch is None:64 kind = "minor"65 elif ch <= -min_pct and (pp - p) >= min_abs:66 kind = "cut"67 elif ch >= min_pct and (p - pp) >= min_abs:68 kind = "rise"69 else:70 kind = "minor"71 ev.append({"ts": t, "price": p, "prev_price": pp, "change_pct": ch, "kind": kind})72 return ev737475def build(price_log: pl.DataFrame, vehicles: pl.DataFrame) -> tuple[pl.DataFrame, pl.DataFrame]:76 """Return (events, summary) DataFrames for all uids in price_log."""77 ls = dict(zip(vehicles["uid"].to_list(), vehicles["last_seen"].to_list()))78 rows_ev, rows_sum = [], []79 grouped = price_log.sort(["uid", "ts"]).group_by("uid", maintain_order=True).agg([pl.col("ts"), pl.col("price")])80 for uid, ts, pr in grouped.iter_rows():81 ps = sustain(ts, pr, ls.get(uid, max(ts)))82 ev = classify(ps.levels)83 cuts = [e for e in ev if e["kind"] == "cut"]84 rises = [e for e in ev if e["kind"] == "rise"]85 entry = ev[0] if ev else None86 for e in ev:87 rows_ev.append({"uid": uid, **e})88 rows_sum.append({89 "uid": uid,90 "entry_price": entry["price"] if entry else None,91 "entry_price_ts": entry["ts"] if entry else None,92 "last_price": ev[-1]["price"] if ev else None,93 "n_levels_raw": len(ts),94 "n_null_price": ps.n_null,95 "n_transients": ps.n_transients,96 "oscillating": ps.n_transients >= C.OSCILLATION_MAX_TRANSIENTS,97 "n_cuts": len(cuts),98 "n_rises": len(rises),99 "first_cut_ts": cuts[0]["ts"] if cuts else None,100 "first_cut_pct": cuts[0]["change_pct"] if cuts else None,101 "total_change_pct": ((ev[-1]["price"] - entry["price"]) / entry["price"]) if (entry and entry["price"]) else None,102 })103 events = pl.DataFrame(rows_ev, schema={"uid": pl.Utf8, "ts": pl.Float64, "price": pl.Float64,104 "prev_price": pl.Float64, "change_pct": pl.Float64, "kind": pl.Utf8})105 summary = pl.DataFrame(rows_sum)106 return events, summary107