brad/LLM_Engineering_OLD
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

week4: ai stock trading

Browse Source
This commit is contained in:
Omar Marie
2025-07-21 20:49:24 +03:00
parent 59f4a5be27
commit c3d0a8e8b7
10 changed files with 3043 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
week4/community-contributions/ai_stock_trading/tools/__init__.py Normal file
View File

@@ -0,0 +1,28 @@
"""
AI Stock Trading Tools
This package contains all the core tools for the AI Stock Trading platform:
- fetching: Stock data fetching and market data
- analysis: Technical analysis and stock metrics
- trading_decisions: AI-powered trading recommendations
- sharia_compliance: Islamic finance compliance checking
- charting: Interactive charts and visualizations
"""
__version__ = "1.0.0"
__author__ = "AI Stock Trading Platform"
# Import main classes and functions for easy access
from .fetching import StockDataFetcher, stock_fetcher, fetch_stock_data, get_available_stocks
from .analysis import StockAnalyzer, stock_analyzer, analyze_stock
from .trading_decisions import TradingDecisionEngine, trading_engine, get_trading_recommendation
from .sharia_compliance import ShariaComplianceChecker, sharia_checker, check_sharia_compliance
from .charting import StockChartGenerator, chart_generator, create_price_chart
__all__ = [
'StockDataFetcher', 'stock_fetcher', 'fetch_stock_data', 'get_available_stocks',
'StockAnalyzer', 'stock_analyzer', 'analyze_stock',
'TradingDecisionEngine', 'trading_engine', 'get_trading_recommendation',
'ShariaComplianceChecker', 'sharia_checker', 'check_sharia_compliance',
'StockChartGenerator', 'chart_generator', 'create_price_chart'
]

316
week4/community-contributions/ai_stock_trading/tools/analysis.py Normal file
View File

@@ -0,0 +1,316 @@
"""
Stock Analysis Module
This module provides enhanced technical and fundamental analysis capabilities
for stock data with advanced metrics and indicators.
"""
import pandas as pd
import numpy as np
from typing import Dict, List, Optional, Tuple, Union, Any
import warnings
warnings.filterwarnings('ignore')
class StockAnalyzer:
"""Enhanced stock analyzer with comprehensive technical indicators"""
def __init__(self):
pass
def analyze_stock(self, data: pd.DataFrame) -> Dict:
"""
Comprehensive stock analysis with enhanced metrics
Args:
data: DataFrame with OHLCV stock data
Returns:
Dictionary with analysis results
"""
if data.empty:
return {'error': 'No data provided for analysis'}
try:
analysis = {}
# Basic price metrics
analysis.update(self._calculate_price_metrics(data))
# Technical indicators
analysis.update(self._calculate_technical_indicators(data))
# Volatility analysis
analysis.update(self._calculate_volatility_metrics(data))
# Volume analysis
analysis.update(self._calculate_volume_metrics(data))
# Trend analysis
analysis.update(self._calculate_trend_metrics(data))
# Risk metrics
analysis.update(self._calculate_risk_metrics(data))
# Performance metrics
analysis.update(self._calculate_performance_metrics(data))
return analysis
except Exception as e:
return {'error': f'Analysis failed: {str(e)}'}
def _calculate_price_metrics(self, data: pd.DataFrame) -> Dict:
"""Calculate basic price metrics"""
close_prices = data['Close']
return {
'current_price': float(close_prices.iloc[-1]),
'start_price': float(close_prices.iloc[0]),
'max_price': float(close_prices.max()),
'min_price': float(close_prices.min()),
'price_range_pct': float(((close_prices.max() - close_prices.min()) / close_prices.min()) * 100),
'total_return_pct': float(((close_prices.iloc[-1] - close_prices.iloc[0]) / close_prices.iloc[0]) * 100)
}
def _calculate_technical_indicators(self, data: pd.DataFrame) -> Dict:
"""Calculate technical indicators"""
close_prices = data['Close']
high_prices = data['High']
low_prices = data['Low']
indicators = {}
# Moving averages
if len(data) >= 20:
sma_20 = close_prices.rolling(window=20).mean()
indicators['sma_20'] = float(sma_20.iloc[-1])
indicators['price_vs_sma_20'] = float(((close_prices.iloc[-1] - sma_20.iloc[-1]) / sma_20.iloc[-1]) * 100)
if len(data) >= 50:
sma_50 = close_prices.rolling(window=50).mean()
indicators['sma_50'] = float(sma_50.iloc[-1])
indicators['price_vs_sma_50'] = float(((close_prices.iloc[-1] - sma_50.iloc[-1]) / sma_50.iloc[-1]) * 100)
# Exponential Moving Average
if len(data) >= 12:
ema_12 = close_prices.ewm(span=12).mean()
indicators['ema_12'] = float(ema_12.iloc[-1])
# RSI (Relative Strength Index)
if len(data) >= 14:
rsi = self._calculate_rsi(pd.Series(close_prices), 14)
indicators['rsi'] = float(rsi.iloc[-1])
indicators['rsi_signal'] = self._interpret_rsi(float(rsi.iloc[-1]))
# MACD
if len(data) >= 26:
macd_line, signal_line, histogram = self._calculate_macd(pd.Series(close_prices))
indicators['macd'] = float(macd_line.iloc[-1])
indicators['macd_signal'] = float(signal_line.iloc[-1])
indicators['macd_histogram'] = float(histogram.iloc[-1])
indicators['macd_trend'] = 'bullish' if float(histogram.iloc[-1]) > 0 else 'bearish'
# Bollinger Bands
if len(data) >= 20:
bb_upper, bb_middle, bb_lower = self._calculate_bollinger_bands(pd.Series(close_prices), 20, 2)
indicators['bb_upper'] = float(bb_upper.iloc[-1])
indicators['bb_middle'] = float(bb_middle.iloc[-1])
indicators['bb_lower'] = float(bb_lower.iloc[-1])
indicators['bb_position'] = self._interpret_bollinger_position(float(close_prices.iloc[-1]), float(bb_upper.iloc[-1]), float(bb_lower.iloc[-1]))
return indicators
def _calculate_volatility_metrics(self, data: pd.DataFrame) -> Dict:
"""Calculate volatility metrics"""
close_prices = data['Close']
daily_returns = close_prices.pct_change().dropna()
return {
'volatility_daily': float(daily_returns.std() * 100),
'volatility_annualized': float(daily_returns.std() * np.sqrt(252) * 100),
'avg_daily_return': float(daily_returns.mean() * 100),
'max_daily_gain': float(daily_returns.max() * 100),
'max_daily_loss': float(daily_returns.min() * 100)
}
def _calculate_volume_metrics(self, data: pd.DataFrame) -> Dict:
"""Calculate volume metrics"""
volume = data['Volume']
metrics: Dict[str, Union[float, str]] = {
'avg_volume': float(volume.mean()),
'current_volume': float(volume.iloc[-1]),
'max_volume': float(volume.max()),
'min_volume': float(volume.min())
}
# Volume trend
if len(volume) >= 10:
recent_avg = volume.tail(10).mean()
overall_avg = volume.mean()
if recent_avg > overall_avg:
metrics['volume_trend'] = 'increasing'
else:
metrics['volume_trend'] = 'decreasing'
metrics['volume_vs_avg'] = float(((recent_avg - overall_avg) / overall_avg) * 100)
return metrics
def _calculate_trend_metrics(self, data: pd.DataFrame) -> Dict:
"""Calculate trend analysis metrics"""
close_prices = data['Close']
# Linear regression for trend
x = np.arange(len(close_prices))
slope, intercept = np.polyfit(x, close_prices, 1)
# Trend strength
correlation = np.corrcoef(x, close_prices)[0, 1]
return {
'trend_slope': float(slope),
'trend_direction': 'upward' if slope > 0 else 'downward',
'trend_strength': float(abs(correlation)),
'trend_angle': float(np.degrees(np.arctan(slope))),
'r_squared': float(correlation ** 2)
}
def _calculate_risk_metrics(self, data: pd.DataFrame) -> Dict:
"""Calculate risk metrics"""
close_prices = data['Close']
daily_returns = close_prices.pct_change().dropna()
# Value at Risk (VaR)
var_95 = np.percentile(daily_returns, 5)
var_99 = np.percentile(daily_returns, 1)
# Maximum Drawdown
cumulative_returns = (1 + daily_returns).cumprod()
running_max = cumulative_returns.expanding().max()
drawdown = (cumulative_returns - running_max) / running_max
max_drawdown = drawdown.min()
# Sharpe Ratio (assuming risk-free rate of 2%)
risk_free_rate = 0.02 / 252 # Daily risk-free rate
excess_returns = daily_returns - risk_free_rate
sharpe_ratio = excess_returns.mean() / daily_returns.std() if daily_returns.std() != 0 else 0
return {
'var_95': float(var_95 * 100),
'var_99': float(var_99 * 100),
'max_drawdown': float(max_drawdown * 100),
'sharpe_ratio': float(sharpe_ratio * np.sqrt(252)), # Annualized
'downside_deviation': float(daily_returns[daily_returns < 0].std() * 100)
}
def _calculate_performance_metrics(self, data: pd.DataFrame) -> Dict:
"""Calculate performance metrics"""
close_prices = data['Close']
# Different period returns
periods = {
'1_week': min(5, len(close_prices) - 1),
'1_month': min(22, len(close_prices) - 1),
'3_months': min(66, len(close_prices) - 1),
'6_months': min(132, len(close_prices) - 1)
}
performance = {}
current_price = close_prices.iloc[-1]
for period_name, days_back in periods.items():
if days_back > 0:
past_price = close_prices.iloc[-(days_back + 1)]
return_pct = ((current_price - past_price) / past_price) * 100
performance[f'return_{period_name}'] = float(return_pct)
return performance
def _calculate_rsi(self, prices: pd.Series, period: int = 14) -> pd.Series:
"""Calculate Relative Strength Index"""
delta = prices.diff()
gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
return rsi
def _interpret_rsi(self, rsi_value: float) -> str:
"""Interpret RSI value"""
if rsi_value >= 70:
return 'overbought'
elif rsi_value <= 30:
return 'oversold'
else:
return 'neutral'
def _calculate_macd(self, prices: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9) -> Tuple[pd.Series, pd.Series, pd.Series]:
"""Calculate MACD indicator"""
ema_fast = prices.ewm(span=fast).mean()
ema_slow = prices.ewm(span=slow).mean()
macd_line = ema_fast - ema_slow
signal_line = macd_line.ewm(span=signal).mean()
histogram = macd_line - signal_line
return macd_line, signal_line, histogram
def _calculate_bollinger_bands(self, prices: pd.Series, period: int = 20, std_dev: int = 2) -> Tuple[pd.Series, pd.Series, pd.Series]:
"""Calculate Bollinger Bands"""
sma = prices.rolling(window=period).mean()
std = prices.rolling(window=period).std()
upper_band = sma + (std * std_dev)
lower_band = sma - (std * std_dev)
return upper_band, sma, lower_band
def _interpret_bollinger_position(self, current_price: float, upper_band: float, lower_band: float) -> str:
"""Interpret position relative to Bollinger Bands"""
if current_price > upper_band:
return 'above_upper_band'
elif current_price < lower_band:
return 'below_lower_band'
else:
return 'within_bands'
def get_analysis_summary(self, analysis: Dict) -> str:
"""Generate a human-readable analysis summary"""
if 'error' in analysis:
return f"Analysis Error: {analysis['error']}"
summary = []
# Price summary
current_price = analysis.get('current_price', 0)
total_return = analysis.get('total_return_pct', 0)
summary.append(f"Current Price: ${current_price:.2f}")
summary.append(f"Total Return: {total_return:.2f}%")
# Trend
trend_direction = analysis.get('trend_direction', 'unknown')
trend_strength = analysis.get('trend_strength', 0)
summary.append(f"Trend: {trend_direction.title()} (Strength: {trend_strength:.2f})")
# Technical indicators
if 'rsi' in analysis:
rsi = analysis['rsi']
rsi_signal = analysis['rsi_signal']
summary.append(f"RSI: {rsi:.1f} ({rsi_signal})")
if 'macd_trend' in analysis:
macd_trend = analysis['macd_trend']
summary.append(f"MACD: {macd_trend}")
# Risk
volatility = analysis.get('volatility_annualized', 0)
max_drawdown = analysis.get('max_drawdown', 0)
summary.append(f"Volatility: {volatility:.1f}% (Annual)")
summary.append(f"Max Drawdown: {max_drawdown:.1f}%")
return "\n".join(summary)
# Global instance for easy import
stock_analyzer = StockAnalyzer()
# Convenience function
def analyze_stock(data: pd.DataFrame) -> Dict:
"""Convenience function to analyze stock data"""
return stock_analyzer.analyze_stock(data)

