避免信息泄漏地训练和评估分类器

from pathlib import Path
import base64, gzip, hashlib, json, platform
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from IPython.display import display, HTML
from IPython.display import display as _display
def display(value):
    if isinstance(value,pd.DataFrame):
        table=value.to_html(max_rows=12,escape=True)
        style="<style>.vibeit-readable-table{max-width:100%;overflow-x:auto}.vibeit-readable-table table{width:max-content;max-width:none;border-collapse:collapse}.vibeit-readable-table td,.vibeit-readable-table th{white-space:nowrap!important;word-break:normal!important;overflow-wrap:normal!important}</style>"
        return _display(HTML(style+'<div class="vibeit-readable-table">'+table+'</div>'))
    return _display(value)
RECIPE_ID='ds03-iris-knn'
OUTPUT_DIR=Path.cwd()/"results"/RECIPE_ID
plt.rcParams.update({"figure.dpi":120,"font.size":10,"axes.spines.top":False,"axes.spines.right":False,
                      "axes.prop_cycle":plt.cycler(color=["#18785f","#416aa6","#bc6541","#9974a5"])})
print("Python:",platform.python_version(),"; NumPy:",np.__version__,"; Pandas:",pd.__version__)
print("Working folder:",Path.cwd())
print("Core data mode: embedded snapshot; no network requests")

Python: 3.13.13 ; NumPy: 2.5.3 ; Pandas: 3.0.6
Working folder: /Users/zhiluo/Developer/vibeit-cookbook/cookbook/downloads
Core data mode: embedded snapshot; no network requests

def load_snapshot(recipe_id, expected_sha256, snapshot_key):
    """Read embedded data from this notebook, including a renamed copy."""
    for path in sorted(Path.cwd().glob("*.ipynb")):
        try:
            notebook = json.loads(path.read_text(encoding="utf-8"))
            metadata = notebook.get("metadata", {}).get("vibeit_cookbook", {})
            blob = metadata.get("snapshots", {}).get(snapshot_key)
            if metadata.get("recipe_id") != recipe_id or not blob:
                continue
            if blob.get("sha256") != expected_sha256:
                continue
            raw = gzip.decompress(base64.b64decode(blob["gzip_base64"], validate=True))
            if hashlib.sha256(raw).hexdigest() != expected_sha256:
                raise ValueError("Embedded snapshot checksum mismatch: " + snapshot_key)
            return json.loads(raw)
        except (OSError, json.JSONDecodeError):
            continue
    raise FileNotFoundError(
        "Save/open the complete .ipynb in the active working folder. "
        "The embedded snapshot could not be located for " + recipe_id)

SNAPSHOT_HASHES={'iris': 'cb2512f34911979eef3a42b7a9e6fbc6ca566fd2c0eaa7bd86daefd7ac1cf5ad'}
def snapshot(key):
    return load_snapshot(RECIPE_ID,SNAPSHOT_HASHES[key],key)
print("Embedded snapshots:",list(SNAPSHOT_HASHES))

Embedded snapshots: ['iris']

def audit_numeric_table(frame,columns):
    """Audit observed rows; never silently impute or remove them."""
    if frame.empty or frame.index.has_duplicates:
        raise ValueError('Expected a nonempty table with unique row IDs')
    missing=[name for name in columns if name not in frame.columns]
    if missing:raise ValueError('Missing required columns: '+str(missing))
    values=frame[columns].to_numpy(dtype=float)
    if not np.isfinite(values).all():raise ValueError('Non-finite measurements; inspect missing values explicitly')
    return {'rows':len(frame),'features':len(columns),'finite':True,'duplicate_measurement_rows':int(frame[columns].duplicated().sum())}

print("Method ready: audit_numeric_table")

Method ready: audit_numeric_table

def stratified_holdout(labels,test_fraction=.3,seed=42):
    """Reserve a deterministic class-stratified test set before fitting anything."""
    labels=np.asarray(labels)
    if labels.ndim!=1 or not 0<test_fraction<1:raise ValueError('Invalid labels or test fraction')
    rng=np.random.default_rng(seed);train=[];test=[]
    for label in np.unique(labels):
        indices=np.flatnonzero(labels==label)
        if len(indices)<3:raise ValueError('Each class needs at least three observations')
        indices=rng.permutation(indices)
        count=max(1,min(len(indices)-1,int(round(len(indices)*test_fraction))))
        test.extend(indices[:count]);train.extend(indices[count:])
    return np.sort(train),np.sort(test)

