Real Estate Market Data Pipeline

Project Overview

Real estate markets move fast. A property listed today might see three price drops in a week, get relisted under a different agent, or vanish from one platform while still appearing on another. For investors, analysts, and proptech companies, having a reliable, up-to-date view of the market isn’t a nice-to-have — it’s the foundation of every decision they make.

This pipeline was built to solve a specific problem: there was no single, normalized dataset covering property listings across multiple public sources with full historical tracking. Commercial APIs existed, but they were expensive, had inconsistent coverage, and offered no visibility into how data was collected or transformed.

The system collects publicly available property listing data from multiple platforms, normalizes it into a consistent schema, deduplicates across sources, and tracks changes over time. The result is a structured dataset that powers downstream analysis: pricing trend models, days-on-market calculations, neighborhood-level supply metrics, and anomaly detection for mispriced listings.

What started as a weekend experiment with a single Scrapy spider grew into a production system processing over 50,000 records daily across multiple geographic markets.

Architecture

The pipeline follows a staged architecture where each layer has a single responsibility. Raw data flows through collection, normalization, deduplication, and enrichment before landing in the analytical store.

flowchart LR
    subgraph Sources["Data Sources"]
        S1["Listing Site A"]
        S2["Listing Site B"]
        S3["Public APIs"]
    end

    subgraph Collection["Scraping Layer"]
        SC["Scrapy Cluster"]
        PW["Playwright\nFallback"]
        RQ["Redis\nRequest Queue"]
    end

    subgraph Processing["Processing"]
        NM["Normalizer"]
        DD["Deduplication\nEngine"]
        CD["Change\nDetection"]
        EN["Enrichment\n(Geocoding, etc.)"]
    end

    subgraph Storage["Storage"]
        PG[("PostgreSQL")]
        RS["Redis Cache"]
    end

    subgraph Analysis["Analysis"]
        PD["Pandas\nPipelines"]
        RP["Reports &\nExports"]
    end

    S1 --> SC
    S2 --> SC
    S3 --> SC
    SC --> PW
    SC --> RQ
    RQ --> NM
    NM --> DD
    DD --> CD
    CD --> EN
    EN --> PG
    PG --> RS
    PG --> PD
    PD --> RP

Each component runs in its own Docker container. The scraping layer and processing layer scale independently — during peak hours (when listings are most frequently updated), the scraping cluster runs more workers while the processing pipeline stays steady.

Technical Deep Dive

Spider Design and Request Handling

The scraping layer is built on Scrapy with a few important additions. Each target source gets its own spider class that inherits from a base spider providing common functionality: request fingerprinting, robots.txt enforcement, and structured logging.

The trickiest part was handling sites that rely heavily on client-side rendering. About 30% of listing pages returned near-empty HTML from a standard HTTP request because the actual content was loaded via JavaScript. Rather than running every request through a headless browser (which would have been prohibitively slow at scale), the system uses a two-pass approach.

import scrapy
from scrapy import Request
from itemloaders import ItemLoader
from pipeline.items import PropertyItem


class BasePropertySpider(scrapy.Spider):
    """Base spider with robots.txt compliance and JS fallback."""

    custom_settings = {
        "ROBOTSTXT_OBEY": True,
        "DOWNLOAD_DELAY": 1.0,
        "CONCURRENT_REQUESTS_PER_DOMAIN": 1,
        "USER_AGENT": "ScrapemBot/1.0 (+https://scrapem.com/bot-info)",
    }

    # Subclasses define these
    source_name: str = ""
    js_render_selectors: list[str] = []

    def parse(self, response):
        """First pass: try static HTML extraction."""
        loader = ItemLoader(item=PropertyItem(), response=response)
        loader.add_value("source", self.source_name)
        loader.add_value("source_url", response.url)

        # Check if critical content is present in static HTML
        if self._has_required_content(response):
            yield from self.parse_listing(response, loader)
        else:
            # Content missing — fall back to Playwright rendering
            self.logger.info(f"JS fallback triggered for {response.url}")
            yield Request(
                url=response.url,
                callback=self.parse_rendered,
                meta={"playwright": True, "playwright_page_methods": [
                    {"method": "wait_for_selector", "args": [self.js_render_selectors[0]]},
                ]},
                dont_filter=True,
            )

    def _has_required_content(self, response) -> bool:
        """Check if the static HTML contains the data we need."""
        for selector in self.js_render_selectors:
            if not response.css(selector):
                return False
        return True

    def parse_listing(self, response, loader):
        """Override in subclass to extract listing-specific fields."""
        raise NotImplementedError

    def parse_rendered(self, response):
        """Second pass: extract from JS-rendered page."""
        loader = ItemLoader(item=PropertyItem(), response=response)
        loader.add_value("source", self.source_name)
        loader.add_value("source_url", response.url)
        yield from self.parse_listing(response, loader)

