apps/api/app/devtools/corpus.py

"""The canonical document corpora.

Two sets, both generated deterministically from a fixed seed so that a run today
and a run next month score the same documents:

* **Seed corpus** — 30 historical invoices across 6 vendors (spec §8), so the
  vendor-spend chart and the per-vendor z-scores are meaningful the first time the
  dashboard is opened, plus a planted near-duplicate pair so the anomaly demo
  always fires.
* **Evaluation corpus** — 10 labelled invoices (spec §8) including two
  near-duplicates and one inflated amount, with the expected anomaly for each
  document recorded up front so `eval/run_eval.py` can compute precision/recall
  rather than eyeballing the output.

Vendors have distinct, stable profiles — currency, payment terms, typical spend
band — because a per-vendor z-score is meaningless if every vendor looks alike.
"""

from __future__ import annotations

import random
import uuid
from collections.abc import Callable
from dataclasses import dataclass, replace
from datetime import date, timedelta
from decimal import Decimal
from typing import Final

from app.devtools.documents import InvoiceSpec, LineSpec
from app.models.enums import AnomalyType
from app.pipeline.anomaly import InvoiceRecord, ScreeningConfig, screen_invoice

SEED: Final = 20260729
SEED_CORPUS_SIZE: Final = 30
EVAL_CORPUS_SIZE: Final = 10


@dataclass(frozen=True, slots=True)
class VendorProfile:
    name: str
    address: str
    trn: str
    currency: str
    payment_terms: str
    vat_rate: Decimal
    catalogue: tuple[tuple[str, Decimal], ...]
    qty_range: tuple[int, int]
    lines_range: tuple[int, int]


VENDORS: Final[tuple[VendorProfile, ...]] = (
    VendorProfile(
        name="Gulf Metals L.L.C.",
        address="Warehouse 14, Industrial Area 12\nAl Quoz, Dubai, UAE",
        trn="100234567800003",
        currency="AED",
        payment_terms="Net 30",
        vat_rate=Decimal("0.05"),
        catalogue=(
            ("Steel plate 10mm x 2400", Decimal("420.00")),
            ("Galvanised bolts M12 (box of 100)", Decimal("95.50")),
            ("Welding consumables set", Decimal("812.00")),
            ("Aluminium channel 6m", Decimal("268.75")),
            ("Stainless sheet 1.5mm", Decimal("534.20")),
        ),
        qty_range=(3, 14),
        lines_range=(3, 5),
    ),
    VendorProfile(
        name="Aurora Systems FZ-LLC",
        address="Office 1108, Dubai Internet City\nDubai, UAE",
        trn="100987654300003",
        currency="AED",
        payment_terms="Net 45",
        vat_rate=Decimal("0.05"),
        catalogue=(
            ("Managed endpoint support (per seat)", Decimal("148.00")),
            ("Cloud backup retention 1TB", Decimal("612.40")),
            ("Security monitoring tier 2", Decimal("1_284.00")),
            ("Network hardware maintenance", Decimal("376.90")),
        ),
        qty_range=(2, 9),
        lines_range=(2, 4),
    ),
    VendorProfile(
        name="Nakheel Facilities Management",
        address="Building 3, Business Bay\nDubai, UAE",
        trn="100445566700003",
        currency="AED",
        payment_terms="Net 15",
        vat_rate=Decimal("0.05"),
        catalogue=(
            ("Daily office cleaning (per week)", Decimal("362.50")),
            ("Deep carpet treatment", Decimal("845.00")),
            ("HVAC filter replacement", Decimal("218.30")),
            ("Pest control visit", Decimal("176.45")),
        ),
        qty_range=(1, 6),
        lines_range=(2, 4),
    ),
    VendorProfile(
        name="Cedar Print House",
        address="Shop 7, Al Karama\nDubai, UAE",
        trn="100112233400003",
        currency="AED",
        payment_terms="Due on receipt",
        vat_rate=Decimal("0.05"),
        catalogue=(
            ("Business cards (500)", Decimal("142.75")),
            ("A2 poster print", Decimal("63.40")),
            ("Bound report, colour", Decimal("88.90")),
            ("Roll-up banner", Decimal("311.20")),
        ),
        qty_range=(1, 8),
        lines_range=(2, 3),
    ),
    VendorProfile(
        name="Meridian Logistics DMCC",
        address="JAFZA One, Tower A\nJebel Ali, Dubai, UAE",
        trn="100778899100003",
        currency="USD",
        payment_terms="Net 30",
        vat_rate=Decimal("0.00"),
        catalogue=(
            ("Sea freight FCL 40ft", Decimal("2_845.00")),
            ("Customs clearance handling", Decimal("418.60")),
            ("Inland haulage per container", Decimal("736.25")),
            ("Warehousing per pallet/month", Decimal("54.80")),
        ),
        qty_range=(1, 5),
        lines_range=(2, 4),
    ),
    VendorProfile(
        name="Palm Office Supplies Trading",
        address="Warehouse 22, Al Qusais\nDubai, UAE",
        trn="100556677800003",
        currency="AED",
        payment_terms="Net 30",
        vat_rate=Decimal("0.05"),
        catalogue=(
            ("A4 paper (box of 5 reams)", Decimal("78.30")),
            ("Toner cartridge, mono", Decimal("246.90")),
            ("Ergonomic chair", Decimal("689.50")),
            ("Desk organiser set", Decimal("41.25")),
        ),
        qty_range=(2, 12),
        lines_range=(2, 4),
    ),
)