print("Method ready: stratified_holdout")

Method ready: stratified_holdout

def stratified_folds(labels,count=5,seed=42):
    """Return validation indices relative to the training subset only."""
    labels=np.asarray(labels);rng=np.random.default_rng(seed);folds=[[] for _ in range(count)]
    if count<2:raise ValueError('At least two folds are required')
    for label in np.unique(labels):
        indices=np.flatnonzero(labels==label)
        if len(indices)<count:raise ValueError('A class has fewer observations than folds')
        for j,chunk in enumerate(np.array_split(rng.permutation(indices),count)):folds[j].extend(chunk)
    return [np.sort(fold) for fold in folds]

print("Method ready: stratified_folds")

Method ready: stratified_folds

def knn_predict(train_values,train_labels,test_values,k=5):
    """Euclidean kNN with a deterministic smallest-label tie rule."""
    train_values=np.asarray(train_values,dtype=float);test_values=np.asarray(test_values,dtype=float)
    train_labels=np.asarray(train_labels)
    if train_values.ndim!=2 or test_values.ndim!=2 or train_values.shape[1]!=test_values.shape[1]:
        raise ValueError('Training/test feature dimensions differ')
    if len(train_labels)!=len(train_values) or not isinstance(k,(int,np.integer)) or not 1<=k<=len(train_values):raise ValueError('Invalid k or labels')
    if not np.isfinite(train_values).all() or not np.isfinite(test_values).all():raise ValueError('Non-finite features')
    classes=np.unique(train_labels);predictions=[]
    for row in test_values:
        neighbors=np.argsort(np.sum((train_values-row)**2,axis=1),kind='stable')[:k]
        votes=[np.sum(train_labels[neighbors]==label) for label in classes]
        predictions.append(classes[np.argmax(votes)])
    return np.asarray(predictions)

print("Method ready: knn_predict")

Method ready: knn_predict

def classification_metrics(truth,predictions,classes):
    truth=np.asarray(truth);predictions=np.asarray(predictions);classes=np.asarray(classes)
    if truth.shape!=predictions.shape or truth.size==0:raise ValueError('Invalid comparison')
    if not np.isin(truth,classes).all() or not np.isin(predictions,classes).all():raise ValueError('Unknown class')
    confusion=np.array([[np.sum((truth==a)&(predictions==b)) for b in classes] for a in classes])
    true_positive=np.diag(confusion)
    precision=np.divide(true_positive,confusion.sum(axis=0),out=np.zeros(len(classes),dtype=float),where=confusion.sum(axis=0)>0)
    recall=np.divide(true_positive,confusion.sum(axis=1),out=np.zeros(len(classes),dtype=float),where=confusion.sum(axis=1)>0)
    f1=np.divide(2*precision*recall,precision+recall,out=np.zeros(len(classes),dtype=float),where=precision+recall>0)
    return dict(confusion=confusion,accuracy=float(np.mean(truth==predictions)),macro_f1=float(f1.mean()),precision=precision,recall=recall)

print("Method ready: classification_metrics")

Method ready: classification_metrics

iris=snapshot("iris")
frame=pd.DataFrame(iris["rows"],columns=iris["columns"])
features=list(frame.columns[:4])
audit=audit_numeric_table(frame,features)
assert len(frame)==150 and frame["species"].nunique()==3
assert (frame[features]>0).all().all()
print("Audit:",audit,"; measurement unit: cm")
print("Source variant:",iris["variant"])
display(frame.head(8))
values=frame[features].to_numpy(float);labels=frame.species.to_numpy()
train_indices,test_indices=stratified_holdout(labels,test_fraction=.3,seed=42)
assert not set(train_indices)&set(test_indices)
assert len(train_indices)+len(test_indices)==150
print("Training/test observations:",len(train_indices),len(test_indices))

Audit: {'rows': 150, 'features': 4, 'finite': True, 'duplicate_measurement_rows': 3} ; measurement unit: cm
Source variant: Original UCI iris.data; no silent correction of the two documented transcription discrepancies

Training/test observations: 105 45

