"""Retrospective active-learning benchmark.
The flagship validation for a next-experiment engine on literature data:
hide the dataset, seed the campaign with a few random experiments, and let
DENIA sequentially "run" experiments by revealing rows from the hidden pool.
The figure of merit is the best target value found as a function of the number
of experiments, compared against random selection - i.e., how much
experimental budget the decision layer saves.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import List, Optional
import numpy as np
import pandas as pd
from .campaign import Campaign
from .schema import CampaignSchema
[docs]
@dataclass
class RetrospectiveResult:
n_experiments: np.ndarray
best_denia: np.ndarray # mean over seeds
best_denia_std: np.ndarray
best_random: np.ndarray
best_random_std: np.ndarray
oracle: float
target: str
curves_denia: List[np.ndarray] = field(default_factory=list)
curves_random: List[np.ndarray] = field(default_factory=list)
[docs]
def savings_at(self, fraction_of_oracle: float = 0.9) -> dict:
"""Experiments needed to reach a fraction of the oracle optimum."""
thresh = fraction_of_oracle * self.oracle
def first_reach(curve):
hit = np.argmax(curve >= thresh)
return int(self.n_experiments[hit]) if curve.max() >= thresh else None
return {
"threshold": thresh,
"denia": first_reach(self.best_denia),
"random": first_reach(self.best_random),
}
def run_retrospective(
df: pd.DataFrame,
schema: CampaignSchema,
target: Optional[str] = None,
n_init: int = 20,
n_iterations: int = 30,
batch_size: int = 5,
n_seeds: int = 5,
pool_size: Optional[int] = None,
surrogate: str = "auto",
) -> RetrospectiveResult:
target = target or schema.targets[0]
y_all = pd.to_numeric(df[target], errors="coerce")
df = df.loc[y_all.notna()].reset_index(drop=True)
if pool_size and len(df) > pool_size:
df = df.sample(pool_size, random_state=0).reset_index(drop=True)
y = pd.to_numeric(df[target], errors="coerce").values
oracle = float(np.nanmax(y))
steps = np.arange(n_init, n_init + n_iterations * batch_size + 1, batch_size)
curves_d, curves_r = [], []
for seed in range(n_seeds):
rng = np.random.default_rng(seed)
init_idx = rng.choice(len(df), size=n_init, replace=False)
# ---- DENIA loop
known = list(init_idx)
best_curve = [y[known].max()]
for _ in range(n_iterations):
pool_idx = np.setdiff1d(np.arange(len(df)), known)
if len(pool_idx) == 0:
best_curve.append(best_curve[-1])
continue
camp = Campaign(df.iloc[known], schema, target=target,
surrogate=surrogate, random_state=seed)
rec = camp.recommend(df.iloc[pool_idx], batch_size=batch_size)
known += list(rec.index)
best_curve.append(y[known].max())
curves_d.append(np.array(best_curve))
# ---- random baseline
known_r = list(init_idx)
best_r = [y[known_r].max()]
for _ in range(n_iterations):
pool_idx = np.setdiff1d(np.arange(len(df)), known_r)
pick = rng.choice(pool_idx, size=min(batch_size, len(pool_idx)), replace=False)
known_r += list(pick)
best_r.append(y[known_r].max())
curves_r.append(np.array(best_r))
D = np.stack(curves_d)
R = np.stack(curves_r)
return RetrospectiveResult(
n_experiments=steps,
best_denia=D.mean(0), best_denia_std=D.std(0),
best_random=R.mean(0), best_random_std=R.std(0),
oracle=oracle, target=target,
curves_denia=curves_d, curves_random=curves_r,
)
[docs]
def plot_retrospective(result: RetrospectiveResult, path: Optional[str] = None):
"""Discovery curve: best-found target vs experimental budget."""
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
fig, ax = plt.subplots(figsize=(7, 4.5), dpi=150)
x = result.n_experiments
ax.plot(x, result.best_denia, lw=2.2, color="#0B6E4F", label="DENIA (active learning)")
ax.fill_between(x, result.best_denia - result.best_denia_std,
result.best_denia + result.best_denia_std, color="#0B6E4F", alpha=0.18)
ax.plot(x, result.best_random, lw=2.2, color="#8D8D8D", ls="--", label="Random selection")
ax.fill_between(x, result.best_random - result.best_random_std,
result.best_random + result.best_random_std, color="#8D8D8D", alpha=0.18)
ax.axhline(result.oracle, color="#B3402A", lw=1.2, ls=":", label="Dataset optimum (oracle)")
ax.set_xlabel("Experiments performed")
ax.set_ylabel(f"Best {result.target} found")
ax.set_title("Retrospective discovery benchmark")
ax.legend(frameon=False)
ax.spines[["top", "right"]].set_visible(False)
fig.tight_layout()
if path:
fig.savefig(path)
return fig