How to Monitor Changes in Real Estate Market: Automated Property Listing Change Detection

import requests
from datetime import datetime, timedelta
import json

class RealEstateMonitoringSystem:
    """Comprehensive real estate market monitoring system"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.supacrawler.com/api/v1"
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
        self.active_monitors = []
    
    def create_market_monitor(self, market_config: dict):
        """Create monitoring for a specific real estate market"""
        
        # Configure market-specific parameters
        watch_config = {
            "url": market_config["listing_url"],
            "frequency": market_config.get("frequency", "hourly"),
            "notify_email": market_config.get("email", "[email protected]"),
            "notification_preference": "changes_only",
            
            # Target specific property data elements
            "selector": self._build_property_selector(market_config["property_type"]),
            
            # Capture visual evidence
            "include_image": True,
            "include_html": True,
            "full_page": False,  # Focus on listings
            "quality": 90,
            
            # Custom headers for some property sites
            "headers": {
                "User-Agent": "RealEstateAnalytics/1.0"
            },
            
            # Webhook for real-time processing
            "webhook_url": market_config.get("webhook_url", ""),
            "webhook_headers": {
                "X-Market": market_config["market_name"],
                "X-Property-Type": market_config["property_type"]
            }
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/watch",
                headers=self.headers,
                json=watch_config
            )
            
            result = response.json()
            
            if result.get('success'):
                monitor_info = {
                    'watch_id': result['watch_id'],
                    'market_name': market_config['market_name'],
                    'property_type': market_config['property_type'],
                    'monitoring_url': market_config['listing_url'],
                    'created_at': datetime.now().isoformat(),
                    'status': 'active'
                }
                
                self.active_monitors.append(monitor_info)
                return monitor_info
                
            else:
                raise Exception(f"Failed to create monitor: {result.get('error')}")
                
        except Exception as e:
            print(f"Error creating market monitor: {e}")
            return None
    
    def _build_property_selector(self, property_type: str) -> str:
        """Build CSS selectors based on property type and common listing sites"""
        
        selectors = {
            "residential": ".property-card, .listing-item, .property-result, .home-card",
            "commercial": ".commercial-listing, .office-space, .retail-property",
            "rental": ".rental-unit, .apartment-listing, .rental-property",
            "land": ".land-listing, .lot-property, .development-opportunity",
            "luxury": ".luxury-home, .premium-property, .executive-home"
        }
        
        # Common price and key info selectors
        common_selectors = [
            ".price, .listing-price, .property-price",
            ".address, .property-address, .location",
            ".beds, .bedrooms, .sqft, .square-feet",
            ".status, .listing-status, .availability"
        ]
        
        base_selector = selectors.get(property_type, ".property-listing")
        all_selectors = [base_selector] + common_selectors
        
        return ", ".join(all_selectors)

# Initialize monitoring system
def setup_comprehensive_monitoring(api_key: str):
    """Setup monitoring for multiple real estate markets"""
    
    monitor_system = RealEstateMonitoringSystem(api_key)
    
    # Define market monitoring configurations
    market_configs = [
        {
            "market_name": "Downtown Austin Condos",
            "property_type": "residential",
            "listing_url": "https://austin-realty.com/downtown-condos",
            "frequency": "hourly",
            "email": "[email protected]",
            "webhook_url": "https://realestate-firm.com/api/austin-updates"
        },
        {
            "market_name": "Miami Beach Luxury",
            "property_type": "luxury", 
            "listing_url": "https://luxury-miami.com/beach-properties",
            "frequency": "every_30_minutes",
            "email": "[email protected]",
            "webhook_url": "https://realestate-firm.com/api/luxury-updates"
        },
        {
            "market_name": "Denver Commercial", 
            "property_type": "commercial",
            "listing_url": "https://denver-commercial.com/office-retail",
            "frequency": "daily",
            "email": "[email protected]",
            "webhook_url": "https://realestate-firm.com/api/commercial-updates"
        },
        {
            "market_name": "Portland Rental Market",
            "property_type": "rental",
            "listing_url": "https://portland-rentals.com/apartments",
            "frequency": "every_2_hours",
            "email": "[email protected]",
            "webhook_url": "https://realestate-firm.com/api/rental-updates"
        },
        {
            "market_name": "Phoenix Development Land",
            "property_type": "land",
            "listing_url": "https://phoenix-land.com/development-opportunities",
            "frequency": "weekly", 
            "email": "[email protected]",
            "webhook_url": "https://realestate-firm.com/api/land-updates"
        }
    ]
    
    created_monitors = []
    
    for config in market_configs:
        print(f"Creating monitor for {config['market_name']}...")
        monitor = monitor_system.create_market_monitor(config)
        
        if monitor:
            created_monitors.append(monitor)
            print(f"✅ {config['market_name']}: {monitor['watch_id']}")
        else:
            print(f"❌ Failed to create monitor for {config['market_name']}")
    
    return monitor_system, created_monitors

# Example usage
api_key = "your_supacrawler_api_key"
monitoring_system, monitors = setup_comprehensive_monitoring(api_key)

print(f"\n🏠 Real Estate Monitoring System Active")
print(f"📊 Monitoring {len(monitors)} markets:")
for monitor in monitors:
    print(f"  • {monitor['market_name']}: {monitor['property_type']} properties")

class PropertyPriceTracker:
    """Specialized system for tracking property price changes"""
    
    def __init__(self, monitoring_system):
        self.monitoring_system = monitoring_system
        self.price_history = {}
        self.alert_thresholds = {}
    
    def setup_price_monitoring(self, properties: list):
        """Setup dedicated price monitoring for specific properties"""
        
        for property_info in properties:
            # Create individual property monitor
            price_config = {
                "market_name": f"Price Tracker: {property_info['address']}",
                "property_type": "individual_property",
                "listing_url": property_info["listing_url"],
                "frequency": "hourly",  # Frequent for price changes
                "email": property_info.get("alert_email", "[email protected]"),
                "webhook_url": property_info.get("webhook_url", "")
            }
            
            monitor = self.monitoring_system.create_market_monitor(price_config)
            
            if monitor:
                # Set up price change thresholds
                self.alert_thresholds[monitor['watch_id']] = {
                    "percentage_change": property_info.get("alert_percentage", 5),  # 5% change
                    "absolute_change": property_info.get("alert_amount", 10000),    # $10k change
                    "price_reduction_only": property_info.get("reduction_only", False)
                }
                
                print(f"✅ Price monitoring: {property_info['address']}")
                
    def analyze_price_changes(self, webhook_data: dict):
        """Analyze incoming price change notifications"""
        
        watch_id = webhook_data.get('watch_id')
        if not watch_id or watch_id not in self.alert_thresholds:
            return None
        
        # Extract price information from change data
        current_price = self._extract_price(webhook_data.get('new_content', ''))
        previous_price = self._extract_price(webhook_data.get('previous_content', ''))
        
        if not current_price or not previous_price:
            return None
        
        # Calculate price change
        price_change = current_price - previous_price
        percentage_change = (price_change / previous_price) * 100 if previous_price > 0 else 0
        
        thresholds = self.alert_thresholds[watch_id]
        
        # Check if change meets alert criteria
        alert_triggered = (
            abs(percentage_change) >= thresholds["percentage_change"] or
            abs(price_change) >= thresholds["absolute_change"]
        )
        
        if thresholds["price_reduction_only"] and price_change >= 0:
            alert_triggered = False
        
        if alert_triggered:
            alert_data = {
                "watch_id": watch_id,
                "property_url": webhook_data.get('url'),
                "previous_price": previous_price,
                "current_price": current_price,
                "price_change": price_change,
                "percentage_change": percentage_change,
                "change_direction": "increase" if price_change > 0 else "decrease",
                "screenshot_url": webhook_data.get('screenshot_url'),
                "detected_at": datetime.now().isoformat()
            }
            
            self._send_price_alert(alert_data)
            return alert_data
            
        return None
    
    def _extract_price(self, content: str) -> float:
        """Extract price from listing content using regex patterns"""
        import re
        
        # Common price patterns in real estate listings
        price_patterns = [
            r'\$([0-9,]+)',  # $1,250,000
            r'Price[:\s]*\$([0-9,]+)',  # Price: $1,250,000
            r'([0-9,]+)\s*USD',  # 1,250,000 USD
            r'\$([0-9]+\.[0-9]{2})',  # $1250000.00
        ]
        
        for pattern in price_patterns:
            matches = re.findall(pattern, content)
            if matches:
                # Convert first match to float
                price_str = matches[0].replace(',', '')
                try:
                    return float(price_str)
                except ValueError:
                    continue
        
        return None
    
    def _send_price_alert(self, alert_data: dict):
        """Send price change alert via multiple channels"""
        
        # Email alert
        email_subject = f"Price {alert_data['change_direction'].title()}: {alert_data['property_url']}"
        email_body = f"""
        Property Price Change Detected
        
        Property: {alert_data['property_url']}
        Previous Price: ${alert_data['previous_price']:,.0f}
        Current Price: ${alert_data['current_price']:,.0f}
        Change Amount: ${alert_data['price_change']:,.0f}
        Percentage Change: {alert_data['percentage_change']:.1f}%
        
        Screenshot: {alert_data['screenshot_url']}
        Detected: {alert_data['detected_at']}
        """
        
        # Slack notification
        slack_payload = {
            "text": f"🏠 Property Price {alert_data['change_direction'].title()}",
            "attachments": [
                {
                    "color": "danger" if alert_data['change_direction'] == 'decrease' else "warning",
                    "fields": [
                        {"title": "Previous Price", "value": f"${alert_data['previous_price']:,.0f}", "short": True},
                        {"title": "Current Price", "value": f"${alert_data['current_price']:,.0f}", "short": True},
                        {"title": "Change", "value": f"${alert_data['price_change']:,.0f} ({alert_data['percentage_change']:.1f}%)", "short": True},
                        {"title": "Property", "value": alert_data['property_url'], "short": False}
                    ]
                }
            ]
        }
        
        print(f"🚨 Price Alert: {alert_data['change_direction']} of ${abs(alert_data['price_change']):,.0f}")
        print(f"   Property: {alert_data['property_url']}")

# Example price monitoring setup
properties_to_monitor = [
    {
        "address": "123 Luxury Lane, Beverly Hills",
        "listing_url": "https://luxury-listings.com/property/123-luxury-lane",
        "alert_percentage": 3,  # Alert on 3% change
        "alert_amount": 50000,  # Alert on $50k change
        "reduction_only": True,  # Only alert on price reductions
        "alert_email": "[email protected]"
    },
    {
        "address": "456 Investment Ave, Austin",
        "listing_url": "https://austin-properties.com/456-investment-ave",
        "alert_percentage": 5,  # Alert on 5% change
        "alert_amount": 15000,  # Alert on $15k change
        "reduction_only": False, # Alert on any significant change
        "alert_email": "[email protected]"
    }
]

# Setup price tracking
price_tracker = PropertyPriceTracker(monitoring_system)
price_tracker.setup_price_monitoring(properties_to_monitor)

class MarketTrendAnalyzer:
    """Advanced market trend analysis using property data"""
    
    def __init__(self, monitoring_system):
        self.monitoring_system = monitoring_system
        self.market_data = {}
        self.trend_thresholds = {
            "inventory_change": 10,      # 10% inventory change threshold
            "price_trend": 5,            # 5% price movement threshold
            "days_on_market": 7,         # Days on market change threshold
            "absorption_rate": 15        # Market absorption rate threshold
        }
    
    def setup_market_analysis(self, markets: list):
        """Setup comprehensive market trend monitoring"""
        
        for market in markets:
            # Create market overview monitors
            market_monitors = self._create_market_overview_monitors(market)
            
            # Create segment-specific monitors
            segment_monitors = self._create_segment_monitors(market)
            
            # Store market configuration
            self.market_data[market['market_id']] = {
                'market_name': market['market_name'],
                'overview_monitors': market_monitors,
                'segment_monitors': segment_monitors,
                'baseline_data': {},
                'trend_history': []
            }
    
    def _create_market_overview_monitors(self, market: dict):
        """Create monitors for overall market metrics"""
        
        overview_configs = [
            {
                "monitor_type": "inventory_levels",
                "url": f"{market['base_url']}/search?location={market['location']}",
                "selector": ".search-results-count, .total-listings, .inventory-summary",
                "frequency": "daily"
            },
            {
                "monitor_type": "price_distribution", 
                "url": f"{market['base_url']}/market-stats?area={market['location']}",
                "selector": ".price-stats, .market-summary, .pricing-data",
                "frequency": "weekly"
            },
            {
                "monitor_type": "days_on_market",
                "url": f"{market['base_url']}/market-trends?region={market['location']}",
                "selector": ".market-velocity, .days-on-market, .selling-time",
                "frequency": "weekly"
            }
        ]
        
        created_monitors = []
        
        for config in overview_configs:
            monitor_config = {
                "market_name": f"{market['market_name']} - {config['monitor_type']}",
                "property_type": "market_analysis",
                "listing_url": config["url"],
                "frequency": config["frequency"],
                "email": f"market-analysis-{market['market_id']}@company.com",
                "webhook_url": f"https://company.com/api/market-trends/{market['market_id']}/{config['monitor_type']}"
            }
            
            monitor = self.monitoring_system.create_market_monitor(monitor_config)
            if monitor:
                monitor['monitor_type'] = config['monitor_type']
                created_monitors.append(monitor)
        
        return created_monitors
    
    def _create_segment_monitors(self, market: dict):
        """Create monitors for specific market segments"""
        
        segments = [
            {"name": "entry_level", "price_range": "0-300000", "segment_focus": "first-time buyers"},
            {"name": "mid_market", "price_range": "300000-600000", "segment_focus": "move-up buyers"},
            {"name": "luxury", "price_range": "600000-plus", "segment_focus": "luxury market"},
            {"name": "investment", "criteria": "rental-property", "segment_focus": "investors"}
        ]
        
        segment_monitors = []
        
        for segment in segments:
            # Build segment-specific URL
            if "price_range" in segment:
                segment_url = f"{market['base_url']}/search?location={market['location']}&price={segment['price_range']}"
            else:
                segment_url = f"{market['base_url']}/search?location={market['location']}&type={segment['criteria']}"
            
            monitor_config = {
                "market_name": f"{market['market_name']} - {segment['name']} Segment",
                "property_type": "market_segment",
                "listing_url": segment_url,
                "frequency": "daily",
                "email": f"segment-analysis-{market['market_id']}@company.com",
                "webhook_url": f"https://company.com/api/segments/{market['market_id']}/{segment['name']}"
            }
            
            monitor = self.monitoring_system.create_market_monitor(monitor_config)
            if monitor:
                monitor['segment_name'] = segment['name']
                monitor['segment_focus'] = segment['segment_focus']
                segment_monitors.append(monitor)
        
        return segment_monitors
    
    def analyze_market_trends(self, webhook_data: dict):
        """Analyze market trends from incoming data"""
        
        market_id = self._extract_market_id(webhook_data.get('webhook_headers', {}))
        monitor_type = self._extract_monitor_type(webhook_data.get('webhook_headers', {}))
        
        if not market_id or market_id not in self.market_data:
            return None
        
        # Extract relevant metrics from the content
        current_metrics = self._extract_market_metrics(webhook_data.get('new_content', ''), monitor_type)
        previous_metrics = self._extract_market_metrics(webhook_data.get('previous_content', ''), monitor_type)
        
        if not current_metrics or not previous_metrics:
            return None
        
        # Calculate trend indicators
        trend_analysis = self._calculate_trends(current_metrics, previous_metrics, monitor_type)
        
        # Store trend data
        market_data = self.market_data[market_id]
        market_data['trend_history'].append({
            'timestamp': datetime.now().isoformat(),
            'monitor_type': monitor_type,
            'metrics': current_metrics,
            'trend_analysis': trend_analysis
        })
        
        # Check for significant trend alerts
        alerts = self._check_trend_alerts(trend_analysis, monitor_type)
        
        if alerts:
            self._send_trend_alerts(market_id, alerts, trend_analysis)
        
        return trend_analysis
    
    def _extract_market_metrics(self, content: str, monitor_type: str):
        """Extract market metrics from listing content"""
        import re
        
        metrics = {}
        
        if monitor_type == "inventory_levels":
            # Extract total listings count
            count_patterns = [
                r'([0-9,]+)\s*(?:properties|listings|homes)',
                r'Total[:\s]*([0-9,]+)',
                r'([0-9,]+)\s*results'
            ]
            
            for pattern in count_patterns:
                matches = re.findall(pattern, content, re.IGNORECASE)
                if matches:
                    metrics['total_listings'] = int(matches[0].replace(',', ''))
                    break
        
        elif monitor_type == "price_distribution":
            # Extract price statistics
            price_patterns = {
                'median_price': r'Median[:\s]*\$([0-9,]+)',
                'average_price': r'Average[:\s]*\$([0-9,]+)',
                'price_per_sqft': r'\$([0-9,]+)[\s/]*(?:sq\.?\s*ft|sqft|per\s*sq)'
            }
            
            for key, pattern in price_patterns.items():
                matches = re.findall(pattern, content, re.IGNORECASE)
                if matches:
                    metrics[key] = int(matches[0].replace(',', ''))
        
        elif monitor_type == "days_on_market":
            # Extract time-on-market data
            dom_patterns = [
                r'([0-9]+)\s*days?\s*on\s*market',
                r'Average[:\s]*([0-9]+)\s*days',
                r'DOM[:\s]*([0-9]+)'
            ]
            
            for pattern in dom_patterns:
                matches = re.findall(pattern, content, re.IGNORECASE)
                if matches:
                    metrics['average_days_on_market'] = int(matches[0])
                    break
        
        return metrics
    
    def _calculate_trends(self, current: dict, previous: dict, monitor_type: str):
        """Calculate trend indicators from metric comparison"""
        
        trends = {}
        
        for metric, current_value in current.items():
            if metric in previous:
                previous_value = previous[metric]
                
                if previous_value > 0:
                    change = current_value - previous_value
                    percentage_change = (change / previous_value) * 100
                    
                    trends[metric] = {
                        'current_value': current_value,
                        'previous_value': previous_value,
                        'absolute_change': change,
                        'percentage_change': percentage_change,
                        'trend_direction': 'up' if change > 0 else 'down' if change < 0 else 'stable'
                    }
        
        # Calculate market temperature score
        trends['market_temperature'] = self._calculate_market_temperature(trends)
        
        return trends
    
    def _calculate_market_temperature(self, trends: dict) -> str:
        """Calculate overall market temperature from trend indicators"""
        
        score = 0
        
        # Inventory trends (higher inventory = cooler market)
        if 'total_listings' in trends:
            inventory_change = trends['total_listings']['percentage_change']
            if inventory_change > 10:
                score -= 2  # Cooling
            elif inventory_change < -10:
                score += 2  # Heating
        
        # Price trends (increasing prices = hotter market)
        if 'median_price' in trends:
            price_change = trends['median_price']['percentage_change']
            if price_change > 5:
                score += 2  # Heating
            elif price_change < -5:
                score -= 2  # Cooling
        
        # Days on market (fewer days = hotter market)
        if 'average_days_on_market' in trends:
            dom_change = trends['average_days_on_market']['percentage_change']
            if dom_change < -15:
                score += 2  # Heating
            elif dom_change > 15:
                score -= 2  # Cooling
        
        # Convert score to temperature
        if score >= 4:
            return "hot"
        elif score >= 2:
            return "warm"
        elif score <= -4:
            return "cold"
        elif score <= -2:
            return "cool"
        else:
            return "balanced"

# Example market analysis setup
markets_to_analyze = [
    {
        "market_id": "austin_central",
        "market_name": "Austin Central",
        "location": "austin-tx-downtown",
        "base_url": "https://austin-realty.com"
    },
    {
        "market_id": "miami_beach", 
        "market_name": "Miami Beach",
        "location": "miami-beach-fl",
        "base_url": "https://miami-luxury.com"
    }
]

# Setup market trend analysis
trend_analyzer = MarketTrendAnalyzer(monitoring_system)
trend_analyzer.setup_market_analysis(markets_to_analyze)

print("📈 Market Trend Analysis System Active")
print("Monitoring market indicators:")
print("  • Inventory levels and changes")
print("  • Price distribution and trends") 
print("  • Days on market velocity")
print("  • Market segment performance")
print("  • Market temperature scoring")

import { SupacrawlerClient } from '@supacrawler/js'

const client = new SupacrawlerClient({ apiKey: process.env.SUPACRAWLER_API_KEY })

// Comprehensive market intelligence vs manual searches
async function createRealEstateIntelligence() {
  // Multi-market price monitoring
  const priceMonitors = await Promise.all([
    client.createWatchJob({
      url: 'https://austin-luxury.com/west-lake-hills',
      frequency: 'hourly',
      notify_email: '[email protected]',
      
      // Precise targeting vs broad searches
      selector: '.property-price, .listing-price, .price-change',
      notification_preference: 'changes_only',
      
      // Rich market context vs text alerts
      include_image: true,
      include_html: true,
      quality: 95,
      
      // Business intelligence integration
      webhook_url: 'https://realestate-firm.com/api/price-alerts',
      webhook_headers: {
        'X-Market': 'Austin-WestLakeHills',
        'X-Alert-Type': 'price-change'
      }
    }),
    
    // Investment property monitoring
    client.createWatchJob({
      url: 'https://investment-properties.com/phoenix-rental-market',
      frequency: 'daily',
      notify_email: '[email protected]',
      
      // Investment-specific data capture
      selector: '.cap-rate, .rental-yield, .cash-flow, .property-metrics',
      include_html: true,
      
      // Investment analysis webhook
      webhook_url: 'https://realestate-firm.com/api/investment-analysis'
    })
  ])
  
  // New listing alerts with instant notification
  const inventoryMonitor = await client.createWatchJob({
    url: 'https://miami-beach-luxury.com/new-listings',
    frequency: 'every_15_minutes',
    notify_email: '[email protected]',
    
    // New inventory detection
    selector: '.new-listing, .just-listed, .property-card[data-status="new"]',
    notification_preference: 'changes_only',
    
    // Immediate competitive advantage
    include_image: true,
    webhook_url: 'https://realestate-firm.com/api/new-listings',
    webhook_headers: {
      'X-Priority': 'high',
      'X-Market': 'Miami-Beach-Luxury'
    }
  })
  
  // Market trend analysis
  const trendMonitor = await client.createWatchJob({
    url: 'https://market-stats.com/denver-commercial/quarterly-report',
    frequency: 'weekly',
    notify_email: '[email protected]',
    
    // Market intelligence data
    selector: '.market-stats, .trend-indicators, .absorption-rates',
    include_html: true,
    full_page: true,
    
    // Strategic planning integration
    webhook_url: 'https://realestate-firm.com/api/market-intelligence'
  })
  
  return {
    priceMonitors: priceMonitors.map(m => m.watch_id),
    inventoryMonitor: inventoryMonitor.watch_id,
    trendMonitor: trendMonitor.watch_id
  }
}

// Competitive intelligence monitoring
async function createCompetitiveIntelligence(competitorFirms: string[]) {
  const competitorWatches = await Promise.all(
    competitorFirms.map(firm => client.createWatchJob({
      url: `https://${firm}.com/listings`,
      frequency: 'daily',
      notify_email: '[email protected]',
      
      // Competitor activity tracking
      selector: '.listing-count, .new-listings, .agent-activity, .market-share',
      include_image: true,
      
      // Competitive analysis integration
      webhook_url: 'https://realestate-firm.com/api/competitive-intelligence',
      webhook_headers: {
        'X-Competitor': firm,
        'X-Analysis-Type': 'market-activity'
      }
    }))
  )
  
  return competitorWatches.map(w => w.watch_id)
}