主页 / coding / 问题

问题[scikit-learn](coding)

Martin Hope
FelipeB
Asked: 2024-12-21 00:26:44 +0800 CST
  • 5

我正在开发一个模仿https://www.mtgassist.com/的 Python 项目。对于那些不太熟悉的人：Magic 是一款交易卡牌游戏，其中有可收藏的卡牌，价格可能非常昂贵。该项目应该采用一张卡牌的名称，并根据几个功能列出具有类似机制（并且希望更便宜）的其他卡牌，包括描述卡牌功能的“oracle_text”。

  1. 我正在使用 sklearn 的 TfidfVectorizer，其参数如下：
porter = PorterStemmer()
def tokenizer_stemmer(text: str) -> str:
    stop = stopwords.words('english')
    return [porter.stem(word) for word in text.split() if word not in stop]

tfidf = TfidfVectorizer(
    ngram_range=(1,2)
    , tokenizer=tokenizer_stemmer
    , stop_words=stopwords.words('english')
)
  1. 然后我将 TfidfVectorizer.fit_transform 与我之前加载的约 20k 行 pandas DataFrame 一起使用。此过程大约需要25 秒：

token_mat = tfidf.fit_transform(df_not_na['oracle_text'])

  1. 接下来，我将其转换token_mat为形状为 ~(20_000, 90_000) 的 numpy 数组 ( token_arr)，并计算所选卡片与数组中所有卡片之间的欧几里得距离（这需要额外的25 秒）。最后，我打印出前 5 张“最接近”的卡片的名称：
token_arr = token_mat.toarray()

distances = []
for _card in tqdm(token_mat):
    distances.append(np.linalg.norm(_card - chosen_card_array))

nearest_5 = np.argpartition(distances, 10)[:10]
print(df_not_na.iloc[nearest_5][['name', 'oracle_text']])

我的目标是优化这个过程并减少创建特征向量和计算距离的时间。

我尝试使用二元组而不是 ngram_range=(1,2)，但效果并不大。

我也想过使用 numba，但读到 sklearn/numpy 具有类似的嵌入功能，因此不会有太大的好处。

也请告诉我其他建议！谢谢

scikit-learn
Martin Hope
Pau Vila Soler
Asked: 2024-11-29 22:45:51 +0800 CST
  • 5

我不确定我是否遗漏了一些非常基本的东西，但我已经开始注意到 sklearn 的 r2_score() 函数返回的 R2 分数与 PyTorch 的 MSELoss() 计算出的 R2 分数之间存在一些细微的差异（在 statistics.variance() 的额外帮助下）。

基本上，sklearn 方法返回的 R2 分数始终（略微）低于通过 MSELoss() 返回的分数。

以下是一些重现差异的基本代码。

from sklearn.metrics import r2_score
from torch.nn import MSELoss
import statistics 
import random
import torch
import numpy as np 

actuals = random.sample(range(1, 50), 40)

preds = []

for value in actuals:
    pred = value * 0.70
    preds.append(pred)

loss = MSELoss()

mse = loss(torch.tensor(preds), torch.tensor(actuals))

r2 = 1 - mse / statistics.variance(actuals)

score = r2_score(actuals, preds)

print(f'R2 Score using (PyTorch) MSELoss: {r2}')
print(f'R2 Score using (sklearn) r2_score: {score}')

示例输出：

R2 Score using (PyTorch) MSELoss: 0.6261289715766907 R2 Score using (sklearn) r2_score: 0.6165425269729996

我认为这可能与 MSELoss() 将张量作为输入有关（但 sklearn 不这样做），但我真的不知道为什么或如何。

如果有人能帮忙/澄清一下就好了。

谢谢！

版本：

  • PyTorch == 2.1.0
  • scikit-learn == 1.4.2
  • Python == 3.9.18
scikit-learn
Martin Hope
Simon
Asked: 2024-09-25 03:51:17 +0800 CST
  • 7

我正在尝试使用 scikit-learnLabelEncoder和 Polars DataFrame 来编码分类列。我正在使用以下代码。

import polars as pl

from sklearn.preprocessing import LabelEncoder

df = pl.DataFrame({
    "Color" : ["red","white","blue"]
})

enc = LabelEncoder()

但是，会出现错误。

ValueError: y should be a 1d array, got an array of shape () instead.

