<jupyter_start><jupyter_text>非监督学习方法<jupyter_code>from sklearn.preprocessing import StandardScaler, MinMaxScaler
from sklearn.model_selection import train_test_split, GridSearchCV
## 数据准备
## 训练集：60%正常数据
## 验证集：20%正常+50%异常
## 测试集：20%正常+50%异常
class DATA1():
    def __init__(self,raw=raw):
        raw['Amount'] = np.log10(raw.Amount+1e-5)
        raw.drop('Time',axis=1,inplace=True)
        
        Fraud, Normal = raw.query('Class==1'), raw.query('Class==0')
        self.X_train,Normal40 = train_test_split(Normal,test_size=.4)
        Fraud1,Fraud2 = train_test_split(Fraud,test_size=.5)
        Normal1,Normal2 = train_test_split(Normal40,test_size=.5)
        self.X_valid = pd.concat([Fraud1,Normal1])
        self.X_test = pd.concat([Fraud2,Normal2])
        
        self.y_train = self.X_train.pop('Class')
        self.y_valid = self.X_valid.pop('Class')
        self.y_test = self.X_test.pop('Class')
        
    def fit_transform(self,estimator=StandardScaler()):
        self.estimator = estimator
        self.estimator.fit(self.X_train)
        
        self.X_train = self.estimator.transform(self.X_train)
        self.X_valid = self.estimator.transform(self.X_valid)
        self.X_test = self.estimator.transform(self.X_test)
        
        return self.estimator
        
data1 = DATA1(raw.copy())
data1.fit_transform(StandardScaler())
tsne_plot(tsne,data1.X_test[:1000],data1.y_test[:1000])
## 大致可分<jupyter_output>D:\Program Files\Anaconda3\envs\pytorch\lib\site-packages\pandas\plotting\_matplotlib\tools.py:307: MatplotlibDeprecationWarning: 
The rowNum attribute was deprecated in Matplotlib 3.2 and will be removed two minor releases later. Use ax.get_subplotspec().rowspan.start instead.
  layout[ax.rowNum, ax.colNum] = ax.get_visible()
D:\Program Files\Anaconda3\envs\pytorch\lib\site-packages\pandas\plotting\_matplotlib\tools.py:307: MatplotlibDeprecationWarning: 
The colNum attribute was deprecated in Matplotlib 3.2 and will be removed two minor releases later. Use ax.get_subplotspec().colspan.start instead.
  layout[ax.rowNum, ax.colNum] = ax.get_visible()
D:\Program Files\Anaconda3\envs\pytorch\lib\site-packages\pandas\plotting\_matplotlib\tools.py:313: MatplotlibDeprecationWarning: 
The rowNum attribute was deprecated in Matplotlib 3.2 and will be removed two minor releases later. Use ax.get_subplotspec().rowspan.start instead.
  if not layout[ax.rowNum + 1, ax.colNum]:
D:\Program Files\Ana[...]<jupyter_text>AutoEncoder<jupyter_code>import torch as t
import torch.utils.data as Data<jupyter_output><empty_output><jupyter_text>定义<jupyter_code>BATCH_SIZE=256
EPOCHS = 100

class CCFD(Data.Dataset):
    def __init__(self,X_train:np.array):
        self.data = t.from_numpy(X_train).float()
        
    def __len__(self):
        return len(self.data)
    
    def __getitem__(self,i):
        return self.data[i]
    
train_db = CCFD(data1.X_train)
valid_db = CCFD(data1.X_valid)
class AE(t.nn.Module):
    def __init__(self, input_dim):
        super().__init__()
        
        self.encoder = t.nn.Sequential(
            t.nn.Linear(input_dim,16),
            t.nn.ReLU(),
            t.nn.Linear(16,8),
            t.nn.ReLU(),
            t.nn.Linear(8,2),
            t.nn.ReLU())
        
        self.decoder = t.nn.Sequential(
            t.nn.Linear(2,8),
            t.nn.ReLU(),
            t.nn.Linear(8,16),
            t.nn.ReLU(),
            t.nn.Linear(16,input_dim))
        
    def forward(self,x):
        encoded = self.encoder(x)
        decoded = self.decoder(encoded)
        return encoded, decoded
    
