🕵🏻 Building an AI-Powered Stock Analyser using OpenAI Agents SDK and Dappier

import os
import getpass
import nest_asyncio
import asyncio
from agents import Agent, FunctionTool, Runner, function_tool
from colorama import Fore, Style
from dappier import Dappier
from openai.types.responses import ResponseTextDeltaEvent

os.environ["DAPPIER_API_KEY"] = getpass.getpass("Enter your Dappier API Key: ")
os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API Key: ")

dappier_client = Dappier(api_key=os.environ["DAPPIER_API_KEY"])

from agents import set_tracing_export_api_key

set_tracing_export_api_key(os.environ["OPENAI_API_KEY"])

@function_tool
def dappier_real_time_search(query: str, ai_model_id: str) -> str:
    """Fetches real-time search results from Dappier based on the user's query.

    Args:
        query (str): The user's query for real-time data.
        ai_model_id (str): The AI model ID to use for the query, dynamically set by the agent.
            Possible values:
            - "am_01j0rzq4tvfscrgzwac7jv1p4c" (General real-time web search, including news, weather, and events.)
            - "am_01j749h8pbf7ns8r1bq9s2evrh" (Stock market data with real-time financial news and stock prices.)

    Returns:
        str: A response message containing the real-time data results.
    """
    print(Fore.RED + f"CALLING TOOL - dappier_real_time_search: {ai_model_id}\n" + Style.RESET_ALL)
    print(Fore.GREEN + f"Query: {query}\n" + Style.RESET_ALL)
    response = dappier_client.search_real_time_data(query=query, ai_model_id=ai_model_id)
    return response.message if response else "No real-time data found."

@function_tool
def dappier_ai_recommendations(query: str, data_model_id: str) -> str:
    """Fetches AI-powered content recommendations from Dappier based on the user's query.

    Args:
        query (str): The user's query for recommendations.
        data_model_id (str): The Data Model ID to use for the query, dynamically set by the agent.
            Possible values:
            - "dm_01j0pb465keqmatq9k83dthx34" (Sports news)
            - "dm_01j0q82s4bfjmsqkhs3ywm3x6y" (Lifestyle news)
            - "dm_01j1sz8t3qe6v9g8ad102kvmqn" (Dog care advice from iHeartDogs)
            - "dm_01j1sza0h7ekhaecys2p3y0vmj" (Cat care advice from iHeartCats)
            - "dm_01j5xy9w5sf49bm6b1prm80m27" (Eco-friendly content from GreenMonster)
            - "dm_01jagy9nqaeer9hxx8z1sk1jx6" (General news from WISH-TV)
            - "dm_01jhtt138wf1b9j8jwswye99y5" (Local news from 9 and 10 News)

    Returns:
        str: A formatted response containing AI-powered recommendations.
    """
    print(Fore.RED + f"CALLING TOOL: dappier_ai_recommendations: {data_model_id}\n" + Style.RESET_ALL)
    print(Fore.GREEN + f"Query: {query}\n" + Style.RESET_ALL)
    response = dappier_client.get_ai_recommendations(query=query, data_model_id=data_model_id, similarity_top_k=5)
    results = response.response.results
    formatted_text = ""
    for result in results:
        formatted_text += (f"Title: {result.title}\n"
                           f"Author: {result.author}\n"
                           f"Published on: {result.pubdate}\n"
                           f"URL: {result.source_url}\n"
                           f"Image URL: {result.image_url}\n"
                           f"Summary: {result.summary}\n\n")
    return formatted_text or "No recommendations found."

agent = Agent(
    name="Dappier Assistant",
    instructions="""
    You analyze the user's query and determine whether to use real-time search or AI recommendations.
    If the query involves stocks, finance, or current events, use `dappier_real_time_search`.
    If the query is about recommendations (e.g., news, lifestyle, sports, pets), use `dappier_ai_recommendations`.
    You MUST provide `ai_model_id` or `data_model_id` as necessary.
    Format responses in structured Markdown.
    """,
    tools=[dappier_real_time_search, dappier_ai_recommendations],
)

def generate_stock_agent_prompt(stock_name: str) -> str:
    return f"""Generate a detailed stock market analysis and trading strategy for {stock_name}, using real-time financial data. Follow these steps:

### **1. Fetch Financial News:**
Use **Dappier's real-time search** to retrieve the latest financial news related to {stock_name}. Identify key insights such as earnings reports, company announcements, regulatory changes, and macroeconomic factors affecting the stock.

### **2. Fetch Market Trends:**
Analyze **real-time market trends** for {stock_name} using Dappier's data. Extract key performance indicators such as:
   - Recent price movements (intraday, weekly, monthly)
   - Trading volume fluctuations
   - Support and resistance levels
   - Institutional investor activity

### **3. Fetch Breaking News & AI Recommendations:**
Use **Dappier AI Recommendations API** to detect any **breaking news** or high-impact reports that could influence stock price movement, such as:
   - Mergers & acquisitions
   - SEC filings and insider trading reports
   - Macroeconomic events affecting the industry

### **4. Find Similar Stocks with Strong Performance:**
Identify **similar stocks** that have historically performed well under similar conditions. Compare:
   - Price correlation with {stock_name}
   - Industry trends and sector-wide movements
   - Alternative investment opportunities based on risk-reward assessment

### **5. Generate a Trading Strategy:**
Based on all the gathered insights, generate a detailed **trading strategy** that includes:
   - **Entry Points:** Optimal buy price based on historical data & support levels
   - **Exit Points:** Ideal selling price based on resistance levels & profit margins
   - **Risk Management:** Suggested stop-loss levels & risk-reward ratio
   - **Investment Timeframe:** Short-term (day/swing trading) vs. long-term (investment)
   - **News-Driven Adjustments:** How to react to sudden news or volatility

### **6. Output:**
Present the findings in a structured format:
   - **Latest financial news for {stock_name}**
   - **Market trends & technical indicators**
   - **Breaking news alerts from Dappier AI**
   - **Comparison with similar high-performing stocks**
   - **A complete trading strategy tailored to {stock_name}**

Ensure that the recommendations are **data-driven**, clear, and actionable for the user.
    """

async def main():
    """Asks user for stock details dynamically and executes AI agent"""

    # Ask for stock details dynamically
    stock_name = input("Enter the stock that you are intersted to invest in: ")

    # Generate task prompt
    task_prompt = generate_stock_agent_prompt(stock_name)

    print(Fore.BLUE + f"\nExecuting AI Agent with prompt:\n{task_prompt}" + Style.RESET_ALL)

    # Execute AI agent with streaming results
    result = Runner.run_streamed(agent, task_prompt)

    print("\n\n=== Streaming Start ===\n\n")

    async for event in result.stream_events():
        if event.type == "raw_response_event" and isinstance(event.data, ResponseTextDeltaEvent):
            print(event.data.delta, end="", flush=True)

    print("\n\n=== Streaming Complete ===")

nest_asyncio.apply()
asyncio.run(main())