Spaces:
Sleeping
Sleeping
File size: 4,109 Bytes
f65b816 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 |
import streamlit as st
from transformers import AutoTokenizer, AutoModelForSequenceClassification
import requests
import json
import os
# --- API Key ---
CMC_API_KEY = os.environ.get("CMC_API_KEY") # Store your API key as a Hugging Face Secret
if not CMC_API_KEY:
st.warning("Please add your CoinMarketCap API key as a Secret in Hugging Face Spaces.")
else:
headers = {
'X-CMC_PRO_API_KEY': CMC_API_KEY,
'Accepts': 'application/json'
}
# --- Data Fetching ---
@st.cache_data(ttl=60) # Cache data for 60 seconds
def get_crypto_price(symbol):
url = f'https://pro-api.coinmarketcap.com/v1/cryptocurrency/quotes/latest?symbol={symbol}'
try:
response = requests.get(url, headers=headers)
response.raise_for_status()
data = json.loads(response.text)
if data['status']['error_code'] == 0:
price = data['data'][symbol]['quote']['USD']['price']
return price
else:
return f"Error fetching data: {data['status']['error_message']}"
except requests.exceptions.RequestException as e:
return f"Error connecting to CoinMarketCap API: {e}"
# --- AI Model Integration ---
@st.cache_resource
def load_sentiment_model():
tokenizer = AutoTokenizer.from_pretrained("ElKulako/cryptobert")
model = AutoModelForSequenceClassification.from_pretrained("ElKulako/cryptobert")
return tokenizer, model
@st.cache_data
def analyze_sentiment(text, tokenizer, model):
try:
inputs = tokenizer(text, return_tensors="pt")
outputs = model(**inputs)
logits = outputs.logits
predicted_class_id = logits.argmax().item()
return model.config.id2label[predicted_class_id]
except Exception as e:
return f"Error analyzing sentiment: {e}"
tokenizer, sentiment_model = load_sentiment_model()
# --- Main Chatbot Logic ---
def process_user_message(user_input):
user_input_lower = user_input.lower()
if "current price of" in user_input_lower:
symbol = user_input_lower.split("current price of")[1].strip().upper()
price = get_crypto_price(symbol)
if isinstance(price, str) and "Error" in price:
return price
else:
sentiment_summary = analyze_sentiment(f"Recent news about {symbol}", tokenizer, sentiment_model)
return f"The current price of {symbol} is ${price:.2f}. Market sentiment is currently {sentiment_summary}."
elif "should i buy" in user_input_lower:
return "I am currently unable to provide buy/sell recommendations without technical analysis capabilities in this deployment."
elif "rsi say about" in user_input_lower:
return "I am currently unable to analyze RSI without the necessary libraries in this deployment."
else:
return "I'm still learning! I can currently tell you the price of a cryptocurrency and analyze the sentiment of related news."
# --- Streamlit UI ---
st.title("Crypto Trading Assistant")
st.markdown("Ask me about cryptocurrency prices and market sentiment.")
user_query = st.text_input("Your question:", "")
if CMC_API_KEY:
if user_query:
with st.spinner("Thinking..."):
bot_response = process_user_message(user_query)
st.write(f"**Bot:** {bot_response}")
# Simple price chart example if the user asked for the price
if "price of" in user_query.lower():
symbol = user_query.lower().split("price of")[1].strip().upper()
price_data = get_crypto_price(symbol)
if not isinstance(price_data, str):
st.subheader(f"Current Price of {symbol}: ${price_data:.2f}")
st.line_chart([price_data]) # Simple single point chart
else:
st.error("CoinMarketCap API key is missing. Please add it as a Secret.") |