nb6.5 — The AI Co-Pilot Lab

Companion to Ch 6.5, “The AI Co-Pilot for Research (LLM-in-the-Loop).”

Leah can ask a large language model to read 40,000 filings and label each one’s tone in an afternoon. That is genuinely new power — and it is exactly the place an LLM is most dangerous, because the output is a column of numbers that flows straight into a regression and you cannot eyeball 40,000 labels to check them. This lab turns the chapter’s advice into a working, reproducible measurement instrument you calibrate yourself.

We build five things, all of which run offline, with no API key:

1. A small hand-labeled dataset of synthetic “8-K snippets,” split into a gold/dev set and a held-out test set.
2. A tiny RAG pipeline over synthetic “10-K passages” (chunk already done → embed with TF-IDF → retrieve top-k by cosine → ground a templated, cited answer).
3. A classify(text) function with a provider switch whose default is a deterministic local classifier, so the whole notebook runs with no network and no key.
4. The out-of-sample validation harness — confusion matrix, precision/recall/F1 (per class + macro), and Cohen’s \kappa against the hand labels. This is the pedagogical core.
5. An audit log of every classify call, a look-ahead / leakage demo with a date-filter defense, and the real-API code for Anthropic and GMU Azure — written out in full but gated behind environment variables so they never run offline.

The one sentence to remember (Ch 6.5.4): an LLM-produced label is a measurement, not the truth — so before any LLM label enters a regression, you validate it against a hand-labeled gold standard and report its error rate on held-out data.

1. Setup

We fix one seeded generator (rng = np.random.default_rng(42)) so every run produces the identical sample — the reproducibility rule from CONVENTIONS.md §5. We force matplotlib’s non-interactive Agg backend so the notebook runs headless anywhere. We import only the standard scientific stack plus scikit-learn; nothing here touches the network.

import os
import json
import hashlib
import datetime

import matplotlib
matplotlib.use("Agg")  # headless, non-interactive backend

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

from sklearn.feature_extraction.text import TfidfVectorizer, CountVectorizer
from sklearn.metrics.pairwise import cosine_similarity
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import (
    confusion_matrix,
    precision_recall_fscore_support,
    cohen_kappa_score,
)

rng = np.random.default_rng(42)  # one generator, seeded, for the whole notebook

LABELS = ["negative", "neutral", "positive"]

print("numpy", np.__version__, "| pandas", pd.__version__)
print("ANTHROPIC_API_KEY set?  ", bool(os.environ.get("ANTHROPIC_API_KEY")))
print("AZURE_OPENAI_KEY  set?  ", bool(os.environ.get("AZURE_OPENAI_KEY")))
print("-> If both say False, every real-API cell below is SKIPPED and we run fully offline.")

2. The provider-switch design — why this notebook is safe to run anywhere

A research notebook that requires a paid API key to run is not reproducible: a referee, a CI server, or a student without a key cannot execute it. So we use a provider switch. There is a single public function

classify(text) -> label

that dispatches to one of three backends, chosen by the provider argument (default "local"):

provider	what it does	needs a key?
"local"	a deterministic, seeded classifier trained on the gold set	no (default)
"anthropic"	the real Anthropic Messages API (Claude)	yes — gated
"azure"	GMU’s Azure OpenAI gateway	yes — gated

The two real backends are written out in full further down so you can see the production patterns from Ch 6.5.6 — prompt caching, cache_read_input_tokens, the audit log. But every live call is gated behind if os.environ.get(...), so with no key set they are defined and never executed. That is the whole trick: the local provider is a faithful stand-in that lets the entire validation pipeline — confusion matrix, F1, \kappa — run identically whether or not you have a key. When you do have a key, you flip one argument and re-run, and the same harness validates the real model.

3. A small hand-labeled dataset of 8-K snippets

The honest validation protocol (Ch 6.5.4, Step 1) starts with a gold standard: a random sample of documents that you label by hand, the closest thing you have to ground truth. Here we write 45 short synthetic “8-K snippets” — the kind of one-paragraph event disclosure a firm files — and a hand label for each: negative, neutral, or positive sentiment about the firm’s prospects. These are invented for the lab (no licensed data leaves anyone’s machine), but they are written to look like real 8-K language: earnings beats and dividend hikes (positive), litigation, going-concern doubt, and guidance cuts (negative), and routine boilerplate — a new director, a bylaw amendment, a venue change (neutral).

Each row carries a date, which we will use in the leakage demo (§9). Because we wrote the labels by hand, we are the ground truth the LLM (or local classifier) must be validated against.

