AI Agents in Finance Section 1 & Section 2 — the finance-first, code-first edition

pip install openai openai-agents google-genai yfinance financetoolkit chromadb llama-index transformers torch

import os
# Put keys in environment variables (or Colab secrets) — never paste them into code you share
OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]
FINANCIAL_MODELING_PREP_API_KEY = os.environ["FINANCIAL_MODELING_PREP_API_KEY"]

import asyncio
from agents import Agent, Runner

async def main():
    result = await Runner.run(my_agent, "Hello")   # 'await' is allowed inside an 'async def'
    print(result.final_output)

asyncio.run(main())                                 # starts the async world and runs main()

from openai import OpenAI
import json, os

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

# WHO the model is + WHAT SHAPE the answer must have
system_prompt = """You are a finance analyst. Produce a concise, decision-useful summary of an
earnings call for portfolio managers. Return strict JSON with keys:
company, quarter, revenue, eps, guidance, highlights[], risks[], outlook."""

# The actual job + the data. The f"..." drops the transcript text into the {...} slot.
user_prompt = f"""Summarize the following earnings call transcript:
{transcript_apple_earnings_call}"""

response = client.responses.create(
    model="gpt-4.1-mini",                                    # cheap + fast: fine for summarising
    input=[{"role": "system", "content": system_prompt},
           {"role": "user",   "content": user_prompt}],
    text={"format": {"type": "json_object"}},                # JSON mode: the reply must be valid JSON
)

raw_text = response.output[0].content[0].text                # dig the text out of the response box
summary  = json.loads(raw_text)                              # JSON text  →  Python dictionary
print(json.dumps(summary, indent=2))                         # dictionary →  pretty, indented text

from google import genai

gclient = genai.Client(api_key=os.environ["GOOGLE_API_KEY"])   # separate name, so we don't overwrite 'client'
resp_genai = gclient.models.generate_content(
    model="gemini-2.5-flash",
    contents=system_prompt + "\n\n" + user_prompt,             # here: both prompts glued into one text
    config={"response_mime_type": "application/json"},         # Gemini's version of JSON mode
)
print(resp_genai.text)

import torch, transformers

pipe = transformers.pipeline(
    "text-generation",
    model="meta-llama/Llama-3.1-8B-Instruct",       # downloaded to YOUR machine — nothing leaves it
    model_kwargs={"torch_dtype": torch.float16},     # half-precision numbers → fits on a 16 GB GPU
    device_map="auto",                               # place it on the GPU automatically
)
messages = [{"role": "system", "content": system_prompt},
            {"role": "user",   "content": user_prompt}]
outputs = pipe(messages, max_new_tokens=1024)
print(outputs[0]["generated_text"][-1]["content"])  # see the note below

system_prompt = """You are a finance analyst writing for portfolio managers.
Return strict JSON with exactly these keys and types:
  company: string
  quarter: string, like "FY2025 Q3"
  revenue_usd: number in US dollars, no symbols or words (98700000000, not "$98.7B")
  revenue_growth_yoy_pct: number (6 means +6%)
  eps_usd: number
  guidance: string
  highlights: list of strings
  risks: list of strings
  outlook: one of "positive", "neutral", "negative"
"""

{"messages": [{"role": "system", "content": "You are a senior financial analyst."}, {"role": "user", "content": "Quarterly operating profit rose to EUR 13.1 mn from EUR 8.7 mn a year earlier."}, {"role": "assistant", "content": "positive"}]}

response = client.responses.create(
    model="gpt-4.1",
    input="Give me the latest news about NVIDIA.",
)
print(response.output_text)

response = client.responses.create(model="gpt-4.1", input="What is the last price of NVIDIA.")
print(response.output_text)

from agents import Agent, Runner, function_tool
import yfinance as yf

@function_tool                                    # turns this function into a tool the LLM can call
def get_latest_stock_price(symbol: str) -> dict:
    """Fetch the current stock price for the given symbol."""   # the LLM reads this to decide when to use it
    ticker = yf.Ticker(symbol)                    # a handle for one stock, e.g. "NVDA"
    hist = ticker.history(period="1d")            # a table of the last trading day
    latest_close = hist["Close"].iloc[-1]         # column "Close", last row  → latest price
    latest_date = hist.index[-1]                  # label of that last row    → its date
    return {"price": float(latest_close), "date": str(latest_date.date())}

