How to Scrape Redfin Real Estate Data in 2026 (API + Web Scraping)
How to Scrape Redfin Real Estate Data in 2026 (API + Web Scraping)
Real estate data is among the most valuable structured information on the public internet. Property listings, price history, days on market, neighborhood statistics, and comparable sales feed a wide range of high-value applications: automated valuation models, investment opportunity screening, rental yield calculators, market timing signals, and competitive analysis tools. Redfin is one of the richest free sources for this data — and unlike many platforms, it exposes a surprising amount of that richness through discoverable internal API endpoints.
Redfin was built as a technology-first brokerage from day one. Their engineering team chose a service-oriented architecture where the frontend communicates with well-structured backend APIs, and those APIs — while undocumented publicly — follow consistent patterns that have remained stable for years. The same "stingray" API endpoints that power Redfin's map search, property detail pages, and neighborhood statistics are accessible with properly crafted HTTP requests. You do not need to parse HTML or execute JavaScript to get most of Redfin's data — you can access it as clean JSON by targeting the APIs directly.
The challenge is getting through the anti-bot layer that sits in front of those APIs. Redfin runs Akamai Bot Manager, one of the most sophisticated CDN-level anti-bot systems in use. Akamai evaluates dozens of signals per request — TLS fingerprint, HTTP/2 frame priorities, header ordering, IP reputation, behavioral patterns, and JavaScript execution characteristics — to determine whether a visitor is a real browser. A plain Python HTTP request hits several of these tripwires simultaneously and gets blocked long before reaching Redfin's actual servers.
This guide covers the complete picture: Redfin's internal API structure, how to make authenticated requests that bypass Akamai's basic checks, proxy configuration with ThorData's residential network for sustained access, and five complete use cases with working Python code for property research workflows.
Understanding Redfin's data model is worth a brief detour before the code. Redfin structures real estate data around three core objects: the property (a physical address with stable propertyId), the listing (a specific sale event with listingId — the same property can have multiple listings over time), and the market (a geographic region used for statistical aggregates). Most API endpoints accept one or more of these identifiers as parameters. Property IDs are stable and persistent; listing IDs change each time a property goes on market. When scraping at scale, you typically collect propertyId values from search endpoints and then use those IDs to fetch detail, history, and comps data.
Redfin's Stingray API Architecture
Redfin's backend system is internally called "stingray." All stingray endpoints share a common URL prefix and response format:
https://www.redfin.com/stingray/api/...
Every response — whether from a search endpoint, a property detail endpoint, or a stats endpoint — is prefixed with the string {}&& before the JSON payload. This is a JSON hijacking countermeasure (the same technique Google uses for its APIs). You must strip these first four characters before parsing. Forgetting this is the most common cause of json.JSONDecodeError when first working with Redfin's APIs.
import httpx
import json
def parse_redfin_response(text: str) -> dict:
"""Strip the {}&& prefix and parse Redfin's JSON response."""
if text.startswith("{}&&"):
text = text[4:]
elif text.startswith("{}&&\n"):
text = text[5:]
try:
return json.loads(text)
except json.JSONDecodeError as e:
raise ValueError(f"Failed to parse Redfin response: {e}\nFirst 200 chars: {text[:200]}")
The base HTTP headers you need to send with every stingray request:
REDFIN_HEADERS = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36",
"Accept": "application/json, text/plain, */*",
"Accept-Language": "en-US,en;q=0.9",
"Accept-Encoding": "gzip, deflate, br",
"Referer": "https://www.redfin.com/",
"sec-ch-ua": '"Chromium";v="134", "Google Chrome";v="134", "Not:A-Brand";v="99"',
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": '"macOS"',
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
}
Property Search API
The geographic search endpoint returns property listings for a bounding box or polygon. This is how Redfin's map view works — as you pan and zoom the map, the frontend calls this endpoint with updated boundary coordinates.
import httpx
import json
from typing import Optional
def parse_redfin_response(text: str) -> dict:
"""Strip {}&& prefix and parse JSON."""
return json.loads(text[4:] if text.startswith("{}&&") else text)
def search_properties_by_polygon(
polygon: str,
market: str = "seattle",
num_homes: int = 350,
status: int = 9,
property_types: str = "1,2,3",
min_price: Optional[int] = None,
max_price: Optional[int] = None,
min_beds: Optional[int] = None,
sort: str = "redfin-recommended-asc",
client: httpx.Client = None,
) -> list[dict]:
"""
Search for properties within a polygon boundary.
Args:
polygon: Space-separated lon/lat pairs defining the search area
e.g. "-122.459 47.481,-122.459 47.735,-122.224 47.735,-122.224 47.481"
market: Redfin market slug (e.g., "seattle", "portland", "boston")
num_homes: Max results per request (Redfin caps this at 350)
status: 9 = active listings, 130 = sold listings
property_types: Comma-separated: 1=house, 2=condo, 3=townhouse, 4=multi-family
sort: Sort order — "redfin-recommended-asc", "price-asc", "price-desc", etc.
Returns:
List of property dicts with id, address, price, beds, baths, sqft, etc.
"""
params = {
"al": 1,
"market": market,
"num_homes": num_homes,
"ord": sort,
"page_number": 1,
"poly": polygon,
"sf": "1,2,3,5,6,7",
"start": 0,
"status": status,
"uipt": property_types,
"v": 8,
}
if min_price:
params["min_price"] = min_price
if max_price:
params["max_price"] = max_price
if min_beds:
params["min_beds"] = min_beds
own_client = client is None
if own_client:
client = httpx.Client(headers=REDFIN_HEADERS, follow_redirects=True, timeout=httpx.Timeout(connect=8.0, read=30.0))
try:
response = client.get("https://www.redfin.com/stingray/api/gis", params=params)
response.raise_for_status()
data = parse_redfin_response(response.text)
homes_raw = data.get("payload", {}).get("homes", [])
properties = []
for home in homes_raw:
properties.append({
"property_id": home.get("propertyId", ""),
"listing_id": home.get("listingId", ""),
"mls_id": home.get("mlsId", {}).get("value", ""),
"address": home.get("streetLine", {}).get("value", ""),
"city": home.get("cityStateZip", {}).get("value", ""),
"zip": home.get("zip", ""),
"lat": home.get("lat", 0),
"lon": home.get("lon", 0),
"price": home.get("price", {}).get("value", 0),
"beds": home.get("beds", 0),
"baths": home.get("baths", 0),
"sqft": home.get("sqFt", {}).get("value", 0),
"lot_sqft": home.get("lotSize", {}).get("value", 0),
"year_built": home.get("yearBuilt", {}).get("value", 0),
"days_on_market": home.get("dom", {}).get("value", 0),
"url": "https://www.redfin.com" + home.get("url", ""),
"status": home.get("listingType", ""),
"hoa_fee": home.get("hoaFee", 0),
"price_per_sqft": home.get("pricePerSqFt", {}).get("value", 0),
})
return properties
finally:
if own_client:
client.close()
# Example: Search for houses in Seattle's Capitol Hill neighborhood
seattle_polygon = "-122.327 47.609,-122.327 47.637,-122.295 47.637,-122.295 47.609"
with httpx.Client(headers=REDFIN_HEADERS, follow_redirects=True) as client:
listings = search_properties_by_polygon(
polygon=seattle_polygon,
market="seattle",
property_types="1,2", # houses and condos
min_beds=2,
client=client,
)
print(f"Found {len(listings)} listings")
for prop in listings[:5]:
print(f" {prop['address']}: ${prop['price']:,} | {prop['beds']}bd/{prop['baths']}ba | {prop['sqft']:,} sqft")
Property Detail Endpoint
Once you have a propertyId and listingId, fetch the complete property record:
def get_property_detail(
property_id: str,
listing_id: Optional[str] = None,
client: Optional[httpx.Client] = None,
) -> dict:
"""
Fetch complete property details including all listing info, school ratings,
walkability scores, and neighborhood data.
"""
params = {
"propertyId": property_id,
"accessLevel": 1,
}
if listing_id:
params["listingId"] = listing_id
own_client = client is None
if own_client:
client = httpx.Client(headers=REDFIN_HEADERS, follow_redirects=True, timeout=httpx.Timeout(connect=8.0, read=30.0))
try:
response = client.get(
"https://www.redfin.com/stingray/api/home/details/aboveTheFold",
params=params,
)
response.raise_for_status()
data = parse_redfin_response(response.text)
payload = data.get("payload", {})
main_info = payload.get("mainHouseInfo", {})
address_info = main_info.get("addressInfo", {})
price_info = main_info.get("priceInfo", {})
return {
"property_id": property_id,
"listing_id": listing_id,
"address": address_info.get("streetAddress", ""),
"city": address_info.get("city", ""),
"state": address_info.get("state", ""),
"zip": address_info.get("zip", ""),
"county": address_info.get("county", ""),
"current_price": price_info.get("displayedPrice", {}).get("displayValue", ""),
"price_per_sqft": price_info.get("pricePerSqFt", {}).get("displayValue", ""),
"status": payload.get("listingStatus", ""),
"listing_type": payload.get("listingType", ""),
"beds": main_info.get("numBeds", 0),
"baths": main_info.get("numBaths", 0),
"sqft": main_info.get("sqFt", {}).get("value", 0),
"lot_size": main_info.get("lotSqFt", {}).get("value", 0),
"year_built": main_info.get("yearBuilt", 0),
"hoa_fee": main_info.get("hoaFee", {}).get("value", 0),
"parking": main_info.get("parking", ""),
"description": payload.get("publicRemarksInfo", {}).get("remarks", ""),
"list_date": payload.get("listingDate", ""),
"days_on_market": payload.get("dom", 0),
"photo_count": payload.get("primaryPhotoDisplayLevel", 0),
"url": "https://www.redfin.com" + payload.get("url", ""),
"walk_score": payload.get("walkScore", 0),
"transit_score": payload.get("transitScore", 0),
"bike_score": payload.get("bikeScore", 0),
"school_rating": payload.get("elementarySchoolRating", 0),
}
finally:
if own_client:
client.close()
Price History Endpoint
The price history endpoint returns every listing and sale event for a property — invaluable for understanding a property's market trajectory.
def get_price_history(
property_id: str,
client: Optional[httpx.Client] = None,
) -> list[dict]:
"""
Fetch the complete price history for a property.
Returns all list/delist/sale events with dates and prices.
"""
params = {"propertyId": property_id, "accessLevel": 1}
own_client = client is None
if own_client:
client = httpx.Client(headers=REDFIN_HEADERS, follow_redirects=True, timeout=httpx.Timeout(connect=8.0, read=30.0))
try:
response = client.get(
"https://www.redfin.com/stingray/api/home/details/propertyParcelInfo",
params=params,
)
response.raise_for_status()
data = parse_redfin_response(response.text)
history_raw = data.get("payload", {}).get("priceHistory", [])
events = []
for event in history_raw:
events.append({
"date": event.get("date", ""),
"price": event.get("price", 0),
"price_display": event.get("priceDisplayValue", ""),
"event_type": event.get("eventDescription", ""), # "Listed", "Sold", "Price Drop", etc.
"price_change": event.get("priceChangeVal", 0),
"price_change_pct": event.get("priceChangePct", 0),
"mls_number": event.get("mlsId", {}).get("value", ""),
"source": event.get("source", ""),
"listing_agent": event.get("listingAgent", {}).get("agentName", ""),
"buyer_agent": event.get("buyerAgent", {}).get("agentName", ""),
})
return events
finally:
if own_client:
client.close()
Comparable Sales (Comps) Endpoint
Comps — recently sold similar properties — are the foundation of real estate valuation. Redfin's comps endpoint returns the homes used internally for their Redfin Estimate calculation.
def get_comparable_sales(
property_id: str,
listing_id: Optional[str] = None,
client: Optional[httpx.Client] = None,
) -> list[dict]:
"""
Fetch comparable recently sold properties for valuation.
Returns the same comps data used by Redfin's AVM.
"""
params = {
"propertyId": property_id,
"accessLevel": 1,
"pageType": 1,
}
if listing_id:
params["listingId"] = listing_id
own_client = client is None
if own_client:
client = httpx.Client(headers=REDFIN_HEADERS, follow_redirects=True, timeout=httpx.Timeout(connect=8.0, read=30.0))
try:
response = client.get(
"https://www.redfin.com/stingray/api/home/details/similarsAndNearby",
params=params,
)
response.raise_for_status()
data = parse_redfin_response(response.text)
payload = data.get("payload", {})
section_data = payload.get("sectionPreviewData", {}).get("sectionData", {})
comps = []
for section_key, section in section_data.items():
homes = section.get("homes", [])
for home in homes:
comps.append({
"property_id": home.get("propertyId", ""),
"address": home.get("address", {}).get("streetAddress", ""),
"city": home.get("address", {}).get("city", ""),
"sold_price": home.get("price", {}).get("value", 0),
"sold_date": home.get("soldDate", ""),
"beds": home.get("beds", 0),
"baths": home.get("baths", 0),
"sqft": home.get("sqFt", {}).get("value", 0),
"price_per_sqft": home.get("pricePerSqFt", {}).get("value", 0),
"distance_miles": home.get("distance", {}).get("value", 0),
"similarity_score": home.get("similarityScore", 0),
"url": "https://www.redfin.com" + home.get("url", ""),
"section": section_key, # "recentlySold", "similarHomes", etc.
})
return comps
finally:
if own_client:
client.close()
Neighborhood Market Statistics
Redfin exposes market aggregate statistics at the neighborhood level — median sale price, days on market distribution, list-to-sale price ratio, and inventory counts.
def get_neighborhood_stats(
region_id: str,
region_type: int = 6, # 6 = neighborhood, 2 = city, 4 = zip code
client: Optional[httpx.Client] = None,
) -> dict:
"""
Fetch market statistics for a neighborhood or region.
region_type: 2=city, 4=zip code, 5=county, 6=neighborhood, 7=metro area
To get region_id: search Redfin for a neighborhood, then find the region ID
in the URL (e.g., redfin.com/neighborhood/12345/WA/Seattle/Capitol-Hill)
"""
params = {
"region_id": region_id,
"region_type": region_type,
"tz": True,
"last_sold_within_years": 1,
}
own_client = client is None
if own_client:
client = httpx.Client(headers=REDFIN_HEADERS, follow_redirects=True, timeout=httpx.Timeout(connect=8.0, read=30.0))
try:
response = client.get(
"https://www.redfin.com/stingray/api/v1/market/marketInsightsPanel",
params=params,
)
response.raise_for_status()
data = parse_redfin_response(response.text)
stats = data.get("payload", {}).get("marketStats", {})
return {
"region_id": region_id,
"region_type": region_type,
"median_sale_price": stats.get("medianSalePrice", {}).get("value", 0),
"median_sale_price_yoy": stats.get("medianSalePriceYoY", 0),
"median_days_on_market": stats.get("medianDom", 0),
"median_list_price": stats.get("medianListPrice", {}).get("value", 0),
"sale_to_list_ratio": stats.get("saleToListRatio", 0),
"homes_sold_count": stats.get("homesSoldCount", 0),
"active_listings": stats.get("activeListings", 0),
"months_of_supply": stats.get("monthsOfSupply", 0),
"pct_homes_above_list": stats.get("pctHomesAboveList", 0),
"pct_homes_below_list": stats.get("pctHomesBelowList", 0),
"avg_sale_price": stats.get("avgSalePrice", {}).get("value", 0),
}
finally:
if own_client:
client.close()
CSV Export Endpoint
Redfin provides a CSV export of search results accessible without login. This is the fastest way to get flat tabular data for a region:
def download_listings_csv(
polygon: str,
market: str = "seattle",
status: int = 9,
property_types: str = "1,2,3",
output_path: str = "listings.csv",
client: Optional[httpx.Client] = None,
) -> int:
"""
Download listings as CSV directly from Redfin's export endpoint.
Limited to 350 rows per request. Returns number of rows written.
"""
params = {
"al": 1,
"market": market,
"num_homes": 350,
"poly": polygon,
"sf": "1,2,3,5,6,7",
"status": status,
"uipt": property_types,
"v": 8,
}
own_client = client is None
if own_client:
client = httpx.Client(headers=REDFIN_HEADERS, follow_redirects=True, timeout=httpx.Timeout(connect=8.0, read=60.0))
try:
response = client.get(
"https://www.redfin.com/stingray/api/gis-csv",
params=params,
)
response.raise_for_status()
with open(output_path, "wb") as f:
f.write(response.content)
# Count rows (subtract 1 for header)
row_count = response.text.count("\n") - 1
print(f"Downloaded {row_count} listings to {output_path}")
return row_count
finally:
if own_client:
client.close()
Anti-Bot Measures: What You Are Dealing With
Redfin runs Akamai Bot Manager at the CDN edge. Here is what Akamai analyzes for each request:
TLS fingerprint (JA3/JA4). The TLS ClientHello message contains the cipher suite list, extension ordering, and supported groups. Python's ssl library generates a fingerprint that is well-known to Akamai's database as non-browser traffic. Chrome generates a different fingerprint. This is the most fundamental detection signal, and it operates at the network layer before Redfin's servers ever see the request.
HTTP/2 frame priorities. HTTP/2 connections send SETTINGS and PRIORITY frames with values that vary by client. Chrome's HTTP/2 implementation sends specific SETTINGS values (initial window size, header table size, etc.) that differ from Python's httpcore implementation. Akamai checks these values.
Header ordering and casing. HTTP/2 encodes headers as lowercase; HTTP/1.1 is case-insensitive. Akamai checks the ordering of headers and whether certain headers appear in the expected sequence for the claimed browser.
Cookie presence and values. Redfin sets RF_BROWSER_ID and Akamai sets ak_bmsc on the first page load. These cookies encode browser fingerprint data. Subsequent requests that arrive without these cookies (or with invalid/expired values) trigger detection. A scraper that goes directly to the API endpoints without first obtaining these cookies via a homepage visit will have a much lower success rate.
JavaScript sensor data. The Akamai collector script (/_/smap) runs 500ms of browser fingerprint detection — checking navigator.plugins, canvas rendering, audio context behavior, WebGL renderer, and dozens of other browser APIs. The results are encoded and sent as a hidden form field or cookie. Requests that arrive without this sensor data are flagged.
IP reputation. Datacenter IP ranges (AWS, GCP, DigitalOcean, Hetzner, all major VPN providers) are pre-blocked at Akamai's edge before any fingerprint analysis happens. The block is invisible — Akamai serves a fake success response or a 403, depending on configuration.
Bypassing Bot Detection with Residential Proxies
The practical approach for sustained Redfin access is:
- Route through residential IPs to pass the IP reputation check
- Use
curl_cffito match Chrome's TLS fingerprint - Seed the session with a homepage visit to collect legitimate Akamai cookies
- Add realistic delays between requests
For residential proxies, ThorData provides geo-targeted residential IPs. For Redfin specifically, matching your proxy's location to the market you're querying (e.g., a Seattle-area IP for Seattle listings) reduces anomaly scoring because it matches expected user behavior.
import httpx
try:
from curl_cffi import requests as curl_requests
HAS_CURL_CFFI = True
except ImportError:
HAS_CURL_CFFI = False
import time
import random
import json
class RedfinClient:
"""
Production Redfin scraping client with anti-bot mitigation.
Uses curl_cffi for Chrome TLS impersonation when available.
"""
BASE_URL = "https://www.redfin.com"
HEADERS = {
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/134.0.0.0 Safari/537.36",
"Accept": "application/json, text/plain, */*",
"Accept-Language": "en-US,en;q=0.9",
"Accept-Encoding": "gzip, deflate, br",
"sec-ch-ua": '"Chromium";v="134", "Google Chrome";v="134", "Not:A-Brand";v="99"',
"sec-ch-ua-mobile": "?0",
"sec-ch-ua-platform": '"macOS"',
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-origin",
}
def __init__(
self,
proxy_url: Optional[str] = None,
use_curl_cffi: bool = True,
warm_up: bool = True,
):
self.proxy_url = proxy_url
self.use_curl_cffi = use_curl_cffi and HAS_CURL_CFFI
self._session = None
self._last_request_time = 0.0
if self.use_curl_cffi:
self._session = curl_requests.Session(impersonate="chrome134")
if proxy_url:
self._session.proxies = {"http": proxy_url, "https": proxy_url}
self._session.headers.update(self.HEADERS)
else:
self._httpx = httpx.Client(
headers=self.HEADERS,
proxy=proxy_url,
http2=True,
follow_redirects=True,
timeout=httpx.Timeout(connect=8.0, read=30.0),
)
if warm_up:
self._warm_up_session()
def _warm_up_session(self):
"""Visit Redfin homepage to collect Akamai cookies before making API calls."""
try:
homepage_headers = {
**self.HEADERS,
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
"Sec-Fetch-Dest": "document",
"Sec-Fetch-Mode": "navigate",
"Sec-Fetch-Site": "none",
"Sec-Fetch-User": "?1",
}
if self.use_curl_cffi:
self._session.get(
"https://www.redfin.com/",
headers=homepage_headers,
timeout=15,
)
else:
self._httpx.get(
"https://www.redfin.com/",
headers=homepage_headers,
)
# Pause to simulate human reading the page
time.sleep(random.uniform(1.5, 3.5))
except Exception as e:
print(f"Warm-up failed (continuing anyway): {e}")
def _rate_limit(self, min_interval: float = 1.5):
"""Enforce minimum time between requests with jitter."""
now = time.time()
elapsed = now - self._last_request_time
if elapsed < min_interval:
jitter = random.uniform(0, min_interval * 0.3)
time.sleep(min_interval - elapsed + jitter)
self._last_request_time = time.time()
def get(self, endpoint: str, params: dict = None) -> dict:
"""Make an authenticated GET request to a Redfin stingray endpoint."""
self._rate_limit()
url = self.BASE_URL + endpoint
try:
if self.use_curl_cffi:
response = self._session.get(url, params=params, timeout=30)
response.raise_for_status()
text = response.text
else:
response = self._httpx.get(url, params=params)
response.raise_for_status()
text = response.text
return parse_redfin_response(text)
except Exception as e:
raise RuntimeError(f"Redfin API error for {endpoint}: {e}")
def close(self):
if hasattr(self, "_httpx"):
self._httpx.close()
def __enter__(self):
return self
def __exit__(self, *args):
self.close()
# ThorData proxy setup for residential IP routing
# Sign up at https://thordata.partnerstack.com/partner/0a0x4nzh
def get_thordata_proxy(
username: str,
password: str,
country: str = "US",
state: Optional[str] = None,
) -> str:
"""Build a ThorData residential proxy URL with geo-targeting."""
user_parts = [username, f"country-{country}"]
if state:
user_parts.append(f"state-{state.lower()}")
encoded_user = "-".join(user_parts)
return f"http://{encoded_user}:{password}@rotating.thordata.net:9080"
# Usage
proxy = get_thordata_proxy(
username="your_username",
password="your_password",
country="US",
state="WA", # Washington state for Seattle searches
)
with RedfinClient(proxy_url=proxy) as client:
data = client.get("/stingray/api/gis", params={
"al": 1,
"market": "seattle",
"poly": "-122.327 47.609,-122.327 47.637,-122.295 47.637,-122.295 47.609",
"status": 9,
"uipt": "1,2",
"v": 8,
})
homes = data.get("payload", {}).get("homes", [])
print(f"Found {len(homes)} listings")
Use Case 1: Investment Property Screening
Screen active listings for investment potential using cap rate and rental yield calculations. Combines active listing data with price history to identify undervalued properties.
import sqlite3
import json
from datetime import datetime, timedelta
def setup_investment_db(db_path: str = "investment_screening.db") -> sqlite3.Connection:
conn = sqlite3.connect(db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS screened_properties (
id INTEGER PRIMARY KEY AUTOINCREMENT,
property_id TEXT UNIQUE,
address TEXT,
city TEXT,
zip TEXT,
price INTEGER,
beds INTEGER,
baths REAL,
sqft INTEGER,
year_built INTEGER,
days_on_market INTEGER,
price_per_sqft REAL,
walk_score INTEGER,
estimated_rent INTEGER,
estimated_gross_yield REAL,
price_drop_count INTEGER,
total_price_drop_pct REAL,
url TEXT,
screened_at TEXT
)
""")
conn.execute("CREATE INDEX IF NOT EXISTS idx_yield ON screened_properties(estimated_gross_yield)")
conn.commit()
return conn
def estimate_monthly_rent(beds: int, baths: float, sqft: int, zip_code: str) -> int:
"""
Rough rent estimate based on property characteristics.
In production, integrate with a rent estimation API or Zillow's Zestimate.
"""
base_rent_per_sqft = {
"98101": 2.80, # Seattle downtown
"98103": 2.40, # Seattle Fremont
"98115": 2.10, # Seattle NE
}.get(zip_code, 1.80) # default for unknown zip
if sqft <= 0:
sqft = beds * 500 # rough estimate
base = sqft * base_rent_per_sqft
# Adjust for bedroom count
bed_premium = {1: 0.9, 2: 1.0, 3: 1.05, 4: 1.08}.get(beds, 1.0)
return int(base * bed_premium)
def screen_properties_for_investment(
polygon: str,
market: str,
min_gross_yield_pct: float = 5.0,
max_days_on_market: int = 90,
proxy_url: Optional[str] = None,
db_path: str = "investment_screening.db",
) -> list[dict]:
"""
Fetch active listings, calculate investment metrics, filter by criteria.
"""
conn = setup_investment_db(db_path)
qualifying = []
with RedfinClient(proxy_url=proxy_url) as client:
# Get active listings
listings = search_properties_by_polygon(polygon, market, client=client)
print(f"Screening {len(listings)} properties...")
time.sleep(random.uniform(2, 4))
for prop in listings:
# Skip properties that don't meet basic criteria
if prop["days_on_market"] > max_days_on_market:
continue
if prop["price"] <= 0 or prop["beds"] < 1:
continue
# Estimate rent and yield
monthly_rent = estimate_monthly_rent(
prop["beds"], prop["baths"], prop["sqft"], prop["zip"]
)
annual_rent = monthly_rent * 12
gross_yield = (annual_rent / prop["price"]) * 100 if prop["price"] > 0 else 0
if gross_yield < min_gross_yield_pct:
continue
# Fetch price history to check for prior price reductions
time.sleep(random.uniform(1.5, 3.0))
history = []
try:
history_data = client.get(
"/stingray/api/home/details/propertyParcelInfo",
params={"propertyId": prop["property_id"], "accessLevel": 1},
)
history = history_data.get("payload", {}).get("priceHistory", [])
except Exception as e:
print(f"Price history failed for {prop['property_id']}: {e}")
# Count price drops in current listing
price_drops = [e for e in history if "Price Drop" in e.get("eventDescription", "")]
total_drop_pct = sum(abs(e.get("priceChangePct", 0)) for e in price_drops)
record = {
**prop,
"estimated_monthly_rent": monthly_rent,
"estimated_gross_yield": round(gross_yield, 2),
"price_drop_count": len(price_drops),
"total_price_drop_pct": round(total_drop_pct, 1),
}
qualifying.append(record)
# Save to DB
conn.execute("""
INSERT OR REPLACE INTO screened_properties
(property_id, address, city, zip, price, beds, baths, sqft, year_built,
days_on_market, price_per_sqft, walk_score, estimated_rent,
estimated_gross_yield, price_drop_count, total_price_drop_pct, url, screened_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
prop["property_id"], prop["address"], prop["city"], prop["zip"],
prop["price"], prop["beds"], prop["baths"], prop["sqft"],
prop["year_built"], prop["days_on_market"], prop["price_per_sqft"],
prop.get("walk_score", 0), monthly_rent, round(gross_yield, 2),
len(price_drops), round(total_drop_pct, 1), prop["url"],
datetime.now().isoformat(),
))
conn.commit()
conn.close()
# Sort by yield descending
qualifying.sort(key=lambda x: x["estimated_gross_yield"], reverse=True)
return qualifying
Use Case 2: Price Drop Monitoring
Monitor a set of tracked properties for price reductions — useful for buyers watching specific listings or investment researchers tracking motivated sellers.
import json
def monitor_tracked_properties(
tracked_property_ids: list[str],
previous_prices_file: str = "tracked_prices.json",
alert_threshold_pct: float = 2.0,
proxy_url: Optional[str] = None,
) -> list[dict]:
"""
Check a list of tracked property IDs for price changes.
Returns properties where price dropped by at least alert_threshold_pct.
"""
# Load previous prices
try:
with open(previous_prices_file) as f:
previous_prices = json.load(f)
except FileNotFoundError:
previous_prices = {}
alerts = []
current_prices = {}
with RedfinClient(proxy_url=proxy_url) as client:
for property_id in tracked_property_ids:
time.sleep(random.uniform(2.0, 4.0))
try:
data = client.get(
"/stingray/api/home/details/aboveTheFold",
params={"propertyId": property_id, "accessLevel": 1},
)
payload = data.get("payload", {})
price_info = payload.get("mainHouseInfo", {}).get("priceInfo", {})
price_raw = price_info.get("displayedPrice", {}).get("value", 0)
address = payload.get("mainHouseInfo", {}).get("addressInfo", {}).get("streetAddress", property_id)
status = payload.get("listingStatus", "")
current_prices[property_id] = {
"price": price_raw,
"address": address,
"status": status,
"checked_at": datetime.now().isoformat(),
}
# Check for price drop
prev = previous_prices.get(property_id, {})
prev_price = prev.get("price", 0)
if prev_price > 0 and price_raw > 0 and price_raw < prev_price:
drop_pct = (prev_price - price_raw) / prev_price * 100
if drop_pct >= alert_threshold_pct:
alerts.append({
"property_id": property_id,
"address": address,
"previous_price": prev_price,
"current_price": price_raw,
"drop_amount": prev_price - price_raw,
"drop_pct": round(drop_pct, 1),
"status": status,
})
print(f"PRICE DROP: {address} — ${prev_price:,} -> ${price_raw:,} ({drop_pct:.1f}% off)")
# Check for status changes (e.g., went under contract)
prev_status = prev.get("status", "")
if prev_status and status != prev_status:
print(f"STATUS CHANGE: {address} — {prev_status} -> {status}")
except Exception as e:
print(f"Error checking {property_id}: {e}")
current_prices[property_id] = previous_prices.get(property_id, {})
# Save current prices for next run
merged = {**previous_prices, **current_prices}
with open(previous_prices_file, "w") as f:
json.dump(merged, f, indent=2)
return alerts
Use Case 3: Market Statistics Dashboard
Collect neighborhood-level market stats across multiple regions for a market overview dashboard.
import csv
from typing import Optional
SEATTLE_NEIGHBORHOODS = {
"Capitol Hill": "12345",
"Fremont": "12346",
"Ballard": "12347",
"Queen Anne": "12348",
"Madison Park": "12349",
"Columbia City": "12350",
"West Seattle": "12351",
"South Lake Union": "12352",
}
def collect_market_stats(
neighborhoods: dict,
output_csv: str = "market_stats.csv",
proxy_url: Optional[str] = None,
) -> list[dict]:
"""Collect market statistics for a set of neighborhoods."""
stats_list = []
with RedfinClient(proxy_url=proxy_url, warm_up=True) as client:
for name, region_id in neighborhoods.items():
time.sleep(random.uniform(3.0, 6.0)) # Longer pause for stats endpoint
try:
stats = get_neighborhood_stats(region_id, region_type=6, client=None)
stats["neighborhood"] = name
stats["collected_at"] = datetime.now().isoformat()
stats_list.append(stats)
print(f"{name}: median ${stats['median_sale_price']:,} | "
f"{stats['median_days_on_market']} DOM | "
f"{stats['sale_to_list_ratio']:.1%} list/sale")
except Exception as e:
print(f"Error collecting stats for {name}: {e}")
# Write to CSV
if stats_list:
fieldnames = list(stats_list[0].keys())
with open(output_csv, "w", newline="") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(stats_list)
print(f"Saved stats to {output_csv}")
return stats_list
Output Schema
A complete property record from the search + detail endpoints:
{
"property_id": "12345678",
"listing_id": "87654321",
"mls_id": "NW24061234",
"address": "1234 E Pine St",
"city": "Seattle, WA 98122",
"state": "WA",
"zip": "98122",
"county": "King",
"lat": 47.6152,
"lon": -122.3192,
"current_price": "$895,000",
"price": 895000,
"price_per_sqft": 748,
"beds": 3,
"baths": 2.0,
"sqft": 1196,
"lot_size": 2800,
"year_built": 1924,
"hoa_fee": 0,
"days_on_market": 12,
"status": "Active",
"listing_type": "MLS",
"parking": "None",
"list_date": "2026-03-19",
"walk_score": 95,
"transit_score": 72,
"bike_score": 88,
"school_rating": 7,
"url": "https://www.redfin.com/WA/Seattle/1234-E-Pine-St-98122/home/12345678",
"price_history": [
{
"date": "Mar 19, 2026",
"price": 895000,
"event_type": "Listed",
"price_change": 0,
"price_change_pct": 0
}
],
"comps": [
{
"address": "1111 E Olive St",
"sold_price": 871000,
"sold_date": "Feb 14, 2026",
"sqft": 1140,
"distance_miles": 0.3,
"similarity_score": 0.87
}
]
}
Error Handling and Reliability
import time
import random
from typing import Optional
class RedfinAPIError(Exception):
pass
class RedfinBlockedError(RedfinAPIError):
pass
class RedfinNotFoundError(RedfinAPIError):
pass
def safe_api_call(
client: RedfinClient,
endpoint: str,
params: dict,
max_retries: int = 4,
) -> Optional[dict]:
"""
Make a Redfin API call with comprehensive error handling and retry logic.
Returns None if all retries are exhausted.
"""
for attempt in range(max_retries):
try:
return client.get(endpoint, params=params)
except RuntimeError as e:
error_str = str(e)
# 404 — property or listing not found (no retry)
if "404" in error_str:
raise RedfinNotFoundError(f"Not found: {endpoint} {params}")
# 403 or Akamai block — need fresh session/IP
if "403" in error_str or "blocked" in error_str.lower():
if attempt < max_retries - 1:
wait = 30 * (2 ** attempt) # 30s, 60s, 120s
print(f"Blocked on attempt {attempt + 1}, waiting {wait}s before retry")
time.sleep(wait)
# Signal to caller that a new client may be needed
raise RedfinBlockedError(f"Akamai block: {error_str}")
# Rate limit (429)
if "429" in error_str:
wait = 60 * (attempt + 1) # 60s, 120s, 180s
print(f"Rate limited, waiting {wait}s")
time.sleep(wait)
continue
# Transient errors — retry with backoff
if attempt < max_retries - 1:
wait = 5 * (2 ** attempt) # 5s, 10s, 20s
print(f"Error on attempt {attempt + 1}/{max_retries}: {e}. Retry in {wait}s")
time.sleep(wait)
else:
print(f"All {max_retries} attempts failed for {endpoint}: {e}")
return None
return None
Practical Notes for Production Use
Region IDs for the stats endpoint are not exposed in the main search API. You find them by navigating to a Redfin neighborhood page and extracting the ID from the URL path (e.g., /neighborhood/12345/WA/Seattle/Capitol-Hill — 12345 is the region ID). Build a lookup table for your target markets.
The 350-row CSV limit can be worked around by subdividing your target bounding box into smaller tiles and making one request per tile. A city-sized area typically requires 4-16 tiles at this limit to get full coverage.
Price history timestamps are returned as display strings ("Mar 19, 2026") rather than ISO dates. Use Python's dateutil.parser.parse() or explicit strptime with fallback parsing to normalize these to timestamps.
Redfin's stingray API has been stable for years, but treat it as undocumented infrastructure. The {}&& prefix, endpoint paths, and core response structure have been consistent, but specific field names within payloads change occasionally. Add null-safe access (.get() with defaults) for every field you read, and emit a warning when expected fields are missing rather than letting KeyError crash your pipeline.
Request spacing — for light-duty use (a few hundred properties per day), 2-4 second delays between requests is fine. For heavier workloads, use the concurrent pipeline pattern with a semaphore and randomized delays. Never make requests faster than one per second to a single domain.
Redfin's stingray API is one of the most accessible and data-rich real estate APIs available for those willing to work with undocumented endpoints. The data quality is high — Redfin aggregates MLS data directly as a licensed brokerage, so their listings are current and accurate. Combined with ThorData residential proxies for Akamai bypass and curl_cffi for TLS fingerprint matching, you can build production pipelines that reliably extract property data at scale for investment screening, market analysis, and real-time price tracking.