import numpy as np
import pandas as pd
import yfinance as yf
from sklearn.preprocessing import MinMaxScaler
import torch
import torch.nn as nn
from datetime import datetime
import smtplib
from email.mime.text import MIMEText
import matplotlib.pyplot as plt
def get_historical_prices(symbol, start_date, end_date):
df = yf.download(symbol, start=start_date, end=end_date)
return df['Close'].values.reshape(-1, 1)
class LSTM(nn.Module):
def __init__(self, input_size=1, hidden_layer_size=100, output_size=1):
super().__init__()
self.hidden_layer_size = hidden_layer_size
self.lstm = nn.LSTM(input_size, hidden_layer_size)
self.linear = nn.Linear(hidden_layer_size, output_size)
self.hidden_cell = (torch.zeros(1, 1, self.hidden_layer_size),
torch.zeros(1, 1, self.hidden_layer_size))
def forward(self, input_seq):
lstm_out, self.hidden_cell = self.lstm(input_seq.view(len(input_seq), 1,
1), self.hidden_cell)
predictions = self.linear(lstm_out.view(len(input_seq), -1))
return predictions[-1]
def send_email(subject, body):
sender = "dhxnugowdaa@gmail.com"
receiver = "dhxnugowdaa@gmail.com"
sender_password = "royd fvte qwpt cmut"
msg = MIMEText(body)
msg['Subject'] = subject
msg['From'] = sender
msg['To'] = receiver
try:
with smtplib.SMTP('smtp.gmail.com', 587) as server:
server.starttls()
server.login(sender, sender_password)
server.sendmail(sender, receiver, msg.as_string())
print("Email sent successfully.")
except Exception as e:
print(f"Email failed: {e}")
crypto_symbol = 'BTC-USD'
start_date = '2025-01-01'
end_date = datetime.now().strftime('%Y-%m-%d')
prices = get_historical_prices(crypto_symbol, start_date, end_date)
plt.plot(prices)
plt.title(f'{crypto_symbol} Historical Prices')
plt.xlabel('Days')
plt.ylabel('Price (USD)')
plt.show()
scaler = MinMaxScaler(feature_range=(0, 1))
prices_scaled = scaler.fit_transform(prices)
X_train, y_train = [], []
for i in range(60, len(prices_scaled)):
X_train.append(prices_scaled[i-60:i, 0])
y_train.append(prices_scaled[i, 0])
X_train = torch.from_numpy(np.array(X_train)).float()
y_train = torch.from_numpy(np.array(y_train)).float()
model = LSTM()
loss_function = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
epochs = 25
for epoch in range(epochs):
for seq, labels in zip(X_train, y_train):
optimizer.zero_grad()
model.hidden_cell = (torch.zeros(1, 1, model.hidden_layer_size),
torch.zeros(1, 1, model.hidden_layer_size))
y_pred = model(seq)
single_loss = loss_function(y_pred, labels)
single_loss.backward()
optimizer.step()
model.eval()
with torch.no_grad():
last_60_days = prices[-60:].reshape(-1, 1)
last_60_days_scaled = scaler.transform(last_60_days)
X_test = torch.from_numpy(last_60_days_scaled).float()
model.hidden_cell = (torch.zeros(1, 1, model.hidden_layer_size),
torch.zeros(1, 1, model.hidden_layer_size))
predicted_price_scaled = model(X_test)
predicted_price =
scaler.inverse_transform(np.array([[predicted_price_scaled.item()]]))
threshold = 50000
if predicted_price[0][0] > threshold:
subject = "Crypto Price Alert"
body = f"The predicted price of {crypto_symbol} has crossed ${threshold}.\
nPredicted Price: ${predicted_price[0][0]:,.2f}"
send_email(subject, body)
else:
print(f"No alert. Predicted Price: ${predicted_price[0][0]:,.2f}")