stock_price_agent = Agent(
    name="Stock Price Agent",                                          # a label, shows up in logs
    instructions="Fetch Historical prices for a given stock symbol",   # its job description
    model="gpt-4.1",                                                   # the brain
    tools=[get_latest_stock_price],                                    # what it may call
)

output = Runner.run_sync(                                              # run_sync: no 'await' needed
    starting_agent=stock_price_agent,
    input="What is today's date, and what is the current stock price of Nvidia?",
)
print(output.final_output)

import os
import pandas as pd
from financetoolkit import Toolkit
from agents import Agent, Runner, function_tool

OPENAI_API_KEY = os.environ["OPENAI_API_KEY"]
FINANCIAL_MODELING_PREP_API_KEY = os.environ["FINANCIAL_MODELING_PREP_API_KEY"]

@function_tool
def get_company_fundamentals(ticker: str) -> dict[str, pd.DataFrame]:
    """Get profitability, valuation and liquidity ratios for a given ticker."""
    companies = Toolkit(
        [ticker], api_key=FINANCIAL_MODELING_PREP_API_KEY, start_date="2022-01-01"
    )
    ratios = companies.ratios.collect_all_ratios()     # ONE big table: rows = ratio names, columns = years

    # Our three "shopping lists"
    profitability_ratios_names = ["Return on Equity", "Return on Assets", "Gross Margin",
                                  "Operating Margin", "Net Profit Margin"]
    valuation_ratios_names = ["Price-to-Earnings", "Price-to-Book", "Price-to-Cash-Flow",
                              "Price-to-Free-Cash-Flow", "EV-to-EBITDA", "EV-to-Sales"]
    liquidity_ratios_names = ["Current Ratio", "Quick Ratio", "Cash Ratio", "Working Capital",
                              "Operating Cash Flow Ratio", "Operating Cash Flow to Sales Ratio",
                              "Short Term Coverage Ratio"]

    return {
        "profitability_ratios": ratios.loc[ratios.index.intersection(profitability_ratios_names)],
        "valuation_ratios":     ratios.loc[ratios.index.intersection(valuation_ratios_names)],
        "liquidity_ratios":     ratios.loc[ratios.index.intersection(liquidity_ratios_names)],
    }

fundamental_retrieval_agent = Agent(
    name="Fundamental Retrieval Agent",
    model="gpt-4o-mini",
    instructions="""
You are a Fundamental Retrieval Agent.
Tasks:
1) Call the fundamentals tool for the specified ticker.
2) Output the desired set of ratios.
""",
    tools=[get_company_fundamentals],
)

fundamental_results = await Runner.run(fundamental_retrieval_agent, """Proceed step by step:
1- List 3 liquidity ratios and explain their significance.
2- Provide Apple's values for the three listed liquidity ratios for the most
recent fiscal year, knowing that we are in 2026.
""")
print(fundamental_results.final_output)

from agents import Agent, Runner, SQLiteSession

portfolio_agent = Agent(
    name="Portfolio Rebalancing Agent",
    model="gpt-4o-mini",
    instructions="""You are a Portfolio Rebalancing Agent for a professional investor.
Your goal is to propose clear, explainable rebalancing actions that keep the
portfolio aligned with the target risk/return profile and constraints.
Your role
- Analyze the current portfolio (weights, asset classes, sectors, factors, cash).
- Incorporate market regime information (volatility, trends, macro environment).
- Compare current allocation to:
  - Target allocation or risk budget (if provided),
  - Or to a reasonably diversified allocation based on the information you have.
- Propose specific rebalancing trades with approximate percentage changes and short rationales.
- Prioritize diversification, risk control, and alignment with the investor's objectives.
- Keep answers concise and structured, and state assumptions when information is missing.
- Treat the allocation provided in the user's prompt as the portfolio's current allocation.
Always explain your reasoning.""",
)

session_ptf_rebal = SQLiteSession("conversation_ptf")    # a fresh notepad named "conversation_ptf"