# (text, hand_label, date) — hand labels are the GOLD STANDARD (we wrote them).
# Sentiment is about the firm's prospects as implied by the disclosed event.
raw = [
    # --- positive ---
    ("The company reported quarterly earnings of $2.10 per share, well above the $1.65 consensus, and raised full-year guidance.", "positive", "2019-04-22"),
    ("Our board approved a 15% increase in the quarterly cash dividend and authorized a new $500 million share-repurchase program.", "positive", "2019-05-03"),
    ("The previously disclosed patent litigation has been settled on favorable terms with no material payment by the company.", "positive", "2019-06-11"),
    ("The company closed the acquisition of a profitable competitor, expected to be immediately accretive to earnings.", "positive", "2019-07-18"),
    ("Standard & Poor's upgraded the company's credit rating to investment grade, citing improved cash flow and lower leverage.", "positive", "2019-08-02"),
    ("Quarterly revenue rose 28% year over year, a record, driven by strong demand and expanding operating margins.", "positive", "2019-09-14"),
    ("The company received FDA approval for its lead product, clearing the path to commercial launch next quarter.", "positive", "2019-10-05"),
    ("Management announced that the restructuring is complete and the company has returned to profitability ahead of schedule.", "positive", "2019-11-21"),
    ("A long-running regulatory investigation was closed with no enforcement action and no fine against the company.", "positive", "2019-12-09"),
    ("The company signed a multi-year supply agreement with a major customer, expected to add $200 million in annual revenue.", "positive", "2020-01-15"),
    ("Free cash flow doubled and the company fully repaid its revolving credit facility ahead of maturity.", "positive", "2020-02-03"),
    ("The board appointed a new CEO with a strong turnaround record; the stock of the firm has historically responded well to such hires.", "positive", "2020-03-12"),
    ("Same-store sales grew 9%, beating expectations, and the company reinstated its suspended dividend.", "positive", "2020-04-08"),
    ("The company resolved its going-concern doubt after a successful equity raise and now reports adequate liquidity.", "positive", "2020-05-19"),
    ("Operating income exceeded the high end of guidance and the company announced a special dividend.", "positive", "2020-06-22"),
    # --- negative ---
    ("The company announced it has substantial doubt about its ability to continue as a going concern over the next twelve months.", "negative", "2019-04-30"),
    ("The CEO and CFO both resigned effective immediately amid an internal investigation into accounting irregularities.", "negative", "2019-05-27"),
    ("The company cut full-year revenue guidance by 20% and warned of a material decline in operating margins.", "negative", "2019-06-30"),
    ("A jury awarded $450 million in damages against the company in a product-liability suit; the company intends to appeal.", "negative", "2019-07-25"),
    ("The company received a delisting notice from the exchange after failing to meet minimum bid-price requirements.", "negative", "2019-08-30"),
    ("Quarterly results showed a wider-than-expected loss, and the company drew down its entire credit facility to fund operations.", "negative", "2019-09-28"),
    ("The company disclosed a material weakness in internal control over financial reporting and will restate prior results.", "negative", "2019-10-31"),
    ("Its largest customer, representing 35% of revenue, terminated its contract effective next quarter.", "negative", "2019-11-30"),
    ("Moody's downgraded the company's debt two notches to junk, citing deteriorating liquidity and rising leverage.", "negative", "2019-12-20"),
    ("The company suspended its dividend and announced layoffs of 12% of its workforce amid falling demand.", "negative", "2020-01-31"),
    ("A data breach exposed customer records; the company expects significant remediation costs and regulatory scrutiny.", "negative", "2020-02-28"),
    ("The company missed a scheduled interest payment and entered a forbearance agreement with its lenders.", "negative", "2020-03-31"),
    ("Goodwill impairment of $1.1 billion was recorded, reflecting a sustained decline in the value of an acquired business.", "negative", "2020-04-30"),
    ("The SEC issued a Wells notice indicating staff intends to recommend an enforcement action against the company.", "negative", "2020-05-29"),
    ("The company withdrew its full-year outlook entirely, citing extreme uncertainty and weakening order trends.", "negative", "2020-06-30"),
    # --- neutral ---
    ("The board appointed a new independent director to fill an existing vacancy; committee assignments will follow.", "neutral", "2019-05-10"),
    ("The company amended its bylaws to update advance-notice procedures for shareholder proposals.", "neutral", "2019-06-14"),
    ("The annual meeting of shareholders has been rescheduled and will now be held at the company's new headquarters.", "neutral", "2019-07-12"),
    ("The company entered into a routine indemnification agreement with its newly elected officers, consistent with past practice.", "neutral", "2019-08-16"),
    ("The company adopted a new equity incentive plan, replacing the prior plan that had reached its expiration date.", "neutral", "2019-09-20"),
    ("The company changed the state of its incorporation from one jurisdiction to another for administrative reasons.", "neutral", "2019-10-18"),
    ("The audit committee approved the reappointment of the existing independent registered public accounting firm.", "neutral", "2019-11-15"),
    ("The company announced the date of its next quarterly earnings call and webcast details.", "neutral", "2019-12-13"),
    ("A new revenue-recognition accounting standard was adopted on the required effective date with no restatement of prior periods.", "neutral", "2020-01-17"),
    ("The company relocated one of its administrative offices; operations and headcount are unaffected.", "neutral", "2020-02-14"),
    ("The board ratified the compensation arrangements for non-employee directors at the previously disclosed levels.", "neutral", "2020-03-13"),
    ("The company filed a routine registration statement to register shares issuable under its existing employee stock plan.", "neutral", "2020-04-17"),
    ("Shareholders approved all proposals at the annual meeting, including the routine ratification of auditors.", "neutral", "2020-05-15"),
    ("The company updated its corporate website and investor-relations contact information.", "neutral", "2020-06-12"),
    ("The board declared the regular quarterly dividend at the same rate as the prior quarter.", "neutral", "2020-06-26"),
]

