import streamlit as st
import requests
import json
import plotly.graph_objects as go
import plotly.express as px
from typing import Dict, List, Any
import time
import pandas as pd
# Page configuration
st.set_page_config(
page_title="Market Research & Analysis Platform",
layout="wide",
initial_sidebar_state="expanded"
)
# Custom styling (updated for dark mode)
st.markdown("""
""", unsafe_allow_html=True)
def query_perplexity(query: str, context: Dict) -> Dict:
url = "https://api.perplexity.ai/chat/completions"
headers = {
"Authorization": f"Bearer {st.secrets['PERPLEXITY_API_KEY']}",
"Content-Type": "application/json"
}
payload = {
"model": "llama-3.1-sonar-small-128k-online",
"messages": [
{"role": "system", "content": get_system_prompt(context)},
{"role": "user", "content": query}
],
"temperature": 0.2,
"max_tokens": 4096,
"top_p": 0.9,
"search_domain_filter": ["perplexity.ai"],
"return_images": False,
"return_related_questions": False,
"search_recency_filter": context.get('timeframe', 'month')
}
try:
response = requests.post(url, headers=headers, json=payload)
if response.status_code == 200:
return {
"status": "success",
"data": response.json(),
"citations": response.json().get("citations", [])
}
else:
return {"status": "error", "message": f"API Error: {response.status_code}"}
except Exception as e:
return {"status": "error", "message": str(e)}
def get_system_prompt(context: Dict) -> str:
# Adjust the prompt if Executive style and Comprehensive depth are selected.
if context.get("style", "").lower() == "executive" and context.get("depth", "").lower() == "comprehensive":
return f"""You are an expert market research analyst focusing on {context['focus_area']}.
Provide a high-level executive summary with key insights and concrete metrics.
Analysis style: Executive
Depth: Comprehensive
Timeline: Past {context['timeframe']}
Format your response with clear sections:
1. Executive Summary
2. Key Metrics
3. Market Position
4. Growth Analysis
5. Strategic Recommendations
Include specific numbers, percentages, and actionable insights."""
else:
return f"""You are an expert market research analyst focusing on {context['focus_area']}.
Provide detailed analysis with concrete metrics and specific insights.
Analysis style: {context['style']}
Depth: {context['depth']}
Timeline: Past {context['timeframe']}
Format your response with clear sections:
1. Key Metrics
2. Market Position
3. Growth Analysis
4. Competitive Insights
5. Strategic Recommendations
Include specific numbers, percentages, and actionable insights."""
def parse_perplexity_response(response: Dict) -> Dict:
try:
content = response['data']['choices'][0]['message']['content']
citations = response.get('citations', [])
sections = {
'key_metrics': [],
'market_position': [],
'growth_analysis': [],
'competitive_insights': [],
'recommendations': []
}
current_section = None
for line in content.split('\n'):
line = line.strip()
if not line:
continue
lower_line = line.lower()
if 'key metric' in lower_line:
current_section = 'key_metrics'
elif 'market position' in lower_line:
current_section = 'market_position'
elif 'growth' in lower_line:
current_section = 'growth_analysis'
elif 'competiti' in lower_line:
current_section = 'competitive_insights'
elif 'recommend' in lower_line:
current_section = 'recommendations'
elif current_section and line.startswith(('-', '•', '*')):
sections[current_section].append(line.lstrip('-•* '))
return {"status": "success", "sections": sections, "citations": citations}
except Exception as e:
return {"status": "error", "message": str(e)}
def extract_metrics(content: Dict) -> Dict:
metrics = {
'market_share': [],
'growth_rate': [],
'competitive_position': [],
'innovation_score': []
}
try:
for section in content['sections'].values():
for line in section:
if '%' in line or any(char.isdigit() for char in line):
if 'market share' in line.lower():
metrics['market_share'].append(extract_number(line))
elif 'growth' in line.lower():
metrics['growth_rate'].append(extract_number(line))
elif 'position' in line.lower():
metrics['competitive_position'].append(extract_number(line))
elif 'innovation' in line.lower():
metrics['innovation_score'].append(extract_number(line))
return metrics
except Exception as e:
st.error(f"Error extracting metrics: {str(e)}")
return metrics
def extract_number(text: str) -> float:
import re
numbers = re.findall(r'[-+]?\d*\.?\d+%?', text)
if numbers:
number = numbers[0]
return float(number.replace('%', '')) if '%' in number else float(number)
return 0.0
def create_visualizations(metrics: Dict, context: Dict) -> Dict:
charts = {}
if metrics['market_share'] and metrics['competitive_position']:
fig = go.Figure()
categories = ['Market Share', 'Growth Rate', 'Competitive Position', 'Innovation Score']
values = [
metrics['market_share'][0] if metrics['market_share'] else 0,
metrics['growth_rate'][0] if metrics['growth_rate'] else 0,
metrics['competitive_position'][0] if metrics['competitive_position'] else 0,
metrics['innovation_score'][0] if metrics['innovation_score'] else 0
]
fig.add_trace(go.Scatterpolar(
r=values,
theta=categories,
fill='toself',
name=context['company_name']
))
fig.update_layout(
polar=dict(radialaxis=dict(visible=True, range=[0, 100])),
showlegend=False,
title=f"Market Position Analysis - {context['company_name']}"
)
charts['market_position'] = fig
if metrics['growth_rate']:
fig = go.Figure()
fig.add_trace(go.Scatter(
y=metrics['growth_rate'],
mode='lines+markers',
name='Growth Rate'
))
fig.update_layout(
title=f"Growth Trend Analysis - {context['company_name']}",
yaxis_title='Growth Rate (%)',
showlegend=True
)
charts['growth_trend'] = fig
return charts
def format_insights(content: Dict) -> str:
"""Format detailed analysis with sub-titles for each section"""
formatted = ""
section_titles = {
'key_metrics': '📊 Key Metrics',
'market_position': '🎯 Market Position',
'growth_analysis': '📈 Growth Analysis',
'competitive_insights': '🔍 Competitive Insights',
'recommendations': '💡 Strategic Recommendations'
}
for section, title in section_titles.items():
if content['sections'].get(section):
formatted += f"\n### {title}\n\n"
for idx, point in enumerate(content['sections'][section], start=1):
formatted += f"- **{idx}.** {point}\n"
return formatted
def main():
st.title("Market Research & Analysis Platform")
st.markdown("Real-time market insights with data-driven analysis")
with st.sidebar:
st.header("Analysis Parameters")
company_name = st.text_input("Company/Product Name", placeholder="e.g., Tesla, OpenAI, Snowflake")
industry = st.selectbox("Industry", ["Technology", "AI/ML", "SaaS", "Fintech", "E-commerce", "Healthcare", "Energy", "Other"])
st.markdown("### Analysis Configuration")
focus_area = st.multiselect("Focus Areas", ["Market Position", "Growth Trajectory", "Technology Stack", "Competitive Analysis", "Innovation Trends", "Investment Outlook"], default=["Market Position", "Growth Trajectory"])
timeframe = st.select_slider("Analysis Timeframe", options=["week", "month", "quarter", "year"], value="month")
depth = st.select_slider("Analysis Depth", options=["Brief", "Detailed", "Comprehensive"], value="Detailed")
style = st.selectbox("Report Style", ["Technical", "Business", "Executive"], index=1)
competitors = st.text_input("Key Competitors (optional)", placeholder="Comma-separated names")
if st.button("Generate Analysis", type="primary"):
if not company_name:
st.warning("Please enter a company name.")
return
analysis_context = {
"company_name": company_name,
"industry": industry,
"focus_area": ", ".join(focus_area),
"timeframe": timeframe,
"depth": depth,
"style": style,
"competitors": competitors
}
with st.spinner("Generating market analysis..."):
progress_bar = st.progress(0)
status_text = st.empty()
try:
status_text.text("Gathering market intelligence...")
progress_bar.progress(20)
research_response = query_perplexity(
f"Provide a detailed market analysis for {company_name} in the {industry} industry, focusing on {', '.join(focus_area)}",
analysis_context
)
if research_response["status"] != "success":
st.error("Failed to gather market intelligence.")
return
status_text.text("Processing insights...")
progress_bar.progress(40)
parsed_content = parse_perplexity_response(research_response)
if parsed_content["status"] != "success":
st.error("Failed to process insights.")
return
status_text.text("Generating visualizations...")
progress_bar.progress(60)
metrics = extract_metrics(parsed_content)
charts = create_visualizations(metrics, analysis_context)
status_text.text("Preparing analysis report...")
progress_bar.progress(80)
tabs = st.tabs(["Overview", "Detailed Analysis", "Visualizations"])
with tabs[0]:
st.markdown("## Executive Summary", unsafe_allow_html=True)
if metrics:
cols = st.columns(len(metrics))
for col, (metric, values) in zip(cols, metrics.items()):
if values:
col.metric(metric.replace('_', ' ').title(), f"{values[0]:.1f}%")
if parsed_content["citations"]:
st.markdown("### Sources")
for citation in parsed_content["citations"]:
st.markdown(f'{citation}
', unsafe_allow_html=True)
with tabs[1]:
st.markdown(format_insights(parsed_content), unsafe_allow_html=True)
with tabs[2]:
for chart_name, fig in charts.items():
st.plotly_chart(fig, use_container_width=True)
progress_bar.progress(100)
status_text.text("Analysis complete!")
st.download_button(
"Download Analysis",
data=json.dumps({
"context": analysis_context,
"insights": parsed_content["sections"],
"metrics": metrics,
"citations": parsed_content["citations"]
}, indent=2),
file_name=f"market_analysis_{company_name}.json",
mime="application/json"
)
except Exception as e:
st.error(f"An error occurred: {str(e)}")
return
if __name__ == "__main__":
main()