# Same, but saved in a file so it survives a restart:
# session_ptf_rebal = SQLiteSession("conversation_ptf", "portfolio_sessions.db")

input_user_1 = """Current Market Situation:
- Volatility Regime: high
- Market Trend: declining
Current Portfolio State:
- Portfolio: {"stocks": 0.70, "bonds": 0.20, "cash": 0.10}
What rebalancing action should I take?"""

output_1 = await Runner.run(
    starting_agent=portfolio_agent,
    input=input_user_1,
    session=session_ptf_rebal,      # read what's on the notepad, then write this turn onto it
)
print(output_1.final_output)        # suggests ~50% stocks / 40% bonds / 10% cash (defensive)

input_user_2 = "In this case, I'm reducing equity to 50%, increasing bonds to 35% and cash to 15%."
output_2 = await Runner.run(
    starting_agent=portfolio_agent,
    input=input_user_2,
    session=session_ptf_rebal,      # SAME notepad → the agent still knows the 70/20/10 starting point
)
print(output_2.final_output)

all_items = await session_ptf_rebal.get_items()    # everything on the notepad, oldest first
conversation_history = []                           # we'll fill this with (role, text) pairs

for i, msg in enumerate(all_items, 1):              # i counts 1, 2, 3 … (only for printing)
    role = msg.get("role", "unknown")               # "user" or "assistant"; tool-call items have no role
    content = msg.get("content", "")

    if role == "user":
        text = content                              # a letter: already a string
    elif role == "assistant":
        text = content[0]["text"]                   # an envelope: open it, take page 0's text
    else:
        continue                                    # skip tool calls and other item types

    print(f"{i}. {role}: {text}")
    print("--" * 50)
    conversation_history.append((role, text))       # save a clean (role, text) pair

# Original one-liner:
conversation_history_string = "\n\n".join([f'{elem[0]}: """{elem[1]}"""' for elem in conversation_history])

# Exactly the same thing, with friendlier names:
conversation_history_string = "\n\n".join(
    [f'{role}: """{text}"""' for role, text in conversation_history]
)

episodic_memory_prompt_template = """
You will read a conversation between a user and a portfolio-rebalancing agent.
Extract an episodic memory that will help the agent in future sessions.

OUTPUT FORMAT:
Return only valid JSON in this exact schema:
{{
  "conversation_summary": string,            // 3–5 sentences: market conditions, portfolio situation, actions taken
  "portfolio_allocation_trajectory": string, // how allocations changed, with the market situation each time
  "what_worked": string,                     // alignment with goals, risk management, clear decisions
  "what_didnt_work": string,                 // unclear strategy, over/under-reaction, concentration risk, data issues
  "key_points": ["string"]                   // the most important insights to keep for future sessions
}}

(… a few examples of GOOD and POOR summaries go here — few-shot …)

Here is the conversation between the user and the principal portfolio-rebalancing agent:
{conversation}
"""

user_prompt = episodic_memory_prompt_template.format(conversation=conversation_history_string)

import json
from openai import OpenAI

client = OpenAI(api_key=OPENAI_API_KEY)
response = client.responses.create(
    model="gpt-4.1-mini",
    input=[{"role": "user", "content": user_prompt}],
    text={"format": {"type": "json_object"}},              # JSON mode, as in Lab 1A
)
episodic_memory = json.loads(response.output[0].content[0].text)   # JSON text → Python dict
print(json.dumps(episodic_memory, indent=2))

# Version A: one readable text — this is what gets embedded and searched by meaning
document_text = f"""
Conversation Summary: {episodic_memory['conversation_summary']}
Portfolio Allocation Trajectory: {episodic_memory['portfolio_allocation_trajectory']}
What Worked: {episodic_memory['what_worked']}
What Didn't Work: {episodic_memory['what_didnt_work']}
Key Points: {', '.join(episodic_memory['key_points'])}
"""

# Version B: the same facts as separate fields — the labels on the folder
metadata_clean = {
    "conversation_summary": episodic_memory["conversation_summary"],
    "portfolio_allocation_trajectory": episodic_memory["portfolio_allocation_trajectory"],
    "what_worked": episodic_memory["what_worked"],
    "what_didnt_work": episodic_memory["what_didnt_work"],
    "key_points": " | ".join(episodic_memory["key_points"]),   # list → one string
}