接下来，我尝试将该列转换为 NumPy。

df.with_columns(
    enc.fit_transform(pl.col("Color").to_numpy()) 
)

现在，出现了一个不同的错误。

AttributeError: 'Expr' object has no attribute 'to_numpy'

注意。我发现.cast(pl.Categorical).to_physical()可以使用来获得所需的结果。不过，我更喜欢transform()在我的测试数据集上使用类似的东西。

df.with_columns(
    pl.col("Color").cast(pl.Categorical).to_physical().alias("Color_encoded")
)
scikit-learn
Martin Hope
Qazi Fahim Farhan
Asked: 2024-05-04 13:09:47 +0800 CST
  • 4

我正在尝试使用 CNN 在 DNA 序列中查找模式（例如“CTCATGTCA”）来进行二元分类。我用pytorch写了一个模型。当模式位于序列的中心时，模型会检测到它。但如果该模式出现在随机位置，则该模型不起作用。如何使CNN对模式的位置不变？

这是我的代码：

import logging

import torch
import torch.nn as nn
import torch.optim as optim
import pandas as pd
from sklearn import metrics
from skorch import NeuralNetClassifier
from skorch.callbacks import EpochScoring
from torch.utils.data import DataLoader, Dataset
import numpy as np

import constants

timber = logging.getLogger()
logging.basicConfig(level=logging.INFO)  # change to level=logging.DEBUG to print more logs...


# utils

def one_hot_e(dna_seq: str) -> np.ndarray:
  mydict = {'A': np.asarray([1.0, 0.0, 0.0, 0.0]), 'C': np.asarray([0.0, 1.0, 0.0, 0.0]),
            'G': np.asarray([0.0, 0.0, 1.0, 0.0]), 'T': np.asarray([0.0, 0.0, 0.0, 1.0]),
            'N': np.asarray([0.0, 0.0, 0.0, 0.0]), 'H': np.asarray([0.0, 0.0, 0.0, 0.0]),
            'a': np.asarray([1.0, 0.0, 0.0, 0.0]), 'c': np.asarray([0.0, 1.0, 0.0, 0.0]),
            'g': np.asarray([0.0, 0.0, 1.0, 0.0]), 't': np.asarray([0.0, 0.0, 0.0, 1.0]),
            'n': np.asarray([0.0, 0.0, 0.0, 0.0]), '-': np.asarray([0.0, 0.0, 0.0, 0.0])}

  size_of_a_seq: int = len(dna_seq)

  # forward = np.zeros(shape=(size_of_a_seq, 4))

  forward_list: list = [mydict[dna_seq[i]] for i in range(0, size_of_a_seq)]
  encoded = np.asarray(forward_list)
  return encoded


def one_hot_e_column(column: pd.Series) -> np.ndarray:
  tmp_list: list = [one_hot_e(seq) for seq in column]
  encoded_column = np.asarray(tmp_list)
  return encoded_column


def reverse_dna_seq(dna_seq: str) -> str:
  # m_reversed = ""
  # for i in range(0, len(dna_seq)):
  #     m_reversed = dna_seq[i] + m_reversed
  # return m_reversed
  return dna_seq[::-1]


def complement_dna_seq(dna_seq: str) -> str:
  comp_map = {"A": "T", "C": "G", "T": "A", "G": "C",
              "a": "t", "c": "g", "t": "a", "g": "c",
              "N": "N", "H": "H", "-": "-",
              "n": "n", "h": "h"
              }

  comp_dna_seq_list: list = [comp_map[nucleotide] for nucleotide in dna_seq]
  comp_dna_seq: str = "".join(comp_dna_seq_list)
  return comp_dna_seq


def reverse_complement_dna_seq(dna_seq: str) -> str:
  return reverse_dna_seq(complement_dna_seq(dna_seq))


def reverse_complement_dna_seqs(column: pd.Series) -> pd.Series:
  tmp_list: list = [reverse_complement_dna_seq(seq) for seq in column]
  rc_column = pd.Series(tmp_list)
  return rc_column