This fallback mechanism cut Playwright usage to under 15% of total requests, which kept resource consumption manageable while still capturing all the data.

Data Parsing and Normalization

Every source formats data differently. One site lists square footage as "1,250 sqft", another as "116 m²", and a third buries it inside a JSON-LD block. Prices come as "$425,000", "425000", or "$425K". Addresses might be a single string or split across five fields.

The normalization layer handles this through a chain of field processors that run after extraction. Each processor handles one transformation, and they compose together.

import re
from decimal import Decimal


def normalize_price(values: list[str]) -> list[Decimal]:
    """Convert price strings to Decimal values in USD."""
    results = []
    for raw in values:
        cleaned = re.sub(r"[^\d.,KMkm]", "", raw)
        if not cleaned:
            continue

        multiplier = 1
        if cleaned[-1].upper() == "K":
            multiplier = 1_000
            cleaned = cleaned[:-1]
        elif cleaned[-1].upper() == "M":
            multiplier = 1_000_000
            cleaned = cleaned[:-1]

        cleaned = cleaned.replace(",", "")
        try:
            results.append(Decimal(cleaned) * multiplier)
        except Exception:
            continue
    return results


def normalize_area(values: list[str]) -> list[Decimal]:
    """Convert area strings to square feet."""
    SQM_TO_SQFT = Decimal("10.7639")
    results = []
    for raw in values:
        raw_lower = raw.lower()
        cleaned = re.sub(r"[^\d.,]", "", raw)
        if not cleaned:
            continue
        try:
            area = Decimal(cleaned.replace(",", ""))
            if "m²" in raw_lower or "m2" in raw_lower or "sqm" in raw_lower:
                area = area * SQM_TO_SQFT
            results.append(area.quantize(Decimal("1")))
        except Exception:
            continue
    return results


def normalize_address(values: list[str]) -> list[str]:
    """Standardize address formatting."""
    results = []
    for raw in values:
        addr = " ".join(raw.split())  # collapse whitespace
        # Standardize common abbreviations
        replacements = {
            r"\bSt\b": "Street", r"\bAve\b": "Avenue",
            r"\bDr\b": "Drive", r"\bBlvd\b": "Boulevard",
            r"\bLn\b": "Lane", r"\bRd\b": "Road",
            r"\bCt\b": "Court", r"\bPl\b": "Place",
        }
        for pattern, replacement in replacements.items():
            addr = re.sub(pattern, replacement, addr)
        results.append(addr)
    return results

The key insight was making processors composable and stateless. When a new source required a quirky transformation (one site encoded price ranges as "$400-450K"), adding support was a 15-line function rather than a structural change.

Deduplication Strategy

Deduplication was the hardest problem in the pipeline. The same property can appear on multiple listing sites with different photos, slightly different descriptions, and even different prices (one agent updates faster than another). There’s no universal property ID across platforms.

The system uses a multi-signal scoring approach rather than a single matching key.

from dataclasses import dataclass
from decimal import Decimal


@dataclass
class DeduplicationScore:
    address_similarity: float  # 0-1, fuzzy string match
    geo_distance_m: float      # meters between coordinates
    price_delta_pct: float     # percentage difference
    area_delta_pct: float      # percentage difference
    bedrooms_match: bool
    bathrooms_match: bool

    @property
    def is_duplicate(self) -> bool:
        """Determine if two listings represent the same property."""
        score = 0.0

        # Strong signal: address similarity
        if self.address_similarity > 0.92:
            score += 40
        elif self.address_similarity > 0.85:
            score += 25

        # Strong signal: geographic proximity
        if self.geo_distance_m < 50:
            score += 35
        elif self.geo_distance_m < 200:
            score += 15

        # Supporting signals
        if self.price_delta_pct < 2.0:
            score += 10
        if self.area_delta_pct < 5.0:
            score += 10
        if self.bedrooms_match:
            score += 5
        if self.bathrooms_match:
            score += 5

        return score >= 65

    @property
    def confidence(self) -> str:
        score = 0.0
        if self.address_similarity > 0.92:
            score += 40
        elif self.address_similarity > 0.85:
            score += 25
        if self.geo_distance_m < 50:
            score += 35
        elif self.geo_distance_m < 200:
            score += 15
        if self.price_delta_pct < 2.0:
            score += 10
        if self.area_delta_pct < 5.0:
            score += 10
        if self.bedrooms_match:
            score += 5
        if self.bathrooms_match:
            score += 5

        if score >= 85:
            return "high"
        elif score >= 65:
            return "medium"
        return "low"

