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
from pydantic import BaseModel, Field
from typing import List, Optional
 
class PropertyListing(BaseModel):
    address: str = Field(description="Full street address")
    city: str = Field(description="City name")
    state: str = Field(description="State abbreviation")
    zip_code: str = Field(description="ZIP code")
    price: int = Field(description="Listing price in dollars")
    bedrooms: int = Field(description="Number of bedrooms")
    bathrooms: float = Field(description="Number of bathrooms")
    sqft: int = Field(description="Square footage")
    lot_size: Optional[str] = Field(description="Lot size")
    year_built: Optional[int] = Field(description="Year property was built")
    days_on_market: Optional[int] = Field(description="Days since listing")
    property_type: str = Field(description="House, condo, townhouse, etc.")
    features: List[str] = Field(description="Key property features")
    listing_url: Optional[str] = Field(description="URL to full listing")
 
class ListingsResponse(BaseModel):
    listings: List[PropertyListing] = Field(description="All property listings found")
    total_count: int = Field(description="Total number of listings")
 
client = Client(api_key="your-api-key-here")
 
response = client.smartscraper(
    website_url="https://www.zillow.com/san-francisco-ca/",
    user_prompt="Extract all property listings with full details",
    output_schema=ListingsResponse
)
 
data = ListingsResponse(**response["result"])
for prop in data.listings:
    price_per_sqft = prop.price / prop.sqft if prop.sqft else 0
    print(f"{prop.address}: ${prop.price:,} (${price_per_sqft:.0f}/sqft)")

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