"""Holdout Data Contract v1: validator for client data folders.

Usage (toolkit):   holdout-audit validate <folder> [--json] [--max-errors N]
Usage (one file):  python holdout_validate.py <folder>

The contract itself is documented in docs/DATA-CONTRACT.md (public copy: holdoutlabs.io/data-contract/).
This module uses the Python standard library only, so the same file can be handed to a client and run
without installing anything. Parquet input additionally needs pandas with pyarrow.

Files (CSV, UTF-8, comma-separated, header row; or the same name with .parquet):
  returns.csv      Tier 1, always required
  instruments.csv  required whenever positions, trades or baskets are sent
  positions.csv    Tier 2
  trades.csv       optional
  baskets.csv      optional (custom baskets / swap look-through)

Exit code 0 when there are no errors (warnings allowed), 1 otherwise.
"""

from __future__ import annotations

import argparse
import csv
import io
import json
import math
import re
import sys
from dataclasses import dataclass, field
from datetime import date, datetime
from pathlib import Path

CONTRACT_VERSION = "1.0"

ID_TYPES = ("ISIN", "CUSIP", "FIGI", "TICKER_MIC", "OCC", "FUTURES", "CUSTOM")
ASSET_CLASSES = ("equity", "etf", "index", "future", "option", "fx", "bond", "commodity", "swap", "basket",
                 "cash", "other")
# Asset classes whose market_value is a notional exposure (margined); their P&L sits in cash, so they are
# left out when positions are summed to NAV.
NOTIONAL_CLASSES = ("future", "swap")

RETURN_TOL = 1e-4          # |return - NAV-implied return|, i.e. 1 basis point
MV_REL_TOL = 0.005         # market_value vs price x quantity x multiplier: 0.5% relative ...
MV_ABS_TOL = 0.01          # ... or one cent, whichever is larger
WEIGHT_TOL = 0.01          # basket weights must sum to 1 +/- this
NAV_RECON_TOL = 0.005      # sum of position values vs NAV, when cash rows are supplied
MIN_AUDIT_PERIODS = 60


@dataclass(frozen=True)
class Field:
    name: str
    kind: str              # date | number | text | enum | ccy | putcall
    required: bool = False  # the column must exist and every cell must be filled
    enum: tuple = ()
    doc: str = ""