The threshold of 65 was calibrated against a hand-labeled set of 2,000 listing pairs. At this threshold, the system achieves a 97.3% precision rate (very few false positives) with 91.8% recall (catches most true duplicates). The remaining 8.2% of missed duplicates are typically cases where the same property is listed with substantially different addresses (e.g., a corner lot listed under two different street names).

When a duplicate is found, the system doesn’t discard it — it merges it. The canonical record keeps the most complete set of fields, the earliest listing date, and a full history of price observations from every source.

Storage Schema Design

The PostgreSQL schema was designed around two principles: track everything that changes, and never lose raw data.

-- Core property record — the "canonical" listing
CREATE TABLE properties (
    id              UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    canonical_addr  TEXT NOT NULL,
    latitude        NUMERIC(10, 7),
    longitude       NUMERIC(10, 7),
    property_type   TEXT,
    bedrooms        SMALLINT,
    bathrooms       NUMERIC(3, 1),
    area_sqft       INTEGER,
    lot_size_sqft   INTEGER,
    year_built      SMALLINT,
    market          TEXT NOT NULL,           -- e.g., "austin_tx"
    first_seen      TIMESTAMPTZ NOT NULL,
    last_updated    TIMESTAMPTZ NOT NULL,
    status          TEXT DEFAULT 'active',   -- active, sold, delisted
    created_at      TIMESTAMPTZ DEFAULT NOW()
);

-- Every observation from every source
CREATE TABLE price_history (
    id              BIGSERIAL PRIMARY KEY,
    property_id     UUID REFERENCES properties(id),
    source          TEXT NOT NULL,
    source_url      TEXT,
    listed_price    NUMERIC(12, 2),
    observed_at     TIMESTAMPTZ NOT NULL,
    price_change    NUMERIC(12, 2),          -- delta from previous
    created_at      TIMESTAMPTZ DEFAULT NOW()
);

-- Raw listing data before normalization
CREATE TABLE raw_listings (
    id              BIGSERIAL PRIMARY KEY,
    source          TEXT NOT NULL,
    source_url      TEXT NOT NULL,
    raw_data        JSONB NOT NULL,
    fingerprint     TEXT NOT NULL,           -- content hash for dedup
    scraped_at      TIMESTAMPTZ NOT NULL,
    processed       BOOLEAN DEFAULT FALSE,
    property_id     UUID REFERENCES properties(id)
);

-- Indexes for common query patterns
CREATE INDEX idx_properties_market ON properties(market);
CREATE INDEX idx_properties_status ON properties(status);
CREATE INDEX idx_price_history_property ON price_history(property_id, observed_at DESC);
CREATE INDEX idx_raw_listings_fingerprint ON raw_listings(fingerprint);
CREATE INDEX idx_raw_listings_unprocessed ON raw_listings(processed) WHERE NOT processed;

The raw_listings table stores the original JSONB payload from every scrape. This turned out to be invaluable — when a normalization bug was discovered three months in, it was possible to reprocess historical data without re-scraping anything. The fingerprint column (a hash of the relevant content fields) provides fast deduplication at the ingestion level before the more expensive multi-signal matching runs.

The price_history table is append-only. Every time a listing is observed, a new row is inserted. This makes time-series analysis straightforward: finding all properties with more than two price reductions is a simple window function query.

-- Find properties with multiple price drops in the last 30 days
SELECT
    p.id,
    p.canonical_addr,
    COUNT(*) AS price_drops,
    MIN(ph.listed_price) AS lowest_price,
    MAX(ph.listed_price) AS highest_price
