Coding Problems

import csv
from datetime import datetime
from typing import Iterable

DATE_FORMATS = ("%Y-%m-%d", "%m/%d/%Y", "%d-%b-%Y", "%B %d, %Y", "%d/%m/%Y")
CORP_SUFFIXES = (", Inc.", " Inc.", ", LLC", " LLC", ", Ltd", " Ltd", " Corp.", ", Corp.")

def clean_date(s: str | None) -> str | None:
    if not s or not s.strip():
        return None
    s = s.strip()
    for fmt in DATE_FORMATS:
        try:
            return datetime.strptime(s, fmt).date().isoformat()
        except ValueError:
            continue
    return None

def normalize_supplier(name: str | None) -> str | None:
    if not name: return None
    cleaned = " ".join(name.split())
    for suffix in CORP_SUFFIXES:
        if cleaned.lower().endswith(suffix.lower()):
            cleaned = cleaned[:-len(suffix)].strip()
            break
    return cleaned or None

def parse_value(v: str | None) -> float | None:
    if not v or not v.strip():
        return None
    cleaned = v.replace("$", "").replace(",", "").strip()
    try:
        return float(cleaned)
    except ValueError:
        return None

def clean_contracts(rows: Iterable[dict]) -> list[dict]:
    return [
        {
            "contract_id": row["id"].strip(),
            "supplier": normalize_supplier(row.get("supplier_name")),
            "effective_date": clean_date(row.get("effective_date")),
            "value_usd": parse_value(row.get("value")),
        }
        for row in rows
    ]

import json
from typing import Iterable

def reshape(extractions: Iterable[dict]) -> Iterable[dict]:
    for e in extractions:
        doc_id = e["document_id"]
        for field_name, payload in e.get("fields", {}).items():
            value = payload.get("value")
            # Stringify list values for narrow storage (or split if downstream requires)
            value_str = json.dumps(value) if isinstance(value, list) else value
            yield {
                "document_id": doc_id,
                "field_name": field_name,
                "value": value_str,
                "confidence": payload.get("confidence"),
            }

-- Approach 1: anti-join via LEFT JOIN
SELECT a.agreement_id, a.parent_agreement_id
FROM agreements a
LEFT JOIN agreements parent
  ON parent.agreement_id = a.parent_agreement_id
WHERE a.parent_agreement_id IS NOT NULL
  AND parent.agreement_id IS NULL;

-- Approach 2: NOT EXISTS (NULL-safer)
SELECT a.agreement_id, a.parent_agreement_id
FROM agreements a
WHERE a.parent_agreement_id IS NOT NULL
  AND NOT EXISTS (
    SELECT 1 FROM agreements p WHERE p.agreement_id = a.parent_agreement_id
  );

from difflib import SequenceMatcher

def similarity(a: str, b: str) -> float:
    return SequenceMatcher(None, a.lower().strip(), b.lower().strip()).ratio()

def match_suppliers(
    raw_names: list[str],
    canonical: list[str],
    threshold: float = 0.8,
) -> list[dict]:
    """Map each raw name to its best canonical match above threshold, or None."""
    results = []
    for raw in raw_names:
        scored = [(c, similarity(raw, c)) for c in canonical]
        scored.sort(key=lambda x: -x[1])
        if scored and scored[0][1] >= threshold:
            results.append({"raw": raw, "canonical": scored[0][0], "confidence": scored[0][1]})
        else:
            results.append({"raw": raw, "canonical": None, "confidence": scored[0][1] if scored else 0.0})
    return results

WITH RECURSIVE chain AS (
  SELECT supplier_id, supplier_id AS root_id, 0 AS depth
  FROM suppliers WHERE parent_supplier_id IS NULL
  UNION ALL
  SELECT s.supplier_id, c.root_id, c.depth + 1
  FROM suppliers s
  JOIN chain c ON c.supplier_id = s.parent_supplier_id
)
SELECT supplier_id, root_id, depth FROM chain;

from collections import defaultdict
from datetime import datetime

def dedupe(extractions: list[dict]) -> list[dict]:
    """Keep latest per (document_id, field). Flag drops > 0.1."""
    grouped = defaultdict(list)
    for e in extractions:
        grouped[(e["document_id"], e["field"])].append(e)

    result = []
    for key, items in grouped.items():
        items.sort(key=lambda x: x["extracted_at"], reverse=True)
        latest = items[0]
        result.append(latest)
        # If a prior extraction had meaningfully higher confidence, flag both
        if len(items) > 1:
            prior = items[1]
            if prior["confidence"] - latest["confidence"] > 0.1:
                result[-1]["confidence_regressed"] = True
                result.append({**prior, "shadow": True})
    return result

SELECT
  sp.supplier_id,
  SUM(sp.amount) AS off_contract_total
FROM spend sp
LEFT JOIN agreements a
  ON a.supplier_id = sp.supplier_id
 -- no status filter: status is today's state, so the date window decides coverage
 AND sp.accrued_at BETWEEN a.effective_date AND COALESCE(a.current_term_end, '9999-12-31')
WHERE a.agreement_id IS NULL
GROUP BY sp.supplier_id
ORDER BY off_contract_total DESC;

from collections import defaultdict

def field_accuracy(
    extracted: list[dict],
    gold: list[dict],
    tolerance: dict[str, float] | None = None,
) -> dict[str, float]:
    """
    extracted, gold: list of {document_id, field, value}
    tolerance: per-field tolerance (e.g., {"amount": 0.01} for 1% of value)
    Returns per-field accuracy.
    """
    gold_lookup = {(g["document_id"], g["field"]): g["value"] for g in gold}
    per_field_correct: dict[str, int] = defaultdict(int)
    per_field_total: dict[str, int] = defaultdict(int)
    for e in extracted:
        key = (e["document_id"], e["field"])
        if key not in gold_lookup:
            continue
        per_field_total[e["field"]] += 1
        gold_val = gold_lookup[key]
        if matches(e["value"], gold_val, tolerance.get(e["field"]) if tolerance else None):
            per_field_correct[e["field"]] += 1
    return {f: per_field_correct[f] / per_field_total[f] for f in per_field_total}

def matches(extracted, gold, tol: float | None) -> bool:
    if extracted == gold:
        return True
    if tol is not None and isinstance(extracted, (int, float)) and isinstance(gold, (int, float)):
        return abs(extracted - gold) <= abs(gold) * tol
    return False

from collections import defaultdict

def reliability(extracted: list[dict], n_bins: int = 10) -> list[dict]:
    """Each extracted item: {confidence: float, was_correct: bool}"""
    buckets = defaultdict(list)
    for e in extracted:
        bucket = min(int(e["confidence"] * n_bins), n_bins - 1)
        buckets[bucket].append(e)
    out = []
    for b in range(n_bins):
        items = buckets.get(b, [])
        if not items:
            continue
        mean_conf = sum(i["confidence"] for i in items) / len(items)
        mean_acc = sum(1 for i in items if i["was_correct"]) / len(items)
        out.append({
            "bucket": b, "n": len(items),
            "mean_confidence": mean_conf, "actual_accuracy": mean_acc,
            "gap": mean_acc - mean_conf,
        })
    return out