df = pd.DataFrame(raw, columns=["text", "gold", "date"])
df["date"] = pd.to_datetime(df["date"])
print("dataset size:", len(df))
print(df["gold"].value_counts().reindex(LABELS))
df.head(3)

Split: development (gold/dev) vs. held-out test

Ch 6.5.4, Step 2: set aside a held-out test set you do not look at while building the classifier. We use a stratified split (equal class proportions in both halves) so the small test set still contains all three classes. The dev set is for building and tuning the local classifier; the test set is touched once, at the end, for the honest out-of-sample numbers.

# Stratified split: ~1/3 of each class held out, using the seeded rng (reproducible).
test_idx = []
for lab in LABELS:
    idx = df.index[df["gold"] == lab].to_numpy()
    n_test = max(1, round(len(idx) * 0.34))
    chosen = rng.choice(idx, size=n_test, replace=False)
    test_idx.extend(chosen.tolist())
test_idx = sorted(test_idx)

dev_df = df.drop(index=test_idx).copy()      # .copy() — no chained-indexing surprises
test_df = df.loc[test_idx].copy()

print(f"dev set:  {len(dev_df)} docs ->", dev_df["gold"].value_counts().reindex(LABELS).to_dict())
print(f"test set: {len(test_df)} docs ->", test_df["gold"].value_counts().reindex(LABELS).to_dict())
print("\nRule: we TUNE on dev, and REPORT on test (evaluated once).")

4. RAG mini-demo (offline): grounding an answer in retrieved 10-K passages

Before classification, the other big co-pilot use is question answering over your own documents via retrieval-augmented generation (RAG) (Ch 6.5.3). Instead of asking the model to recall facts from its blurry training memory (fabrication-prone), we hand it the relevant source text at query time and make it answer from that. The four stages are chunk → embed → retrieve → ground; chunking is already done here (each passage is one chunk), so we demonstrate the last three.

We embed with a local TF-IDF vectorizer (no API, no network) — the same cosine-similarity idea you met in Ch 6.2. Each chunk carries metadata (firm, fiscal year, 10-K item) so the grounded answer can cite a retrievable source. The honest caveat: grounding reduces hallucination by removing the recall failure mode, but it does not eliminate it — retrieval can miss the right chunk, and the model can still misread the chunks it was given.

# A tiny doc store of synthetic "10-K passages." Each is one chunk with metadata.
doc_store = [
    {"id": 1, "firm": "AcmeCorp", "fy": 2019, "item": "Item 1A Risk Factors",
     "text": "A downgrade of our credit rating could increase our borrowing costs and limit our access to capital markets, harming liquidity."},
    {"id": 2, "firm": "AcmeCorp", "fy": 2019, "item": "Item 7 MD&A",
     "text": "Our liquidity depends on cash from operations and our revolving credit facility; as of fiscal year end we held $320 million in cash and equivalents."},
    {"id": 3, "firm": "AcmeCorp", "fy": 2019, "item": "Item 1A Risk Factors",
     "text": "We face significant litigation exposure, including product-liability claims that, if adversely resolved, could materially affect our results."},
    {"id": 4, "firm": "BetaIndustries", "fy": 2019, "item": "Item 7 MD&A",
     "text": "Revenue grew 12% driven by demand in our cloud segment; gross margin expanded 180 basis points year over year."},
    {"id": 5, "firm": "BetaIndustries", "fy": 2019, "item": "Item 1A Risk Factors",
     "text": "We depend on a small number of large customers; the loss of any one of them could materially reduce our revenue."},
    {"id": 6, "firm": "BetaIndustries", "fy": 2019, "item": "Item 7 MD&A",
     "text": "We invested $90 million in research and development this year, focused on machine-learning features for our platform."},
]
store_df = pd.DataFrame(doc_store)