autoencoder = AE(input_dim = data1.X_train.shape[1])
optimizer = t.optim.Adam(autoencoder.parameters(),lr=5e-3)
loss_func = t.nn.MSELoss()

train_loader = Data.DataLoader(train_db,batch_size=BATCH_SIZE,shuffle=True)<jupyter_output><empty_output><jupyter_text>训练集训练<jupyter_code>for epoch in range(EPOCHS):
    
    for step, x in enumerate(train_loader):
        
        b_x, b_y = x.view(*x.shape), x.view(*x.shape)
        
        ## 计算输出
        encoded, decoded = autoencoder.forward(b_x)
        ## 计算误差
        loss = loss_func(decoded, b_y)
        ## 反向传播
        loss.backward()
        ## 更新参数
        optimizer.step()
        ## 清除梯度
        optimizer.zero_grad()
        
    y_valid = autoencoder.forward(valid_db.data)[1]
    loss_valid = loss_func(y_valid, valid_db.data).detach().numpy()
#     print(f'Epoch: {epoch}, | val loss: {loss_valid:.4f}')<jupyter_output><empty_output><jupyter_text>验证集选阈值<jupyter_code>from sklearn.metrics import PrecisionRecallDisplay, precision_recall_curve, confusion_matrix, precision_recall_fscore_support
def ae_error(estimator, X, y):
    _, pred = estimator.forward(t.Tensor(X))
    residual = (pred.detach()-X).square().mean(1).detach().numpy()
    error = y.to_frame().assign(pred = residual)
    return error
error = ae_error(autoencoder, data1.X_valid, data1.y_valid)
error.head()
def thre_f1(error):
    p, r, thre = precision_recall_curve(error.Class,error.pred)
    f1 = 2*p*r/(p+r+1e-5)
    ax=PrecisionRecallDisplay(p,r).plot()
    plt.plot(r, f1,label='f1-score')
    idmax = f1.argmax()
    plt.scatter(r[idmax], f1[idmax], marker='o', color='g', 
                label=f'Best f1-score:{f1[idmax]:.3f}\nThreshold:{thre[idmax]:.3f}')
    plt.legend()
    return thre[idmax]
thre = thre_f1(error)
thre<jupyter_output><empty_output><jupyter_text>测试集表现<jupyter_code>test_error = ae_error(autoencoder,data1.X_test,data1.y_test)
test_error['PRED'] = test_error.pred>thre
a=confusion_matrix(test_error.Class,test_error.PRED);a
def summary(confusion):
    r = confusion[1,1]/confusion[1,:].sum()
    p = confusion[1,1]/confusion[:,1].sum()
    f1 = 2*p*r/(p+r)
    print(f'P: {p:.3f},\nR: {r:.3f},\nF1-SCORE: {f1:.3f}')
    return
summary(a) ## 效果太差了，放弃<jupyter_output>P: 0.344,
R: 0.492,
F1-SCORE: 0.405<jupyter_text>Gaussian Mixture  模型定义<jupyter_code>from scipy.stats import multivariate_normal as mn

class myGM:
    def __init__(self,thre=.5):
        self.thre = thre
        pass
    
    def fit(self, X, y=None):
        X_ = pd.DataFrame(X)
        self.mu = X_.mean().values
        self.cov = X_.cov().values
        return self
    
    def predict_pdf(self, X, y=None):
        pdf = mn.pdf(X, self.mu, self.cov)
        return pdf
    
    def predict(self,X):
        pdf = self.predict_pdf(X)
        return (pdf<self.thre).astype(int)<jupyter_output><empty_output><jupyter_text>训练集训练<jupyter_code>mygm = myGM()
