"""CLEARLY LABELLED SYNTHETIC TEST FIXTURE — not empirical data. Generates a small SQLite file with the Auto-Ka schema (vehicles, price_log, sync_log, dealers) so that the pipeline and tests run without the restricted snapshot. Values are random; nothing here is a finding. """ from __future__ import annotations import json import random import sqlite3 from pathlib import Path SCHEMA = """ CREATE TABLE vehicles (uid TEXT PRIMARY KEY, source TEXT NOT NULL, external_id TEXT NOT NULL, url TEXT, title TEXT, make TEXT, model TEXT, trim TEXT, year INTEGER, price REAL, price_label TEXT, mileage_km REAL, mileage_label TEXT, transmission TEXT, fuel TEXT, drivetrain TEXT, body_type TEXT, exterior_color TEXT, interior_color TEXT, engine TEXT, doors INTEGER, seats INTEGER, vin TEXT, stock_number TEXT, dealer_name TEXT, city TEXT, region TEXT, description TEXT, features TEXT, details TEXT, images TEXT, carfax_url TEXT, content_hash TEXT, first_seen REAL, last_seen REAL, updated_at REAL, miss_count INTEGER DEFAULT 0, active INTEGER DEFAULT 1, kind TEXT DEFAULT 'auto', lat REAL, lng REAL, dup_of TEXT, dup_sources TEXT); CREATE TABLE sync_log (id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT, ts REAL, found INTEGER, added INTEGER, updated INTEGER, removed INTEGER, ok INTEGER, message TEXT, stats TEXT); CREATE TABLE price_log (uid TEXT NOT NULL, ts REAL NOT NULL, price REAL); CREATE TABLE dealers (source TEXT PRIMARY KEY, name TEXT, address TEXT, phone TEXT, website TEXT, city TEXT, lat REAL, lng REAL, fetched_at REAL); CREATE TABLE vin_decode (vin TEXT PRIMARY KEY, data TEXT, fetched_at REAL); """ T0 = 1786537835.0 # same window start as the real panel (for code paths only) def make(path: Path, n_listings: int = 3000, seed: int = 36) -> Path: rng = random.Random(seed) path.parent.mkdir(parents=True, exist_ok=True) if path.exists(): path.unlink() con = sqlite3.connect(path) con.executescript(SCHEMA) makes = {"Honda": ["Civic", "CR-V"], "Toyota": ["Corolla", "RAV4"], "Ford": ["F-150", "Escape"], "Hyundai": ["Elantra", "Tucson"]} regions = {"Montréal": (45.5, -73.6), "Capitale-Nationale": (46.8, -71.2), "Outaouais": (45.4, -75.7), "Estrie": (45.4, -71.9)} sources = [f"synthdealer{i:02d}" for i in range(20)] + ["otogo"] dealer_geo = {} for s in sources[:-1]: reg = rng.choice(list(regions)) lat, lng = regions[reg] dealer_geo[s] = (reg, lat + rng.uniform(-0.3, 0.3), lng + rng.uniform(-0.3, 0.3)) con.execute("INSERT INTO dealers VALUES (?,?,?,?,?,?,?,?,?)", (s, s, "", "", "", reg, dealer_geo[s][1], dealer_geo[s][2], T0)) for s in sources: for c in range(585): con.execute("INSERT INTO sync_log (source, ts, found, added, updated, removed, ok, message, stats) VALUES (?,?,?,?,?,?,?,?,?)", (s, T0 + c * 3.3 * 3600, 100, 5, 10, 4, 1, "ok", json.dumps({"null_price_rate": 0.0}))) end = T0 + 81 * 86400 for i in range(n_listings): src = rng.choice(sources) if src == "otogo": d = rng.choice(sources[:-1]); reg, lat, lng = dealer_geo[d]; dealer_name = d.upper() else: reg, lat, lng = dealer_geo[src]; dealer_name = src mk = rng.choice(list(makes)); md = rng.choice(makes[mk]); yr = rng.randint(2012, 2025) km = max(2000, rng.gauss(18000 * (2026 - yr), 20000)) price = max(3000, 38000 * (0.86 ** (2026 - yr)) * (1 - 0.000002 * km) * rng.lognormvariate(0, 0.12)) entry = T0 + rng.uniform(0, 75 * 86400) dur = rng.expovariate(1 / 20) * 86400 exit_ts = entry + dur active = 1 if exit_ts > end - 2 * 86400 else 0 last_seen = min(exit_ts, end) vin = "".join(rng.choice("ABCDEFGHJKLMNPRSTUVWXYZ0123456789") for _ in range(17)) uid = f"{src}:{i}" con.execute("""INSERT INTO vehicles (uid, source, external_id, title, make, model, trim, year, price, mileage_km, transmission, fuel, drivetrain, body_type, vin, dealer_name, city, region, first_seen, last_seen, updated_at, miss_count, active, kind, lat, lng) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", (uid, src, str(i), f"{mk} {md} {yr}", mk, md, "LX", yr, round(price), round(km), "Automatique", "Essence", "Traction", "VUS" if md in {"CR-V", "RAV4", "Escape", "Tucson"} else "Berline", vin, dealer_name, "Ville", reg, entry, last_seen, last_seen if active == 0 else entry, 0, active, "auto", lat, lng)) p = price con.execute("INSERT INTO price_log VALUES (?,?,?)", (uid, entry, round(p))) t = entry while rng.random() < 0.35 and t + 5 * 86400 < last_seen: t += rng.uniform(5, 15) * 86400 p *= 1 - rng.uniform(0.01, 0.06) con.execute("INSERT INTO price_log VALUES (?,?,?)", (uid, t, round(p))) con.execute("UPDATE vehicles SET price=? WHERE uid=?", (round(p), uid)) con.commit(); con.close() (path.parent / "README.md").write_text("# SYNTHETIC FIXTURE\n\nRandomly generated data with the Auto-Ka schema for tests only. " "No empirical content. Generated by `ucml.synthetic.make`.\n") return path if __name__ == "__main__": from . import config as C print(make(C.SYNTHETIC / "autoka-synthetic.db"))