# Stage 2 (Embed): fit a local TF-IDF "embedding" over the chunk texts.
vectorizer = TfidfVectorizer(stop_words="english")
chunk_vecs = vectorizer.fit_transform(store_df["text"])
print("embedded", chunk_vecs.shape[0], "chunks into a", chunk_vecs.shape[1], "-dim TF-IDF space")

# Stage 3 (Retrieve): embed the QUESTION with the SAME vectorizer, take top-k by cosine.
def retrieve(question, k=3):
    qv = vectorizer.transform([question])
    sims = cosine_similarity(qv, chunk_vecs).ravel()
    order = np.argsort(sims)[::-1][:k]
    out = store_df.iloc[order].copy()
    out["cosine"] = sims[order]
    return out

question = "What liquidity risks did AcmeCorp disclose in 2019?"
hits = retrieve(question, k=3)
hits[["id", "firm", "fy", "item", "cosine", "text"]]

# Stage 4 (Ground): build a CITED answer from ONLY the retrieved passages.
# Offline, we template the answer instead of calling a model — but the SHAPE is identical
# to the grounding prompt in Ch 6.5.3: every claim points to a retrievable chunk id.
def grounded_answer(question, k=3, min_cosine=0.05):
    hits = retrieve(question, k=k)
    used = hits[hits["cosine"] >= min_cosine]
    if used.empty:
        return "Not answerable from the provided passages.", hits
    lines = [f"Answer to: {question!r}", "(grounded only in retrieved passages)\n"]
    for _, r in used.iterrows():
        lines.append(f'- {r["text"]}  [{r["id"]}] ({r["firm"]} 10-K FY{r["fy"]}, {r["item"]})')
    return "\n".join(lines), hits

ans, _ = grounded_answer(question, k=3)
print(ans)

Why this is safer — and where it can still fail. Every bullet ends in a bracketed citation like [2] that points to a real chunk in store_df; a reader can open passage 2 and confirm the claim in seconds. That is the discipline that makes RAG trustworthy. But notice the two residual risks from Ch 6.5.3: if the relevant chunk never makes the top-k (retrieval failure), the answer is built from incomplete information; and a real model — unlike our template — could still stitch two passages into a claim neither supports. So we spot-check the citations, and our prompt (below, in the real-API section) forbids any claim without one.

# A query with NO supporting chunk should DECLINE, not fabricate. This is the cite-or-decline
# behavior of Ch 6.5.2, Pattern 2 — proven here with a deliberately off-topic question.
off_topic = "What is AcmeCorp's executive compensation policy?"
ans2, hits2 = grounded_answer(off_topic, k=3, min_cosine=0.15)
print(ans2)
print("\n(top retrieved cosine was only", round(float(hits2['cosine'].max()), 3),
      "-> below threshold, so we decline rather than guess)")

5. The classify function with a provider switch

Now the core instrument. The default provider is a deterministic local classifier so the whole notebook runs offline. We build it two ways and let it fall back gracefully:

1. A tiny bag-of-words logistic regression trained on the dev set (seeded, reproducible).
2. A transparent keyword/rule backstop for any document the model is unsure about.

Both are deterministic given the seed — re-running gives identical labels, which is the reproducibility property a stochastic API does not give you for free (Ch 6.5.5). The local model is intentionally simple and weak; that is the point — the validation harness (§6) will tell us exactly how weak, the same way it would grade a frontier model.

# A transparent keyword rule — the kind of thing the Loughran-McDonald dictionary does (Ch 6.3).
NEG_WORDS = ["going concern", "resigned", "investigation", "cut", "loss", "delisting",
             "downgrad", "material weakness", "restate", "terminated", "suspended",
             "layoff", "breach", "missed", "impairment", "wells notice", "withdrew",
             "damages", "doubt", "forbearance"]
POS_WORDS = ["above", "raised", "increase", "favorable", "accretive", "upgrade", "record",
             "approval", "profitab", "closed with no", "no enforcement", "no fine",
             "doubled", "repaid", "beating", "special dividend", "exceeded", "grew",
             "resolved", "settled on favorable"]

def keyword_label(text):
    t = text.lower()
    neg = sum(w in t for w in NEG_WORDS)
    pos = sum(w in t for w in POS_WORDS)
    if neg > pos:
        return "negative"
    if pos > neg:
        return "positive"
    return "neutral"

