1#!/usr/bin/env python32# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca3"""Phase 6 — prospective evaluation set: sales present in the *live* Vrai Prix database but absent from the42026-10-21 snapshot (registered after the snapshot, hence after the protocol freeze), joined to their roll unit.56Run only AFTER `av freeze`. Reads the live SQLite over SSH (read-only), never modifies it.7Output: data/raw/prospective_<date>.parquet (raw columns, same vocabulary as the snapshot) + a manifest.8"""9from __future__ import annotations1011import json12import subprocess13import sys14from datetime import date15from pathlib import Path1617import polars as pl1819sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))20from agentic_valuation.config import RAW, RESULTS # noqa: E4022122NODE = "M3U96a"23DB = "~/apps/vrai-prix/data/vraiprix.db"24SINCE = "2026-07-01" # registration lag is 2-8 weeks; anything older is already in the snapshot252627def ssh_query(sql: str) -> str:28 cmd = ["ssh", "-o", "ConnectTimeout=20", NODE, f"sqlite3 -json -readonly {DB} \"{sql}\""]29 return subprocess.run(cmd, capture_output=True, text=True, check=True).stdout303132def main() -> None:33 freeze = RESULTS / "freeze.json"34 if not freeze.exists():35 sys.exit("results/freeze.json not found: run `av freeze` before fetching the prospective set")36 snap = pl.read_parquet(RAW / "wp22_clean_transactions.parquet", columns=["tx_id"])37 live = pl.DataFrame(json.loads(ssh_query(f"select * from transactions where date >= '{SINCE}'") or "[]"))38 new = live.filter(~pl.col("id").is_in(snap["tx_id"].implode()))39 ids = [i for i in new["id_provinc"].drop_nulls().unique().to_list() if i]40 units = pl.DataFrame()41 if ids:42 chunks = [ids[i : i + 500] for i in range(0, len(ids), 500)]43 parts = []44 for ch in chunks:45 inl = ",".join(f"'{i}'" for i in ch)46 parts.append(pl.DataFrame(json.loads(ssh_query(f"select * from units where id_provinc in ({inl})") or "[]")))47 units = pl.concat([p for p in parts if p.height], how="diagonal_relaxed") if any(p.height for p in parts) else pl.DataFrame()48 stamp = date.today().isoformat()49 out = RAW / f"prospective_{stamp}.parquet"50 new.write_parquet(out)51 if units.height:52 units.write_parquet(RAW / f"prospective_units_{stamp}.parquet")53 man = {"fetched": stamp, "node": NODE, "since": SINCE, "live_rows_since": live.height, "new_rows": new.height, "new_date_min": str(new["date"].min()) if new.height else None, "new_date_max": str(new["date"].max()) if new.height else None, "units_rows": units.height, "freeze_hash": json.loads(freeze.read_text()).get("freeze_digest")}54 (RAW / f"prospective_{stamp}.manifest.json").write_text(json.dumps(man, indent=1))55 print(man)565758if __name__ == "__main__":59 main()60