483
week4/community-contributions/ai_stock_trading/tools/charting.py Normal file
View File

@@ -0,0 +1,483 @@
"""
Charting Module
This module provides comprehensive charting and visualization capabilities
for stock analysis with interactive dashboards using Plotly.
"""
import pandas as pd
import numpy as np
import plotly.graph_objects as go
import plotly.express as px
from plotly.subplots import make_subplots
import streamlit as st
from typing import Dict, List, Optional, Tuple
import warnings
warnings.filterwarnings('ignore')
class StockChartGenerator:
"""Enhanced stock chart generator with interactive dashboards"""
def __init__(self):
self.color_scheme = {
'primary': '#1f77b4',
'secondary': '#ff7f0e',
'success': '#2ca02c',
'danger': '#d62728',
'warning': '#ff7f0e',
'info': '#17a2b8',
'background': '#f8f9fa'
}
def create_price_chart(self, data: pd.DataFrame, symbol: str, analysis: Dict = None) -> go.Figure:
"""
Create comprehensive price chart with technical indicators
Args:
data: Stock price data
symbol: Stock symbol
analysis: Technical analysis results
Returns:
Plotly figure object
"""
if data.empty:
return self._create_empty_chart("No data available")
# Create subplots
fig = make_subplots(
rows=3, cols=1,
shared_xaxes=True,
vertical_spacing=0.05,
subplot_titles=(f'{symbol} Price Chart', 'Volume', 'Technical Indicators'),
row_heights=[0.6, 0.2, 0.2]
)
# Main price chart (candlestick)
fig.add_trace(
go.Candlestick(
x=data.index,
open=data['Open'],
high=data['High'],
low=data['Low'],
close=data['Close'],
name='Price',
increasing_line_color=self.color_scheme['success'],
decreasing_line_color=self.color_scheme['danger']
),
row=1, col=1
)
# Add moving averages if available
if 'SMA_20' in data.columns:
fig.add_trace(
go.Scatter(
x=data.index,
y=data['SMA_20'],
mode='lines',
name='SMA 20',
line=dict(color=self.color_scheme['primary'], width=1)
),
row=1, col=1
)
if 'SMA_50' in data.columns:
fig.add_trace(
go.Scatter(
x=data.index,
y=data['SMA_50'],
mode='lines',
name='SMA 50',
line=dict(color=self.color_scheme['secondary'], width=1)
),
row=1, col=1
)
# Volume chart
colors = ['red' if close < open else 'green' for close, open in zip(data['Close'], data['Open'])]
fig.add_trace(
go.Bar(
x=data.index,
y=data['Volume'],
name='Volume',
marker_color=colors,
opacity=0.7
),
row=2, col=1
)
# Technical indicators (RSI if available in analysis)
if analysis and 'rsi' in analysis:
# Create RSI line (simplified - would need full RSI calculation for time series)
rsi_value = analysis['rsi']
rsi_line = [rsi_value] * len(data)
fig.add_trace(
go.Scatter(
x=data.index,
y=rsi_line,
mode='lines',
name=f'RSI ({rsi_value:.1f})',
line=dict(color=self.color_scheme['info'], width=2)
),
row=3, col=1
)
# Add RSI reference lines
fig.add_hline(y=70, line_dash="dash", line_color="red", opacity=0.5, row=3, col=1)
fig.add_hline(y=30, line_dash="dash", line_color="green", opacity=0.5, row=3, col=1)
# Update layout
fig.update_layout(
title=f'{symbol} Stock Analysis Dashboard',
xaxis_title='Date',
yaxis_title='Price ($)',
template='plotly_white',
height=800,
showlegend=True,
hovermode='x unified'
)
# Remove rangeslider for cleaner look
fig.update_layout(xaxis_rangeslider_visible=False)
return fig
def create_performance_chart(self, data: pd.DataFrame, symbol: str, analysis: Dict) -> go.Figure:
"""
Create performance analysis chart
Args:
data: Stock price data
symbol: Stock symbol
analysis: Analysis results with performance metrics
Returns:
Plotly figure object
"""
if data.empty:
return self._create_empty_chart("No data available for performance analysis")
# Calculate cumulative returns
daily_returns = data['Close'].pct_change().fillna(0)
cumulative_returns = (1 + daily_returns).cumprod() - 1
fig = go.Figure()
# Cumulative returns line
fig.add_trace(
go.Scatter(
x=data.index,
y=cumulative_returns * 100,
mode='lines',
name='Cumulative Returns (%)',
line=dict(color=self.color_scheme['primary'], width=2),
fill='tonexty',
fillcolor='rgba(31, 119, 180, 0.1)'
)
)
# Add benchmark line (0% return)
fig.add_hline(y=0, line_dash="dash", line_color="gray", opacity=0.5)
# Add performance annotations
if analysis:
total_return = analysis.get('total_return_pct', 0)
fig.add_annotation(
x=data.index[-1],
y=total_return,
text=f"Total Return: {total_return:.1f}%",
showarrow=True,
arrowhead=2,
arrowcolor=self.color_scheme['primary'],
bgcolor="white",
bordercolor=self.color_scheme['primary']
)
fig.update_layout(
title=f'{symbol} Performance Analysis',
xaxis_title='Date',
yaxis_title='Cumulative Returns (%)',
template='plotly_white',
height=500,
hovermode='x'
)
return fig
def create_risk_analysis_chart(self, analysis: Dict, symbol: str) -> go.Figure:
"""
Create risk analysis visualization
Args:
analysis: Analysis results with risk metrics
symbol: Stock symbol
Returns:
Plotly figure object
"""
if not analysis or 'error' in analysis:
return self._create_empty_chart("No risk data available")
# Prepare risk metrics
risk_metrics = {
'Volatility (Annual)': analysis.get('volatility_annualized', 0),
'Max Drawdown': abs(analysis.get('max_drawdown', 0)),
'VaR 95%': abs(analysis.get('var_95', 0)),
'VaR 99%': abs(analysis.get('var_99', 0))
}
# Create radar chart for risk metrics
categories = list(risk_metrics.keys())
values = list(risk_metrics.values())
fig = go.Figure()
fig.add_trace(go.Scatterpolar(
r=values,
theta=categories,
fill='toself',
name=f'{symbol} Risk Profile',
line_color=self.color_scheme['danger'],
fillcolor='rgba(214, 39, 40, 0.1)'
))
fig.update_layout(
polar=dict(
radialaxis=dict(
visible=True,
range=[0, max(values) * 1.2] if values else [0, 100]
)
),
title=f'{symbol} Risk Analysis Chart',
template='plotly_white',
height=500
)
return fig
def create_comparison_chart(self, data_dict: Dict[str, pd.DataFrame], symbols: List[str]) -> go.Figure:
"""
Create comparison chart for multiple stocks
Args:
data_dict: Dictionary of stock data {symbol: dataframe}
symbols: List of stock symbols to compare
Returns:
Plotly figure object
"""
fig = go.Figure()
colors = [self.color_scheme['primary'], self.color_scheme['secondary'],
self.color_scheme['success'], self.color_scheme['danger']]
for i, symbol in enumerate(symbols):
if symbol in data_dict and not data_dict[symbol].empty:
data = data_dict[symbol]
# Normalize prices to start at 100 for comparison
normalized_prices = (data['Close'] / data['Close'].iloc[0]) * 100
fig.add_trace(
go.Scatter(
x=data.index,
y=normalized_prices,
mode='lines',
name=symbol,
line=dict(color=colors[i % len(colors)], width=2)
)
)
fig.update_layout(
title='Stock Price Comparison (Normalized to 100)',
xaxis_title='Date',
yaxis_title='Normalized Price',
template='plotly_white',
height=600,
hovermode='x unified'
)
return fig
def create_sector_analysis_chart(self, sector_data: Dict) -> go.Figure:
"""
Create sector analysis visualization
Args:
sector_data: Dictionary with sector analysis data
Returns:
Plotly figure object
"""
# This would typically show sector performance, P/E ratios, etc.
# For now, create a placeholder
fig = go.Figure()
fig.add_annotation(
x=0.5, y=0.5,
text="Sector Analysis<br>Coming Soon",
showarrow=False,
font=dict(size=20),
xref="paper", yref="paper"
)
fig.update_layout(
title='Sector Analysis Dashboard',
template='plotly_white',
height=400,
showticklabels=False
)
return fig
def create_trading_signals_chart(self, data: pd.DataFrame, analysis: Dict, trading_decision: Dict, symbol: str) -> go.Figure:
"""
Create trading signals visualization
Args:
data: Stock price data
analysis: Technical analysis results
trading_decision: Trading recommendation
symbol: Stock symbol
Returns:
Plotly figure object
"""
if data.empty:
return self._create_empty_chart("No data available for trading signals")
fig = go.Figure()
# Price line
fig.add_trace(
go.Scatter(
x=data.index,
y=data['Close'],
mode='lines',
name='Price',
line=dict(color=self.color_scheme['primary'], width=2)
)
)
# Add trading signal
recommendation = trading_decision.get('recommendation', 'HOLD')
current_price = data['Close'].iloc[-1]
signal_color = {
'BUY': self.color_scheme['success'],
'SELL': self.color_scheme['danger'],
'HOLD': self.color_scheme['warning']
}.get(recommendation, self.color_scheme['info'])
fig.add_trace(
go.Scatter(
x=[data.index[-1]],
y=[current_price],
mode='markers',
name=f'{recommendation} Signal',
marker=dict(
color=signal_color,
size=15,
symbol='triangle-up' if recommendation == 'BUY' else
'triangle-down' if recommendation == 'SELL' else 'circle'
)
)
)
# Add price target if available
price_target = trading_decision.get('price_target')
if price_target:
fig.add_hline(
y=price_target,
line_dash="dash",
line_color=self.color_scheme['success'],
annotation_text=f"Target: ${price_target:.2f}"
)
# Add stop loss if available
stop_loss = trading_decision.get('stop_loss')
if stop_loss:
fig.add_hline(
y=stop_loss,
line_dash="dash",
line_color=self.color_scheme['danger'],
annotation_text=f"Stop Loss: ${stop_loss:.2f}"
)
fig.update_layout(
title=f'{symbol} Trading Signals',
xaxis_title='Date',
yaxis_title='Price ($)',
template='plotly_white',
height=500,
hovermode='x'
)
return fig
def create_dashboard_summary(self, symbol: str, analysis: Dict, trading_decision: Dict, sharia_compliance: Dict) -> Dict:
"""
Create summary metrics for dashboard display
Args:
symbol: Stock symbol
analysis: Technical analysis results
trading_decision: Trading recommendation
sharia_compliance: Sharia compliance results
Returns:
Dictionary with summary metrics
"""
summary = {
'symbol': symbol,
'current_price': analysis.get('current_price', 0),
'total_return': analysis.get('total_return_pct', 0),
'volatility': analysis.get('volatility_annualized', 0),
'trading_recommendation': trading_decision.get('recommendation', 'HOLD'),
'trading_confidence': trading_decision.get('confidence', 0) * 100,
'sharia_ruling': sharia_compliance.get('ruling', 'UNCERTAIN'),
'sharia_confidence': sharia_compliance.get('confidence', 0) * 100,
'risk_level': trading_decision.get('risk_level', 'medium'),
'trend_direction': analysis.get('trend_direction', 'unknown'),
'rsi': analysis.get('rsi', 50),
'max_drawdown': analysis.get('max_drawdown', 0)
}
return summary
def _create_empty_chart(self, message: str) -> go.Figure:
"""Create an empty chart with a message"""
fig = go.Figure()
fig.add_annotation(
x=0.5, y=0.5,
text=message,
showarrow=False,
font=dict(size=16),
xref="paper", yref="paper"
)
fig.update_layout(
template='plotly_white',
height=400,
showticklabels=False
)
return fig
# Global instance for easy import
chart_generator = StockChartGenerator()
# Convenience functions
def create_price_chart(data: pd.DataFrame, symbol: str, analysis: Dict = None) -> go.Figure:
"""Convenience function to create price chart"""
return chart_generator.create_price_chart(data, symbol, analysis)
def create_performance_chart(data: pd.DataFrame, symbol: str, analysis: Dict) -> go.Figure:
"""Convenience function to create performance chart"""
return chart_generator.create_performance_chart(data, symbol, analysis)
def create_trading_signals_chart(data: pd.DataFrame, analysis: Dict, trading_decision: Dict, symbol: str) -> go.Figure:
"""Convenience function to create trading signals chart"""
return chart_generator.create_trading_signals_chart(data, analysis, trading_decision, symbol)

