Blame for scripts/stress_test_precision.py

"""Precision stress test: how often are GENUINE returns flagged when the return photo is

taken differently?



Precision = TP / (TP + FP) only drops when a genuine customer's return is flagged, so this

test changes *how the returned item is photographed* (one augmentation type at a time, at a

mild and a strong level) and measures the false flags. Fraud pairs are scored under the same

augmentation, so precision per condition is exact, not estimated.



Only the return side is augmented. One augmented photo set per item therefore serves both its

genuine pair (X -> X) and every fraud pair in which X is the returned item (Y -> X): per

condition, 8 genuine + 56 fraud pairs from 24 augmented photos.



This runs the free local stage (background removal, quality gate, object-only DINOv2, colour,

evidence-weighted decision) with no LLM calls. For each genuine return it also records what the

full pipeline would do next: fast-accept, send to Gemini, or fast-reject (a hard false

rejection no LLM can undo).



    python scripts/stress_test_precision.py

"""
import json
import math
import random
import sys
import time
import zlib
from pathlib import Path

import numpy as np
from PIL import Image, ImageDraw, ImageEnhance, ImageFilter

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(ROOT / "scripts"))

from build_dataset import ANGLES, ITEMS, has_second_set, load  # noqa: E402
from returnguard import imaging  # noqa: E402
from returnguard.agents import decision  # noqa: E402
from returnguard.agents.screen import visual_screen  # noqa: E402
from returnguard.embeddings import view_features  # noqa: E402

CATEGORY = {"speaker": "electronics", "headphones": "electronics", "shaker": "home_kitchen", "bottle": "home_kitchen"}
FRAUD_PREVALENCE = 0.15   # Appriss Retail 2024: 15.14% of returns were fraudulent (US); used for "implied precision"
RESULTS = ROOT / "results"


# ── augmentations ──────────────────────────────────────────────────────────
def _fill(img):
    edges = np.array(img)[[0, -1]].reshape(-1, 3)
    return tuple(int(c) for c in np.median(edges, axis=0))


def rotate(img, rng, deg):
    return img.rotate(rng.choice([-1, 1]) * deg, resample=Image.BICUBIC, fillcolor=_fill(img))


def perspective(img, rng, amount):
    """Viewpoint tilt: one edge of the frame comes closer to the camera than the opposite edge."""
    w, h = img.size
    d = amount * min(w, h)
    src = [(0, 0), (w, 0), (w, h), (0, h)]
    side = rng.choice(["top", "bottom", "left", "right"])
    dst = {"top": [(d, 0), (w - d, 0), (w, h), (0, h)], "bottom": [(0, 0), (w, 0), (w - d, h), (d, h)],
           "left": [(0, d), (w, 0), (w, h), (0, h - d)], "right": [(0, 0), (w, d), (w, h - d), (0, h)]}[side]
    rows, rhs = [], []
    for (x, y), (u, v) in zip(dst, src):            # PIL maps output -> input pixels
        rows += [[x, y, 1, 0, 0, 0, -u * x, -u * y], [0, 0, 0, x, y, 1, -v * x, -v * y]]
        rhs += [u, v]
    coeffs = np.linalg.solve(np.array(rows, float), np.array(rhs, float))
    return img.transform(img.size, Image.PERSPECTIVE, tuple(coeffs), Image.BICUBIC, fillcolor=_fill(img))


def zoom_out(img, rng, scale):
    """Photo taken from further away: the product is smaller in the frame."""
    w, h = img.size
    small = img.resize((int(w * scale), int(h * scale)), Image.LANCZOS)
    canvas = Image.new("RGB", (w, h), _fill(img))
    dx = int(rng.uniform(0.2, 0.8) * (w - small.width))
    dy = int(rng.uniform(0.2, 0.8) * (h - small.height))
    canvas.paste(small, (dx, dy))
    return canvas


def crop_cut(img, rng, frac):
    """Product partly out of frame: one side cut off, then the photo is the usual size again."""
    w, h = img.size
    side = rng.choice(["top", "bottom", "left", "right"])
    box = {"top": (0, int(h * frac), w, h), "bottom": (0, 0, w, int(h * (1 - frac))),
           "left": (int(w * frac), 0, w, h), "right": (0, 0, int(w * (1 - frac)), h)}[side]
    return img.crop(box).resize((w, h), Image.LANCZOS)


