Signal Quest: Build an Honest Machine with an LLM¶

Dr. Mallarapu · SEAS-8414 · Research and paper-trading only

This capstone teaches the complete Signal Quest textbook as one reproducible investigation. You will define a settlement rule, freeze an evidence boundary, build causal features, compare models, calibrate probabilities, replay a conservative policy, and monitor the system. The data are synthetic. A good conclusion may be NO_TRADE.

Non-negotiable rule: an LLM may draft bounded code, but it never receives secrets, production data, execution authority, or permission to run its own output. Every candidate is reviewed and tested before use.

Learning map¶

Textbook movement Notebook evidence LLM role
ML, labels, and clocks Synthetic event ledger and label contract Draft a pure feature function
Metrics and fair tests Baseline, chronological split, calibration Explain a failed test, not invent a result
Three model families Tabular baseline, encoder scaffold, causal sequence scaffold Draft isolated modules and tests
Replay and paper trading Conservative cost-aware NO_TRADE policy Propose a test case only
Agentic monitoring Read-only health checks and reversible containment Summarize evidence for a human

Completion bar: all tests pass, the manifest is complete, the LLM trace is saved, and every conclusion is labeled simulated, illustrative, or design target as appropriate.

In [1]:
# Optional student environment. Run once in a fresh notebook kernel if needed.
# %pip install --upgrade numpy pandas scikit-learn
# Optional model-family extensions: %pip install catboost torch

import ast
import hashlib
import json
import os
import textwrap
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path

import numpy as np
import pandas as pd

SEED = 8414
rng = np.random.default_rng(SEED)
WORKDIR = Path("signal_quest_lab")
WORKDIR.mkdir(exist_ok=True)
print(f"Working directory: {WORKDIR.resolve()}")
print("Epistemic status: simulated teaching lab; no live market or order authority.")

Working directory: $TEXTBOOK_ROOT/btc-polymarket-ml/labs/signal_quest_lab
Epistemic status: simulated teaching lab; no live market or order authority.

1. The contract comes before the model¶

At decision time t, a model may receive only events whose receive time is no later than t. The label resolves later. Here, UP means the synthetic settlement price at the end of a five-minute horizon is greater than or equal to the start price. This is a teaching contract, not a claim about a real venue or market.

In [2]:
@dataclass(frozen=True)
class DecisionContract:
    horizon_minutes: int = 5
    tie_rule: str = "UP"
    receive_time_required: bool = True
    research_only: bool = True

CONTRACT = DecisionContract()
print(json.dumps(asdict(CONTRACT), indent=2))

def label_up(start_price: float, end_price: float) -> int:
    """Settlement contract: equality resolves UP."""
    return int(end_price >= start_price)

assert label_up(100.0, 100.0) == 1
assert label_up(100.0, 99.9) == 0

{
  "horizon_minutes": 5,
  "tie_rule": "UP",
  "receive_time_required": true,
  "research_only": true
}

2. Build an event ledger and stop time travel¶

The ledger deliberately contains some events that occurred before a cutoff but arrived too late to be used. This distinction is the central causal test: occurrence time alone is not eligibility.

In [3]:
def synthetic_events(n: int = 720) -> pd.DataFrame:
    event_time = pd.date_range("2026-01-01", periods=n, freq="min", tz="UTC")
    mid = 100_000 + np.cumsum(rng.normal(0, 14, n))
    spread = np.maximum(0.5, rng.lognormal(mean=0.4, sigma=0.35, size=n))
    imbalance = np.tanh(rng.normal(0, 0.9, n))
    receive_delay_seconds = rng.integers(0, 4, n)
    receive_delay_seconds[::71] += 180  # deliberately late records
    return pd.DataFrame({
        "event_time": event_time,
        "receive_time": event_time + pd.to_timedelta(receive_delay_seconds, unit="s"),
        "mid": mid,
        "spread": spread,
        "imbalance": imbalance,
        "event_id": [f"evt-{i:04d}" for i in range(n)],
    })

