Если ваши часы следят за тем, как вы едите…

Начав с простого устройства для измерения времени, часы превратились в незаменимые модные аксессуары как для мужчин, так и для женщин. Однако с появлением смарт-часов и носимых технологий часы взяли на себя новую роль — «наблюдать» за людьми.

В этой статье я покажу вам, как шаг за шагом обучить ваши драгоценные умные часы или любое другое носимое устройство, которое может записывать ваши физические данные на машину наблюдения. Здесь, в частности, применяются контролируемые алгоритмы машинного обучения (логистическая регрессия и нейронные сети) для отслеживания вашего ежедневного статуса питания. При этом ваши часы будут знать, когда вы едите в течение дня.

Сбор данных

Устройство, которое я использовал для сбора своих физиологических данных, — это Empatica E4, носимое устройство, которое в основном используется в исследовательских целях для сбора цифровых биомаркеров. Он собирал информацию о частоте сердечных сокращений (HR), пульсе объема крови (BVP), электрокожной активности (EDA), температуре (TEMP) и данных акселерометра (ACC). Здесь я отброшу данные ACC, потому что во время приема пищи я не двигаюсь активно. График приема пищи такой: прием пищи в течение 2 минут; отдых 2 минуты; прием пищи за 2 минуты; отдых 2 минуты; есть еду до конца. В общей сложности я собрал 253 715 точек данных, учитывая, что носимое устройство имеет частоту дискретизации 64 Гц.

Предварительная обработка данных

Исходные необработанные данные выглядят так:

Первый индекс — это текущее время в секундах с начала эпохи, а второй — частота дискретизации.

Во-первых, я объединяю три сеанса данных вместе путем конкатенации. Некоторые функции собраны в разных герцах, таких как EDA с частотой дискретизации 4 Гц (как показано на рисунке выше) и HR с частотой дискретизации 1 Гц.

import tensorflow as tf
import pandas as pd
import matplotlib
from matplotlib import pyplot as plt
import seaborn as sns
import sklearn.metrics as sk_metrics
import tempfile
import os
import numpy as np

# My original TEMP file is named as TEMP.csv
file_name = 'TEMP.csv'
# My final TEMP file name:
export_name = 'TEMP64_02.csv'

# Desired sampling rate is 64
target_hz = 64
df = pd.read_csv(file_name)

# In my file, the first cell gives the sampling rate
current_hz = df.iloc[0,0]

# This gives how much the ratio of current and desired Hz
multiplication_ratio = target_hz / current_hz
ur = multiplication_ratio

column_values = df.iloc[1:,:].to_numpy()

# This for loop repeately insert value to fill up the
# undersampling gap
for col in range(np.shape(column_values)[1]):
  curr_col = column_values[:,col]
  new_col = np.repeat(curr_col, ur)
  
  new_df[curr_df_headers[col]] = new_col

# adding a row with value of Hz and set index to be -1
new_df.loc[-1] = target_hz 
# shifting index by 1
new_df.index = new_df.index + 1

Используя приведенный выше код, мы можем повысить частоту дискретизации любых данных ниже 64 Гц ровно до 64 Гц.

Помните, что мы используем контролируемые методы машинного обучения, поэтому для обучения наших данных нам требуется метка истинности. Таким образом, следующим шагом является присвоение метки истинности каждой из наших точек данных. Я не буду вдаваться в подробности, потому что я предпочел бы оставить больше времени для изучения методов обучения машинному обучению. Подсказка: вы можете просто записать временной интервал, в течение которого вы активно едите, а затем присвоить метки «Истина» или «Ложь» (1/0) рядом с каждой точкой в ​​качестве нового столбца в файле CSV.

Чтобы объединить все сеансы данных:

df1 = pd.read_csv('session1.csv', index_col=False)
df2 = pd.read_csv('session2.csv',index_col=False)
df3 = pd.read_csv('session3.csv',index_col=False)
frames = [df1, df2, df3]
dataset = pd.concat(frames, ignore_index=True)

