Signal Quest Technical Masterclass

Executable teaching artifact · research only
Every numerical result in this notebook is Simulated or Illustrative. Nothing is measured BTC or Polymarket performance, a live fill, financial advice, or permission to trade.

This notebook turns Chapters 4–13 into one deterministic evidence trace: metrics → temporal evaluation → a demanding tabular fallback → calibration → abstention → representation objectives → causal attention → cost-aware replay → fail-closed monitoring.

Epistemic vocabulary. `Simulated` means generated by the seeded synthetic process below. `Illustrative` means formula-derived teaching arithmetic. `Implemented` means only that this notebook component executes; it does not establish market validity. `Design target` describes the absent research system.

Authority boundary. The notebook has no market data, credentials, order API, live execution path, or autonomous code-execution authority.

Learning contract

By the end, you should be able to:

1. reconstruct threshold metrics from a confusion matrix and distinguish ROC, precision–recall, and calibration;
2. derive purge and embargo gaps from a temporal contract;
3. explain why the scikit-learn gradient-boosting model here is a teaching fallback—not CatBoost and not a production candidate;
4. calculate a causal attention mask and a masked-sequence objective;
5. audit a replay that preserves fees, latency, partial fills, no fills, and no-trade decisions; and
6. prove that a guardian fails closed and lacks order authority.

The positive label is `Up = 1`. Threshold selection uses validation only. The test interval remains untouched until the complete forecast rule is frozen.

from __future__ import annotations

import contextlib
import io
import math
import os
import platform
import re
import warnings
import tempfile
from pathlib import Path
from importlib.metadata import distributions
from dataclasses import dataclass
from enum import Enum

os.environ.setdefault("MPLCONFIGDIR", str(Path(tempfile.gettempdir()) / "signal-quest-mpl"))

import matplotlib
matplotlib.use("module://matplotlib_inline.backend_inline")
warnings.filterwarnings("ignore", message="FigureCanvasAgg is non-interactive")
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
from IPython.display import Markdown, display

SEED = 8414
rng = np.random.default_rng(SEED)
np.set_printoptions(precision=4, suppress=True)
pd.set_option("display.max_columns", 30)
pd.set_option("display.float_format", lambda value: f"{value:,.4f}")

COLORS = {
    "ink": "#17212B", "teal": "#0E928C", "coral": "#DF5D4D",
    "gold": "#FFC857", "blue": "#415EB1", "paper": "#FFFFFF",
    "mist": "#E9F5F4", "gray": "#7A8793",
}
plt.rcParams.update({
    "figure.facecolor": COLORS["paper"], "axes.facecolor": COLORS["paper"],
    "axes.edgecolor": COLORS["ink"], "axes.labelcolor": COLORS["ink"],
    "text.color": COLORS["ink"], "xtick.color": COLORS["ink"],
    "ytick.color": COLORS["ink"], "font.size": 10.5,
    "axes.titleweight": "bold", "axes.spines.top": False,
    "axes.spines.right": False, "figure.dpi": 115,
})

EXPECTED_RUNTIME = {"python_major_minor": "3.11"}
EXPECTED_LOCK = {'appnope': '0.1.4', 'asttokens': '3.0.2', 'attrs': '26.1.0', 'beautifulsoup4': '4.15.0', 'bleach': '6.4.0', 'comm': '0.2.3', 'contourpy': '1.3.3', 'cycler': '0.12.1', 'debugpy': '1.8.21', 'defusedxml': '0.7.1', 'executing': '2.2.1', 'fastjsonschema': '2.22.1', 'fonttools': '4.63.0', 'ipykernel': '7.2.0', 'ipython': '9.16.1', 'ipython-pygments-lexers': '1.1.1', 'jedi': '0.20.0', 'jinja2': '3.1.6', 'joblib': '1.5.3', 'jsonschema': '4.26.0', 'jsonschema-specifications': '2025.9.1', 'jupyter-client': '8.9.1', 'jupyter-core': '5.9.1', 'jupyterlab-pygments': '0.3.0', 'kiwisolver': '1.5.0', 'markupsafe': '3.0.3', 'matplotlib': '3.11.1', 'matplotlib-inline': '0.2.2', 'mistune': '3.3.4', 'narwhals': '2.24.0', 'nbclient': '0.11.0', 'nbconvert': '7.17.1', 'nbformat': '5.10.4', 'nest-asyncio': '1.6.0', 'numpy': '1.26.4', 'packaging': '26.3', 'pandas': '3.0.5', 'pandocfilters': '1.5.1', 'parso': '0.8.7', 'pexpect': '4.9.0', 'pillow': '12.3.0', 'platformdirs': '4.11.0', 'prompt-toolkit': '3.0.53', 'psutil': '7.2.2', 'ptyprocess': '0.7.0', 'pure-eval': '0.2.3', 'pygments': '2.20.0', 'pyparsing': '3.3.2', 'python-dateutil': '2.9.0.post0', 'pyzmq': '27.1.0', 'referencing': '0.37.0', 'rpds-py': '2026.6.3', 'scikit-learn': '1.9.0', 'scipy': '1.17.1', 'six': '1.17.0', 'soupsieve': '2.9.1', 'stack-data': '0.6.3', 'threadpoolctl': '3.6.0', 'tinycss2': '1.5.1', 'tornado': '6.5.7', 'traitlets': '5.16.1', 'typing-extensions': '4.16.0', 'wcwidth': '0.8.2', 'webencodings': '0.5.1'}

def canonical_package_name(name: str) -> str:
    return re.sub(r"[-_.]+", "-", name).lower()

installed_lock = {
    canonical_package_name(dist.metadata["Name"]): dist.version
    for dist in distributions()
    if dist.metadata.get("Name")
}
lock_missing = sorted(set(EXPECTED_LOCK) - set(installed_lock))
lock_extra = sorted(set(installed_lock) - set(EXPECTED_LOCK))
lock_mismatches = {
    name: {"expected": version, "observed": installed_lock.get(name)}
    for name, version in EXPECTED_LOCK.items()
    if installed_lock.get(name) != version
}
runtime_matches = (
    platform.python_version().startswith(EXPECTED_RUNTIME["python_major_minor"] + ".")
    and not lock_missing
    and not lock_extra
    and not lock_mismatches
)

versions = pd.Series({
    "python": platform.python_version(),
    "implementation": platform.python_implementation(),
    "locked_packages_verified": f"{len(installed_lock)}/{len(EXPECTED_LOCK)} exact",
    "numpy": np.__version__,
    "pandas": pd.__version__,
    "matplotlib": matplotlib.__version__,
    "seed": SEED,
}, name="value")
display(Markdown("**Implemented — portable runtime identity.** Python is checked by major/minor and every installed distribution must match the complete 64-package lock exactly. No interpreter path is retained."))
display(versions.to_frame())
assert runtime_matches, {
    "python": platform.python_version(),
    "expected_python": EXPECTED_RUNTIME,
    "missing": lock_missing,
    "extra": lock_extra,
    "mismatches": lock_mismatches,
}

Implemented — portable runtime identity. Python is checked by major/minor and every installed distribution must match the complete 64-package lock exactly. No interpreter path is retained.

	value
python	3.11.15
implementation	CPython
locked_packages_verified	64/64 exact
numpy	1.26.4
pandas	3.0.5
matplotlib	3.11.1
seed	8414

1 · Build synthetic evidence with clocks

Simulated — not market data. The generator below creates 960 ordered teaching rows. Its features resemble generic tabular signals only to make the evaluation pipeline concrete. The target comes from a declared latent Bernoulli process. It does not model a venue, order book, settlement source, or executable market.

The feature lookback is 12 rows, label horizon is 5 rows, and illustrative availability allowance is 3 rows. The boundary gap is therefore conservatively set to 20 rows: `12 + 5 + 3`.

def sigmoid(x: np.ndarray | float) -> np.ndarray:
    x_arr = np.asarray(x, dtype=float)
    return 1.0 / (1.0 + np.exp(-np.clip(x_arr, -30, 30)))

N = 960
latent = np.zeros(N)
innovations = rng.normal(0, 0.58, N)
for t in range(1, N):
    latent[t] = 0.88 * latent[t - 1] + innovations[t]

momentum = pd.Series(latent).diff().rolling(4, min_periods=1).mean().fillna(0).to_numpy()
imbalance = np.tanh(0.72 * latent + rng.normal(0, 0.50, N))
spread = np.clip(0.045 + 0.018 * np.abs(latent) + rng.normal(0, 0.006, N), 0.008, None)
activity = np.exp(np.clip(1.65 + 0.20 * latent + rng.normal(0, 0.22, N), 0.4, 3.0))
time_wave = np.sin(np.arange(N) / 33.0)

true_logit = -0.28 + 1.12 * momentum + 0.76 * imbalance - 3.2 * spread + 0.22 * time_wave
true_probability = sigmoid(true_logit)
y = rng.binomial(1, true_probability)
reference_probability = np.clip(0.24 + 0.08 * sigmoid(0.55 * latent + 0.35 * time_wave), 0.05, 0.95)