# A bag-of-words logistic regression trained on the DEV set only (never the test set).
_bow = CountVectorizer(stop_words="english", ngram_range=(1, 2), min_df=1)
_X_dev = _bow.fit_transform(dev_df["text"])
_clf = LogisticRegression(max_iter=1000, random_state=42)
_clf.fit(_X_dev, dev_df["gold"])
print("trained local bag-of-words logistic regression on", _X_dev.shape[0], "dev docs;",
      _X_dev.shape[1], "features.")

# In-memory audit log (filled by every classify call). See §8 for the JSONL writer.
AUDIT_LOG = []

def _log_call(provider, text, label, usage=None, logfile=None):
    record = {
        "timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
        "provider": provider,
        "input_sha256": hashlib.sha256(text.encode("utf-8")).hexdigest(),  # hash, not raw text
        "label": label,
        # token usage is only present for real API calls; None offline.
        "input_tokens": getattr(usage, "input_tokens", None) if usage else None,
        "output_tokens": getattr(usage, "output_tokens", None) if usage else None,
        "cache_read_input_tokens": getattr(usage, "cache_read_input_tokens", None) if usage else None,
    }
    AUDIT_LOG.append(record)
    if logfile is not None:
        with open(logfile, "a") as f:
            f.write(json.dumps(record) + "\n")
    return record


def classify_local(text):
    '''Deterministic, offline classifier: trust the BoW model when it is confident,
    otherwise fall back to the transparent keyword rule.'''
    proba = _clf.predict_proba(_bow.transform([text]))[0]
    top = proba.max()
    model_label = _clf.classes_[proba.argmax()]
    if top >= 0.55:
        return model_label
    return keyword_label(text)  # low-confidence backstop

The public entry point dispatches on provider. Offline, provider="local" is the only branch that executes; the "anthropic" and "azure" branches call functions defined in §10 that themselves check for a key and refuse to run without one. Every call is logged.

def classify(text, provider="local", logfile=None):
    '''Classify one snippet's sentiment. Default provider is the offline local model.'''
    if provider == "local":
        label = classify_local(text)
        usage = None
    elif provider == "anthropic":
        label, usage = classify_anthropic(text)   # defined in §10; gated on the key
    elif provider == "azure":
        label, usage = classify_azure(text)       # defined in §10; gated on the key
    else:
        raise ValueError(f"unknown provider: {provider!r}")
    _log_call(provider, text, label, usage=usage, logfile=logfile)
    return label

# Smoke test on three obvious cases (runs offline).
for t in ["The company raised full-year guidance after a record quarter.",
          "The company disclosed substantial doubt about its ability to continue as a going concern.",
          "The board appointed a new independent director to fill a vacancy."]:
    print(f"{classify(t):>9}  <-  {t[:70]}")

6. Out-of-sample validation harness — the pedagogical core

This is the heart of the chapter. We treat the local classifier’s labels as a measurement and ask: how accurate is this instrument, on my data, against truth? We run classify over the held-out test set (touched once), then compute, against the hand labels:

• a confusion matrix (rows = true/gold, columns = predicted),
• precision, recall, F1 per class and macro-averaged — never bare accuracy, which lies under class imbalance (Ch 6.5.4, Step 4),
• Cohen’s \kappa, which measures agreement corrected for chance — a sterner test than raw agreement.

# Run the classifier over the HELD-OUT test set (evaluated once, as the protocol demands).
test_pred = [classify(t, provider="local") for t in test_df["text"]]
test_gold = test_df["gold"].tolist()

cm = confusion_matrix(test_gold, test_pred, labels=LABELS)
cm_df = pd.DataFrame(cm, index=[f"gold:{l}" for l in LABELS],
                         columns=[f"pred:{l}" for l in LABELS])
print("Confusion matrix (rows = hand/gold truth, cols = classifier prediction):")
cm_df

# Precision / recall / F1, per class and macro-averaged.
prec, rec, f1, support = precision_recall_fscore_support(
    test_gold, test_pred, labels=LABELS, zero_division=0)
report = pd.DataFrame(
    {"precision": prec, "recall": rec, "f1": f1, "support": support},
    index=LABELS)

macro = report[["precision", "recall", "f1"]].mean()
report.loc["macro avg"] = [macro["precision"], macro["recall"], macro["f1"], support.sum()]

accuracy = np.mean(np.array(test_pred) == np.array(test_gold))
kappa = cohen_kappa_score(test_gold, test_pred)

print(report.round(3).to_string())
print(f"\nplain accuracy : {accuracy:.3f}   <- do NOT report this alone under class imbalance")
print(f"Cohen's kappa  : {kappa:.3f}   <- chance-corrected agreement (1.0 = perfect, 0 = chance)")