Вот небольшая часть того, как выглядит наш предварительно обработанный набор данных:

# A glimpse of data information

dataset.info()

Следующим шагом, который нам нужно будет сделать, является разработка фильтра скользящего среднего для временного разделения. Без него обучение будет бессмысленным, потому что мы хотим, чтобы наша модель могла предсказывать наш пищевой статус по равномерным интервалам, а не по ряду моментов времени.

# Here we get half size of data to mimic the stride of 2
# and using rolling(window=32) function to define a time space of 0.5s
dataset.loc[dataset.index[np.arange(len(dataset))%2==1],'hr64_w_s_32']=dataset.hr.rolling(window=32).mean()
dataset.loc[dataset.index[np.arange(len(dataset))%2==1],'bvp64_w_s_32']=dataset.bvp.rolling(window=32).mean()
dataset.loc[dataset.index[np.arange(len(dataset))%2==1],'eda64_w_s_32']=dataset.eda.rolling(window=32).mean()
dataset.loc[dataset.index[np.arange(len(dataset))%2==1],'temp64_w_s_32']=dataset.temp.rolling(window=32).mean()

# Creat a new dataframe and drop rows containing NaN
clean_dataset = dataset[['hr64_w_s_32', 'bvp64_w_s_32', 'eda64_w_s_32', 'temp64_w_s_32', 'eating_status']]
clean_dataset.dropna(inplace=True)
clean_dataset

Прежде чем начать наш процесс обучения, нам нужно разделить данные на обучающие наборы (80% данных) и тестовые наборы (20% данных) и определить несколько функций оценки. Поскольку наши переменные имеют разные значения, необходима нормализация. Он преобразует функции, чтобы они были в аналогичном масштабе.

train_dataset = dataset[0:int(len(clean_dataset)*0.80)]
test_dataset = dataset.drop(train_dataset.index)

# set x to be the variable columns and set y to be ground truth column
x_train, y_train = train_dataset.iloc[:, [0,1,2,3]], train_dataset.iloc[:,4]
x_test, y_test = test_dataset.iloc[:, [0,1,2,3]], test_dataset.iloc[:,4]

# convert to trainable tensors
x_train, y_train = tf.convert_to_tensor(x_train, dtype=tf.float32), tf.convert_to_tensor(y_train, dtype=tf.float32)
x_test, y_test = tf.convert_to_tensor(x_test, dtype=tf.float32), tf.convert_to_tensor(y_test, dtype=tf.float32)


class Normalize(tf.Module):
  def __init__(self, x):
    # Initialize the mean and standard deviation for normalization
    self.mean = tf.Variable(tf.math.reduce_mean(x, axis=0))
    self.std = tf.Variable(tf.math.reduce_std(x, axis=0))

  def norm(self, x):
    # Normalize the input
    return (x - self.mean)/self.std

  def unnorm(self, x):
    # Unnormalize the input
    return (x * self.std) + self.mean

norm_x = Normalize(x_train)
x_train_norm, x_test_norm = norm_x.norm(x_train), norm_x.norm(x_test)

def log_loss(y_pred, y):
  # Compute the log loss function
  ce = tf.nn.sigmoid_cross_entropy_with_logits(labels=y, logits=y_pred)
  return tf.reduce_mean(ce)

# define logistic regression moduels
class LogisticRegression(tf.Module):

  def __init__(self):
    self.built = False

  def __call__(self, x, train=True):
    # Initialize the model parameters on the first call
    if not self.built:
      # Randomly generate the weights and the bias term
      rand_w = tf.random.uniform(shape=[x.shape[-1], 1], seed=22)
      rand_b = tf.random.uniform(shape=[], seed=22)
      self.w = tf.Variable(rand_w)
      self.b = tf.Variable(rand_b)
      self.built = True
    # Compute the model output
    z = tf.add(tf.matmul(x, self.w), self.b)
    z = tf.squeeze(z, axis=1)
    if train:
      return z
    return tf.sigmoid(z)

