Trading_Guru / app.py
shaheerawan3's picture
Update app.py
47fa6c2 verified
# At the top of app.py, make sure you have all these imports:
import streamlit as st
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import plotly.graph_objects as go
import plotly.express as px
from typing import Dict, List, Optional, Tuple
import yfinance as yf
from prophet import Prophet
import tensorflow as tf
from scipy import stats
import seaborn as sns
import ta
from concurrent.futures import ThreadPoolExecutor
import logging
logging.basicConfig(level=logging.INFO)
import joblib
import warnings
warnings.filterwarnings('ignore')
import concurrent.futures
import time
import requests
from data_fetcher import EnhancedDataFetcher
# Import custom modules
from models.trading_models import EnhancedTradingModels
from config import Config
from models.technical_analysis import EnhancedTechnicalAnalysis
from models.portfolio_optimization import PortfolioOptimizer
from models.risk_metrics import RiskMetrics
from models.pattern_recognition import PatternRecognition
class DataProcessor:
@staticmethod
def prepare_data(data: pd.DataFrame) -> pd.DataFrame:
"""Prepare and validate data for analysis."""
try:
# Ensure data has the correct columns
required_columns = ['Date', 'Open', 'High', 'Low', 'Close', 'Volume']
# Reset index if Date is in index
if isinstance(data.index, pd.DatetimeIndex):
data = data.reset_index()
# Rename columns if they're uppercase
data.columns = [col.capitalize() if col.isupper() else col for col in data.columns]
# Ensure Date column is datetime
data['Date'] = pd.to_datetime(data['Date'])
# Convert price columns to float
numeric_columns = ['Open', 'High', 'Low', 'Close', 'Volume']
for col in numeric_columns:
if col in data.columns:
data[col] = pd.to_numeric(data[col], errors='coerce')
# Forward fill missing values
data = data.fillna(method='ffill')
# Drop any remaining NaN values
data = data.dropna()
return data
except Exception as e:
print(f"Error preparing data: {str(e)}")
return pd.DataFrame()
# Add this to your app.py file
# Key fixes in the DataManager class
class DataManager:
@staticmethod
def load_data(ticker: str, start_date: str) -> Optional[pd.DataFrame]:
"""Load financial data with improved error handling and retry logic."""
try:
max_retries = 3
for attempt in range(max_retries):
try:
# Handle crypto tickers
if 'USD' in ticker:
symbol = ticker
else:
symbol = ticker
# Download data
data = yf.download(
symbol,
start=start_date,
end=datetime.now().strftime('%Y-%m-%d'),
progress=False,
interval='1d'
)
if data.empty:
print(f"Attempt {attempt + 1}: No data received for {ticker}")
if attempt < max_retries - 1:
time.sleep(2)
continue
return None
# Process the data
data = data.reset_index()
# Standardize column names
data.columns = [col if col == 'Date' else col.title() for col in data.columns]
# Ensure datetime format
data['Date'] = pd.to_datetime(data['Date'])
# Add Adj Close if missing
if 'Adj Close' not in data.columns:
data['Adj Close'] = data['Close']
# Forward fill any missing values
data = data.fillna(method='ffill').fillna(method='bfill')
return data
except Exception as e:
print(f"Attempt {attempt + 1} failed for {ticker}: {str(e)}")
if attempt < max_retries - 1:
time.sleep(2)
continue
raise e
except Exception as e:
print(f"Failed to load data for {ticker}: {str(e)}")
return None
def validate_data(data: pd.DataFrame) -> bool:
"""Validate the loaded data."""
required_columns = ['Date', 'Open', 'High', 'Low', 'Close', 'Volume']
# Check for required columns
if not all(col in data.columns for col in required_columns):
print(f"Missing columns. Available columns: {data.columns.tolist()}")
return False
# Check for minimum data points
if len(data) < 30: # Require at least 30 days of data
print(f"Insufficient data points: {len(data)}")
return False
# Check for missing values
if data[required_columns].isnull().any().any():
print("Data contains missing values")
return False
return True
class AdvancedVisualization:
@staticmethod
def plot_candlestick_with_patterns(data: pd.DataFrame, patterns: pd.DataFrame, title: str) -> go.Figure:
"""Create advanced candlestick chart with pattern annotations."""
fig = go.Figure()
# Add candlestick
fig.add_trace(go.Candlestick(
x=data['Date'],
open=data['Open'],
high=data['High'],
low=data['Low'],
close=data['Close'],
name='Price'
))
# Add patterns
for pattern in ['Double_Top', 'Double_Bottom', 'Head_And_Shoulders', 'Triangle', 'Channel']:
if pattern in patterns.columns:
pattern_points = patterns[patterns[pattern] == 1]
if not pattern_points.empty:
fig.add_trace(go.Scatter(
x=pattern_points['Date'],
y=pattern_points['High'],
mode='markers',
name=pattern,
marker=dict(size=10, symbol='triangle-down')
))
# Update layout
fig.update_layout(
title=title,
yaxis_title='Price',
template='plotly_dark' if st.session_state.theme == 'dark' else 'plotly_white',
xaxis_rangeslider_visible=False,
height=600
)
return fig
@staticmethod
def plot_correlation_matrix(returns: pd.DataFrame) -> go.Figure:
"""Create correlation matrix heatmap."""
corr = returns.corr()
fig = px.imshow(
corr,
labels=dict(color="Correlation"),
x=corr.columns,
y=corr.columns,
color_continuous_scale='RdBu'
)
fig.update_layout(
title='Asset Correlation Matrix',
template='plotly_dark' if st.session_state.theme == 'dark' else 'plotly_white'
)
return fig
class TradingStrategy:
def __init__(self, data: pd.DataFrame):
self.data = DataProcessor.prepare_data(data.copy())
self.signals = pd.DataFrame(index=self.data.index)
def generate_signals(self) -> pd.DataFrame:
"""Generate trading signals based on multiple strategies."""
try:
# Calculate technical indicators
self.data['SMA20'] = self.data['Close'].rolling(window=20).mean()
self.data['SMA50'] = self.data['Close'].rolling(window=50).mean()
self.data['RSI'] = self._calculate_rsi()
# Generate individual signals
self.signals['MA_Cross'] = self._moving_average_crossover()
self.signals['RSI'] = self._rsi_signals()
self.signals['Trend'] = self._trend_signals()
# Combine signals
self.signals['Position'] = 0
# Generate combined signals
buy_condition = (
(self.signals['MA_Cross'] == 1) &
(self.signals['RSI'] == 1) |
(self.signals['Trend'] == 1)
)
sell_condition = (
(self.signals['MA_Cross'] == -1) &
(self.signals['RSI'] == -1) |
(self.signals['Trend'] == -1)
)
self.signals.loc[buy_condition, 'Position'] = 1
self.signals.loc[sell_condition, 'Position'] = -1
return self.signals
except Exception as e:
print(f"Error generating signals: {str(e)}")
return pd.DataFrame()
def _calculate_rsi(self, periods=14):
"""Calculate RSI indicator."""
delta = self.data['Close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=periods).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=periods).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
return rsi
def _moving_average_crossover(self) -> pd.Series:
"""Generate signals based on MA crossover."""
signals = pd.Series(0, index=self.data.index)
signals.loc[self.data['SMA20'] > self.data['SMA50']] = 1
signals.loc[self.data['SMA20'] < self.data['SMA50']] = -1
return signals
def _rsi_signals(self) -> pd.Series:
"""Generate signals based on RSI."""
signals = pd.Series(0, index=self.data.index)
signals.loc[self.data['RSI'] < 30] = 1 # Oversold
signals.loc[self.data['RSI'] > 70] = -1 # Overbought
return signals
def _trend_signals(self) -> pd.Series:
"""Generate signals based on trend."""
signals = pd.Series(0, index=self.data.index)
price_sma = self.data['Close'] > self.data['SMA50']
signals.loc[price_sma] = 1
signals.loc[~price_sma] = -1
return signals
# In the TradingStrategy class, update the _bollinger_signals method:
def _bollinger_signals(self) -> pd.Series:
"""Generate signals based on Bollinger Bands."""
# Calculate Bollinger Bands using ta library
indicator_bb = ta.volatility.BollingerBands(close=self.data['Close'], window=20, window_dev=2)
upper = indicator_bb.bollinger_hband()
lower = indicator_bb.bollinger_lband()
signals = pd.Series(0, index=self.data.index)
signals[self.data['Close'] > upper] = -1 # Sell signal
signals[self.data['Close'] < lower] = 1 # Buy signal
return signals
def _pattern_signals(self) -> pd.Series:
"""Generate signals based on pattern recognition."""
pattern_recognition = PatternRecognition(self.data)
patterns = pattern_recognition.detect_patterns()
signals = pd.Series(0, index=self.data.index)
signals[patterns['Double_Top'] == 1] = -1
signals[patterns['Double_Bottom'] == 1] = 1
return signals
def initialize_session_state():
"""Initialize session state variables."""
if 'theme' not in st.session_state:
st.session_state.theme = Config.THEME
if 'selected_assets' not in st.session_state:
st.session_state.selected_assets = []
if 'optimization_results' not in st.session_state:
st.session_state.optimization_results = None
def render_sidebar() -> Tuple[str, str, List[str]]:
"""Render enhanced sidebar with additional controls."""
st.sidebar.title("โš™๏ธ Configuration")
# Theme selection
theme = st.sidebar.selectbox(
"Theme",
["light", "dark"],
key="theme"
)
# Asset selection
asset_type = st.sidebar.radio("Select Asset Type", list(Config.ASSETS.keys()))
# Multi-asset selection
selected_assets = st.sidebar.multiselect(
'Select Assets for Analysis',
options=list(Config.ASSETS[asset_type].keys()),
format_func=lambda x: f"{x} ({Config.ASSETS[asset_type][x]})"
)
# Analysis Options
st.sidebar.subheader("Analysis Options")
st.sidebar.checkbox("Technical Analysis", value=True)
st.sidebar.checkbox("Pattern Recognition", value=True)
st.sidebar.checkbox("Portfolio Optimization", value=True)
st.sidebar.checkbox("Risk Metrics", value=True)
return theme, asset_type, selected_assets
# Add this function before the main() function:
def plot_signals(data: pd.DataFrame, signals: pd.DataFrame, asset: str) -> go.Figure:
"""Plot trading signals on candlestick chart."""
fig = go.Figure()
# Add candlestick
fig.add_trace(go.Candlestick(
x=data['Date'],
open=data['Open'],
high=data['High'],
low=data['Low'],
close=data['Close'],
name='Price'
))
# Add buy signals
buy_points = data[signals['Position'] == 1]
if not buy_points.empty:
fig.add_trace(go.Scatter(
x=buy_points['Date'],
y=buy_points['High'],
mode='markers',
name='Buy Signal',
marker=dict(
symbol='triangle-up',
size=15,
color='green',
)
))
# Add sell signals
sell_points = data[signals['Position'] == -1]
if not sell_points.empty:
fig.add_trace(go.Scatter(
x=sell_points['Date'],
y=sell_points['Low'],
mode='markers',
name='Sell Signal',
marker=dict(
symbol='triangle-down',
size=15,
color='red',
)
))
# Update layout
fig.update_layout(
title=f"{asset} Trading Signals",
yaxis_title='Price',
xaxis_title='Date',
template='plotly_dark' if st.session_state.theme == 'dark' else 'plotly_white',
xaxis_rangeslider_visible=False,
height=600
)
return fig
def main():
data_manager = DataManager()
# Initialize session state
initialize_session_state()
# Page config
st.set_page_config(
page_title=Config.APP_TITLE,
page_icon=Config.APP_ICON,
layout="wide"
)
# Custom CSS
st.markdown("""
<style>
.stButton>button {
width: 100%;
}
.reportview-container {
background: var(--background-color)
}
.sidebar .sidebar-content {
background: var(--background-color)
}
</style>
""", unsafe_allow_html=True)
# Sidebar
theme, asset_type, selected_assets = render_sidebar()
# Main content
st.title(f"{Config.APP_TITLE} - Advanced Market Analysis")
if not selected_assets:
st.warning("Please select at least one asset to analyze.")
return
failed_assets = []
# Tabs for different analyses
tabs = st.tabs([
"๐Ÿ“Š Technical Analysis",
"๐ŸŽฏ Pattern Recognition",
"๐Ÿ’ผ Portfolio Optimization",
"โš ๏ธ Risk Analysis",
"๐Ÿ“ˆ Trading Signals"
])
# Load data for selected assets
with st.spinner('Loading data...'):
data_dict = {}
returns_dict = {}
failed_assets = []
# Use shorter timeframe for initial load
start_date = (datetime.now() - timedelta(days=365)).strftime('%Y-%m-%d')
for asset in selected_assets:
try:
st.text(f"Loading data for {asset}...") # Debug output
# Get the data
data = data_manager.load_data(asset, start_date)
if data is not None and not data.empty:
# Basic validation
required_cols = ['Date', 'Open', 'High', 'Low', 'Close', 'Volume']
if all(col in data.columns for col in required_cols):
data_dict[asset] = data
returns_dict[asset] = data['Close'].pct_change().dropna()
st.success(f"Successfully loaded data for {asset}")
else:
failed_assets.append(asset)
st.error(f"Missing required columns for {asset}")
else:
failed_assets.append(asset)
st.error(f"No data received for {asset}")
except Exception as e:
failed_assets.append(asset)
st.error(f"Error loading data for {asset}: {str(e)}")
continue
if not data_dict:
st.error("No data available for analysis. Please try different assets or check your internet connection.")
return
# Portfolio Optimization Tab - FIX 5: Add proper error handling and data validation
def handle_portfolio_optimization(tabs, data_dict, returns_dict):
with tabs[2]:
st.subheader("Portfolio Optimization")
if len(data_dict) < 2:
st.warning("Please select at least 2 assets for portfolio optimization.")
return
try:
# Create returns DataFrame with proper index
returns_df = pd.DataFrame(returns_dict)
# Initialize optimizer
optimizer = PortfolioOptimizer(returns_df)
# Calculate reasonable bounds for target return
min_return = max(returns_df.mean().min() * 252, -0.5) # Limit to -50%
max_return = min(returns_df.mean().max() * 252, 0.5) # Limit to +50%
col1, col2 = st.columns(2)
with col1:
target_return = st.slider(
"Target Annual Return (%)",
min_value=int(min_return * 100),
max_value=int(max_return * 100),
value=10
) / 100
# Run optimization
optimization_results = optimizer.optimize_portfolio(target_return)
if optimization_results['success']:
st.success("Portfolio optimization completed successfully!")
# Display optimization results
st.subheader("Optimal Portfolio Allocation")
# Plot allocation pie chart
fig = px.pie(
values=list(optimization_results['weights'].values()),
names=list(optimization_results['weights'].keys()),
title="Optimal Portfolio Weights"
)
st.plotly_chart(fig, use_container_width=True)
# Display metrics
col1, col2, col3 = st.columns(3)
col1.metric("Expected Annual Return", f"{optimization_results['return']:.2%}")
col2.metric("Annual Volatility", f"{optimization_results['volatility']:.2%}")
col3.metric("Sharpe Ratio", f"{optimization_results['sharpe_ratio']:.2f}")
else:
st.warning(f"Optimization failed: {optimization_results['message']}")
except Exception as e:
st.error(f"Portfolio optimization error: {str(e)}")
st.info("Try selecting different assets or adjusting the target return.")
# Trading Signals Tab - FIX 6: Add proper signal generation
# In your main() function, replace the trading signals tab section with:
# Inside the `Trading Signals` Tab
with tabs[4]:
st.subheader("Trading Signals")
for asset, data in data_dict.items():
try:
st.subheader(f"Trading Signals - {asset}")
# Use EnhancedTradingModels to generate signals and metrics
model = EnhancedTradingModels(data)
signals, metrics = model.generate_signals()
if signals.empty:
st.warning(f"No trading signals generated for {asset}")
continue
# Plot signals
fig = plot_signals(data, signals, asset)
st.plotly_chart(fig, use_container_width=True)
# Display signal statistics
st.subheader("Signal Metrics")
metrics_df = pd.DataFrame.from_dict(metrics, orient='index', columns=['Value'])
st.dataframe(metrics_df)
except Exception as e:
st.error(f"Error analyzing {asset}: {str(e)}")
# Footer
st.markdown("""
---
### Trading Guru Pro - Advanced Market Analysis Platform
Features:
- Real-time technical analysis
- Pattern recognition with machine learning
- Portfolio optimization
- Risk metrics and analysis
- Automated trading signals
Created by Muhammad Shaheer
# Continuing main.py...
๐Ÿ“ง For support and feature requests: [Your Contact Info]
*Disclaimer: This tool is for educational purposes only. Always conduct your own research before making investment decisions.*
""")
# Add helpful tooltips
st.sidebar.markdown("""
### ๐Ÿ“š Analysis Guide
#### Technical Indicators
- **RSI**: Measures momentum (Oversold < 30, Overbought > 70)
- **MACD**: Shows trend direction and momentum
- **Bollinger Bands**: Indicates volatility and potential reversals
#### Pattern Recognition
- **Double Top/Bottom**: Potential reversal patterns
- **Head & Shoulders**: Major reversal pattern
- **Triangles**: Continuation or reversal patterns
#### Portfolio Metrics
- **Sharpe Ratio**: Risk-adjusted return measure
- **Max Drawdown**: Largest peak-to-trough decline
- **VaR**: Potential loss in value
#### Trading Signals
- **Green Triangle**: Buy signal
- **Red Triangle**: Sell signal
*Remember to use multiple indicators for confirmation*
""")
class BacktestEngine:
"""Backtesting engine for trading strategies."""
def __init__(self, data: pd.DataFrame, initial_capital: float = 100000):
self.data = data
self.initial_capital = initial_capital
self.positions = pd.DataFrame(index=data.index).fillna(0)
self.portfolio_value = pd.DataFrame(index=data.index).fillna(0)
def run_backtest(self, signals: pd.DataFrame) -> Dict:
"""Run backtest simulation."""
# Initialize portfolio
portfolio = self.initial_capital
position = 0
trades = []
for i in range(len(self.data)):
if signals['Position'].iloc[i] == 1 and position == 0: # Buy signal
shares = portfolio / self.data['Close'].iloc[i]
position = shares
trades.append({
'date': self.data['Date'].iloc[i],
'type': 'buy',
'price': self.data['Close'].iloc[i],
'shares': shares,
'value': portfolio
})
elif signals['Position'].iloc[i] == -1 and position > 0: # Sell signal
portfolio = position * self.data['Close'].iloc[i]
position = 0
trades.append({
'date': self.data['Date'].iloc[i],
'type': 'sell',
'price': self.data['Close'].iloc[i],
'shares': shares,
'value': portfolio
})
# Update portfolio value
self.portfolio_value.iloc[i] = portfolio if position == 0 else position * self.data['Close'].iloc[i]
# Calculate performance metrics
returns = self.portfolio_value.pct_change().dropna()
metrics = {
'total_return': (self.portfolio_value.iloc[-1] - self.initial_capital) / self.initial_capital,
'annual_return': returns.mean() * 252,
'volatility': returns.std() * np.sqrt(252),
'sharpe_ratio': (returns.mean() - Config.RISK_FREE_RATE/252) / (returns.std()) * np.sqrt(252),
'max_drawdown': (self.portfolio_value / self.portfolio_value.cummax() - 1).min(),
'trades': trades
}
return metrics
class MarketSentiment:
"""Analyze market sentiment using various indicators."""
@staticmethod
def calculate_sentiment_score(data: pd.DataFrame) -> float:
"""Calculate overall market sentiment score."""
# Technical indicators sentiment
rsi = ta.momentum.RSIIndicator(data['Close']).rsi().iloc[-1]
macd = ta.trend.MACD(data['Close']).macd().iloc[-1]
# Volume sentiment
volume_ma = data['Volume'].rolling(window=20).mean().iloc[-1]
current_volume = data['Volume'].iloc[-1]
volume_sentiment = 1 if current_volume > volume_ma else -1
# Trend sentiment
sma_20 = data['Close'].rolling(window=20).mean().iloc[-1]
sma_50 = data['Close'].rolling(window=50).mean().iloc[-1]
trend_sentiment = 1 if sma_20 > sma_50 else -1
# Combine sentiments
technical_sentiment = (
(1 if 40 < rsi < 60 else -1) +
(1 if macd > 0 else -1) +
volume_sentiment +
trend_sentiment
) / 4
return technical_sentiment
def add_market_sentiment_analysis(data_dict: Dict[str, pd.DataFrame]) -> None:
"""Add market sentiment analysis to the app."""
st.subheader("๐Ÿ“Š Market Sentiment Analysis")
sentiment_scores = {}
for asset, data in data_dict.items():
sentiment_score = MarketSentiment.calculate_sentiment_score(data)
sentiment_scores[asset] = sentiment_score
# Create sentiment gauge charts
cols = st.columns(len(sentiment_scores))
for i, (asset, score) in enumerate(sentiment_scores.items()):
with cols[i]:
fig = go.Figure(go.Indicator(
mode="gauge+number",
value=score * 100,
domain={'x': [0, 1], 'y': [0, 1]},
gauge={
'axis': {'range': [-100, 100]},
'bar': {'color': "darkblue"},
'steps': [
{'range': [-100, -33], 'color': "red"},
{'range': [-33, 33], 'color': "gray"},
{'range': [33, 100], 'color': "green"}
],
}
))
fig.update_layout(
title=f"{asset} Sentiment",
height=250
)
st.plotly_chart(fig, use_container_width=True)
def add_portfolio_metrics(returns_dict: Dict[str, pd.Series]) -> None:
"""Add detailed portfolio metrics analysis."""
st.subheader("๐Ÿ“ˆ Portfolio Performance Metrics")
# Calculate metrics for each asset
metrics_dict = {}
for asset, returns in returns_dict.items():
metrics = {
'Annual Return': returns.mean() * 252,
'Volatility': returns.std() * np.sqrt(252),
'Sharpe Ratio': (returns.mean() * 252 - Config.RISK_FREE_RATE) / (returns.std() * np.sqrt(252)),
'Sortino Ratio': (returns.mean() * 252 - Config.RISK_FREE_RATE) / (returns[returns < 0].std() * np.sqrt(252)),
'Max Drawdown': (returns.cumsum() - returns.cumsum().cummax()).min(),
'Win Rate': len(returns[returns > 0]) / len(returns)
}
metrics_dict[asset] = metrics
# Create metrics table
metrics_df = pd.DataFrame(metrics_dict).T
# Style the dataframe
styled_df = metrics_df.style.format({
'Annual Return': '{:.2%}',
'Volatility': '{:.2%}',
'Sharpe Ratio': '{:.2f}',
'Sortino Ratio': '{:.2f}',
'Max Drawdown': '{:.2%}',
'Win Rate': '{:.2%}'
}).background_gradient(cmap='RdYlGn')
st.dataframe(styled_df)
if __name__ == "__main__":
main()