import chromadb
from chromadb.utils import embedding_functions

# The translator: turns any text into a list of 1,536 numbers (a vector) that captures its meaning
embedding_function = embedding_functions.OpenAIEmbeddingFunction(
    api_key=OPENAI_API_KEY,
    model_name="text-embedding-3-small",
)

collection_name = "episodic_memory_ptf_rebal"                         # the drawer's name
persistent_client = chromadb.PersistentClient(path="./chroma_ptf_rebal")  # the cabinet, saved in a folder
collection = persistent_client.get_or_create_collection(               # open the drawer (create it if new)
    collection_name, embedding_function=embedding_function
)

collection.add(
    ids=["memory_1"],               # a unique name for this page
    documents=[document_text],      # the text → Chroma embeds it automatically with our translator
    metadatas=[metadata_clean],     # the folder labels
)

input_user_init = """Current Market Situation:
- Volatility Regime: low
- Market Trend: bullish
What rebalancing action should I take?"""

episodic_memory_recap = collection.query(
    query_texts=[input_user_init],   # a LIST: you could ask several questions at once
    n_results=3,                     # up to 3 closest pages per question
)

session_ptf_rebal_2 = SQLiteSession("conversation_ptf_2")      # a brand-new, EMPTY notepad

input_user_with_memory = input_user_init + f"""
Here are relevant insights from past rebalancing episodes. Use them as guidance, not strict rules:
{episodic_memory_recap['documents'][0][0]}
Argue your proposition using past rebalancing episodes if needed.
"""

output_using_memory = await Runner.run(
    starting_agent=portfolio_agent,
    input=input_user_with_memory,
    session=session_ptf_rebal_2,
)
print(output_using_memory.final_output)

import yfinance as yf

prices = yf.Ticker("AAPL").history(period="3mo")["Close"]      # step 1: ~63 daily closing prices
ma20 = prices.rolling(window=20).mean()                        # step 2: 20-day moving average
today_ma, week_ago_ma = ma20.iloc[-1], ma20.iloc[-6]           # step 3: today vs 5 trading days earlier
trend = "Uptrend" if today_ma > week_ago_ma else "Downtrend"   # step 4: label it
print(f"MA20 today={today_ma:.2f}, a week ago={week_ago_ma:.2f} → {trend}")

from llama_index.core.agent.workflow import ReActAgent, AgentStream
from llama_index.core.tools import FunctionTool
from llama_index.llms.openai import OpenAI
import yfinance as yf

def get_latest_price(ticker_symbol: str):
    """Fetch the latest price for the given ticker symbol."""
    ticker = yf.Ticker(ticker_symbol)
    data = ticker.history(period="1d", interval="1m")         # today, one row per MINUTE
    if not data.empty:                                         # did any rows come back?
        latest_row = data.iloc[-1]                             # the most recent minute
        return {
            "ticker": ticker_symbol,
            "datetime": latest_row.name.strftime("%Y-%m-%d %H:%M"),   # .name = this row's label = its timestamp
            "price": float(latest_row["Close"]),
        }
    else:
        return {"error": f"No data found for ticker: {ticker_symbol}"}  # the agent can read this and react

tools = [FunctionTool.from_defaults(get_latest_price)]       # LlamaIndex's way to wrap a function as a tool

agent_prices = ReActAgent(
    llm=OpenAI(model="gpt-4o", api_key=OPENAI_API_KEY),
    tools=tools,
    timeout=120,              # give up after 120 s instead of hanging forever
    verbose=False,
)

handler = agent_prices.run("What is the last price of Apple?")   # start the job, get a tracking handle

async for ev in handler.stream_events():       # live updates while the agent works
    if isinstance(ev, AgentStream):            # only the "text chunk" events
        print(ev.delta, end="", flush=True)    # print each new chunk on the same line, immediately

response = await handler                       # wait for the final, complete answer

from collections import Counter
from openai import OpenAI

client = OpenAI()

