AI Agent for Property Market Analysis: Neighborhood Data, Trends, and Investment Insights

Why AI Market Analysis Matters for Real Estate

Traditional market analysis relies on quarterly MLS reports and gut instinct. By the time a market report is published, the data is weeks old. An AI market analysis agent pulls data from multiple sources in real time, identifies emerging trends, scores neighborhoods for investment potential, and generates reports that would take an analyst days to compile manually.

Data Aggregation Layer

The agent needs to pull from multiple data sources and normalize them into a unified format.

from dataclasses import dataclass, field
from datetime import date
from typing import Optional

@dataclass
class NeighborhoodMetrics:
    neighborhood: str
    city: str
    state: str
    median_home_price: float
    median_rent: float
    price_change_yoy: float  # year-over-year percent
    rent_change_yoy: float
    days_on_market: int
    inventory_count: int
    sale_to_list_ratio: float  # 1.02 = 2% over asking
    population_growth: float
    median_income: float
    crime_rate_per_1000: float
    school_rating: float  # 1-10
    walk_score: int
    transit_score: int

@dataclass
class MarketDataSource:
    name: str
    data_type: str  # sales, rentals, demographics, crime, schools
    freshness_days: int  # how old the data is

async def aggregate_neighborhood_data(
    neighborhood: str,
    city: str,
    state: str,
    pool=None,
) -> NeighborhoodMetrics:
    """Pull and aggregate data from multiple sources."""
    # Sales data from MLS feed
    sales_data = await pool.fetchrow("""
        SELECT
            PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY sale_price) as median_price,
            AVG(days_on_market) as avg_dom,
            COUNT(*) as sale_count,
            AVG(sale_price::float / list_price) as sale_to_list
        FROM sales
        WHERE neighborhood = $1
          AND sale_date >= NOW() - INTERVAL '6 months'
    """, neighborhood)

    # Rental data
    rental_data = await pool.fetchrow("""
        SELECT PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY rent) as median_rent
        FROM active_rentals
        WHERE neighborhood = $1
    """, neighborhood)

    # YoY comparison
    prior_year = await pool.fetchrow("""
        SELECT
            PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY sale_price) as median_price
        FROM sales
        WHERE neighborhood = $1
          AND sale_date BETWEEN NOW() - INTERVAL '18 months'
                             AND NOW() - INTERVAL '12 months'
    """, neighborhood)

    current_price = float(sales_data["median_price"])
    prior_price = float(prior_year["median_price"])
    yoy_change = ((current_price - prior_price) / prior_price) * 100

    return NeighborhoodMetrics(
        neighborhood=neighborhood,
        city=city,
        state=state,
        median_home_price=current_price,
        median_rent=float(rental_data["median_rent"]),
        price_change_yoy=round(yoy_change, 1),
        rent_change_yoy=0.0,  # similar calculation
        days_on_market=int(sales_data["avg_dom"]),
        inventory_count=int(sales_data["sale_count"]),
        sale_to_list_ratio=round(float(sales_data["sale_to_list"]), 3),
        population_growth=0.0,  # from census API
        median_income=0.0,      # from census API
        crime_rate_per_1000=0.0, # from crime API
        school_rating=0.0,      # from school API
        walk_score=0,           # from Walk Score API
        transit_score=0,        # from Walk Score API
    )

Investment Scoring Algorithm

The agent scores neighborhoods on investment potential using a weighted multi-factor model.

@dataclass
class InvestmentScore:
    neighborhood: str
    overall_score: float  # 0-100
    appreciation_score: float
    cash_flow_score: float
    stability_score: float
    growth_score: float
    risk_factors: list[str]
    opportunity_signals: list[str]