data = pd.DataFrame({
    "row_id": np.arange(N),
    "event_time": pd.date_range("2026-01-01", periods=N, freq="5min", tz="UTC"),
    "receive_delay_ms": 35 + (np.arange(N) * 17) % 180,
    "momentum": momentum,
    "imbalance": imbalance,
    "spread": spread,
    "activity": activity,
    "time_wave": time_wave,
    "reference_probability": reference_probability,
    "label_up": y,
})
FEATURES = ["momentum", "imbalance", "spread", "activity", "time_wave"]

display(Markdown("**Simulated — seeded synthetic generator.**"))
display(data.head(6))
print(f"Rows: {len(data):,} | positive prevalence: {data.label_up.mean():.3f} | seed: {SEED}")

Simulated — seeded synthetic generator.

	row_id	event_time	receive_delay_ms	momentum	imbalance	spread	activity	time_wave	reference_probability	label_up
0	0	2026-01-01 00:00:00+00:00	35	0.0000	-0.5642	0.0512	4.1862	0.0000	0.2800	1
1	1	2026-01-01 00:05:00+00:00	52	-1.0184	-0.7903	0.0627	4.4279	0.0303	0.2693	0
2	2	2026-01-01 00:10:00+00:00	69	-0.3108	0.0621	0.0543	6.9573	0.0606	0.2736	1
3	3	2026-01-01 00:15:00+00:00	86	-0.0320	0.7418	0.0467	4.6290	0.0908	0.2796	1
4	4	2026-01-01 00:20:00+00:00	103	-0.0143	-0.8465	0.0434	5.1543	0.1209	0.2802	0
5	5	2026-01-01 00:25:00+00:00	120	-0.0642	-0.8803	0.0577	3.0065	0.1509	0.2675	0

Rows: 960 | positive prevalence: 0.371 | seed: 8414

fig, axes = plt.subplots(2, 1, figsize=(11, 5.8), sharex=True, constrained_layout=True)
axes[0].plot(data.row_id, data.imbalance, color=COLORS["teal"], lw=1.4, label="synthetic imbalance")
axes[0].plot(data.row_id, data.momentum, color=COLORS["blue"], lw=1.0, alpha=0.8, label="synthetic momentum")
axes[0].axhline(0, color=COLORS["ink"], lw=0.8)
axes[0].set_ylabel("feature value")
axes[0].legend(frameon=False, ncol=2, loc="upper right")
axes[1].plot(data.row_id, true_probability, color=COLORS["coral"], lw=1.4, label="latent event probability")
axes[1].scatter(data.row_id[::8], data.label_up[::8], s=10, color=COLORS["ink"], alpha=0.50, label="sampled label")
axes[1].set(xlabel="ordered synthetic decision row", ylabel="probability / label", ylim=(-0.05, 1.05))
axes[1].legend(frameon=False, ncol=2, loc="upper right")
fig.suptitle("SIMULATED · Seeded teaching evidence, not market observations", fontsize=14, fontweight="bold")
plt.show()

Exercise 1 — contract defense

Change the label horizon from 5 to 8 rows. Which split boundaries must move, and why would changing only the displayed gap fail to protect the experiment? Then identify one way `receive_delay_ms` could invalidate an event-time-only feature builder.

2 · Temporal split, purge, and embargo

Illustrative contract; implemented split. Training ends before a 20-row purge. Validation follows. A separate 20-row embargo precedes the untouched test. These widths teach derivation from declared windows; they are not recommended durations for any real system.

LOOKBACK_ROWS = 12
LABEL_HORIZON_ROWS = 5
AVAILABILITY_ALLOWANCE_ROWS = 3
GAP_ROWS = LOOKBACK_ROWS + LABEL_HORIZON_ROWS + AVAILABILITY_ALLOWANCE_ROWS

split = {
    "train": np.arange(0, 540),
    "purge": np.arange(540, 560),
    "validation": np.arange(560, 720),
    "embargo": np.arange(720, 740),
    "test": np.arange(740, 960),
}
assert GAP_ROWS == len(split["purge"]) == len(split["embargo"])
assert max(split["train"]) < min(split["purge"]) < min(split["validation"])
assert max(split["validation"]) < min(split["embargo"]) < min(split["test"])

def support_interval(decision_row: int) -> dict[str, int]:
    return {
        "feature_start": decision_row - LOOKBACK_ROWS + 1,
        "feature_end": decision_row,
        "label_start": decision_row + 1,
        "label_end": decision_row + LABEL_HORIZON_ROWS,
        "available_end": decision_row + LABEL_HORIZON_ROWS + AVAILABILITY_ALLOWANCE_ROWS,
    }

boundary_support_audit = pd.DataFrame([
    {"boundary": "train→validation",
     "left_available_end": support_interval(int(max(split["train"])))["available_end"],
     "right_feature_start": support_interval(int(min(split["validation"])))["feature_start"]},
    {"boundary": "validation→test",
     "left_available_end": support_interval(int(max(split["validation"])))["available_end"],
     "right_feature_start": support_interval(int(min(split["test"])))["feature_start"]},
])
boundary_support_audit["disjoint"] = (
    boundary_support_audit["left_available_end"] < boundary_support_audit["right_feature_start"]
)
assert boundary_support_audit["disjoint"].all(), boundary_support_audit.to_dict("records")

segments = [
    ("TRAIN", 0, 540, COLORS["teal"]),
    ("PURGE", 540, 560, COLORS["coral"]),
    ("VALIDATION", 560, 720, COLORS["blue"]),
    ("EMBARGO", 720, 740, COLORS["gold"]),
    ("UNTOUCHED TEST", 740, 960, COLORS["ink"]),
]
fig, ax = plt.subplots(figsize=(11, 2.8), constrained_layout=True)
for label, start, end, color in segments:
    ax.barh(0, end - start, left=start, height=0.48, color=color, edgecolor="white")
    ax.text((start + end) / 2, 0, f"{label}\n{end-start} rows", ha="center", va="center",
            color="white" if label != "EMBARGO" else COLORS["ink"], fontweight="bold", fontsize=9)
ax.annotate(f"gap = lookback {LOOKBACK_ROWS} + horizon {LABEL_HORIZON_ROWS} + availability {AVAILABILITY_ALLOWANCE_ROWS}",
            xy=(550, 0.31), xytext=(480, 0.85), arrowprops={"arrowstyle": "->", "color": COLORS["coral"]},
            ha="center", color=COLORS["coral"], fontweight="bold")
ax.set(xlim=(0, N), ylim=(-0.55, 1.1), xlabel="ordered synthetic row", yticks=[])
ax.set_title("ILLUSTRATIVE CONTRACT · Chronology with purge and embargo")
plt.show()

display(pd.DataFrame({
    "component": ["feature lookback", "label horizon", "availability allowance", "derived gap"],
    "rows": [LOOKBACK_ROWS, LABEL_HORIZON_ROWS, AVAILABILITY_ALLOWANCE_ROWS, GAP_ROWS],
    "epistemic_status": ["Illustrative"] * 4,
}))
display(boundary_support_audit)

	component	rows	epistemic_status
0	feature lookback	12	Illustrative
1	label horizon	5	Illustrative
2	availability allowance	3	Illustrative
3	derived gap	20	Illustrative

	boundary	left_available_end	right_feature_start	disjoint
0	train→validation	547	549	True
1	validation→test	727	729	True

Exercise 2 — leakage audit

A row at index 538 uses a 12-row lookback and its label resolves 5 rows later. Draw its full support. Explain why chronology without purging can still leak. Then propose a boundary assertion that operates on raw-event identifiers rather than row numbers.

3 · Demanding tabular fallback

Design boundary. CatBoost is absent and is not emulated. The intended teaching baseline is scikit-learn's `GradientBoostingClassifier`, clearly labeled a fallback challenger rather than CatBoost. A complex model earns attention only after it survives the same temporal split and calibration/replay contract.

If the verified environment cannot import scikit-learn, the cell fails safely into a small NumPy boosted-stump teaching surrogate so later arithmetic remains inspectable. That branch is an environment limitation, not a substitute for the requested scikit-learn evidence.

SKLEARN_OK = False
SKLEARN_ERROR = None
_import_stderr = io.StringIO()
with warnings.catch_warnings(), contextlib.redirect_stderr(_import_stderr):
    warnings.simplefilter("ignore")
    try:
        import sklearn
        from sklearn.ensemble import GradientBoostingClassifier
        SKLEARN_OK = True
    except Exception as exc:
        SKLEARN_ERROR = f"{type(exc).__name__}: {exc}"

