eval/run_eval.py

"""FinSight v3 prompt & retrieval evaluation harness.

Zero-dependency (stdlib only), JSON-driven, and gracefully degrading so it can
run anywhere from a laptop with no credentials up to full CI. It measures the
three things v3's accuracy rests on:

  1. detection  — resolve_ticker() maps names/aliases to the right symbol.
                  Needs a DB connection (the companies directory), no vector
                  index and no API key.
  2. retrieval  — search_passages(ticker=...) never leaks another company's
                  filings (no cross-company contamination). Needs the DB +
                  populated pgvector index + the embedding model.
  3. answer     — run_agent() produces a grounded answer end-to-end: every
                  figure verifies against its source and the expected topic
                  keywords appear. Needs the DB + a GEMINI_API_KEY.

Each layer that can't run (missing DB / index / key) is SKIPPED, not failed.
The process exits non-zero if any case FAILS, so it can gate CI.

Run from finsight-v3/backend:
    python -m eval.run_eval              # full run
    python -m eval.run_eval --no-answer  # skip the LLM layer (no API cost)
    python -m eval.run_eval --only detection
"""

import argparse
import json
import sys
from pathlib import Path

CASES = json.loads((Path(__file__).parent / "cases.json").read_text(encoding="utf-8"))

PASS, FAIL, SKIP = "PASS", "FAIL", "SKIP"


class Tally:
    def __init__(self) -> None:
        self.passed = 0
        self.failed = 0
        self.skipped = 0

    def record(self, status: str) -> None:
        if status == PASS:
            self.passed += 1
        elif status == FAIL:
            self.failed += 1
        else:
            self.skipped += 1


def _line(status: str, label: str, detail: str = "") -> None:
    tail = f"  — {detail}" if detail else ""
    print(f"  [{status}] {label}{tail}")


def _db_available() -> tuple[bool, str]:
    try:
        from db import query

        query("select 1")
        return True, ""
    except Exception as exc:  # noqa: BLE001 — any failure means "can't reach DB"
        return False, str(exc).splitlines()[0]


def _skip_layer(name: str, cases: list, reason: str, tally: Tally) -> None:
    print(f"\n{name} — SKIPPED ({reason})")
    for case in cases:
        tally.record(SKIP)


# ---------------------------------------------------------------- layers


def run_detection(tally: Tally) -> None:
    print("\ndetection — resolve_ticker() name/alias correctness")
    from tools import resolve_ticker

    for case in CASES["detection"]:
        got = resolve_ticker(case["input"])
        if got == case["expected"]:
            tally.record(PASS)
            _line(PASS, f"{case['input']!r} -> {got}")
        else:
            tally.record(FAIL)
            _line(FAIL, f"{case['input']!r} -> {got}", f"expected {case['expected']}")


def run_retrieval(tally: Tally) -> None:
    print("\nretrieval — search_passages() company scoping (no contamination)")
    from retrieval import search_passages

    for case in CASES["retrieval"]:
        ticker = case["ticker"].upper()
        hits = search_passages(case["query"], ticker=ticker, k=6)
        stray = sorted({h["ticker"].upper() for h in hits if h["ticker"].upper() != ticker})
        if stray:
            tally.record(FAIL)
            _line(FAIL, f"{ticker}: {case['query']!r}", f"leaked {stray}")
        else:
            tally.record(PASS)
            note = f"{len(hits)} passages, all {ticker}" if hits else "no chunks (still clean)"
            _line(PASS, f"{ticker}: {case['query']!r}", note)


def run_answer(tally: Tally) -> None:
    print("\nanswer — run_agent() grounded end-to-end answers")
    from agent import run_agent

    for case in CASES["answer"]:
        question = case["question"]
        try:
            events = list(run_agent([{"role": "user", "content": question}]))
        except Exception as exc:  # noqa: BLE001
            tally.record(FAIL)
            _line(FAIL, question, f"agent raised: {str(exc).splitlines()[0]}")
            continue

        answer_event = next((e for e in reversed(events) if e.type == "answer"), None)
        if answer_event is None:
            error_event = next((e for e in events if e.type == "error"), None)
            reason = error_event.data.get("message", "no answer") if error_event else "no answer"
            tally.record(FAIL)
            _line(FAIL, question, reason)
            continue

        answer = (answer_event.data.get("answer") or "").strip()
        verification = answer_event.data.get("verification", {})
        unverified = verification.get("unverified", [])
        missing = [kw for kw in case["expect_keywords"] if kw.lower() not in answer.lower()]

        problems = []
        if not answer:
            problems.append("empty answer")
        if unverified:
            problems.append(f"unverified figures {unverified}")
        if missing:
            problems.append(f"missing keywords {missing}")

        if problems:
            tally.record(FAIL)
            _line(FAIL, question, "; ".join(problems))
        else:
            tally.record(PASS)
            checked = verification.get("figures_checked", 0)
            _line(PASS, question, f"{checked} figures, all verified")


# ---------------------------------------------------------------- driver


def main() -> int:
    parser = argparse.ArgumentParser(description="FinSight v3 eval harness")
    parser.add_argument("--no-answer", action="store_true",
                        help="skip the LLM answer layer (no API cost)")
    parser.add_argument("--only", choices=["detection", "retrieval", "answer"],
                        help="run only one layer")
    args = parser.parse_args()

    layers = ["detection", "retrieval", "answer"]
    if args.only:
        layers = [args.only]

    db_ok, db_reason = _db_available()
    from config import get_settings

    key_ok = bool(get_settings().gemini_api_key)
    tally = Tally()

    print("=" * 72)
    print("FinSight v3 evaluation")
    print(f"  database: {'reachable' if db_ok else f'unavailable ({db_reason})'}")
    print(f"  gemini key: {'present' if key_ok else 'missing'}")
    print("=" * 72)

    if "detection" in layers:
        if db_ok:
            run_detection(tally)
        else:
            _skip_layer("detection", CASES["detection"], "no database", tally)

    if "retrieval" in layers:
        if not db_ok:
            _skip_layer("retrieval", CASES["retrieval"], "no database", tally)
        else:
            try:
                run_retrieval(tally)
            except Exception as exc:  # noqa: BLE001 — missing index / embedding model
                _skip_layer("retrieval", CASES["retrieval"],
                            f"index/embedder unavailable: {str(exc).splitlines()[0]}", tally)

    if "answer" in layers:
        if args.no_answer:
            _skip_layer("answer", CASES["answer"], "--no-answer", tally)
        elif not db_ok:
            _skip_layer("answer", CASES["answer"], "no database", tally)
        elif not key_ok:
            _skip_layer("answer", CASES["answer"], "no GEMINI_API_KEY", tally)
        else:
            run_answer(tally)

    print("\n" + "=" * 72)
    print(f"{tally.passed} passed, {tally.failed} failed, {tally.skipped} skipped")
    print("=" * 72)
    return 1 if tally.failed else 0


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