mygm.fit(data1.X_train)<jupyter_output><empty_output><jupyter_text>验证集找阈值<jupyter_code>def gm_thre_fi1(estimator, X, y):
    pred = estimator.predict_pdf(X)
    error = y.to_frame().assign(pred = pred)
    
    error.sort_values('pred', ascending=True,inplace=True)
    t = error.pred
    p = error.expanding()['Class'].mean()
    r = error.Class.cumsum()/error.Class.sum()
    f1 = 2*p*r/(p+r)
    
    PrecisionRecallDisplay(p,r).plot()
    plt.plot(r,f1,label='f1-score')
    idmax = f1.idxmax()
    plt.scatter(r[idmax], f1[idmax], marker='o', color='g', 
                label=f'Best f1-score:{f1[idmax]:.3f}')
    plt.legend()
    return t[idmax]
mygm.thre = gm_thre_fi1(mygm, data1.X_valid, data1.y_valid)<jupyter_output><empty_output><jupyter_text>测试集评估<jupyter_code>print(confusion_matrix(data1.y_test, mygm.predict(data1.X_test)))
summary(confusion_matrix(data1.y_test, mygm.predict(data1.X_test)))
## 比随机分要好一点<jupyter_output>[[56758   105]
 [   75   171]]
P: 0.620,
R: 0.695,
F1-SCORE: 0.655<jupyter_text>监督学习方法  重新划分<jupyter_code>## 数据准备
## 训练集：75%正常+75%异常
## 测试集：25%正常+25%异常
class DATA2():
    def __init__(self,raw=raw):
        raw['Amount'] = np.log10(raw.Amount+1e-5)
        raw.drop('Time',axis=1,inplace=True)
        
        self.X_train, self.X_test = train_test_split(raw, 
                                        stratify=raw.Class)       
        self.y_train = self.X_train.pop('Class')
        self.y_test = self.X_test.pop('Class')
        
    def fit_transform(self,estimator=StandardScaler()):
        self.estimator = estimator
        self.estimator.fit(self.X_train)
        
        self.X_train = self.estimator.transform(self.X_train)
        self.X_test = self.estimator.transform(self.X_test)
        
        return self.estimator
data2 = DATA2(raw.copy())
data2.fit_transform(StandardScaler())
## 定义评估函数
def metric(estimator, X_test, y_test):
    plt.figure(figsize=(16,5))
    ax1=plt.subplot(121)
    plot_precision_recall_curve(estimator,X_test,y_test,ax=ax1)
    ax2 = plt.subplot(122)
    ax = plot_confusion_matrix(estimator,X_test,y_test,ax=ax2)
    summary(ax.confusion_matrix)
    return 
from sklearn.model_selection import train_test_split, GridSearchCV
from sklearn.metrics import plot_confusion_matrix, plot_precision_recall_curve<jupyter_output><empty_output><jupyter_text>LogisticRegression<jupyter_code>from sklearn.linear_model import LogisticRegression
lr = LogisticRegression()
params = {'C':[.1,1,10]}
lr_clf = GridSearchCV(lr,params,scoring='f1',n_jobs=-1)
lr_clf.fit(data2.X_train,data2.y_train)
print(clf.best_params_, clf.best_score_)
metric(lr_clf.best_estimator_, data2.X_test, data2.y_test)<jupyter_output>P: 0.894,
R: 0.618,
F1-SCORE: 0.731<jupyter_text>SMOTE OR NOT<jupyter_code>from imblearn.over_sampling import SMOTE,RandomOverSampler

smote = RandomOverSampler(sampling_strategy=.3)
# smote = SMOTE(sampling_strategy=.3)
train_smoted = smote.fit_resample(X_train, df_train.Class)

lr = LogisticRegression()
params = {'C':[.001,.01,.1,1,10]}
clf = GridSearchCV(lr,params,scoring='f1',n_jobs=-1)
clf.fit(*train_smoted)

