crupo23 / app.py
tahatehrani's picture
Update app.py
3f4d14f verified
raw
history blame
27.7 kB
import streamlit as st
import requests
import json
import os
import numpy as np
import yfinance as yf
import datetime as dt
import pandas as pd
import pandas_ta as ta
from pytz import timezone
import plotly.graph_objects as go
from sklearn.linear_model import LinearRegression
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score
USERS_FILE = 'users.json'
API_KEYS_FILE = 'api_keys.json'
def load_users():
if not os.path.exists(USERS_FILE):
with open(USERS_FILE, 'w') as file:
json.dump({"users": []}, file)
with open(USERS_FILE, 'r') as file:
return json.load(file)
def save_users(users):
with open(USERS_FILE, 'w') as file:
json.dump(users, file)
def login(username, password):
users = load_users()
for user in users['users']:
if user['username'] == username and user['password'] == password:
return True
return False
def signup(username, password):
users = load_users()
for user in users['users']:
if user['username'] == username:
return False
users['users'].append({"username": username, "password": password})
save_users(users)
return True
def admin_login(username, password):
if username == "admin" and password == "admin":
return True
return False
def load_api_keys():
if not os.path.exists(API_KEYS_FILE):
with open(API_KEYS_FILE, 'w') as file:
json.dump({"newsapi_key": "", "coinmarketcap_key": ""}, file)
with open(API_KEYS_FILE, 'r') as file:
return json.load(file)
def save_api_keys(newsapi_key, coinmarketcap_key):
api_keys = load_api_keys()
api_keys['newsapi_key'] = newsapi_key
api_keys['coinmarketcap_key'] = coinmarketcap_key
with open(API_KEYS_FILE, 'w') as file:
json.dump(api_keys, file)
def get_crypto_news(api_key, crypto_symbol, articles_count=10):
url = f"https://newsapi.org/v2/everything?q={crypto_symbol}&apiKey={api_key}&language=en&sortBy=publishedAt&pageSize={articles_count}"
response = requests.get(url)
if response.status_code == 200:
news_data = response.json()
articles = news_data.get('articles', [])
crypto_news = []
for article in articles:
title = article.get('title', 'No Title')
description = article.get('description', 'No Description')
url = article.get('url', '#')
published_at = article.get('publishedAt', 'No Date')
relevancy = article.get('relevancy', 'unknown')
popularity = article.get('popularity', 'unknown')
crypto_news.append({
"title": title,
"description": description,
"url": url,
"publishedAt": published_at,
"relevancy": relevancy,
"popularity": popularity
})
return crypto_news
else:
return []
def custom_sentiment_analysis(news, domain_lexicon):
analyzer = SentimentIntensityAnalyzer()
for article in news:
title = article['title']
description = article['description']
sentiment_score = analyzer.polarity_scores(title + " " + description)
# Use the domain-specific lexicon to adjust the sentiment score
for term, weight in domain_lexicon.items():
if term.lower() in (title + " " + description).lower():
sentiment_score['compound'] += weight
if sentiment_score['compound'] >= 0.5:
article['sentiment'] = 'positive'
elif sentiment_score['compound'] <= -0.5:
article['sentiment'] = 'negative'
else:
article['sentiment'] = 'neutral'
return news
def train_price_prediction_model(data):
X = data[['Open', 'High', 'Low', 'Volume']]
y = data['Close']
model = LinearRegression()
model.fit(X, y)
return model
def predict_crypto_price(data, model):
latest_data = data.iloc[-1]
latest_features = latest_data[['Open', 'High', 'Low', 'Volume']].values.reshape(1, -1)
predicted_price = model.predict(latest_features)[0]
return predicted_price
def analyze_indicators(data):
# محاسبه و اضافه کردن شاخص‌های تکنیکال
if 'Close' in data:
data['RSI'] = ta.rsi(data['Close'], length=14)
data['Stochastic'] = ta.stoch(data['High'], data['Low'], data['Close'], k=14, d=3)['STOCHk_14_3_3']
macd = ta.macd(data['Close'], fast=12, slow=26, signal=9)
data['MACD'] = macd['MACD_12_26_9']
data['SMA'] = ta.sma(data['Close'], length=50)
data['EMA'] = ta.ema(data['Close'], length=50)
return data
def calculate_indicators(data):
data['MA'] = data['Close'].rolling(window=10).mean()
data['CCI'] = (data['Close'] - data['Close'].rolling(window=20).mean()) / (0.015 * data['Close'].rolling(window=20).std())
data['MACD'] = data['Close'].ewm(span=12, adjust=False).mean() - data['Close'].ewm(span=26, adjust=False).mean()
return data
def generate_signals(data, news_sentiment):
buy_signal = None
sell_signal = None
confidence = None
data = analyze_indicators(data)
data = calculate_indicators(data)
data.dropna(inplace=True)
# چک کردن وجود ستون‌های لازم
required_cols = ['RSI', 'Stochastic', 'MA', 'CCI', 'MACD', 'news_sentiment']
for col in required_cols:
if col not in data.columns:
data[col] = pd.Series([None] * len(data), index=data.index)
labels = ((data['RSI'] < 30) & (data['Stochastic'] < 20)).astype(int) - ((data['RSI'] > 70) & (data['Stochastic'] > 80)).astype(int)
# Check if data is not empty
if data.empty or labels.empty or len(data) == 0:
st.error("Not enough data to generate signals.")
return buy_signal, sell_signal
X_train, X_test, y_train, y_test = train_test_split(data[required_cols], labels, test_size=0.2, random_state=42)
if len(X_train) == 0 or len(y_train) == 0:
st.error("Training set is empty after train/test split. Adjust parameters.")
return buy_signal, sell_signal
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X_train, y_train)
y_pred = model.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)
if 0.8 <= accuracy <= 1.0:
latest_data = data.iloc[-1]
prediction = model.predict([latest_data[required_cols].values])
confidence = model.predict_proba([latest_data[required_cols].values])[0][abs(prediction[0])]
if prediction[0] == 1:
buy_signal = (latest_data.name, latest_data['Close'], latest_data['Close'] * 0.95, "High Risk", confidence)
elif prediction[0] == -1:
sell_signal = (latest_data.name, latest_data['Close'], latest_data['Close'] * 1.05, "High Risk", confidence)
if buy_signal is None and sell_signal is None:
if 'RSI' in data.columns and 'Stochastic' in data.columns:
if data['RSI'].iloc[-1] < 30 and data['Stochastic'].iloc[-1] < 20:
buy_signal = (data.index[-1], data['Close'].iloc[-1], data['Close'].iloc[-1] * 0.95, "Low Confidence", 0.5)
elif data['RSI'].iloc[-1] > 70 and data['Stochastic'].iloc[-1] > 80:
sell_signal = (data.index[-1], data['Close'].iloc[-1], data['Close'].iloc[-1] * 1.05, "Low Confidence", 0.5)
return buy_signal, sell_signal
def get_fear_and_greed_index():
response = requests.get("https://api.alternative.me/fng/?limit=1")
if response.status_code == 200:
return response.json()["data"][0]["value"]
else:
return None
def get_crypto_data_from_coinmarketcap(api_key, crypto_symbol):
url = "https://pro-api.coinmarketcap.com/v1/cryptocurrency/quotes/latest"
parameters = {'symbol': crypto_symbol, 'convert': 'USD'}
headers = {'Accepts': 'application/json', 'X-CMC_PRO_API_KEY': api_key}
response = requests.get(url, headers=headers, params=parameters)
data = response.json()
return data['data'][crypto_symbol]['quote']['USD']
def display_time_information(language):
if language == "English":
st.subheader("Time Information")
st.write("Below are the current times for different global markets and the best trading time in Iran.")
else:
st.subheader("Information on Time")
iran_tz = timezone('Asia/Tehran')
utc_tz = timezone('UTC')
japan_tz = timezone('Asia/Tokyo')
europe_tz = timezone('Europe/Berlin')
us_tz = timezone('America/New_York')
iran_time = dt.datetime.now(iran_tz).strftime('%H:%M:%S')
utc_time = dt.datetime.now(utc_tz).strftime('%H:%M:%S')
japan_open = dt.datetime.now(japan_tz).replace(hour=9, minute=0, second=0, microsecond=0).strftime('%H:%M:%S')
europe_open = dt.datetime.now(europe_tz).replace(hour=8, minute=0, second=0, microsecond=0).strftime('%H:%M:%S')
us_open = dt.datetime.now(us_tz).replace(hour=9, minute=30, second=0, microsecond=0).strftime('%H:%M:%S')
if language == "English":
st.write(f"Iran Time: {iran_time}")
st.write(f"UTC Time: {utc_time}")
st.subheader("Global Crypto Markets Open Times")
data = {
"Country": ["Japan", "Europe", "USA"],
"Open Time": [japan_open, europe_open, us_open]
}
df = pd.DataFrame(data)
st.table(df)
st.subheader("Best Trading Time in Iran")
st.write("The best time for trading in Iran is when the global crypto markets are active, especially during the overlapping hours of the European and American markets.")
else:
st.write(f"زمان ایران: {iran_time}")
st.write(f"زمان هماهنگ جهانی: {utc_time}")
st.subheader("زمان باز شدن بازارهای جهانی ارز دیجیتال")
data = {
"کشور": ["ژاپن", "اروپا", "آمریکا"],
"زمان باز شدن": [japan_open, europe_open, us_open]
}
df = pd.DataFrame(data)
st.table(df)
st.subheader("بهترین زمان معامله در ایران")
st.write("بهترین زمان برای معامله در ایران زمانی است که بازارهای جهانی ارز دیجیتال فعال هستند، به ویژه در ساعت های همپوشانی بازارهای اروپا و آمریکا.")
def generate_learning_tips(language):
tips = [
{"en": "Diversify your portfolio to manage risk effectively.", "fa": "سبد سرمایه‌گذاری خود را متنوع کنید تا ریسک را به طور مؤثری مدیریت کنید."},
{"en": "Use technical analysis to identify market trends.", "fa": "از تحلیل تکنیکال برای شناسایی روندهای بازار استفاده کنید."},
{"en": "Stay updated with the latest news in the crypto world.", "fa": "با آخرین اخبار دنیای ارز دیجیتال به‌روز باشید."},
{"en": "Understand the fundamentals of the cryptocurrencies you invest in.", "fa": "اصول اولیه ارزهای دیجیتالی که در آن‌ها سرمایه‌گذاری می‌کنید را درک کنید."},
{"en": "Use stop-loss orders to protect your investments.", "fa": "از دستورات توقف ضرر برای محافظت از سرمایه‌گذاری‌های خود استفاده کنید."},
{"en": "Regularly review your investment strategy and adjust as needed.", "fa": "استراتژی سرمایه‌گذاری خود را به طور منظم بازبینی کنید و در صورت نیاز آن را تنظیم کنید."},
{"en": "Don't invest more than you can afford to lose.", "fa": "بیش از آنچه که می‌توانید از دست بدهید سرمایه‌گذاری نکنید."}
]
if language == "English":
st.subheader("Learning Tips")
for tip in tips:
st.write(f"- {tip['en']}")
else:
st.subheader("نکات آموزشی")
for tip in tips:
st.write(f"- {tip['fa']}")
def get_bitcoin_price(time_frame='1h'):
base_url = 'https://api.pro.coinbase.com/products/NOT-USD/candles'
response = requests.get(base_url, params={'granularity': time_frame})
data = response.json()
df = pd.DataFrame(data, columns=['epoch', 'low', 'high', 'open', 'close', 'volume'])
df['epoch'] = pd.to_datetime(df['epoch'], unit='s', utc=True)
df.set_index('epoch', inplace=True)
return df
def get_current_bitcoin_price():
url = "https://api.coindesk.com/v1/bpi/currentprice/BTC.json"
response = requests.get(url)
data = response.json()
price = data['bpi']['USD']['rate_float']
return price
def calculate_indicators(price):
# فرض کنید 'price' یک آرایه دو بعدی با شکل (366, 11) است
# استفاده از اولین ستون داده‌ها برای محاسبه اندیکاتورها
if isinstance(price, np.ndarray) and price.ndim == 2:
price = price[:, 0] # انتخاب اولین ستون
# تولید یک بازه زمانی برای اندیس
index = pd.date_range(start=pd.Timestamp.now(), periods=len(price), freq='D')
# ایجاد سری با اندیس‌های صحیح
prices = pd.Series(price, index=index)
# محاسبه SMA و EMA
sma_12 = prices.rolling(window=12).mean()
sma_26 = prices.rolling(window=26).mean()
ema_12 = prices.ewm(span=12, adjust=False).mean()
ema_26 = prices.ewm(span=26, adjust=False).mean()
# محاسبه MACD و خط سیگنال و هیستوگرام
macd = ema_12 - ema_26
signal_line = macd.ewm(span=9, adjust=False).mean()
histogram = macd - signal_line
# محاسبه RSI
delta = prices.diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
# بازگرداندن آخرین مقادیر محاسبه شده
return {
'sma_12': sma_12.iloc[-1],
'sma_26': sma_26.iloc[-1],
'ema_12': ema_12.iloc[-1],
'ema_26': ema_26.iloc[-1],
'macd': macd.iloc[-1],
'signal_line': signal_line.iloc[-1],
'histogram': histogram.iloc[-1],
'rsi': rsi.iloc[-1]
}
def main():
st.title("Crypto Trading Dashboard")
language = st.sidebar.selectbox("Select Language", ("English", "Persian"))
menu = ["Home", "Login", "SignUp", "Admin", "Time", "Charts", "Market Data", "News" , "signal"]
choice = st.sidebar.selectbox("Menu", menu)
if choice == "Home":
if language == "English":
st.subheader("Welcome to the Crypto Trading Dashboard")
st.write("""
This dashboard provides you with tools and insights to trade cryptocurrencies effectively.
You can track prices, perform technical analysis, get buy/sell signals, predict prices, and stay updated with the latest news.
Use the sidebar to navigate through different sections.
""")
st.write("Website: [Taha Tehrani Nasab](https://ththt.ir)")
st.write("© 2024 Taha Tehrani Nasab. All rights reserved.")
else:
st.subheader("به داشبورد معاملات ارز دیجیتال خوش آمدید")
st.write("""
این داشبورد ابزارها و بینش‌هایی را برای تجارت ارزهای دیجیتال به شما ارائه می‌دهد.
می‌توانید قیمت‌ها را پیگیری کنید، تحلیل تکنیکال انجام دهید، سیگنال‌های خرید/فروش دریافت کنید، قیمت‌ها را پیش‌بینی کنید و با آخرین اخبار به‌روز باشید.
از نوار کناری برای پیمایش در بخش‌های مختلف استفاده کنید.
""")
st.write("وبسایت: [Taha Tehrani nasab](https://ththt.ir)")
st.write("© 2024 Taha Tehrani Nasab. تمامی حقوق محفوظ است.")
elif choice == "Login":
if language == "English":
st.subheader("Login Section")
else:
st.subheader("بخش ورود")
username = st.sidebar.text_input("Username")
password = st.sidebar.text_input("Password", type='password')
if st.sidebar.checkbox("Login"):
if login(username, password):
st.success(f"Logged in as {username}")
if language == "English":
st.subheader("Select Cryptocurrency")
else:
st.subheader("انتخاب ارز دیجیتال")
crypto_symbol = st.selectbox("Cryptocurrency Symbol", ["BTC", "ETH", "LTC", "BCH" , "TON", "NOT"])
end_date = dt.datetime.now()
start_date = end_date - dt.timedelta(days=365)
data = yf.download(crypto_symbol + "-USD", start=start_date, end=end_date)
if language == "English":
st.subheader(f"Price Data for {crypto_symbol}")
else:
st.subheader(f"داده‌های قیمت برای {crypto_symbol}")
st.write(data.tail())
if language == "English":
st.subheader(f"Technical Analysis for {crypto_symbol}")
else:
st.subheader(f"تحلیل تکنیکال برای {crypto_symbol}")
data = analyze_indicators(data)
st.write(data[['RSI', 'Stochastic', 'MACD', 'SMA', 'EMA']].tail())
if language == "English":
st.subheader("Buy/Sell Signals")
else:
st.subheader("سیگنال‌های خرید/فروش")
buy_signal, sell_signal = generate_signals(data, None)
if buy_signal:
st.success(f"Buy Signal: {buy_signal}")
if sell_signal:
st.error(f"Sell Signal: {sell_signal}")
if language == "English":
st.subheader("Price Prediction")
else:
st.subheader("پیش‌بینی قیمت")
model = train_price_prediction_model(data)
predicted_price = predict_crypto_price(data, model)
if language == "English":
st.write(f"The predicted price for the next trading day is: ${predicted_price:.2f}")
else:
st.write(f"قیمت پیش‌بینی شده برای روز معاملاتی بعدی: ${predicted_price:.2f}")
if language == "English":
st.subheader("Fear and Greed Index")
else:
st.subheader("شاخص ترس و طمع")
fear_and_greed_index = get_fear_and_greed_index()
if fear_and_greed_index:
st.write(f"The current Fear and Greed Index is: {fear_and_greed_index}")
else:
if language == "English":
st.write("Could not retrieve the Fear and Greed Index.")
else:
st.write("امکان دریافت شاخص ترس و طمع وجود ندارد.")
else:
if language == "English":
st.warning("Incorrect Username/Password")
else:
st.warning("نام کاربری/رمز عبور اشتباه است")
elif choice == "SignUp":
if language == "English":
st.subheader("Create a New Account")
else:
st.subheader("ایجاد حساب جدید")
new_user = st.text_input("Username")
new_password = st.text_input("Password", type='password')
if st.button("Sign Up"):
if signup(new_user, new_password):
if language == "English":
st.success("Account created successfully. You can now log in.")
else:
st.success("حساب با موفقیت ایجاد شد. اکنون می‌توانید وارد شوید.")
else:
if language == "English":
st.warning("Username already exists. Please choose another.")
else:
st.warning("نام کاربری از قبل وجود دارد. لطفاً نام دیگری انتخاب کنید.")
elif choice == "Admin":
if language == "English":
st.subheader("Admin Section")
else:
st.subheader("بخش مدیریت")
username = st.sidebar.text_input("Admin Username")
password = st.sidebar.text_input("Admin Password", type='password')
if st.sidebar.checkbox("Login"):
if admin_login(username, password):
if language == "English":
st.success("Admin login successful")
st.subheader("Set API Keys")
else:
st.success("ورود مدیر موفقیت‌آمیز بود")
st.subheader("تنظیم کلیدهای API")
newsapi_key = st.text_input("NewsAPI Key")
coinmarketcap_key = st.text_input("CoinMarketCap Key")
if st.button("Save API Keys"):
save_api_keys(newsapi_key, coinmarketcap_key)
if language == "English":
st.success("API keys saved successfully")
else:
st.success("کلیدهای API با موفقیت ذخیره شد")
else:
if language == "English":
st.warning("Incorrect Admin Username/Password")
else:
st.warning("نام کاربری/رمز عبور مدیر اشتباه است")
elif choice == "Time":
display_time_information(language)
generate_learning_tips(language)
elif choice == "Charts":
if language == "English":
st.subheader("Cryptocurrency Charts")
else:
st.subheader("نمودارهای ارز دیجیتال")
crypto_symbol = st.selectbox("Cryptocurrency Symbol", ["BTC", "ETH", "LTC", "BCH", "TON", "NOT"])
end_date = dt.datetime.now()
start_date = end_date - dt.timedelta(days=365)
data = yf.download(crypto_symbol + "-USD", start=start_date, end=end_date)
if language == "English":
st.subheader(f"{crypto_symbol} TradingView-like Chart")
else:
st.subheader(f"نمودار شبیه TradingView برای {crypto_symbol}")
fig1 = go.Figure(data=[go.Candlestick(x=data.index,
open=data['Open'],
high=data['High'],
low=data['Low'],
close=data['Close'])])
st.plotly_chart(fig1)
elif choice == "Market Data":
if language == "English":
st.subheader("Cryptocurrency Market Data")
else:
st.subheader("داده‌های بازار ارز دیجیتال")
crypto_symbol = st.selectbox("Cryptocurrency Symbol", ["BTC", "ETH", "LTC", "BCH" , "TON", "NOT"])
api_keys = load_api_keys()
if 'coinmarketcap_key' in api_keys and api_keys['coinmarketcap_key']:
market_data = get_crypto_data_from_coinmarketcap(api_keys['coinmarketcap_key'], crypto_symbol)
if language == "English":
st.write(f"Price: ${market_data['price']:.2f}")
st.write(f"Market Cap: ${market_data['market_cap']:.2f}")
st.write(f"24h Volume: ${market_data['volume_24h']:.2f}")
st.write(f"Change (24h): {market_data['percent_change_24h']:.2f}%")
else:
st.write(f"قیمت: ${market_data['price']:.2f}")
st.write(f"ارزش بازار: ${market_data['market_cap']:.2f}")
st.write(f"حجم معاملات 24 ساعته: ${market_data['volume_24h']:.2f}")
st.write(f"تغییرات (24 ساعت): {market_data['percent_change_24h']:.2f}%")
else:
if language == "English":
st.warning("API key for CoinMarketCap is not set. Please contact the admin.")
else:
st.warning("کلید API برای CoinMarketCap تنظیم نشده است. لطفاً با مدیر تماس بگیرید.")
elif choice == "News":
if language == "English":
st.subheader("Cryptocurrency News")
else:
st.subheader("اخبار ارز دیجیتال")
crypto_symbol = st.selectbox("Cryptocurrency Symbol", ["BTC", "ETH", "LTC", "BCH" , "TON"])
end_date = dt.datetime.now()
start_date = end_date - dt.timedelta(days=365)
data = yf.download(crypto_symbol + "-USD", start=start_date, end=end_date)
#نیازمند تغییر
api_keys = load_api_keys()
if 'newsapi_key' in api_keys and api_keys['newsapi_key']:
news = get_crypto_news(api_keys['newsapi_key'], crypto_symbol)
news = custom_sentiment_analysis(news, {
"cryptocurrency": 0.5,
"bullish": 0.4,
"bearish": -0.4
})
buy_signal, sell_signal = generate_signals(data, news)
else:
buy_signal, sell_signal = generate_signals(data, None)
#نیاز مند تغییر بالا
# Sorting and categorizing news
sort_by = st.radio("Sort News By", ("publishedAt", "relevancy", "popularity"), index=0)
news = sorted(news, key=lambda x: x[sort_by])
if language == "English":
st.subheader(f"News for {crypto_symbol}")
else:
st.subheader(f"اخبار برای {crypto_symbol}")
# Display news with confidence level
buy_signal, sell_signal = generate_signals(data, news)
if buy_signal:
st.success(f"Buy Signal: {buy_signal}")
if sell_signal:
st.error(f"Sell Signal: {sell_signal}")
#نیاز مند تتغییر بالا
# Paginate news
page = st.slider("Select page", min_value=1, max_value=(len(news) // 5) + 1)
news_to_display = news[(page - 1) * 5: page * 5]
for article in news_to_display:
st.write(f"Title: {article['title']}")
st.write(f"Description: {article['description']}")
st.write(f"Sentiment: {article['sentiment']}")
st.write(f"Published At: {article['publishedAt']}")
st.write(f"Read more: [Link]({article['url']})")
else:
if language == "English":
st.warning("API key for NewsAPI is not set. Please contact the admin.")
else:
st.warning("کلید API برای NewsAPI تنظیم نشده است. لطفاً با مدیر تماس بگیرید.")
if __name__ == '__main__':
main()