import dupehell
import polars as pl
import os, sys, json
from pathlib import Path
BASE = Path(r"C:\Users\Admin\Desktop\dupehell\test_gt_out")
os.makedirs(BASE, exist_ok=True)
DOMAINS = ["kyc", "healthcare", "fintech", "ecommerce", "blockchain",
"cybersecurity", "academia", "publishing", "insurance", "telecom"]
FAILURES = []
def fail(domain, part, test, detail):
FAILURES.append((domain, part, test, detail))
def read_gt(domain, d: Path):
gt_path = sorted(d.glob(f"*{domain}*ground_truth*"))[0]
return pl.read_parquet(gt_path)
def read_data(domain, d: Path):
data_paths = sorted(d.glob(f"*{domain}*.parquet"))
data_paths = [p for p in data_paths if "ground_truth" not in p.name]
return pl.read_parquet(data_paths[0])
print("=" * 70)
print("DUPEHELL v0.5.0 - GT INTEGRITY VALIDATION (10 domains)")
print("=" * 70)
results = {}
for dom in DOMAINS:
out_dir = BASE / dom
out_dir.mkdir(parents=True, exist_ok=True)
print(f"\nGenerating {dom} (size=2000, seed=42)... ", end="", flush=True)
try:
res = dupehell.generate(domain=dom, size=2000, seed=42,
output_format="parquet", output_dir=str(out_dir),
difficulty="medium")
results[dom] = res
gt = read_gt(dom, out_dir)
print(f"OK ({gt.height} rows, cols={gt.columns})")
except Exception as e:
print(f"FAILED: {e}")
fail(dom, "generation", "generate()", str(e))
print("\n" + "=" * 70)
print("PART A: GT STRUCTURAL INTEGRITY")
print("=" * 70)
for dom in DOMAINS:
if dom not in results:
continue
d = BASE / dom
print(f"\n--- {dom} ---")
try:
gt = read_gt(dom, d)
ds = read_data(dom, d)
except Exception as e:
fail(dom, "A", "load", str(e))
continue
actual_cols = set(gt.columns)
print(f" GT columns: {gt.columns}")
expected = {"record_id", "master_id", "match_type"}
if not expected.issubset(actual_cols):
fail(dom, "A", "A1:missing GT columns", f"missing {expected - actual_cols}")
else:
print(f" A1: has GT core columns OK")
nulls = gt.select(pl.all().is_null().sum()).to_dicts()[0]
bad = {k: v for k, v in nulls.items() if v > 0}
if bad:
fail(dom, "A", "A2:nulls in GT", str(bad))
else:
print(f" A2: no nulls OK")
valid_types = {"exact_dup", "hard_neg", "unique", "canary", "fuzzy_dup"}
actual_types = set(gt["match_type"].unique().to_list())
bad_types = actual_types - valid_types
if bad_types:
fail(dom, "A", "A3:invalid match_type", f"invalid: {bad_types}")
else:
print(f" A3: valid match_types OK ({actual_types})")
gt_ids = set(gt["record_id"].to_list())
ds_ids = set(ds["record_id"].to_list())
orphans = gt_ids - ds_ids
if orphans:
fail(dom, "A", "A4:orphan record_ids in GT", f"{len(orphans)} orphans: {list(orphans)[:5]}")
else:
print(f" A4: no orphans ({len(gt_ids)} ids match) OK")
dup_gt = gt.filter(pl.count("record_id").over("record_id") > 1)
if dup_gt.height > 0:
fail(dom, "A", "A5:dup record_id in GT", f"{dup_gt.height} dups")
else:
print(" A5: GT record_ids unique OK")
dup_ds = ds.filter(pl.count("record_id").over("record_id") > 1)
if dup_ds.height > 0:
fail(dom, "A", "A6:dup record_id in dataset", f"{dup_ds.height} dups")
else:
print(" A6: DS record_ids unique OK")
if gt.height != ds.height:
fail(dom, "A", "A7:row count mismatch", f"GT:{gt.height} vs DS:{ds.height}")
else:
print(f" A7: GT rows == DS rows ({gt.height}) OK")
print("\n" + "=" * 70)
print("PART B: MATCH TYPE DISTRIBUTION")
print("=" * 70)
for dom in DOMAINS:
if dom not in results:
continue
d = BASE / dom
gt = read_gt(dom, d)
counts = gt.group_by("match_type").len().sort("match_type")
total = counts["len"].sum()
print(f"\n--- {dom} (total={total}) ---")
for row in counts.to_dicts():
print(f" {row['match_type']}: {row['len']} ({row['len']/total*100:.1f}%)")
canary_ct = gt.filter(pl.col("match_type") == "canary").height
if canary_ct == 0:
fail(dom, "B", "B2:no canary", "0 canary records")
else:
print(f" canary: {canary_ct} OK")
sum_check = counts["len"].sum()
if sum_check != gt.height:
fail(dom, "B", "B3:count sum mismatch", f"{sum_check} vs {gt.height}")
hn_ct = gt.filter(pl.col("match_type") == "hard_neg").height
hn_ratio = hn_ct / gt.height * 100
print(f" hard_neg ratio: {hn_ratio:.2f}%")
if hn_ratio <= 5.0:
fail(dom, "B", "B4:low hard_neg ratio", f"{hn_ratio:.2f}% (threshold >5%)")
print("\n" + "=" * 70)
print("PART C: CLUSTER CONSISTENCY")
print("=" * 70)
for dom in DOMAINS:
if dom not in results:
continue
d = BASE / dom
gt = read_gt(dom, d)
print(f"\n--- {dom} ---")
mt_consistency = gt.group_by("master_id").agg(pl.col("match_type").unique().alias("mts"))
bad_mt = mt_consistency.filter(pl.col("mts").list.len() > 1)
if bad_mt.height > 0:
fail(dom, "C", "C1:inconsistent match_type within cluster",
f"{bad_mt.height} master_ids have mixed match_types")
else:
print(f" C1: all clusters have consistent match_type OK")
for mt in ["exact_dup", "fuzzy_dup"]:
sub = gt.filter(pl.col("match_type") == mt)
if sub.height == 0:
continue
cs = sub.group_by("master_id").len().rename({"len": "cluster_size"})
bad_sz = cs.filter(pl.col("cluster_size") < 2)
if bad_sz.height > 0:
fail(dom, "C", f"C2:{mt} cluster_size<2", f"{bad_sz.height} records in undersized clusters")
uniq_sub = gt.filter(pl.col("match_type") == "unique")
uniq_sizes = uniq_sub.group_by("master_id").len().rename({"len": "cluster_size"})
uniq_bad = uniq_sizes.filter(pl.col("cluster_size") != 1)
if uniq_bad.height > 0:
fail(dom, "C", "C3:unique cluster_size!=1", f"{uniq_bad.height} unique master_ids have != 1 record")
else:
print(f" C3: all unique clusters have size 1 OK")
print(" C1-C4 checks: see above for details")
print("\n" + "=" * 70)
print("PART D: EXACT_DUP CLUSTER QUALITY")
print("=" * 70)
QUAL_DOMAINS = ["kyc", "healthcare", "fintech"]
for dom in QUAL_DOMAINS:
if dom not in results:
continue
d = BASE / dom
gt = read_gt(dom, d)
ds = read_data(dom, d)
print(f"\n--- {dom} ---")
merged = ds.join(gt, on="record_id", how="inner")
data_cols = [c for c in ds.columns if c not in ("record_id",)]
dups = merged.filter(pl.col("match_type") == "exact_dup")
total_dups = dups.height
print(f" exact_dup records: {total_dups}")
if total_dups > 0:
cluster_sizes = dups.group_by("master_id").len().sort("len", descending=True)
print(f" Top cluster sizes: {cluster_sizes['len'].to_list()[:10]}")
top_clusters = cluster_sizes.head(min(5, cluster_sizes.height))["master_id"].to_list()
for mid in top_clusters:
cluster_recs = merged.filter(pl.col("master_id") == mid)
if cluster_recs.height < 2:
continue
rows = cluster_recs.to_dicts()
ref = rows[0]
identical_counts = []
for r in rows[1:]:
same = sum(1 for c in data_cols if c in ref and c in r and ref[c] == r[c])
identical_counts.append(same)
if identical_counts:
avg_identical = sum(identical_counts) / len(identical_counts)
print(f" mid={mid} ({cluster_recs.height} rows): ~{avg_identical:.1f}/{len(data_cols)} cols identical")
if avg_identical < 3:
fail(dom, "D", f"cluster {mid}",
f"only ~{avg_identical:.1f}/{len(data_cols)} cols identical")
uniq_recs = merged.filter(pl.col("match_type") == "unique")
critical_cols = data_cols[:min(5, len(data_cols))]
uniq_samples = uniq_recs["record_id"].unique().to_list()[:10]
for uid in uniq_samples:
row = merged.filter(pl.col("record_id") == uid)
if row.height == 0:
continue
row_dict = row.to_dicts()[0]
crit_vals = {c: row_dict[c] for c in critical_cols if c in row_dict and row_dict[c] is not None}
if not crit_vals:
continue
matches = merged
for c, v in crit_vals.items():
matches = matches.filter(pl.col(c) == v)
non_self = matches.filter(pl.col("record_id") != uid)
if non_self.height > 0:
fail(dom, "D", f"unique record {uid} has dupes",
f"{non_self.height} other records share {len(crit_vals)} critical cols: {list(crit_vals.keys())}")
print("\n" + "=" * 70)
print("PART E: GT SCALE WITH SIZE")
print("=" * 70)
SIZES = [100, 1000, 10000]
scale_results = {}
for sz in SIZES:
out_dir = BASE / f"scale_kyc_{sz}"
out_dir.mkdir(parents=True, exist_ok=True)
print(f"\nGenerating kyc size={sz}... ", end="", flush=True)
try:
res = dupehell.generate(domain="kyc", size=sz, seed=42,
output_format="parquet", output_dir=str(out_dir),
difficulty="medium")
gt = pl.read_parquet(res.ground_truth)
counts = gt.group_by("match_type").len().sort("match_type")
total = counts["len"].sum()
ratios = {row["match_type"]: row["len"]/total*100 for row in counts.to_dicts()}
scale_results[sz] = {"total": total, "ratios": ratios, "gt_height": gt.height}
print(f"total={total}, ratios={ {k: f'{v:.1f}%' for k, v in ratios.items()} }")
except Exception as e:
fail(f"scale_kyc_{sz}", "E", "generate", str(e))
if 100 in scale_results and 10000 in scale_results:
r100 = scale_results[100]["ratios"]
r10000 = scale_results[10000]["ratios"]
print("\nRatio comparison (size=100 vs size=10000):")
for mt in set(list(r100.keys()) + list(r10000.keys())):
r_small = r100.get(mt, 0)
r_large = r10000.get(mt, 0)
diff = abs(r_small - r_large)
status = "OK" if diff <= 15 else "WARN"
print(f" {mt}: {r_small:.1f}% -> {r_large:.1f}% (delta={diff:.1f}pp) {status}")
for sz in SIZES:
if sz not in scale_results:
continue
ratios = scale_results[sz]["ratios"]
unique_pct = ratios.get("unique", 0)
if unique_pct > 99:
fail(f"scale_kyc_{sz}", "E", "E3:nearly all unique",
f"{unique_pct:.1f}% unique - dataset degenerate")
print(f" size={sz}: unique={unique_pct:.1f}%")
prev_total = None
for sz in SIZES:
if sz not in scale_results:
continue
if prev_total is not None:
if scale_results[sz]["total"] <= prev_total:
fail(f"scale_kyc_{sz}", "E", "E1:entity count not monotonic",
f"size={sz}: {scale_results[sz]['total']} <= prev {prev_total}")
prev_total = scale_results[sz]["total"]
print("\n" + "=" * 70)
print("SUMMARY")
print("=" * 70)
by_domain = {}
for d, p, t, det in FAILURES:
by_domain.setdefault(d, []).append((p, t, det))
for dom in DOMAINS:
if dom in by_domain:
print(f"\n{dom}: FAILURES={len(by_domain[dom])}")
for p, t, det in by_domain[dom]:
print(f" {p}/{t}: {det}")
if FAILURES:
print(f"\nTOTAL FAILURES: {len(FAILURES)}")
else:
print("\nALL CHECKS PASSED")
with open(BASE / "failures.json", "w") as f:
json.dump([{"domain": d, "part": p, "test": t, "detail": det} for d, p, t, det in FAILURES], f, indent=2, default=str)