_BILL_TO: Final = "Northline Trading FZE\nJebel Ali Free Zone\nDubai, UAE"
_ROUND_GUARD: Final = Decimal("500")
# Smallest recurring-basket quantity, so a +-1 unit step stays a small move.
_MIN_BASKET_QTY: Final = 12
# How far a month may drift from the vendor baseline. A ~12% uniform band is
# ~7% standard deviation, which keeps an ordinary month under the 2-sigma
# z-score threshold while still looking like real month-to-month movement.
_SPEND_BAND: Final = Decimal("0.12")


@dataclass(frozen=True, slots=True)
class CorpusItem:
    """A document plus the anomalies it is expected to raise."""

    spec: InvoiceSpec
    as_scan: bool
    expected_anomalies: frozenset[AnomalyType]

    @property
    def stem(self) -> str:
        safe = self.spec.invoice_number.replace("/", "-")
        return f"{safe}{'-scan' if self.as_scan else ''}"

    @property
    def extension(self) -> str:
        return ".jpg" if self.as_scan else ".pdf"

    @property
    def filename(self) -> str:
        return f"{self.stem}{self.extension}"


def _build_lines(profile: VendorProfile, rng: random.Random) -> tuple[LineSpec, ...]:
    """A one-off invoice with quantities drawn freely across the vendor's range."""
    count = rng.randint(*profile.lines_range)
    chosen = rng.sample(profile.catalogue, min(count, len(profile.catalogue)))
    return tuple(
        LineSpec(
            description=description,
            qty=Decimal(rng.randint(*profile.qty_range)),
            unit_price=price,
        )
        for description, price in chosen
    )