SPEC: dict[str, list[Field]] = {
    "returns": [
        Field("date", "date", True, doc="Period end date, ISO-8601 (YYYY-MM-DD, or a full timestamp for intraday)."),
        Field("nav", "number", doc="Net asset value at the period end, in the base currency (> 0). Send nav or return, or both."),
        Field("return", "number", doc="Simple decimal return for the period (0.012 = +1.2%), time-weighted, net of costs unless agreed."),
        Field("flow", "number", doc="External cash flow at the period end (+ subscription, - redemption), base currency. Used with nav."),
        Field("currency", "ccy", True, doc="Base currency (ISO 4217), the same on every row."),
    ],
    "instruments": [
        Field("instrument_id", "text", True, doc="Your identifier; must match positions, trades and baskets exactly."),
        Field("id_type", "enum", True, ID_TYPES, doc="What kind of identifier instrument_id is."),
        Field("description", "text", doc="Free text. Leave blank or generic if anonymising."),
        Field("asset_class", "enum", True, ASSET_CLASSES, doc="Instrument type."),
        Field("currency", "ccy", True, doc="Currency the instrument is priced in (ISO 4217)."),
        Field("multiplier", "number", doc="Contract multiplier (> 0); blank means 1."),
        Field("underlying_id", "text", doc="For derivatives and swaps: the instrument_id of the underlying."),
        Field("expiry", "date", doc="Expiry / maturity date; required for futures and options."),
        Field("strike", "number", doc="Option strike; required for options."),
        Field("put_call", "putcall", doc="P or C; required for options."),
        Field("exchange", "text", doc="Listing venue, ideally an ISO 10383 MIC (e.g. XNYS)."),
    ],
    "positions": [
        Field("date", "date", True, doc="Holdings date (end of day), ISO-8601."),
        Field("instrument_id", "text", True, doc="Must exist in instruments.csv."),
        Field("quantity", "number", True, doc="Signed units or contracts; negative = short. For cash, the amount."),
        Field("price", "number", doc="Price per unit in the instrument currency. Optional if market_value is given."),
        Field("market_value", "number", doc="Position value in the base currency (notional for futures and swaps). Optional if price is given."),
        Field("currency", "ccy", True, doc="Currency of price (the instrument currency)."),
        Field("sleeve", "text", doc="Optional sub-strategy / book label."),
        Field("delta", "number", doc="Options: per-unit delta of the option (-1 to 1)."),
    ],
    "trades": [
        Field("trade_date", "date", True, doc="Trade date, ISO-8601."),
        Field("instrument_id", "text", True, doc="Must exist in instruments.csv."),
        Field("quantity", "number", True, doc="Signed: + buy, - sell. Never zero."),
        Field("price", "number", True, doc="Execution price in the instrument currency."),
        Field("fees", "number", doc="Commissions and fees for the trade, base currency, as a positive number."),
        Field("sleeve", "text", doc="Optional sub-strategy / book label."),
        Field("trade_id", "text", doc="Optional unique trade reference."),
    ],
    "baskets": [
        Field("basket_id", "text", True, doc="instrument_id of the basket (or swap reference basket) in instruments.csv."),
        Field("date", "date", True, doc="Composition date (effective from this date until the next one)."),
        Field("constituent_id", "text", True, doc="instrument_id of the constituent; must exist in instruments.csv."),
        Field("weight", "number", doc="Decimal weight (weights per basket and date sum to 1). Send weight or units."),
        Field("units", "number", doc="Units of the constituent per one unit of the basket. Send weight or units."),
    ],
}
FILE_ORDER = ("returns", "instruments", "positions", "trades", "baskets")

_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}([T ]\d{2}:\d{2}(:\d{2}(\.\d+)?)?(Z|[+-]\d{2}:?\d{2})?)?$")
_CCY_RE = re.compile(r"^[A-Z]{3}$")
_ID_FORMATS = {
    "ISIN": (re.compile(r"^[A-Z]{2}[A-Z0-9]{9}[0-9]$"), "12 characters: 2-letter country, 9 alphanumerics, check digit"),
    "CUSIP": (re.compile(r"^[A-Z0-9]{8}[0-9]$"), "9 characters: 8 alphanumerics and a check digit"),
    "FIGI": (re.compile(r"^BBG[A-Z0-9]{9}$"), "12 characters starting with BBG"),
    "TICKER_MIC": (re.compile(r"^[A-Z0-9][A-Z0-9./-]*\.[A-Z0-9]{4}$"), "TICKER.MIC, e.g. SPY.ARCX"),
}


# ----------------------------------------------------------------------------------------------- results
@dataclass
class Issue:
    level: str        # "error" | "warning"
    file: str         # e.g. "positions.csv" or "" for folder-level
    line: int | None  # 1-based line in the file (header = 1); None for file-level
    message: str

    def where(self) -> str:
        if not self.file:
            return "folder"
        return f"{self.file}:{self.line}" if self.line else self.file

    def to_dict(self) -> dict:
        return {"level": self.level, "file": self.file, "line": self.line, "message": self.message}


@dataclass
class Report:
    folder: str
    files: dict[str, dict] = field(default_factory=dict)   # name -> {"path", "rows"}
    issues: list[Issue] = field(default_factory=list)
    base_currency: str | None = None

    @property
    def errors(self) -> list[Issue]:
        return [i for i in self.issues if i.level == "error"]

    @property
    def warnings(self) -> list[Issue]:
        return [i for i in self.issues if i.level == "warning"]

    @property
    def ok(self) -> bool:
        return not self.errors

    @property
    def tier(self) -> str:
        if "positions" in self.files or "trades" in self.files:
            return "Tier 2 (positions / trades)"
        return "Tier 1 (returns only)"

    def to_dict(self) -> dict:
        return {"contract_version": CONTRACT_VERSION, "folder": self.folder, "result": "PASS" if self.ok else "FAIL",
                "tier": self.tier, "base_currency": self.base_currency,
                "files": {k: v["rows"] for k, v in self.files.items()},
                "errors": len(self.errors), "warnings": len(self.warnings),
                "issues": [i.to_dict() for i in self.issues]}