384
week4/community-contributions/ai_stock_trading/tools/fetching.py Normal file
View File

@@ -0,0 +1,384 @@
"""
Stock Data Fetching Module
This module handles fetching stock data from various sources including yfinance
and provides enhanced data retrieval capabilities for different markets.
"""
import yfinance as yf
import pandas as pd
import numpy as np
import requests
from typing import Dict, List, Optional, Tuple
import warnings
warnings.filterwarnings('ignore')
class StockDataFetcher:
"""Enhanced stock data fetcher with multi-market support"""
# Stock symbols for different markets
STOCK_SYMBOLS = {
'USA': {
# Technology
'Apple Inc.': 'AAPL',
'Microsoft Corporation': 'MSFT',
'NVIDIA Corporation': 'NVDA',
'Alphabet Inc. (Class A)': 'GOOGL',
'Alphabet Inc. (Class C)': 'GOOG',
'Meta Platforms Inc.': 'META',
'Tesla Inc.': 'TSLA',
'Amazon.com Inc.': 'AMZN',
'Netflix Inc.': 'NFLX',
'Adobe Inc.': 'ADBE',
'Salesforce Inc.': 'CRM',
'Oracle Corporation': 'ORCL',
'Cisco Systems Inc.': 'CSCO',
'Intel Corporation': 'INTC',
'Advanced Micro Devices': 'AMD',
'Qualcomm Inc.': 'QCOM',
'Texas Instruments': 'TXN',
'Broadcom Inc.': 'AVGO',
'ServiceNow Inc.': 'NOW',
'Palantir Technologies': 'PLTR',
# Financial Services
'JPMorgan Chase & Co.': 'JPM',
'Bank of America Corp': 'BAC',
'Wells Fargo & Company': 'WFC',
'Goldman Sachs Group': 'GS',
'Morgan Stanley': 'MS',
'Citigroup Inc.': 'C',
'American Express Company': 'AXP',
'Berkshire Hathaway Inc.': 'BRK.B',
'BlackRock Inc.': 'BLK',
'Charles Schwab Corporation': 'SCHW',
'Visa Inc.': 'V',
'Mastercard Inc.': 'MA',
# Healthcare & Pharmaceuticals
'Johnson & Johnson': 'JNJ',
'UnitedHealth Group': 'UNH',
'Pfizer Inc.': 'PFE',
'AbbVie Inc.': 'ABBV',
'Merck & Co Inc.': 'MRK',
'Eli Lilly and Company': 'LLY',
'Abbott Laboratories': 'ABT',
'Thermo Fisher Scientific': 'TMO',
'Danaher Corporation': 'DHR',
'Gilead Sciences Inc.': 'GILD',
# Consumer & Retail
'Walmart Inc.': 'WMT',
'Procter & Gamble Co': 'PG',
'Coca-Cola Company': 'KO',
'PepsiCo Inc.': 'PEP',
'Home Depot Inc.': 'HD',
'McDonald\'s Corporation': 'MCD',
'Nike Inc.': 'NKE',
'Costco Wholesale Corp': 'COST',
'TJX Companies Inc.': 'TJX',
'Lowe\'s Companies Inc.': 'LOW',
# Industrial & Energy
'Exxon Mobil Corporation': 'XOM',
'Chevron Corporation': 'CVX',
'ConocoPhillips': 'COP',
'Caterpillar Inc.': 'CAT',
'Boeing Company': 'BA',
'General Electric': 'GE',
'Honeywell International': 'HON',
'Deere & Company': 'DE',
'Union Pacific Corporation': 'UNP',
'Lockheed Martin Corp': 'LMT',
# Communication & Media
'AT&T Inc.': 'T',
'Verizon Communications': 'VZ',
'T-Mobile US Inc.': 'TMUS',
'Comcast Corporation': 'CMCSA',
'Walt Disney Company': 'DIS'
},
'Egypt': {
# Banking & Financial Services
'Commercial International Bank': 'COMI.CA',
'QNB Alahli Bank': 'QNBE.CA',
'Housing and Development Bank': 'HDBK.CA',
'Abu Dhabi Islamic Bank Egypt': 'ADIB.CA',
'Egyptian Gulf Bank': 'EGBE.CA',
# Real Estate & Construction
'Talaat Moustafa Group Holding': 'TMGH.CA',
'Palm Hills Developments': 'PHDC.CA',
'Orascom Construction': 'ORAS.CA',
'Orascom Development Holding': 'ORHD.CA',
'Six of October Development': 'SCTS.CA',
'Heliopolis Housing': 'HELI.CA',
'Rooya Group': 'RMDA.CA',
# Industrial & Manufacturing
'Eastern Company': 'EAST.CA',
'El Sewedy Electric Company': 'SWDY.CA',
'Ezz Steel': 'ESRS.CA',
'Iron and Steel Company': 'IRON.CA',
'Alexandria Containers': 'ALCN.CA',
'Sidi Kerir Petrochemicals': 'SKPC.CA',
# Chemicals & Fertilizers
'Abu Qir Fertilizers and Chemical Industries': 'ABUK.CA',
'Egyptian Chemical Industries (Kima)': 'KIMA.CA',
'Misr Fertilizers Production': 'MFPC.CA',
# Telecommunications & Technology
'Telecom Egypt': 'ETEL.CA',
'Raya Holding': 'RAYA.CA',
'E-Finance for Digital Payments': 'EFIH.CA',
'Fawry for Banking Technology': 'FWRY.CA',
# Food & Beverages
'Juhayna Food Industries': 'JUFO.CA',
'Edita Food Industries': 'EFID.CA',
'Cairo Poultry Company': 'POUL.CA',
'Upper Egypt Flour Mills': 'UEFM.CA',
'Ismailia Misr Poultry': 'ISPH.CA',
# Healthcare & Pharmaceuticals
'Cleopatra Hospital Group': 'CLHO.CA',
'Cairo Pharmaceuticals': 'PHAR.CA',
# Energy & Utilities
'Egyptian Natural Gas Company': 'EGAS.CA',
'Suez Cement Company': 'SCEM.CA',
'Arabian Cement Company': 'ARCC.CA',
# Investment & Holding Companies
'Egyptian Financial Group-Hermes': 'HRHO.CA',
'Citadel Capital': 'CCAP.CA',
'Beltone Financial Holding': 'BTFH.CA'
}
}
# Currency mapping for different markets
MARKET_CURRENCIES = {
'USA': 'USD',
'Egypt': 'EGP'
}
def __init__(self):
self.cache = {}
def get_available_stocks(self, country: str) -> Dict[str, str]:
"""Get available stocks for a specific country"""
return self.STOCK_SYMBOLS.get(country, {})
def get_market_currency(self, country: str) -> str:
"""Get the currency for a specific market"""
return self.MARKET_CURRENCIES.get(country, 'USD')
def format_price_with_currency(self, price: float, country: str) -> str:
"""Format price with appropriate currency symbol"""
currency = self.get_market_currency(country)
if currency == 'EGP':
return f"{price:.2f} EGP"
elif currency == 'USD':
return f"${price:.2f}"
else:
return f"{price:.2f} {currency}"
def fetch_stock_data(self, symbol: str, period: str = "1y", interval: str = "1d") -> pd.DataFrame:
"""
Fetch historical stock data with enhanced error handling
Args:
symbol: Stock symbol (e.g., 'AAPL', 'COMI.CA')
period: Time period ('1d', '5d', '1mo', '3mo', '6mo', '1y', '2y', '5y', '10y', 'ytd', 'max')
interval: Data interval ('1m', '2m', '5m', '15m', '30m', '60m', '90m', '1h', '1d', '5d', '1wk', '1mo', '3mo')
Returns:
DataFrame with OHLCV data
"""
cache_key = f"{symbol}_{period}_{interval}"
# Check cache first
if cache_key in self.cache:
return self.cache[cache_key]
try:
# Create ticker object
ticker = yf.Ticker(symbol)
# Fetch historical data
data = ticker.history(period=period, interval=interval)
if data.empty:
print(f"⚠️ No data found for {symbol}")
return pd.DataFrame()
# Clean and enhance data
data = self._clean_data(data)
# Cache the result
self.cache[cache_key] = data
print(f"✅ Successfully fetched {len(data)} data points for {symbol} ({period})")
return data
except Exception as e:
print(f"❌ Error fetching data for {symbol}: {str(e)}")
return pd.DataFrame()
def get_stock_info(self, symbol: str, country: Optional[str] = None) -> Dict:
"""
Get comprehensive stock information
Args:
symbol: Stock symbol
country: Market country (USA, Egypt) for currency handling
Returns:
Dictionary with stock information
"""
try:
ticker = yf.Ticker(symbol)
info = ticker.info
# Detect country if not provided
if country is None:
country = self._detect_country_from_symbol(symbol)
# Get market currency
market_currency = self.get_market_currency(country)
# Extract key information
stock_info = {
'symbol': symbol,
'company_name': info.get('longName', 'N/A'),
'sector': info.get('sector', 'N/A'),
'industry': info.get('industry', 'N/A'),
'market_cap': info.get('marketCap', 0),
'pe_ratio': info.get('trailingPE', 0),
'dividend_yield': info.get('dividendYield', 0),
'beta': info.get('beta', 0),
'fifty_two_week_high': info.get('fiftyTwoWeekHigh', 0),
'fifty_two_week_low': info.get('fiftyTwoWeekLow', 0),
'current_price': info.get('currentPrice', 0),
'currency': market_currency, # Use detected market currency
'exchange': info.get('exchange', 'N/A'),
'country': country,
'market_country': country # Add explicit market country
}
return stock_info
except Exception as e:
print(f"❌ Error fetching info for {symbol}: {str(e)}")
return {'symbol': symbol, 'error': str(e)}
def _detect_country_from_symbol(self, symbol: str) -> str:
"""
Detect country from stock symbol
Args:
symbol: Stock symbol
Returns:
Country name (USA or Egypt)
"""
# Check if symbol exists in any country's stock list
for country, stocks in self.STOCK_SYMBOLS.items():
if symbol in stocks.values():
return country
# Default to USA if not found
return 'USA'
def fetch_multiple_periods(self, symbol: str) -> Dict[str, pd.DataFrame]:
"""
Fetch data for multiple time periods
Args:
symbol: Stock symbol
Returns:
Dictionary with DataFrames for different periods
"""
periods = ['1mo', '1y', '5y']
data = {}
for period in periods:
df = self.fetch_stock_data(symbol, period)
if not df.empty:
data[period] = df
return data
def _clean_data(self, data: pd.DataFrame) -> pd.DataFrame:
"""
Clean and enhance the stock data
Args:
data: Raw stock data DataFrame
Returns:
Cleaned DataFrame
"""
# Remove rows with all NaN values
data = data.dropna(how='all')
# Forward fill missing values
data = data.fillna(method='ffill')
# Add technical indicators
if len(data) > 0:
# Simple moving averages
if len(data) >= 20:
data['SMA_20'] = data['Close'].rolling(window=20).mean()
if len(data) >= 50:
data['SMA_50'] = data['Close'].rolling(window=50).mean()
# Daily returns
data['Daily_Return'] = data['Close'].pct_change()
# Price change from previous day
data['Price_Change'] = data['Close'].diff()
data['Price_Change_Pct'] = (data['Price_Change'] / data['Close'].shift(1)) * 100
return data
def get_real_time_price(self, symbol: str) -> Optional[float]:
"""
Get real-time stock price
Args:
symbol: Stock symbol
Returns:
Current stock price or None if error
"""
try:
ticker = yf.Ticker(symbol)
data = ticker.history(period="1d", interval="1m")
if not data.empty:
return float(data['Close'].iloc[-1])
return None
except Exception as e:
print(f"❌ Error fetching real-time price for {symbol}: {str(e)}")
return None
# Global instance for easy import
stock_fetcher = StockDataFetcher()
# Convenience functions
def fetch_stock_data(symbol: str, period: str = "1y", interval: str = "1d") -> pd.DataFrame:
"""Convenience function to fetch stock data"""
return stock_fetcher.fetch_stock_data(symbol, period, interval)
def get_available_stocks(country: str) -> Dict[str, str]:
"""Convenience function to get available stocks"""
return stock_fetcher.get_available_stocks(country)
def get_stock_info(symbol: str) -> Dict:
"""Convenience function to get stock info"""
return stock_fetcher.get_stock_info(symbol)

