agentic-system / ui /venture_ui.py
Cascade Bot
Added Groq streaming support and optimizations - clean version
1d75522
"""UI components for venture strategies and analysis."""
import gradio as gr
import json
from typing import Dict, Any, List
import plotly.graph_objects as go
import plotly.express as px
import pandas as pd
from datetime import datetime
class VentureUI:
"""UI for venture strategies and analysis."""
def __init__(self, api_client):
self.api_client = api_client
def create_interface(self):
"""Create Gradio interface."""
with gr.Blocks(title="Venture Strategy Optimizer") as interface:
gr.Markdown("# Venture Strategy Optimizer")
with gr.Tabs():
# Venture Analysis Tab
with gr.Tab("Venture Analysis"):
with gr.Row():
with gr.Column():
venture_type = gr.Dropdown(
choices=self._get_venture_types(),
label="Venture Type"
)
query = gr.Textbox(
lines=3,
label="Analysis Query"
)
analyze_btn = gr.Button("Analyze Venture")
with gr.Column():
analysis_output = gr.JSON(label="Analysis Results")
metrics_plot = gr.Plot(label="Key Metrics")
analyze_btn.click(
fn=self._analyze_venture,
inputs=[venture_type, query],
outputs=[analysis_output, metrics_plot]
)
# Market Analysis Tab
with gr.Tab("Market Analysis"):
with gr.Row():
with gr.Column():
segment = gr.Textbox(
label="Market Segment"
)
market_btn = gr.Button("Analyze Market")
with gr.Column():
market_output = gr.JSON(label="Market Analysis")
market_plot = gr.Plot(label="Market Trends")
market_btn.click(
fn=self._analyze_market,
inputs=[segment],
outputs=[market_output, market_plot]
)
# Portfolio Optimization Tab
with gr.Tab("Portfolio Optimization"):
with gr.Row():
with gr.Column():
ventures = gr.CheckboxGroup(
choices=self._get_venture_types(),
label="Select Ventures"
)
optimize_btn = gr.Button("Optimize Portfolio")
with gr.Column():
portfolio_output = gr.JSON(label="Portfolio Strategy")
portfolio_plot = gr.Plot(label="Portfolio Allocation")
optimize_btn.click(
fn=self._optimize_portfolio,
inputs=[ventures],
outputs=[portfolio_output, portfolio_plot]
)
# Monetization Strategy Tab
with gr.Tab("Monetization Strategy"):
with gr.Row():
with gr.Column():
monetization_type = gr.Dropdown(
choices=self._get_venture_types(),
label="Venture Type"
)
monetize_btn = gr.Button("Optimize Monetization")
with gr.Column():
monetization_output = gr.JSON(label="Monetization Strategy")
revenue_plot = gr.Plot(label="Revenue Projections")
monetize_btn.click(
fn=self._optimize_monetization,
inputs=[monetization_type],
outputs=[monetization_output, revenue_plot]
)
# Insights Dashboard Tab
with gr.Tab("Insights Dashboard"):
with gr.Row():
refresh_btn = gr.Button("Refresh Insights")
with gr.Row():
with gr.Column():
market_insights = gr.JSON(label="Market Insights")
market_trends = gr.Plot(label="Market Trends")
with gr.Column():
portfolio_insights = gr.JSON(label="Portfolio Insights")
portfolio_trends = gr.Plot(label="Portfolio Performance")
refresh_btn.click(
fn=self._refresh_insights,
outputs=[
market_insights, market_trends,
portfolio_insights, portfolio_trends
]
)
return interface
def _get_venture_types(self) -> List[str]:
"""Get available venture types."""
try:
response = self.api_client.list_strategies()
return response.get("strategies", [])
except Exception as e:
print(f"Error getting venture types: {e}")
return []
def _analyze_venture(self,
venture_type: str,
query: str) -> tuple[Dict[str, Any], go.Figure]:
"""Analyze venture opportunity."""
try:
# Get analysis
response = self.api_client.analyze_venture({
"venture_type": venture_type,
"query": query
})
result = response.get("result", {})
# Create visualization
fig = self._create_venture_plot(result)
return result, fig
except Exception as e:
print(f"Error in venture analysis: {e}")
return {"error": str(e)}, go.Figure()
def _analyze_market(self,
segment: str) -> tuple[Dict[str, Any], go.Figure]:
"""Analyze market opportunity."""
try:
# Get analysis
response = self.api_client.analyze_market({
"segment": segment
})
result = response.get("result", {})
# Create visualization
fig = self._create_market_plot(result)
return result, fig
except Exception as e:
print(f"Error in market analysis: {e}")
return {"error": str(e)}, go.Figure()
def _optimize_portfolio(self,
ventures: List[str]) -> tuple[Dict[str, Any], go.Figure]:
"""Optimize venture portfolio."""
try:
# Get optimization
response = self.api_client.optimize_portfolio({
"ventures": ventures
})
result = response.get("result", {})
# Create visualization
fig = self._create_portfolio_plot(result)
return result, fig
except Exception as e:
print(f"Error in portfolio optimization: {e}")
return {"error": str(e)}, go.Figure()
def _optimize_monetization(self,
venture_type: str) -> tuple[Dict[str, Any], go.Figure]:
"""Optimize monetization strategy."""
try:
# Get optimization
response = self.api_client.optimize_monetization({
"venture_type": venture_type
})
result = response.get("result", {})
# Create visualization
fig = self._create_revenue_plot(result)
return result, fig
except Exception as e:
print(f"Error in monetization optimization: {e}")
return {"error": str(e)}, go.Figure()
def _refresh_insights(self) -> tuple[Dict[str, Any], go.Figure,
Dict[str, Any], go.Figure]:
"""Refresh insights dashboard."""
try:
# Get insights
market_response = self.api_client.get_market_insights()
portfolio_response = self.api_client.get_portfolio_insights()
market_insights = market_response.get("insights", {})
portfolio_insights = portfolio_response.get("insights", {})
# Create visualizations
market_fig = self._create_market_trends_plot(market_insights)
portfolio_fig = self._create_portfolio_trends_plot(portfolio_insights)
return market_insights, market_fig, portfolio_insights, portfolio_fig
except Exception as e:
print(f"Error refreshing insights: {e}")
return (
{"error": str(e)}, go.Figure(),
{"error": str(e)}, go.Figure()
)
def _create_venture_plot(self, data: Dict[str, Any]) -> go.Figure:
"""Create venture analysis visualization."""
try:
metrics = data.get("metrics", {})
fig = go.Figure()
fig.add_trace(go.Scatterpolar(
r=[
metrics.get("market_score", 0),
metrics.get("opportunity_score", 0),
metrics.get("risk_score", 0),
metrics.get("growth_potential", 0),
metrics.get("profitability", 0)
],
theta=[
"Market Score",
"Opportunity Score",
"Risk Score",
"Growth Potential",
"Profitability"
],
fill='toself'
))
fig.update_layout(
polar=dict(
radialaxis=dict(
visible=True,
range=[0, 1]
)
),
showlegend=False
)
return fig
except Exception as e:
print(f"Error creating venture plot: {e}")
return go.Figure()
def _create_market_plot(self, data: Dict[str, Any]) -> go.Figure:
"""Create market analysis visualization."""
try:
trends = data.get("trend_analysis", {})
df = pd.DataFrame([
{
"Trend": trend["name"],
"Impact": trend["impact"],
"Potential": trend["market_potential"],
"Risk": trend["risk_level"]
}
for trend in trends
])
fig = px.scatter(
df,
x="Impact",
y="Potential",
size="Risk",
hover_data=["Trend"],
title="Market Trends Analysis"
)
return fig
except Exception as e:
print(f"Error creating market plot: {e}")
return go.Figure()
def _create_portfolio_plot(self, data: Dict[str, Any]) -> go.Figure:
"""Create portfolio optimization visualization."""
try:
allocation = data.get("allocation", {})
fig = go.Figure(data=[
go.Bar(
name=venture,
x=["Resources", "Priority", "Risk"],
y=[
sum(resources.values()),
priority,
len(constraints)
]
)
for venture, (resources, priority, constraints) in allocation.items()
])
fig.update_layout(
barmode='group',
title="Portfolio Allocation"
)
return fig
except Exception as e:
print(f"Error creating portfolio plot: {e}")
return go.Figure()
def _create_revenue_plot(self, data: Dict[str, Any]) -> go.Figure:
"""Create revenue projection visualization."""
try:
projections = data.get("projections", {})
months = list(range(12))
revenue = [
projections.get("monthly_revenue", {}).get(str(m), 0)
for m in months
]
fig = go.Figure()
fig.add_trace(go.Scatter(
x=months,
y=revenue,
mode='lines+markers',
name='Revenue'
))
fig.update_layout(
title="Revenue Projections",
xaxis_title="Month",
yaxis_title="Revenue ($)"
)
return fig
except Exception as e:
print(f"Error creating revenue plot: {e}")
return go.Figure()
def _create_market_trends_plot(self, data: Dict[str, Any]) -> go.Figure:
"""Create market trends visualization."""
try:
trends = data.get("trend_insights", [])
df = pd.DataFrame(trends)
fig = px.scatter(
df,
x="impact",
y="potential",
size="risk",
hover_data=["name"],
title="Market Trends Overview"
)
return fig
except Exception as e:
print(f"Error creating market trends plot: {e}")
return go.Figure()
def _create_portfolio_trends_plot(self, data: Dict[str, Any]) -> go.Figure:
"""Create portfolio trends visualization."""
try:
metrics = data.get("portfolio_metrics", {})
fig = go.Figure()
fig.add_trace(go.Indicator(
mode="gauge+number",
value=metrics.get("total_revenue", 0),
title={'text': "Total Revenue ($M)"},
gauge={'axis': {'range': [None, 10]}}
))
return fig
except Exception as e:
print(f"Error creating portfolio trends plot: {e}")
return go.Figure()