import os import time import uuid from typing import List, Tuple, Optional, Dict, Union import google.generativeai as genai import gradio as gr from PIL import Image from langchain.chat_models import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.chains import LLMChain, SequentialChain from textwrap import dedent import google.generativeai as genai import yfinance as yf from pypfopt.discrete_allocation import DiscreteAllocation, get_latest_prices from pypfopt import EfficientFrontier from pypfopt import risk_models from pypfopt import expected_returns from pypfopt import plotting import copy import numpy as np import pandas as pd import plotly.express as px import matplotlib.pyplot as plt from datetime import datetime import datetime # Tool import from crewai.tools.gemini_tools import GeminiSearchTools from langchain.tools.yahoo_finance_news import YahooFinanceNewsTool from crewai.tools.browser_tools import BrowserTools from crewai.tools.sec_tools import SECTools # Google Langchain from langchain_google_genai import GoogleGenerativeAI #Crew imports from crewai import Agent, Task, Crew, Process # Retrieve API Key from Environment Variable GOOGLE_AI_STUDIO = os.environ.get('GOOGLE_API_KEY') # Ensure the API key is available if not GOOGLE_AI_STUDIO: raise ValueError("API key not found. Please set the GOOGLE_AI_STUDIO2 environment variable.") # LangChain function for company analysis def company_analysis(api_key: str, company_name: str) -> dict: os.environ['OPENAI_API_KEY'] = api_key # Set the OpenAI API key as an environment variable llm = ChatOpenAI() ''' # Identify the email's language template1 = "Return the language this email is written in:\n{email}.\nONLY return the language it was written in." prompt1 = ChatPromptTemplate.from_template(template1) chain_1 = LLMChain(llm=llm, prompt=prompt1, output_key="language") # Translate the email to English template2 = "Translate this email from {language} to English. Here is the email:\n" + email prompt2 = ChatPromptTemplate.from_template(template2) chain_2 = LLMChain(llm=llm, prompt=prompt2, output_key="translated_email") # Provide a summary in English template3 = "Create a short summary of this email:\n{translated_email}" prompt3 = ChatPromptTemplate.from_template(template3) chain_3 = LLMChain(llm=llm, prompt=prompt3, output_key="summary") # Provide a reply in English template4 = "Reply to the sender of the email giving a plausible reply based on the {summary} and a promise to address issues" prompt4 = ChatPromptTemplate.from_template(template4) chain_4 = LLMChain(llm=llm, prompt=prompt4, output_key="reply") # Provide a translation back to the original language template5 = "Translate the {reply} back to the original {language} of the email." prompt5 = ChatPromptTemplate.from_template(template5) chain_5 = LLMChain(llm=llm, prompt=prompt5, output_key="translated_reply") seq_chain = SequentialChain(chains=[chain_1, chain_2, chain_3, chain_4, chain_5], input_variables=['email'], output_variables=['language', 'translated_email', 'summary', 'reply', 'translated_reply'], verbose=True) ''' return seq_chain(email) print("google-generativeai:", genai.__version__) GOOGLE_API_KEY = os.environ.get("GOOGLE_API_KEY") TITLE_INTRO = """

Introduction to Financial Manager

""" TITLE1 = """

Company Analysis

""" TITLE2 = """

Investment Strategy

""" TITLE3 = """

Profit Prophet

""" SUBTITLE = """

Strategy Agent built with Gemini Pro and Gemini Pro Vision API