class _Ctx:
    def __init__(self, report: Report, fname: str):
        self.r, self.fname = report, fname

    def err(self, line, msg):
        self.r.issues.append(Issue("error", self.fname, line, msg))

    def warn(self, line, msg):
        self.r.issues.append(Issue("warning", self.fname, line, msg))


# ----------------------------------------------------------------------------------------------- parsing
def _parse_date(s: str):
    if not _DATE_RE.match(s):
        return None
    try:
        if len(s) == 10:
            return date.fromisoformat(s)
        return datetime.fromisoformat(s.replace("Z", "+00:00"))
    except ValueError:
        return None


def _day(d) -> date:
    return d.date() if isinstance(d, datetime) else d


def _parse_number(s: str):
    try:
        v = float(s)
    except ValueError:
        return None
    return v if math.isfinite(v) else None


def _number_hint(s: str) -> str:
    if s.endswith("%"):
        return " (drop the % sign and send decimals: 1.2% -> 0.012)"
    if re.match(r"^-?[\d.]+,\d", s) or re.match(r"^-?\d{1,3}(,\d{3})+", s):
        return " (no thousands separators or decimal commas: use 1234.5)"
    if s.startswith("(") and s.endswith(")"):
        return " (use a minus sign, not brackets, for negatives)"
    return ""


def _read_table(path: Path, ctx: _Ctx):
    """Return (header, rows) where rows are (line_no, dict of str) or None if unreadable."""
    if path.suffix.lower() == ".parquet":
        try:
            import pandas as pd  # noqa: PLC0415
            df = pd.read_parquet(path)
        except Exception as e:  # pragma: no cover - depends on optional engine
            ctx.err(None, f"cannot read Parquet ({e}); install pandas and pyarrow, or send CSV")
            return None
        header = [str(c).strip().lower() for c in df.columns]
        rows = []
        for i, rec in enumerate(df.itertuples(index=False), start=2):
            vals = {}
            for h, v in zip(header, rec):
                if v is None or (isinstance(v, float) and math.isnan(v)):
                    vals[h] = ""
                elif hasattr(v, "isoformat"):
                    iso = v.isoformat()
                    vals[h] = iso[:10] if iso.endswith("T00:00:00") else iso
                else:
                    vals[h] = str(v).strip()
            rows.append((i, vals))
        return header, rows

    raw = path.read_bytes()
    if raw.startswith(b"\xef\xbb\xbf"):
        ctx.warn(None, "file starts with a UTF-8 byte-order mark (common in Excel exports); accepted")
        raw = raw[3:]
    try:
        text = raw.decode("utf-8")
    except UnicodeDecodeError as e:
        ctx.err(None, f"not valid UTF-8 (byte {e.start}); re-save as 'CSV UTF-8'")
        return None
    lines = text.splitlines()
    if not lines or not lines[0].strip():
        ctx.err(1, "file is empty or has no header row")
        return None
    first = lines[0]
    if "," not in first and (";" in first or "\t" in first):
        ctx.err(1, "header is not comma-separated (found ';' or tab); use commas")
        return None
    reader = csv.reader(io.StringIO(text))
    header = [h.strip().lower() for h in next(reader)]
    rows = []
    for rec in reader:
        line = reader.line_num
        if not any(c.strip() for c in rec):
            continue  # blank line
        if len(rec) != len(header):
            ctx.err(line, f"has {len(rec)} fields but the header has {len(header)} "
                          "(stray comma, unquoted text or merged cells?)")
            continue
        rows.append((line, {h: c.strip() for h, c in zip(header, rec)}))
    return header, rows


def _find(folder: Path, name: str, report: Report):
    found = [p for p in (folder / f"{name}.csv", folder / f"{name}.parquet") if p.exists()]
    if len(found) > 1:
        report.issues.append(Issue("error", found[0].name, None,
                                   f"both {name}.csv and {name}.parquet exist; send one"))
    return found[0] if found else None