def score_investment_potential(
    metrics: NeighborhoodMetrics,
) -> InvestmentScore:
    """Score a neighborhood for investment potential."""
    risk_factors = []
    opportunities = []

    # Appreciation potential (0-25)
    if metrics.price_change_yoy > 10:
        appreciation = 15  # already appreciated a lot
        risk_factors.append("Rapid appreciation may indicate overheating")
    elif metrics.price_change_yoy > 5:
        appreciation = 25
        opportunities.append("Strong but sustainable appreciation trend")
    elif metrics.price_change_yoy > 0:
        appreciation = 20
    else:
        appreciation = 5
        risk_factors.append("Declining property values")

    # Cash flow potential (0-25)
    if metrics.median_rent > 0 and metrics.median_home_price > 0:
        gross_yield = (metrics.median_rent * 12) / metrics.median_home_price * 100
        if gross_yield > 8:
            cash_flow = 25
            opportunities.append(f"High gross yield: {gross_yield:.1f}%")
        elif gross_yield > 5:
            cash_flow = 18
        else:
            cash_flow = 8
            risk_factors.append(f"Low gross yield: {gross_yield:.1f}%")
    else:
        cash_flow = 0

    # Market stability (0-25)
    stability = 15  # baseline
    if metrics.days_on_market < 14:
        stability += 5
        opportunities.append("Fast-moving market (seller's market)")
    elif metrics.days_on_market > 60:
        stability -= 5
        risk_factors.append("Slow market — properties sit long")
    if metrics.sale_to_list_ratio > 1.0:
        stability += 5

    # Growth indicators (0-25)
    growth = 12  # baseline
    if metrics.population_growth > 2:
        growth += 8
        opportunities.append("Strong population growth")
    if metrics.walk_score > 70:
        growth += 5

    overall = appreciation + cash_flow + stability + growth

    return InvestmentScore(
        neighborhood=metrics.neighborhood,
        overall_score=min(100, overall),
        appreciation_score=appreciation,
        cash_flow_score=cash_flow,
        stability_score=stability,
        growth_score=growth,
        risk_factors=risk_factors,
        opportunity_signals=opportunities,
    )

Trend Detection

The agent identifies emerging trends by comparing rolling metrics.

from typing import Optional as Opt

async def detect_market_trends(
    neighborhood: str,
    pool=None,
) -> list[dict]:
    """Detect emerging market trends from historical data."""
    rows = await pool.fetch("""
        SELECT
            DATE_TRUNC('month', sale_date) as month,
            PERCENTILE_CONT(0.5) WITHIN GROUP (ORDER BY sale_price) as median_price,
            AVG(days_on_market) as avg_dom,
            COUNT(*) as volume
        FROM sales
        WHERE neighborhood = $1
          AND sale_date >= NOW() - INTERVAL '24 months'
        GROUP BY DATE_TRUNC('month', sale_date)
        ORDER BY month
    """, neighborhood)

    trends = []
    if len(rows) >= 6:
        recent_3 = rows[-3:]
        prior_3 = rows[-6:-3]
        recent_avg = sum(r["median_price"] for r in recent_3) / 3
        prior_avg = sum(r["median_price"] for r in prior_3) / 3
        price_momentum = ((recent_avg - prior_avg) / prior_avg) * 100

        if price_momentum > 5:
            trends.append({
                "type": "price_acceleration",
                "description": f"Prices accelerating: {price_momentum:.1f}% gain in last 3 months vs prior 3",
                "confidence": "high" if price_momentum > 10 else "medium",
            })
        elif price_momentum < -3:
            trends.append({
                "type": "price_deceleration",
                "description": f"Prices cooling: {price_momentum:.1f}% change in last 3 months",
                "confidence": "high" if price_momentum < -8 else "medium",
            })

        recent_dom = sum(r["avg_dom"] for r in recent_3) / 3
        prior_dom = sum(r["avg_dom"] for r in prior_3) / 3
        if recent_dom < prior_dom * 0.8:
            trends.append({
                "type": "market_tightening",
                "description": "Days on market dropping — demand increasing",
                "confidence": "medium",
            })
    return trends