class NumpyBoostedStumps:
    """Small deterministic teaching surrogate; not sklearn and not CatBoost."""
    def __init__(self, n_estimators: int = 45, learning_rate: float = 0.08):
        self.n_estimators = n_estimators
        self.learning_rate = learning_rate
        self.stumps: list[tuple[int, float, float, float]] = []

    def fit(self, x: np.ndarray, target: np.ndarray) -> "NumpyBoostedStumps":
        prevalence = np.clip(target.mean(), 1e-5, 1 - 1e-5)
        self.base_logit = float(np.log(prevalence / (1 - prevalence)))
        score = np.full(len(target), self.base_logit)
        quantiles = np.linspace(0.15, 0.85, 8)
        for _ in range(self.n_estimators):
            residual = target - sigmoid(score)
            best = None
            for feature in range(x.shape[1]):
                for threshold in np.unique(np.quantile(x[:, feature], quantiles)):
                    left = x[:, feature] <= threshold
                    if left.sum() < 8 or (~left).sum() < 8:
                        continue
                    left_value = float(residual[left].mean())
                    right_value = float(residual[~left].mean())
                    prediction = np.where(left, left_value, right_value)
                    loss = float(np.mean((residual - prediction) ** 2))
                    if best is None or loss < best[0]:
                        best = (loss, feature, float(threshold), left_value, right_value)
            _, feature, threshold, left_value, right_value = best
            self.stumps.append((feature, threshold, left_value, right_value))
            score += self.learning_rate * np.where(x[:, feature] <= threshold, left_value, right_value)
        return self

    def predict_proba(self, x: np.ndarray) -> np.ndarray:
        score = np.full(len(x), self.base_logit)
        for feature, threshold, left_value, right_value in self.stumps:
            score += self.learning_rate * np.where(x[:, feature] <= threshold, left_value, right_value)
        probability = sigmoid(score)
        return np.column_stack([1 - probability, probability])

x_train = data.loc[split["train"], FEATURES].to_numpy()
y_train = data.loc[split["train"], "label_up"].to_numpy()
x_val = data.loc[split["validation"], FEATURES].to_numpy()
y_val = data.loc[split["validation"], "label_up"].to_numpy()
x_test = data.loc[split["test"], FEATURES].to_numpy()
y_test = data.loc[split["test"], "label_up"].to_numpy()

if SKLEARN_OK:
    model = GradientBoostingClassifier(
        random_state=SEED, n_estimators=90, learning_rate=0.05,
        max_depth=2, min_samples_leaf=18, subsample=0.85,
    )
    model.fit(x_train, y_train)
    model_name = "scikit-learn GradientBoostingClassifier fallback (not CatBoost)"
    model_status = "Implemented teaching fallback"
else:
    model = NumpyBoostedStumps().fit(x_train, y_train)
    model_name = "NumPy boosted-stump environment surrogate (not sklearn; not CatBoost)"
    model_status = "LIMITATION — sklearn import failed closed"

p_val = model.predict_proba(x_val)[:, 1]
p_test = model.predict_proba(x_test)[:, 1]
model_record = pd.Series({
    "model": model_name,
    "status": model_status,
    "CatBoost imported": False,
    "scikit-learn available": SKLEARN_OK,
    "scikit-learn version": sklearn.__version__ if SKLEARN_OK else "unavailable",
    "failure detail": SKLEARN_ERROR or "none",
    "fit rows": len(x_train),
    "validation rows": len(x_val),
    "test rows": len(x_test),
})
display(model_record.to_frame("value"))
print("SIMULATED teaching output. No market-performance claim.")

	value
model	scikit-learn GradientBoostingClassifier fallba...
status	Implemented teaching fallback
CatBoost imported	False
scikit-learn available	True
scikit-learn version	1.9.0
failure detail	none
fit rows	540
validation rows	160
test rows	220

SIMULATED teaching output. No market-performance claim.

4 · Confusion matrix and threshold metric table

Simulated held-out result. The classification threshold is selected on validation by maximum F1, with deterministic tie-breaking toward the higher threshold. The frozen threshold is then applied once to the untouched synthetic test interval. Accuracy, precision, recall, specificity, F1, and MCC answer different questions; none certifies calibration or value.

def confusion_counts(target: np.ndarray, probability: np.ndarray, threshold: float) -> dict[str, int]:
    pred = probability >= threshold
    return {
        "TN": int(np.sum((target == 0) & (~pred))),
        "FP": int(np.sum((target == 0) & pred)),
        "FN": int(np.sum((target == 1) & (~pred))),
        "TP": int(np.sum((target == 1) & pred)),
    }

def threshold_metrics(target: np.ndarray, probability: np.ndarray, threshold: float) -> dict[str, float]:
    c = confusion_counts(target, probability, threshold)
    tn, fp, fn, tp = c["TN"], c["FP"], c["FN"], c["TP"]
    safe = lambda numerator, denominator: numerator / denominator if denominator else np.nan
    precision = safe(tp, tp + fp)
    recall = safe(tp, tp + fn)
    specificity = safe(tn, tn + fp)
    denominator = math.sqrt((tp + fp) * (tp + fn) * (tn + fp) * (tn + fn))
    return {
        "threshold": threshold,
        "accuracy": safe(tp + tn, tp + tn + fp + fn),
        "precision": precision,
        "recall": recall,
        "specificity": specificity,
        "f1": safe(2 * precision * recall, precision + recall),
        "mcc": safe(tp * tn - fp * fn, denominator),
        **c,
    }

candidate_thresholds = np.linspace(0.05, 0.95, 181)
val_sweep = pd.DataFrame([threshold_metrics(y_val, p_val, t) for t in candidate_thresholds])
selected_threshold = float(val_sweep.sort_values(["f1", "threshold"], ascending=[False, False]).iloc[0].threshold)
test_metric = threshold_metrics(y_test, p_test, selected_threshold)
metric_table = pd.DataFrame({
    "metric": ["accuracy", "precision", "recall", "specificity", "f1", "mcc"],
    "value": [test_metric[k] for k in ["accuracy", "precision", "recall", "specificity", "f1", "mcc"]],
    "question": [
        "fraction of all hard decisions correct", "reliability of positive decisions",
        "share of positives detected", "share of negatives rejected",
        "harmonic precision–recall summary", "association using all four cells",
    ],
    "epistemic_status": "Simulated",
})
cm = np.array([[test_metric["TN"], test_metric["FP"]], [test_metric["FN"], test_metric["TP"]]])

fig, ax = plt.subplots(figsize=(5.4, 4.6), constrained_layout=True)
image = ax.imshow(cm, cmap="GnBu", vmin=0)
for (row, col), value in np.ndenumerate(cm):
    ax.text(col, row, str(value), ha="center", va="center", fontsize=20, fontweight="bold",
            color="white" if value > cm.max() * 0.55 else COLORS["ink"])
ax.set(xticks=[0, 1], xticklabels=["Predicted Down", "Predicted Up"],
       yticks=[0, 1], yticklabels=["Actual Down", "Actual Up"],
       xlabel=f"Frozen validation-selected threshold = {selected_threshold:.3f}",
       ylabel="Observed synthetic label")
ax.set_title("SIMULATED · Untouched-test confusion matrix")
fig.colorbar(image, ax=ax, shrink=0.74, label="count")
plt.show()

display(Markdown("**Simulated metric table — same matrix, different denominators.**"))
display(metric_table)
assert cm.sum() == len(y_test)

Simulated metric table — same matrix, different denominators.

	metric	value	question	epistemic_status
0	accuracy	0.5500	fraction of all hard decisions correct	Simulated
1	precision	0.4648	reliability of positive decisions	Simulated
2	recall	0.7416	share of positives detected	Simulated
3	specificity	0.4198	share of negatives rejected	Simulated
4	f1	0.5714	harmonic precision–recall summary	Simulated
5	mcc	0.1656	association using all four cells	Simulated

Exercise 3 — denominators and policy

Recompute precision and specificity directly from the displayed matrix. Then lower the threshold by 0.05 and predict which cells must weakly increase or decrease. Explain why a higher F1 would still not establish a calibrated probability or positive expected value.

5 · ROC and precision–recall threshold sweeps

Simulated ranking evidence. ROC plots true-positive rate against false-positive rate. Precision–recall centers the positive class. The scalar reported here is stepwise average precision (AP), not an unnamed trapezoidal “PR-AUC.” Both plots show the frozen operating point; neither proves calibration or usefulness after costs.

def roc_auc_rank(target: np.ndarray, probability: np.ndarray) -> float:
    positives = probability[target == 1]
    negatives = probability[target == 0]
    comparisons = (positives[:, None] > negatives[None, :]).mean()
    ties = (positives[:, None] == negatives[None, :]).mean()
    return float(comparisons + 0.5 * ties)

def average_precision_stepwise(target: np.ndarray, probability: np.ndarray) -> float:
    order = np.argsort(-probability, kind="mergesort")
    sorted_target = target[order]
    cumulative_tp = np.cumsum(sorted_target)
    ranks = np.arange(1, len(target) + 1)
    precision_at_rank = cumulative_tp / ranks
    return float(precision_at_rank[sorted_target == 1].mean())

