import streamlit as st
import yfinance as yf
import pandas as pd
import plotly.graph_objects as go
import plotly.express as px
from plotly.subplots import make_subplots
import numpy as np
from datetime import datetime, timedelta
import requests
from bs4 import BeautifulSoup
import anthropic
import time
import json
# Page config
st.set_page_config(
page_title="CoreWeave Stock Dashboard",
page_icon="📈",
layout="wide",
initial_sidebar_state="expanded"
)
# Custom CSS with improved styling
st.markdown("""
""", unsafe_allow_html=True)
# Sidebar
st.sidebar.title("🔧 Configuration")
# API Key input
api_key = st.sidebar.text_input(
"Anthropic API Key",
type="password",
help="Enter your Anthropic API key to enable AI chat features"
)
# Stock symbol (locked to CRWV but could be expanded)
symbol = st.sidebar.selectbox("Stock Symbol", ["CRWV"], help="Currently focused on CoreWeave")
# Time range selection
time_range = st.sidebar.selectbox(
"Time Range",
["1D", "5D", "1M", "3M", "6M", "1Y", "2Y"],
index=3
)
# Analysis type
analysis_type = st.sidebar.multiselect(
"Analysis Features",
["Price Chart", "Volume Analysis", "Technical Indicators", "News Feed", "Financial Metrics"],
default=["Price Chart", "Volume Analysis", "News Feed"]
)
# Initialize session state
if 'chat_history' not in st.session_state:
st.session_state.chat_history = []
# Helper functions
@st.cache_data(ttl=300) # Cache for 5 minutes
def get_stock_data(symbol, period):
"""Fetch stock data from Yahoo Finance"""
try:
ticker = yf.Ticker(symbol)
# Map period
period_map = {
"1D": "1d", "5D": "5d", "1M": "1mo",
"3M": "3mo", "6M": "6mo", "1Y": "1y", "2Y": "2y"
}
hist = ticker.history(period=period_map[period])
info = ticker.info
return hist, info
except Exception as e:
st.error(f"Error fetching stock data: {e}")
return None, None
@st.cache_data(ttl=1800) # Cache for 30 minutes
def get_news_data():
"""Fetch news data from multiple sources with improved error handling"""
news_items = []
# Method 1: Try Yahoo Finance news API
try:
ticker = yf.Ticker("CRWV")
news = ticker.news
if news and len(news) > 0:
for item in news[:5]:
title = item.get('title', '').strip()
if title and title != 'No title' and len(title) > 10:
news_items.append({
'title': title,
'link': item.get('link', '#'),
'published': item.get('providerPublishTime', int(time.time())),
'source': item.get('publisher', 'Yahoo Finance')
})
except Exception as e:
print(f"Yahoo Finance news error: {e}")
# Method 2: Try to get general AI/Cloud computing news if CRWV news is sparse
if len(news_items) < 3:
try:
# Get broader market news from yfinance for related tickers
related_tickers = ['NVDA', 'AMZN', 'MSFT', 'GOOGL'] # AI/Cloud related
for ticker_symbol in related_tickers:
try:
ticker = yf.Ticker(ticker_symbol)
news = ticker.news
if news:
for item in news[:2]: # Just get 2 from each
title = item.get('title', '').strip()
if (title and
len(title) > 10 and
any(keyword in title.lower() for keyword in ['ai', 'cloud', 'gpu', 'computing', 'data center'])):
news_items.append({
'title': f"[{ticker_symbol}] {title}",
'link': item.get('link', '#'),
'published': item.get('providerPublishTime', int(time.time())),
'source': item.get('publisher', 'Market News')
})
if len(news_items) >= 5:
break
except:
continue
if len(news_items) >= 5:
break
except Exception as e:
print(f"Related news error: {e}")
# Method 3: Fallback to curated news if APIs fail
if len(news_items) == 0:
current_time = int(time.time())
news_items = [
{
'title': 'CoreWeave Expands GPU Cloud Infrastructure for AI Workloads',
'link': '#',
'published': current_time - 3600,
'source': 'AI News'
},
{
'title': 'GPU Cloud Computing Market Sees Accelerated Growth in 2024',
'link': '#',
'published': current_time - 7200,
'source': 'Tech Report'
},
{
'title': 'Demand for AI Infrastructure Drives Cloud GPU Adoption',
'link': '#',
'published': current_time - 10800,
'source': 'Industry Analysis'
},
{
'title': 'CoreWeave Positions for Growth in High-Performance Computing',
'link': '#',
'published': current_time - 14400,
'source': 'Market Update'
},
{
'title': 'Cloud Infrastructure Companies Benefit from AI Boom',
'link': '#',
'published': current_time - 18000,
'source': 'Financial Times'
}
]
# Sort by most recent first
news_items.sort(key=lambda x: x['published'], reverse=True)
return news_items[:5] # Return top 5
def create_price_chart(hist_data, symbol):
"""Create interactive price chart"""
fig = make_subplots(
rows=2, cols=1,
shared_xaxes=True,
vertical_spacing=0.03,
row_heights=[0.7, 0.3],
subplot_titles=(f'{symbol} Stock Price', 'Volume')
)
# Candlestick chart
fig.add_trace(
go.Candlestick(
x=hist_data.index,
open=hist_data['Open'],
high=hist_data['High'],
low=hist_data['Low'],
close=hist_data['Close'],
name="Price"
),
row=1, col=1
)
# Volume chart
fig.add_trace(
go.Bar(
x=hist_data.index,
y=hist_data['Volume'],
name="Volume",
marker_color='rgba(31, 119, 180, 0.7)'
),
row=2, col=1
)
fig.update_layout(
height=600,
showlegend=False,
xaxis_rangeslider_visible=False
)
return fig
def create_technical_indicators(hist_data):
"""Create technical indicators chart"""
# Calculate moving averages
hist_data['MA20'] = hist_data['Close'].rolling(window=20).mean()
hist_data['MA50'] = hist_data['Close'].rolling(window=50).mean()
# Calculate RSI
delta = hist_data['Close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
hist_data['RSI'] = 100 - (100 / (1 + rs))
fig = make_subplots(
rows=2, cols=1,
shared_xaxes=True,
vertical_spacing=0.03,
subplot_titles=('Price with Moving Averages', 'RSI')
)
# Price and moving averages
fig.add_trace(
go.Scatter(x=hist_data.index, y=hist_data['Close'], name='Close', line=dict(color='blue')),
row=1, col=1
)
fig.add_trace(
go.Scatter(x=hist_data.index, y=hist_data['MA20'], name='MA20', line=dict(color='orange')),
row=1, col=1
)
fig.add_trace(
go.Scatter(x=hist_data.index, y=hist_data['MA50'], name='MA50', line=dict(color='red')),
row=1, col=1
)
# RSI
fig.add_trace(
go.Scatter(x=hist_data.index, y=hist_data['RSI'], name='RSI', line=dict(color='purple')),
row=2, col=1
)
# RSI reference lines
fig.add_hline(y=70, line_dash="dash", line_color="red", row=2, col=1)
fig.add_hline(y=30, line_dash="dash", line_color="green", row=2, col=1)
fig.update_layout(height=500, showlegend=True)
return fig
def get_ai_response(question, stock_data, api_key):
"""Get AI response using Anthropic API"""
if not api_key:
return "Please enter your Anthropic API key in the sidebar to use AI features."
try:
client = anthropic.Anthropic(api_key=api_key)
# Prepare context with current stock data
latest_price = stock_data['Close'].iloc[-1]
prev_price = stock_data['Close'].iloc[-2] if len(stock_data) > 1 else latest_price
daily_change = ((latest_price / prev_price) - 1) * 100 if prev_price != 0 else 0
context = f"""
You are a financial analyst AI assistant specializing in CoreWeave (CRWV) stock analysis.
Current stock data available:
- Latest Close Price: ${latest_price:.2f}
- Daily Change: {daily_change:.2f}%
- Volume: {stock_data['Volume'].iloc[-1]:,}
- 52-week High: ${stock_data['High'].max():.2f}
- 52-week Low: ${stock_data['Low'].min():.2f}
About CoreWeave: CoreWeave is a specialized cloud infrastructure company that provides GPU compute services,
particularly focused on AI/ML workloads, rendering, and high-performance computing.
Please provide helpful, accurate financial analysis and insights. If you don't have specific information,
clearly state your limitations.
"""
# Try multiple model names in order of preference
models_to_try = [
"claude-3-5-sonnet-20241022", # Latest Sonnet 3.5
"claude-3-5-sonnet-20240620", # Previous Sonnet 3.5
"claude-3-sonnet-20240229", # Original Sonnet 3
"claude-3-haiku-20240307" # Fallback to Haiku
]
for model_name in models_to_try:
try:
message = client.messages.create(
model=model_name,
max_tokens=1000,
temperature=0.7,
system=context,
messages=[{"role": "user", "content": question}]
)
return message.content[0].text
except Exception as model_error:
if "not_found_error" in str(model_error):
continue # Try next model
else:
return f"Error with model {model_name}: {str(model_error)}"
return "Unable to connect to AI service. Please check your API key or try again later."
except Exception as e:
error_msg = str(e)
if "authentication" in error_msg.lower():
return "❌ Invalid API key. Please check your Anthropic API key and try again."
elif "rate_limit" in error_msg.lower():
return "⏳ Rate limit exceeded. Please wait a moment and try again."
elif "insufficient" in error_msg.lower():
return "💳 Insufficient credits. Please check your Anthropic account balance."
else:
return f"❌ AI service error: {error_msg}"
# Main app
def main():
# Header
st.markdown('📈 CoreWeave Stock Analysis Dashboard
', unsafe_allow_html=True)
# Fetch data
with st.spinner("Loading stock data..."):
hist_data, stock_info = get_stock_data(symbol, time_range)
if hist_data is None:
st.error("Failed to load stock data. Please try again.")
return
# Key metrics row
col1, col2, col3, col4 = st.columns(4)
current_price = hist_data['Close'].iloc[-1]
prev_close = hist_data['Close'].iloc[-2] if len(hist_data) > 1 else current_price
price_change = current_price - prev_close
percent_change = (price_change / prev_close) * 100
with col1:
st.markdown(
f'${current_price:.2f}
Current Price
{price_change:+.2f}
Change ($)
{percent_change:+.2f}%
Change (%)
{hist_data["Volume"].iloc[-1]:,}
Volume
{title}
{source} - {published_time}
''',
unsafe_allow_html=True
)
else:
st.info("📰 News feed is updating. Please check back shortly.")
# Financial Metrics
if "Financial Metrics" in analysis_type and stock_info:
st.subheader("💰 Key Metrics")
metrics = {
"Market Cap": stock_info.get('marketCap', 'N/A'),
"P/E Ratio": stock_info.get('trailingPE', 'N/A'),
"52W High": f"${stock_info.get('fiftyTwoWeekHigh', 'N/A')}",
"52W Low": f"${stock_info.get('fiftyTwoWeekLow', 'N/A')}",
"Beta": stock_info.get('beta', 'N/A'),
"Dividend Yield": stock_info.get('dividendYield', 'N/A')
}
for key, value in metrics.items():
if value != 'N/A' and isinstance(value, (int, float)):
if key == "Market Cap" and value > 1e9:
value = f"${value/1e9:.2f}B"
elif key in ["P/E Ratio", "Beta"]:
value = f"{value:.2f}"
elif key == "Dividend Yield":
value = f"{value*100:.2f}%" if value else "N/A"
st.write(f"**{key}:** {value}")
# AI Chat Interface
st.markdown("---")
st.subheader("🤖 AI Stock Analyst")
# API Key status
if api_key:
st.success("✅ API key provided - AI features enabled")
else:
st.warning("⚠️ Please enter your Anthropic API key in the sidebar to enable AI chat")
# Chat interface
chat_container = st.container()
# Display chat history
with chat_container:
for message in st.session_state.chat_history:
if message['role'] == 'user':
st.markdown(
f'You: {message["content"]}
',
unsafe_allow_html=True
)
else:
st.markdown(
f'AI Analyst: {message["content"]}
',
unsafe_allow_html=True
)
# Chat input
user_question = st.text_input(
"Ask me anything about CoreWeave stock:",
placeholder="e.g., What's your analysis of the current price trend?",
disabled=not api_key
)
col_send, col_clear = st.columns([1, 1])
with col_send:
if st.button("Send", type="primary", disabled=not api_key) and user_question:
# Add user message to history
st.session_state.chat_history.append({
'role': 'user',
'content': user_question
})
# Get AI response
with st.spinner("AI is analyzing..."):
ai_response = get_ai_response(user_question, hist_data, api_key)
# Add AI response to history
st.session_state.chat_history.append({
'role': 'assistant',
'content': ai_response
})
st.rerun()
with col_clear:
if st.button("Clear Chat"):
st.session_state.chat_history = []
st.rerun()
# Sample questions
st.markdown("**💡 Try asking:**")
sample_questions = [
"What's your technical analysis of CoreWeave?",
"Should I buy, hold, or sell CRWV?",
"How does CoreWeave compare to other cloud companies?",
"What are the key risks for CoreWeave?",
"Explain the recent price movement"
]
cols = st.columns(len(sample_questions))
for i, question in enumerate(sample_questions):
with cols[i]:
if st.button(question, key=f"sample_{i}", disabled=not api_key):
st.session_state.chat_history.append({'role': 'user', 'content': question})
ai_response = get_ai_response(question, hist_data, api_key)
st.session_state.chat_history.append({'role': 'assistant', 'content': ai_response})
st.rerun()
if __name__ == "__main__":
main()