How to read this. Look at the per-class recall, especially for the class your downstream result depends on. Suppose your study hinges on finding negative 8-Ks: a high accuracy number can hide a low recall on negative (you are missing the events you care about). That is exactly the trap Ch 6.5.4 warns about, and the chapter’s Question 1 — “precision 0.88, recall 0.61 on the class you care about: would you use these labels?” The macro-F1 and \kappa summarize the whole instrument in one number each; if they are too low, you do not have a measurement, you have noise, and no amount of fluent output changes that. We also visualize the confusion matrix.

fig, ax = plt.subplots(figsize=(5.2, 4.4))
im = ax.imshow(cm, cmap="Blues")
ax.set_xticks(range(len(LABELS)), labels=LABELS)
ax.set_yticks(range(len(LABELS)), labels=LABELS)
ax.set_xlabel("predicted (classifier)")
ax.set_ylabel("true (hand / gold)")
ax.set_title(f"Confusion matrix — local provider\nmacro-F1={macro['f1']:.2f}, kappa={kappa:.2f}")
for i in range(len(LABELS)):
    for j in range(len(LABELS)):
        ax.text(j, i, int(cm[i, j]), ha="center", va="center",
                color="white" if cm[i, j] > cm.max() / 2 else "black")
fig.colorbar(im, ax=ax, fraction=0.046, pad=0.04)
fig.tight_layout()
fig.savefig("/tmp/nb65_confusion.png", dpi=90)
plt.close(fig)
print("saved confusion-matrix figure to /tmp/nb65_confusion.png")

The numbers have error bars too

Our test set is tiny (that is realistic — hand labeling is expensive). A precision of 0.80 measured on a handful of predicted-positives has a very wide confidence interval (Ch 6.5.4). A quick way to feel that uncertainty is a nonparametric bootstrap: resample the test set with replacement many times and look at the spread of macro-F1.

gold_arr = np.array(test_gold)
pred_arr = np.array(test_pred)
n = len(gold_arr)
boot_f1 = []
for _ in range(2000):
    samp = rng.integers(0, n, size=n)
    _, _, f1_b, _ = precision_recall_fscore_support(
        gold_arr[samp], pred_arr[samp], labels=LABELS, zero_division=0)
    boot_f1.append(f1_b.mean())
boot_f1 = np.array(boot_f1)
lo, hi = np.percentile(boot_f1, [2.5, 97.5])
print(f"macro-F1 point estimate : {macro['f1']:.3f}")
print(f"bootstrap 95% interval  : [{lo:.3f}, {hi:.3f}]  (n_test = {n})")
print("-> The interval is wide BECAUSE the test set is small. Report it; do not over-claim.")

8. Audit logging — making the pipeline reproducible

For IRB-style reproducibility (Ch 6.5.6), we log every classify call: a timestamp, the provider, a SHA-256 hash of the input (not the raw text, so licensed data never lands in a log), the label, and — for real API calls — the token usage including cache_read_input_tokens. We have been appending to an in-memory AUDIT_LOG all along; here we also write it to a JSONL file (one JSON object per line), the AI-pipeline equivalent of a Stata .log.

AUDIT_PATH = "/tmp/nb65_audit_log.jsonl"
with open(AUDIT_PATH, "w") as f:
    for rec in AUDIT_LOG:
        f.write(json.dumps(rec) + "\n")

print(f"logged {len(AUDIT_LOG)} classify calls to {AUDIT_PATH}")
print("\nfirst record:")
print(json.dumps(AUDIT_LOG[0], indent=2))

audit_df = pd.DataFrame(AUDIT_LOG)
print("\ncalls by provider:")
print(audit_df["provider"].value_counts().to_string())
print("\n-> Offline, every call is provider='local'. With a key, real calls would show "
      "token counts and cache_read_input_tokens here.")

9. Look-ahead bias and the date-filter defense

The subtlest, most finance-specific failure (Ch 6.5.5): a model is trained on text up to some cutoff date, so for a forward-looking task it may “know the future” relative to your sample — it has read what actually happened. Labeling a 2019 filing’s contemporaneous tone is comparatively safe; predicting 2020 returns from a 2019 filing is exactly where leakage bites.

The cleanest mechanical defense, especially in RAG, is to filter retrieval by date: when answering a question dated t, never retrieve a chunk dated after t. We demonstrate it on our dated dataset — a retriever that respects an “as-of” date can only ever see the past.

# A leakage-aware retriever over the 8-K snippets: only documents dated <= as_of are visible.
snippet_vecs = vectorizer.fit_transform(df["text"])  # re-fit over the 8-K corpus for this demo

