import gradio as gr
import yfinance as yf
import pandas as pd
import numpy as np
from datetime import datetime
import plotly.graph_objects as go
from plotly.subplots import make_subplots
import pixeltable as pxt
from pixeltable.functions import openai
import json
import os
import getpass
from typing import Dict, Any
# Set up OpenAI API key
if 'OPENAI_API_KEY' not in os.environ:
os.environ['OPENAI_API_KEY'] = getpass.getpass('Enter your OpenAI API key: ')
class NumpyEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, (np.int_, np.intc, np.intp, np.int8,
np.int16, np.int32, np.int64, np.uint8,
np.uint16, np.uint32, np.uint64)):
return int(obj)
elif isinstance(obj, (np.float_, np.float16, np.float32, np.float64)):
return float(obj)
elif isinstance(obj, (np.ndarray,)):
return obj.tolist()
return json.JSONEncoder.default(self, obj)
def safe_json_serialize(obj):
return json.dumps(obj, cls=NumpyEncoder)
def calculate_basic_indicators(data: pd.DataFrame) -> pd.DataFrame:
df = data.copy()
# Moving averages
df['MA20'] = df['Close'].rolling(window=20).mean()
df['MA50'] = df['Close'].rolling(window=50).mean()
df['MA200'] = df['Close'].rolling(window=200).mean()
# RSI
delta = df['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
df['RSI'] = 100 - (100 / (1 + rs))
# MACD
exp1 = df['Close'].ewm(span=12, adjust=False).mean()
exp2 = df['Close'].ewm(span=26, adjust=False).mean()
df['MACD'] = exp1 - exp2
df['MACD_Signal'] = df['MACD'].ewm(span=9, adjust=False).mean()
return df.ffill().bfill()
@pxt.udf
def generate_analysis_prompt(data: str, analysis_type: str, time_horizon: str,
risk_tolerance: str, investment_style: str,
technical_depth: str) -> list[dict]:
"""Generate a structured prompt for AI analysis"""
# Create specific guidance based on parameters
time_horizon_guidance = {
'short': 'Focus on short-term trading opportunities within 1-3 months. Emphasize technical signals and immediate catalysts.',
'medium': 'Balance short-term opportunities with medium-term trends over 3-12 months. Consider both technical and fundamental factors.',
'long': 'Prioritize long-term growth potential over 1+ years. Emphasize fundamental analysis and strategic positioning.'
}
risk_tolerance_guidance = {
'conservative': 'Prioritize capital preservation. Focus on stable, large-cap stocks with strong fundamentals. Suggest conservative options strategies like covered calls.',
'moderate': 'Balance growth opportunities with risk management. Consider mid-cap stocks and moderate options strategies.',
'aggressive': 'Seek high-growth opportunities. Include small-cap stocks and more aggressive options strategies in the analysis.'
}
investment_style_guidance = {
'value': 'Focus on valuation metrics, margin of safety, and undervalued opportunities.',
'growth': 'Emphasize revenue growth, market expansion, and future potential.',
'momentum': 'Focus on price trends, relative strength, and technical indicators.',
'balanced': 'Consider both value and growth factors, maintaining a balanced perspective.',
'income': 'Prioritize dividend yield, payout ratio, and income-generating options strategies.'
}
analysis_depth = {
'comprehensive': 'Provide detailed analysis across all categories.',
'quantitative': 'Focus on numerical metrics and statistical analysis.',
'technical': 'Emphasize technical analysis and chart patterns.'
}
technical_guidance = {
'basic': 'Focus on essential technical indicators (MA, RSI, MACD).',
'advanced': 'Include advanced technical analysis including Fibonacci levels, Elliott Wave patterns, and advanced options analysis.'
}
system_prompt = f'''You are a senior investment advisor and market analyst with decades of experience, holding CFA and CMT certifications. Provide analysis tailored to the following parameters:
TIME HORIZON: {time_horizon.upper()}
{time_horizon_guidance[time_horizon]}
RISK PROFILE: {risk_tolerance.upper()}
{risk_tolerance_guidance[risk_tolerance]}
INVESTMENT STYLE: {investment_style.upper()}
{investment_style_guidance[investment_style]}
ANALYSIS FOCUS: {analysis_type.upper()}
{analysis_depth[analysis_type]}
TECHNICAL DEPTH: {technical_depth.upper()}
{technical_guidance[technical_depth]}
Structure your response using EXACTLY the following format and sections:
SUMMARY
Provide a clear 2-3 sentence executive summary aligned with the specified time horizon ({time_horizon}) and risk tolerance ({risk_tolerance}).
TECHNICAL ANALYSIS
{technical_guidance[technical_depth]}
• Moving Averages: Analyze trends with emphasis on {time_horizon} timeframe
• RSI Analysis: Current RSI level and implications
• MACD Analysis: Signal trends relevant to {time_horizon} horizon
• Volume Analysis: Notable volume patterns and implications
{"• Advanced Patterns: Fibonacci levels, Elliott Wave analysis" if technical_depth == "advanced" else ""}
MARKET CONTEXT
• Sector Analysis: {"Long-term industry trends and positioning" if time_horizon == "long" else "Short-term sector momentum" if time_horizon == "short" else "Medium-term sector outlook"}
• {"Fundamental Drivers: Key growth catalysts" if investment_style == "growth" else "Value Metrics: Key valuation catalysts" if investment_style == "value" else "Market Dynamics: Key price catalysts"}
• Economic Impact: Factors relevant to {time_horizon} horizon
RISKS
• Identify and quantify risks specific to {risk_tolerance} risk tolerance
• Focus on {time_horizon} horizon risks
• {"Include volatility analysis" if technical_depth == "advanced" else ""}
OPPORTUNITIES
• Align opportunities with {investment_style} style
• Focus on {time_horizon} timeline
• Match risk/reward to {risk_tolerance} profile
OPTIONS STRATEGY
{"• Advanced Options Analysis: Complex strategies and volatility analysis" if technical_depth == "advanced" else "• Basic Options Strategies: Simple hedging and income generation"}
• Tailor strategies to {risk_tolerance} risk tolerance
• Focus on {time_horizon} expiration cycles
• Match strategies to {investment_style} style objectives
RECOMMENDATION
• Provide specific recommendations aligned with:
- {time_horizon.capitalize()} time horizon
- {risk_tolerance.capitalize()} risk tolerance
- {investment_style.capitalize()} investment style
• Include position sizing appropriate for risk level
• Specify entry, exit, and stop-loss levels
• {"Include advanced technical levels" if technical_depth == "advanced" else "Focus on key support/resistance levels"}
IMPORTANT: This analysis is for informational purposes only. All investments carry risk. Please consult with a licensed financial advisor before making investment decisions.'''
return [
{'role': 'system', 'content': system_prompt},
{'role': 'user', 'content': f'Analyze this market data considering the specified parameters:\n{data}'}
]
def parse_analysis_response(response: str) -> Dict[str, str]:
"""Parse the structured AI response into sections with support for markdown formatting"""
sections = {
'SUMMARY': None,
'TECHNICAL ANALYSIS': None,
'MARKET CONTEXT': None,
'RISKS': None,
'OPPORTUNITIES': None,
'OPTIONS STRATEGY': None,
'RECOMMENDATION': None
}
current_section = None
buffer = []
if not response or not response.strip():
return {k: "Analysis not available" for k in sections.keys()}
for line in response.split('\n'):
line = line.strip()
matched_section = None
for section in sections.keys():
# Remove asterisks and check for exact match
cleaned_line = line.replace('*', '').strip()
if cleaned_line == section:
matched_section = section
break
if matched_section:
# Save previous section if exists
if current_section and buffer:
sections[current_section] = '\n'.join(buffer).strip()
current_section = matched_section
buffer = []
elif current_section and line:
# Clean up markdown formatting in content
cleaned_content = line.replace('*', '').strip()
if cleaned_content: # Only add non-empty lines
buffer.append(cleaned_content)
if current_section and buffer:
sections[current_section] = '\n'.join(buffer).strip()
section_messages = {
'SUMMARY': 'Market analysis summary not available',
'TECHNICAL ANALYSIS': 'Technical analysis not available',
'MARKET CONTEXT': 'Market context information not available',
'RISKS': 'Risk assessment not available',
'OPPORTUNITIES': 'Opportunity analysis not available',
'OPTIONS STRATEGY': 'Options trading analysis not available',
'RECOMMENDATION': 'Investment recommendation not available'
}
for key in sections:
if sections[key] is None or not sections[key].strip():
sections[key] = section_messages[key]
return sections
def create_visualization(data: pd.DataFrame, technical_depth: str) -> go.Figure:
fig = make_subplots(
rows=3 if technical_depth == 'advanced' else 2,
cols=1,
shared_xaxes=True,
vertical_spacing=0.05,
subplot_titles=('Price & Moving Averages', 'Volume', 'RSI' if technical_depth == 'advanced' else None)
)
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='#26A69A',
decreasing_line_color='#EF5350'
),
row=1, col=1
)
colors = {'MA20': '#1E88E5', 'MA50': '#FFC107', 'MA200': '#7B1FA2'}
for ma, color in colors.items():
fig.add_trace(
go.Scatter(
x=data.index,
y=data[ma],
name=ma,
line=dict(color=color, width=1.5)
),
row=1, col=1
)
colors = ['#26A69A' if close >= open_price else '#EF5350'
for close, open_price in zip(data['Close'].values, data['Open'].values)]
fig.add_trace(
go.Bar(
x=data.index,
y=data['Volume'],
name='Volume',
marker_color=colors
),
row=2, col=1
)
if technical_depth == 'advanced':
fig.add_trace(
go.Scatter(
x=data.index,
y=data['RSI'],
name='RSI',
line=dict(color='#7C4DFF', width=1.5)
),
row=3, col=1
)
fig.add_hline(y=70, line_dash="dash", line_color="red", row=3, col=1)
fig.add_hline(y=30, line_dash="dash", line_color="green", row=3, col=1)
fig.update_layout(
height=800,
template='plotly_white',
showlegend=True,
legend=dict(
orientation="h",
yanchor="bottom",
y=1.02,
xanchor="right",
x=1
)
)
fig.update_yaxes(title_text="Price", row=1, col=1)
fig.update_yaxes(title_text="Volume", row=2, col=1)
if technical_depth == 'advanced':
fig.update_yaxes(title_text="RSI", row=3, col=1)
return fig
def process_outputs(ticker_symbol, analysis_type, time_horizon, risk_tolerance,
investment_style, technical_depth, include_market_context=True,
max_positions=3):
try:
# Initialize Pixeltable
pxt.drop_dir('financial_analysis', force=True)
pxt.create_dir('financial_analysis')
data_table = pxt.create_table(
'financial_analysis.stock_data',
{
'ticker': pxt.StringType(),
'data': pxt.StringType(),
'timestamp': pxt.TimestampType()
}
)
# Fetch and process data
stock = yf.Ticker(ticker_symbol.strip().upper())
market_data = stock.history(period='1y')
if market_data.empty:
raise ValueError("No data found for the specified ticker symbol.")
options_data = stock.options # Get available expiration dates
if len(options_data) > 0:
nearest_expiry = options_data[0]
opt_chain = stock.option_chain(nearest_expiry)
options_info = {
'Nearest Expiry': nearest_expiry,
'Call Volume': int(opt_chain.calls['volume'].sum()),
'Put Volume': int(opt_chain.puts['volume'].sum()),
'Put/Call Ratio': float(opt_chain.puts['volume'].sum() / opt_chain.calls['volume'].sum()),
'Implied Volatility': float(opt_chain.calls['impliedVolatility'].mean())
}
else:
options_info = {
'Nearest Expiry': 'N/A',
'Call Volume': 0,
'Put Volume': 0,
'Put/Call Ratio': 0,
'Implied Volatility': 0
}
technical_data = calculate_basic_indicators(market_data)
market_data_json = technical_data.to_json(date_format='iso')
# Store data and generate analysis
data_table.insert([{
'ticker': ticker_symbol.upper(),
'data': market_data_json,
'timestamp': datetime.now()
}])
data_table['prompt'] = generate_analysis_prompt(
data_table.data,
analysis_type,
time_horizon,
risk_tolerance,
investment_style,
technical_depth
)
data_table['analysis'] = openai.chat_completions(
messages=data_table.prompt,
model='gpt-4o-mini-2024-07-18',
temperature=0.7,
max_tokens=1000
)
try:
analysis_text = data_table.select(
analysis=data_table.analysis.choices[0].message.content
).tail(1)['analysis'][0]
parsed_analysis = parse_analysis_response(analysis_text)
except Exception as analysis_error:
print(f"Analysis error: {str(analysis_error)}")
parsed_analysis = parse_analysis_response("") # This will return default messages
# Prepare company info with proper JSON formatting
company_info_data = {
'Name': str(stock.info.get('longName', 'N/A')),
'Sector': str(stock.info.get('sector', 'N/A')),
'Industry': str(stock.info.get('industry', 'N/A')),
'Exchange': str(stock.info.get('exchange', 'N/A'))
}
raw_llm_output = ""
try:
raw_llm_output = data_table.select(
analysis=data_table.analysis.choices[0].message.content
).tail(1)['analysis'][0]
parsed_analysis = parse_analysis_response(raw_llm_output)
except Exception as analysis_error:
print(f"Analysis error: {str(analysis_error)}")
parsed_analysis = parse_analysis_response("")
raw_llm_output = f"Error processing analysis: {str(analysis_error)}"
try:
current_price = float(technical_data['Close'].iloc[-1])
previous_price = float(technical_data['Close'].iloc[-2])
daily_change = float((current_price / previous_price - 1) * 100)
volume = int(technical_data['Volume'].iloc[-1])
rsi = float(technical_data['RSI'].iloc[-1])
except (IndexError, KeyError, TypeError):
current_price = daily_change = volume = rsi = 0
market_stats_data = {
'Current Price': f"${current_price:.2f}",
'Daily Change': f"{daily_change:.2f}%",
'Volume': f"{volume:,}",
'RSI': f"{rsi:.2f}"
}
technical_data_with_time = technical_data.reset_index()
technical_data_with_time['Date'] = technical_data_with_time['Date'].dt.strftime('%Y-%m-%d %H:%M:%S')
# Create visualization
plot = create_visualization(technical_data, technical_depth)
return (
json.dumps(company_info_data),
json.dumps(market_stats_data),
json.dumps(options_info), # Add options info to return values
plot,
parsed_analysis['SUMMARY'],
parsed_analysis['TECHNICAL ANALYSIS'],
parsed_analysis['MARKET CONTEXT'],
parsed_analysis['RISKS'],
parsed_analysis['OPPORTUNITIES'],
parsed_analysis['OPTIONS STRATEGY'], # Add options strategy
parsed_analysis['RECOMMENDATION'],
technical_data_with_time,
raw_llm_output
)
except Exception as e:
error_msg = f"Error processing data: {str(e)}"
empty_json = json.dumps({})
no_data_msg = "Analysis not available due to data processing error"
empty_df = pd.DataFrame()
return (
empty_json,
empty_json,
empty_json,
None,
no_data_msg,
no_data_msg,
no_data_msg,
no_data_msg,
no_data_msg,
no_data_msg,
no_data_msg,
empty_df,
f"Error occurred: {str(e)}"
)
def create_interface() -> gr.Blocks:
"""Create the production-ready Gradio interface"""
with gr.Blocks(theme=gr.themes.Base()) as demo:
# Header
gr.Markdown(
"""
# 📈 AI Financial Analysis Platform
AI-powered market analysis and technical indicators powered by OpenAI GPT models and Pixeltable.
"""
)
gr.HTML(
"""
⚠️ Important Data Reliability Notice:
This is a demonstration platform showcasing how to leverage LLMs for financial analysis.
The accuracy of information, data retrieved from yfinance, and calculated metrics are NOT guaranteed.
Use at your own risk. The creators and operators of this tool are not responsible for any financial losses or decisions made based on this analysis. This tool is intended for demonstration and educational purposes only.
"""
)
with gr.Row():
with gr.Column():
with gr.Accordion("🎯 What does it do?", open=False):
gr.Markdown("""
This platform provides comprehensive financial analysis tools:
1. 📊 **Technical Analysis**: Advanced indicators, e.g. RSI, and MACD
2. 🤖 **AI-Powered Insights**: Intelligent market analysis/recommendations
3. 📈 **Interactive Charts**: Visual representation of movements/indicators
4. 💡 **Investment Context**: Market conditions and sector analysis
5. ⚡ **Real-time Data**: Up-to-date information through Yahoo Finance
6. 🎯 **Personalized Analysis**: Tailored to your style/risk tolerance
""")
with gr.Column():
with gr.Accordion("🛠️ How does it work?", open=False):
gr.Markdown("""
The platform leverages several advanced technologies:
1. 📦 **Data Processing**: Pixeltable manages and orchestrate data
2. 🔍 **Technical Indicators**: Custom algorithms calculate market metrics
3. 🤖 **AI Analysis**: Advanced language models provide market insights
4. 📊 **Visualization**: Interactive charts using Plotly
5. 🔄 **Real-time Updates**: Direct connection to market data feeds
6. 💾 **Data Persistence**: Reliable storage and retrieval of insights
""")
# Disclaimer
gr.HTML(
"""
⚠️ Disclaimer:
This tool provides financial analysis for informational purposes only and should not be considered as financial advice, please:
- Consult with qualified financial advisors
- Conduct your own research
- Consider your personal financial situation
- Be aware that past performance does not guarantee future results
- Understand that all investments carry risk
"""
)
with gr.Row():
# Left sidebar for inputs
with gr.Column(scale=1):
with gr.Row():
gr.Markdown("### 📊 Analysis Parameters")
with gr.Row():
ticker_input = gr.Textbox(
label="Stock Ticker",
placeholder="e.g., AAPL",
max_lines=1
)
analysis_type = gr.Radio(
choices=['comprehensive', 'quantitative', 'technical'],
label="Analysis Type",
value='comprehensive'
)
technical_depth = gr.Radio(
choices=['basic', 'advanced'],
label="Technical Depth",
value='advanced'
)
with gr.Row():
gr.Markdown("### 🎯 Investment Profile")
with gr.Row():
time_horizon = gr.Radio(
choices=['short', 'medium', 'long'],
label="Time Horizon",
value='medium'
)
risk_tolerance = gr.Radio(
choices=['conservative', 'moderate', 'aggressive'],
label="Risk Tolerance",
value='moderate'
)
investment_style = gr.Dropdown(
choices=['value', 'growth', 'momentum', 'balanced', 'income'],
label="Investment Style",
value='balanced'
)
analyze_btn = gr.Button("📊 Analyze Stock", variant="primary")
with gr.Row():
with gr.Column(scale=3):
with gr.Tabs() as tabs:
with gr.TabItem("📊 Analysis Dashboard"):
# Top row with company info and market stats
with gr.Row(equal_height=True):
with gr.Column(scale=1):
company_info = gr.JSON(
label="Company Information",
height=150
)
with gr.Column(scale=1):
market_stats = gr.JSON(
label="Market Statistics",
height=150
)
with gr.Column(scale=1):
options_info = gr.JSON(
label="Options Market Data",
height=150
)
with gr.TabItem("📑 Historical Data"):
technical_data = gr.DataFrame(
headers=["Date", "Open", "High", "Low", "Close",
"Volume", "MA20", "MA50", "MA200", "RSI",
"MACD", "MACD_Signal"],
)
with gr.TabItem("🔍 Debug View"):
raw_output = gr.Textbox(
label="Raw LLM Output",
lines=10,
max_lines=20,
show_label=True,
interactive=False
)
gr.Markdown("""
### Debug Information
This tab shows the raw output from the language model before parsing.
Use this to diagnose any issues with the analysis display.
""")
# Technical analysis chart
with gr.Row():
with gr.Column(scale=1):
with gr.Row():
gr.Markdown("### 📈 Technical Analysis Chart")
with gr.Row():
plot_output = gr.Plot()
# AI Analysis section
with gr.Row():
with gr.Column(scale=2):
with gr.Row():
gr.Markdown("### 🤖 AI Analysis")
with gr.Row():
summary = gr.Textbox(
label="Executive Summary",
lines=3,
max_lines=5,
show_label=True
)
with gr.Row():
with gr.Column(scale=1):
tech_analysis = gr.Textbox(
label="Technical Analysis",
lines=8,
max_lines=10,
show_label=True
)
market_context = gr.Textbox(
label="Market Context",
lines=4,
max_lines=6,
show_label=True
)
with gr.Column(scale=1):
risks = gr.Textbox(
label="Key Risks",
lines=5,
max_lines=7,
show_label=True
)
opportunities = gr.Textbox(
label="Key Opportunities",
lines=5,
max_lines=7,
show_label=True
)
with gr.Row():
options_strategy = gr.Textbox(
label="Options Trading Strategy",
lines=8,
max_lines=10,
show_label=True
)
with gr.Row():
recommendation = gr.Textbox(
label="Investment Recommendation",
lines=3,
max_lines=5,
show_label=True
)
gr.Examples(
examples=[
["AAPL", "comprehensive", "medium", "moderate", "balanced", "advanced"],
["MSFT", "technical", "short", "aggressive", "momentum", "basic"],
["GOOGL", "quantitative", "long", "conservative", "value", "advanced"]
],
inputs=[
ticker_input, analysis_type, time_horizon, risk_tolerance,
investment_style, technical_depth
]
)
# Footer
gr.HTML(
"""
🚀 Built with Pixeltable
Open Source AI Data infrastructure for building intelligent applications.
© 2024 AI Financial Analysis Platform powered by Pixeltable.
This work is licensed under the Apache License 2.0.
"""
)
analyze_btn.click(
process_outputs,
inputs=[
ticker_input, analysis_type, time_horizon, risk_tolerance,
investment_style, technical_depth
],
outputs=[
company_info, market_stats, options_info, plot_output,
summary, tech_analysis, market_context,
risks, opportunities, options_strategy, recommendation,
technical_data, raw_output
]
)
return demo
if __name__ == "__main__":
demo = create_interface()
demo.launch()