cot_prompt = """You are a financial analysis agent.
Classify the earnings result as: Positive surprise, Neutral surprise or Negative surprise.
Reason step by step before producing the final classification.

Input:
EPS: 1.25 vs consensus 1.20
Revenue: +4% YoY
Margins: Declining
Guidance: Raised
Management: CEO departure announced

End with exactly one line: FINAL: Positive surprise | Neutral surprise | Negative surprise"""

answers = []
for _ in range(5):                                            # 5 independent reasoning paths
    r = client.responses.create(model="gpt-4.1-mini", input=cot_prompt, temperature=1.0)
    last_line = r.output_text.strip().splitlines()[-1]        # e.g. "FINAL: Positive surprise"
    answers.append(last_line.replace("FINAL:", "").strip())

votes = Counter(answers)                                      # counts each distinct answer
print(votes)                                                  # e.g. Counter({'Positive surprise': 3, 'Neutral surprise': 2})
print("Majority:", votes.most_common(1)[0][0])

from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import Literal
from agents import Agent, Runner, WebSearchTool, ItemHelpers, TResponseInputItem

today_date   = datetime.now().strftime("%Y-%m-%d")                         # e.g. "2026-10-22"
two_days_ago = (datetime.now() - timedelta(days=2)).strftime("%Y-%m-%d")   # e.g. "2026-10-20"

INSTRUCTIONS_NEWS_SEARCH = (
    "You are a research news assistant specialized in Finance or any news that can have an impact "
    "on the financial health of a given company or a region. "
    "You search the web for a given region to collect the most important and recent news. "
    "Produce a concise summary of the results with citations. The summary must be "
    "2-3 paragraphs and less than 300 words. Capture the main points. Write succinctly. "
    f"Get only the latest news between 2 days ago {two_days_ago} and today {today_date}. "
    "Use only this resource: https://www.reuters.com/. "
    "For each item you are providing, specify the date on which the article was published. "
    "For each item you are providing, specify the complete external link to the article. "
    "This is important. Otherwise, don't give me a summary, if you don't provide a date and an external link. "
    "If you don't give a source link, it means that you are hallucinating. "
)

web_news_searcher = Agent(
    name="web_news_searcher",
    instructions=INSTRUCTIONS_NEWS_SEARCH,
    tools=[WebSearchTool()],          # OpenAI's hosted web search — no code of your own needed
)

@dataclass
class EvaluationFeedback:
    feedback: str                                    # free text: why it passed or failed
    score: Literal["successful", "unsuccessful"]     # exactly one of these two words — nothing else

news_evaluator = Agent(
    name="news_evaluator",
    instructions=(
        "You will be given a financial news summary in a given region (Euro, US, Asia) and you are "
        "asked to evaluate the completeness of external source links. "
        "Check whether the news summary is supported by external article links from https://www.reuters.com/. "
        f"Check whether the news summary was published between 2 days ago {two_days_ago} and today {today_date}. "
        "For each item of news, the date of the news must be explicitly provided. "
        "Check whether the news summary corresponds to the region or regions requested by the user. "
        "If the user requested 'n' news items, each item must be supported by an external source link, "
        "so you need to ensure that there are 'n' source links. "
        "External links are required for the user to verify the accuracy of the news. "
    ),
    output_type=EvaluationFeedback,                  # forces the answer into the form above
)

msg = input("User's request:")    # e.g. "Provide the latest 5 news in the Eurozone and specify the source for each one."
input_items: list[TResponseInputItem] = [{"content": msg, "role": "user"}]   # the searcher's running chat
latest_outline: str | None = None   # will hold the newest news summary
max_iteration = 4                   # retry limit
nbr_iterations = 0                  # retries so far

while True:
    # 1) The searcher writes (or rewrites) the summary, seeing the whole chat so far
    news_searcher_result = await Runner.run(web_news_searcher, input_items)
    input_items = news_searcher_result.to_input_list()                       # old chat + the new reply
    latest_outline = ItemHelpers.text_message_outputs(news_searcher_result.new_items)  # just the new text

    # 2) The evaluator grades ONLY the newest summary
    news_evaluator_result = await Runner.run(news_evaluator, latest_outline)
    result: EvaluationFeedback = news_evaluator_result.final_output         # the filled-in form

    # 3) Passed? Stop.
    if result.score == "successful":
        print("Links check is good ==> stopping the iteration.")
        break

    # 4) Failed: add the summary and the judge's feedback to the chat, then go round again
    input_items.append({"content": f"News Summary: {latest_outline}", "role": "system"})
    input_items.append({"content": f"Feedback: {result.feedback}\n for each item in the news summary, "
                                   "provide an external source link:", "role": "user"})
    nbr_iterations += 1
    if nbr_iterations > max_iteration:
        print("Reached max_iteration ==> stopping the iteration.")
        break