def _typed_rows(fname: str, spec: list[Field], header, rows, ctx: _Ctx):
    """Check columns and cell types; return list of (line, dict of parsed values)."""
    names = [f.name for f in spec]
    dups = sorted({h for h in header if header.count(h) > 1})
    for h in dups:
        ctx.err(1, f"column '{h}' appears more than once")
    missing = [f.name for f in spec if f.required and f.name not in header]
    for name in missing:
        ctx.err(1, f"missing required column '{name}'")
    extra = [h for h in header if h not in names and h]
    if extra:
        ctx.warn(1, f"unknown column(s) ignored: {', '.join(extra)}")
    present = [f for f in spec if f.name in header]
    out = []
    for line, vals in rows:
        parsed, bad = {}, False
        for f in present:
            s = vals.get(f.name, "")
            if s == "":
                if f.required:
                    ctx.err(line, f"'{f.name}' is empty")
                    bad = True
                parsed[f.name] = None
                continue
            if f.kind == "date":
                v = _parse_date(s)
                if v is None:
                    hint = " (looks like an Excel serial number)" if re.match(r"^\d{5}(\.\d+)?$", s) else ""
                    ctx.err(line, f"'{f.name}' = '{s}' is not an ISO-8601 date (YYYY-MM-DD){hint}")
                    bad = True
            elif f.kind == "number":
                v = _parse_number(s)
                if v is None:
                    ctx.err(line, f"'{f.name}' = '{s}' is not a number{_number_hint(s)}")
                    bad = True
            elif f.kind == "enum":
                v = s
                if s not in f.enum:
                    low = {e.lower(): e for e in f.enum}
                    hint = f" (did you mean '{low[s.lower()]}'?)" if s.lower() in low else ""
                    ctx.err(line, f"'{f.name}' = '{s}' is not one of {', '.join(f.enum)}{hint}")
                    bad = True
            elif f.kind == "ccy":
                v = s
                if not _CCY_RE.match(s):
                    ctx.err(line, f"'{f.name}' = '{s}' is not a 3-letter ISO 4217 code (e.g. USD)")
                    bad = True
            elif f.kind == "putcall":
                v = s
                if s not in ("P", "C"):
                    ctx.err(line, f"'{f.name}' = '{s}' must be P or C")
                    bad = True
            else:
                v = s
            parsed[f.name] = v
        out.append((line, parsed, bad or bool(missing)))  # rows are unusable for cross-checks without them
    return out


