Что объединяет 1 февраля, 22 марта, 3 мая и 14 июня 2023 года? Независимо от их конкретной направленности на классы активов — облигации, акции, иностранную валюту, товары и т. д. — миллионы участников финансового рынка с нетерпением ждут одного и того же события: завершения заседания FOMC.
На каждом из этих мероприятий выпускается заявление и проводится пресс-конференция. Мы собираемся использовать методы обработки естественного языка (NLP), чтобы анализировать заявления FOMC и алгоритмически присваивать рейтинг настроений каждому из них. Наш метод включает анализ каждого утверждения на уровне предложений за предложением с последующим объединением этих оценок в составную оценку утверждения.
Учитывая количество необходимых процедур, мы разобьем их на несколько частей. В этой статье мы будем использовать дополнение текстовых данных с помощью библиотеки NLPAug с WordNet, чтобы решить проблему ограниченного набора обучающих данных. Мы также будем использовать трансферное обучение с помощью модели BERT.
В части 2 из 3 мы настроим модель GPT-3 для этой задачи классификации через API OpenAI. Мы также поделимся тем, как мы создаем метод для агрегирования рейтингов тональности на уровне предложений в оценку на уровне утверждений.
Наконец, в части 3 из 3 мы углубимся в то, как мы можем напрямую прогнозировать движение доходности 10-летних государственных облигаций США после публикации заявления FOMC по сравнению с уровнями предыдущего дня.
Подведение итогов набора данных
Заявления FOMC доступны на веб-сайте Федеральной резервной системы в форматах PDF и HTML. Извлечение текстовых данных из PDF и удаление из HTML выходит за рамки данной статьи. Заявления FOMC состоят из предложений; каждое утверждение состоит из различного количества предложений.
Наши обучающие данные представляют собой набор данных из 200 предложений, которые были предварительно помечены как голубиные и ястребиные. Ниже мы показываем пример некоторых предложений в категории настроений. По сути, это проблема машинного обучения с многоклассовой классификацией.
Приговоры были отмечены четырьмя рецензентами, имеющими опыт работы в области финансов и экономики, в рамках исследовательской работы (Хансен, А. и Казинник, С. (2023)) сотрудников Федеральной резервной системы. Хотя мы не можем поделиться базовым помеченным набором данных (они были предоставлены нам в рамках академического исследования), в нашей следующей статье мы покажем, как мы можем использовать финансовый рынок, чтобы помочь нам пометить наш набор данных; эта статья нацелена на то, чтобы поделиться основными методами НЛП, используемыми при анализе заявлений FOMC.
import pandas as pd
labelled_sentences = pd.read_excel('FOMC Labelled Sentences.xlsx')
labelled_sentences_aug = labelled_sentences.copy() #Create a copy to avoid altering the original data
labelled_sentences_aug
Добавление данных для дополнения ограниченного набора обучающих данных
Как правило, чем больше у нас данных, тем лучше мы можем обучать наши модели. Учитывая ограниченное количество предложений всего в 200, нам нужно искусственно увеличить наш набор обучающих данных. Перед увеличением данных мы должны принять к сведению разделение исходного набора данных на обучающие (80%) и тестовые компоненты (20%); расширение данных следует применять только к обучающему набору данных, а тестовый набор данных должен оставаться в чистом виде. Это сделано для того, чтобы в дальнейшем сохранить целостность точности нашей модели, тестовый набор данных должен состоять из предложений непосредственно из заявлений FOMC.
from sklearn.model_selection import train_test_split # Keep sentence and score columns; meeting_date is not relevant to this analysis labelled_sentences_aug['Score'] = labelled_sentences_aug['Score'].astype(str) labelled_sentences_aug = labelled_sentences_aug[['Sentence', 'Score']] ### Split the original dataset into train-test prior to augmentation ### train_aug, test_aug = train_test_split(labelled_sentences_aug, test_size=0.2, random_state=23, stratify=labelled_sentences_aug['Score'])
Наш подход к расширению данных включает технику замены на основе лексики Wordnet из библиотеки NLPAug; он определяет синонимы для токена/слов, которые должны быть заменены в исходных предложениях.
import nlpaug.augmenter.word as naw
# Note that this function elects to create 2 sets of augmented data on top of our original data
def augment_dataset(data, augmenter, num_augmented=2):
data_augmented = data.copy()
augmented_data = []
for index, row in data_augmented.iterrows():
for _ in range(num_augmented):
augmented_data.append([ ' '.join(augmenter.augment(row['Sentence'])), row['Score']])
augmented_data_df = pd.DataFrame(augmented_data, columns=['Sentence', 'Score'])
return augmented_data_df
aug = naw.SynonymAug(aug_src='wordnet') # Initialize the augmenter
augmented_data = augment_dataset(train_aug, aug)
augmented_data
Учитывая, что наш исходный набор данных составляет 200 предложений, наш обучающий набор данных составляет 80% от этого и состоит из 160 предложений. Мы создали расширенный набор данных, в два раза превышающий исходный набор данных для обучения (320 предложений), чтобы дополнить наш исходный набор данных для обучения. Это доводит наш общий расширенный набор данных для обучения до 480 предложений.
def combine_datasets(original_data, augmented_data):
# Combine the original dataset and the augmented dataset into a single dataframe
combined_data = pd.concat([original_data, augmented_data], ignore_index=True)
return combined_data
train_augmented_aug = combine_datasets(train_aug, augmented_data)
train_augmented_aug
Причина, по которой мы преодолеваем все эти трудности, чтобы увеличить набор данных, заключается в том, что это улучшает производительность модели. Эти характеристики точности показывают производительность точно настроенной модели GPT3 Ada с различными методами увеличения данных; Подстановка слов WordNet, которую мы использовали выше, дала наилучшее повышение точности по сравнению с базовой производительностью без увеличения данных (73% против 62%).
Применение модели BERT
Первая модель, которую мы продемонстрируем для использования, — это Представления двунаправленного кодировщика от преобразователей (BERT), где мы будем использовать передачу обучения из модели на основе bert-uncased. Модель BERT была предварительно обучена на BookCorpus, наборе данных из 11 038 неопубликованных книг и английской Википедии. Модель bert-base-uncased, с которой мы работаем, имеет 110 миллионов параметров.
import torch
from torch.utils.data import TensorDataset, DataLoader, RandomSampler, SequentialSampler
from keras.utils import pad_sequences
from sklearn.model_selection import train_test_split
from transformers import BertTokenizer, BertForSequenceClassification, AdamW, get_linear_schedule_with_warmup
import pandas as pd
import numpy as np
from collections import defaultdict
import random
import time
import datetime
from sklearn.metrics import f1_score, balanced_accuracy_score, confusion_matrix, precision_score, recall_score, classification_report
from torch.nn import functional as F
class MCBert(BertForSequenceClassification):
def train(self):
self.training = True
def eval(self):
self.training = True
class BERTSentimentClassifier:
def __init__(self, model_name='bert-base-uncased', num_labels=6, device='cuda', max_length=100):
self.device = torch.device(device if torch.cuda.is_available() else 'cpu')
self.tokenizer = BertTokenizer.from_pretrained(model_name, do_lower_case=True)
self.model = MCBert.from_pretrained(
model_name,
num_labels = num_labels,
output_attentions = False,
output_hidden_states = False
).to(self.device)
self.max_length = max_length
def tokenize(self, sentences):
input_ids, attention_masks = [], []
for sent in sentences:
encoded_dict = self.tokenizer.encode_plus(
sent,
add_special_tokens=True,
max_length=self.max_length,
padding='max_length',
truncation=True,
return_attention_mask=True,
return_tensors='pt',
)
input_ids.append(encoded_dict['input_ids'])
attention_masks.append(encoded_dict['attention_mask'])
input_ids = torch.cat(input_ids, dim=0)
attention_masks = torch.cat(attention_masks, dim=0)
return input_ids, attention_masks
def prepare_data(self, df, test_size=0.2, batch_size=32):
x_train, x_test, y_train, y_test = train_test_split(df['Sentence'], df['Score'], test_size=test_size, random_state=23)
score_mapping = {'-1': 0, '-0.5': 1, '0': 2, '0.5': 3, '1': 4, 'Remove': 5}
y_train = y_train.map(score_mapping)
y_test = y_test.map(score_mapping)
train_input_ids, train_attention_mask = self.tokenize(x_train.values.tolist())
dev_input_ids, dev_attention_mask = self.tokenize(x_test.values.tolist())
train_labels = torch.tensor(y_train.values)
# train_labels = F.one_hot(train_labels, num_classes=6)
dev_labels = torch.tensor(y_test.values)
# dev_labels = F.one_hot(dev_labels, num_classes=6)
train_data = TensorDataset(train_input_ids, train_attention_mask, train_labels)
train_sampler = RandomSampler(train_data, replacement=False, generator=torch.Generator().manual_seed(23))
self.train_dataloader = DataLoader(train_data, sampler=train_sampler, batch_size=batch_size)
dev_data = TensorDataset(dev_input_ids, dev_attention_mask, dev_labels)
dev_sampler = SequentialSampler(dev_data)
self.dev_dataloader = DataLoader(dev_data, sampler=dev_sampler, batch_size=batch_size)
return self
def train(self, epochs, learning_rate=5e-5, eps=1e-8):
optimizer = AdamW(self.model.parameters(), lr=learning_rate, eps=eps)
total_steps = len(self.train_dataloader) * epochs
# Create the learning rate scheduler
scheduler = get_linear_schedule_with_warmup(optimizer, num_warmup_steps=0, num_training_steps=total_steps)
for epoch_i in range(0, epochs):
print('\n======== Epoch {:} / {:} ========'.format(epoch_i + 1, epochs))
t0 = time.time()
total_train_loss = 0
self.model.train()
for step, batch in enumerate(self.train_dataloader):
b_input_ids = batch[0].to(self.device)
b_input_mask = batch[1].to(self.device)
b_labels = batch[2].to(self.device)
self.model.zero_grad()
outputs = self.model(b_input_ids,
token_type_ids=None,
attention_mask=b_input_mask,
labels=b_labels)
loss = outputs.loss
total_train_loss += loss.item()
loss.backward()
torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0)
optimizer.step()
scheduler.step()
avg_train_loss = total_train_loss / len(self.train_dataloader)
print("\n Average training loss: {0:.2f}".format(avg_train_loss))
print(" Training epoch took: {:}".format(self.format_time(time.time() - t0)))
t0 = time.time()
self.model.eval()
total_eval_accuracy = 0
total_eval_loss = 0
correct_preds_per_class = defaultdict(int)
total_preds_per_class = defaultdict(int)
for batch in self.dev_dataloader:
b_input_ids = batch[0].to(self.device)
b_input_mask = batch[1].to(self.device)
b_labels = batch[2].to(self.device)
with torch.no_grad():
outputs = self.model(b_input_ids,
token_type_ids=None,
attention_mask=b_input_mask,
labels=b_labels)
loss = outputs.loss
logits = outputs.logits
total_eval_loss += loss.item()
logits = logits.detach().cpu().numpy()
label_ids = b_labels.to('cpu').numpy()
pred_flat = np.argmax(logits, axis=1).flatten()
for label in np.unique(label_ids):
correct_preds_per_class[label] += np.sum((pred_flat == label_ids) & (label_ids == label))
total_preds_per_class[label] += np.sum(label_ids == label)
total_eval_accuracy += self.flat_accuracy(logits, label_ids)
avg_val_accuracy = total_eval_accuracy / len(self.dev_dataloader)
print("\n Accuracy: {0:.2f}".format(avg_val_accuracy))
avg_val_loss = total_eval_loss / len(self.dev_dataloader)
for label in total_preds_per_class.keys():
accuracy = correct_preds_per_class[label] / total_preds_per_class[label]
print(f"Accuracy for class {label}: {accuracy:.2f}")
print(" Validation Loss: {0:.2f}".format(avg_val_loss))
print(" Validation took: {:}".format(self.format_time(time.time() - t0)))
print("\nTraining complete!")
return self
def predict(self, test):
test_input_ids, test_attention_mask = self.tokenize(test['Sentence'].values.tolist())
# Convert inputs to tensors
test_inputs = torch.tensor(test_input_ids)
test_masks = torch.tensor(test_attention_mask)
# Create DataLoader for the test data
batch_size = 32
test_data = TensorDataset(test_inputs, test_masks)
test_sampler = SequentialSampler(test_data)
test_dataloader = DataLoader(test_data, sampler=test_sampler, batch_size=batch_size)
# For the BERT model, we keep it in train mode to enable dropout
self.model.train()
predicted_scores = []
uncertainties = []
n_mc_samples = 30
for batch in test_dataloader:
batch = tuple(t.to(self.device) for t in batch)
b_input_ids, b_input_mask = batch
mc_samples = []
for _ in range(n_mc_samples):
with torch.no_grad():
outputs = self.model(b_input_ids, token_type_ids=None, attention_mask=b_input_mask)
logits = outputs[0]
logits = logits.detach().cpu().numpy()
mc_samples.append(logits)
mc_samples = np.array(mc_samples)
predicted_score = mc_samples.mean(axis=0)
uncertainty = mc_samples.std(axis=0)
predicted_scores.extend(np.argmax(predicted_score, axis=1).flatten())
uncertainties.extend(uncertainty.max(axis=1).flatten())
score_mapping = {'-1': 0, '-0.5': 1, '0': 2, '0.5': 3, '1': 4, 'Remove': 5}
inverse_class_mapping = {v: k for k, v in score_mapping.items()}
# Inverse map the classes to their original values
predicted_scores = np.vectorize(inverse_class_mapping.get)(predicted_scores)
accuracy = np.mean(predicted_scores == test['Score'].values)
print(f'Test Accuracy: {accuracy*100:.2f}%')
# Add the original sentences, their predicted scores and uncertainties to a DataFrame
predictions_df = pd.DataFrame({'Sentence': test['Sentence'], 'Ground Truth': test['Score'], 'Predicted_Score': predicted_scores, 'Uncertainty': uncertainties})
return predictions_df
def predict_scores(self, test):
test_input_ids, test_attention_mask = self.tokenize(test['Sentence'].values.tolist())
# Convert inputs to tensors
test_inputs = test_input_ids.clone().detach()
test_masks = test_attention_mask.clone().detach()
# Create DataLoader for the test data
batch_size = 32
test_data = TensorDataset(test_inputs, test_masks)
test_sampler = SequentialSampler(test_data)
test_dataloader = DataLoader(test_data, sampler=test_sampler, batch_size=batch_size)
# For the BERT model, we keep it in train mode to enable dropout
self.model.train()
predicted_scores = []
uncertainties = []
n_mc_samples = 30
for batch in test_dataloader:
batch = tuple(t.to(self.device) for t in batch)
b_input_ids, b_input_mask = batch
mc_samples = []
for _ in range(n_mc_samples):
with torch.no_grad():
outputs = self.model(b_input_ids, token_type_ids=None, attention_mask=b_input_mask)
logits = outputs[0]
logits = logits.detach().cpu().numpy()
mc_samples.append(logits)
mc_samples = np.array(mc_samples)
predicted_score = mc_samples.mean(axis=0)
uncertainty = mc_samples.std(axis=0)
predicted_scores.extend(np.argmax(predicted_score, axis=1).flatten())
uncertainties.extend(uncertainty.max(axis=1).flatten())
score_mapping = {'-1': 0, '-0.5': 1, '0': 2, '0.5': 3, '1': 4, 'Remove': 5}
inverse_class_mapping = {v: k for k, v in score_mapping.items()}
# Inverse map the classes to their original values
predicted_scores = np.vectorize(inverse_class_mapping.get)(predicted_scores)
return predicted_scores
def compute_metrics(self, y_true, predicted_scores):
# Convert arrays to string type
y_true_str = y_true
predicted_scores_str = predicted_scores
# Compute F1 score and Balanced Accuracy
f1 = f1_score(y_true_str, predicted_scores_str, average='weighted')
balanced_accuracy = balanced_accuracy_score(y_true_str, predicted_scores_str)
# Compute Precision and Recall
precision = precision_score(y_true_str, predicted_scores_str, average='weighted')
recall = recall_score(y_true_str, predicted_scores_str, average='weighted')
# Compute Confusion Matrix
confusion_mat = confusion_matrix(y_true_str, predicted_scores_str)
# Compute classification report
class_report = classification_report(y_true_str, predicted_scores_str)
# Print the scores
print(f'F1 Score: {f1*100:.2f}%')
print(f'Balanced Accuracy: {balanced_accuracy*100:.2f}%')
print(f'Precision: {precision*100:.2f}%')
print(f'Recall: {recall*100:.2f}%')
print('Classification Report:')
print(class_report)
return {
"F1 Score": f1,
"Balanced Accuracy": balanced_accuracy,
"Precision": precision,
"Recall": recall,
"Confusion Matrix": confusion_mat,
"Classification Report": class_report
}
@staticmethod
def flat_accuracy(preds, labels):
pred_flat = np.argmax(preds, axis=1).flatten()
labels_flat = labels.flatten()
return np.sum(pred_flat == labels_flat) / len(labels_flat)
@staticmethod
def flat_accuracy_per_class(preds, labels):
pred_flat = np.argmax(preds, axis=1).flatten()
labels_flat = labels.flatten()
correct_per_class = defaultdict(int)
total_per_class = defaultdict(int)
for pred, label in zip(pred_flat, labels_flat):
if pred == label:
correct_per_class[label] += 1
total_per_class[label] += 1
accuracies_per_class = {label: correct / total for label, correct, total in zip(correct_per_class.keys(), correct_per_class.values(), total_per_class.values())}
return accuracies_per_class
@staticmethod
# Function to format time
def format_time(elapsed):
elapsed_rounded = int(round((elapsed)))
return str(datetime.timedelta(seconds=elapsed_rounded))
Приведенный выше фрагмент содержит код класса BERTSentimentClassifier, который мы будем вызывать для обучения нашего корпуса текстов. Осталось только вызвать модель и обучить ее! Обратите внимание, что число меток равно 6, поскольку оно включает +1, +0,5, 0, -0,5, -1 и «неизвестно» для предложений, не содержащих значения тональности.
model_aug = BERTSentimentClassifier(model_name='bert-base-uncased', num_labels=6, device='cuda', max_length=200) model_aug.prepare_data(train_aug, test_size=0.2, batch_size=32) model_aug.train(epochs=12, learning_rate=5e-5, eps=1e-8)
После того, как модель будет обучена, мы применим ее к нашему тестовому набору данных, который не подвергается дополнению данных, и модель также не использовалась ранее, что исключает утечку данных. И это все! Точность нашего теста составляет 57,5%. В нашей следующей статье, часть 2 из 3, вы можете увидеть, как тонко настроенный GPT-3 способен значительно превзойти это. На момент написания статьи исходные модели GPT-3 в настоящее время являются единственными, доступными для тонкой настройки.
Это завершает часть 1 из 3, где мы использовали увеличение текстовых данных, чтобы дополнить наш набор обучающих данных, а также использовали трансферное обучение с нашей моделью BERT. Если эта статья принесла вам какую-то пользу, рассмотрите возможность подписаться на меня на Medium, чтобы узнать больше!
Ссылки
Хансен А. и Казинник С. (2023). Может ли ChatGPT расшифровать федеральный язык? Доступно онлайн по адресу https://papers.ssrn.com/sol3/papers.cfm?abstract_id=4399406