log_reg = LogisticRegression()


def predict_class(y_pred, thresh=0.5):
  # Return a tensor with  `1` if `y_pred` > `0.5`, and `0` otherwise
  return tf.cast(y_pred > thresh, tf.float32)

def accuracy(y_pred, y):
  # Return the proportion of matches between `y_pred` and `y`
  y_pred = tf.math.sigmoid(y_pred)
  y_pred_class = predict_class(y_pred)
  check_equal = tf.cast(y_pred_class == y,tf.float32)
  acc_val = tf.reduce_mean(check_equal)
  return acc_val



batch_size = 64
train_dataset = tf.data.Dataset.from_tensor_slices((x_train_norm, y_train))
train_dataset = train_dataset.shuffle(buffer_size=x_train.shape[0]).batch(batch_size)
test_dataset = tf.data.Dataset.from_tensor_slices((x_test_norm, y_test))
test_dataset = test_dataset.shuffle(buffer_size=x_test.shape[0]).batch(batch_size)

С набором для обучения и набором для тестирования, определенными выше, мы готовы обучить наши невинные умные часы маленькому злобному сталкеру.

Обучение данным

1. Логистическая регрессия

Наша цель — максимально уменьшить функцию потерь. Поэтому мы используем функцию выше, чтобы определить нашу функцию потерь. Y со шляпой — это наше предсказанное значение, а y без шляпы — это истинное значение.

Подставив y в простое линейное уравнение Xw+b, мы получим новую функцию потерь, как указано выше.

Функции градиентного спуска для логистической регрессии определены выше. Частичные функции используются для обновления параметров в новой итерации.

# Set training parameters
from tqdm import tqdm
epochs = 200
learning_rate = 0.01
train_losses, test_losses = [], []
train_accs, test_accs = [], []

# Set up the training loop and begin training
for epoch in tqdm(range(epochs)):
  batch_losses_train, batch_accs_train = [], []
  batch_losses_test, batch_accs_test = [], []

  # Iterate over the training data
  for x_batch, y_batch in train_dataset:
    with tf.GradientTape() as tape:
      y_pred_batch = log_reg(x_batch)
      batch_loss = log_loss(y_pred_batch, y_batch)
    batch_acc = accuracy(y_pred_batch, y_batch)
    # Update the parameters with respect to the gradient calculations
    grads = tape.gradient(batch_loss, log_reg.variables)
    for g,v in zip(grads, log_reg.variables):
      v.assign_sub(learning_rate * g)
    # Keep track of batch-level training performance
    batch_losses_train.append(batch_loss)
    batch_accs_train.append(batch_acc)

  # Iterate over the testing data
  for x_batch, y_batch in test_dataset:
    y_pred_batch = log_reg(x_batch)
    batch_loss = log_loss(y_pred_batch, y_batch)
    batch_acc = accuracy(y_pred_batch, y_batch)
    # Keep track of batch-level testing performance
    batch_losses_test.append(batch_loss)
    batch_accs_test.append(batch_acc)

  # Keep track of epoch-level model performance
  train_loss, train_acc = tf.reduce_mean(batch_losses_train), tf.reduce_mean(batch_accs_train)
  test_loss, test_acc = tf.reduce_mean(batch_losses_test), tf.reduce_mean(batch_accs_test)
  train_losses.append(train_loss)
  train_accs.append(train_acc)
  test_losses.append(test_loss)
  test_accs.append(test_acc)
  if epoch % 20 == 0:
    print(f"Epoch: {epoch}, Training log loss: {train_loss:.3f}")

Мы установили эпохи на 200 и скорость обучения на 0,01 в нашем первом испытании. Поскольку наш набор данных больше, мы ожидаем длительного времени обучения. Если вы хотите тренировать его быстрее, не стесняйтесь уменьшать эпохи и увеличивать скорость обучения. Однако вы также должны получить возможную более низкую точность в своем собственном наборе данных. Для получения дополнительной информации о фундаментальном алгоритме логистической регрессии см.: Логистическая регрессия для бинарной классификации с помощью Core API.

