Dashboard + adaptive retune loop fixes (Increments 1-2 follow-ups)

.python-version

@@ -0,0 +1 @@
+3.13

README.md

@@ -1 +1,50 @@
-# NLP_lab
+# NLP_lab
+
+Predict next-day stock move (**up / down / neutral**) from financial-news
+headlines with FinBERT, then explain each prediction in plain text. Five agents
+pass contract files to each other in a retune loop driven by the Manager.
+
+See [`CLAUDE.md`](./CLAUDE.md), [`docs/architecture.md`](./docs/architecture.md),
+and [`docs/data_contracts.md`](./docs/data_contracts.md) for the full design.
+
+## Pipeline flow
+
+```mermaid
+flowchart TD
+    raw[("FNSPID headlines<br/>+ yfinance prices")] --> aurora
+
+    aurora["**Aurora** — Processing<br/>join + label move"]
+    nadi["**Nadi** — Classifier<br/>FinBERT inference"]
+    sabina["**Sabina** — Evaluator<br/>accuracy + per-class metrics"]
+    manager{"**Jack** — Manager<br/>threshold gate<br/>accuracy ≥ 0.60?"}
+    freddi["**Freddi** — Explanation<br/>Ollama justification"]
+    final[("final_results.csv<br/>final_report.json")]
+
+    aurora -->|processed_data.csv| nadi
+    nadi -->|classifier.py +<br/>predictions_test.csv| sabina
+    sabina -->|evaluation_report.json| manager
+    manager -->|"retune_request.json<br/>(below target, &lt; 5 iters)"| nadi
+    manager -->|"sample_for_explanation.csv<br/>(cleared / cap hit)"| freddi
+    freddi -->|explanations.csv| manager
+    manager -->|finalize| final
+```
+
+**Loop:** the Manager gates on accuracy. Below the 0.60 target it writes a
+`retune_request.json` and sends Nadi back around; once the target clears, the
+accuracy plateaus (convergence early-stop), or the 5-iteration cap forces it, it
+samples rows for Freddi, then joins the explanations into the final outputs. Each
+retune escalates the classifier's hyperparameters instead of repeating, so the
+loop actually explores rather than spinning.
+
+The whole thing is one compiled **LangGraph** (`agents/pipeline_graph.py`) whose
+retune loop is a real graph *cycle* (`gate → classify → evaluate → gate`), not a
+Python loop. `main.py` just loads secrets, parses flags, and invokes the graph.
+
+## Setup & run
+
+Python 3.13, uv-managed. From repo root:
+
+```bash
+uv sync
+uv run main.py          # drives the full pipeline
+```

agents/jack_manager.py

@@ -41,8 +41,18 @@ class ManagerState(TypedDict):
     overrides: dict              # fields Jack changed; {} if accept
     notes: str                   # rationale (filled by the LLM node, step 5)
 
-    # --- running history (reducer field → appended every iteration) ---
-    decision_log: Annotated[list, operator.add]
+    # --- convergence config (set once at start; see `decide`) ---
+    patience: int                # how many recent iterations to watch for progress
+    min_delta: float             # smallest accuracy gain that counts as "progress"
+
+    # --- running history (reducer fields → APPENDED every iteration) ------------
+    # A reducer field is merged with `operator.add` (list concatenation) instead of
+    # being overwritten, so each node returns a *one-element list* and LangGraph
+    # appends it. This is how the Manager remembers what happened in earlier
+    # iterations — the raw material for the convergence and adaptation decisions.
+    decision_log: Annotated[list, operator.add]      # human-readable note per step
+    accuracy_history: Annotated[list, operator.add]  # one accuracy per iteration
+    tried_params: Annotated[list, operator.add]      # retune params used per retune
 
     # --- I/O config (paths to contract files; optional, read via .get) ---
     predictions_path: str        # Nadi's predictions_test.csv (to sample / finalize)
@@ -57,7 +67,10 @@ def _write_json(name: str, obj: dict) -> None:
 
 
 def _write_decision(state: "ManagerState") -> None:
