{ "cells": [ { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "import pickle" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def is_bearish_candle(candle):\n", " start, end, high, low = candle\n", " return (end < start) \n", "\n", "def is_bullish_candle(candle):\n", " start, end, high, low = candle\n", " return (end > start)\n", "\n", "def is_doji(candle):\n", " start, end, high, low = candle\n", " diff = abs(end - start)\n", " return (10 * diff < (high - low)) and (abs(high - max(end, start)) > 10 * diff or abs(low - min(end, start)) > 10 * diff)\n", "\n", "def is_long_legged_doji(candle):\n", " return is_doji(candle) # long legged doji is characterized primarily by its long shadows\n", "\n", "def is_dragonfly_doji(candle):\n", " start, end, high, low = candle\n", " return is_doji(candle) and (high - max(end, start)) < (3 * abs(end - start))\n", "\n", "def is_gravestone_doji(candle):\n", " start, end, high, low = candle\n", " return is_doji(candle) and (min(end, start) - low) < (3 * abs(end - start))\n", "\n", "def is_hammer(candle):\n", " start, end, high, low = candle\n", " diff = abs(end - start)\n", " return (high - max(end, start)) < (3 * diff) and (low - min(end, start)) > (3 * diff)\n", "\n", "def is_inverted_hammer(candle):\n", " start, end, high, low = candle\n", " diff = abs(end - start)\n", " return (low - min(end, start)) < (3 * diff) and (high - max(end, start)) > (3 * diff)\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def did_candle_go_up(sequence):\n", " # Return True if the last day's candle went up, False otherwise\n", " return sequence[2][1] > sequence[1][1]\n", "\n", "def analyze_patterns(sequences):\n", " results = {}\n", " up_counts = {} # Stores the number of times the last day's candle went up for each pattern\n", " total_counts = {} # Stores the total number of occurrences of each pattern\n", " \n", " for seq in sequences:\n", " if len(seq) != 3:\n", " continue\n", "\n", " pattern_name = None\n", "\n", " # Bearish Candle -> 3 different classes of dogi\n", " if is_bearish_candle(seq[0]):\n", " if is_long_legged_doji(seq[1]):\n", " pattern_name = 'Bearish -> Long-legged Doji'\n", "\n", " print(seq)\n", " \n", " elif is_dragonfly_doji(seq[1]):\n", " pattern_name = 'Bearish -> Dragonfly Doji'\n", " elif is_gravestone_doji(seq[1]):\n", " pattern_name = 'Bearish -> Gravestone Doji'\n", " \n", " # Hammer\n", " if is_hammer(seq[1]):\n", " pattern_name = 'Bearish -> Hammer'\n", " elif is_inverted_hammer(seq[1]):\n", " pattern_name = 'Bearish -> Inverted Hammer'\n", " \n", " # Bullish Candle -> 3 different classes of dogi\n", " elif is_bullish_candle(seq[0]):\n", " if is_long_legged_doji(seq[1]):\n", " pattern_name = 'Bullish -> Long-legged Doji'\n", " elif is_dragonfly_doji(seq[1]):\n", " pattern_name = 'Bullish -> Dragonfly Doji'\n", " elif is_gravestone_doji(seq[1]):\n", " pattern_name = 'Bullish -> Gravestone Doji'\n", " \n", " # Hammer\n", " if is_hammer(seq[1]):\n", " pattern_name = 'Bullish -> Hammer'\n", " elif is_inverted_hammer(seq[1]):\n", " pattern_name = 'Bullish -> Inverted Hammer'\n", "\n", " # Update the counts based on whether the last day's candle went up\n", " if pattern_name:\n", " total_counts[pattern_name] = total_counts.get(pattern_name, 0) + 1\n", " if did_candle_go_up(seq):\n", " up_counts[pattern_name] = up_counts.get(pattern_name, 0) + 1\n", " \n", " # Calculate percentage of times the last day's candle went up for each pattern\n", " for pattern, count in up_counts.items():\n", " results[pattern] = count / total_counts[pattern]\n", "\n", " return results" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Load all_sequences from the pickle file\n", "with open('data_3candles.pkl', 'rb') as f:\n", " all_sequences = pickle.load(f)\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "results = analyze_patterns(all_sequences)\n", "for pattern, percent_up in results.items():\n", " print(f\"{pattern}: {percent_up * 100:.2f}% went up\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "# Optimization" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "from bayes_opt import BayesianOptimization\n", "\n", "def is_bearish_candle(candle):\n", " start, end, high, low = candle\n", " return (end < start) \n", "\n", "def is_bullish_candle(candle):\n", " start, end, high, low = candle\n", " return (end > start)\n", "\n", "def is_doji(candle, doji_diff_multiplier):\n", " start, end, high, low = candle\n", " diff = abs(end - start)\n", " return (doji_diff_multiplier * diff < (high - low)) and \\\n", " (abs(high - max(end, start)) > doji_diff_multiplier * diff or \\\n", " abs(low - min(end, start)) > doji_diff_multiplier * diff)\n", "\n", "def is_long_legged_doji(candle, doji_diff_multiplier):\n", " return is_doji(candle, doji_diff_multiplier)\n", "\n", "def is_dragonfly_doji(candle, doji_diff_multiplier, doji_abs_diff_multiplier):\n", " start, end, high, low = candle\n", " return is_doji(candle, doji_diff_multiplier) and \\\n", " (high - max(end, start)) < (doji_abs_diff_multiplier * abs(end - start))\n", "\n", "def is_gravestone_doji(candle, doji_diff_multiplier, doji_abs_diff_multiplier):\n", " start, end, high, low = candle\n", " return is_doji(candle, doji_diff_multiplier) and \\\n", " (min(end, start) - low) < (doji_abs_diff_multiplier * abs(end - start))\n", "\n", "def is_hammer(candle, hammer_abs_diff_multiplier):\n", " start, end, high, low = candle\n", " diff = abs(end - start)\n", " return (high - max(end, start)) < (hammer_abs_diff_multiplier * diff) and \\\n", " (low - min(end, start)) > (hammer_abs_diff_multiplier * diff)\n", "\n", "def is_inverted_hammer(candle, hammer_abs_diff_multiplier):\n", " start, end, high, low = candle\n", " diff = abs(end - start)\n", " return (low - min(end, start)) < (hammer_abs_diff_multiplier * diff) and \\\n", " (high - max(end, start)) > (hammer_abs_diff_multiplier * diff)\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Load sequences from the pickle file\n", "with open('data_3candles.pkl', 'rb') as f:\n", " candles = pickle.load(f)\n", "\n", "def objective(doji_multiplier, hammer_abs_diff_multiplier):\n", " # Counting the times when the patterns indicate a reversal\n", " flipped_signals = 0\n", " \n", " for seq in candles:\n", " # The sequence length should be 3 for the given logic\n", " if len(seq) != 3:\n", " continue\n", "\n", " # Bearish Candle followed by a potential bullish reversal pattern and then a bullish candle\n", " if is_bearish_candle(seq[0]):\n", " if (is_long_legged_doji(seq[1], doji_multiplier) or \n", " # is_dragonfly_doji(seq[1], doji_multiplier, doji_abs_diff_multiplier) or \n", " # is_gravestone_doji(seq[1], doji_multiplier, doji_abs_diff_multiplier) or \n", " is_hammer(seq[1], hammer_abs_diff_multiplier) or \n", " is_inverted_hammer(seq[1], hammer_abs_diff_multiplier)) and is_bullish_candle(seq[2]):\n", " flipped_signals += 1\n", "\n", " # Bullish Candle followed by a potential bearish reversal pattern and then a bearish candle\n", " elif is_bullish_candle(seq[0]):\n", " if (is_long_legged_doji(seq[1], doji_multiplier) or \n", " # is_dragonfly_doji(seq[1], doji_multiplier, doji_abs_diff_multiplier) or \n", " # is_gravestone_doji(seq[1], doji_multiplier, doji_abs_diff_multiplier) or \n", " is_hammer(seq[1], hammer_abs_diff_multiplier) or \n", " is_inverted_hammer(seq[1], hammer_abs_diff_multiplier)) and is_bearish_candle(seq[2]):\n", " flipped_signals += 1\n", "\n", " return -flipped_signals\n", "\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "pbounds = {\n", " 'doji_multiplier': (4, 20),\n", " 'hammer_abs_diff_multiplier': (2, 8)\n", "}\n", "\n", "optimizer = BayesianOptimization(\n", " f=objective,\n", " pbounds=pbounds,\n", " random_state=0,\n", " allow_duplicate_points=True\n", ")\n", "\n", "optimizer.maximize(\n", " init_points=4,\n", " n_iter=8,\n", ")\n", "print(optimizer.max)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def is_bearish_candle(candle):\n", " start, end, high, low = candle\n", " return (end < start) \n", "\n", "def is_bullish_candle(candle):\n", " start, end, high, low = candle\n", " return (end > start)\n", "\n", "def is_doji(candle):\n", " start, end, high, low = candle\n", " diff = abs(end - start)\n", " return (11 * diff < (high - low)) and (abs(high - max(end, start)) > 11 * diff or abs(low - min(end, start)) > 10 * diff)\n", "\n", "def is_long_legged_doji(candle):\n", " return is_doji(candle) # long legged doji is characterized primarily by its long shadows\n", "\n", "def is_dragonfly_doji(candle):\n", " start, end, high, low = candle\n", " return is_doji(candle) and (high - max(end, start)) < (2 * abs(end - start))\n", "\n", "def is_gravestone_doji(candle):\n", " start, end, high, low = candle\n", " return is_doji(candle) and (min(end, start) - low) < (2 * abs(end - start))\n", "\n", "def is_hammer(candle):\n", " start, end, high, low = candle\n", " diff = abs(end - start)\n", " return (high - max(end, start)) < (4 * diff) and (low - min(end, start)) > (4 * diff)\n", "\n", "def is_inverted_hammer(candle):\n", " start, end, high, low = candle\n", " diff = abs(end - start)\n", " return (low - min(end, start)) < (4 * diff) and (high - max(end, start)) > (4 * diff)\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "def did_candle_go_up(sequence):\n", " # Return True if the last day's candle went up, False otherwise\n", " return sequence[2][1] > sequence[2][0]\n", "\n", "def analyze_patterns(sequences):\n", " results = {}\n", " up_counts = {} # Stores the number of times the last day's candle went up for each pattern\n", " total_counts = {} # Stores the total number of occurrences of each pattern\n", " \n", " for seq in sequences:\n", " if len(seq) != 3:\n", " continue\n", "\n", " pattern_name = None\n", "\n", " # Bearish Candle -> 3 different classes of dogi\n", " if is_bearish_candle(seq[0]):\n", " if is_long_legged_doji(seq[1]):\n", " pattern_name = 'Bearish -> Long-legged Doji'\n", " # elif is_dragonfly_doji(seq[1]):\n", " # pattern_name = 'Bearish -> Dragonfly Doji'\n", " # elif is_gravestone_doji(seq[1]):\n", " # pattern_name = 'Bearish -> Gravestone Doji'\n", " \n", " # Hammer\n", " if is_hammer(seq[1]):\n", " pattern_name = 'Bearish -> Hammer'\n", " elif is_inverted_hammer(seq[1]):\n", " pattern_name = 'Bearish -> Inverted Hammer'\n", " \n", " # Bullish Candle -> 3 different classes of dogi\n", " elif is_bullish_candle(seq[0]):\n", " if is_long_legged_doji(seq[1]):\n", " pattern_name = 'Bullish -> Long-legged Doji'\n", " # elif is_dragonfly_doji(seq[1]):\n", " # pattern_name = 'Bullish -> Dragonfly Doji'\n", " # elif is_gravestone_doji(seq[1]):\n", " # pattern_name = 'Bullish -> Gravestone Doji'\n", " \n", " # Hammer\n", " if is_hammer(seq[1]):\n", " pattern_name = 'Bullish -> Hammer'\n", " elif is_inverted_hammer(seq[1]):\n", " pattern_name = 'Bullish -> Inverted Hammer'\n", "\n", " # Update the counts based on whether the last day's candle went up\n", " if pattern_name:\n", " total_counts[pattern_name] = total_counts.get(pattern_name, 0) + 1\n", " if did_candle_go_up(seq):\n", " up_counts[pattern_name] = up_counts.get(pattern_name, 0) + 1\n", " \n", " # Calculate percentage of times the last day's candle went up for each pattern\n", " for pattern, count in up_counts.items():\n", " results[pattern] = count / total_counts[pattern]\n", "\n", " return results" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "results = analyze_patterns(all_sequences)\n", "for pattern, percent_up in results.items():\n", " print(f\"{pattern}: {percent_up * 100:.2f}% went up\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [] }, { "cell_type": "code", "execution_count": 1, "metadata": {}, "outputs": [], "source": [ "import pandas as pd\n", "import numpy as np\n", "from sklearn.model_selection import train_test_split\n", "import pickle" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 데이터 전처리 (Data Preprocessing)" ] }, { "cell_type": "code", "execution_count": 2, "metadata": {}, "outputs": [], "source": [ "def preprocess_dataset(file_path):\n", " df = pd.read_csv(file_path, parse_dates=[0])\n", "\n", " result = {}\n", " for ticker in df['tck_iem_cd'].unique():\n", " ticker_data = df[df['tck_iem_cd'] == ticker]\n", " result[ticker] = ticker_data[['gts_iem_ong_pr', 'gts_iem_end_pr', 'gts_iem_hi_pr', 'gts_iem_low_pr']].values.tolist()\n", "\n", " return result" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 데이터 인코딩 (Data Encoding)" ] }, { "cell_type": "code", "execution_count": 3, "metadata": {}, "outputs": [], "source": [ "def encode_data(data):\n", " encoded_data = {}\n", " for ticker, sequences in data.items():\n", " encoded_sequences = []\n", " for sequence in sequences:\n", " encoded_sequence = []\n", " for day in sequence:\n", " open_price, close_price, high_price, low_price = day\n", " \n", " # Calculate percentage change (multiply by 100 for percentage representation)\n", " body_size = 100 * ((close_price - open_price) / open_price)\n", " upper_shadow = 100 * ((high_price - max(open_price, close_price)) / max(open_price, close_price))\n", " lower_shadow = 100 * ((min(open_price, close_price) - low_price) / min(open_price, close_price))\n", " \n", " encoded_sequence.append([body_size, upper_shadow, lower_shadow])\n", " encoded_sequences.append(encoded_sequence)\n", " encoded_data[ticker] = encoded_sequences\n", " return encoded_data" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 데이터 라벨링 (Labeling the Data)" ] }, { "cell_type": "code", "execution_count": 4, "metadata": {}, "outputs": [], "source": [ "def create_labels(data):\n", " labels = {}\n", " for ticker, sequences in data.items():\n", " sequence_labels = []\n", " for i, sequence in enumerate(sequences[:-1]): # Excluding the last sequence since we don't have the next day data for it\n", " current_end_price = sequence[-1][1]\n", " next_end_price = sequences[i+1][-1][1]\n", " price_change = (next_end_price - current_end_price) / current_end_price\n", " if price_change >= 0:\n", " label = 0\n", " else:\n", " label = 1\n", " sequence_labels.append(label)\n", " \n", " labels[ticker] = sequence_labels\n", " return labels" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 데이터 정규화 및 훈련 준비 (Data Normalization and Training Preparation)" ] }, { "cell_type": "code", "execution_count": 5, "metadata": {}, "outputs": [], "source": [ "def chunk_sequences(data, window_size=5):\n", " \"\"\"\n", " Chunks the data into sequences of a given window size.\n", " \"\"\"\n", " chunks = []\n", " for i in range(len(data) - window_size + 1):\n", " chunks.append(data[i:i+window_size])\n", " return chunks\n", "\n", "\n", "def normalize_chunk(chunk):\n", " \"\"\"\n", " Normalizes a given chunk using the first day's start price.\n", " If the price went up or down by a percentage from the first day's price, \n", " that percentage is added to or subtracted from 100.\n", " \"\"\"\n", " first_day_start_price = chunk[0][0]\n", " normalized_chunk = []\n", "\n", " for day in chunk:\n", " normalized_day = [\n", " 100 * (day[i] / first_day_start_price) if i == 0 and day_idx == 0 else \n", " (100 * (day[i] / first_day_start_price) if day[i] != 0 else 0)\n", " for day_idx, i in enumerate(range(4))\n", " ]\n", " normalized_chunk.append(normalized_day)\n", "\n", " # Setting first day's start price to 100\n", " normalized_chunk[0][0] = 100\n", " \n", " return normalized_chunk\n", "\n", "\n", "def original_chunk(chunk):\n", " \"\"\"\n", " This function will just return the chunk as it is without performing any normalization.\n", " \"\"\"\n", " return chunk\n", " \n", "\n", "def preprocess_data_for_training(data, window_size=5):\n", " # Existing functions: chunk_sequences, original_chunk, encode_data, create_labels\n", "\n", " training_data = []\n", "\n", " for ticker, prices in data.items():\n", " chunks = chunk_sequences(prices, window_size)\n", " normalized_chunks = [normalize_chunk(chunk) for chunk in chunks]\n", " original_chunks = [original_chunk(chunk) for chunk in chunks]\n", " encoded_chunks = encode_data({'temp': original_chunks})['temp'] # Using a temp key to fit the function's format\n", " \n", " # Combine original and encoded data\n", " combined_chunks = [\n", " [day + encoded_day for day, encoded_day in zip(orig_chunk, encoded_chunk)] \n", " for orig_chunk, encoded_chunk in zip(normalized_chunks, encoded_chunks)\n", " ]\n", "\n", " # Create labels for these chunks\n", " labels_for_chunks = create_labels({'temp': original_chunks})['temp']\n", "\n", " # Add to the training data\n", " for combined_chunk, label in zip(combined_chunks, labels_for_chunks):\n", " training_data.append((combined_chunk, label))\n", "\n", " return training_data" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 데이터 분할 및 저장 (Data Splitting and Saving)" ] }, { "cell_type": "code", "execution_count": 6, "metadata": {}, "outputs": [], "source": [ "file_path = \"NASDAQ_DT_FC_STK_QUT.csv\"\n", "preprocessed_data = preprocess_dataset(file_path)\n", "training_data = preprocess_data_for_training(preprocessed_data)\n", "\n", "# 데이터 분할\n", "train_data, test_data = train_test_split(training_data, test_size=0.2, random_state=42)\n", "\n", "# train_data 저장\n", "with open('train_data.pkl', 'wb') as f:\n", " pickle.dump(train_data, f)\n", "\n", "# test_data 저장\n", "with open('test_data.pkl', 'wb') as f:\n", " pickle.dump(test_data, f)" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### 모델 생성 및 훈련 (Model Creation and Training)" ] }, { "cell_type": "code", "execution_count": 7, "metadata": {}, "outputs": [], "source": [ "import pandas as pd\n", "import numpy as np\n", "import pickle\n", "import copy\n", "import math\n", "\n", "import torch.nn as nn\n", "from torch.utils.data import DataLoader, TensorDataset\n", "import torch.optim as optim\n", "import torch" ] }, { "cell_type": "code", "execution_count": 8, "metadata": {}, "outputs": [], "source": [ "# Importing necessary libraries\n", "import torch.nn as nn\n", "import torch\n", "import math\n", "\n", "# PositionalEncoding Class\n", "class PositionalEncoding(nn.Module):\n", " def __init__(self, d_model, dropout=0.1, max_len=5000):\n", " super(PositionalEncoding, self).__init__()\n", " # Dropout layer\n", " self.dropout = nn.Dropout(p=dropout)\n", "\n", " # Initialize a zero tensor for positional encoding\n", " pe = torch.zeros(max_len, d_model)\n", "\n", " # Create a tensor containing positions for max_len values\n", " position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)\n", "\n", " # Create a division term for the sinusoidal function\n", " div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))\n", " \n", " # Calculate positional encoding for even indices\n", " pe[:, 0::2] = torch.sin(position * div_term)\n", "\n", " # Calculate positional encoding for odd indices\n", " # Adjust for d_model values that are odd\n", " if d_model % 2 == 0:\n", " pe[:, 1::2] = torch.cos(position * div_term)\n", " else:\n", " pe[:, 1::2] = torch.cos(position * div_term)[:,:-1]\n", "\n", " # Transpose the PE for addition with token embeddings\n", " pe = pe.unsqueeze(0).transpose(0, 1)\n", "\n", " # Register buffer is a way to include tensors that are not parameters (don't need gradients)\n", " # but you still want them to be part of the module's state.\n", " self.register_buffer('pe', pe)\n", "\n", " def forward(self, x):\n", " # Add positional encoding to the input and apply dropout\n", " x = x + self.pe[:x.size(0), :]\n", " return self.dropout(x)\n", "\n", "\n", "# TransformerPatternRecognition Class\n", "class TransformerPatternRecognition(nn.Module):\n", " def __init__(self, d_model, nhead, num_encoder_layers, num_classes, dropout=0.1):\n", " super(TransformerPatternRecognition, self).__init__()\n", " # Positional encoder\n", " self.pos_encoder = PositionalEncoding(d_model, dropout)\n", "\n", " # Transformer layer\n", " # This uses multi-head self attention mechanism internally.\n", " self.transformer = nn.Transformer(d_model, nhead, num_encoder_layers, dropout=dropout)\n", "\n", " # Batch normalization stabilizes the activations of the model\n", " self.bn = nn.BatchNorm1d(d_model)\n", "\n", " # Fully connected layer for classification\n", " self.fc = nn.Linear(d_model, num_classes)\n", " \n", " # Initialize weights using kaiming normal (also known as He initialization).\n", " # It's especially used for deep networks to prevent the vanishing gradient problem.\n", " for p in self.parameters():\n", " if p.dim() > 1:\n", " nn.init.kaiming_normal_(p)\n", "\n", " def forward(self, x):\n", " # Apply positional encoding\n", " x = self.pos_encoder(x)\n", "\n", " # Pass the input through the transformer\n", " output = self.transformer(x, x)\n", "\n", " # Apply batch normalization on the output of the last token \n", " # (often [CLS] token in some architectures, but here just the last token's output)\n", " output = self.bn(output[:, -1, :])\n", "\n", " # Pass the output through the fully connected layer\n", " output = self.fc(output)\n", "\n", " return output\n" ] }, { "cell_type": "code", "execution_count": 9, "metadata": {}, "outputs": [], "source": [ "def to_1d_tensor(t):\n", " t = t.squeeze()\n", " if t.dim() == 0:\n", " t = t.unsqueeze(0)\n", " return t\n", "\n", "\n", "def calculate_accuracy(outputs, labels):\n", " _, predicted = torch.max(outputs.data, 1)\n", " total = labels.size(0)\n", " correct = (predicted == labels).sum().item()\n", " accuracy = 100 * correct / total\n", " return accuracy\n", "\n", "\n", "def evaluate_model(model, dataloader, criterion):\n", " total_loss = 0.0\n", " total_correct = 0\n", " total_items = 0\n", "\n", " with torch.no_grad():\n", " for inputs, labels in dataloader:\n", " inputs, labels = inputs.to(device), labels.to(device) # Move inputs and labels to the model's device\n", " outputs = model(inputs)\n", "\n", " loss = criterion(outputs, labels)\n", " total_loss += loss.item()\n", "\n", " accuracy = calculate_accuracy(outputs, labels)\n", " total_correct += accuracy * len(labels) / 100 # Convert accuracy percentage back to count\n", " total_items += len(labels)\n", "\n", " avg_loss = total_loss / len(dataloader)\n", " overall_accuracy = 100 * total_correct / total_items\n", " return avg_loss, overall_accuracy" ] }, { "cell_type": "code", "execution_count": 10, "metadata": {}, "outputs": [], "source": [ "def get_dataloader(training_data, batch_size=32):\n", " input_tensors = torch.tensor([item[0] for item in training_data], dtype=torch.float32)\n", " label_tensors = torch.tensor([item[1] for item in training_data], dtype=torch.long)\n", " dataset = TensorDataset(input_tensors, label_tensors)\n", " dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True)\n", " return dataloader\n", "\n", "\n", "# Load the train and test data using pickle\n", "with open('train_data.pkl', 'rb') as f:\n", " training_data = pickle.load(f)\n", "with open('test_data.pkl', 'rb') as f:\n", " test_data = pickle.load(f)" ] }, { "cell_type": "code", "execution_count": 11, "metadata": {}, "outputs": [], "source": [ "# Parameters\n", "d_model = 7\n", "nhead = 1\n", "num_encoder_layers = 2\n", "num_classes = 2\n", "learning_rate = 0.0001\n", "num_epochs = 100\n", "\n", "device = torch.device(\"cuda\" if torch.cuda.is_available() else \"cpu\")\n", "\n", "# Model, Optimizer\n", "model = TransformerPatternRecognition(d_model, nhead, num_encoder_layers, num_classes).to(device)" ] }, { "cell_type": "code", "execution_count": 12, "metadata": {}, "outputs": [], "source": [ "# Weights for classes\n", "criterion = nn.CrossEntropyLoss()\n", "optimizer = optim.Adam(model.parameters(), lr=learning_rate)\n", "\n", "# Training loop with batching\n", "train_dataloader = get_dataloader(training_data)\n", "test_dataloader = get_dataloader(test_data)\n", "\n", "all_outputs = []\n", "all_labels = []" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Training" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# train\n", "for epoch in range(num_epochs):\n", " total_loss = 0.0\n", " iter_count = 0\n", " total_iters = len(train_dataloader)\n", "\n", " for inputs, labels in train_dataloader:\n", " inputs, labels = inputs.to(device), labels.to(device)\n", " \n", " iter_count += 1\n", " optimizer.zero_grad()\n", " outputs = model(inputs)\n", " loss = criterion(outputs, labels)\n", " \n", " loss.backward()\n", " torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)\n", " optimizer.step()\n", " total_loss += loss.item()\n", "\n", " all_outputs.append(outputs)\n", " all_labels.append(labels)\n", "\n", " # Periodically backup the model's weights\n", " if iter_count % 1000 == 0:\n", " accumulated_outputs = torch.cat(all_outputs, dim=0)\n", " accumulated_labels = torch.cat(all_labels, dim=0)\n", " accuracy = calculate_accuracy(accumulated_outputs, accumulated_labels)\n", " print(f\"Epoch {epoch+1}/{num_epochs}, Iteration {iter_count}/{total_iters}, Loss: {total_loss/iter_count:.4f}, Accuracy: {accuracy:.2f}%\")\n", " \n", " # Reset the lists for the next 1000 iterations\n", " all_outputs = []\n", " all_labels = []\n", "\n", " avg_loss = total_loss / len(train_dataloader)\n", " print(f\"Epoch {epoch+1}/{num_epochs} - Avg Loss: {avg_loss:.4f}\")\n", " \n", " # Evaluate on test set\n", " test_loss, test_accuracy = evaluate_model(model, test_dataloader, criterion)\n", " print(f\"Epoch {epoch+1}/{num_epochs} - Test Loss: {test_loss:.4f}, Test Accuracy: {test_accuracy:.2f}%\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [] } ], "metadata": { "kernelspec": { "display_name": "prac", "language": "python", "name": "prac" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.9.0" } }, "nbformat": 4, "nbformat_minor": 4 }