Чтобы просмотреть наши результаты, просто используйте пакет matplotlib для отображения потерь и точности:

plt.plot(range(epochs), train_losses, label = "Training loss")
plt.plot(range(epochs), test_losses, label = "Testing loss")
plt.xlabel("Epoch")
plt.ylabel("Log loss")
plt.legend()
plt.title("Log loss vs training iterations");

plt.plot(range(epochs), train_accs, label = "Training accuracy")
plt.plot(range(epochs), test_accs, label = "Testing accuracy")
plt.xlabel("Epoch")
plt.ylabel("Accuracy (%)")
plt.legend()
plt.title("Accuracy vs training iterations");

print(f"Final training log loss: {train_losses[-1]:.3f}")
print(f"Final testing log Loss: {test_losses[-1]:.3f}")
print(f"Final training accuracy: {train_accs[-1]:.3f}")
print(f"Final testing accuracy: {test_accs[-1]:.3f}")
Final training log loss: 0.570
Final testing log Loss: 0.743
Final training accuracy: 0.701
Final testing accuracy: 0.594

В результате эта модель могла приблизительно предсказывать результат, но не очень хорошо, и быстро достигала переобучения в эпоху числа 2. Возможно, потому, что признак данных нуждается в лучшей функции потерь, чтобы классифицировать карту признаков по меткам истинности, модель могла бы только результат около 59% точности. Другая причина может заключаться в том, что логистическая регрессия требует, чтобы независимая переменная не имела мультиколлинеарности или имела очень небольшую мультиколлинеарность для каждого признака. В нашем случае весьма вероятно, что 4 функции реагируют на сигналы приема пищи сходным образом. Так как это не дает большого результата, давайте попробуем более сложную модель.

2. Нейронные сети (NN)

Сначала мы предварительно обрабатываем данные, назначая обучающий набор 80% от общего объема данных, а тестовый набор — 20%. Использование StandardScaler для масштабирования данных, готовых к обучению машинному обучению.

from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
import tensorflow as tf
import math

df = clean_dataset

X = df.drop('eating_status', axis=1)
y = df['eating_status']

X_train, X_test, y_train, y_test = train_test_split(
    X, y, 
    test_size=0.2, random_state=42
)

scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)

Общая настройка NN начинается с назначения весов w и смещения b. Каждое оцененное значение будет передано в функцию активации для увеличения сложности, которая была бы невозможна при использовании простых алгоритмов линейной регрессии или логистической регрессии.

Оцененное значение y шляпа будет сравниваться с наземной меткой истинности y и вычислять разницу и обновлять скорость обучения в зависимости от того, насколько велико значение потери. Здесь мы используем ту же логарифмическую потерю из логистической регрессии, что и наша функция потерь. Это можно определить из встроенного пакета TensorFlow keras.losses.binary_crossentropy.

Слой NN установлен на 128x256x256. Ясно, что слой слишком сложный и есть риск переобучения. Чтобы узнать больше о переобучении, вы можете перейти на этот веб-сайт: Полное руководство по предотвращению переобучения в нейронных сетях. Не стесняйтесь изменять размерность NN и попробуйте свой собственный набор данных, чтобы лучше понять, как размерность слоя может повлиять на выходные данные. (Здесь предлагается отличное анимированное объяснение того, как работает слой с веб-сайта Tensorflow: Слои нейронной сети от Tensorflow)

import tensorflow as tf
tf.random.set_seed(42)


model = tf.keras.Sequential([
    tf.keras.layers.Dense(128, activation='relu'),
    tf.keras.layers.Dense(256, activation='relu'),
    tf.keras.layers.Dense(256, activation='relu'),
    tf.keras.layers.Dense(1, activation='sigmoid')
])

