AI Agent for Market Research: Competitive Analysis and Trend Detection

Why Automate Market Research

Manual market research involves scanning news sites, reading analyst reports, tracking competitor product launches, and monitoring social sentiment — then synthesizing everything into a coherent strategic narrative. This process takes a skilled analyst one to two weeks and is outdated by the time the report is finished.

An AI market research agent continuously pulls data from multiple sources, extracts structured insights, detects emerging trends, and produces reports on demand. The analyst shifts from data gathering to strategic interpretation.

Data Source Tools

The agent needs tools to gather information from different channels. Here is a news search tool and a company profile fetcher:

import httpx
from datetime import datetime, timedelta
from agents import Agent, Runner, function_tool

@function_tool
async def search_news(query: str, days_back: int = 30) -> str:
    """Search recent news articles about a company or industry topic."""
    # Using a news API (NewsAPI, Bing News, or similar)
    since = (datetime.now() - timedelta(days=days_back)).strftime("%Y-%m-%d")
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            "https://newsapi.org/v2/everything",
            params={
                "q": query,
                "from": since,
                "sortBy": "relevancy",
                "pageSize": 10,
                "apiKey": "YOUR_API_KEY",
            },
        )
        resp.raise_for_status()

    articles = resp.json().get("articles", [])
    results = []
    for a in articles:
        results.append(
            f"Title: {a['title']}\n"
            f"Source: {a['source']['name']}\n"
            f"Date: {a['publishedAt'][:10]}\n"
            f"Summary: {(a.get('description') or 'N/A')[:300]}"
        )
    return "\n---\n".join(results) if results else "No news articles found."

@function_tool
async def get_company_info(company_name: str) -> str:
    """Fetch basic company information from a public data source."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            "https://api.crunchbase.com/api/v4/autocompletes",
            params={"query": company_name, "user_key": "YOUR_KEY"},
        )
        if resp.status_code != 200:
            return f"Could not fetch info for {company_name}."
        entities = resp.json().get("entities", [])
        if not entities:
            return "No matching companies found."
        e = entities[0]["properties"]
        return (
            f"Name: {e.get('name')}\n"
            f"Description: {e.get('short_description', 'N/A')}\n"
            f"Founded: {e.get('founded_on', 'N/A')}\n"
            f"Location: {e.get('location_identifiers', [{}])[0].get('value', 'N/A')}"
        )

Competitive Analysis Framework

The core of market research is structured comparison. This tool organizes competitor data into a framework:

_competitor_data: list[dict] = []

@function_tool
def add_competitor(
    name: str,
    strengths: str,
    weaknesses: str,
    market_position: str,
    recent_moves: str,
) -> str:
    """Add a competitor profile to the analysis framework."""
    _competitor_data.append({
        "name": name,
        "strengths": strengths,
        "weaknesses": weaknesses,
        "market_position": market_position,
        "recent_moves": recent_moves,
    })
    return f"Added {name}. Total competitors tracked: {len(_competitor_data)}"

@function_tool
def get_competitive_matrix() -> str:
    """Return a formatted competitive analysis matrix."""
    if not _competitor_data:
        return "No competitors added yet."
    lines = []
    for c in _competitor_data:
        lines.append(
            f"## {c['name']}\n"
            f"**Position:** {c['market_position']}\n"
            f"**Strengths:** {c['strengths']}\n"
            f"**Weaknesses:** {c['weaknesses']}\n"
            f"**Recent moves:** {c['recent_moves']}"
        )
    return "\n\n".join(lines)

Trend Detection Tool

Trend detection looks for patterns across the collected news and data. This tool uses keyword frequency analysis to surface emerging themes:

from collections import Counter
import re

_collected_texts: list[str] = []

@function_tool
def record_text_for_trends(text: str) -> str:
    """Store a text snippet for later trend analysis."""
    _collected_texts.append(text)
    return f"Recorded. Total snippets: {len(_collected_texts)}"

@function_tool
def detect_trends(min_frequency: int = 3) -> str:
    """Analyze all recorded texts to find frequently mentioned topics."""
    if not _collected_texts:
        return "No texts recorded yet."

    all_text = " ".join(_collected_texts).lower()
    # Extract bigrams for more meaningful trends
    words = re.findall(r"[a-z]+", all_text)
    bigrams = [f"{words[i]} {words[i+1]}" for i in range(len(words) - 1)]
    counter = Counter(bigrams)

    trending = [
        (phrase, count)
        for phrase, count in counter.most_common(20)
        if count >= min_frequency
    ]
    if not trending:
        return "No strong trends detected. Try collecting more data."

    lines = ["Detected trends (phrase: frequency):"]
    for phrase, count in trending:
        lines.append(f"  - '{phrase}': mentioned {count} times")
    return "\n".join(lines)

Assembling the Market Research Agent

market_researcher = Agent(
    name="Market Research Agent",
    instructions="""You are a market research analyst agent. When given a company or industry:
1. Use search_news to gather recent news about the target and its competitors.
2. Use get_company_info to pull structured data on each company.
3. Record all relevant text with record_text_for_trends.
4. For each competitor, call add_competitor with a SWOT-style profile.
5. Call detect_trends to identify emerging patterns.
6. Call get_competitive_matrix to compile the analysis.
7. Produce a final report with: Executive Summary, Competitive Landscape,
   Emerging Trends, Strategic Recommendations.""",
    tools=[
        search_news, get_company_info, add_competitor,
        get_competitive_matrix, record_text_for_trends, detect_trends,
    ],
)

Generating the Report

import asyncio

async def main():
    result = await Runner.run(
        market_researcher,
        "Conduct a competitive analysis of the AI code assistant market. "
        "Compare GitHub Copilot, Cursor, and Codeium. Identify trends "
        "and recommend positioning strategies for a new entrant.",
    )
    print(result.final_output)

asyncio.run(main())

The agent searches news for each competitor, builds company profiles, populates the competitive matrix, runs trend detection across all collected articles, and outputs a structured report with an executive summary, individual competitor assessments, trend analysis, and strategic recommendations.

FAQ

How do I keep the research data fresh?

Schedule the agent to run on a cron job — daily or weekly — and compare new results against previous reports. Store each run's output in a database and have the agent highlight changes since the last analysis.

What data sources work best for market research agents?

News APIs (NewsAPI, Bing News) for recent developments, Crunchbase or PitchBook for company data, SEC EDGAR for financial filings, and social media APIs for sentiment. The more diverse your sources, the richer the analysis.

Can this agent produce charts and visualizations?

Yes. Add a chart generation tool (like the one in the data analysis agent post) and instruct the agent to create bar charts comparing competitors on key metrics or line charts showing trend frequency over time.

---

#MarketResearch #CompetitiveAnalysis #TrendDetection #WebScraping #AIAgents #AgenticAI #LearnAI #AIEngineering