test_sweep = pd.DataFrame([threshold_metrics(y_test, p_test, t) for t in candidate_thresholds])
test_sweep["fpr"] = 1 - test_sweep["specificity"]
roc_auc = roc_auc_rank(y_test, p_test)
average_precision = average_precision_stepwise(y_test, p_test)
positive_prevalence = float(y_test.mean())
operating = test_sweep.iloc[(test_sweep.threshold - selected_threshold).abs().argmin()]

fig, axes = plt.subplots(1, 2, figsize=(11, 4.7), constrained_layout=True)
roc_view = test_sweep.sort_values(["fpr", "recall"])
axes[0].plot(roc_view.fpr, roc_view.recall, color=COLORS["blue"], lw=2.4)
axes[0].plot([0, 1], [0, 1], "--", color=COLORS["gray"], lw=1)
axes[0].scatter([operating.fpr], [operating.recall], s=75, color=COLORS["coral"], zorder=3, label="frozen threshold")
axes[0].set(xlabel="false-positive rate", ylabel="true-positive rate", xlim=(-0.02, 1.02), ylim=(-0.02, 1.02),
            title=f"ROC threshold sweep · rank AUC = {roc_auc:.3f}")
axes[0].legend(frameon=False)
pr_view = test_sweep.sort_values("recall")
axes[1].plot(pr_view.recall, pr_view.precision, color=COLORS["teal"], lw=2.4)
axes[1].axhline(positive_prevalence, ls="--", color=COLORS["gray"], lw=1, label=f"prevalence = {positive_prevalence:.3f}")
axes[1].scatter([operating.recall], [operating.precision], s=75, color=COLORS["coral"], zorder=3, label="frozen threshold")
axes[1].set(xlabel="recall", ylabel="precision", xlim=(-0.02, 1.02), ylim=(-0.02, 1.02),
            title=f"Precision–recall sweep · stepwise AP = {average_precision:.3f}")
axes[1].legend(frameon=False)
fig.suptitle("SIMULATED · Ranking views, not calibration or decision value", fontsize=14, fontweight="bold")
plt.show()

display(test_sweep.iloc[::30][["threshold", "recall", "fpr", "precision", "f1"]].reset_index(drop=True))

	threshold	recall	fpr	precision	f1
0	0.0500	1.0000	1.0000	0.4045	0.5761
1	0.2000	0.9101	0.7939	0.4378	0.5912
2	0.3500	0.6180	0.4046	0.5093	0.5584
3	0.5000	0.3034	0.1374	0.6000	0.4030
4	0.6500	0.0674	0.0382	0.5455	0.1200
5	0.8000	0.0000	0.0076	0.0000	NaN
6	0.9500	0.0000	0.0000	NaN	NaN

6 · Reliability, Brier score, and log loss

Simulated probability evidence. Equal-width reliability bins compare mean prediction with observed frequency and expose sample counts. Brier score and log loss assess the probability forecast jointly; a lower Brier score alone must not be described as proof of better calibration.

def calibration_table(target: np.ndarray, probability: np.ndarray, bins: int = 8) -> pd.DataFrame:
    frame = pd.DataFrame({"target": target, "probability": probability})
    frame["bin"] = pd.cut(frame.probability, bins=np.linspace(0, 1, bins + 1), include_lowest=True)
    result = frame.groupby("bin", observed=False).agg(
        count=("target", "size"), mean_prediction=("probability", "mean"), observed_frequency=("target", "mean")
    ).reset_index()
    result["calibration_gap"] = result.mean_prediction - result.observed_frequency
    return result[result["count"] > 0].reset_index(drop=True)

calibration = calibration_table(y_test, p_test)
brier = float(np.mean((p_test - y_test) ** 2))
clipped = np.clip(p_test, 1e-12, 1 - 1e-12)
logloss = float(-np.mean(y_test * np.log(clipped) + (1 - y_test) * np.log(1 - clipped)))

fig, axes = plt.subplots(1, 2, figsize=(11, 4.6), constrained_layout=True)
axes[0].plot([0, 1], [0, 1], "--", color=COLORS["gray"], label="perfect calibration")
axes[0].plot(calibration.mean_prediction, calibration.observed_frequency, "o-", color=COLORS["teal"], lw=2, ms=7)
for row in calibration.itertuples():
    axes[0].annotate(f"n={row.count}", (row.mean_prediction, row.observed_frequency), xytext=(4, 5), textcoords="offset points", fontsize=8)
axes[0].set(xlabel="mean predicted probability", ylabel="observed positive frequency", xlim=(0, 1), ylim=(0, 1),
            title="Reliability diagram")
axes[0].legend(frameon=False)
axes[1].hist(p_test[y_test == 0], bins=14, alpha=0.72, color=COLORS["blue"], label="Down labels")
axes[1].hist(p_test[y_test == 1], bins=14, alpha=0.72, color=COLORS["coral"], label="Up labels")
axes[1].set(xlabel="predicted probability", ylabel="count", title="Forecast sharpness and overlap")
axes[1].legend(frameon=False)
fig.suptitle(f"SIMULATED · Brier = {brier:.4f} · log loss = {logloss:.4f}", fontsize=14, fontweight="bold")
plt.show()

display(calibration)
display(pd.DataFrame({
    "score": ["Brier score", "log loss"], "value": [brier, logloss],
    "interpretation": ["mean squared probability error", "proper score with sharp penalty for confident errors"],
    "epistemic_status": ["Simulated", "Simulated"],
}))

	bin	count	mean_prediction	observed_frequency	calibration_gap
0	(-0.001, 0.125]	8	0.1020	0.1250	-0.0230
1	(0.125, 0.25]	45	0.1855	0.3111	-0.1256
2	(0.25, 0.375]	74	0.3158	0.3378	-0.0221
3	(0.375, 0.5]	48	0.4292	0.4583	-0.0291
4	(0.5, 0.625]	33	0.5571	0.6364	-0.0792
5	(0.625, 0.75]	9	0.6946	0.5556	0.1390
6	(0.75, 0.875]	3	0.7822	0.3333	0.4489

	score	value	interpretation	epistemic_status
0	Brier score	0.2319	mean squared probability error	Simulated
1	log loss	0.6585	proper score with sharp penalty for confident ...	Simulated

Exercise 4 — ranking versus probability

Apply any strictly increasing transformation to the test scores. Explain why rank AUC remains unchanged while Brier score, log loss, and the reliability curve may change. Which evidence would you require before using probability language on a selected high-score region?

7 · Expected-value triggers and abstention

Illustrative formula, simulated inputs. Under a teaching-only binary $1/$0 payoff, conservative per-unit edge is

[ e_c = p_{lower} - q_{exec} - c_{fees,slippage,latency}. ]

The deterministic policy acts only if data, artifact, calibration, and risk gates pass and `e_c ≥ margin`. Abstention is a policy output with a reason code—not a third settlement label and never permission for live execution.

policy = pd.DataFrame({
    "row_id": split["test"],
    "probability": p_test,
    "probability_lower": np.clip(p_test - 0.035, 0, 1),
    "q_exec": data.loc[split["test"], "reference_probability"].to_numpy(),
    "fee_allowance": 0.010,
    "slippage_allowance": 0.008 + 0.06 * data.loc[split["test"], "spread"].to_numpy(),
    "latency_allowance": 0.006 + data.loc[split["test"], "receive_delay_ms"].to_numpy() / 50_000,
    "data_valid": (split["test"] % 37 != 0),
    "artifact_match": (split["test"] % 53 != 0),
    "calibration_valid": True,
    "risk_ok": (split["test"] % 71 != 0),
    "label_up": y_test,
})
policy["total_cost_allowance"] = policy[["fee_allowance", "slippage_allowance", "latency_allowance"]].sum(axis=1)
policy["conservative_edge"] = policy.probability_lower - policy.q_exec - policy.total_cost_allowance
POLICY_MARGIN = 0.010

def policy_reason(row: pd.Series) -> str:
    if not row.data_valid:
        return "ABSTAIN_STALE_DATA"
    if not row.artifact_match:
        return "ABSTAIN_ARTIFACT_MISMATCH"
    if not row.calibration_valid:
        return "ABSTAIN_CALIBRATION_INVALID"
    if not row.risk_ok:
        return "ABSTAIN_RISK_LIMIT"
    if row.conservative_edge < POLICY_MARGIN:
        return "NO_TRADE_INSUFFICIENT_EDGE"
    return "TAKE_PAPER_INTENT"

policy["decision"] = policy.apply(policy_reason, axis=1)
policy["eligible"] = policy.decision.eq("TAKE_PAPER_INTENT")

