Real Estate Data Scraping: Track Listings, Prices & Market Trends

from scrapegraph_py import Client
 
# Initialize the client with your API key
client = Client(api_key="your-api-key-here")
 
# SmartScraper request to extract property listings
response = client.smartscraper(
    website_url="https://www.zillow.com/san-francisco-ca/",
    user_prompt="""Extract all property listings including:
    - Address
    - Price
    - Bedrooms and bathrooms
    - Square footage
    - Lot size
    - Year built
    - Days on market
    - Listing agent
    - Property type (house, condo, townhouse)
    - Key features
    - Listing URL
    """
)
 
print("Result:", response)

{
  "listings": [
    {
      "address": "123 Market Street, San Francisco, CA 94102",
      "price": "$1,250,000",
      "bedrooms": 2,
      "bathrooms": 2,
      "sqft": 1450,
      "year_built": 2018,
      "days_on_market": 12,
      "property_type": "Condo",
      "features": ["City View", "Parking", "Gym Access"]
    }
  ]
}

from scrapegraph_py import Client
 
# Initialize the client
client = Client(api_key="your-api-key-here")
 
# SearchScraper request to find properties
response = client.searchscraper(
    user_prompt="Find 3-bedroom houses for sale in Austin, TX under $500,000 with prices, addresses, and square footage",
    num_results=5
)
 
print("Result:", response)

from scrapegraph_py import Client
 
client = Client(api_key="your-api-key-here")
 
# Deep dive into a specific listing
response = client.smartscraper(
    website_url="https://www.redfin.com/CA/San-Francisco/123-Main-St-94102/home/12345678",
    user_prompt="""Extract complete property details:
    - Full address
    - Current listing price
    - Price history (all previous prices and dates)
    - Tax assessment value
    - HOA fees if applicable
    - Property description
    - All interior features
    - All exterior features
    - Heating and cooling
    - Parking details
    - School ratings (elementary, middle, high)
    - Walk score, transit score, bike score
    - Nearby amenities
    - Recent sales in the area
    - Estimated rent value
    """
)
 
print("Result:", response)

search_criteria = {
    "locations": [
        "San Francisco, CA",
        "Oakland, CA",
        "San Jose, CA"
    ],
    "property_types": ["house", "condo"],
    "min_beds": 2,
    "max_beds": 4,
    "min_price": 500000,
    "max_price": 1500000,
    "min_sqft": 1000
}

def monitor_all_platforms(criteria):
    platforms = {
        "zillow": f"https://www.zillow.com/homes/{criteria['location']}/",
        "redfin": f"https://www.redfin.com/city/{criteria['location']}/",
        "realtor": f"https://www.realtor.com/realestateandhomes-search/{criteria['location']}"
    }
    
    all_listings = []
    
    for platform, url in platforms.items():
        listings = client.smartscraper(
            website_url=url,
            user_prompt=f"""Find properties matching:
            - {criteria['min_beds']}-{criteria['max_beds']} bedrooms
            - Price ${criteria['min_price']:,} to ${criteria['max_price']:,}
            - Minimum {criteria['min_sqft']} sqft
            
            Extract: address, price, beds, baths, sqft, listing URL"""
        )
        
        for listing in listings.get("properties", []):
            listing["source"] = platform
            all_listings.append(listing)
    
    return all_listings

import json
from datetime import datetime
 
def track_price_changes(property_id, current_price, history_file="price_history.json"):
    # Load existing history
    try:
        with open(history_file, 'r') as f:
            history = json.load(f)
    except FileNotFoundError:
        history = {}
    
    # Add new price point
    if property_id not in history:
        history[property_id] = []
    
    history[property_id].append({
        "price": current_price,
        "date": datetime.now().isoformat()
    })
    
    # Check for price drop
    if len(history[property_id]) > 1:
        previous = history[property_id][-2]["price"]
        if current_price < previous:
            price_drop = previous - current_price
            return {"alert": "price_drop", "amount": price_drop}
    
    # Save updated history
    with open(history_file, 'w') as f:
        json.dump(history, f)
    
    return None

def check_new_listings(current_listings, known_listings_file="known_listings.json"):
    # Load known listings
    try:
        with open(known_listings_file, 'r') as f:
            known = set(json.load(f))
    except FileNotFoundError:
        known = set()
    
    new_listings = []
    
    for listing in current_listings:
        listing_id = listing.get("address") or listing.get("url")
        if listing_id and listing_id not in known:
            new_listings.append(listing)
            known.add(listing_id)
    
    # Save updated known listings
    with open(known_listings_file, 'w') as f:
        json.dump(list(known), f)
    
    return new_listings