events = synthetic_events()
events.head()
Out[3]:
event_time receive_time mid spread imbalance event_id
0 2026-01-01 00:00:00+00:00 2026-01-01 00:03:01+00:00 99983.577156 1.086732 -0.531457 evt-0000
1 2026-01-01 00:01:00+00:00 2026-01-01 00:01:00+00:00 99958.995144 0.978767 -0.589098 evt-0001
2 2026-01-01 00:02:00+00:00 2026-01-01 00:02:03+00:00 99965.623805 1.235367 -0.837308 evt-0002
3 2026-01-01 00:03:00+00:00 2026-01-01 00:03:03+00:00 99976.513167 2.715420 -0.673456 evt-0003
4 2026-01-01 00:04:00+00:00 2026-01-01 00:04:02+00:00 99977.172296 0.827841 0.606708 evt-0004
In [4]:
def eligible_events(events: pd.DataFrame, cutoff: pd.Timestamp) -> pd.DataFrame:
    required = {"event_time", "receive_time", "mid", "spread", "imbalance", "event_id"}
    missing = required.difference(events.columns)
    if missing:
        raise ValueError(f"Missing required evidence fields: {sorted(missing)}")
    return events.loc[events["receive_time"] <= cutoff].copy()

cutoff = events.loc[72, "event_time"]  # evt-0071 occurred but is received later.
eligible = eligible_events(events, cutoff)
assert (eligible["receive_time"] <= cutoff).all()
assert len(eligible) < len(events.loc[events["event_time"] <= cutoff])
print(f"Cutoff: {cutoff}; eligible rows: {len(eligible)}")

Cutoff: 2026-01-01 01:12:00+00:00; eligible rows: 71

3. Ask the LLM for a bounded code draft¶

The LLM receives a small, explicit contract. It must return only one pure Python function and tests. The notebook stores its response, hashes it, parses it, and refuses to execute it automatically. Set SIGNAL_QUEST_USE_LLM=1 and OLLAMA_API_KEY only when you are ready to make an API request. Otherwise, the cell emits a reviewable placeholder.

The prompt is outcome-first: required inputs, allowed output, safety boundary, and completion test are explicit.

In [5]:
OLLAMA_MODEL = os.getenv("OLLAMA_MODEL", "kimi-k3:cloud")
OLLAMA_HOST = os.getenv("OLLAMA_HOST", "https://ollama.com").rstrip("/")
USE_LLM = os.getenv("SIGNAL_QUEST_USE_LLM", "0") == "1"

FEATURE_PROMPT = """
Mission: draft one pure Python function named build_features(rows).

Input contract:
- rows is a pandas DataFrame with mid, spread, imbalance, receive_time, and event_time.
- rows contains only events eligible at a decision cutoff.

Output contract:
- Return a dict with float keys: last_mid, mean_spread, mean_imbalance, and return_3.
- Raise ValueError for fewer than four rows or missing fields.

Hard rules:
- Use no imports, file access, network calls, randomness, globals, eval, exec, classes, or model fitting.
- Do not mention market performance or trading.
- Return code only, followed by exactly three assert-style tests in comments.

Completion bar: a reviewer can read the function, parse it, and test it deterministically.
""".strip()