fig, axes = plt.subplots(1, 2, figsize=(11, 4.5), constrained_layout=True)
axes[0].hist(policy.conservative_edge, bins=24, color=COLORS["teal"], alpha=0.86)
axes[0].axvspan(policy.conservative_edge.min(), POLICY_MARGIN, color=COLORS["coral"], alpha=0.12, label="no-trade / abstain region")
axes[0].axvline(POLICY_MARGIN, color=COLORS["coral"], lw=2, label=f"margin = {POLICY_MARGIN:.3f}")
axes[0].set(xlabel="conservative per-unit edge", ylabel="opportunities", title="Expected-value trigger")
axes[0].legend(frameon=False)
counts = policy.decision.value_counts().sort_values()
axes[1].barh(counts.index.str.replace("_", " "), counts.values, color=[COLORS["blue"] if "NO_TRADE" in x else COLORS["coral"] if "ABSTAIN" in x else COLORS["teal"] for x in counts.index])
axes[1].set(xlabel="decision count", title="Reason-coded outcomes")
fig.suptitle("SIMULATED · A probability cannot override a failed deterministic gate", fontsize=14, fontweight="bold")
plt.show()

display(policy[["row_id", "probability_lower", "q_exec", "total_cost_allowance", "conservative_edge", "decision"]].head(12))
print(f"Coverage (paper intents / test opportunities): {policy.eligible.mean():.3f}")
assert policy.decision.str.startswith(("ABSTAIN", "NO_TRADE", "TAKE")).all()

	row_id	probability_lower	q_exec	total_cost_allowance	conservative_edge	decision
0	740	0.3375	0.2840	0.0315	0.0220	ABSTAIN_STALE_DATA
1	741	0.5089	0.2900	0.0321	0.1868	TAKE_PAPER_INTENT
2	742	0.6224	0.2842	0.0284	0.3097	ABSTAIN_ARTIFACT_MISMATCH
3	743	0.2429	0.2736	0.0287	-0.0594	NO_TRADE_INSUFFICIENT_EDGE
4	744	0.0562	0.2600	0.0304	-0.2343	NO_TRADE_INSUFFICIENT_EDGE
5	745	0.0733	0.2645	0.0301	-0.2212	NO_TRADE_INSUFFICIENT_EDGE
6	746	0.1719	0.2698	0.0295	-0.1274	NO_TRADE_INSUFFICIENT_EDGE
7	747	0.2281	0.2656	0.0310	-0.0686	NO_TRADE_INSUFFICIENT_EDGE
8	748	0.1745	0.2579	0.0323	-0.1158	NO_TRADE_INSUFFICIENT_EDGE
9	749	0.2858	0.2627	0.0315	-0.0084	NO_TRADE_INSUFFICIENT_EDGE
10	750	0.0617	0.2576	0.0325	-0.2284	NO_TRADE_INSUFFICIENT_EDGE
11	751	0.2226	0.2588	0.0319	-0.0681	NO_TRADE_INSUFFICIENT_EDGE

Coverage (paper intents / test opportunities): 0.436

Exercise 5 — abstention is not free

Increase the latency allowance and recompute coverage. Report conditional accuracy together with coverage; explain why conditional accuracy can rise merely because the policy acts less often. Add one unknown predicate and specify the only fail-closed interpretation.

8 · Illustrative self-supervised masked-sequence objective

Illustrative — no encoder is trained. A self-supervised objective derives a target from permitted inputs rather than the later settlement label. Here a causal prefix rule reconstructs selected masked values from the two most recent visible values. Low reconstruction loss would not establish directional skill, calibration, or replay value.

sequence_t = np.arange(24)
sequence = np.sin(sequence_t / 3.0) + 0.03 * sequence_t
masked_positions = np.array([5, 10, 15, 20])
visible = sequence.copy()
visible[masked_positions] = np.nan
reconstruction = visible.copy()
for position in masked_positions:
    causal_history = reconstruction[:position][~np.isnan(reconstruction[:position])]
    reconstruction[position] = causal_history[-2:].mean()
masked_mse = float(np.mean((reconstruction[masked_positions] - sequence[masked_positions]) ** 2))

fig, ax = plt.subplots(figsize=(10.8, 4.0), constrained_layout=True)
ax.plot(sequence_t, sequence, color=COLORS["blue"], lw=2, label="permitted sequence target")
ax.scatter(masked_positions, sequence[masked_positions], s=90, facecolor="white", edgecolor=COLORS["coral"], lw=2.5, label="masked target")
ax.scatter(masked_positions, reconstruction[masked_positions], s=75, marker="x", color=COLORS["teal"], lw=2.5, label="causal-prefix reconstruction")
for position in masked_positions:
    ax.axvline(position, color=COLORS["gray"], lw=0.7, alpha=0.25)
ax.set(xlabel="sequence position", ylabel="illustrative scalar attribute",
       title=f"ILLUSTRATIVE · Masked-sequence objective · masked MSE = {masked_mse:.4f}")
ax.legend(frameon=False, ncol=3)
plt.show()

display(pd.DataFrame({
    "position": masked_positions,
    "target": sequence[masked_positions],
    "reconstruction": reconstruction[masked_positions],
    "squared_error": (reconstruction[masked_positions] - sequence[masked_positions]) ** 2,
    "epistemic_status": "Illustrative",
}))

	position	target	reconstruction	squared_error	epistemic_status
0	5	1.1454	1.0117	0.0179	Illustrative
1	10	0.1094	0.5542	0.1978	Illustrative
2	15	-0.5089	-0.5590	0.0025	Illustrative
3	20	0.9742	0.4404	0.2849	Illustrative

9 · Causal attention mask and tensor calculation

Illustrative tensor arithmetic. For one attention head, `Q=XW_Q`, `K=XW_K`, `V=XW_V`, logits are `QKᵀ/√d`, and an upper-triangular `−∞` mask removes future keys before row-wise softmax. A mask constrains only presented tokens; it cannot repair preprocessing leakage.

X = np.array([
    [1.0, 0.0, 0.5],
    [0.8, 0.4, 0.1],
    [0.2, 1.0, 0.3],
    [0.5, 0.3, 1.0],
])
W_Q = np.array([[0.7, -0.2], [0.1, 0.8], [0.5, 0.3]])
W_K = np.array([[0.6, 0.1], [-0.3, 0.7], [0.4, 0.5]])
W_V = np.array([[0.5, 0.2], [0.2, 0.9], [0.8, -0.1]])

def causal_attention(tokens: np.ndarray) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    q, k, v = tokens @ W_Q, tokens @ W_K, tokens @ W_V
    logits = q @ k.T / np.sqrt(q.shape[1])
    permitted = np.tril(np.ones_like(logits, dtype=bool))
    masked_logits = np.where(permitted, logits, -np.inf)
    row_max = np.max(masked_logits, axis=1, keepdims=True)
    exp_logits = np.exp(masked_logits - row_max)
    weights = exp_logits / exp_logits.sum(axis=1, keepdims=True)
    return weights, weights @ v, permitted

attention_weights, attention_output, causal_mask = causal_attention(X)
X_future_changed = X.copy()
X_future_changed[3] = np.array([99.0, -50.0, 27.0])
changed_weights, changed_output, _ = causal_attention(X_future_changed)
future_perturbation_delta = float(np.max(np.abs(attention_output[:3] - changed_output[:3])))
assert future_perturbation_delta < 1e-12

fig, axes = plt.subplots(1, 2, figsize=(10.5, 4.2), constrained_layout=True)
axes[0].imshow(causal_mask, cmap="Greens", vmin=0, vmax=1)
axes[0].set(title="Permission mask", xlabel="key position", ylabel="query position", xticks=range(4), yticks=range(4))
for i in range(4):
    for j in range(4):
        axes[0].text(j, i, "allow" if causal_mask[i, j] else "block", ha="center", va="center", fontsize=8,
                     color="white" if causal_mask[i, j] else COLORS["ink"])
heat = axes[1].imshow(attention_weights, cmap="YlGnBu", vmin=0, vmax=1)
axes[1].set(title="Row-wise causal attention weights", xlabel="key position", ylabel="query position", xticks=range(4), yticks=range(4))
for i in range(4):
    for j in range(4):
        axes[1].text(j, i, f"{attention_weights[i,j]:.2f}", ha="center", va="center", fontsize=8)
fig.colorbar(heat, ax=axes[1], shrink=0.78)
fig.suptitle("ILLUSTRATIVE · Architectural masking ≠ causal-effect identification", fontsize=13.5, fontweight="bold")
plt.show()

display(Markdown(f"**Implemented toy invariant.** Changing only token 4 changed outputs at positions 1–3 by \`{future_perturbation_delta:.2e}\`."))
display(pd.DataFrame(attention_output, columns=["output_dim_0", "output_dim_1"]).rename_axis("query_position"))

Implemented toy invariant. Changing only token 4 changed outputs at positions 1–3 by `0.00e+00`.

	output_dim_0	output_dim_1
query_position
0	0.9000	0.1500
1	0.7449	0.3142
2	0.6659	0.5337
3	0.8124	0.4164

Exercise 6 — test time itself

