File size: 3,636 Bytes
e23cf4c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 | """Đánh giá hiệu năng model bằng linear probe.
Đặc trưng được trích riêng từng split (train/val/test) — đảm bảo không có
identity leakage. Quy trình:
1. Load features train (đã trích từ backbone)
2. Train LogisticRegression (linear probe, class_weight='balanced') trên train
3. Đánh giá trên test (báo cáo cả val nếu có): acc, precision, recall, F1, ROC-AUC, confusion matrix
"""
import argparse
import json
import os
import numpy as np
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import (
accuracy_score,
confusion_matrix,
f1_score,
precision_score,
recall_score,
roc_auc_score,
)
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
CLASS_NAMES = {0: "real", 1: "fake"}
def load_features(path: str):
data = np.load(path, allow_pickle=True)
X, y = data["features"], data["labels"]
print(f" {os.path.basename(path)}: {X.shape} | real={int((y == 0).sum())}, fake={int((y == 1).sum())}")
return X, y
def eval_split(clf, X, y):
y_pred = clf.predict(X)
y_prob = clf.predict_proba(X)[:, 1]
return {
"accuracy": float(accuracy_score(y, y_pred)),
"precision": float(precision_score(y, y_pred)),
"recall": float(recall_score(y, y_pred)),
"f1": float(f1_score(y, y_pred)),
"roc_auc": float(roc_auc_score(y, y_prob)),
"confusion_matrix": confusion_matrix(y, y_pred).tolist(),
}, y_pred, y_prob
def print_results(name: str, m: dict):
cm = m["confusion_matrix"]
print(f"\n--- {name.upper()} ---")
print(f" accuracy : {m['accuracy']:.4f}")
print(f" precision: {m['precision']:.4f} (fake=positive)")
print(f" recall : {m['recall']:.4f}")
print(f" f1 : {m['f1']:.4f}")
print(f" roc_auc : {m['roc_auc']:.4f}")
print(f" CM [TN FP; FN TP]: TN={cm[0][0]} FP={cm[0][1]} FN={cm[1][0]} TP={cm[1][1]}")
def main():
parser = argparse.ArgumentParser(description="Linear probe evaluation (train / eval tách riêng)")
parser.add_argument("--train-features", required=True, help="features.npz của tập train")
parser.add_argument("--test-features", required=True, help="features.npz của tập test")
parser.add_argument("--val-features", default=None, help="features.npz của tập val (tùy chọn)")
parser.add_argument("--output", default="outputs/results/evaluation_report.json")
parser.add_argument("--seed", type=int, default=42)
args = parser.parse_args()
print("Load features:")
X_train, y_train = load_features(args.train_features)
X_test, y_test = load_features(args.test_features)
X_val, y_val = (load_features(args.val_features) if args.val_features else (None, None))
# ---------- Train linear probe ----------
clf = make_pipeline(
StandardScaler(),
LogisticRegression(class_weight="balanced", max_iter=2000, random_state=args.seed),
)
clf.fit(X_train, y_train)
print("\nĐã train xong linear probe.")
# ---------- Đánh giá ----------
report = {"seed": args.seed, "splits": {}}
for name, X, y in [("val", X_val, y_val), ("test", X_test, y_test)]:
if X is None:
continue
m, _, _ = eval_split(clf, X, y)
report["splits"][name] = {"metrics": m}
print_results(name, m)
os.makedirs(os.path.dirname(args.output), exist_ok=True)
with open(args.output, "w") as f:
json.dump(report, f, indent=2, ensure_ascii=False)
print(f"\nĐã lưu báo cáo: {args.output}")
if __name__ == "__main__":
main()
|