Did the property change, or did your collection mode?

You're about to publish a count of homes that “switched from rent to sale”. The join on propertyId looks sensible. The change flag comes from comparing mode.

That's where the small dread belongs. A property can advertise rent and sale together, and the Actor includes both prices when the source publishes them.

A change in collection mode does not establish a change in the property's advertised offers.

A collection label isn't an offer event

The awkward value is byUrls. Alongside rent and buy, it describes loading specific listings, rather than an advertised transaction.

Treating that field as a property's exclusive status would turn a different retrieval path into a market event. SQL will count it quite happily. SQL has no reputation to protect.

The separation looks like this:

propertyId ├── rent collection → rentPrice, salePrice, offer flags ├── buy collection → rentPrice, salePrice, offer flags └── URL collection → rentPrice, salePrice, offer flags 

Those branches are observations of an entity. They aren't mutually exclusive identities.

Your change feed needs to name the attribute that changed: an advertised rent, an advertised sale price, or an explicit availability flag.

Use the property ID without flattening the prices

Use propertyId as the source identifier linking observations. Within storage, qualify it with the collection identifier so a later observation doesn't overwrite an earlier one.

The useful distinctions fit in a small table:

For this notebook, keep an in-memory database open across the cells.

import os import json import re from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from uuid import uuid4 import pandas as pd import duckdb from apify_client import ApifyClient client = ApifyClient(os.environ["APIFY_TOKEN"]) con = duckdb.connect() 

At operational scale, the same tables belong in durable storage. An in-memory notebook is useful for inspecting the model, but it isn't a historical archive.

Store collections without inventing simultaneity

A snapshot identifier groups a collection round. It doesn't make sequential requests simultaneous.

Keep the collection's start and finish times, its complete input, and each record's scrapedAt. A shared round identifier is an organisational convenience, not a time machine.

Create separate tables for collection provenance, property observations and the attributes being tracked.

con.execute(""" CREATE TABLE collections ( snapshot_id VARCHAR, collection_id VARCHAR PRIMARY KEY, started_at TIMESTAMPTZ, finished_at TIMESTAMPTZ, comparison_scope VARCHAR, input_mode VARCHAR, input_json VARCHAR ); CREATE TABLE observations ( collection_id VARCHAR, property_id VARCHAR, scraped_at VARCHAR, property_url VARCHAR, payload VARCHAR, PRIMARY KEY (collection_id, property_id) ); CREATE TABLE offer_fields ( collection_id VARCHAR, property_id VARCHAR, attribute VARCHAR, value VARCHAR, state VARCHAR, PRIMARY KEY (collection_id, property_id, attribute) ); """) 

The attribute table stores prices separately even when they arrived together. It also keeps a parsing state, because a missing price and a valid price are different evidence.

The full payload remains available when a manager asks what the source actually returned.

Make missing prices earn their meaning

Empty fields are omitted. A missing salePrice therefore doesn't justify “sale offer withdrawn”, and replacing it with a numeric default would manufacture a price movement.

Formatting needs similar restraint. The price contract is BRL; unexpected currencies or unsupported formats should enter review rather than acquire a guessed meaning.

These normalisers distinguish usable values from missing or invalid evidence.

def normalise_price(value, currency): if value is None or value == "": return None, "missing" if currency != "BRL" or isinstance(value, bool): return None, "invalid" try: if isinstance(value, str): text = value.strip() pattern = r"(?:R\$\s*)?(\d+|\d{1,3}(?:\.\d{3})+)(?:,(\d{1,2}))?" match = re.fullmatch(pattern, text) if match is None: return None, "invalid" whole, fraction = match.groups() amount = Decimal( whole.replace(".", "") + "." + (fraction or "0") ) elif isinstance(value, (int, float, Decimal)): amount = Decimal(str(value)) else: return None, "invalid" if not amount.is_finite() or amount < 0: return None, "invalid" rounded = amount.quantize(Decimal("0.01")) if rounded != amount: return None, "invalid" return format(rounded, ".2f"), "value" except (InvalidOperation, ValueError): return None, "invalid" def normalise_flag(value): if value is None or value == "": return None, "missing" if isinstance(value, bool): return str(value).lower(), "value" return None, "invalid" 

This deliberately leaves unfamiliar flag representations unresolved. Converting every non-empty string to True is an efficient way to advertise everything.

A field becoming missing still deserves attention. It belongs in a data-quality queue until you can distinguish publication changes from collection differences.

Reject duplicate keys before the overlap join

Check uniqueness inside each collection before joining feeds. A duplicate property ID can multiply rows and make a clean-looking overlap count indefensible.

The capture function validates identifiers, retains raw records and writes each accepted collection in a transaction.

Example collection inputs are defined inline below; the consecutive captures exercise the comparison mechanics.