Modify a future raw event before tokenization while keeping the eligible prefix fixed. Which components must rerun for an end-to-end perturbation test? Explain why testing only the triangular matrix cannot detect full-period normalization leakage.

10 · Cost-aware paper replay

Simulated — not a backtest or live fill claim. This teaching replay consumes synthetic ask depth only after a declared latency haircut, applies a 1% illustrative fee, preserves partial and no fills, and records every no-trade decision. A binary contract pays one unit per filled unit only when the later synthetic label is positive.

@dataclass(frozen=True)
class ReplayAssumptions:
    requested_units: float = 80.0
    fee_rate: float = 0.01
    latency_depth_haircut: float = 10.0

REPLAY = ReplayAssumptions()

def consume_asks(levels: list[tuple[float, float]], requested: float) -> tuple[float, float, list[tuple[float, float]]]:
    remaining = requested
    gross = 0.0
    fills: list[tuple[float, float]] = []
    for price, available in levels:
        quantity = min(max(available, 0.0), remaining)
        if quantity > 0:
            fills.append((price, quantity))
            gross += price * quantity
            remaining -= quantity
        if remaining <= 1e-12:
            break
    return requested - remaining, gross, fills

ledger_rows = []
take_counter = 0
for row in policy.itertuples(index=False):
    if row.decision != "TAKE_PAPER_INTENT":
        ledger_rows.append({
            "decision_id": f"SQ-{int(row.row_id):04d}", "policy_decision": row.decision,
            "execution_status": "NO_TRADE", "requested_units": 0.0, "filled_units": 0.0,
            "unfilled_units": 0.0, "gross_cost": 0.0, "fee": 0.0, "net_result": 0.0,
            "latency_ms": 125, "fill_detail": [], "epistemic_status": "Simulated",
        })
        continue
    take_counter += 1
    ask = float(row.q_exec + 0.008)
    if take_counter == 1:
        raw_depths = [0.0, 0.0, 0.0]
    elif take_counter == 2:
        raw_depths = [20.0, 15.0, 0.0]
    else:
        raw_depths = [40.0, 35.0, 30.0]
    post_latency_depths = raw_depths.copy()
    post_latency_depths[0] = max(0.0, post_latency_depths[0] - REPLAY.latency_depth_haircut)
    levels = [(ask, post_latency_depths[0]), (ask + 0.018, post_latency_depths[1]), (ask + 0.038, post_latency_depths[2])]
    filled, gross, fills = consume_asks(levels, REPLAY.requested_units)
    fee = REPLAY.fee_rate * gross
    unfilled = REPLAY.requested_units - filled
    if filled == 0:
        status = "NO_FILL"
    elif unfilled > 1e-12:
        status = "PARTIAL_FILL"
    else:
        status = "FULL_FILL"
    settlement = float(row.label_up) * filled
    ledger_rows.append({
        "decision_id": f"SQ-{int(row.row_id):04d}", "policy_decision": row.decision,
        "execution_status": status, "requested_units": REPLAY.requested_units,
        "filled_units": filled, "unfilled_units": unfilled, "gross_cost": gross,
        "fee": fee, "net_result": settlement - gross - fee, "latency_ms": 125,
        "fill_detail": fills, "epistemic_status": "Simulated",
    })

replay_ledger = pd.DataFrame(ledger_rows)
status_order = ["NO_TRADE", "NO_FILL", "PARTIAL_FILL", "FULL_FILL"]
status_counts = replay_ledger.execution_status.value_counts().reindex(status_order, fill_value=0)
assert (status_counts > 0).all(), status_counts.to_dict()

fig, axes = plt.subplots(1, 2, figsize=(11, 4.6), constrained_layout=True)
axes[0].bar(status_counts.index.str.replace("_", " "), status_counts.values,
            color=[COLORS["gray"], COLORS["coral"], COLORS["gold"], COLORS["teal"]])
axes[0].tick_params(axis="x", rotation=20)
axes[0].set(ylabel="decision records", title="Replay outcome states")
axes[1].plot(replay_ledger.net_result.cumsum().to_numpy(), color=COLORS["blue"], lw=2)
axes[1].axhline(0, color=COLORS["ink"], lw=0.8)
axes[1].set(xlabel="ordered replay record", ylabel="cumulative simulated units", title="Conditional result under declared assumptions")
fig.suptitle("SIMULATED · Fees + latency + partial/no fills + no-trade retained", fontsize=14, fontweight="bold")
plt.show()

display(replay_ledger[replay_ledger.execution_status.ne("NO_TRADE")].head(8))
display(pd.DataFrame({
    "assumption": ["requested units", "fee rate", "latency", "latency first-level depth haircut"],
    "value": [REPLAY.requested_units, REPLAY.fee_rate, 125, REPLAY.latency_depth_haircut],
    "status": ["Illustrative"] * 4,
}))

	decision_id	policy_decision	execution_status	requested_units	filled_units	unfilled_units	gross_cost	fee	net_result	latency_ms	fill_detail	epistemic_status