def brightness(img, rng, f):
    return ImageEnhance.Brightness(img).enhance(f)


def contrast(img, rng, f):
    return ImageEnhance.Contrast(img).enhance(f)


def white_balance(img, rng, red, blue):
    arr = np.array(img, dtype=np.float32)
    arr[..., 0] *= red
    arr[..., 2] *= blue
    return Image.fromarray(np.clip(arr, 0, 255).astype(np.uint8))


def shadow(img, rng, strength):
    """A soft shadow falling across the scene (multiplicative gradient)."""
    w, h = img.size
    ang = rng.uniform(0, 2 * math.pi)
    yy, xx = np.mgrid[0:h, 0:w]
    t = ((xx - w / 2) * math.cos(ang) + (yy - h / 2) * math.sin(ang)) / (0.5 * math.hypot(w, h))
    mult = 1 - strength * np.clip((t + 0.2) / 0.8, 0, 1)
    return Image.fromarray(np.clip(np.array(img, np.float32) * mult[..., None], 0, 255).astype(np.uint8))


def glare(img, rng, strength):
    """A bright reflection (e.g. a ceiling light) over part of the product."""
    w, h = img.size
    r = int(min(w, h) * 0.18)
    cx, cy = int(w * rng.uniform(0.35, 0.65)), int(h * rng.uniform(0.3, 0.6))
    mask = Image.new("L", (w, h), 0)
    ImageDraw.Draw(mask).ellipse((cx - r, cy - int(r * 0.6), cx + r, cy + int(r * 0.6)), fill=255)
    m = np.array(mask.filter(ImageFilter.GaussianBlur(r * 0.5)), np.float32)[..., None] / 255 * strength
    arr = np.array(img, np.float32)
    return Image.fromarray(np.clip(arr * (1 - m) + 255 * m, 0, 255).astype(np.uint8))


def gaussian_blur(img, rng, radius):
    return img.filter(ImageFilter.GaussianBlur(radius))