def capture_snapshot(): snapshot_id = str(uuid4()) for input_mode in ("rent", "buy"): run_input = { "mode": input_mode, "city": "sao-paulo-sp-brasil", "maxItems": 200, "includeDetails": True, } input_json = json.dumps(run_input, sort_keys=True) collection_id = str(uuid4()) started_at = datetime.now(timezone.utc) run = client.actor( "dataspryva/brazil-rental-and-sale-listings-by-city" ).call(run_input=run_input) if run["status"] != "SUCCEEDED": raise RuntimeError(f"Collection failed: {run['status']}") items = list( client.dataset(run["defaultDatasetId"]).iterate_items() ) finished_at = datetime.now(timezone.utc) ids = pd.Series([ None if item.get("propertyId") is None else str(item["propertyId"]).strip() for item in items ], dtype="string") bad = ids.isna() | ids.eq("") | ids.duplicated(keep=False) if bad.any(): raise ValueError( f"Missing or duplicate propertyId values: {ids[bad].tolist()}" ) observation_rows = [] field_rows = [] for property_id, item in zip(ids.tolist(), items): observation_rows.append(( collection_id, property_id, item.get("scrapedAt"), item.get("propertyUrl"), json.dumps(item), )) for attribute in ("rentPrice", "salePrice", "forRent", "forSale"): raw_value = item.get(attribute) if attribute in ("rentPrice", "salePrice"): value, state = normalise_price( raw_value, item.get("currency") ) else: value, state = normalise_flag(raw_value) field_rows.append(( collection_id, property_id, attribute, value, state, )) con.execute("BEGIN") try: con.execute( "INSERT INTO collections VALUES (?, ?, ?, ?, ?, ?, ?)", [snapshot_id, collection_id, started_at, finished_at, input_json, input_mode, input_json], ) if observation_rows: con.executemany( "INSERT INTO observations VALUES (?, ?, ?, ?, ?)", observation_rows, ) con.executemany( "INSERT INTO offer_fields VALUES (?, ?, ?, ?, ?)", field_rows, ) con.execute("COMMIT") except Exception: con.execute("ROLLBACK") raise capture_snapshot() capture_snapshot() 

A failed key check stops the collection rather than selecting an arbitrary “winner”. If duplicates represent meaningful distinctions, investigate those distinctions before changing the key.

The notebook takes adjacent captures. A scheduled tracker needs a deliberate observation interval; back-to-back captures don't establish a market trend.

Now an overlap join can retain separate timestamps instead of coalescing the feeds into a supposedly simultaneous property state.

overlap = con.execute(""" SELECT r.snapshot_id, ro.property_id, r.collection_id AS rent_collection_id, b.collection_id AS buy_collection_id, ro.scraped_at AS rent_scraped_at, bo.scraped_at AS buy_scraped_at FROM collections r JOIN collections b ON r.snapshot_id = b.snapshot_id AND r.input_mode = 'rent' AND b.input_mode = 'buy' JOIN observations ro ON ro.collection_id = r.collection_id JOIN observations bo ON bo.collection_id = b.collection_id AND bo.property_id = ro.property_id """).df() 

This gives you linked observations for inspection. It doesn't choose which observation is authoritative.

Diff adjacent collections with the same inputs

The comparison scope here's the complete serialised input. That keeps retrieval-mode changes outside the offer-change count, along with changes to filters, enrichment settings or collection caps.

It's conservative by design. Reaching a property through a different path can be useful corroboration, but it isn't automatically a comparable observation.

Choose the previous collection before joining property rows. Otherwise, a per-property LAG can skip an intervening absence and quietly compare non-adjacent collections.

Build the comparison pairs and separate observed changes from unresolved fields.

con.execute(""" CREATE TEMP VIEW field_pairs AS WITH ordered AS ( SELECT *, lag(collection_id) OVER ( PARTITION BY comparison_scope ORDER BY started_at, finished_at, collection_id ) AS previous_collection_id FROM collections ) SELECT c.input_mode, c.previous_collection_id, c.collection_id, n.property_id, n.attribute, p.value AS previous_value, n.value AS current_value, p.state AS previous_state, n.state AS current_state, po.scraped_at AS previous_scraped_at, no.scraped_at AS current_scraped_at FROM ordered c JOIN offer_fields n ON n.collection_id = c.collection_id JOIN offer_fields p ON p.collection_id = c.previous_collection_id AND p.property_id = n.property_id AND p.attribute = n.attribute JOIN observations po ON po.collection_id = p.collection_id AND po.property_id = p.property_id JOIN observations no ON no.collection_id = n.collection_id AND no.property_id = n.property_id """) changes = con.execute(""" SELECT * FROM field_pairs WHERE previous_state = 'value' AND current_state = 'value' AND previous_value IS DISTINCT FROM current_value """).df() review_fields = con.execute(""" SELECT * FROM field_pairs WHERE previous_state <> 'value' OR current_state <> 'value' """).df() con.close() 

changes names the changed attribute and retains its observation timestamps. review_fields keeps uncertainty visible without turning it into an offer event.

A property missing from an adjacent collection isn't included in either attribute comparison. Track collection presence separately; don't convert absence into forSale = false.

The worked feed is Brazil Rental & Sale Listings by City (QuintoAndar). Its collection cap and finite source inventory also mean that unchanged inputs don't guarantee identical coverage.

The change feed stops at the advertised offer

A changed asking price doesn't establish a completed transaction. An explicit flag change doesn't explain the owner's reason, and the scrape timestamps only bracket when you observed the difference.

The property identifier connects source records. It doesn't independently verify physical identity across the property's entire history.

Your report can defend “the advertised sale price differed between these observations”. It needs more evidence to defend “this home switched markets”.

A collection setting shouldn't get promoted into a property story.