def request_llm_code(prompt: str) -> str:
    """Request an untrusted teaching-code draft from Ollama."""
    if not USE_LLM:
        return "# LLM disabled. Set SIGNAL_QUEST_USE_LLM=1 after reviewing the prompt.\n"
    api_key = os.getenv("OLLAMA_API_KEY")
    if not api_key:
        raise RuntimeError("LLM mode requested but OLLAMA_API_KEY is not set.")
    import urllib.error
    import urllib.request

    payload = json.dumps({
        "model": OLLAMA_MODEL,
        "messages": [{"role": "user", "content": prompt}],
        "stream": False,
        "options": {"temperature": 0},
    }).encode("utf-8")
    request = urllib.request.Request(
        f"{OLLAMA_HOST}/api/chat",
        data=payload,
        headers={"Content-Type": "application/json", "Authorization": f"Bearer {api_key}"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=120) as response:
            result = json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        raise RuntimeError(f"Ollama request failed with HTTP {exc.code}.") from exc
    except urllib.error.URLError as exc:
        raise RuntimeError(f"Ollama is unreachable at {OLLAMA_HOST}.") from exc
    content = result.get("message", {}).get("content")
    if not isinstance(content, str) or not content.strip():
        raise RuntimeError("Ollama returned no message content.")
    return content

llm_draft = request_llm_code(FEATURE_PROMPT)
trace = {
    "timestamp_utc": datetime.now(timezone.utc).isoformat(),
    "provider": "ollama",
    "host": OLLAMA_HOST,
    "model": OLLAMA_MODEL if USE_LLM else "disabled",
    "prompt": FEATURE_PROMPT,
    "response": llm_draft,
    "response_sha256": hashlib.sha256(llm_draft.encode()).hexdigest(),
    "status": "review_required",
}
(WORKDIR / "llm_feature_draft.json").write_text(json.dumps(trace, indent=2))
print(trace["status"], trace["response_sha256"])

review_required a31c43d91a6436fa799ea9d740e80d7ac9841fcf86376d36620b0194e7aa771c
In [6]:
# The LLM's text is data. It is not executed. Validate its AST before human review.
FORBIDDEN = (ast.Import, ast.ImportFrom, ast.Global, ast.Nonlocal, ast.ClassDef, ast.AsyncFunctionDef)
FORBIDDEN_CALLS = {"eval", "exec", "open", "compile", "__import__", "input"}

def static_code_review(candidate: str) -> list[str]:
    findings = []
    try:
        tree = ast.parse(candidate)
    except SyntaxError as exc:
        return [f"syntax error: {exc}"]
    functions = [n for n in tree.body if isinstance(n, ast.FunctionDef)]
    if len(functions) != 1 or functions[0].name != "build_features":
        findings.append("require exactly one function named build_features")
    for node in ast.walk(tree):
        if isinstance(node, FORBIDDEN):
            findings.append(f"forbidden syntax: {type(node).__name__}")
        if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id in FORBIDDEN_CALLS:
            findings.append(f"forbidden call: {node.func.id}")
    return sorted(set(findings))

review_findings = static_code_review(llm_draft) if USE_LLM else ["LLM disabled: no candidate submitted"]
print(review_findings)
print("Human gate: copy an approved candidate into the next cell; do not exec model output.")

['LLM disabled: no candidate submitted']
Human gate: copy an approved candidate into the next cell; do not exec model output.
In [7]:
# Student-controlled implementation after review. This is the only feature code used below.
def build_features(rows: pd.DataFrame) -> dict[str, float]:
    required = {"mid", "spread", "imbalance"}
    if missing := required.difference(rows.columns):
        raise ValueError(f"missing fields: {sorted(missing)}")
    if len(rows) < 4:
        raise ValueError("need at least four rows")
    mid = rows["mid"].astype(float)
    return {
        "last_mid": float(mid.iloc[-1]),
        "mean_spread": float(rows["spread"].mean()),
        "mean_imbalance": float(rows["imbalance"].mean()),
        "return_3": float(mid.iloc[-1] / mid.iloc[-4] - 1.0),
    }

assert set(build_features(eligible.tail(10))) == {"last_mid", "mean_spread", "mean_imbalance", "return_3"}
try:
    build_features(eligible.head(3))
    raise AssertionError("short-window test did not fail")
except ValueError:
    pass

3A. On-demand Code Studio¶

This is the notebook's LLM workbench for the entire book. Choose one topic, write a narrow learning request, and set SIGNAL_QUEST_USE_LLM=1 only after you have read the generated prompt. The LLM may draft a small teaching artifact; it cannot access files, networks, credentials, live markets, or a trading account. Its draft is saved with a trace and is never executed automatically.

In [8]:
LESSON_CATALOG = {
    "contract": {
        "chapter": "Observations, decision time, later settlement",
        "function": "define_decision_contract",
        "goal": "Create a dataclass that rejects a label horizon, tie rule, or evidence boundary that is missing.",
    },
    "features": {
        "chapter": "Causal features and the evidence boundary",
        "function": "build_causal_features",
        "goal": "Create pure rolling features from rows already eligible at a decision cutoff.",
    },
    "leakage": {
        "chapter": "Time travel and leakage tests",
        "function": "assert_no_future_events",
        "goal": "Create a test that fails when an event received after the cutoff is present.",
    },
    "catboost": {
        "chapter": "Tabular probability model",
        "function": "train_tabular_candidate",
        "goal": "Draft a chronological CatBoost training function that returns validation probabilities only.",
    },
    "encoder": {
        "chapter": "Self-supervised limit-order-book encoder",
        "function": "masked_window_loss",
        "goal": "Draft a small tensor-only masked reconstruction loss with shape checks.",
    },
    "transformer": {
        "chapter": "Causal limit-order-book Transformer",
        "function": "causal_attention_mask",
        "goal": "Create a strictly upper-triangular attention mask for an ordered sequence.",
    },
    "calibration": {
        "chapter": "Probabilities, calibration, and abstention",
        "function": "calibration_summary",
        "goal": "Calculate reliability-bin summaries without choosing an action.",
    },
    "replay": {
        "chapter": "Conservative replay and costs",
        "function": "paper_replay_row",
        "goal": "Score one paper-trade record with supplied price, conservative cost, and later outcome.",
    },
    "guardian": {
        "chapter": "Agentic monitoring and reversible containment",
        "function": "guardian_check",
        "goal": "Return an evidence-only incident record; never modify a model, policy, or account.",
    },
}

print("Available topics:", ", ".join(LESSON_CATALOG))

Available topics: contract, features, leakage, catboost, encoder, transformer, calibration, replay, guardian
In [9]:
ON_DEMAND_TOPIC = "leakage"  # Change to any key in LESSON_CATALOG.
ON_DEMAND_REQUEST = "Explain the idea with one small, deterministic function and three assert-style tests."

def on_demand_prompt(topic: str, learner_request: str) -> str:
    if topic not in LESSON_CATALOG:
        raise ValueError(f"Unknown topic {topic!r}. Choose one of: {sorted(LESSON_CATALOG)}")
    lesson = LESSON_CATALOG[topic]
    return f"""
You are the Signal Quest teaching-code assistant. Produce one small, reviewable Python teaching artifact.

Book chapter: {lesson['chapter']}
Required function name: {lesson['function']}
Teaching goal: {lesson['goal']}
Student request: {learner_request}

Hard safety and evidence rules:
- Return code only, followed by exactly three assert-style tests in comments.
- Define exactly one function. Use only parameters passed to that function.
- No imports, files, network, environment variables, randomness, subprocesses, eval, exec, classes, model persistence, live data, or order placement.
- Do not claim predictive performance, profitability, or deployment readiness.
- For model topics, return probabilities or tensors only; do not make a trading decision.
- For guardian topics, return observations and a human-review requirement only.

The output will be statically reviewed by a human before any manual copy into an approved cell.
""".strip()

ON_DEMAND_PROMPT = on_demand_prompt(ON_DEMAND_TOPIC, ON_DEMAND_REQUEST)
on_demand_draft = request_llm_code(ON_DEMAND_PROMPT)
on_demand_trace = {
    "topic": ON_DEMAND_TOPIC,
    "chapter": LESSON_CATALOG[ON_DEMAND_TOPIC]["chapter"],
    "provider": "ollama",
    "model": OLLAMA_MODEL if USE_LLM else "disabled",
    "prompt": ON_DEMAND_PROMPT,
    "response": on_demand_draft,
    "response_sha256": hashlib.sha256(on_demand_draft.encode()).hexdigest(),
    "status": "review_required",
}
(WORKDIR / "llm_on_demand_code_trace.json").write_text(json.dumps(on_demand_trace, indent=2))
print(on_demand_trace["status"], on_demand_trace["topic"], on_demand_trace["response_sha256"])

review_required leakage a31c43d91a6436fa799ea9d740e80d7ac9841fcf86376d36620b0194e7aa771c
In [10]:
ON_DEMAND_FORBIDDEN = {
    "eval", "exec", "open", "compile", "__import__", "input", "requests", "urlopen",
    "subprocess", "system", "popen", "fit", "predict", "save", "load",
}

def review_on_demand_draft(candidate: str, required_function: str) -> list[str]:
    try:
        tree = ast.parse(candidate)
    except SyntaxError as exc:
        return [f"syntax error: {exc}"]
    findings = []
    functions = [node for node in tree.body if isinstance(node, ast.FunctionDef)]
    if len(functions) != 1 or functions[0].name != required_function:
        findings.append(f"require exactly one function named {required_function}")
    for node in ast.walk(tree):
        if isinstance(node, FORBIDDEN):
            findings.append(f"forbidden syntax: {type(node).__name__}")
        if isinstance(node, ast.Attribute) and node.attr.startswith("__"):
            findings.append("forbidden dunder attribute")
        if isinstance(node, ast.Call):
            name = node.func.id if isinstance(node.func, ast.Name) else getattr(node.func, "attr", "")
            if name in ON_DEMAND_FORBIDDEN:
                findings.append(f"forbidden call: {name}")
    return sorted(set(findings))

if USE_LLM:
    on_demand_findings = review_on_demand_draft(
        on_demand_draft, LESSON_CATALOG[ON_DEMAND_TOPIC]["function"]
    )
else:
    on_demand_findings = ["LLM disabled: set SIGNAL_QUEST_USE_LLM=1 and supply OLLAMA_API_KEY."]
print(on_demand_findings)
print("Human gate: inspect the trace, review the draft, then manually copy an approved function below.")

['LLM disabled: set SIGNAL_QUEST_USE_LLM=1 and supply OLLAMA_API_KEY.']
Human gate: inspect the trace, review the draft, then manually copy an approved function below.

Human approval gate¶

The LLM draft is not program behavior. Read the saved trace, check the static findings, test the artifact in isolation, and then manually copy only an approved version into a new cell. If the draft fails the evidence boundary or safety contract, reject it and improve the prompt—do not weaken the guardrail.

4. Freeze features, then define a later label¶

Each row below is a decision record. Features are built from an eligible history at the cutoff. The label is resolved five minutes later. The notebook discards unresolved horizons instead of guessing.

In [11]:
def make_dataset(events: pd.DataFrame, lookback: int = 20, horizon: int = 5) -> pd.DataFrame:
    rows = []
    for i in range(lookback, len(events) - horizon):
        cutoff = events.loc[i, "event_time"]
        history = eligible_events(events.iloc[: i + 1], cutoff).tail(lookback)
        if len(history) < lookback:
            continue
        features = build_features(history)
        start_mid = float(events.loc[i, "mid"])
        end_mid = float(events.loc[i + horizon, "mid"])
        rows.append({**features, "cutoff": cutoff, "label": label_up(start_mid, end_mid)})
    return pd.DataFrame(rows)

dataset = make_dataset(events)
assert dataset["cutoff"].is_monotonic_increasing
assert dataset["label"].isin([0, 1]).all()
dataset.head()
Out[11]:
last_mid mean_spread mean_imbalance return_3 cutoff label
0 99960.588725 1.535299 0.049990 0.000086 2026-01-01 00:20:00+00:00 0
1 99976.596972 1.615599 0.083696 0.000199 2026-01-01 00:21:00+00:00 1
2 99965.585331 1.677618 0.080641 0.000144 2026-01-01 00:22:00+00:00 0
3 99978.688297 1.675101 0.120764 0.000181 2026-01-01 00:23:00+00:00 1
4 99954.378232 1.583342 0.110286 -0.000222 2026-01-01 00:24:00+00:00 1

5. Evaluate fairly: chronological split, simple baseline, and calibration¶

Random splitting would let nearby observations leak across train and test. We use time order. The baseline is the training prevalence; a score only earns attention if it beats that reference under the same evidence contract.

In [12]:
def chronological_split(frame: pd.DataFrame, train_fraction: float = 0.60, valid_fraction: float = 0.20):
    n = len(frame)
    a, b = int(n * train_fraction), int(n * (train_fraction + valid_fraction))
    return frame.iloc[:a].copy(), frame.iloc[a:b].copy(), frame.iloc[b:].copy()

train, valid, test = chronological_split(dataset)
base_probability = float(train["label"].mean())

def brier(y: pd.Series, p: np.ndarray) -> float:
    return float(np.mean((np.asarray(y) - np.asarray(p)) ** 2))

baseline_brier = brier(test["label"], np.full(len(test), base_probability))
print({"train": len(train), "valid": len(valid), "test": len(test), "base_rate": base_probability, "baseline_brier": baseline_brier})

{'train': 417, 'valid': 139, 'test': 139, 'base_rate': 0.4460431654676259, 'baseline_brier': 0.2571813053154599}
In [13]:
# A small transparent research baseline, not a production model.
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score, confusion_matrix

FEATURES = ["mean_spread", "mean_imbalance", "return_3"]
baseline_model = LogisticRegression(random_state=SEED, class_weight="balanced")
baseline_model.fit(train[FEATURES], train["label"])
valid_probability = baseline_model.predict_proba(valid[FEATURES])[:, 1]
test_probability = baseline_model.predict_proba(test[FEATURES])[:, 1]

print("Validation Brier:", brier(valid["label"], valid_probability))
print("Test Brier:", brier(test["label"], test_probability))
print("Test accuracy:", accuracy_score(test["label"], test_probability >= 0.5))
print("Confusion matrix:\n", confusion_matrix(test["label"], test_probability >= 0.5))
print("Interpretation: simulated result only. Accuracy alone is not a claim of usefulness.")

Validation Brier: 0.25171696377318237
Test Brier: 0.2710190132809431
Test accuracy: 0.39568345323741005
Confusion matrix:
 [[44 20]
 [64 11]]
Interpretation: simulated result only. Accuracy alone is not a claim of usefulness.
In [14]:
def calibration_table(y: pd.Series, p: np.ndarray, bins: int = 5) -> pd.DataFrame:
    frame = pd.DataFrame({"y": np.asarray(y), "p": np.asarray(p)})
    frame["bin"] = pd.cut(frame["p"], bins=np.linspace(0, 1, bins + 1), include_lowest=True)
    return frame.groupby("bin", observed=False).agg(count=("y", "size"), predicted=("p", "mean"), observed=("y", "mean")).reset_index()

calibration_table(valid["label"], valid_probability)
Out[14]:
bin count predicted observed
0 (-0.001, 0.2] 0 NaN NaN
1 (0.2, 0.4] 0 NaN NaN
2 (0.4, 0.6] 139 0.490995 0.388489
3 (0.6, 0.8] 0 NaN NaN
4 (0.8, 1.0] 0 NaN NaN

6. The three-model ladder¶

The first rung is a tabular candidate. The second learns a representation from permitted unlabeled sequence windows. The third applies a causal mask to ordered limit-order-book states. None proves an edge by existing. Each must survive the same temporal, calibration, replay, and safety gates.

Use the LLM to draft one module at a time from the prompts below. Review the output exactly as you reviewed build_features; do not paste generated code directly into a training run.

In [15]:
MODEL_PROMPTS = {
    "catboost": "Draft a single Python function train_tabular_candidate(X_train, y_train, X_valid, y_valid). Use CatBoostClassifier only; no file, network, or plotting access. Return model and validation probabilities. Include an early-stopping parameter and do not claim performance.",
    "encoder": "Draft a PyTorch nn.Module named MaskedLOBEncoder. It accepts [batch, time, features], masks only supplied positions, and returns embeddings. Do not create data loaders, files, or training loops. Include shape assertions in comments.",
    "causal_transformer": "Draft a PyTorch nn.Module named CausalLOBTransformer. It accepts ordered [batch, time, features] data and applies a strictly upper-triangular causal attention mask. Return one probability logit per sequence. No file, network, or execution code.",
}
for name, prompt in MODEL_PROMPTS.items():
    print(f"\n{name.upper()} PROMPT\n{textwrap.fill(prompt, width=100)}")

CATBOOST PROMPT
Draft a single Python function train_tabular_candidate(X_train, y_train, X_valid, y_valid). Use
CatBoostClassifier only; no file, network, or plotting access. Return model and validation
probabilities. Include an early-stopping parameter and do not claim performance.

ENCODER PROMPT
Draft a PyTorch nn.Module named MaskedLOBEncoder. It accepts [batch, time, features], masks only
supplied positions, and returns embeddings. Do not create data loaders, files, or training loops.
Include shape assertions in comments.

CAUSAL_TRANSFORMER PROMPT
Draft a PyTorch nn.Module named CausalLOBTransformer. It accepts ordered [batch, time, features]
data and applies a strictly upper-triangular causal attention mask. Return one probability logit per
sequence. No file, network, or execution code.
In [16]:
# Design-target acceptance tests for every advanced candidate.
# These tests are intentionally stated before implementation.
ADVANCED_MODEL_GATES = [
    "chronological train/validation/test split is recorded",
    "normalization fit uses training period only",
    "sequence window contains no record received after its cutoff",
    "causal mask blocks every future position",
    "probabilities are calibrated on validation data only",
    "comparison includes baseline and conservative replay",
    "artifact manifest records code, config, data, and model hashes",
]
for gate in ADVANCED_MODEL_GATES:
    print("[ ]", gate)

[ ] chronological train/validation/test split is recorded
[ ] normalization fit uses training period only
[ ] sequence window contains no record received after its cutoff
[ ] causal mask blocks every future position
[ ] probabilities are calibrated on validation data only
[ ] comparison includes baseline and conservative replay
[ ] artifact manifest records code, config, data, and model hashes

7. Probability is not permission¶

A model probability becomes a possible action only after conservative cost, calibration, data-health, and risk gates. The policy below is deliberately fail-closed: missing evidence, a bad calibration result, or an unsafe cost assumption produces NO_TRADE. It does not place orders.

In [17]:
@dataclass(frozen=True)
class PolicyInput:
    probability_up: float
    executable_up_price: float
    conservative_cost: float
    calibration_ok: bool
    data_health_ok: bool
    risk_ok: bool

def decide_paper_only(x: PolicyInput, buffer: float = 0.03) -> str:
    if not all([x.calibration_ok, x.data_health_ok, x.risk_ok]):
        return "NO_TRADE"
    conservative_probability = x.probability_up - buffer
    return "PAPER_UP" if conservative_probability > x.executable_up_price + x.conservative_cost else "NO_TRADE"

assert decide_paper_only(PolicyInput(.80, .60, .02, True, True, True)) == "PAPER_UP"
assert decide_paper_only(PolicyInput(.80, .60, .02, False, True, True)) == "NO_TRADE"

8. Replay the policy under a reconstructed clock¶

This minimal replay uses synthetic executable prices and conservative costs. It does not model a live book and it does not predict a real outcome. Its purpose is to make assumptions inspectable and to demonstrate that a score may still lead to abstention.

In [18]:
def paper_replay(frame: pd.DataFrame, probabilities: np.ndarray, cost: float = 0.02) -> pd.DataFrame:
    ledger = []
    for (_, row), probability in zip(frame.iterrows(), probabilities):
        synthetic_price = float(np.clip(0.50 + 4 * row["return_3"], 0.05, 0.95))
        decision = decide_paper_only(PolicyInput(float(probability), synthetic_price, cost, True, True, True))
        ledger.append({
            "cutoff": row["cutoff"],
            "probability_up": float(probability),
            "synthetic_executable_price": synthetic_price,
            "cost": cost,
            "decision": decision,
            "later_label": int(row["label"]),
            "epistemic_status": "simulated",
        })
    return pd.DataFrame(ledger)

ledger = paper_replay(test, test_probability)
assert set(ledger["decision"]) <= {"PAPER_UP", "NO_TRADE"}
ledger["decision"].value_counts(dropna=False)
Out[18]:
decision
NO_TRADE    139
Name: count, dtype: int64

9. Agentic monitoring: observe, contain, escalate¶

The guardian may summarize evidence, quarantine a bad artifact, and request human review. It may not alter a model, policy, dataset, account, or external system. The LLM may draft a human-readable incident summary only from the provided facts.

In [19]:
def guardian_check(frame: pd.DataFrame, calibration_ok: bool, manifest_ok: bool) -> dict:
    issues = []
    if not frame["cutoff"].is_monotonic_increasing:
        issues.append("non_monotonic_cutoff")
    if not calibration_ok:
        issues.append("calibration_gate_failed")
    if not manifest_ok:
        issues.append("artifact_manifest_missing")
    return {
        "status": "HEALTHY" if not issues else "QUARANTINE_AND_ESCALATE",
        "issues": issues,
        "allowed_actions": ["write_incident_record", "quarantine_research_artifact", "request_human_review"],
        "forbidden_actions": ["place_order", "change_model", "change_policy", "delete_evidence"],
    }

guardian = guardian_check(ledger, calibration_ok=True, manifest_ok=True)
print(json.dumps(guardian, indent=2))

{
  "status": "HEALTHY",
  "issues": [],
  "allowed_actions": [
    "write_incident_record",
    "quarantine_research_artifact",
    "request_human_review"
  ],
  "forbidden_actions": [
    "place_order",
    "change_model",
    "change_policy",
    "delete_evidence"
  ]
}
In [20]:
INCIDENT_PROMPT = """
Summarize this research incident for a human reviewer. Use only supplied facts.
State: observed evidence, allowed reversible containment, required human decision, and what cannot be inferred.
Do not recommend trading, policy changes, or automatic remediation.
Facts: {facts}
""".strip()

incident_trace = {
    "prompt": INCIDENT_PROMPT.format(facts=json.dumps(guardian)),
    "allowed_output": "human-readable incident summary",
    "requires_human_review": True,
    "epistemic_status": "design target",
}
(WORKDIR / "guardian_llm_prompt.json").write_text(json.dumps(incident_trace, indent=2))
print("Saved bounded incident-summary prompt; no remediation is invoked.")

Saved bounded incident-summary prompt; no remediation is invoked.

10. Reproducibility and doctoral defense¶

A result without a data boundary, code trace, configuration, and limitations is not a result that another student can audit. The final manifest records this lab's synthetic inputs and prompts. It is not a benchmark report.

In [21]:
def sha256_path(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()

manifest = {
    "title": "Signal Quest LLM-guided capstone",
    "epistemic_status": "simulated teaching lab",
    "decision_contract": asdict(CONTRACT),
    "seed": SEED,
    "rows": {"events": len(events), "dataset": len(dataset), "test": len(test)},
    "feature_draft": "llm_feature_draft.json",
    "feature_draft_sha256": sha256_path(WORKDIR / "llm_feature_draft.json"),
    "policy": "paper-only, fail-closed",
    "guardian": guardian,
    "limitations": [
        "synthetic data",
        "no executable venue data",
        "no live performance claim",
        "no order authority",
        "advanced-model cells are design-target scaffolds until independently implemented and tested",
    ],
}
(WORKDIR / "experiment_manifest.json").write_text(json.dumps(manifest, indent=2))
print(json.dumps(manifest, indent=2))

{
  "title": "Signal Quest LLM-guided capstone",
  "epistemic_status": "simulated teaching lab",
  "decision_contract": {
    "horizon_minutes": 5,
    "tie_rule": "UP",
    "receive_time_required": true,
    "research_only": true
  },
  "seed": 8414,
  "rows": {
    "events": 720,
    "dataset": 695,
    "test": 139
  },
  "feature_draft": "llm_feature_draft.json",
  "feature_draft_sha256": "b74624f36136a6c2a34ab5df574afcaceae1e4f09471393cf1bab87181b2bcf3",
  "policy": "paper-only, fail-closed",
  "guardian": {
    "status": "HEALTHY",
    "issues": [],
    "allowed_actions": [
      "write_incident_record",
      "quarantine_research_artifact",
      "request_human_review"
    ],
    "forbidden_actions": [
      "place_order",
      "change_model",
      "change_policy",
      "delete_evidence"
    ]
  },
  "limitations": [
    "synthetic data",
    "no executable venue data",
    "no live performance claim",
    "no order authority",
    "advanced-model cells are design-target scaffolds until independently implemented and tested"
  ]
}

Capstone defense prompts¶

  1. Show one event that occurred before the cutoff but arrived too late. Why is it forbidden?
  2. Read the LLM prompt and trace. What exact code behavior was requested, and what behavior was prohibited?
  3. Why is the student-controlled feature function the only function used in the experiment?
  4. Compare the baseline Brier score to the candidate score. What does the comparison fail to prove?
  5. Which advanced model gate would catch a future-looking sequence implementation?
  6. Find a case where the policy abstains despite a high model probability.
  7. What can the guardian do automatically, and what is it structurally unable to do?
  8. State the strongest defensible conclusion from this notebook without using the words “profitable,” “edge,” or “prediction success.”