def motion_blur(img, rng, length):
    """Camera shake: average along one direction over `length` pixels."""
    arr = np.array(img, np.float32)
    axis = rng.choice([0, 1])
    pad = [(0, 0)] * 3
    pad[axis] = (length // 2, length - length // 2)
    padded = np.pad(arr, pad, mode="edge")
    c = np.cumsum(padded, axis=axis, dtype=np.float64)
    c = np.concatenate([np.zeros_like(np.take(c, [0], axis=axis)), c], axis=axis)
    out = (np.take(c, range(length, c.shape[axis]), axis=axis) - np.take(c, range(0, c.shape[axis] - length), axis=axis)) / length
    out = np.take(out, range(arr.shape[axis]), axis=axis)
    return Image.fromarray(np.clip(out, 0, 255).astype(np.uint8))


def noise(img, rng, sigma):
    nrng = np.random.default_rng(rng.randint(0, 2 ** 31))
    arr = np.array(img, np.float32) + nrng.normal(0, sigma, (img.height, img.width, 3))
    return Image.fromarray(np.clip(arr, 0, 255).astype(np.uint8))


def jpeg(img, rng, quality):
    import io
    buf = io.BytesIO()
    img.save(buf, "JPEG", quality=quality)
    return Image.open(io.BytesIO(buf.getvalue())).convert("RGB")


def low_res(img, rng, f):
    w, h = img.size
    return img.resize((max(8, int(w * f)), max(8, int(h * f))), Image.BILINEAR).resize((w, h), Image.BICUBIC)


def occlusion(img, rng, width_frac):
    """A finger or thumb covering part of the product, entering from one edge to just past the

    centre of the frame; `width_frac` is the finger's width as a share of the photo's short side."""
    w, h = img.size
    out = img.copy()
    layer = Image.new("RGBA", (w, h), (0, 0, 0, 0))
    d = ImageDraw.Draw(layer)
    skin = (214, 160, 125, 255)
    side = rng.choice(["left", "right", "bottom"])
    cy, cx = int(h * rng.uniform(0.4, 0.6)), int(w * rng.uniform(0.4, 0.6))
    # the fingertip reaches the middle of the frame, where the product is
    length = {"left": cx, "right": w - cx, "bottom": h - cy}[side] + int(0.08 * min(w, h))
    thick = max(8, int(width_frac * min(w, h)))
    if side == "left":
        box = (0, cy - thick // 2, length, cy + thick // 2)
    elif side == "right":
        box = (w - length, cy - thick // 2, w, cy + thick // 2)
    else:
        box = (cx - thick // 2, h - length, cx + thick // 2, h)
    d.rounded_rectangle(box, radius=thick // 2, fill=skin)
    layer = layer.filter(ImageFilter.GaussianBlur(2))
    out.paste(layer, mask=layer.getchannel("A"))
    return out


_BG_CACHE = {}


def _background(kind, size):
    w, h = size
    if kind == "plain":
        grad = np.linspace(0.92, 1.05, h)[:, None, None]
        bg = np.ones((h, w, 3), np.float32) * np.array([205, 196, 182], np.float32) * grad
        return Image.fromarray(np.clip(bg, 0, 255).astype(np.uint8))
    if "cluttered" not in _BG_CACHE:
        # patterned tablecloth from one of the real photo sessions (no product in this region)
        src = Image.open(ROOT / "Dataset" / "items" / "headphones_blue" / "set2" / "front.jpeg").convert("RGB")
        _BG_CACHE["cluttered"] = src.crop((0, int(src.height * 0.82), src.width, src.height))
    tex = _BG_CACHE["cluttered"]
    scale = max(w / tex.width, h / tex.height)
    tex = tex.resize((int(tex.width * scale) + 1, int(tex.height * scale) + 1), Image.LANCZOS)
    return tex.crop((0, 0, w, h))


def background(img, rng, kind):
    """The same product photographed on a different surface (plain floor, or a busy patterned cloth)."""
    from rembg import remove
    mask = remove(img, session=imaging._rembg_session(), only_mask=True).convert("L")
    bg = _background(kind, img.size)
    return Image.composite(img, bg, mask)


CONDITIONS = [  # name, label, function, mild params, strong params
    ("rotation", "Rotated", rotate, (10,), (25,)),
    ("perspective", "Viewpoint tilt", perspective, (0.08,), (0.18,)),
    ("zoom_out", "Photographed from further away", zoom_out, (0.8,), (0.55,)),
    ("crop", "Partly out of frame", crop_cut, (0.10,), (0.25,)),
    ("darker", "Darker room", brightness, (0.7,), (0.45,)),
    ("brighter", "Overexposed", brightness, (1.35,), (1.8,)),
    ("warm", "Warm (yellow) light", white_balance, (1.08, 0.92), (1.18, 0.80)),
    ("cool", "Cool (blue) light", white_balance, (0.92, 1.08), (0.82, 1.18)),
    ("low_contrast", "Low contrast / haze", contrast, (0.7,), (0.45,)),
    ("shadow", "Shadow across the item", shadow, (0.35,), (0.65,)),
    ("glare", "Glare / reflection", glare, (0.45,), (0.8,)),
    ("blur", "Out of focus", gaussian_blur, (1.5,), (3.5,)),
    ("motion_blur", "Camera shake", motion_blur, (9,), (21,)),
    ("noise", "Sensor noise (low light)", noise, (10,), (25,)),
    ("jpeg", "Heavy JPEG compression", jpeg, (30,), (8,)),
    ("low_res", "Low resolution", low_res, (0.5,), (0.25,)),
    ("occlusion", "Finger covering part", occlusion, (0.08,), (0.14,)),
    ("background", "Different background", background, ("plain",), ("cluttered",)),
]


def seed_for(*parts) -> int:
    return zlib.crc32("|".join(map(str, parts)).encode())


# ── scoring ────────────────────────────────────────────────────────────────
def features(img):
    cut = imaging.remove_background(img)
    q = imaging.assess_quality(img, cut)
    return view_features(cut), {"ok": q.ok, "problem": q.problem}


def local_decision(delivered, d_feats, r_feats, r_quality):
    screen = visual_screen(d_feats, r_feats)
    state = {"quality": r_quality, "screen": screen, "llm_calls": [],
             "policy": {"eligible": True, "reason": "changed_mind", "notes": [], "requires_functional_test": False},
             "history": {"tier": "new", "accept_below": 50, "reject_from": 90, "prior_flagged": 0},
             "order": {"category": CATEGORY[delivered.split("_")[0]], "fingerprint": None}}
    if any(not q["ok"] for q in r_quality.values()):
        screen = None
    d = decision.decide(state)
    return d["decision"], screen


def wilson(k, n, z=1.96):
    if n == 0:
        return [0.0, 0.0]
    p, d = k / n, 1 + z * z / n
    c = (p + z * z / (2 * n)) / d
    hw = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n)) / d
    return [round(max(0, c - hw), 4), round(min(1, c + hw), 4)]


def implied_precision(tpr, fpr, prevalence=FRAUD_PREVALENCE):
    denom = tpr * prevalence + fpr * (1 - prevalence)
    return round(tpr * prevalence / denom, 4) if denom else None


def score_condition(name, label, level, returns, deliveries):
    """returns: item -> (features by angle, quality by angle)."""
    genuine, fraud = [], []
    for returned, (r_feats, r_quality) in returns.items():
        for delivered, d_feats in deliveries.items():
            dec, screen = local_decision(delivered, d_feats, r_feats, r_quality)
            row = {"delivered": delivered, "returned": returned, "decision": dec,
                   "screen": screen["verdict"] if screen else "retake",
                   "shape_mean": screen["shape_mean"] if screen else None,
                   "color_shift": screen["color_shift_median"] if screen else None}
            (genuine if delivered == returned else fraud).append(row)
    fp = sum(r["decision"] != "ACCEPT" for r in genuine)
    tp = sum(r["decision"] != "ACCEPT" for r in fraud)
    tn, fn = len(genuine) - fp, len(fraud) - tp
    precision = tp / (tp + fp) if tp + fp else 0.0
    recall = tp / len(fraud)
    fpr = fp / len(genuine)
    f05 = (1.25 * precision * recall / (0.25 * precision + recall)) if precision + recall else 0.0
    by = lambda rows, key, val: sum(r[key] == val for r in rows)
    return {
        "condition": name, "label": label, "level": level,
        "genuine": len(genuine), "fraud": len(fraud),
        "local": {"tp": tp, "fp": fp, "fn": fn, "tn": tn, "precision": round(precision, 4),
                  "precision_ci": wilson(tp, tp + fp), "recall": round(recall, 4), "f0_5": round(f05, 4),
                  "false_flag_rate": round(fpr, 4), "false_flag_ci": wilson(fp, len(genuine)),
                  "implied_precision_at_15pct_fraud": implied_precision(recall, fpr)},
        "genuine_routing": {"fast_accept": by(genuine, "screen", "strong_match"),
                            "sent_to_gemini": by(genuine, "screen", "ambiguous"),
                            "fast_reject": by(genuine, "screen", "clear_mismatch"),
                            "retake": by(genuine, "screen", "retake")},
        "genuine_shape_min": min((r["shape_mean"] for r in genuine if r["shape_mean"] is not None), default=None),
        "genuine_pairs": genuine,
    }


def main():
    t0 = time.time()
    items = list(ITEMS)
    print(f"Precision stress test: {len(items)} items, {len(CONDITIONS)} augmentation types x 2 levels + baseline")
    deliveries = {it: {a: features(load(it, a))[0] for a in ANGLES} for it in items}

    def augmented_returns(fn, params, cname):
        out = {}
        for it in items:
            feats, quals = {}, {}
            for a in ANGLES:
                rng = random.Random(seed_for(cname, it, a, params))
                feats[a], quals[a] = features(fn(load(it, a), rng, *params))
            out[it] = (feats, quals)
        return out

    rows = [score_condition("baseline", "No change (same photo, re-encoded)", "none",
                            augmented_returns(jpeg, (92,), "baseline"), deliveries)]
    print(f"  baseline done ({time.time() - t0:.0f}s)", flush=True)
    for name, label, fn, mild, strong in CONDITIONS:
        for level, params in (("mild", mild), ("strong", strong)):
            row = score_condition(name, label, level, augmented_returns(fn, params, f"{name}-{level}"), deliveries)
            rows.append(row)
            g, l = row["genuine_routing"], row["local"]
            print(f"  {label:34s} {level:6s} genuine flagged {l['fp']}/{row['genuine']}  "
                  f"(fast-reject {g['fast_reject']}, to Gemini {g['sent_to_gemini']}, retake {g['retake']})  "
                  f"precision {l['precision']:.3f}  recall {l['recall']:.3f}  [{time.time() - t0:.0f}s]", flush=True)

    # real second photo session (headphones only): the honest version of all of the above
    real = []
    for it in [i for i in items if has_second_set(i)]:
        r_feats, r_quality = {}, {}
        for a in ANGLES:
            r_feats[a], r_quality[a] = features(load(it, a, 2))
        dec, screen = local_decision(it, deliveries[it], r_feats, r_quality)
        real.append({"item": it, "decision": dec, "screen": screen["verdict"] if screen else "retake",
                     "shape_mean": screen["shape_mean"] if screen else None,
                     "color_shift": screen["color_shift_median"] if screen else None})

    aug_rows = rows[1:]
    total_genuine = sum(r["genuine"] for r in aug_rows)
    summary = {
        "conditions": len(aug_rows),
        "genuine_returns_tested": total_genuine,
        "genuine_flagged_by_local_stage": sum(r["local"]["fp"] for r in aug_rows),
        "hard_false_rejections": sum(r["genuine_routing"]["fast_reject"] for r in aug_rows),
        "sent_to_gemini": sum(r["genuine_routing"]["sent_to_gemini"] for r in aug_rows),
        "retakes": sum(r["genuine_routing"]["retake"] for r in aug_rows),
        "fast_accepted": sum(r["genuine_routing"]["fast_accept"] for r in aug_rows),
        "fraud_prevalence_assumed_for_implied_precision": FRAUD_PREVALENCE,
    }
    results = {"generated_at": time.strftime("%Y-%m-%dT%H:%M:%S"), "stage": "local (no LLM calls)",
               "summary": summary, "conditions": rows, "real_second_session": real,
               "runtime_s": round(time.time() - t0)}
    RESULTS.mkdir(exist_ok=True)
    (RESULTS / "stress_precision.json").write_text(json.dumps(results, indent=2), encoding="utf-8")
    write_markdown(results, RESULTS / "stress_precision.md")
    print(f"\nWrote {RESULTS / 'stress_precision.json'} in {time.time() - t0:.0f}s")


def write_markdown(r, path):
    s = r["summary"]
    lines = [
        "# Precision stress test (local stage, no LLM)", "",
        f"Generated {r['generated_at']}. Each condition changes only how the **returned** item is photographed; "
        "delivery photos are untouched. Per condition: 8 genuine + 56 fraud pairs (8 items). "
        "*Flagged* = not auto-accepted by the local stage (review, reject or retake).", "",
        f"Across {s['conditions']} conditions, {s['genuine_flagged_by_local_stage']} of {s['genuine_returns_tested']} "
        f"genuine returns were flagged by the local stage: {s['hard_false_rejections']} fast-rejected outright (a hard false "
        f"rejection no LLM would undo), {s['sent_to_gemini']} would be sent to Gemini, {s['retakes']} asked for a retake; "
        f"{s['fast_accepted']} were fast-accepted.", "",
        "| Augmentation | Level | Genuine flagged | Fast-reject | To Gemini | Retake | Precision (95% interval) | Recall | Implied precision at 15% fraud |",
        "|---|---|---|---|---|---|---|---|---|",
    ]
    for c in r["conditions"]:
        l, g = c["local"], c["genuine_routing"]
        lines.append(f"| {c['label']} | {c['level']} | {l['fp']}/{c['genuine']} | {g['fast_reject']} | {g['sent_to_gemini']} | "
                     f"{g['retake']} | {l['precision'] * 100:.1f}% ({l['precision_ci'][0] * 100:.0f}–{l['precision_ci'][1] * 100:.0f}%) | "
                     f"{l['recall'] * 100:.1f}% | {(l['implied_precision_at_15pct_fraud'] or 0) * 100:.1f}% |")
    if r["real_second_session"]:
        lines += ["", "## Real second photo session (headphones)", "",
                  "| Item | Local decision | Screen | Shape | Colour ΔE |", "|---|---|---|---|---|"]
        for x in r["real_second_session"]:
            lines.append(f"| {x['item']} | {x['decision']} | {x['screen']} | {x['shape_mean']} | {x['color_shift']} |")
    lines += ["", "*Implied precision at 15% fraud* re-weights the measured false-flag rate and recall to a realistic "
              "share of fraudulent returns (Appriss Retail 2024: 15.14%), because precision depends on how common fraud is."]
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")


if __name__ == "__main__":
    main()