class CNN1D(nn.Module):
  def __init__(self,
               in_channel_num_of_nucleotides=4,
               kernel_size_k_mer_motif=4,
               dnn_size=256,
               num_filters=1,
               lstm_hidden_size=128,
               *args, **kwargs):
    super().__init__(*args, **kwargs)
    self.conv1d = nn.Conv1d(in_channels=in_channel_num_of_nucleotides, out_channels=num_filters,
                            kernel_size=kernel_size_k_mer_motif, stride=2)
    self.activation = nn.ReLU()
    self.pooling = nn.MaxPool1d(kernel_size=kernel_size_k_mer_motif, stride=2)

    self.flatten = nn.Flatten()
    # linear layer

    self.dnn2 = nn.Linear(in_features=14 * num_filters, out_features=dnn_size)
    self.act2 = nn.Sigmoid()
    self.dropout2 = nn.Dropout(p=0.2)

    self.out = nn.Linear(in_features=dnn_size, out_features=1)
    self.out_act = nn.Sigmoid()

    pass

  def forward(self, x):
    timber.debug(constants.magenta + f"h0: {x}")
    h = self.conv1d(x)
    timber.debug(constants.green + f"h1: {h}")
    h = self.activation(h)
    timber.debug(constants.magenta + f"h2: {h}")
    h = self.pooling(h)
    timber.debug(constants.blue + f"h3: {h}")
    timber.debug(constants.cyan + f"h4: {h}")

    h = self.flatten(h)
    timber.debug(constants.magenta + f"h5: {h},\n shape {h.shape}, size {h.size}")
    h = self.dnn2(h)
    timber.debug(constants.green + f"h6: {h}")

    h = self.act2(h)
    timber.debug(constants.blue + f"h7: {h}")

    h = self.dropout2(h)
    timber.debug(constants.cyan + f"h8: {h}")

    h = self.out(h)
    timber.debug(constants.magenta + f"h9: {h}")

    h = self.out_act(h)
    timber.debug(constants.green + f"h10: {h}")
    # h = (h > 0.5).float()  # <---- should this go here?
    # timber.debug(constants.green + f"h11: {h}")

    return h


class CustomDataset(Dataset):
  def __init__(self, dataframe):
    self.x = dataframe["Sequence"]
    self.y = dataframe["class"]

  def __len__(self):
    return len(self.y)

  def preprocessing(self, x1, y1) -> (torch.Tensor, torch.Tensor, torch.Tensor):
    forward_col = x1

    backward_col = reverse_complement_dna_seqs(forward_col)

    forward_one_hot_e_col: np.ndarray = one_hot_e_column(forward_col)
    backward_one_hot_e_col: np.ndarray = one_hot_e_column(backward_col)

    tr_xf_tensor = torch.Tensor(forward_one_hot_e_col).permute(1, 2, 0)
    tr_xb_tensor = torch.Tensor(backward_one_hot_e_col).permute(1, 2, 0)
    # timber.debug(f"y1 {y1}")
    tr_y1 = np.array([y1])  # <--- need to put it inside brackets

    return tr_xf_tensor, tr_xb_tensor, tr_y1

  def __getitem__(self, idx):
    m_seq = self.x.iloc[idx]
    labels = self.y.iloc[idx]
    xf, xb, y = self.preprocessing(m_seq, labels)
    timber.debug(f"xf -> {xf.shape}, xb -> {xb.shape}, y -> {y}")
    return xf, xb, y


def test_dataloader():
  df = pd.read_csv("todo.csv")
  X = df["Sequence"]
  y = df["class"]

  ds = CustomDataset(df)
  loader = DataLoader(ds, shuffle=True, batch_size=16)

  train_loader = loader

  for data in train_loader:
    timber.debug(data)
    # xf, xb, y = data[0], data[1], data[2]
    # timber.debug(f"xf -> {xf.shape}, xb -> {xb.shape}, y -> {y.shape}")
  pass