FROM properties p
JOIN price_history ph ON ph.property_id = p.id
WHERE ph.price_change < 0
  AND ph.observed_at > NOW() - INTERVAL '30 days'
GROUP BY p.id, p.canonical_addr
HAVING COUNT(*) >= 2
ORDER BY COUNT(*) DESC;

Incremental Scraping and Change Detection

Rather than re-scraping entire listing catalogs daily, the pipeline uses incremental scraping to focus resources where they matter most. A Redis sorted set tracks when each URL was last scraped, and a scheduler prioritizes URLs based on expected update frequency.

New listings and recently changed listings get scraped every 4 hours. Listings that haven’t changed in a week drop to once-daily checks. Listings inactive for 30 days move to weekly verification. This prioritization scheme reduced total requests by approximately 60% compared to a naive full-crawl approach — a major factor in keeping rate limits sustainable.

Change detection compares the incoming data fingerprint against the stored one. When a change is detected, the system diffs the normalized fields and records exactly what changed (price, status, description, photos) along with the timestamp. This granular change log is what makes the downstream analysis possible.

Results and Metrics

After six months of production operation:

• 50,000+ records processed daily across three geographic markets
• 2M+ total property records with full history
• 97.3% deduplication precision verified against hand-labeled test sets
• Sub-2-minute latency from scrape to database for new listings
• 99.4% uptime with automated recovery from transient failures
• 60% reduction in request volume from incremental scraping optimization

The dataset has powered several downstream applications: a pricing anomaly detector that flags potentially mispriced listings, a days-on-market predictor, and neighborhood-level supply/demand dashboards updated daily.

Lessons Learned

Define the schema before writing spiders. The first iteration focused on getting data in as fast as possible, with normalization as an afterthought. That led to months of cleanup work. Defining the target schema first — and being strict about what gets stored — would have saved significant time.

Exact-match deduplication doesn’t work for real estate. Early versions used exact address matching, which missed duplicates constantly. The multi-signal scoring approach was more complex to build but dramatically more reliable. The confidence levels (high/medium/low) were essential for downstream consumers to make their own tradeoff decisions.

Keep raw data forever (or at least, much longer than you think). Storing the raw JSONB payloads felt wasteful initially — it was 80% of storage costs. But it paid for itself the first time a normalization bug needed to be fixed retroactively. Storage is cheap; re-scraping historical data is impossible.

Polite scraping is more reliable than fast scraping. Being conservative with request rates (1 per second per domain) wasn’t just about ethics — it made the system more reliable. Aggressive scraping triggers anti-bot measures, leading to IP bans and degraded data quality. Slower, polite scraping produced more consistent results with fewer operational headaches.

We never needed Kubernetes. The initial deployment was Docker Compose on a single machine. It handled 50K records/day comfortably. The temptation to jump to Kubernetes was real, but the operational simplicity of Compose was worth more than theoretical scaling headroom.

Ethical Compliance Details

This project was built with a strict ethical framework from day one. Every design decision was evaluated against the question: “Would the site operator consider this reasonable?”

robots.txt is non-negotiable. Every spider has ROBOTSTXT_OBEY = True in its settings. This is enforced at the framework level — there is no override mechanism. If a site disallows scraping a path, that path is never requested, period.

Rate limiting goes beyond minimums. The 1 request/second per domain limit is a hard ceiling, not a target. In practice, most spiders average 0.5-0.7 requests/second due to processing delays. During the incremental scraping redesign, the rate limit was actually lowered from 2 req/s to 1 req/s after observing that the faster rate occasionally triggered rate-limit responses from a source.

Transparent identification. The bot uses a descriptive User-Agent string (ScrapemBot/1.0 (+https://scrapem.com/bot-info)) that links to a page explaining what the bot does, what data it collects, and how to request exclusion. Two site operators have reached out through this page — one to request exclusion (honored immediately) and one to offer API access instead.

Public data only. The system only collects information that is publicly visible without authentication. No login walls are bypassed, no CAPTCHAs are solved, and no paywalled content is accessed. Listing agent contact information is explicitly excluded from collection even when publicly displayed.

Data retention and privacy. Raw data older than 12 months is archived and purged from the active database. Any inadvertently collected personal information (which the parsers are designed to avoid) is flagged by an automated PII scanner and removed within 24 hours. The system maintains a processing log for GDPR compliance, documenting what data was collected, from where, and when it was deleted.