-    """decision.json — Jack's record, written every iteration (Handoff 3b)."""
+    """decision.json — Jack's record, written every iteration (Handoff 3b). The
+    file is overwritten each iteration, so `accuracy_history` (cumulative through
+    the current iteration) is the only place the per-iteration trend survives on
+    disk once the loop moves on."""
     report = state["evaluation_report"]
     _write_json("decision.json", {
         "iteration": state["iteration"],
@@ -66,50 +79,133 @@ def _write_decision(state: "ManagerState") -> None:
         "based_on_proposal": report.get("proposal", {}),
         "overrides": state.get("overrides", {}),
         "notes": state["notes"],
+        "accuracy_history": state.get("accuracy_history", []),
     })
 
 
+# Ordered escalation schedule for the retune loop. The first retune re-uses
+# Sabina's proposed params as-is; every retune AFTER that walks down this list
+# (skipping anything already tried) so each attempt is genuinely different rather
+# than a repeat. The levers: lower `threshold` so fewer rows get forced to the
+# `neutral` fallback class, widen `max_length` so longer headlines aren't
+# truncated, and raise `boost_factor` so Nadi's focus-label boost (applied to
+# whatever Sabina flagged as weakest) has more effect each retune. Tune these
+# values here — nothing downstream is hardcoded to them.
+_RETUNE_SCHEDULE = [
+    {"threshold": 0.45, "max_length": 128, "boost_factor": 1.25},
+    {"threshold": 0.40, "max_length": 160, "boost_factor": 1.35},
+    {"threshold": 0.35, "max_length": 192, "boost_factor": 1.45},
+    {"threshold": 0.30, "max_length": 224, "boost_factor": 1.55},
+    {"threshold": 0.25, "max_length": 256, "boost_factor": 1.65},
+    {"threshold": 0.20, "max_length": 256, "boost_factor": 1.75},
+]
+
+
+def _next_params(tried: list) -> dict:
+    """Pick the next retune params: the first schedule entry not already tried,
+    or the last entry once the schedule is exhausted (the iteration cap still
+    bounds the loop, so returning a repeat here is safe).
+
+    "Already tried" compares only the keys both dicts share: Sabina's proposal
+    omits params she doesn't set (e.g. `boost_factor`), and Nadi fills those from
+    the same defaults the schedule starts at — so a schedule entry that matches a
+    tried set on every shared key would regenerate an identical classifier."""
+    def _same(a: dict, b: dict) -> bool:
+        shared = a.keys() & b.keys()
+        return bool(shared) and all(a[k] == b[k] for k in shared)
+
+    for params in _RETUNE_SCHEDULE:
+        if not any(_same(params, t) for t in tried):
+            return params
+    return _RETUNE_SCHEDULE[-1]
+
+
+def _converged(history: list, patience: int, min_delta: float) -> bool:
+    """True when retuning has stopped paying off, so we should proceed instead of
+    burning the rest of the iteration budget.
+
+    Rule: once we have more than `patience` accuracies, compare the best of the
+    last `patience` iterations against the best of everything before them. If the
+    recent window failed to beat the earlier best by at least `min_delta`, the
+    loop has plateaued. `history` includes the current iteration's accuracy.
+    """
+    if len(history) <= patience:
+        return False  # not enough history yet — let the loop keep exploring
+    recent_best = max(history[-patience:])
+    earlier_best = max(history[:-patience])
+    return recent_best - earlier_best < min_delta
+
+
 def decide(state: ManagerState) -> dict:
-    """Threshold gate (deterministic). Returns only the keys it changed."""
+    """Threshold gate (deterministic). Decides retune vs. proceed, and when
+    retuning, chooses the next hyperparameters from history. Returns only the
+    keys it changed (LangGraph merges them into the running state).
+    """
     report = state["evaluation_report"]
     accuracy = report["accuracy"]
-    # count retune cycles only: once we've proceeded, the finalize pass is not
-    # a new iteration, so don't bump the counter past convergence.
+    # Count retune cycles only: once we've proceeded, the finalize pass is not a
+    # new iteration, so don't bump the counter past convergence.
     iteration = state.get("iteration", 0)
     if state.get("final_action") != "proceed":
         iteration += 1
 