def retrieve_asof(query, as_of, k=3):
    as_of = pd.Timestamp(as_of)
    visible = df.index[df["date"] <= as_of].to_numpy()
    if len(visible) == 0:
        return df.iloc[0:0].copy()
    qv = vectorizer.transform([query])
    sims = cosine_similarity(qv, snippet_vecs[visible]).ravel()
    order = visible[np.argsort(sims)[::-1][:k]]
    out = df.loc[order].copy()
    out["cosine"] = sims[np.argsort(sims)[::-1][:k]]
    return out

q = "dividend increase"
as_of = "2019-06-01"
hits_asof = retrieve_asof(q, as_of=as_of, k=4)
n_future = int((df["date"] > pd.Timestamp(as_of)).sum())
print(f"Query {q!r} as of {as_of}: retriever can see {int((df['date'] <= pd.Timestamp(as_of)).sum())} "
      f"past docs and is BLIND to {n_future} future docs.")
print("\nretrieved (all dated on/before the as-of date):")
print(hits_asof[["date", "gold", "cosine", "text"]].assign(
    text=lambda d: d["text"].str.slice(0, 55) + "...").to_string(index=False))
assert (hits_asof["date"] <= pd.Timestamp(as_of)).all(), "leakage! a future doc was retrieved"
print("\nassertion passed: no future document leaked into the as-of retrieval.")

The conceptual point in one line: without the date filter, a “prediction” task can retrieve a document from after the decision date and produce spectacular, entirely fake backtest performance — the model is remembering the future, not predicting it (Ch 6.5.5, Sam’s 2021 backtest). The assert above is the kind of mechanical guard you should put in any as-of pipeline so leakage fails loudly instead of silently inflating your results.

10. The real API providers (env-gated — skipped offline)

Below are the production implementations from Ch 6.5.6. They are written out in full so you can see the real patterns — prompt caching, cache_read_input_tokens, the Azure gateway — but every live call is gated behind an environment-variable check, so with no key set (as in offline verification) they are defined and never executed. No key is ever hard-coded; the SDKs read it from the environment.

requires an API key; skipped offline. If you set ANTHROPIC_API_KEY / AZURE_OPENAI_KEY in your shell, the gated blocks below light up and the same validation harness in §6 will then grade the real model.

First, the shared rubric (Ch 6.5.2, Pattern 4) — the stable instruction that goes in the cached prefix.

RUBRIC = '''You are classifying the sentiment of a corporate 8-K event disclosure
about the filing firm's prospects. Assign exactly one label:

  NEGATIVE  - the event signals new or worsening harm to the firm (going-concern
              doubt, litigation losses, guidance cuts, downgrades, executive
              departures amid investigations, delisting, breaches).
  POSITIVE  - the event signals improvement (earnings beats, dividend increases,
              upgrades, favorable settlements, accretive acquisitions, approvals).
  NEUTRAL   - routine, administrative, or boilerplate events with no clear
              directional implication (director appointments, bylaw amendments,
              venue changes, routine auditor ratification).

Rules:
- Judge only the text provided. Do not use outside knowledge about the firm.
- If genuinely ambiguous, choose NEUTRAL.
- Return only the lowercase label word: negative, neutral, or positive.'''

def _normalize_label(raw_text):
    '''Map a model's free-text reply to one of our three labels.'''
    t = raw_text.strip().lower()
    for lab in LABELS:
        if lab in t:
            return lab
    return "neutral"  # safe default if the model rambled

print("rubric ready:", len(RUBRIC), "chars. (Freeze this against the gold set BEFORE looking at "
      "any downstream regression -- avoids prompt-induced p-hacking, Ch 6.5.5.)")

(a) Anthropic Messages API — with prompt caching

The anthropic SDK reads ANTHROPIC_API_KEY from the environment automatically. The stable rubric goes in system with a cache_control breakpoint; the volatile snippet goes in messages. Because the API caches a prefix, the rubric is processed once and reused on every later call — verify with response.usage.cache_read_input_tokens > 0. We keep max_tokens tiny because the output is one word.

def classify_anthropic(text, model="claude-opus-4-7"):
    '''Real Anthropic call. Returns (label, usage). Requires ANTHROPIC_API_KEY.'''
    from anthropic import Anthropic
    client = Anthropic()  # reads ANTHROPIC_API_KEY from the environment; never hard-coded
    response = client.messages.create(
        model=model,
        max_tokens=16,  # one word out: keep it tiny and cheap
        system=[
            {
                "type": "text",
                "text": RUBRIC,
                "cache_control": {"type": "ephemeral"},  # <- cache the stable rubric prefix
            }
        ],
        messages=[{"role": "user", "content": text}],
    )
    label = _normalize_label(response.content[0].text)
    return label, response.usage