def _vendor_basket(profile: VendorProfile, rng: random.Random) -> tuple[LineSpec, ...]:
    """The recurring basket a vendor bills month after month.

    Real suppliers invoice a fairly stable book of business, and the per-vendor
    z-score detector depends on that: if every historical invoice were drawn
    independently across the full quantity range, ordinary months would sit two
    standard deviations from the mean and the review queue would fill with noise
    that buries the findings that matter.
    """
    count = rng.randint(*profile.lines_range)
    chosen = rng.sample(profile.catalogue, min(count, len(profile.catalogue)))
    low, high = profile.qty_range
    # Sit in the upper half of the vendor's range. A recurring basket needs enough
    # units that a one-unit month-to-month step is a small relative move; anchored
    # at a quantity of 3, +-1 is a 33% swing and the spend band stops being a band.
    base = max(low + (high - low) * 2 // 3, _MIN_BASKET_QTY)
    return tuple(
        LineSpec(description=description, qty=Decimal(base), unit_price=price)
        for description, price in chosen
    )


def _vary_basket(
    basket: tuple[LineSpec, ...], rng: random.Random, *, spread: float = 0.12
) -> tuple[LineSpec, ...]:
    """Month-to-month wobble around the recurring basket, bounded at ±`spread`."""
    varied: list[LineSpec] = []
    for line in basket:
        delta = Decimal(str(1.0 + rng.uniform(-spread, spread)))
        # With a basket anchored at _MIN_BASKET_QTY a +-12% swing spans three
        # integers, so quantising produces genuine variety. Forcing a step
        # instead would delete the midpoint and collapse the space to 2^n
        # combinations, which collide within a five-month history.
        qty = (line.qty * delta).quantize(Decimal("1"))
        varied.append(LineSpec(line.description, max(Decimal("1"), qty), line.unit_price))
    return tuple(varied)


def _make_spec(
    profile: VendorProfile,
    *,
    number: str,
    issued: date,
    rng: random.Random,
    lines: tuple[LineSpec, ...] | None = None,
    payment_terms: str | None = None,
    forced_total: Decimal | None = None,
    note: str | None = None,
    tags: tuple[str, ...] = (),
) -> InvoiceSpec:
    terms = payment_terms or profile.payment_terms
    due_days = {"Net 15": 15, "Net 30": 30, "Net 45": 45}.get(terms, 0)
    body = lines if lines is not None else _build_lines(profile, rng)
    return InvoiceSpec(
        vendor=profile.name,
        vendor_address=profile.address,
        trn=profile.trn,
        invoice_number=number,
        issue_date=issued,
        due_date=issued + timedelta(days=due_days),
        bill_to=_BILL_TO,
        lines=body,
        currency=profile.currency,
        vat_rate=profile.vat_rate,
        payment_terms=terms,
        forced_total=forced_total,
        note=note,
        tags=tags,
    )


def _near_duplicate_lines(
    lines: tuple[LineSpec, ...], vat_rate: Decimal, target_gap: Decimal = Decimal("0.004")
) -> tuple[LineSpec, ...]:
    """Rebuild an item list so the invoice total lands `target_gap` above the original.

    A fixed multiplier on the last line is unreliable: how far it moves the total
    depends on that line's share of the invoice, which is exactly how the eval pair
    once drifted past the detector's 1% window. Solving for the unit price from the
    desired *total* delta keeps the pair inside the window whatever the basket
    looks like.
    """
    subtotal = sum((line.amount for line in lines), start=Decimal("0"))
    total = subtotal * (Decimal("1") + vat_rate)
    tail = lines[-1]
    if tail.qty <= 0 or total <= 0:
        return lines

    delta_unit = (total * target_gap) / (tail.qty * (Decimal("1") + vat_rate))
    bumped_price = (tail.unit_price + delta_unit).quantize(Decimal("0.01"))
    return (*lines[:-1], LineSpec(tail.description, tail.qty, bumped_price))


def _assert_within_duplicate_window(
    original: InvoiceSpec, copy: InvoiceSpec, tolerance: Decimal = Decimal("0.01")
) -> None:
    """Fail loudly at build time if a planted pair could not actually be detected."""
    gap = abs(copy.total - original.total) / max(copy.total, original.total)
    assert gap <= tolerance, (
        f"planted duplicate {copy.invoice_number} is {gap:.2%} from "
        f"{original.invoice_number}, outside the {tolerance:.0%} detector window"
    )
    days = abs((copy.issue_date - original.issue_date).days)
    assert days <= 7, f"planted duplicate is {days} days apart, outside the 7-day window"


def _distinct_variation(
    profile: VendorProfile,
    basket: tuple[LineSpec, ...],
    rng: random.Random,
    seen_totals: set[Decimal],
    attempts: int = 40,
) -> tuple[LineSpec, ...]:
    """Draw a month's basket whose total is new for this vendor and not round.

    Repeating a total across a vendor's history is the tell that data was
    generated rather than observed, and a coincidentally round total would fire
    the round-number detector on a document the corpus labelled clean. Redrawing
    until both hold keeps the seeded ledger credible without widening the spend
    band enough to disturb the z-score baseline.
    """

    def probe_total(lines: tuple[LineSpec, ...]) -> Decimal:
        return InvoiceSpec(
            vendor=profile.name,
            vendor_address=profile.address,
            trn=profile.trn,
            invoice_number="PROBE",
            issue_date=date(2026, 1, 1),
            due_date=date(2026, 1, 31),
            bill_to=_BILL_TO,
            lines=lines,
            currency=profile.currency,
            vat_rate=profile.vat_rate,
            payment_terms=profile.payment_terms,
        ).total

    baseline = probe_total(basket)
    best: tuple[Decimal, tuple[LineSpec, ...]] | None = None

    for _ in range(attempts):
        lines = _vary_basket(basket, rng)
        total = probe_total(lines)
        if total in seen_totals:
            continue
        drift = abs(total - baseline) / baseline if baseline else Decimal("0")
        if best is None or drift < best[0]:
            best = (drift, lines)
        if drift <= _SPEND_BAND and total % _ROUND_GUARD != 0:
            seen_totals.add(total)
            return lines

    # Nothing landed inside the band; take the closest distinct draw so the
    # history still varies rather than repeating a total.
    chosen = best[1] if best is not None else _vary_basket(basket, rng)
    seen_totals.add(probe_total(chosen))
    return chosen


def _is_suspiciously_round(spec: InvoiceSpec) -> bool:
    """Guard: an unintentionally round total would fire the round-number detector."""
    return spec.total >= _ROUND_GUARD and spec.total % _ROUND_GUARD == 0


def _resample_until_not_round(
    profile: VendorProfile, rng: random.Random, attempts: int = 24
) -> tuple[LineSpec, ...]:
    """Redraw line items until the total is not an exact multiple of 500.

    A total that happens to land on a round figure would fire the round-number
    detector, which would then count as a false positive against a document the
    corpus labelled clean.
    """
    for _ in range(attempts):
        lines = _build_lines(profile, rng)
        probe = InvoiceSpec(
            vendor=profile.name,
            vendor_address=profile.address,
            trn=profile.trn,
            invoice_number="PROBE",
            issue_date=date(2026, 1, 1),
            due_date=date(2026, 1, 31),
            bill_to=_BILL_TO,
            lines=lines,
            currency=profile.currency,
            vat_rate=profile.vat_rate,
            payment_terms=profile.payment_terms,
        )
        if not _is_suspiciously_round(probe):
            return lines
    return _build_lines(profile, rng)


def _draw_seed_corpus(reference: date | None, seed: int) -> list[CorpusItem]:
    """Exactly 30 historical invoices across 6 vendors (spec §8).

    Two of the thirty are a **planted near-duplicate pair**: the last Gulf Metals
    invoice is rewritten as a re-issue of the one before it, so the anomaly demo
    fires on a freshly seeded database without inflating the corpus past the
    thirty documents the spec calls for.

    History is dated backwards from `reference` so the dashboard always shows a
    recent, plausible six months of activity.
    """
    today = reference or date(2026, 7, 20)
    rng = random.Random(seed)
    items: list[CorpusItem] = []

    per_vendor = SEED_CORPUS_SIZE // len(VENDORS)
    for vendor_index, profile in enumerate(VENDORS):
        basket = _vendor_basket(profile, rng)
        seen_totals: set[Decimal] = set()
        for occurrence in range(per_vendor):
            # Oldest first, roughly monthly, with a few days of natural jitter.
            # The ~22-day minimum gap keeps ordinary months well outside the
            # 7-day duplicate window.
            months_ago = per_vendor - occurrence
            issued = today - timedelta(days=months_ago * 30 + rng.randint(-4, 4))
            number = f"{profile.name[:2].upper()}-2026-{vendor_index + 1}{occurrence + 1:02d}"
            lines = _distinct_variation(profile, basket, rng, seen_totals)
            spec = _make_spec(
                profile, number=number, issued=issued, rng=rng, lines=lines, tags=("seed",)
            )
            items.append(CorpusItem(spec=spec, as_scan=False, expected_anomalies=frozenset()))

    # --- Plant the near-duplicate *inside* the thirty, by rewriting the most
    # recent Gulf Metals invoice as a re-issue of the one before it.
    metals_positions = [
        index for index, item in enumerate(items) if item.spec.vendor == VENDORS[0].name
    ]
    source_index, planted_index = metals_positions[-2], metals_positions[-1]
    source = items[source_index].spec

    planted_spec = _make_spec(
        VENDORS[0],
        number=f"{source.invoice_number}-R",
        # Inside the 7-day duplicate window, so the detector must catch it.
        issued=source.issue_date + timedelta(days=3),
        rng=rng,
        lines=_near_duplicate_lines(source.lines, VENDORS[0].vat_rate),
        note="Re-issued copy of a previously submitted invoice.",
        tags=("seed", "planted-duplicate"),
    )
    _assert_within_duplicate_window(source, planted_spec)
    items[planted_index] = CorpusItem(
        spec=planted_spec,
        as_scan=False,
        expected_anomalies=frozenset({AnomalyType.DUPLICATE}),
    )
    assert len(items) == SEED_CORPUS_SIZE
    return items


def _draw_eval_corpus(reference: date | None, seed: int) -> list[CorpusItem]:
    """10 labelled invoices: 2 near-duplicates and 1 inflated amount (spec §8).

    Order matters — a duplicate can only be detected against a document the
    pipeline has already seen, so the pair is emitted original-then-copy.
    """
    today = reference or date(2026, 7, 20)
    rng = random.Random(seed + 1)
    metals, aurora, nakheel, cedar = VENDORS[0], VENDORS[1], VENDORS[2], VENDORS[3]
    items: list[CorpusItem] = []

    # 1-4: Gulf Metals history on a stable recurring basket, so the inflated
    # invoice at the end has a tight baseline to stand out against.
    metals_basket = _vendor_basket(metals, rng)
    metals_totals: set[Decimal] = set()
    for index in range(4):
        issued = today - timedelta(days=(6 - index) * 21)
        items.append(
            CorpusItem(
                spec=_make_spec(
                    metals,
                    number=f"EV-GM-{index + 1:03d}",
                    issued=issued,
                    rng=rng,
                    lines=_distinct_variation(metals, metals_basket, rng, metals_totals),
                    tags=("eval", "normal"),
                ),
                # Mixed media: the vision lane must be exercised too.
                as_scan=index in {1, 3},
                expected_anomalies=frozenset(),
            )
        )

    # 5-6: two other vendors, clean.
    for index, profile in enumerate((nakheel, cedar)):
        items.append(
            CorpusItem(
                spec=_make_spec(
                    profile,
                    number=f"EV-{profile.name[:2].upper()}-{index + 1:03d}",
                    issued=today - timedelta(days=30 + index * 9),
                    rng=rng,
                    lines=_resample_until_not_round(profile, rng),
                    tags=("eval", "normal"),
                ),
                as_scan=index == 1,
                expected_anomalies=frozenset(),
            )
        )

    # 7: Aurora original — the first half of the near-duplicate pair.
    aurora_issue = today - timedelta(days=18)
    aurora_lines = _resample_until_not_round(aurora, rng)
    items.append(
        CorpusItem(
            spec=_make_spec(
                aurora,
                number="EV-AU-101",
                issued=aurora_issue,
                rng=rng,
                lines=aurora_lines,
                tags=("eval", "duplicate-original"),
            ),
            as_scan=False,
            expected_anomalies=frozenset(),
        )
    )

    # 8: Aurora near-duplicate — 4 days later, ~0.4% apart, new reference number.
    aurora_copy = _make_spec(
        aurora,
        number="EV-AU-102",
        issued=aurora_issue + timedelta(days=4),
        rng=rng,
        lines=_near_duplicate_lines(aurora_lines, aurora.vat_rate),
        note="Duplicate submission of the same engagement.",
        tags=("eval", "duplicate-copy"),
    )
    _assert_within_duplicate_window(items[-1].spec, aurora_copy)
    items.append(
        CorpusItem(
            spec=aurora_copy,
            as_scan=False,
            expected_anomalies=frozenset({AnomalyType.DUPLICATE}),
        )
    )

    # 9: broken arithmetic — must route to NEEDS_REVIEW, not be silently repaired.
    broken_lines = _resample_until_not_round(cedar, rng)
    broken = _make_spec(
        cedar,
        number="EV-CP-901",
        issued=today - timedelta(days=11),
        rng=rng,
        lines=broken_lines,
        tags=("eval", "arithmetic-defect"),
    )
    # The printed total disagrees with subtotal + tax by a wide margin, so the
    # deterministic validator must reject it rather than the model "fixing" it.
    broken = replace(broken, forced_total=broken.subtotal + broken.tax + Decimal("450.00"))
    items.append(CorpusItem(spec=broken, as_scan=False, expected_anomalies=frozenset()))

    # 10: inflated amount — same vendor as 1-4, roughly 8x their usual spend.
    baseline = sum(
        (item.spec.total for item in items if item.spec.vendor == metals.name),
        start=Decimal("0"),
    ) / Decimal(4)
    inflated_lines = (
        LineSpec("Steel plate 10mm x 2400", Decimal("96"), Decimal("420.00")),
        LineSpec("Stainless sheet 1.5mm", Decimal("41"), Decimal("534.20")),
    )
    items.append(
        CorpusItem(
            spec=_make_spec(
                metals,
                number="EV-GM-999",
                issued=today - timedelta(days=3),
                rng=rng,
                lines=inflated_lines,
                note=f"Baseline for this vendor is approximately {baseline:,.0f}.",
                tags=("eval", "inflated-amount"),
            ),
            as_scan=False,
            expected_anomalies=frozenset({AnomalyType.AMOUNT_ZSCORE}),
        )
    )

    return items


# ---------------------------------------------------------------------------
# Self-verification
# ---------------------------------------------------------------------------


def _unexpected_findings(items: list[CorpusItem]) -> list[str]:
    """Run the *real* detector over a corpus and report label violations.

    Statistical guarantees are fragile: with only four priors a vendor whose
    fifth invoice lands at the edge of an otherwise reasonable spend band can
    still exceed two standard deviations. Rather than hand-tuning constants until
    a particular seed happens to behave, the corpus is checked against the engine
    that will actually score it, and redrawn if it does not match its own labels.

    This also means the corpora self-heal if a detector threshold is ever tuned.
    """
    config = ScreeningConfig()
    records = [
        InvoiceRecord(
            document_id=uuid.uuid5(uuid.NAMESPACE_OID, item.spec.invoice_number),
            vendor=item.spec.vendor,
            invoice_number=item.spec.invoice_number,
            issue_date=item.spec.issue_date,
            total=item.spec.total,
            currency=item.spec.currency,
            payment_terms=item.spec.payment_terms,
            filename=item.filename,
        )
        for item in items
    ]

    violations: list[str] = []
    for index, (item, record) in enumerate(zip(items, records, strict=True)):
        # Documents are screened against only what the pipeline had already seen.
        observed = {
            finding.anomaly_type for finding in screen_invoice(record, records[:index], config)
        }
        expected = set(item.expected_anomalies)
        if observed != expected:
            violations.append(
                f"{item.spec.invoice_number}: expected {sorted(expected)}, got {sorted(observed)}"
            )
    return violations


def _build_verified(
    draw: Callable[[date | None, int], list[CorpusItem]],
    reference: date | None,
    attempts: int = 40,
) -> list[CorpusItem]:
    """Redraw with successive seeds until the corpus matches its own labels."""
    last: list[CorpusItem] = []
    for offset in range(attempts):
        last = draw(reference, SEED + offset * 101)
        if not _unexpected_findings(last):
            return last
    problems = "; ".join(_unexpected_findings(last))
    msg = f"Could not generate a corpus matching its labels after {attempts} draws: {problems}"
    raise RuntimeError(msg)


def build_seed_corpus(reference: date | None = None) -> list[CorpusItem]:
    """30 historical invoices across 6 vendors, verified against the real detector."""
    return _build_verified(_draw_seed_corpus, reference)


def build_eval_corpus(reference: date | None = None) -> list[CorpusItem]:
    """10 labelled invoices, verified against the real detector."""
    return _build_verified(_draw_eval_corpus, reference)