Spaces:
Sleeping
Sleeping
| 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 --- | |
| # 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 --- | |
| def load_sentiment_model(): | |
| tokenizer = AutoTokenizer.from_pretrained("ElKulako/cryptobert") | |
| model = AutoModelForSequenceClassification.from_pretrained("ElKulako/cryptobert") | |
| return tokenizer, model | |
| 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.") |