""" GETKEY = """
Get an API key GOOGLE API KEY.
""" AVATAR_IMAGES = ( None, "https://media.roboflow.com/spaces/gemini-icon.png" ) movie_script_analysis = "" IMAGE_CACHE_DIRECTORY = "/tmp" IMAGE_WIDTH = 512 CHAT_HISTORY = List[Tuple[Optional[Union[Tuple[str], str]], Optional[str]]] def preprocess_stop_sequences(stop_sequences: str) -> Optional[List[str]]: if not stop_sequences: return None return [sequence.strip() for sequence in stop_sequences.split(",")] def preprocess_image(image: Image.Image) -> Optional[Image.Image]: image_height = int(image.height * IMAGE_WIDTH / image.width) return image.resize((IMAGE_WIDTH, image_height)) def cache_pil_image(image: Image.Image) -> str: image_filename = f"{uuid.uuid4()}.jpeg" os.makedirs(IMAGE_CACHE_DIRECTORY, exist_ok=True) image_path = os.path.join(IMAGE_CACHE_DIRECTORY, image_filename) image.save(image_path, "JPEG") return image_path def preprocess_chat_history( history: CHAT_HISTORY ) -> List[Dict[str, Union[str, List[str]]]]: messages = [] for user_message, model_message in history: if isinstance(user_message, tuple): pass elif user_message is not None: messages.append({'role': 'user', 'parts': [user_message]}) if model_message is not None: messages.append({'role': 'model', 'parts': [model_message]}) return messages def upload(files: Optional[List[str]], chatbot: CHAT_HISTORY) -> CHAT_HISTORY: for file in files: image = Image.open(file).convert('RGB') image = preprocess_image(image) image_path = cache_pil_image(image) chatbot.append(((image_path,), None)) return chatbot def user(text_prompt: str, chatbot: CHAT_HISTORY): if text_prompt: chatbot.append((text_prompt, None)) return "", chatbot def bot( google_key: str, files: Optional[List[str]], temperature: float, max_output_tokens: int, stop_sequences: str, top_k: int, top_p: float, chatbot: CHAT_HISTORY ): if len(chatbot) == 0: return chatbot google_key = google_key if google_key else GOOGLE_API_KEY if not google_key: raise ValueError( "GOOGLE_API_KEY is not set. " "Please follow the instructions in the README to set it up.") genai.configure(api_key=google_key) generation_config = genai.types.GenerationConfig( temperature=temperature, max_output_tokens=max_output_tokens, stop_sequences=preprocess_stop_sequences(stop_sequences=stop_sequences), top_k=top_k, top_p=top_p) if files: text_prompt = [chatbot[-1][0]] \ if chatbot[-1][0] and isinstance(chatbot[-1][0], str) \ else [] image_prompt = [Image.open(file).convert('RGB') for file in files] model = genai.GenerativeModel('gemini-pro-vision') response = model.generate_content( text_prompt + image_prompt, stream=True, generation_config=generation_config) else: messages = preprocess_chat_history(chatbot) model = genai.GenerativeModel('gemini-pro') response = model.generate_content( messages, stream=True, generation_config=generation_config) # streaming effect chatbot[-1][1] = "" for chunk in response: for i in range(0, len(chunk.text), 10): section = chunk.text[i:i + 10] chatbot[-1][1] += section time.sleep(0.01) yield chatbot google_key_component = gr.Textbox( label="GOOGLE API KEY", value="", type="password", placeholder="...", info="You have to provide your own GOOGLE_API_KEY for this app to function properly", visible=GOOGLE_API_KEY is None ) chatbot_component = gr.Chatbot( label='Gemini Pro Vision', bubble_full_width=False, avatar_images=AVATAR_IMAGES, scale=2, height=400 ) text_prompt_component = gr.Textbox(value=movie_script_analysis, show_label=False, autofocus=True, scale=8, lines=8 ) upload_button_component = gr.UploadButton( label="Upload Images", file_count="multiple", file_types=["image"], scale=1 ) run_button_component = gr.Button(value="Run", variant="primary", scale=1) run_button_analysis = gr.Button(value="Run", variant="primary", scale=1) temperature_component = gr.Slider( minimum=0, maximum=1.0, value=0.4, step=0.05, label="Temperature", info=( "Temperature controls the degree of randomness in token selection. Lower " "temperatures are good for prompts that expect a true or correct response, " "while higher temperatures can lead to more diverse or unexpected results. " )) max_output_tokens_component = gr.Slider( minimum=1, maximum=2048, value=1024, step=1, label="Token limit", info=( "Token limit determines the maximum amount of text output from one prompt. A " "token is approximately four characters. The default value is 2048." )) stop_sequences_component = gr.Textbox( label="Add stop sequence", value="", type="text", placeholder="STOP, END", info=( "A stop sequence is a series of characters (including spaces) that stops " "response generation if the model encounters it. The sequence is not included " "as part of the response. You can add up to five stop sequences." )) top_k_component = gr.Slider( minimum=1, maximum=40, value=32, step=1, label="Top-K", info=( "Top-k changes how the model selects tokens for output. A top-k of 1 means the " "selected token is the most probable among all tokens in the model’s " "vocabulary (also called greedy decoding), while a top-k of 3 means that the " "next token is selected from among the 3 most probable tokens (using " "temperature)." )) top_p_component = gr.Slider( minimum=0, maximum=1, value=1, step=0.01, label="Top-P", info=( "Top-p changes how the model selects tokens for output. Tokens are selected " "from most probable to least until the sum of their probabilities equals the " "top-p value. For example, if tokens A, B, and C have a probability of .3, .2, " "and .1 and the top-p value is .5, then the model will select either A or B as " "the next token (using temperature). " )) user_inputs = [ text_prompt_component, chatbot_component ] bot_inputs = [ google_key_component, upload_button_component, temperature_component, max_output_tokens_component, stop_sequences_component, top_k_component, top_p_component, chatbot_component ] # Gmix ++++++++++++++++++++++++++++++++++++++++++++++++ #Crew imports from crewai import Agent, Task, Crew, Process # Set gemini_llm gemini_llm = GoogleGenerativeAI(model="gemini-pro", google_api_key=GOOGLE_AI_STUDIO) def crewai_process(research_topic): # Define your agents with roles and goals researcher = Agent( role='Senior Research Analyst', goal=f'Uncover cutting-edge developments in {research_topic}', backstory="""You are a Senior Research Analyst at a leading think tank. Your expertise lies in identifying emerging trends. You have a knack for dissecting complex data and presenting actionable insights.""", verbose=True, allow_delegation=False, llm = gemini_llm, tools=[ GeminiSearchTools.gemini_search ] ) writer = Agent( role='Tech Content Strategist', goal='Craft compelling content on tech advancements', backstory="""You are a renowned Tech Social Media Content Writer and Strategist, known for your insightful and engaging articles on technology and innovation. With a deep understanding of the tech industry and how people are impacted by it, you transform complex concepts into compelling narratives.""", verbose=True, allow_delegation=True, llm = gemini_llm # Add tools and other optional parameters as needed ) # Create tasks for your agents task1 = Task( description=f"""Conduct a comprehensive analysis of the latest advancements in {research_topic}. Compile your findings in a detailed report. Your final answer MUST be a full analysis report""", agent=researcher ) task2 = Task( description="""Using the insights from the researcher's report, develop an engaging blog post that highlights the most significant advancements. Your post should be informative yet accessible, catering to a tech-savvy audience. Aim for a narrative that captures the essence of these breakthroughs and their implications for the future. Your final answer MUST be the full blog post of at least 3 paragraphs.""", agent=writer ) # Instantiate your crew with a sequential process crew = Crew( agents=[researcher, writer], tasks=[task1, task2], verbose=2, process=Process.sequential ) # Get your crew to work! result = crew.kickoff() return result # Portfolio Analysis +++++++++++++++++++++++++++++++++++ def plot_cum_returns(data, title): daily_cum_returns = 1 + data.dropna().pct_change() daily_cum_returns = daily_cum_returns.cumprod()*100 fig = px.line(daily_cum_returns, title=title) return fig def plot_efficient_frontier_and_max_sharpe(mu, S): # Optimize portfolio for max Sharpe ratio and plot it out with efficient frontier curve ef = EfficientFrontier(mu, S) fig, ax = plt.subplots(figsize=(6,4)) ef_max_sharpe = copy.deepcopy(ef) plotting.plot_efficient_frontier(ef, ax=ax, show_assets=False) # Find the max sharpe portfolio ef_max_sharpe.max_sharpe(risk_free_rate=0.02) ret_tangent, std_tangent, _ = ef_max_sharpe.portfolio_performance() ax.scatter(std_tangent, ret_tangent, marker="*", s=100, c="r", label="Max Sharpe") # Generate random portfolios with random weights n_samples = 1000 w = np.random.dirichlet(np.ones(ef.n_assets), n_samples) rets = w.dot(ef.expected_returns) stds = np.sqrt(np.diag(w @ ef.cov_matrix @ w.T)) sharpes = rets / stds ax.scatter(stds, rets, marker=".", c=sharpes, cmap="viridis_r") # Output ax.legend() return fig def output_results(start_date, end_date, tickers_string): tickers = tickers_string.split(',') # Get Stock Prices stocks_df = yf.download(tickers, start=start_date, end=end_date)['Adj Close'] # Plot Individual Stock Prices fig_indiv_prices = px.line(stocks_df, title='Price of Individual Stocks') # Plot Individual Cumulative Returns fig_cum_returns = plot_cum_returns(stocks_df, 'Cumulative Returns of Individual Stocks Starting with $100') # Calculatge and Plot Correlation Matrix between Stocks corr_df = stocks_df.corr().round(2) fig_corr = px.imshow(corr_df, text_auto=True, title = 'Correlation between Stocks') # Calculate expected returns and sample covariance matrix for portfolio optimization later mu = expected_returns.mean_historical_return(stocks_df) S = risk_models.sample_cov(stocks_df) # Plot efficient frontier curve fig_efficient_frontier = plot_efficient_frontier_and_max_sharpe(mu, S) # Get optimized weights ef = EfficientFrontier(mu, S) ef.max_sharpe(risk_free_rate=0.04) weights = ef.clean_weights() expected_annual_return, annual_volatility, sharpe_ratio = ef.portfolio_performance() expected_annual_return, annual_volatility, sharpe_ratio = '{}%'.format((expected_annual_return*100).round(2)), \ '{}%'.format((annual_volatility*100).round(2)), \ '{}%'.format((sharpe_ratio*100).round(2)) weights_df = pd.DataFrame.from_dict(weights, orient = 'index') weights_df = weights_df.reset_index() weights_df.columns = ['Tickers', 'Weights'] # Calculate returns of portfolio with optimized weights stocks_df['Optimized Portfolio'] = 0 for ticker, weight in weights.items(): stocks_df['Optimized Portfolio'] += stocks_df[ticker]*weight # Plot Cumulative Returns of Optimized Portfolio fig_cum_returns_optimized = plot_cum_returns(stocks_df['Optimized Portfolio'], 'Cumulative Returns of Optimized Portfolio Starting with $100') return fig_cum_returns_optimized, weights_df, fig_efficient_frontier, fig_corr, \ expected_annual_return, annual_volatility, sharpe_ratio, fig_indiv_prices, fig_cum_returns # Interface ============================================= with gr.Tab("Introduction"): gr.HTML(TITLE_INTRO) with gr.Tab("Your Portfolio"): gr.HTML(TITLE_INTRO) fn=crewai_process, inputs=gr.Textbox(lines=2, placeholder="Enter Research Topic Here..."), outputs="text", title="CrewAI on Gemini (Blog Post Writer)", description="Input a research topic to get a comprehensive analysis (in logs) and a blog post draft (in output). To lea with gr.Blocks() as demo: with gr.Tab("Portfolio Analysis"): with gr.Blocks() as app: with gr.Row(): gr.HTML("

