import os
import json
import hashlib
import datetime
import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import classification_report, confusion_matrix
# 1. 计算文件/数据的 MD5 哈希
def compute_data_hash(df: pd.DataFrame) -> str:
data_str = pd.util.hash_pandas_object(df).values.tobytes()
return hashlib.md5(data_str).hexdigest()[:10]
# 2. 结构化实验记录器类
class ExperimentTracker:
def __init__(self, exp_name: str, base_dir: str = "runs"):
self.exp_name = exp_name
self.run_id = f"{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}_{exp_name}"
self.run_dir = os.path.join(base_dir, self.run_id)
os.makedirs(self.run_dir, exist_ok=True)
def log_run(self, config: dict, data_hash: str, metrics: dict):
meta_data = {
"run_id": self.run_id,
"timestamp": str(datetime.datetime.now()),
"git_commit": "mock_commit_a1b2c3d", # 可用 git rev-parse 获取
"data_hash": data_hash,
"config": config,
"metrics": metrics
}
json_path = os.path.join(self.run_dir, "run_meta.json")
with open(json_path, "w", encoding="utf-8") as f:
json.dump(meta_data, f, indent=2, ensure_ascii=False)
print(f"✅ 结构化实验日志已成功记录至: {json_path}")
# 3. 实战运行与错误分析
np.random.seed(42)
N = 200
X = np.random.randn(N, 2)
y = (X[:, 0] * 1.5 + X[:, 1] > 0.5).astype(int)
df = pd.DataFrame(X, columns=['feat1', 'feat2'])
df['target'] = y
d_hash = compute_data_hash(df)
X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.3, random_state=42)
model = LogisticRegression().fit(X_tr, y_tr)
preds = model.predict(X_te)
# 提取错判样本 (Error Analysis)
te_df = pd.DataFrame(X_te, columns=['feat1', 'feat2'])
te_df['y_true'] = y_te
te_df['y_pred'] = preds
errors_df = te_df[te_df['y_true'] != te_df['y_pred']]
# 计算指标并记录实验
acc = float((preds == y_te).mean())
tracker = ExperimentTracker(exp_name="baseline_lr")
tracker.log_run(
config={"model": "LogisticRegression", "seed": 42},
data_hash=d_hash,
metrics={"accuracy": round(acc, 4), "num_errors": len(errors_df)}
)
print("\n================ 错判样本分析 (Error Analysis 示例) ================")
print(f"测试集总数: {len(y_te)} | 错判样本数: {len(errors_df)} (错误率: {len(errors_df)/len(y_te)*100:.1f}%)")
print("前 3 条错判样本明细:")
print(errors_df.head(3).to_string(index=False))