model.compile(
    loss=tf.keras.losses.binary_crossentropy,
    optimizer=tf.keras.optimizers.Adam(lr=0.03),
    metrics=[
        tf.keras.metrics.BinaryAccuracy(name='accuracy'),
        tf.keras.metrics.Precision(name='precision'),
        tf.keras.metrics.Recall(name='recall')
    ]
)

history = model.fit(X_train_scaled, y_train, validation_split=0.2, batch_size=32, epochs=100)

После шага, описанного выше, у вас должна быть под рукой надежная обученная модель. Теперь давайте посмотрим, как наша модель работает с некоторыми инструментами визуализации:

from matplotlib import rcParams

rcParams['figure.figsize'] = (18, 8)
rcParams['axes.spines.top'] = False
rcParams['axes.spines.right'] = False

plt.plot(
    np.arange(1, 101), 
    history.history['loss'], label='Training Loss'
)
plt.plot(
    np.arange(1, 101), 
    history.history['accuracy'], label='Training Accuracy'
)
plt.plot(
    np.arange(1, 101), 
    history.history['val_loss'], label='Validation Loss'
)
plt.plot(
    np.arange(1, 101), 
    history.history['val_accuracy'], label='Validation Accuracy'
)
plt.title('Evaluation metrics', size=20)
plt.xlabel('Epoch', size=14)
plt.legend();

График потерь и точности показывает на удивление хорошие результаты. Точность приближается к 100% с увеличением эпох. Модель не переоснащается, потому что потери при проверке все еще уменьшаются, а точность приближается к 100%. Модель выглядит многообещающе. Итак, давайте проверим нашу модель с помощью нашего тестового набора и посмотрим, как она работает:

# simply using predict methos to feed in our testing set
predictions = model.predict(X_test_scaled)

# The predicted values are simple numbers and we need to classify each number
# based on the value. If it is larger than 0.5, we classify it to positive(True),
# otherwise it is negative(False).
prediction_classes = [
    1 if prob > 0.5 else 0 for prob in np.ravel(predictions)
]

Мы также могли бы использовать метрики путаницы, чтобы оценить нашу производительность и дать нам лучшее представление о кривой AUC/ROC (вы можете попробовать ее самостоятельно с помощью этого веб-сайта: Как использовать кривые ROC и кривые Precision-Recall для классификации. на Питоне»):

from sklearn.metrics import confusion_matrix
print(confusion_matrix(y_test, prediction_classes))
[[14869   368]
 [  240  9892]]
from sklearn.metrics import accuracy_score, precision_score, recall_score

print(f'Accuracy: {accuracy_score(y_test, prediction_classes):.2f}')
print(f'Precision: {precision_score(y_test, prediction_classes):.2f}')
print(f'Recall: {recall_score(y_test, prediction_classes):.2f}')
Accuracy: 0.98
Precision: 0.96
Recall: 0.98

Из приведенного выше результата мы могли видеть огромное увеличение производительности наших моделей логистической регрессии. Вы также можете попробовать изменить сложность сетей и увидеть увеличение/уменьшение производительности и найти лучшую модель, подходящую для вашего набора данных.

Заключение

На этом мы завершаем этот урок о тренировке состояния питания на основе ваших собственных данных. Из набора данных можно настроить очень многое, например размер окна фильтра скользящего среднего, длину шага, оптимизаторы, количество эпох, размеры пакетов и т. д.

Это моя первая статья, размещенная на Medium, и я надеюсь, что она была достаточно полезной. Если у вас есть какие-либо сомнения или опасения, вы можете напрямую прокомментировать эту статью, и я буду более чем счастлив обсудить ее.

Ресурсы

[1] Логистическая регрессия для бинарной классификации с помощью Core API

[2] Нежное введение в сигмовидную функцию

[3] Полное руководство по предотвращению переобучения в нейронных сетях (часть 1)

[4] Уровни нейронной сети от Tensorflow

[5] Как использовать кривые ROC и кривые Precision-Recall для классификации в Python