-    cleared = accuracy >= state["target_accuracy"]
-    cap_hit = iteration >= state["max_iterations"]
-    final_action = "proceed" if (cleared or cap_hit) else "retune"
+    # Full accuracy trend INCLUDING this iteration — the input to convergence.
+    history = state.get("accuracy_history", []) + [accuracy]
+    patience = state.get("patience", 2)
+    min_delta = state.get("min_delta", 0.01)
+
+    # Three independent reasons to stop retuning and move on.
+    cleared = accuracy >= state["target_accuracy"]       # good enough
+    cap_hit = iteration >= state["max_iterations"]        # out of budget
+    converged = _converged(history, patience, min_delta)  # progress has stalled
+    final_action = "proceed" if (cleared or cap_hit or converged) else "retune"
 
     if cleared:
         why = "cleared target"
+    elif converged:
+        why = "converged (no improvement), proceeding"
     elif cap_hit:
         why = "cap hit, forcing proceed"
     else:
         why = "below target, retuning"
     note = (f"iteration {iteration}: accuracy {accuracy:.2f} vs target "
             f"{state['target_accuracy']:.2f} — {why}")
 
-    # Override = the gate's action disagrees with Sabina's recommendation
-    # (e.g. she says retune but the cap forces proceed). Else accept.
-    recommended = report.get("proposal", {}).get("recommended_action")
-    if recommended is not None and recommended != final_action:
-        decision, overrides = "override", {"final_action": final_action}
-    else:
-        decision, overrides = "accept", {}
-
-    return {
-        "iteration": iteration,     # saved → next invocation resumes from here
+    out = {
+        "iteration": iteration,          # saved → next invocation resumes from here
         "accuracy": accuracy,
         "final_action": final_action,
-        "decision": decision,
-        "overrides": overrides,
-        "notes": note,              # plain field → overwrites
-        "decision_log": [note],     # reducer field → appended
+        "notes": note,                   # plain field → overwrites
+        "decision_log": [note],          # reducer field → appended
+        "accuracy_history": [accuracy],  # reducer field → appended
     }
 
+    if final_action == "retune":
+        # First retune trusts Sabina's proposal as-is (accept); every retune after
+        # that adapts the params (override), because a repeat proposal has already
+        # failed once. `tried_params` records what we actually used either way.
+        proposal = report.get("proposal", {})
+        tried = state.get("tried_params", [])
+        if not tried:
+            used = proposal.get("suggested_params", {})
+            out["decision"], out["overrides"] = "accept", {}
+        else:
+            used = _next_params(tried)
+            out["decision"] = "override"
+            out["overrides"] = {"suggested_params": used,
+                                "focus_labels": proposal.get("focus_labels", [])}
+        out["tried_params"] = [used]
+    else:
+        # Proceeding. Override only when the gate overrules Sabina's recommendation
+        # (e.g. she says retune but the cap/convergence forces proceed); else accept.
+        recommended = report.get("proposal", {}).get("recommended_action")
+        if recommended is not None and recommended != final_action:
+            out["decision"], out["overrides"] = "override", {"final_action": final_action}
+        else:
+            out["decision"], out["overrides"] = "accept", {}
+
+    return out
+
 
 def route_after_decide(state: ManagerState) -> str:
     """Router: retune, or — on proceed — finalize if Freddi's explanations are
@@ -283,13 +379,18 @@ class ManagerAgent(Agent):
     """
 
     def __init__(self, *, target_accuracy=0.60, max_iterations=5,
+                 patience=2, min_delta=0.01,
                  predictions_path="mock_data/predictions_test.csv",
                  sample_size=300, checkpointer=None, thread_id="manager"):
         # Set once and merged into every run's state; the iteration counter and
-        # decision_log accumulate across runs via the checkpointer.
+        # the history reducers accumulate across runs via the checkpointer.
+        # `patience`/`min_delta` tune early-stopping: proceed once accuracy hasn't
+        # gained `min_delta` over the best of the last `patience` iterations.
         self._defaults = {
             "target_accuracy": target_accuracy,
             "max_iterations": max_iterations,
+            "patience": patience,
+            "min_delta": min_delta,
             "predictions_path": predictions_path,
             "sample_size": sample_size,
         }

agents/nadi_classifier.py

@@ -24,10 +24,7 @@
 
 OUTPUT_DIR = "outputs"
 
-CLASSIFIER_TEMPLATE = """\"\"\"Generated Classifier Script.
-Runs FinBERT inference on processed_data.csv and writes predictions_test.csv.
-\"\"\"
-
+CLASSIFIER_TEMPLATE = """
 import csv
 import os
 import sys
@@ -38,10 +35,15 @@
 MAX_LENGTH = {max_length}
 THRESHOLD = {threshold}
 FOCUS_LABELS = {focus_labels}
+BOOST_FACTOR = {boost_factor}
 SENTIMENT_TO_LABEL = {{"positive": "up", "negative": "down", "neutral": "neutral"}}
 
-tokenizer = AutoTokenizer.from_pretrained(MODEL)
-model = AutoModelForSequenceClassification.from_pretrained(MODEL)
+# Authenticate to the HF Hub when a token is in the env (higher rate limits,
+# faster downloads); fall back to unauthenticated access when it is absent.
+HF_TOKEN = os.environ.get("HF_TOKEN") or os.environ.get("HUGGING_FACE_HUB_TOKEN")
+
+tokenizer = AutoTokenizer.from_pretrained(MODEL, token=HF_TOKEN)
+model = AutoModelForSequenceClassification.from_pretrained(MODEL, token=HF_TOKEN)
 model.eval()
 
 ID2OURS = {{i: SENTIMENT_TO_LABEL[model.config.id2label[i].lower()] for i in model.config.id2label}}
@@ -57,7 +59,7 @@ def classify(title: str) -> dict:
     # Apply class boost to focus labels if specified
     for fl in FOCUS_LABELS:
         if fl in by_label:
-            by_label[fl] *= 1.25
+            by_label[fl] *= BOOST_FACTOR
 
     # Normalize probabilities after boosting
     total_prob = sum(by_label.values())
@@ -115,13 +117,15 @@ def generate_code(state: PipelineState) -> dict:
     threshold = 0.5
     max_length = 128
     focus_labels = []
+    boost_factor = 1.25
 
     # Read from retune request if it exists in state
     retune_req = state.get("retune_request")
     if retune_req:
         suggested = retune_req.get("suggested_params", {})
         threshold = suggested.get("threshold", threshold)
         max_length = suggested.get("max_length", max_length)
+        boost_factor = suggested.get("boost_factor", boost_factor)
         focus_labels = retune_req.get("focus_labels", focus_labels)
 
     code_path = state.get("classifier_code_path") or os.path.join(OUTPUT_DIR, "classifier.py")
@@ -130,25 +134,38 @@ def generate_code(state: PipelineState) -> dict:
     formatted_code = CLASSIFIER_TEMPLATE.format(
         threshold=threshold,
         max_length=max_length,
-        focus_labels=repr(focus_labels)
+        focus_labels=repr(focus_labels),
+        boost_factor=boost_factor
     )
 
     with open(code_path, "w", encoding="utf-8") as f:
         f.write(formatted_code)
 
     print(f"[nadi] Generated classifier code at: {code_path}")
 
+    # Keep a per-iteration copy so past retune attempts aren't lost when
+    # classifier.py (the single contract file Sabina reads) gets overwritten.
+    # iteration 0 = first pass before any retune; N = the Nth retune's code.
+    iteration = retune_req.get("iteration", 0) if retune_req else 0
+    history_dir = os.path.join(os.path.dirname(code_path) or ".", "classifier_history")
+    os.makedirs(history_dir, exist_ok=True)
+    history_path = os.path.join(history_dir, f"classifier_iter{iteration}.py")
+    with open(history_path, "w", encoding="utf-8") as f:
+        f.write(formatted_code)
+
     metadata = {
         "model_name": "ProsusAI/finbert",
         "fine_tuning_params": {
             "threshold": threshold,
             "max_length": max_length,
-            "focus_labels": focus_labels
+            "focus_labels": focus_labels,
+            "boost_factor": boost_factor
         }
     }
 
     return {
         "classifier_code_path": code_path,
+        "classifier_history_path": history_path,
         "classifier_metadata": metadata
     }