print(latest_outline)

@function_tool
def get_company_fundamentals(ticker: str) -> dict[str, pd.DataFrame]:
    """Get the profitability and valuation fundamental ratios for a given ticker."""
    companies = Toolkit([ticker], api_key=FINANCIAL_MODELING_PREP_API_KEY, start_date="2022-01-01")
    ratios = companies.ratios.collect_all_ratios()
    profitability_ratios_names = ["Return on Equity", "Return on Assets", "Gross Margin",
                                  "Operating Margin", "Net Profit Margin"]
    valuation_ratios_names = ["Price-to-Earnings", "Price-to-Book", "Price-to-Cash-Flow",
                              "Price-to-Free-Cash-Flow", "EV-to-EBITDA", "EV-to-Sales"]
    return {
        "profitability_ratios": ratios.loc[ratios.index.intersection(profitability_ratios_names)],
        "valuation_ratios":     ratios.loc[ratios.index.intersection(valuation_ratios_names)],
    }

MODEL = "gpt-4o-mini"

fundamental_retrieval_agent = Agent(
    name="Fundamental Retrieval Agent", model=MODEL, tools=[get_company_fundamentals],
    instructions="""
You are the Fundamental Retrieval Agent.
Tasks:
1) Call the fundamentals tool for the specified ticker.
2) Output ONLY valid JSON with this schema. Keep the output strictly factual: no interpretation.
{ "ticker": "...", "fundamentals": { ... } }   // raw numbers from tools
""",
)

fundamental_analysis_agent = Agent(
    name="Fundamental Analysis Agent", model=MODEL,
    instructions="""
You are the Fundamental Analysis Agent.
Input: JSON from the Retrieval Agent.
Tasks:
1) Interpret the fundamentals along 2 dimensions:
   - Profitability (profitability, margins, ROE)
   - Valuation (P/E, EV/Sales)
2) Identify 2-3 key risks (e.g., valuation risk, margin sustainability).
3) Output ONLY valid JSON with this schema:
{
  "ticker": "...",
  "fundamentals": { ... },          // keep from input
  "analysis": {
    "Profitability": { "rating": "strong|neutral|weak", "evidence": [string, ...] },
    "valuation": { "rating": "cheap|fair|expensive", "evidence": [string, ...] },
    "risks": [string, ...]
  }
}
Rules:
- Keep evidence short and specific.
""",
)

decision_agent = Agent(
    name="Decision Agent", model=MODEL,
    instructions="""
You are the Decision Agent.
Input: JSON from the Fundamental Analysis Agent.
Tasks: Based only on the input coming from the Fundamental Analysis Agent:
1) Produce a simple stance: BUY, HOLD, or SELL.
2) Provide 3 bullet reasons grounded in the analysis.
3) Provide a "watchlist" section: what metrics would change your stance.
Output ONLY valid JSON:
{ "ticker": "...", "stance": "BUY|HOLD|SELL",
  "summary": { "reasons": [string, string, string], "watchlist": [string, string, string] } }
""",
)

fundamental_results  = await Runner.run(fundamental_retrieval_agent, "Provide Apple fundamentals")
fundamental_analysis = await Runner.run(fundamental_analysis_agent, fundamental_results.final_output)
final_decision       = await Runner.run(decision_agent, fundamental_analysis.final_output)
print(final_decision.final_output)

import json

decision = json.loads(final_decision.final_output)            # crashes here = the agent broke the JSON contract
assert decision["stance"] in {"BUY", "HOLD", "SELL"}, "Unexpected stance!"
assert len(decision["summary"]["watchlist"]) == 3, "Watchlist must have 3 items"