eval/run_eval.py

# eval/run_eval.py
"""
Measured, reproducible numbers for FinSight — extraction accuracy and
token efficiency against hand-labeled ground truth (eval/ground_truth.json,
every value read from the actual filing with a page reference).

Run from the repo root:
    python -m eval.run_eval            # full: re-parses each filing (slow)
    python -m eval.run_eval --fast     # skips re-parsing, uses stored graph

Reports:
  1. Extraction accuracy per metric: TOP_CORRECT (top candidate right,
     within 1%), IN_ALTERNATIVES (right value present but not chosen),
     WRONG, or MISSING. "Right but not chosen" is reported separately
     because it measures ranking quality, not recall.
  2. Token efficiency: prompt tokens for FinSight's /query context
     (metrics header + top-3 chunks) vs a naive-RAG baseline (top-20
     chunks), plus the share of metric questions answered by direct
     lookup with ZERO LLM tokens.

Token counting is len(text)/4 (chars per token approximation) — stated
here so the number is reproducible and honest about its precision.
"""

import argparse
import json
import sys
from pathlib import Path

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

from backend.graph import FinancialGraph          # noqa: E402
from backend.parser import parse_document          # noqa: E402
from backend.red_flags import get_value            # noqa: E402
from backend.main import try_direct_metric_answer  # noqa: E402

TOLERANCE = 0.01  # 1% — display rounding, not different figures


def tokens(text: str) -> int:
    return len(text) // 4


def within(a: float, b: float) -> bool:
    if b == 0:
        return a == 0
    return abs(a - b) / abs(b) <= TOLERANCE


def grade_metric(extracted, truth_value) -> str:
    if extracted is None:
        return "MISSING"
    value, _ = get_value(extracted)
    if value is not None and within(value, truth_value):
        return "TOP_CORRECT"
    if isinstance(extracted, dict):
        for alt in extracted.get("alternatives", []):
            if isinstance(alt, dict) and alt.get("value") is not None:
                if within(alt["value"], truth_value):
                    return "IN_ALTERNATIVES"
    return "WRONG"


def build_query_context(fg: FinancialGraph, company: str, question: str, top_k: int) -> str:
    """Mirror main.py's /query context construction."""
    from backend.entity_resolver import format_money

    chunks = fg.get_relevant_chunks(question, company, top_k=top_k)
    lines = []
    for year, metrics in fg.get_company_metrics(company).items():
        for key, raw in metrics.items():
            value, _ = get_value(raw)
            if value is None:
                continue
            currency = raw.get("currency", "USD") if isinstance(raw, dict) else "USD"
            lines.append(f"{company} {year} {key}: {format_money(value, currency)}")
    metrics_context = ("Key Metrics:\n" + "\n".join(lines) + "\n\n") if lines else ""
    return metrics_context + "\n\n".join(c["text"] for c in chunks)


def build_baseline_context(fg: FinancialGraph, company: str, question: str) -> str:
    """Naive RAG baseline: stuff top-20 chunks, no metrics header."""
    chunks = fg.get_relevant_chunks(question, company, top_k=20)
    return "\n\n".join(c["text"] for c in chunks)


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--fast", action="store_true",
                    help="skip re-parsing filings; use the stored graph for accuracy too")
    args = ap.parse_args()

    gt = json.loads((ROOT / "eval" / "ground_truth.json").read_text(encoding="utf-8"))

    fg = FinancialGraph()
    graph_path = ROOT / "data" / "graph.json"
    if graph_path.exists():
        fg.load(str(graph_path))

    grades = []          # (company, metric, grade)
    token_rows = []      # (question, finsight_tokens, baseline_tokens)
    direct_hits = 0
    direct_total = 0

    for filing in gt["filings"]:
        company, year = filing["company"], filing["year"]

        # ── extraction accuracy ──
        if args.fast:
            extracted = fg.get_company_metrics(company).get(str(year), {})
        else:
            parsed = parse_document(str(ROOT / filing["file"]), company, year)
            extracted = parsed["metrics"]

        for key, truth in filing["metrics"].items():
            grades.append((company, key, grade_metric(extracted.get(key), truth["value"])))

        # ── direct-lookup rate (zero LLM tokens) ──
        metrics_by_year = fg.get_company_metrics(company)
        for q in filing.get("direct_questions", []):
            direct_total += 1
            if try_direct_metric_answer(q, company, metrics_by_year):
                direct_hits += 1

        # ── token efficiency on open questions ──
        for q in filing.get("open_questions", []):
            fs = tokens(build_query_context(fg, company, q, top_k=3))
            base = tokens(build_baseline_context(fg, company, q))
            token_rows.append((q, fs, base))

    # ── report ──
    print("\n=== Extraction accuracy (vs hand-labeled filing values) ===")
    counts = {}
    for company, key, grade in grades:
        counts[grade] = counts.get(grade, 0) + 1
        print(f"  {company:12s} {key:22s} {grade}")
    total = len(grades)
    top = counts.get("TOP_CORRECT", 0)
    in_alt = counts.get("IN_ALTERNATIVES", 0)
    print(f"\n  top-candidate accuracy : {top}/{total} ({100*top/total:.0f}%)")
    print(f"  value found (top+alt)  : {top+in_alt}/{total} ({100*(top+in_alt)/total:.0f}%)")

    if token_rows:
        print("\n=== Token efficiency (open questions, FinSight vs naive top-20 RAG) ===")
        fs_total = base_total = 0
        for q, fs, base in token_rows:
            fs_total += fs
            base_total += base
            print(f"  {fs:6d} vs {base:6d} tokens  | {q[:60]}")
        reduction = 100 * (1 - fs_total / base_total) if base_total else 0
        print(f"\n  avg prompt tokens      : {fs_total//len(token_rows)} vs {base_total//len(token_rows)}")
        print(f"  token reduction        : {reduction:.1f}% (len/4 approximation)")

    if direct_total:
        print(f"\n=== Zero-token direct lookup ===")
        print(f"  {direct_hits}/{direct_total} metric questions answered with 0 LLM tokens")

    results = {
        "extraction": {"top_correct": top, "in_alternatives": in_alt, "total": total},
        "token_reduction_pct": round(reduction, 1) if token_rows else None,
        "direct_lookup": {"hits": direct_hits, "total": direct_total},
    }
    out = ROOT / "eval" / "results.json"
    out.write_text(json.dumps(results, indent=2), encoding="utf-8")
    print(f"\nSaved: {out}")


if __name__ == "__main__":
    main()