Bohmian's Stock Portfolio Optimizer

") with gr.Row(): start_date = gr.Textbox("2013-01-01", label="Start Date") end_date = gr.Textbox(datetime.datetime.now().date(), label="End Date") with gr.Row(): tickers_string = gr.Textbox("MA,META,V,AMZN,JPM,BA", label='Enter all stock tickers to be included in portfolio separated \ by commas WITHOUT spaces, e.g. "MA,META,V,AMZN,JPM,BA"') btn = gr.Button("Get Optimized Portfolio") with gr.Row(): gr.HTML("

Optimizied Portfolio Metrics

") with gr.Row(): expected_annual_return = gr.Text(label="Expected Annual Return") annual_volatility = gr.Text(label="Annual Volatility") sharpe_ratio = gr.Text(label="Sharpe Ratio") with gr.Row(): fig_cum_returns_optimized = gr.Plot(label="Cumulative Returns of Optimized Portfolio (Starting Price of $100)") weights_df = gr.DataFrame(label="Optimized Weights of Each Ticker") with gr.Row(): fig_efficient_frontier = gr.Plot(label="Efficient Frontier") fig_corr = gr.Plot(label="Correlation between Stocks") with gr.Row(): fig_indiv_prices = gr.Plot(label="Price of Individual Stocks") fig_cum_returns = gr.Plot(label="Cumulative Returns of Individual Stocks Starting with $100") btn.click(fn=output_results, inputs=[start_date, end_date, tickers_string], outputs=[fig_cum_returns_optimized, weights_df, fig_efficient_frontier, fig_corr, \ expected_annual_return, annual_volatility, sharpe_ratio, fig_indiv_prices, fig_cum_returns]) with gr.Tab("Company Analysis"): gr.HTML(TITLE1) run_button_analysis.click( fn=company_analysis, inputs=[ gr.Textbox(label="Enter your OpenAI API Key:", type="password"), gr.Textbox(label="Enter the Company Name:") ], outputs=[ gr.Textbox(label="Language"), gr.Textbox(label="Summary"), gr.Textbox(label="Translated Email"), gr.Textbox(label="Reply in English"), gr.Textbox(label="Reply in Original Language") ] ) with gr.Tab("Investment Strategy"): gr.HTML(TITLE2) gr.HTML(SUBTITLE) gr.HTML(GETKEY) with gr.Column(): google_key_component.render() chatbot_component.render() with gr.Row(): text_prompt_component.render() upload_button_component.render() run_button_component.render() with gr.Accordion("Parameters", open=False): temperature_component.render() max_output_tokens_component.render() stop_sequences_component.render() with gr.Accordion("Advanced", open=False): top_k_component.render() top_p_component.render() run_button_component.click( fn=user, inputs=user_inputs, outputs=[text_prompt_component, chatbot_component], queue=False ).then( fn=bot, inputs=bot_inputs, outputs=[chatbot_component], ) text_prompt_component.submit( fn=user, inputs=user_inputs, outputs=[text_prompt_component, chatbot_component], queue=False ).then( fn=bot, inputs=bot_inputs, outputs=[chatbot_component], ) upload_button_component.upload( fn=upload, inputs=[upload_button_component, chatbot_component], outputs=[chatbot_component], queue=False ) with gr.Tab("Profit Prophet"): gr.HTML(TITLE3) with gr.Row(): with gr.Column(scale=1): gr.Image(value = "resources/holder.png") with gr.Column(scale=1): gr.Image(value = "resources/holder.png") demo.queue(max_size=99).launch(debug=False, show_error=True)