1	SQ-0741	TAKE_PAPER_INTENT	NO_FILL	80.0000	0.0000	80.0000	0.0000	0.0000	0.0000	125	[]	Simulated
15	SQ-0755	TAKE_PAPER_INTENT	PARTIAL_FILL	80.0000	25.0000	55.0000	6.9917	0.0699	-7.0616	125	[(0.2688674275036334, 10.0), (0.28686742750363...	Simulated
16	SQ-0756	TAKE_PAPER_INTENT	FULL_FILL	80.0000	80.0000	0.0000	23.4508	0.2345	56.3147	125	[(0.2781354658390098, 30.0), (0.29613546583900...	Simulated
17	SQ-0757	TAKE_PAPER_INTENT	FULL_FILL	80.0000	80.0000	0.0000	23.3855	0.2339	-23.6193	125	[(0.2773181481673054, 30.0), (0.29531814816730...	Simulated
20	SQ-0760	TAKE_PAPER_INTENT	FULL_FILL	80.0000	80.0000	0.0000	23.4648	0.2346	-23.6995	125	[(0.2783105971962925, 30.0), (0.29631059719629...	Simulated
24	SQ-0764	TAKE_PAPER_INTENT	FULL_FILL	80.0000	80.0000	0.0000	23.3726	0.2337	-23.6064	125	[(0.2771578144620177, 30.0), (0.29515781446201...	Simulated
33	SQ-0773	TAKE_PAPER_INTENT	FULL_FILL	80.0000	80.0000	0.0000	24.3732	0.2437	55.3830	125	[(0.28966547684895655, 30.0), (0.3076654768489...	Simulated
34	SQ-0774	TAKE_PAPER_INTENT	FULL_FILL	80.0000	80.0000	0.0000	24.8426	0.2484	54.9090	125	[(0.2955323192124258, 30.0), (0.31353231921242...	Simulated

	assumption	value	status
0	requested units	80.0000	Illustrative
1	fee rate	0.0100	Illustrative
2	latency	125.0000	Illustrative
3	latency first-level depth haircut	10.0000	Illustrative

Exercise 7 — replay skepticism

Double the latency depth haircut and fee rate. Which records change from full to partial or no fill? Explain why a static midpoint table could not reproduce this transition. Identify one missing real-world execution variable that prevents this teaching replay from supporting a live-fill claim.

11 · Fail-closed guardian and state machine

Implemented toy control; simulated health inputs. The guardian may classify evidence, pause research output, quarantine a batch, record, and escalate. It has no credential, network, model-approval, policy-change, or order-control capability. Unknown evidence routes to escalation.

class GuardianState(str, Enum):
    OBSERVE = "OBSERVE"
    VERIFY = "VERIFY"
    HEALTHY = "HEALTHY"
    PAUSED = "PAUSED_RESEARCH"
    QUARANTINED = "QUARANTINED_BATCH"
    VERIFY_CONTAINMENT = "VERIFY_CONTAINMENT"
    RECORDED = "RECORDED_AUDIT"
    ESCALATED = "ESCALATED_HUMAN_REVIEW"

ALLOWED_CAPABILITIES = {
    "read_health_metadata", "pause_research_report", "quarantine_batch",
    "record_audit_event", "request_human_review",
}
PROHIBITED_CAPABILITIES = {
    "place_order", "cancel_order", "resize_order", "approve_model",
    "change_threshold", "modify_evidence",
}
assert ALLOWED_CAPABILITIES.isdisjoint(PROHIBITED_CAPABILITIES)

def guardian_check(*, freshness_ok: bool | None, schema_ok: bool | None,
                   artifact_ok: bool | None, replay_ok: bool | None) -> dict[str, str]:
    evidence = {
        "freshness_ok": freshness_ok, "schema_ok": schema_ok,
        "artifact_ok": artifact_ok, "replay_ok": replay_ok,
    }
    if any(value is None for value in evidence.values()):
        return {"state": GuardianState.ESCALATED.value, "action": "request_human_review", "reason": "INCOMPLETE_EVIDENCE"}
    if not schema_ok:
        return {"state": GuardianState.QUARANTINED.value, "action": "quarantine_batch", "reason": "SCHEMA_CONTRACT_FAILED"}
    if not freshness_ok or not artifact_ok or not replay_ok:
        failed = ",".join(key for key, value in evidence.items() if not value)
        return {"state": GuardianState.PAUSED.value, "action": "pause_research_report", "reason": failed.upper()}
    return {"state": GuardianState.HEALTHY.value, "action": "record_audit_event", "reason": "ALL_PREDICATES_VERIFIED"}

scenarios = pd.DataFrame([
    {"scenario": "healthy fixture", **guardian_check(freshness_ok=True, schema_ok=True, artifact_ok=True, replay_ok=True)},
    {"scenario": "stale source", **guardian_check(freshness_ok=False, schema_ok=True, artifact_ok=True, replay_ok=True)},
    {"scenario": "schema drift", **guardian_check(freshness_ok=True, schema_ok=False, artifact_ok=True, replay_ok=True)},
    {"scenario": "unknown predicate", **guardian_check(freshness_ok=None, schema_ok=True, artifact_ok=True, replay_ok=True)},
])
assert scenarios.loc[scenarios.scenario.eq("unknown predicate"), "state"].item() == GuardianState.ESCALATED.value
assert not (PROHIBITED_CAPABILITIES & ALLOWED_CAPABILITIES)

def verify_containment(action: str, postcondition_verified: bool | None) -> dict[str, str]:
    if action not in {"pause_research_report", "quarantine_batch"}:
        return {"state": GuardianState.ESCALATED.value, "reason": "ACTION_NOT_ALLOWLISTED"}
    if postcondition_verified is not True:
        return {"state": GuardianState.ESCALATED.value, "reason": "CONTAINMENT_UNVERIFIED"}
    return {"state": GuardianState.RECORDED.value, "reason": "CONTAINMENT_VERIFIED_AND_RECORDED"}

containment_checks = pd.DataFrame([
    {"case": "verified pause", **verify_containment("pause_research_report", True)},
    {"case": "unknown quarantine outcome", **verify_containment("quarantine_batch", None)},
])
assert containment_checks.loc[containment_checks.case.eq("unknown quarantine outcome"), "state"].item() == GuardianState.ESCALATED.value

ALLOWED_TRANSITIONS = {
    (GuardianState.OBSERVE, GuardianState.VERIFY),
    (GuardianState.VERIFY, GuardianState.HEALTHY),
    (GuardianState.VERIFY, GuardianState.PAUSED),
    (GuardianState.VERIFY, GuardianState.QUARANTINED),
    (GuardianState.VERIFY, GuardianState.ESCALATED),
    (GuardianState.HEALTHY, GuardianState.RECORDED),
    (GuardianState.PAUSED, GuardianState.VERIFY_CONTAINMENT),
    (GuardianState.QUARANTINED, GuardianState.VERIFY_CONTAINMENT),
    (GuardianState.VERIFY_CONTAINMENT, GuardianState.RECORDED),
    (GuardianState.VERIFY_CONTAINMENT, GuardianState.ESCALATED),
    (GuardianState.RECORDED, GuardianState.ESCALATED),
}

def guardian_runbook(*, scenario: str, freshness_ok: bool | None, schema_ok: bool | None,
                     artifact_ok: bool | None, replay_ok: bool | None,
                     containment_verified: bool | None = True) -> list[dict[str, str]]:
    decision = guardian_check(freshness_ok=freshness_ok, schema_ok=schema_ok,
                              artifact_ok=artifact_ok, replay_ok=replay_ok)
    states = [GuardianState.OBSERVE, GuardianState.VERIFY, GuardianState(decision["state"])]
    if states[-1] is GuardianState.HEALTHY:
        states.append(GuardianState.RECORDED)
    elif states[-1] in {GuardianState.PAUSED, GuardianState.QUARANTINED}:
        states.append(GuardianState.VERIFY_CONTAINMENT)
        verification = verify_containment(decision["action"], containment_verified)
        states.append(GuardianState(verification["state"]))
        if states[-1] is GuardianState.RECORDED:
            states.append(GuardianState.ESCALATED)
    ledger = []
    for sequence, state in enumerate(states):
        if sequence:
            assert (states[sequence - 1], state) in ALLOWED_TRANSITIONS
        ledger.append({"scenario": scenario, "sequence": sequence, "state": state.value,
                       "audit_event": f"{scenario}:{sequence}:{state.value}"})
    return ledger

transition_ledger = pd.DataFrame(
    guardian_runbook(scenario="healthy path", freshness_ok=True, schema_ok=True,
                     artifact_ok=True, replay_ok=True)
    + guardian_runbook(scenario="verified pause", freshness_ok=False, schema_ok=True,
                       artifact_ok=True, replay_ok=True, containment_verified=True)
    + guardian_runbook(scenario="failed verification", freshness_ok=False, schema_ok=True,
                       artifact_ok=True, replay_ok=True, containment_verified=False)
)
healthy_path = transition_ledger.loc[transition_ledger.scenario.eq("healthy path"), "state"].tolist()
verified_path = transition_ledger.loc[transition_ledger.scenario.eq("verified pause"), "state"].tolist()
failed_path = transition_ledger.loc[transition_ledger.scenario.eq("failed verification"), "state"].tolist()
assert healthy_path == ["OBSERVE", "VERIFY", "HEALTHY", "RECORDED_AUDIT"]
assert verified_path == ["OBSERVE", "VERIFY", "PAUSED_RESEARCH", "VERIFY_CONTAINMENT",
                         "RECORDED_AUDIT", "ESCALATED_HUMAN_REVIEW"]
assert failed_path[-2:] == ["VERIFY_CONTAINMENT", "ESCALATED_HUMAN_REVIEW"]

fig, ax = plt.subplots(figsize=(11, 4.8), constrained_layout=True)
ax.axis("off")
positions = {
    "OBSERVE": (0.06, 0.55), "VERIFY": (0.23, 0.55), "HEALTHY": (0.43, 0.82),
    "PAUSED": (0.43, 0.57), "QUARANTINED": (0.43, 0.30),
    "VERIFY_CONTAINMENT": (0.65, 0.43), "RECORDED": (0.82, 0.66), "ESCALATED": (0.92, 0.32),
}
labels = {
    "OBSERVE": "OBSERVE\nread-only", "VERIFY": "VERIFY\npredicates", "HEALTHY": "RECORD\nhealthy",
    "PAUSED": "PAUSE\nresearch", "QUARANTINED": "QUARANTINE\nbatch",
    "VERIFY_CONTAINMENT": "VERIFY\ncontainment", "RECORDED": "RECORD\naudit",
    "ESCALATED": "ESCALATE\nhuman review",
}
node_colors = {"OBSERVE": COLORS["blue"], "VERIFY": COLORS["gold"], "HEALTHY": COLORS["teal"],
               "PAUSED": COLORS["coral"], "QUARANTINED": COLORS["coral"],
               "VERIFY_CONTAINMENT": COLORS["gold"], "RECORDED": COLORS["teal"], "ESCALATED": COLORS["ink"]}
for key, (x, y_pos) in positions.items():
    ax.text(x, y_pos, labels[key], ha="center", va="center", color="white" if key not in {"VERIFY", "VERIFY_CONTAINMENT"} else COLORS["ink"],
            fontweight="bold", bbox={"boxstyle": "round,pad=0.8", "fc": node_colors[key], "ec": COLORS["ink"], "lw": 1.5})
def arrow(a: str, b: str, label: str):
    ax.annotate("", xy=positions[b], xytext=positions[a], arrowprops={"arrowstyle": "->", "lw": 1.7, "color": COLORS["ink"], "shrinkA": 52, "shrinkB": 52})
    mid = ((positions[a][0] + positions[b][0]) / 2, (positions[a][1] + positions[b][1]) / 2)
    ax.text(mid[0], mid[1] + 0.035, label, ha="center", fontsize=8, color=COLORS["ink"])
arrow("OBSERVE", "VERIFY", "evidence")
arrow("VERIFY", "HEALTHY", "all pass")
arrow("VERIFY", "PAUSED", "stale / mismatch")
arrow("VERIFY", "QUARANTINED", "schema fail")
arrow("PAUSED", "VERIFY_CONTAINMENT", "contain")
arrow("QUARANTINED", "VERIFY_CONTAINMENT", "contain")
arrow("VERIFY_CONTAINMENT", "RECORDED", "verified")
arrow("VERIFY_CONTAINMENT", "ESCALATED", "unknown / failed")
arrow("RECORDED", "ESCALATED", "request review")
ax.set_title("DESIGN-TARGET BOUNDARY · Fail closed; no order authority", fontsize=14, fontweight="bold")
plt.show()

display(scenarios)
display(containment_checks)
display(transition_ledger)
display(pd.DataFrame({
    "allowed_capability": sorted(ALLOWED_CAPABILITIES),
}).style.set_caption("Guardian allowlist — order control is structurally absent"))

	scenario	state	action	reason
0	healthy fixture	HEALTHY	record_audit_event	ALL_PREDICATES_VERIFIED
1	stale source	PAUSED_RESEARCH	pause_research_report	FRESHNESS_OK
2	schema drift	QUARANTINED_BATCH	quarantine_batch	SCHEMA_CONTRACT_FAILED
3	unknown predicate	ESCALATED_HUMAN_REVIEW	request_human_review	INCOMPLETE_EVIDENCE

	case	state	reason
0	verified pause	RECORDED_AUDIT	CONTAINMENT_VERIFIED_AND_RECORDED
1	unknown quarantine outcome	ESCALATED_HUMAN_REVIEW	CONTAINMENT_UNVERIFIED

	scenario	sequence	state	audit_event
0	healthy path	0	OBSERVE	healthy path:0:OBSERVE
1	healthy path	1	VERIFY	healthy path:1:VERIFY
2	healthy path	2	HEALTHY	healthy path:2:HEALTHY
3	healthy path	3	RECORDED_AUDIT	healthy path:3:RECORDED_AUDIT
4	verified pause	0	OBSERVE	verified pause:0:OBSERVE
5	verified pause	1	VERIFY	verified pause:1:VERIFY
6	verified pause	2	PAUSED_RESEARCH	verified pause:2:PAUSED_RESEARCH
7	verified pause	3	VERIFY_CONTAINMENT	verified pause:3:VERIFY_CONTAINMENT
8	verified pause	4	RECORDED_AUDIT	verified pause:4:RECORDED_AUDIT
9	verified pause	5	ESCALATED_HUMAN_REVIEW	verified pause:5:ESCALATED_HUMAN_REVIEW
10	failed verification	0	OBSERVE	failed verification:0:OBSERVE
11	failed verification	1	VERIFY	failed verification:1:VERIFY
12	failed verification	2	PAUSED_RESEARCH	failed verification:2:PAUSED_RESEARCH
13	failed verification	3	VERIFY_CONTAINMENT	failed verification:3:VERIFY_CONTAINMENT
14	failed verification	4	ESCALATED_HUMAN_REVIEW	failed verification:4:ESCALATED_HUMAN_REVIEW

Guardian allowlist — order control is structurally absent

	allowed_capability
0	pause_research_report
1	quarantine_batch
2	read_health_metadata
3	record_audit_event
4	request_human_review

12 · Optional local LLM draft path — disabled by default

Design target; no generated-code execution. The next cell can request a text draft from a local Ollama endpoint using a Kimi model name only after a human deliberately changes the flag. The response remains an untrusted string for diff review. It is never passed to `exec`, `eval`, a shell, a file writer, or a tool. Network use is disabled during normal execution.

ENABLE_LOCAL_OLLAMA_KIMI_DRAFT = False
llm_draft_text = None

if ENABLE_LOCAL_OLLAMA_KIMI_DRAFT:
    import json
    import urllib.request

    request_body = json.dumps({
        "model": "kimi",
        "prompt": "Draft one pure Python metric helper. Return text only; do not execute anything.",
        "stream": False,
    }).encode("utf-8")
    request = urllib.request.Request(
        "http://127.0.0.1:11434/api/generate", data=request_body,
        headers={"Content-Type": "application/json"}, method="POST",
    )
    with urllib.request.urlopen(request, timeout=10) as response:
        llm_draft_text = json.loads(response.read().decode("utf-8")).get("response", "")
    print("Draft captured as untrusted text for human review; execution remains prohibited.")
else:
    print("DISABLED · No Ollama/Kimi request made; no generated code obtained or executed.")

assert llm_draft_text is None, "Normal deterministic execution must keep the optional draft path disabled."

DISABLED · No Ollama/Kimi request made; no generated code obtained or executed.

Exercise 8 — authority audit

Add a hypothetical guardian request to “repair the schema automatically.” Explain why this is remediation rather than containment. Identify the missing human approval, tests, rollback, and capability boundary. Then threat-model the disabled LLM cell: what changes would make it unsafe?

13 · Executable evidence gate

The final cell checks narrow invariants. Passing them means only that this notebook's deterministic teaching contracts executed in this environment. It does not promote any result to measured evidence or authorize production use.

audit = pd.DataFrame([
    ("portable locked runtime", runtime_matches, f"Python {platform.python_version()}"),
    ("chronological non-overlap", max(split["train"]) < min(split["validation"]) < min(split["test"]), "ordered"),
    ("purge derived from contract", len(split["purge"]) == GAP_ROWS, f"{GAP_ROWS} rows"),
    ("embargo derived from contract", len(split["embargo"]) == GAP_ROWS, f"{GAP_ROWS} rows"),
    ("raw support intervals disjoint", bool(boundary_support_audit["disjoint"].all()),
     boundary_support_audit.to_dict("records")),
    ("confusion counts complete", int(cm.sum()) == len(y_test), f"{int(cm.sum())}/{len(y_test)}"),
    ("probabilities finite", bool(np.isfinite(p_test).all()), "finite"),
    ("causal future perturbation", future_perturbation_delta < 1e-12, f"{future_perturbation_delta:.2e}"),
    ("replay preserves four states", bool((status_counts > 0).all()), ",".join(status_counts.index)),
    ("guardian fails closed", scenarios.loc[scenarios.scenario.eq("unknown predicate"), "state"].item() == GuardianState.ESCALATED.value, "escalated"),
    ("guardian transition ledger complete", healthy_path == ["OBSERVE", "VERIFY", "HEALTHY", "RECORDED_AUDIT"] and verified_path[-3:] == ["VERIFY_CONTAINMENT", "RECORDED_AUDIT", "ESCALATED_HUMAN_REVIEW"] and failed_path[-1] == "ESCALATED_HUMAN_REVIEW", "healthy, verified-containment, and failed-containment paths recorded"),
    ("no order capability", ALLOWED_CAPABILITIES.isdisjoint(PROHIBITED_CAPABILITIES), "structurally absent"),
    ("optional LLM disabled", ENABLE_LOCAL_OLLAMA_KIMI_DRAFT is False and llm_draft_text is None, "no request / no execution"),
    ("scikit-learn GradientBoosting path", SKLEARN_OK, model_status),
], columns=["check", "passed", "evidence"])
audit["epistemic_status"] = "Implemented notebook check"
display(audit)
critical_without_environment = audit[~audit["check"].eq("scikit-learn GradientBoosting path")]
assert critical_without_environment.passed.all(), audit[~audit.passed].to_dict("records")

if SKLEARN_OK:
    display(Markdown("### Notebook execution gate: **GO for synthetic teaching use**\n\nAll requested executable paths ran. This remains research-only and non-Measured."))
else:
    display(Markdown("### Notebook execution gate: **NO-GO for final delivery**\n\nThe notebook executed its fail-safe teaching surrogate, but the required scikit-learn GradientBoosting path could not run in the exact interpreter. Fix the environment incompatibility, then re-execute without changing notebook source."))

	check	passed	evidence	epistemic_status
0	portable locked runtime	True	Python 3.11.15	Implemented notebook check
1	chronological non-overlap	True	ordered	Implemented notebook check
2	purge derived from contract	True	20 rows	Implemented notebook check
3	embargo derived from contract	True	20 rows	Implemented notebook check
4	raw support intervals disjoint	True	[{'boundary': 'train→validation', 'left_availa...	Implemented notebook check
5	confusion counts complete	True	220/220	Implemented notebook check
6	probabilities finite	True	finite	Implemented notebook check
7	causal future perturbation	True	0.00e+00	Implemented notebook check
8	replay preserves four states	True	NO_TRADE,NO_FILL,PARTIAL_FILL,FULL_FILL	Implemented notebook check
9	guardian fails closed	True	escalated	Implemented notebook check
10	guardian transition ledger complete	True	healthy, verified-containment, and failed-cont...	Implemented notebook check
11	no order capability	True	structurally absent	Implemented notebook check
12	optional LLM disabled	True	no request / no execution	Implemented notebook check
13	scikit-learn GradientBoosting path	True	Implemented teaching fallback	Implemented notebook check

Notebook execution gate: GO for synthetic teaching use

All requested executable paths ran. This remains research-only and non-Measured.

Closing defense

A threshold is not a probability. A probability is not expected value. Expected value is not a fill. A simulated fill is not live execution. A monitoring agent is not an RL policy, and a prompt is not a capability boundary.

The defensible artifact is the trace: declared evidence → frozen procedure → probability assessment → deterministic policy → replay assumptions → reason-coded action or refusal → bounded guardian response.

Current claim boundary: synthetic teaching evidence only. No verified historical corpus, CatBoost model, sequence encoder, TLOB/LiT model, live replay, measured latency, market performance, or trading authority exists in this notebook.