def get_callbacks() -> list:
  # metric.auc ( uses trapezoidal rule) gave an error: x is neither increasing, nor decreasing. so I had to remove it
  return [
    ("tr_acc", EpochScoring(
      metrics.accuracy_score,
      lower_is_better=False,
      on_train=True,
      name="train_acc",
    )),

    ("tr_recall", EpochScoring(
      metrics.recall_score,
      lower_is_better=False,
      on_train=True,
      name="train_recall",
    )),
    ("tr_precision", EpochScoring(
      metrics.precision_score,
      lower_is_better=False,
      on_train=True,
      name="train_precision",
    )),
    ("tr_roc_auc", EpochScoring(
      metrics.roc_auc_score,
      lower_is_better=False,
      on_train=False,
      name="tr_auc"
    )),
    ("tr_f1", EpochScoring(
      metrics.f1_score,
      lower_is_better=False,
      on_train=False,
      name="tr_f1"
    )),
    # ("valid_acc1", EpochScoring(
    #   metrics.accuracy_score,
    #   lower_is_better=False,
    #   on_train=False,
    #   name="valid_acc1",
    # )),
    ("valid_recall", EpochScoring(
      metrics.recall_score,
      lower_is_better=False,
      on_train=False,
      name="valid_recall",
    )),
    ("valid_precision", EpochScoring(
      metrics.precision_score,
      lower_is_better=False,
      on_train=False,
      name="valid_precision",
    )),
    ("valid_roc_auc", EpochScoring(
      metrics.roc_auc_score,
      lower_is_better=False,
      on_train=False,
      name="valid_auc"
    )),
    ("valid_f1", EpochScoring(
      metrics.f1_score,
      lower_is_better=False,
      on_train=False,
      name="valid_f1"
    ))
  ]


def start():

  # df = pd.read_csv("data64.csv")  # use this line
  df = pd.read_csv("data64random.csv")
  X = df["Sequence"]
  y = df["class"]

  npa = np.array([y.values])

  torch_tensor = torch.tensor(npa)  # [0, 1, 1, 0, ... ... ] a simple list
  print(f"torch_tensor: {torch_tensor}")
  # need to transpose it!

  yt = torch.transpose(torch_tensor, 0, 1)

  ds = CustomDataset(df)
  loader = DataLoader(ds, shuffle=True)

  # train_loader = loader
  # test_loader = loader  # todo: load another dataset later

  device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
  model = CNN1D().to(device)
  m_criterion = nn.BCEWithLogitsLoss
  # optimizer = optim.Adam(model.parameters(), lr=0.001)
  m_optimizer = optim.Adam

  net = NeuralNetClassifier(
    model,
    max_epochs=200,
    criterion=m_criterion,
    optimizer=m_optimizer,
    lr=0.01,
    # decay=0.01,
    # momentum=0.9,

    device=device,
    classes=["no_mqtl", "yes_mqtl"],
    verbose=True,
    callbacks=get_callbacks()
  )

  ohe_c = one_hot_e_column(X)
  print(f"ohe_c shape {ohe_c.shape}")
  ohe_c = torch.Tensor(ohe_c)
  ohe_c = ohe_c.permute(0, 2, 1)
  ohe_c = ohe_c.to(device)
  print(f"ohe_c shape {ohe_c.shape}")

  net.fit(X=ohe_c, y=yt)
  y_proba = net.predict_proba(ohe_c)
  # timber.info(f"y_proba = {y_proba}")
  pass


if __name__ == '__main__':
  start()
  # test_dataloader()
  pass

你可以找到2个数据集

  1. dna64random.csv（模型不适用于此）
  2. dna64.csv（模型适用）

您可以使用此要点链接快速下载所有内容

scikit-learn
Martin Hope
Evan Aad
Asked: 2024-04-10 04:07:08 +0800 CST
  • 5

如何从嵌入 ColumnTransformer 中的 OneHotEncoder 获取功能名称？

下面这段代码：

import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import OneHotEncoder

df = pd.DataFrame(data = [[1],[2]], columns = ['C'])
ohe = OneHotEncoder(sparse_output = False)
transformer = ColumnTransformer(transformers = [('encoder', ohe, ['C'])])
transformer.fit_transform(df)
ohe.get_feature_names_out()

结果出现以下错误消息：

NotFittedError                            Traceback (most recent call last)
<ipython-input-1-46f41a3cf71a> in <cell line: 10>()
      8 transformer = ColumnTransformer(transformers = [('encoder', ohe, ['C'])])
      9 transformer.fit_transform(df)
---> 10 ohe.get_feature_names_out()

1 frames
/usr/local/lib/python3.10/dist-packages/sklearn/utils/validation.py in check_is_fitted(estimator, attributes, msg, all_or_any)
   1388 
   1389     if not fitted:
-> 1390         raise NotFittedError(msg % {"name": type(estimator).__name__})
   1391 
   1392 

NotFittedError: This OneHotEncoder instance is not fitted yet. Call 'fit' with appropriate arguments before using this estimator.

为什么？

scikit-learn