candidates=[1,3,5,7,9];folds=stratified_folds(labels[train_indices],5,seed=7)
cv=[]
for k in candidates:
 scores=[]
 for validation in folds:
  fitting=np.setdiff1d(np.arange(len(train_indices)),validation)
  x=values[train_indices[fitting]];mean=x.mean(axis=0);scale=x.std(axis=0)
  prediction=knn_predict((x-mean)/scale,labels[train_indices[fitting]],(values[train_indices[validation]]-mean)/scale,k)
  scores.append(np.mean(prediction==labels[train_indices[validation]]))
 cv.append(scores)
cv=np.asarray(cv);chosen_k=candidates[int(np.argmax(cv.mean(axis=1)))]
print("Training-only selected k:",chosen_k,"; ties choose the smallest candidate")
fig,ax=plt.subplots(figsize=(7,4));ax.errorbar(candidates,cv.mean(axis=1),yerr=cv.std(axis=1,ddof=1),fmt="o-")
ax.set(xlabel="k",ylabel="Fold accuracy",ylim=(0,1.05),title="Training-only stratified CV; bars = fold SD");plt.tight_layout();plt.show()

Training-only selected k: 3 ; ties choose the smallest candidate

mean=values[train_indices].mean(axis=0);scale=values[train_indices].std(axis=0)
prediction=knn_predict((values[train_indices]-mean)/scale,labels[train_indices],(values[test_indices]-mean)/scale,chosen_k)
classes=np.unique(labels);metrics=classification_metrics(labels[test_indices],prediction,classes)
print("Held-out accuracy:",metrics["accuracy"],"; macro-F1:",metrics["macro_f1"])
fig,ax=plt.subplots(figsize=(6,5));im=ax.imshow(metrics["confusion"],cmap="Blues")
short=[v.replace("Iris-","") for v in classes];ax.set_xticks(range(3),short,rotation=25);ax.set_yticks(range(3),short)
for i in range(3):
 for j in range(3):ax.text(j,i,str(metrics["confusion"][i,j]),ha="center",va="center")
ax.set(xlabel="Predicted class",ylabel="True class",title="Untouched test set");fig.colorbar(im,ax=ax,label="Observations");plt.tight_layout();plt.show()

Held-out accuracy: 0.9333333333333333 ; macro-F1: 0.9332591768631814

fig,ax=plt.subplots(figsize=(7,5))
for label in classes:
 mask=labels[test_indices]==label;ax.scatter(values[test_indices][mask,2],values[test_indices][mask,3],label=label,s=35)
wrong=prediction!=labels[test_indices]
ax.scatter(values[test_indices][wrong,2],values[test_indices][wrong,3],marker="x",color="black",s=90,label="Incorrect prediction")
ax.set(xlabel="Petal length (cm)",ylabel="Petal width (cm)",title="Test errors in an observed feature plane");ax.legend(fontsize=8);plt.tight_layout();plt.show()

assert metrics["confusion"].sum()==len(test_indices)
assert np.isclose(np.trace(metrics["confusion"])/len(test_indices),np.mean(prediction==labels[test_indices]))
OUTPUT_DIR.mkdir(parents=True,exist_ok=True)
pd.DataFrame({"source_row":test_indices,"true":labels[test_indices],"predicted":prediction}).to_csv(OUTPUT_DIR/"test-predictions.csv",index=False)
pd.DataFrame(cv,index=candidates).to_csv(OUTPUT_DIR/"training-cv.csv")
(OUTPUT_DIR/"model.json").write_text(json.dumps({"k":chosen_k,"seed":42,"train_indices":train_indices.tolist(),"test_indices":test_indices.tolist(),"training_mean":mean.tolist(),"training_SD":scale.tolist()},indent=2))
print("Metric reconciliation passed; exports:",OUTPUT_DIR)

Metric reconciliation passed; exports: /Users/zhiluo/Developer/vibeit-cookbook/cookbook/downloads/results/ds03-iris-knn

RUN_ONLINE_UPDATE=False
if RUN_ONLINE_UPDATE:
    import requests
    try:
        response=requests.get('https://archive.ics.uci.edu/dataset/53',timeout=20)
        response.raise_for_status()
        print("Source:",response.url)
        print(response.text[:700])
    except requests.RequestException as error:
        print("Online extension unavailable:",type(error).__name__,str(error)[:160])
else:
    print("Online extension skipped; embedded core data unchanged")

Online extension skipped; embedded core data unchanged