# ---- GATED: runs ONLY if a key is present. Skipped during offline verification. ----
if os.environ.get("ANTHROPIC_API_KEY"):
    print("ANTHROPIC_API_KEY found -> running a real Anthropic classification + caching check.")
    demo_text = test_df["text"].iloc[0]
    lab1, usage1 = classify_anthropic(demo_text)
    lab2, usage2 = classify_anthropic(demo_text)  # second call should hit the cached rubric
    print("label:", lab1)
    print("cache_read_input_tokens (2nd call):",
          getattr(usage2, "cache_read_input_tokens", 0),
          "-> >0 means the rubric prefix was cached")
    # To validate the REAL model with the same harness:
    #   real_pred = [classify(t, provider="anthropic") for t in test_df["text"]]
    #   ... then recompute the confusion matrix / F1 / kappa from §6 on real_pred.
else:
    print("requires an API key; skipped offline. "
          "(No ANTHROPIC_API_KEY in the environment -> the gated block did NOT execute.)")

(b) GMU Azure OpenAI gateway

Same shape, OpenAI-family models through GMU’s Azure API Management gateway. The endpoint is fixed; the key is read from AZURE_OPENAI_KEY and appears nowhere in source. The api_version and deployment name below are illustrative / [CHECK] — confirm the current values with GMU’s N1 AI gateway docs before relying on them. Having a second provider is a robustness check: if your headline result survives re-labeling with a different vendor’s model, that is real evidence it is about the documents, not one model’s quirks.

def classify_azure(text, deployment="gpt-4o", api_version="2024-10-21"):
    '''Real GMU Azure call. Returns (label, usage). Requires AZURE_OPENAI_KEY.
    NOTE: deployment name and api_version are illustrative -- [CHECK] GMU N1 AI docs.'''
    from openai import AzureOpenAI
    client = AzureOpenAI(
        azure_endpoint="https://apim-n1ai-use-gmun1.azure-api.net/",  # fixed GMU gateway
        api_key=os.environ["AZURE_OPENAI_KEY"],  # <- read from env, NEVER inline
        api_version=api_version,                 # pin for reproducibility ([CHECK])
    )
    response = client.chat.completions.create(
        model=deployment,  # your GMU deployment name ([CHECK])
        max_tokens=16,
        messages=[
            {"role": "system", "content": RUBRIC},
            {"role": "user", "content": text},
        ],
    )
    label = _normalize_label(response.choices[0].message.content)
    return label, getattr(response, "usage", None)


# ---- GATED: runs ONLY if a key is present. Skipped during offline verification. ----
if os.environ.get("AZURE_OPENAI_KEY"):
    print("AZURE_OPENAI_KEY found -> running a real Azure classification.")
    demo_text = test_df["text"].iloc[0]
    lab, usage = classify_azure(demo_text)
    print("label:", lab)
    # Validate with the same harness:
    #   real_pred = [classify(t, provider="azure") for t in test_df["text"]]
else:
    print("requires an API key; skipped offline. "
          "(No AZURE_OPENAI_KEY in the environment -> the gated block did NOT execute.)")

11. Your Turn

You now have a complete, reproducible measurement instrument with the validation harness that grades it. Extend it:

1. Improve the local classifier. Add bigrams that capture finance idioms (e.g., "going concern", "material weakness"), tune the confidence threshold in classify_local, or swap the logistic regression for a small LinearSVC. Re-run §6 — did macro-F1 and \kappa on the held-out test set improve, or did you just overfit the dev set?
2. Swap in a real provider. Set ANTHROPIC_API_KEY (or AZURE_OPENAI_KEY) in your shell, then run real_pred = [classify(t, provider="anthropic") for t in test_df["text"]] and feed real_pred through the same confusion-matrix / F1 / \kappa code in §6. Does the frontier model clear a higher bar than your local one? By how much, and is the gap inside the bootstrap CI?
3. Stress the rubric. Rewrite RUBRIC to treat dividend suspensions as strongly negative, and check whether your test-set numbers move. This is precisely where prompt-induced p-hacking (Ch 6.5.5) lurks: freeze the rubric against the gold set before you ever look at a downstream regression.
4. Harden the RAG demo. Add a chunk that is almost on-topic and confirm the cosine threshold in grounded_answer still declines when it should. What threshold trades off missed answers (low recall) against fabricated ones (low precision)?

The carry-home from Mentor Session 6: an LLM is a measurement and drafting instrument that you remain responsible for. The guardrails — validation, audit logging, date filters, frozen rubrics — are your job, and they are what make the difference between a co-pilot and an autopilot flown into a mountain with total confidence.