591
week4/community-contributions/ai_stock_trading/tools/sharia_compliance.py Normal file
View File

@@ -0,0 +1,591 @@
"""
Sharia Compliance Module
This module provides comprehensive Islamic finance compliance checking
for stocks and investments according to Islamic principles.
"""
import os
import json
import requests
from typing import Dict, List, Optional, Tuple
import pandas as pd
from openai import OpenAI
from dotenv import load_dotenv
from bs4 import BeautifulSoup
import time
import re
# Load environment variables
load_dotenv()
class ShariaComplianceChecker:
"""Enhanced Sharia compliance checker for Islamic investing"""
def __init__(self):
self.client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
# Sharia compliance criteria weights
self.criteria_weights = {
'business_activity': 0.40, # Most important
'financial_ratios': 0.30,
'debt_levels': 0.20,
'revenue_sources': 0.10
}
# Prohibited business activities (comprehensive list)
self.prohibited_activities = {
# Core prohibitions
'alcohol', 'alcoholic_beverages', 'wine', 'beer', 'spirits', 'liquor',
'gambling', 'casino', 'lottery', 'betting', 'gaming', 'poker',
'tobacco', 'cigarettes', 'smoking', 'nicotine',
'pork', 'pig_farming', 'swine', 'ham', 'bacon',
'adult_entertainment', 'pornography', 'strip_clubs', 'escort_services',
# Financial prohibitions
'conventional_banking', 'interest_based_finance', 'usury', 'riba',
'conventional_insurance', 'life_insurance', 'derivatives_trading',
'forex_trading', 'currency_speculation', 'margin_trading',
'short_selling', 'day_trading', 'high_frequency_trading',
# Weapons and defense
'weapons', 'arms_manufacturing', 'defense_contractors', 'military_equipment',
'ammunition', 'explosives', 'nuclear_weapons',
# Other prohibitions
'nightclubs', 'bars', 'entertainment_venues', 'music_industry',
'film_industry', 'media_entertainment', 'advertising_haram_products'
}
# Sharia-compliant sectors (generally accepted)
self.compliant_sectors = {
'technology', 'healthcare', 'pharmaceuticals', 'telecommunications',
'utilities', 'real_estate', 'construction', 'manufacturing',
'retail', 'food_beverages', 'transportation', 'energy_renewable'
}
# Questionable sectors (need detailed analysis)
self.questionable_sectors = {
'financial_services', 'media', 'hotels', 'airlines',
'oil_gas', 'mining', 'chemicals', 'entertainment',
'restaurants', 'hospitality', 'advertising'
}
# AAOIFI and DSN Sharia standards
self.sharia_standards = {
'max_debt_to_assets': 0.33, # 33% maximum debt-to-assets ratio
'max_interest_income': 0.05, # 5% maximum interest income
'max_non_compliant_income': 0.05, # 5% maximum non-compliant income
'min_tangible_assets': 0.20 # 20% minimum tangible assets
}
def _search_company_business_activities(self, company_name: str, symbol: str) -> Dict:
"""
Search web for company's business activities to verify Sharia compliance
Args:
company_name: Company name
symbol: Stock symbol
Returns:
Dictionary with business activity information
"""
try:
# Search query for company business activities
search_queries = [
f"{company_name} business activities products services",
f"{company_name} {symbol} what does company do",
f"{company_name} revenue sources business model"
]
business_info = {
'activities': [],
'products': [],
'services': [],
'revenue_sources': [],
'prohibited_found': [],
'confidence': 0.5
}
for query in search_queries[:1]: # Limit to 1 search to avoid rate limits
try:
# Simple web search simulation (in production, use proper search API)
search_url = f"https://www.google.com/search?q={query.replace(' ', '+')}"
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
}
# For now, return basic analysis based on company name and sector
# In production, implement actual web scraping with proper rate limiting
business_info['confidence'] = 0.6
break
except Exception as e:
print(f"Web search error: {e}")
continue
return business_info
except Exception as e:
print(f"Error in business activity search: {e}")
return {'activities': [], 'prohibited_found': [], 'confidence': 0.3}
def _estimate_debt_ratio(self, stock_info: Dict) -> float:
"""
Estimate debt-to-assets ratio based on available information
Args:
stock_info: Stock information dictionary
Returns:
Estimated debt-to-assets ratio
"""
try:
# In production, this would fetch actual balance sheet data
# For now, estimate based on sector and other indicators
sector = stock_info.get('sector', '').lower()
industry = stock_info.get('industry', '').lower()
# High debt sectors
if any(x in sector or x in industry for x in ['utility', 'telecom', 'airline', 'real estate']):
return 0.45 # Typically higher debt
# Medium debt sectors
elif any(x in sector or x in industry for x in ['manufacturing', 'retail', 'energy']):
return 0.25
# Low debt sectors
elif any(x in sector or x in industry for x in ['technology', 'healthcare', 'software']):
return 0.15
# Financial sector (different calculation)
elif 'financial' in sector or 'bank' in sector:
return 0.8 # Banks have high leverage by nature
# Default estimate
return 0.3
except Exception:
return 0.3 # Conservative default
def _check_business_activity(self, stock_info: Dict) -> float:
"""
Check if the company's primary business activity is Sharia-compliant
Returns:
Score from 0.0 (non-compliant) to 1.0 (fully compliant)
"""
sector = stock_info.get('sector', '').lower()
industry = stock_info.get('industry', '').lower()
company_name = stock_info.get('company_name', '').lower()
# Check for explicitly prohibited activities
for prohibited in self.prohibited_activities:
if (prohibited.replace('_', ' ') in sector or
prohibited.replace('_', ' ') in industry or
prohibited.replace('_', ' ') in company_name):
return 0.0
# Check for compliant sectors
for compliant in self.compliant_sectors:
if (compliant.replace('_', ' ') in sector or
compliant.replace('_', ' ') in industry):
return 1.0
# Check for questionable sectors
for questionable in self.questionable_sectors:
if (questionable.replace('_', ' ') in sector or
questionable.replace('_', ' ') in industry):
return 0.5
# Default for unknown sectors
return 0.7
def check_sharia_compliance(self, symbol: str, stock_info: Dict, analysis: Dict) -> Dict:
"""
Comprehensive Sharia compliance check
Args:
symbol: Stock symbol
stock_info: Stock information
analysis: Technical analysis results
Returns:
Dictionary with compliance assessment
"""
try:
# Business activity screening
business_score = self._check_business_activity(stock_info)
# Financial ratios screening
financial_score = self._check_financial_ratios(stock_info, analysis)
# Debt levels screening
debt_score = self._check_debt_levels(stock_info)
# Revenue sources screening
revenue_score = self._check_revenue_sources(stock_info)
# Calculate weighted compliance score
total_score = (
business_score * self.criteria_weights['business_activity'] +
financial_score * self.criteria_weights['financial_ratios'] +
debt_score * self.criteria_weights['debt_levels'] +
revenue_score * self.criteria_weights['revenue_sources']
)
# Get AI-powered detailed analysis
ai_analysis = self._get_ai_sharia_analysis(symbol, stock_info)
# Determine final ruling
ruling = self._determine_ruling(total_score, ai_analysis)
return {
'symbol': symbol,
'ruling': ruling['status'],
'confidence': ruling['confidence'],
'compliance_score': total_score,
'detailed_scores': {
'business_activity': business_score,
'financial_ratios': financial_score,
'debt_levels': debt_score,
'revenue_sources': revenue_score
},
'reasoning': ruling['reasoning'],
'key_concerns': ruling.get('concerns', []),
'recommendations': ruling.get('recommendations', []),
'ai_analysis': ai_analysis.get('analysis', ''),
'scholar_consultation_advised': ruling.get('scholar_consultation', False),
'alternative_suggestions': ruling.get('alternatives', [])
}
except Exception as e:
return {
'symbol': symbol,
'ruling': 'UNCERTAIN',
'confidence': 0.0,
'reasoning': f'Error in Sharia compliance analysis: {str(e)}',
'error': str(e)
}
def _check_business_activity(self, stock_info: Dict) -> float:
"""
Check if the company's primary business activity is Sharia-compliant
Returns:
Score from 0.0 (non-compliant) to 1.0 (fully compliant)
"""
sector = stock_info.get('sector', '').lower()
industry = stock_info.get('industry', '').lower()
company_name = stock_info.get('company_name', '').lower()
# Check for explicitly prohibited activities
for prohibited in self.prohibited_activities:
if (prohibited.replace('_', ' ') in sector or
prohibited.replace('_', ' ') in industry or
prohibited.replace('_', ' ') in company_name):
return 0.0
# Check for compliant sectors
for compliant in self.compliant_sectors:
if (compliant.replace('_', ' ') in sector or
compliant.replace('_', ' ') in industry):
return 1.0
# Check for questionable sectors
for questionable in self.questionable_sectors:
if (questionable.replace('_', ' ') in sector or
questionable.replace('_', ' ') in industry):
return 0.5
# Default for unknown sectors
return 0.7
def _check_financial_ratios(self, stock_info: Dict, analysis: Dict) -> float:
"""
Check financial ratios according to AAOIFI and DSN Sharia standards
AAOIFI/DSN Sharia screening ratios:
- Debt/Total Assets < 33%
- Interest Income/Total Revenue < 5%
- Non-compliant Income/Total Revenue < 5%
- Tangible Assets/Total Assets > 20%
Returns:
Score from 0.0 to 1.0
"""
score = 1.0
penalties = []
try:
# Get financial metrics (these would come from detailed financial data)
market_cap = stock_info.get('market_cap', 0)
pe_ratio = stock_info.get('pe_ratio', 0)
# Debt-to-Assets ratio check
# Note: In production, fetch actual balance sheet data
debt_to_assets = self._estimate_debt_ratio(stock_info)
if debt_to_assets > self.sharia_standards['max_debt_to_assets']:
penalty = min(0.5, (debt_to_assets - self.sharia_standards['max_debt_to_assets']) * 2)
score -= penalty
penalties.append(f"High debt ratio: {debt_to_assets:.1%} > {self.sharia_standards['max_debt_to_assets']:.1%}")
# Interest income check (for financial companies)
sector = stock_info.get('sector', '').lower()
if 'financial' in sector or 'bank' in sector:
# Financial companies likely have significant interest income
score -= 0.3
penalties.append("Financial sector - likely high interest income")
# Industry-specific checks
industry = stock_info.get('industry', '').lower()
if any(prohibited in industry for prohibited in ['insurance', 'casino', 'alcohol', 'tobacco']):
score = 0.0
penalties.append(f"Prohibited industry: {industry}")
return max(0.0, score)
except Exception as e:
print(f"Error in financial ratio analysis: {e}")
return 0.5 # Default moderate score if analysis fails
# For now, we'll use available basic metrics and estimate
# PE ratio check (very high PE might indicate speculation)
pe_ratio = stock_info.get('pe_ratio', 0)
if pe_ratio > 50:
score -= 0.2
elif pe_ratio > 30:
score -= 0.1
# Beta check (high beta indicates high speculation/volatility)
beta = stock_info.get('beta', 1.0)
if beta > 2.0:
score -= 0.3
elif beta > 1.5:
score -= 0.1
# Volatility check from analysis
volatility = analysis.get('volatility_annualized', 0)
if volatility > 60:
score -= 0.2
elif volatility > 40:
score -= 0.1
return max(0.0, score)
def _check_debt_levels(self, stock_info: Dict) -> float:
"""
Check debt levels according to Sharia standards
Returns:
Score from 0.0 to 1.0
"""
# Note: In a real implementation, you would fetch debt-to-assets ratio
# For now, we'll use sector-based estimation
sector = stock_info.get('sector', '').lower()
# Sectors typically with high debt
high_debt_sectors = ['utilities', 'real estate', 'telecommunications']
medium_debt_sectors = ['manufacturing', 'transportation', 'energy']
low_debt_sectors = ['technology', 'healthcare', 'retail']
if any(s in sector for s in high_debt_sectors):
return 0.6 # Assume higher debt but may still be acceptable
elif any(s in sector for s in medium_debt_sectors):
return 0.8
elif any(s in sector for s in low_debt_sectors):
return 1.0
else:
return 0.7 # Default assumption
def _check_revenue_sources(self, stock_info: Dict) -> float:
"""
Check revenue sources for non-compliant income
Returns:
Score from 0.0 to 1.0
"""
sector = stock_info.get('sector', '').lower()
industry = stock_info.get('industry', '').lower()
# Industries with potential non-compliant revenue
if 'financial' in sector or 'bank' in industry:
return 0.3 # Banks typically have significant interest income
elif 'insurance' in industry:
return 0.2
elif 'hotel' in industry or 'entertainment' in industry:
return 0.6 # May have some non-compliant revenue sources
else:
return 0.9 # Assume mostly compliant revenue
def _get_ai_sharia_analysis(self, symbol: str, stock_info: Dict) -> Dict:
"""
Get AI-powered detailed Sharia compliance analysis
Args:
symbol: Stock symbol
stock_info: Stock information
Returns:
Dictionary with AI analysis
"""
try:
prompt = f"""
As an Islamic finance expert, analyze the Sharia compliance of {symbol}.
Company Information:
- Name: {stock_info.get('company_name', 'N/A')}
- Sector: {stock_info.get('sector', 'N/A')}
- Industry: {stock_info.get('industry', 'N/A')}
- Country: {stock_info.get('country', 'N/A')}
Please analyze according to Islamic finance principles and provide:
1. Primary business activity assessment
2. Potential Sharia compliance concerns
3. Revenue source analysis
4. Debt and interest exposure concerns
5. Overall compliance recommendation
6. Specific areas requiring scholar consultation
7. Alternative Sharia-compliant investment suggestions
Format your response as JSON:
{{
"compliance_status": "HALAL/HARAM/DOUBTFUL",
"confidence": 85,
"analysis": "Detailed analysis...",
"concerns": ["concern1", "concern2"],
"recommendations": ["rec1", "rec2"],
"scholar_consultation": true/false,
"alternatives": ["alt1", "alt2"]
}}
"""
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "You are an expert in Islamic finance and Sharia compliance for investments."},
{"role": "user", "content": prompt}
],
temperature=0.2,
max_tokens=1000
)
ai_response = response.choices[0].message.content
try:
if ai_response:
return json.loads(ai_response)
else:
return {'analysis': 'No AI response received', 'error': 'Empty response'}
except json.JSONDecodeError:
return {'analysis': ai_response, 'parsed_fallback': True}
except Exception as e:
return {'analysis': f'AI analysis unavailable: {str(e)}', 'error': str(e)}
def _determine_ruling(self, compliance_score: float, ai_analysis: Dict) -> Dict:
"""
Determine final Sharia compliance ruling
Args:
compliance_score: Calculated compliance score
ai_analysis: AI analysis results
Returns:
Dictionary with final ruling
"""
# Get AI recommendation if available
ai_status = ai_analysis.get('compliance_status', 'DOUBTFUL')
ai_confidence = ai_analysis.get('confidence', 50) / 100
# Combine algorithmic score with AI analysis
if compliance_score >= 0.8 and ai_status == 'HALAL':
status = 'HALAL'
confidence = min(0.9, (compliance_score + ai_confidence) / 2)
reasoning = "Company appears to be Sharia-compliant based on business activities and financial structure."
elif compliance_score <= 0.3 or ai_status == 'HARAM':
status = 'HARAM'
confidence = max(0.7, (1 - compliance_score + ai_confidence) / 2)
reasoning = "Company has significant Sharia compliance issues and should be avoided."
else:
status = 'DOUBTFUL'
confidence = 0.6
reasoning = "Company has mixed compliance indicators. Consultation with Islamic scholars recommended."
return {
'status': status,
'confidence': confidence,
'reasoning': reasoning,
'concerns': ai_analysis.get('concerns', []),
'recommendations': ai_analysis.get('recommendations', []),
'scholar_consultation': ai_analysis.get('scholar_consultation', status == 'DOUBTFUL'),
'alternatives': ai_analysis.get('alternatives', [])
}
def get_compliance_summary(self, compliance_result: Dict) -> str:
"""Generate a human-readable compliance summary"""
if 'error' in compliance_result:
return f"Compliance Analysis Error: {compliance_result['error']}"
symbol = compliance_result.get('symbol', 'Unknown')
ruling = compliance_result.get('ruling', 'UNCERTAIN')
confidence = compliance_result.get('confidence', 0) * 100
summary = [f"Sharia Compliance Analysis for {symbol}"]
summary.append(f"Ruling: {ruling} (Confidence: {confidence:.0f}%)")
if ruling == 'HALAL':
summary.append("✅ This investment appears to be permissible under Islamic law.")
elif ruling == 'HARAM':
summary.append("❌ This investment should be avoided due to Sharia non-compliance.")
else:
summary.append("⚠️ This investment requires further investigation and scholar consultation.")
# Add key concerns if any
concerns = compliance_result.get('key_concerns', [])
if concerns:
summary.append(f"Key Concerns: {', '.join(concerns)}")
# Add recommendations
recommendations = compliance_result.get('recommendations', [])
if recommendations:
summary.append(f"Recommendations: {', '.join(recommendations[:2])}")
return "\n".join(summary)
def get_sharia_alternatives(self, sector: str, country: str = 'USA') -> List[str]:
"""
Get Sharia-compliant alternatives in the same sector
Args:
sector: Company sector
country: Market country
Returns:
List of alternative stock symbols
"""
# This would typically connect to a Sharia-compliant stock database
# For now, return some common Sharia-compliant stocks by sector
alternatives = {
'technology': ['AAPL', 'MSFT', 'GOOGL', 'META'],
'healthcare': ['JNJ', 'PFE', 'UNH', 'ABBV'],
'consumer': ['PG', 'KO', 'PEP', 'WMT'],
'industrial': ['BA', 'CAT', 'GE', 'MMM']
}
sector_lower = sector.lower()
for key, stocks in alternatives.items():
if key in sector_lower:
return stocks[:3] # Return top 3 alternatives
return []
# Global instance for easy import
sharia_checker = ShariaComplianceChecker()
# Convenience function
def check_sharia_compliance(symbol: str, stock_info: Dict, analysis: Dict) -> Dict:
"""Convenience function to check Sharia compliance"""
return sharia_checker.check_sharia_compliance(symbol, stock_info, analysis)

