Build a small Zillow price-drop monitor with one Python file. The script requests a saved for-sale search, stores the latest price for each ZPID in SQLite and prints a duplicate-safe alert when a known price falls.
This guide builds a small Zillow price-drop monitor in one Python file. It requests the first page of a saved for-sale search, stores each price in SQLite and prints an alert when the same ZPID returns with a lower price.
The monitor is deliberately small. It runs one request at a time and treats a missing property only as absent from the current first page. It does not prove that a home was sold or removed from Zillow.
What you need
You need Python 3.10 or newer and access to the Real Estate Zillow.com API on RapidAPI. Start with one location and one page. That makes the first version easier to understand and keeps request use predictable.
The code uses the provider's GET /v1/search/sale route. It sends location_or_rid, page, beds_min, price_max and sort. API contracts can change, so compare these names with the RapidAPI playground before running the monitor.
Step 1: create the project
Open PowerShell and run these commands one line at a time:
mkdir zillow-price-monitor
cd zillow-price-monitor
py -m venv .venv
.\.venv\Scripts\Activate.ps1
python -m pip install --upgrade pip
python -m pip install requestsPython already includes SQLite, so you do not install a separate database package. The standard sqlite3 module creates a local file when the monitor first runs.
Step 2: set the API key
Keep the RapidAPI key outside the Python file. These PowerShell commands ask for the key without displaying it and place it in the current terminal session:
$secureKey = Read-Host "RapidAPI key" -AsSecureString
$env:RAPIDAPI_KEY = [System.Net.NetworkCredential]::new("", $secureKey).Password
$env:ZILLOW_LOCATION_OR_RID = "12700"12700 is the example region used by the provider for Miami. Replace it with the location or region ID you have checked for your search. Closing this PowerShell window removes the environment variables.
Step 3: add the complete Python monitor
Create zillow_monitor.py and paste in this code:
from __future__ import annotations
import hashlib
import os
import re
import sqlite3
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
from pathlib import Path
from typing import Any
import requests
API_URL = "https://real-estate-zillow-com.p.rapidapi.com/v1/search/sale"
API_HOST = "real-estate-zillow-com.p.rapidapi.com"
DATABASE_PATH = Path("zillow_monitor.sqlite3")
def utc_now() -> str:
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def price_to_cents(value: Any) -> int | None:
if value is None or isinstance(value, bool):
return None
cleaned = re.sub(r"[^\d.\-]", "", str(value).strip())
if not cleaned:
return None
try:
amount = Decimal(cleaned)
except InvalidOperation:
return None
if not amount.is_finite() or amount < 0:
return None
return int(
(amount * 100).quantize(Decimal("1"), rounding=ROUND_HALF_UP)
)
def find_listing_list(node: Any) -> list[dict[str, Any]]:
if isinstance(node, list):
dictionaries = [item for item in node if isinstance(item, dict)]
if any(item.get("zpid") for item in dictionaries):
return dictionaries
for item in dictionaries:
found = find_listing_list(item)
if found:
return found
if isinstance(node, dict):
for value in node.values():
found = find_listing_list(value)
if found:
return found
return []
def fetch_listings() -> tuple[list[dict[str, Any]], str]:
api_key = os.environ.get("RAPIDAPI_KEY")
if not api_key:
raise RuntimeError("RAPIDAPI_KEY is not set")
headers = {
"X-RapidAPI-Key": api_key,
"X-RapidAPI-Host": API_HOST,
}
params = {
"location_or_rid": os.getenv("ZILLOW_LOCATION_OR_RID", "12700"),
"page": 1,
"beds_min": 2,
"price_max": 500000,
"sort": "newest",
}
response = requests.get(
API_URL,
headers=headers,
params=params,
timeout=(5, 30),
)
response.raise_for_status()
payload = response.json()
if (
not isinstance(payload, dict)
or payload.get("status") != 200
or not isinstance(payload.get("data"), dict)
):
raise RuntimeError("The API returned an unexpected JSON wrapper")
observed_at = utc_now()
by_zpid: dict[str, dict[str, Any]] = {}
for item in find_listing_list(payload["data"]):
zpid = str(item.get("zpid") or "").strip()
if not zpid:
continue
by_zpid[zpid] = {
"zpid": zpid,
"price_cents": price_to_cents(item.get("price")),
"detail_url": item.get("detailUrl"),
"observed_at_utc": observed_at,
}
if not by_zpid:
raise RuntimeError(
"No listings were found. Stop here instead of marking every saved row missing."
)
return list(by_zpid.values()), observed_at
def open_database(path: Path = DATABASE_PATH) -> sqlite3.Connection:
connection = sqlite3.connect(path)
connection.row_factory = sqlite3.Row
connection.executescript(
"""
CREATE TABLE IF NOT EXISTS listing_state (
zpid TEXT PRIMARY KEY,
price_cents INTEGER,
detail_url TEXT,
observed_at_utc TEXT NOT NULL,
is_missing INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS alert_log (
fingerprint TEXT PRIMARY KEY,
zpid TEXT NOT NULL,
old_price_cents INTEGER NOT NULL,
new_price_cents INTEGER NOT NULL,
created_at_utc TEXT NOT NULL
);
"""
)
return connection
def alert_fingerprint(zpid: str, old_price: int, new_price: int) -> str:
text = f"{zpid}:{old_price}:{new_price}"
return hashlib.sha256(text.encode("utf-8")).hexdigest()
def save_snapshot(
connection: sqlite3.Connection,
listings: list[dict[str, Any]],
observed_at: str,
) -> tuple[dict[str, int], list[dict[str, Any]]]:
previous = {
row["zpid"]: row
for row in connection.execute("SELECT * FROM listing_state").fetchall()
}
current_ids = {listing["zpid"] for listing in listings}
counts = {"new": 0, "changed": 0, "unchanged": 0, "missing": 0}
price_drops: list[dict[str, Any]] = []
with connection:
for listing in listings:
old = previous.get(listing["zpid"])
if old is None:
counts["new"] += 1
else:
changed = (
bool(old["is_missing"])
or old["price_cents"] != listing["price_cents"]
or old["detail_url"] != listing["detail_url"]
)
counts["changed" if changed else "unchanged"] += 1
old_price = old["price_cents"]
new_price = listing["price_cents"]
if (
not old["is_missing"]
and old_price is not None
and new_price is not None
and new_price < old_price
):
fingerprint = alert_fingerprint(
listing["zpid"], old_price, new_price
)
inserted = connection.execute(
"""
INSERT OR IGNORE INTO alert_log
(fingerprint, zpid, old_price_cents,
new_price_cents, created_at_utc)
VALUES (?, ?, ?, ?, ?)
""",
(
fingerprint,
listing["zpid"],
old_price,
new_price,
observed_at,
),
).rowcount
if inserted:
price_drops.append(
{
"zpid": listing["zpid"],
"old_price_cents": old_price,
"new_price_cents": new_price,
"detail_url": listing["detail_url"],
}
)
connection.execute(
"""
INSERT INTO listing_state
(zpid, price_cents, detail_url, observed_at_utc, is_missing)
VALUES (?, ?, ?, ?, 0)
ON CONFLICT(zpid) DO UPDATE SET
price_cents = excluded.price_cents,
detail_url = excluded.detail_url,
observed_at_utc = excluded.observed_at_utc,
is_missing = 0
""",
(
listing["zpid"],
listing["price_cents"],
listing["detail_url"],
listing["observed_at_utc"],
),
)
for zpid, old in previous.items():
if zpid not in current_ids and not old["is_missing"]:
counts["missing"] += 1
connection.execute(
"""
UPDATE listing_state
SET is_missing = 1, observed_at_utc = ?
WHERE zpid = ?
""",
(observed_at, zpid),
)
return counts, price_drops
def dollars(cents: int) -> str:
return f"${cents / 100:,.2f}"
def main() -> None:
try:
listings, observed_at = fetch_listings()
except (requests.RequestException, ValueError, RuntimeError) as error:
raise SystemExit(f"Monitor stopped without changing SQLite: {error}")
connection = open_database()
try:
counts, price_drops = save_snapshot(
connection, listings, observed_at
)
finally:
connection.close()
for drop in price_drops:
print(
f"PRICE DROP {drop['zpid']}: "
f"{dollars(drop['old_price_cents'])} -> "
f"{dollars(drop['new_price_cents'])} | "
f"{drop['detail_url'] or 'URL unavailable'}"
)
print(f"{observed_at} | {counts} | alerts={len(price_drops)}")
if __name__ == "__main__":
main()Step 4: run it twice
Run the script from the activated virtual environment:
python .\zillow_monitor.pyThe first successful run creates zillow_monitor.sqlite3. Every listing is new, so there is no price-drop alert yet. Run the script again later with the same search settings. A normal summary looks like this:
2026-09-22T08:00:00+00:00 | {'new': 0, 'changed': 1, 'unchanged': 18, 'missing': 1} | alerts=1This output is illustrative. The count depends on the live search response.
What SQLite stores
listing_state keeps the latest row for each ZPID. A ZPID is stored as text because it is an identifier, not a number you calculate with. The price is stored as integer cents, so the comparison does not use binary floating-point values.
alert_log keeps a SHA-256 fingerprint made from ZPID, old price and new price. Its primary key prevents the same price transition from creating the same alert twice.
The exact price-drop rule
The monitor prints an alert only when the previous and current prices both exist, the property was present in the previous snapshot and the new price is lower. A missing or unreadable price becomes None. It never becomes zero.
A property that returns after being marked missing does not create an immediate price-drop alert. The code first establishes a fresh comparison point. This avoids comparing a new observation with an old price from an uncertain gap.
What missing means in this version
This beginner version requests only page 1. Therefore, is_missing=1 means the ZPID was not on the current first page. The listing may still exist on another page or under different filters.
The code refuses to update SQLite when the response has no recognizable listing objects. That protects the saved state from a broken response, an unexpected JSON shape or an empty challenge page.
Schedule the monitor
After manual runs work, use Windows Task Scheduler or cron to call the script. Keep the API key in the scheduler's protected environment. Do not put it in the Python file, command arguments or logs.
One page every six hours means four API requests per day, or about 120 requests in a 30-day month. Check the current plan allowance in RapidAPI before choosing an interval. If a request fails, let the next scheduled run try again instead of starting a fast retry loop.
Common errors
RAPIDAPI_KEY is not set
Set the key in the same terminal window that runs Python. Environment variables set in another window are not copied into the current process.
The request returns an HTTP error
Check the subscription, host name, endpoint and parameters in the RapidAPI playground. Do not print the request headers while debugging because they contain the API key.
No listings were found
Inspect a sanitized copy of the JSON response from your own subscription. The provider documents fields such as zpid, price and detailUrl, but the enclosing JSON can evolve. Update find_listing_list() only after you see the real response.
SQLite says the database is locked
Make sure only one scheduled copy runs at a time. This tutorial is designed for one local process, not parallel workers.
When you need more than page 1
Add pagination only after the one-page version works. Fetch every configured page before opening SQLite. Mark records missing only when the complete fetch succeeds. The companion Zillow fields and endpoints guide explains which route and parameters belong to each search type.