plt.figure(figsize=(16,5))
ax1 = plt.subplot(121)
plot_precision_recall_curve(clf.best_estimator_,X_test,df_test.Class,ax=ax1)
ax2 = plt.subplot(122)
plot_confusion_matrix(clf.best_estimator_,X_test,df_test.Class,ax=ax2)
summary(np.array([[56429,435],[11,87]])) ## 效果比未拓展时更差了<jupyter_output>P: 0.888,
R: 0.167,
F1-SCORE: 0.281<jupyter_text>SVM<jupyter_code>from sklearn.svm import SVC
svc = SVC(probability=True)
svc.fit(data2.X_train, data2.y_train)
metric(svc,data2.X_test, data2.y_test)<jupyter_output>P: 0.942,
R: 0.659,
F1-SCORE: 0.775<jupyter_text>IsoForest<jupyter_code>from sklearn.ensemble import IsolationForest
iforest = IsolationForest()
iforest.fit(data1.X_train)
summary(confusion_matrix(data1.y_test,(iforest.predict(data1.X_test)==-1).astype(int)))<jupyter_output>P: 0.079,
R: 0.833,
F1-SCORE: 0.144<jupyter_text>XGBoost<jupyter_code>from xgboost import XGBClassifier
xgb = XGBClassifier()
params = {'n_estimators':[100,150],'max_depth':[4,5,6]}
xgb_clf = GridSearchCV(xgb,params,n_jobs=-1,scoring='f1')
xgb_clf.fit(data2.X_train, data2.y_train)
xgb = xgb_clf.best_estimator_
metric(xgb_clf.best_estimator_, data2.X_test, data2.y_test)<jupyter_output>P: 0.925,
R: 0.805,
F1-SCORE: 0.861<jupyter_text>Adaboost<jupyter_code>from sklearn.ensemble import AdaBoostClassifier
adb = AdaBoostClassifier(n_estimators=100)
params = {'n_estimators':[50,100,200]}
adb_clf = GridSearchCV(adb,params,scoring='f1',n_jobs=-1)
adb_clf.fit(data2.X_train, data2.y_train)
metric(adb_clf.best_estimator_, data2.X_test, data2.y_test)<jupyter_output>P: 0.888,
R: 0.772,
F1-SCORE: 0.826<jupyter_text>Random Forest<jupyter_code>from sklearn.ensemble import RandomForestClassifier
rf = RandomForestClassifier()
rf.fit(data2.X_train,data2.y_train)
metric(rf, data2.X_test, data2.y_test)<jupyter_output>P: 0.944,
R: 0.821,
F1-SCORE: 0.878<jupyter_text>Voting<jupyter_code>from sklearn.ensemble import VotingClassifier
from sklearn.preprocessing import LabelEncoder
voting_clf = VotingClassifier([('svc',svc),('xgb',xgb_clf.best_estimator_),('rf',rf)],voting='soft',
                              weights=[.19,.39,.42])
voting_clf.estimators_ = [svc, xgb_clf.best_estimator_, rf]
voting_clf.le_ = LabelEncoder().fit(data2.y_test)
voting_clf.classes_ = voting_clf.le_.classes_
metric(voting_clf, data2.X_test, data2.y_test)<jupyter_output>P: 0.935,
R: 0.821,
F1-SCORE: 0.874
{'fraction_non_alphanumeric': 0.14296501384344323, 'fraction_numerical': 0.035070056212769526, 'mean_word_length': 4.16240796881767, 'pattern_counts': {'":': 0, '<': 67, '<?xml version=': 0, '>': 67, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '19', 'dir': 'jupyter-structured-clean-dedup', 'id': '370038', 'license': 'MIT', 'n_tokens_mistral': 5123, 'n_tokens_neox': 4927, 'n_words': 621, 'path': 'credit card fraud detection.ipynb', 'repo_name': '123385918/Python-Science-Handbook', 'size': '11919'}