491
week4/community-contributions/ai_stock_trading/tools/trading_decisions.py Normal file
View File

@@ -0,0 +1,491 @@
"""
Trading Decisions Module
This module provides AI-powered trading recommendations using OpenAI
and advanced algorithmic decision-making based on technical analysis.
"""
import os
import json
from typing import Dict, List, Optional, Tuple
import pandas as pd
import numpy as np
from openai import OpenAI
from dotenv import load_dotenv
# Load environment variables
load_dotenv()
class TradingDecisionEngine:
"""Enhanced trading decision engine with AI and algorithmic analysis"""
def __init__(self):
self.client = OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
# Trading signal weights
self.signal_weights = {
'trend': 0.25,
'momentum': 0.20,
'volume': 0.15,
'volatility': 0.15,
'technical': 0.25
}
def get_trading_recommendation(self, symbol: str, analysis: Dict, stock_info: Dict) -> Dict:
"""
Get comprehensive trading recommendation
Args:
symbol: Stock symbol
analysis: Technical analysis results
stock_info: Stock information
Returns:
Dictionary with trading recommendation
"""
try:
# Get algorithmic score
algo_decision = self._get_algorithmic_decision(analysis)
# Get AI-powered recommendation
ai_decision = self._get_ai_recommendation(symbol, analysis, stock_info)
# Combine decisions
final_decision = self._combine_decisions(algo_decision, ai_decision)
return {
'symbol': symbol,
'recommendation': final_decision['action'],
'confidence': final_decision['confidence'],
'price_target': final_decision.get('price_target'),
'stop_loss': final_decision.get('stop_loss'),
'reasoning': final_decision['reasoning'],
'algorithmic_score': algo_decision['score'],
'ai_recommendation': ai_decision['recommendation'],
'risk_level': self._assess_risk_level(analysis),
'time_horizon': final_decision.get('time_horizon', 'medium'),
'key_factors': final_decision.get('key_factors', [])
}
except Exception as e:
return {
'symbol': symbol,
'recommendation': 'HOLD',
'confidence': 0.5,
'reasoning': f'Error in analysis: {str(e)}',
'error': str(e)
}
def _get_algorithmic_decision(self, analysis: Dict) -> Dict:
"""
Generate algorithmic trading decision based on technical indicators
Args:
analysis: Technical analysis results
Returns:
Dictionary with algorithmic decision
"""
signals = {}
# Trend signals
trend_score = self._calculate_trend_signal(analysis)
signals['trend'] = trend_score
# Momentum signals
momentum_score = self._calculate_momentum_signal(analysis)
signals['momentum'] = momentum_score
# Volume signals
volume_score = self._calculate_volume_signal(analysis)
signals['volume'] = volume_score
# Volatility signals
volatility_score = self._calculate_volatility_signal(analysis)
signals['volatility'] = volatility_score
# Technical indicator signals
technical_score = self._calculate_technical_signal(analysis)
signals['technical'] = technical_score
# Calculate weighted score
total_score = sum(signals[key] * self.signal_weights[key] for key in signals)
# Determine action
if total_score >= 0.6:
action = 'BUY'
elif total_score <= -0.6:
action = 'SELL'
else:
action = 'HOLD'
return {
'action': action,
'score': total_score,
'signals': signals,
'confidence': min(abs(total_score), 1.0)
}
def _calculate_trend_signal(self, analysis: Dict) -> float:
"""Calculate trend-based signal (-1 to 1)"""
score = 0.0
# Trend direction and strength
if analysis.get('trend_direction') == 'upward':
score += 0.5
elif analysis.get('trend_direction') == 'downward':
score -= 0.5
# Trend strength
trend_strength = analysis.get('trend_strength', 0)
score *= trend_strength
# Moving average signals
if 'price_vs_sma_20' in analysis:
sma_20_signal = analysis['price_vs_sma_20']
if sma_20_signal > 2:
score += 0.2
elif sma_20_signal < -2:
score -= 0.2
if 'price_vs_sma_50' in analysis:
sma_50_signal = analysis['price_vs_sma_50']
if sma_50_signal > 2:
score += 0.3
elif sma_50_signal < -2:
score -= 0.3
return max(-1.0, min(1.0, score))
def _calculate_momentum_signal(self, analysis: Dict) -> float:
"""Calculate momentum-based signal (-1 to 1)"""
score = 0.0
# RSI signal
if 'rsi' in analysis:
rsi = analysis['rsi']
if rsi < 30:
score += 0.4 # Oversold - potential buy
elif rsi > 70:
score -= 0.4 # Overbought - potential sell
# MACD signal
if 'macd_trend' in analysis:
if analysis['macd_trend'] == 'bullish':
score += 0.3
else:
score -= 0.3
# Recent performance
if 'return_1_week' in analysis:
weekly_return = analysis['return_1_week']
if weekly_return > 5:
score += 0.2
elif weekly_return < -5:
score -= 0.2
return max(-1.0, min(1.0, score))
def _calculate_volume_signal(self, analysis: Dict) -> float:
"""Calculate volume-based signal (-1 to 1)"""
score = 0.0
# Volume trend
if analysis.get('volume_trend') == 'increasing':
score += 0.3
elif analysis.get('volume_trend') == 'decreasing':
score -= 0.2
# Volume vs average
if 'volume_vs_avg' in analysis:
vol_vs_avg = analysis['volume_vs_avg']
if vol_vs_avg > 20:
score += 0.2
elif vol_vs_avg < -20:
score -= 0.1
return max(-1.0, min(1.0, score))
def _calculate_volatility_signal(self, analysis: Dict) -> float:
"""Calculate volatility-based signal (-1 to 1)"""
score = 0.0
# High volatility can be both opportunity and risk
volatility = analysis.get('volatility_annualized', 0)
if volatility > 50:
score -= 0.3 # High risk
elif volatility < 15:
score += 0.2 # Low risk
# Max drawdown consideration
max_drawdown = analysis.get('max_drawdown', 0)
if abs(max_drawdown) > 20:
score -= 0.2
return max(-1.0, min(1.0, score))
def _calculate_technical_signal(self, analysis: Dict) -> float:
"""Calculate technical indicator signal (-1 to 1)"""
score = 0.0
# Bollinger Bands
if 'bb_position' in analysis:
bb_pos = analysis['bb_position']
if bb_pos == 'below_lower_band':
score += 0.3 # Potential buy
elif bb_pos == 'above_upper_band':
score -= 0.3 # Potential sell
# Sharpe ratio
sharpe = analysis.get('sharpe_ratio', 0)
if sharpe > 1:
score += 0.2
elif sharpe < 0:
score -= 0.2
return max(-1.0, min(1.0, score))
def _get_ai_recommendation(self, symbol: str, analysis: Dict, stock_info: Dict) -> Dict:
"""
Get AI-powered trading recommendation using OpenAI
Args:
symbol: Stock symbol
analysis: Technical analysis results
stock_info: Stock information
Returns:
Dictionary with AI recommendation
"""
try:
# Prepare analysis data for AI
analysis_summary = self._prepare_analysis_for_ai(analysis, stock_info)
prompt = f"""
You are a senior financial analyst with 15+ years of experience providing institutional-grade trading recommendations.
Analyze {symbol} and provide a professional trading recommendation.
Company Information:
- Name: {stock_info.get('company_name', 'N/A')}
- Sector: {stock_info.get('sector', 'N/A')}
- Market Cap: ${stock_info.get('market_cap', 0):,}
Technical Analysis Data:
{analysis_summary}
REQUIREMENTS:
1. Provide BUY/HOLD/SELL recommendation based on technical analysis
2. Set realistic confidence level (60-95% range)
3. Calculate logical price targets using support/resistance levels
4. Set appropriate stop-loss levels (5-15% below entry for long positions)
5. Consider risk-reward ratios (minimum 1:2 ratio preferred)
6. Provide clear, actionable reasoning without jargon
7. Consider market conditions and sector trends
TRADING STANDARDS:
- BUY: Strong upward momentum, good risk/reward, clear catalysts
- HOLD: Consolidation phase, mixed signals, or fair value
- SELL: Downward trend, poor fundamentals, or overvalued
Return ONLY valid JSON:
{{
"recommendation": "BUY/HOLD/SELL",
"confidence": 85,
"price_target": 150.00,
"stop_loss": 120.00,
"time_horizon": "short/medium/long",
"reasoning": "Professional analysis explaining the recommendation with specific technical factors",
"key_factors": ["specific technical indicator", "market condition", "risk factor"],
"risk_assessment": "low/medium/high"
}}
"""
response = self.client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "You are an expert financial analyst providing professional trading recommendations."},
{"role": "user", "content": prompt}
],
temperature=0.3,
max_tokens=1000
)
# Parse AI response
ai_response = response.choices[0].message.content
if ai_response:
try:
# Clean the response - extract JSON
json_start = ai_response.find('{')
json_end = ai_response.rfind('}') + 1
if json_start != -1 and json_end != -1:
json_str = ai_response[json_start:json_end]
ai_recommendation = json.loads(json_str)
# Validate and clean the response
return {
'recommendation': ai_recommendation.get('recommendation', 'HOLD'),
'confidence': ai_recommendation.get('confidence', 50),
'price_target': ai_recommendation.get('price_target'),
'stop_loss': ai_recommendation.get('stop_loss'),
'time_horizon': ai_recommendation.get('time_horizon', 'medium'),
'reasoning': ai_recommendation.get('reasoning', 'AI analysis completed'),
'key_factors': ai_recommendation.get('key_factors', []),
'risk_assessment': ai_recommendation.get('risk_assessment', 'medium')
}
else:
# No JSON found, use fallback
return self._parse_ai_response_fallback(ai_response)
except json.JSONDecodeError:
# Fallback parsing
return self._parse_ai_response_fallback(ai_response)
else:
return self._parse_ai_response_fallback('No response received')
except Exception as e:
return {
'recommendation': 'HOLD',
'confidence': 50,
'reasoning': f'AI analysis unavailable: {str(e)}',
'error': str(e)
}
def _prepare_analysis_for_ai(self, analysis: Dict, stock_info: Dict) -> str:
"""Prepare analysis summary for AI consumption"""
summary_parts = []
# Price metrics
current_price = analysis.get('current_price', 0)
total_return = analysis.get('total_return_pct', 0)
summary_parts.append(f"Current Price: ${current_price:.2f}")
summary_parts.append(f"Total Return: {total_return:.2f}%")
# Trend analysis
trend_dir = analysis.get('trend_direction', 'unknown')
trend_strength = analysis.get('trend_strength', 0)
summary_parts.append(f"Trend: {trend_dir} (strength: {trend_strength:.2f})")
# Technical indicators
if 'rsi' in analysis:
rsi = analysis['rsi']
rsi_signal = analysis.get('rsi_signal', 'neutral')
summary_parts.append(f"RSI: {rsi:.1f} ({rsi_signal})")
if 'macd_trend' in analysis:
summary_parts.append(f"MACD: {analysis['macd_trend']}")
# Risk metrics
volatility = analysis.get('volatility_annualized', 0)
max_drawdown = analysis.get('max_drawdown', 0)
summary_parts.append(f"Volatility: {volatility:.1f}% (annual)")
summary_parts.append(f"Max Drawdown: {max_drawdown:.1f}%")
# Performance
if 'return_1_month' in analysis:
monthly_return = analysis['return_1_month']
summary_parts.append(f"1-Month Return: {monthly_return:.2f}%")
return "\n".join(summary_parts)
def _parse_ai_response_fallback(self, response: str) -> Dict:
"""Fallback parser for AI response if JSON parsing fails"""
# Simple keyword-based parsing
recommendation = 'HOLD'
confidence = 50
response_lower = response.lower()
if 'buy' in response_lower and 'sell' not in response_lower:
recommendation = 'BUY'
confidence = 70
elif 'sell' in response_lower:
recommendation = 'SELL'
confidence = 70
return {
'recommendation': recommendation,
'confidence': confidence,
'reasoning': response,
'parsed_fallback': True
}
def _combine_decisions(self, algo_decision: Dict, ai_decision: Dict) -> Dict:
"""Combine algorithmic and AI decisions"""
# Weight the decisions (60% algorithmic, 40% AI)
algo_weight = 0.6
ai_weight = 0.4
# Map recommendations to scores
rec_scores = {'BUY': 1, 'HOLD': 0, 'SELL': -1}
algo_score = rec_scores.get(algo_decision['action'], 0)
ai_score = rec_scores.get(ai_decision.get('recommendation', 'HOLD'), 0)
# Calculate combined score
combined_score = (algo_score * algo_weight) + (ai_score * ai_weight)
# Determine final recommendation
if combined_score >= 0.3:
final_action = 'BUY'
elif combined_score <= -0.3:
final_action = 'SELL'
else:
final_action = 'HOLD'
# Calculate confidence
algo_confidence = algo_decision.get('confidence', 0.5)
ai_confidence = ai_decision.get('confidence', 50) / 100
combined_confidence = (algo_confidence * algo_weight) + (ai_confidence * ai_weight)
return {
'action': final_action,
'confidence': combined_confidence,
'combined_score': combined_score,
'reasoning': ai_decision.get('reasoning', 'Combined algorithmic and AI analysis'),
'price_target': ai_decision.get('price_target'),
'stop_loss': ai_decision.get('stop_loss'),
'time_horizon': ai_decision.get('time_horizon', 'medium'),
'key_factors': ai_decision.get('key_factors', [])
}
def _assess_risk_level(self, analysis: Dict) -> str:
"""Assess overall risk level"""
risk_score = 0
# Volatility risk
volatility = analysis.get('volatility_annualized', 0)
if volatility > 40:
risk_score += 2
elif volatility > 25:
risk_score += 1
# Drawdown risk
max_drawdown = abs(analysis.get('max_drawdown', 0))
if max_drawdown > 30:
risk_score += 2
elif max_drawdown > 15:
risk_score += 1
# Sharpe ratio
sharpe = analysis.get('sharpe_ratio', 0)
if sharpe < 0:
risk_score += 1
# Determine risk level
if risk_score >= 4:
return 'high'
elif risk_score >= 2:
return 'medium'
else:
return 'low'
# Global instance for easy import
trading_engine = TradingDecisionEngine()
# Convenience function
def get_trading_recommendation(symbol: str, analysis: Dict, stock_info: Dict) -> Dict:
"""Convenience function to get trading recommendation"""
return trading_engine.get_trading_recommendation(symbol, analysis, stock_info)