# ----------------------------------------------------------------------------------------------- checks
def _check_returns(rows, ctx: _Ctx, report: Report):
    cols = set().union(*(p.keys() for _, p, _ in rows)) if rows else set()
    has_nav, has_ret = "nav" in cols, "return" in cols
    if rows and not (has_nav or has_ret):
        ctx.err(1, "needs a 'nav' column or a 'return' column (or both)")
        return {}
    good = [(ln, p) for ln, p, bad in rows if not bad]
    ccys = sorted({p["currency"] for _, p in good if p.get("currency")})
    if len(ccys) > 1:
        ctx.err(None, f"more than one base currency ({', '.join(ccys)}); declare one base currency")
    report.base_currency = ccys[0] if len(ccys) == 1 else None

    seen, prev_d = {}, None
    for ln, p in good:
        d = p["date"]
        if d in seen:
            ctx.err(ln, f"duplicate date {d} (first on line {seen[d]})")
        else:
            seen[d] = ln
        if prev_d is not None and d < prev_d:
            ctx.warn(ln, f"date {d} is earlier than the row above; rows should be in date order")
        prev_d = d if prev_d is None or d > prev_d else prev_d

    rets = [p["return"] for _, p in good if p.get("return") is not None]
    if rets:
        med = sorted(abs(r) for r in rets)[len(rets) // 2]
        if med > 0.2:
            ctx.err(None, f"returns look like percentages (median |return| = {med:.3g}); send decimals: 1.2% -> 0.012")
            rets = []
    for ln, p in good:
        if has_nav and p.get("nav") is not None and p["nav"] <= 0:
            ctx.err(ln, f"nav = {p['nav']} must be positive")
        r = p.get("return")
        if r is not None and rets:
            if r <= -1:
                ctx.err(ln, f"return = {r} is a loss of 100% or more")
            elif abs(r) > 1:
                ctx.warn(ln, f"return = {r} is more than 100% in one period; check units")

    first_line = good[0][0] if good else None
    for ln, p in good:
        if has_nav and p.get("nav") is None:
            ctx.err(ln, "'nav' is empty (when a nav column is sent, fill every row)")
        if has_ret and not has_nav and p.get("return") is None:
            ctx.err(ln, "'return' is empty")
        if has_ret and has_nav and p.get("return") is None and ln != first_line:
            ctx.err(ln, "'return' is empty (only the first row may be blank when nav is sent)")

    if "flow" in cols and not has_nav:
        ctx.warn(None, "'flow' is ignored without 'nav': a return-only file should already exclude flows")
    if has_nav and has_ret and rets:
        prev = None
        for ln, p in good:
            if prev is not None and p.get("nav") and prev.get("nav") and p.get("return") is not None:
                implied = (p["nav"] - (p.get("flow") or 0.0)) / prev["nav"] - 1
                if abs(implied - p["return"]) > RETURN_TOL:
                    ctx.err(ln, f"return {p['return']:.6f} does not match NAV: (nav - flow) / previous nav - 1 = "
                                f"{implied:.6f} (tolerance {RETURN_TOL:g})")
            prev = p
    if 0 < len(good) < MIN_AUDIT_PERIODS:
        ctx.warn(None, f"{len(good)} periods; an audit needs at least {MIN_AUDIT_PERIODS} (2+ years recommended)")
    return {(_day(p["date"])): p for _, p in good}


def _check_id_format(ln, p, ctx: _Ctx):
    t, iid = p.get("id_type"), p.get("instrument_id") or ""
    if t in _ID_FORMATS:
        rx, desc = _ID_FORMATS[t]
        if not rx.match(iid):
            ctx.err(ln, f"instrument_id '{iid}' does not look like a {t} ({desc}); use CUSTOM for masked or internal ids")
        elif t == "ISIN" and not _isin_ok(iid):
            ctx.warn(ln, f"ISIN '{iid}' fails its check digit")


def _isin_ok(isin: str) -> bool:
    digits = "".join(str(int(c, 36)) for c in isin[:-1])
    total = 0
    for i, ch in enumerate(reversed(digits)):
        n = int(ch)
        if i % 2 == 0:
            n *= 2
            n = n - 9 if n > 9 else n
        total += n
    return (10 - total % 10) % 10 == int(isin[-1])


def _check_instruments(rows, ctx: _Ctx):
    inst, seen = {}, {}
    for ln, p, bad in rows:
        iid = p.get("instrument_id")
        if iid:
            if iid in seen:
                ctx.err(ln, f"duplicate instrument_id '{iid}' (first on line {seen[iid]})")
                continue
            seen[iid] = ln
        if bad:
            continue
        _check_id_format(ln, p, ctx)
        m = p.get("multiplier")
        if m is not None and m <= 0:
            ctx.err(ln, f"multiplier = {m} must be positive")
        ac = p["asset_class"]
        if ac in ("future", "option") and p.get("expiry") is None:
            ctx.err(ln, f"{ac} '{iid}' needs an expiry date")
        if ac == "future" and m is None:
            ctx.warn(ln, f"future '{iid}' has no multiplier; 1 is assumed")
        if ac == "option":
            if p.get("strike") is None:
                ctx.err(ln, f"option '{iid}' needs a strike")
            elif p["strike"] <= 0:
                ctx.err(ln, f"option '{iid}' strike must be positive")
            if p.get("put_call") is None:
                ctx.err(ln, f"option '{iid}' needs put_call (P or C)")
            if m is None:
                ctx.warn(ln, f"option '{iid}' has no multiplier; 1 is assumed (US equity options are usually 100)")
        inst[iid] = dict(p, _line=ln)
    for iid, p in inst.items():
        u = p.get("underlying_id")
        if u and u not in inst:
            ctx.warn(p["_line"], f"underlying_id '{u}' is not in instruments.csv (add it to link exposure to it)")
    return inst


def _require_instrument(ln, iid, inst, ctx: _Ctx, col="instrument_id"):
    if iid not in inst:
        ctx.err(ln, f"{col} '{iid}' is not in instruments.csv")
        return None
    return inst[iid]


def _check_positions(rows, ctx: _Ctx, inst, returns_by_day, base):
    seen, sums, has_cash, off_calendar = {}, {}, set(), []
    for ln, p, bad in rows:
        if bad:
            continue
        iid, qty = p["instrument_id"], p["quantity"]
        key = (p["date"], iid, p.get("sleeve") or "")
        if key in seen:
            ctx.err(ln, f"duplicate position for {iid} on {p['date']}"
                        f"{' in sleeve ' + key[2] if key[2] else ''} (first on line {seen[key]})")
            continue
        seen[key] = ln
        price, mv = p.get("price"), p.get("market_value")
        if price is None and mv is None:
            ctx.err(ln, "needs price or market_value (or both)")
        if qty == 0:
            ctx.warn(ln, "quantity is 0; leave closed positions out")
        if price is not None and price < 0:
            ctx.warn(ln, f"negative price {price}; fine for spreads or some futures, otherwise check the sign convention")
        d = _day(p["date"])
        if returns_by_day and d not in returns_by_day:
            off_calendar.append(ln)
        spec = _require_instrument(ln, iid, inst, ctx) if inst is not None else None
        if spec is None:
            continue
        ac = spec["asset_class"]
        if p["currency"] != spec["currency"]:
            ctx.err(ln, f"currency {p['currency']} differs from {iid}'s currency {spec['currency']} in instruments.csv "
                        "(position currency is the price currency)")
        mult = spec.get("multiplier") or 1.0
        checked = False
        if price is not None and mv is not None and base and p["currency"] == base:
            expected = qty * price * mult
            if abs(mv - expected) > max(MV_ABS_TOL, MV_REL_TOL * abs(expected)):
                ctx.err(ln, f"market_value {mv:,.2f} != quantity x price x multiplier = {qty:g} x {price:g} x {mult:g} "
                            f"= {expected:,.2f} (tolerance 0.5%)")
            checked = True
        if not checked and mv not in (None, 0) and qty != 0 and (price is None or price >= 0):
            if (mv > 0) != (qty > 0):
                ctx.err(ln, f"sign of market_value ({mv:,.2f}) does not match sign of quantity ({qty:g}); "
                            "shorts need negative quantity and negative market_value")
        if p.get("delta") is not None:
            if ac != "option":
                ctx.warn(ln, f"delta given for a non-option ({ac}); ignored")
            elif abs(p["delta"]) > 1:
                ctx.err(ln, f"delta = {p['delta']} should be the per-unit option delta, between -1 and 1")
        elif ac == "option":
            ctx.warn(ln, f"option '{iid}' has no delta; delta exposure will need a model and underlying prices")
        if mv is not None and ac not in NOTIONAL_CLASSES:
            sums[d] = sums.get(d, 0.0) + mv
        if ac == "cash":
            has_cash.add(d)
    if off_calendar:
        ctx.warn(off_calendar[0], f"{len(off_calendar)} position row(s) have dates not in returns.csv "
                                  f"(first on line {off_calendar[0]})")
    # Positions -> NAV reconciliation: only when cash rows are supplied (the holdings then claim to be complete).
    for d in sorted(has_cash):
        r = returns_by_day.get(d) if returns_by_day else None
        if r and r.get("nav"):
            gap = sums[d] - r["nav"]
            if abs(gap) > NAV_RECON_TOL * r["nav"]:
                ctx.warn(None, f"{d}: positions sum to {sums[d]:,.2f} (futures and swaps excluded) but NAV is "
                               f"{r['nav']:,.2f} ({gap / r['nav']:+.2%}); missing cash, accruals or positions?")


def _check_trades(rows, ctx: _Ctx, inst):
    ids, exact = {}, {}
    for ln, p, bad in rows:
        if bad:
            continue
        tid = p.get("trade_id")
        if tid:
            if tid in ids:
                ctx.err(ln, f"duplicate trade_id '{tid}' (first on line {ids[tid]})")
            else:
                ids[tid] = ln
        sig = tuple(sorted((k, str(v)) for k, v in p.items()))
        if not tid and sig in exact:
            ctx.warn(ln, f"identical to line {exact[sig]}; if these are two fills, add a trade_id")
        exact.setdefault(sig, ln)
        if p["quantity"] == 0:
            ctx.err(ln, "quantity is 0; a trade needs a signed non-zero quantity (+ buy, - sell)")
        if p["price"] < 0:
            ctx.warn(ln, f"negative price {p['price']}; check the sign convention")
        if p.get("fees") is not None and p["fees"] < 0:
            ctx.warn(ln, "negative fees (a rebate?); fees are expected as positive costs")
        if inst is not None:
            _require_instrument(ln, p["instrument_id"], inst, ctx)


def _check_baskets(rows, ctx: _Ctx, inst):
    cols = set().union(*(p.keys() for _, p, _ in rows)) if rows else set()
    if rows and not ({"weight", "units"} & cols):
        ctx.err(1, "needs a 'weight' column or a 'units' column")
        return
    seen, wsum, first_line = {}, {}, {}
    for ln, p, bad in rows:
        if bad:
            continue
        b, c, d = p["basket_id"], p["constituent_id"], p["date"]
        key = (b, d, c)
        if key in seen:
            ctx.err(ln, f"duplicate constituent '{c}' in basket '{b}' on {d} (first on line {seen[key]})")
            continue
        seen[key] = ln
        w, u = p.get("weight"), p.get("units")
        if w is None and u is None:
            ctx.err(ln, "needs weight or units")
        if w is not None and u is not None:
            ctx.err(ln, "give weight or units, not both")
        if w is not None:
            wsum[(b, d)] = wsum.get((b, d), 0.0) + w
            first_line.setdefault((b, d), ln)
        if inst is not None:
            spec = _require_instrument(ln, b, inst, ctx, "basket_id")
            if spec is not None and spec["asset_class"] not in ("basket", "swap", "index"):
                ctx.warn(ln, f"basket_id '{b}' has asset_class '{spec['asset_class']}' (expected basket, swap or index)")
            _require_instrument(ln, c, inst, ctx, "constituent_id")
    for (b, d), s in wsum.items():
        if abs(s - 1) > WEIGHT_TOL:
            hint = " (weights look like percent: use decimals)" if abs(s - 100) <= 100 * WEIGHT_TOL else \
                " (for long/short or leveraged baskets send units instead)"
            ctx.err(first_line[(b, d)], f"basket '{b}' weights on {d} sum to {s:.4f}, expected 1 +/- {WEIGHT_TOL}{hint}")


# ----------------------------------------------------------------------------------------------- driver
def validate(folder: str | Path) -> Report:
    folder = Path(folder)
    report = Report(folder=str(folder))
    if not folder.is_dir():
        report.issues.append(Issue("error", "", None, f"'{folder}' is not a folder"))
        return report
    paths = {name: _find(folder, name, report) for name in FILE_ORDER}
    if paths["returns"] is None:
        report.issues.append(Issue("error", "", None, "returns.csv is missing (it is required for every audit)"))
    needs_inst = [n for n in ("positions", "trades", "baskets") if paths[n] is not None]
    if needs_inst and paths["instruments"] is None:
        report.issues.append(Issue("error", "", None,
                                   f"instruments.csv is missing (required with {', '.join(n + '.csv' for n in needs_inst)})"))
    if paths["instruments"] is not None and not needs_inst:
        report.issues.append(Issue("warning", "", None, "instruments.csv sent without positions, trades or baskets; not used"))
    known = {f"{n}.{e}" for n in FILE_ORDER for e in ("csv", "parquet")}
    others = sorted(p.name for p in folder.iterdir() if p.is_file() and p.name.lower() not in known
                    and not p.name.startswith("."))
    if others:
        report.issues.append(Issue("warning", "", None, f"other files ignored: {', '.join(others)}"))

    parsed = {}
    for name in FILE_ORDER:
        path = paths[name]
        if path is None:
            continue
        ctx = _Ctx(report, path.name)
        table = _read_table(path, ctx)
        if table is None:
            report.files[name] = {"path": str(path), "rows": 0}
            continue
        header, rows = table
        report.files[name] = {"path": str(path), "rows": len(rows)}
        if not rows:
            ctx.err(None, "no data rows")
        parsed[name] = (ctx, _typed_rows(name, SPEC[name], header, rows, ctx))

    returns_by_day = _check_returns(parsed["returns"][1], parsed["returns"][0], report) if "returns" in parsed else {}
    inst = _check_instruments(parsed["instruments"][1], parsed["instruments"][0]) if "instruments" in parsed else None
    if "positions" in parsed:
        _check_positions(parsed["positions"][1], parsed["positions"][0], inst, returns_by_day, report.base_currency)
    if "trades" in parsed:
        _check_trades(parsed["trades"][1], parsed["trades"][0], inst)
    if "baskets" in parsed:
        _check_baskets(parsed["baskets"][1], parsed["baskets"][0], inst)
    order = {n: i for i, n in enumerate(FILE_ORDER)}
    report.issues.sort(key=lambda i: (order.get(i.file.split(".")[0], -1) if i.file else -1, i.line or 0))
    return report


def format_report(report: Report, max_errors: int = 20) -> str:
    out = [f"Holdout Data Contract v{CONTRACT_VERSION} validator", f"Folder: {report.folder}"]
    if report.files:
        out.append(f"Detected: {report.tier}; base currency {report.base_currency or '?'}")
    out.append("")
    for name in FILE_ORDER:
        if name not in report.files:
            continue
        fname = Path(report.files[name]["path"]).name
        errs = [i for i in report.errors if i.file == fname]
        warns = [i for i in report.warnings if i.file == fname]
        status = "FAIL" if errs else "PASS"
        tail = f"{len(errs)} error(s), {len(warns)} warning(s)" if (errs or warns) else "ok"
        out.append(f"  {status}  {fname:<17} {report.files[name]['rows']:>6} rows   {tail}")
    folder_issues = [i for i in report.issues if not i.file]
    shown: dict[str, int] = {}
    body = []
    for i in folder_issues + [i for i in report.issues if i.file]:
        k = i.file + i.level
        shown[k] = shown.get(k, 0) + 1
        if shown[k] <= max_errors:
            body.append(f"  {i.level.upper():<7} {i.where():<22} {i.message}")
        elif shown[k] == max_errors + 1:
            n = sum(1 for j in report.issues if j.file + j.level == k)
            body.append(f"  ...     {i.file or 'folder':<22} {n - max_errors} more {i.level}(s) not shown")
    if body:
        out += ["", *body]
    out += ["", f"RESULT: {'PASS' if report.ok else 'FAIL'} "
                f"({len(report.errors)} error(s), {len(report.warnings)} warning(s))"]
    if report.ok and report.warnings:
        out.append("Warnings do not block delivery; read them in case one points to a real problem.")
    return "\n".join(out)


def add_arguments(p: argparse.ArgumentParser) -> None:
    p.add_argument("folder", help="folder holding returns.csv and, for Tier 2, instruments/positions/trades/baskets")
    p.add_argument("--json", action="store_true", help="print a machine-readable report")
    p.add_argument("--max-errors", type=int, default=20, help="issues shown per file and level (default 20)")


def run(a: argparse.Namespace) -> int:
    report = validate(a.folder)
    if a.json:
        print(json.dumps(report.to_dict(), indent=2, default=str))
    else:
        print(format_report(report, a.max_errors))
    return 0 if report.ok else 1


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(prog="holdout_validate",
                                description="Check a data folder against the Holdout Data Contract v1.")
    add_arguments(p)
    return run(p.parse_args(argv))


if __name__ == "__main__":
    sys.exit(main())