The Market Analysis Agent

from agents import Agent, function_tool

@function_tool
async def analyze_neighborhood(
    neighborhood: str,
    city: str,
    state: str,
) -> str:
    """Get comprehensive market data for a neighborhood."""
    # In production: calls aggregate_neighborhood_data
    return (
        f"## {neighborhood}, {city}\n"
        f"Median Home Price: $485,000 (+6.2% YoY)\n"
        f"Median Rent: $2,100/mo\n"
        f"Days on Market: 18\n"
        f"Sale-to-List Ratio: 1.02\n"
        f"Inventory: 45 active listings\n"
        f"Gross Yield: 5.2%\n"
        f"Walk Score: 72 | Transit Score: 55"
    )

@function_tool
async def score_for_investment(neighborhood: str) -> str:
    """Score a neighborhood's investment potential."""
    return (
        f"## Investment Score: {neighborhood}\n"
        f"Overall: 74/100\n"
        f"- Appreciation: 22/25\n"
        f"- Cash Flow: 18/25\n"
        f"- Stability: 19/25\n"
        f"- Growth: 15/25\n\n"
        f"Opportunities: Strong appreciation trend, fast market\n"
        f"Risks: Yield compression as prices outpace rents"
    )

@function_tool
async def compare_neighborhoods(neighborhoods: str) -> str:
    """Compare multiple neighborhoods side by side."""
    areas = [n.strip() for n in neighborhoods.split(",")]
    header = f"Comparing: {', '.join(areas)}\n\n"
    # In production: generates a comparison table
    return header + (
        "| Metric | Area A | Area B |\n"
        "|--------|--------|--------|\n"
        "| Median Price | $485k | $520k |\n"
        "| YoY Change | +6.2% | +3.8% |\n"
        "| Gross Yield | 5.2% | 4.1% |\n"
        "| Inv. Score | 74 | 68 |"
    )

@function_tool
async def get_market_trends(neighborhood: str) -> str:
    """Identify emerging market trends for a neighborhood."""
    return (
        "Detected Trends:\n"
        "1. Price acceleration: +8.3% in last 3mo vs +4.1% prior (HIGH confidence)\n"
        "2. Market tightening: DOM dropped from 28 to 18 days (MEDIUM confidence)\n"
        "3. Investor activity rising: Cash purchases up 15% (MEDIUM confidence)"
    )

market_agent = Agent(
    name="PropertyMarketAnalyst",
    instructions="""You are a real estate market analyst.
    Provide data-driven insights about neighborhoods and
    investment opportunities. Always cite specific metrics.
    Distinguish between facts (data) and analysis (interpretation).
    Include risk factors alongside opportunities.
    Never guarantee returns or make specific price predictions.""",
    tools=[
        analyze_neighborhood,
        score_for_investment,
        compare_neighborhoods,
        get_market_trends,
    ],
)

FAQ

How frequently should the market data be refreshed?

Sales data should refresh daily from MLS feeds. Rental data can refresh weekly. Demographic data (census, crime, schools) updates quarterly or annually. The agent should always display the data freshness date so users know how current the analysis is.

Can the agent predict future property prices?

The agent identifies trends and momentum but should never present point predictions ("this home will be worth $X next year"). Instead, it provides scenario analysis: "If current trends continue, median prices could rise 4-7% over the next 12 months. However, rising interest rates represent a downside risk." Framing analysis as scenarios with conditions is both more accurate and more honest.

How does the investment score handle different investment strategies?

The current scoring model is general-purpose. For specific strategies, you can adjust the weights — a cash flow investor would weight the cash flow score at 40% instead of 25%, while a flip investor would heavily weight days-on-market and appreciation momentum. The agent can accept the investment strategy as input and apply the appropriate weighting profile.

---

#MarketAnalysis #InvestmentInsights #RealEstateAI #Python #DataAnalytics #AgenticAI #LearnAI #AIEngineering