Your daily import can embed years of history again while your evaluation gets suspiciously easy. The culprit can be sitting inside rateTrend, wearing yesterday's observations as today's dataset row.
Lab note 072: can we keep mortgage-history retrieval incremental without sending the same observations through the embedding queue again?
The catch comes early: an observation date doesn't tell us when a revised value became available. Removing duplicate history helps, but it doesn't automatically make an evaluation prospective.
The row is a shipping container
A dataset row containing historical observations is an export unit, not a reliable retrieval identity.
Our mortgage-rate Actor makes the distinction concrete. A 4,000-day daily history request returns 4,001 points beginning on 2015-11-08; monthly history returns 132 points beginning on 2015-11-01.
Scheduled windows can therefore overlap extensively. Giving each import a fresh document ID preserves that overlap beautifully. Unfortunately.
We want observations with stable identities, a revision ledger, and bounded documents whose text stays unchanged when only collection provenance changes.
Example setup: this requests an aggregate history corpus and leaves the unrelated enrichments disabled.
import os
import json
import hashlib
from collections import defaultdict
from datetime import date, datetime, timezone
from decimal import Decimal, InvalidOperation
import tiktoken
from apify_client import ApifyClient
run_input = {
"loanPrograms": ["30yr-fixed"],
"loanPurpose": "purchase",
"states": ["national"],
"creditScore": "veryHigh",
"loanAmount": "conforming",
"includeRateHistory": True,
"historyDays": "4000",
"historyAggregation": "daily",
"includeHomeLoanRates": False,
"includeLenderDirectory": False,
"includeLenderProfiles": False,
}
client = ApifyClient(os.environ["APIFY_TOKEN"])
run = client.actor(
"vexaloom-labs/zillow-daily-mortgage-rate-and-lender-quote"
).call(run_input=run_input)
if run["status"] != "SUCCEEDED":
raise RuntimeError(f"History collection ended: {run['status']}")
items = client.dataset(run["defaultDatasetId"]).iterate_items()
Keep the input configuration alongside each export. In particular, historyAggregation belongs to request provenance; we shouldn't assume it will appear in an output row.
Give each observation an address
The useful identity is the series plus its observation date. The sampling interval belongs in the series: monthly and daily observations shouldn't silently become interchangeable.
For this corpus, we use these derived fields. They're our schema, not additional Actor output fields.
| Derived field | Identifies | Why it exists |
|---|---|---|
series_id |
Product, purpose, geography, assumptions, source, interval | Separates observations answering different questions |
observation_id |
Series and observation date | Survives repeated imports |
value_hash |
Rate and available APR | Detects changed values |
version_id |
Observation and value hash | Preserves revisions without duplicating identical values |
A logical key answers “which observation?” A fingerprint answers “did its values change?” Using either alone loses something.
This cell defines canonical hashing and numeric normalization, so equivalent numeric representations don't create artificial revisions.
def canonical(value):
return json.dumps(
value, sort_keys=True, ensure_ascii=False,
separators=(",", ":"), allow_nan=False
)
def digest(value):
return hashlib.sha256(canonical(value).encode("utf-8")).hexdigest()
def numeric_text(value):
if isinstance(value, bool) or not isinstance(
value, (str, int, float, Decimal)
):
raise ValueError("Expected a numeric rate")
parsed = Decimal(str(value))
if not parsed.is_finite():
raise ValueError("Non-finite rate")
return format(parsed.normalize(), "f")
versions = {}
requests = {}
rejected = defaultdict(int)
The hashes are identifiers, not privacy protection. Hashing a phone number would still be keeping a phone number's fingerprint.
Keep the parent, lose the baggage
Flattening shouldn't mean forgetting where a point came from.
The structure we want looks like this:
export + saved request
-> aggregate series context
-> observation date
-> distinct value versions
-> collection sightings
-> bounded retrieval documents referencing those versions
The parent context carries product, purpose, geography, assumptions, and source. Collection sightings carry the dataset reference and scrapedAt.
Early history can contain a rate without APR. Requiring both would selectively erase older history, so the ingestion function preserves rate-only observations.
This cell merges identical versions while retaining their collection provenance.
def ingest(records, config, dataset_id):
request_id = digest(config)
requests[request_id] = dict(config)
for row in records:
if row.get("recordType") != "aggregate":
continue
try:
series = {
key: row[key]
for key in (
"recordType", "loanType", "productType",
"loanPurpose", "scope", "source"
)
}
series["assumptions"] = {
key: row["assumptions"][key]
for key in (
"creditScoreBucket",
"loanToValueBucket",
"loanAmountBucket",
)
}
series["historyAggregation"] = config["historyAggregation"]
if series["scope"] == "state":
series["state"] = row["state"]
elif series["scope"] != "national":
raise ValueError("Unknown geography")
if series["loanPurpose"] not in ("purchase", "refinance"):
raise ValueError("Ambiguous purpose")
trend = row["rateTrend"]
if not isinstance(trend, list):
raise ValueError("History must be an array")
collected = datetime.fromisoformat(
row["scrapedAt"].replace("Z", "+00:00")
)
if collected.utcoffset() is None:
raise ValueError("Collection time needs a timezone")
stamp = collected.astimezone(timezone.utc).isoformat()
except (KeyError, TypeError, AttributeError, ValueError):
rejected["row_context"] += 1
continue
series_id = digest(series)
sighting = {
"dataset_id": dataset_id,
"request_id": request_id,
"scrapedAt": stamp,
}
for point in trend:
try:
day = date.fromisoformat(point["date"][:10]).isoformat()
values = {"rate": numeric_text(point["rate"])}
if point.get("apr") is not None:
values["apr"] = numeric_text(point["apr"])
except (KeyError, TypeError, AttributeError,
ValueError, InvalidOperation):
rejected["history_point"] += 1
continue
observation_id = digest([series_id, day])
value_hash = digest(values)
version_id = digest([observation_id, value_hash])
version = versions.setdefault(version_id, {
"version_id": version_id,
"observation_id": observation_id,
"series_id": series_id,
"series": series,
"date": day,
"values": values,
"value_hash": value_hash,
"sightings": {},
})
version["sightings"][digest(sighting)] = sighting
ingest(items, run_input, run["defaultDatasetId"])
The allowlist also keeps lender names, contact details, biographies, and review text out of the corpus. Query assumptions describe the rate query, not an individual borrower.
This notebook keeps its ledger in memory. Feed additional exports through the same ingest state; a scheduled service needs durable storage for these keys and sightings. Starting an empty ledger every morning doesn't preserve revision history.
Bound chunks by time and tokens
An observation is a good identity, but sometimes a poor embedding unit. A question about movement across a period needs neighboring observations.
We'd group by series and calendar month, then split further when the token budget requires it. That bounds the time span and keeps unrelated products out of the same document.
The archive retains every distinct value version. The retrieval view below selects the most recently collected version of each observation. Conflicting values with indistinguishable collection times stop the build instead of receiving an arbitrary winner.
This cell creates deterministic, budgeted documents without putting import timestamps into embedding text.
encoding = tiktoken.get_encoding("cl100k_base")
TOKEN_BUDGET = 512
SPLIT_DATE = "2026-10-01"
selected = {}
latest_seen = {}
for version in versions.values():
observation_id = version["observation_id"]
seen_at = max(
sighting["scrapedAt"]
for sighting in version["sightings"].values()
)
previous = selected.get(observation_id)
if previous is not None and seen_at == latest_seen[observation_id]:
if previous["value_hash"] != version["value_hash"]:
raise ValueError("Conflicting versions share a collection time")
if previous is None or seen_at > latest_seen[observation_id]:
selected[observation_id] = version
latest_seen[observation_id] = seen_at
groups = defaultdict(list)
for version in selected.values():
split = "train" if version["date"] < SPLIT_DATE else "test"
groups[(split, version["series_id"], version["date"][:7])].append(version)
def token_count(text):
return len(encoding.encode(text, disallowed_special=()))
def make_document(members, text, split):
return {
"document_id": digest(text),
"series_id": members[0]["series_id"],
"split": split,
"start_date": members[0]["date"],
"end_date": members[-1]["date"],
"observation_ids": [v["observation_id"] for v in members],
"version_ids": [v["version_id"] for v in members],
"text": text,
"token_count": token_count(text),
}
documents = []
for (split, series_id, month), members in sorted(groups.items()):
members.sort(key=lambda version: version["date"])
header = (
"Aggregate mortgage rate index; rate and APR are percentages.\n"
+ canonical(members[0]["series"]) + "\n"
)
current, text = [], header
for version in members:
values = version["values"]
line = f'{version["date"]}: rate={values["rate"]}%'
if "apr" in values:
line += f'; APR={values["apr"]}%'
line += "\n"
if token_count(header + line) > TOKEN_BUDGET:
raise ValueError("Series context and observation exceed budget")
if current and token_count(text + line) > TOKEN_BUDGET:
documents.append(make_document(current, text, split))
current, text = [], header
current.append(version)
text += line
if current:
documents.append(make_document(current, text, split))
Before embedding, look up the embedding model version and document_id together in your durable cache. Queue only missing combinations.
Unchanged text keeps its ID even when another sighting arrives. A changed observation can replace its containing chunk; the ledger still records the older value.
That's the small fix: collection metadata can grow without making the embedding text pretend it's new.
Can an import-date split leak history?
Yes. Splitting exported rows by scrapedAt can put identical historical observations on both sides of the boundary.
Split observations before chunking. Otherwise, a chunk near the boundary can carry training dates into the test partition.
This cell checks logical-key overlap and writes the corpus, revision archive, and request manifest as JSONL.
partition_keys = {"train": set(), "test": set()}
for document in documents:
partition_keys[document["split"]].update(document["observation_ids"])
if partition_keys["train"] & partition_keys["test"]:
raise ValueError("Observation identity crosses evaluation partitions")
def write_jsonl(path, records):
with open(path, "w", encoding="utf-8") as handle:
for record in records:
handle.write(canonical(record) + "\n")
write_jsonl("corpus.jsonl", documents)
write_jsonl("versions.jsonl", (
{
**version,
"sightings": sorted(
version["sightings"].values(), key=canonical
),
}
for version in sorted(
versions.values(), key=lambda entry: entry["version_id"]
)
))
write_jsonl("requests.jsonl", (
{"request_id": request_id, "input": config}
for request_id, config in sorted(requests.items())
))
The overlap check proves a narrow property: the same logical observation doesn't cross partitions. It doesn't prove the series are statistically independent.
For prospective evaluation, freeze the information available at the decision time as well. The code's latest-version view can include later revisions to earlier observations.
scrapedAt records when the run started, not when the source first published a revision. Archived sightings can establish when your pipeline observed a value; they can't reconstruct an unknown publication history.
What this corpus refuses to claim
Matching an observation date doesn't make daily, weekly, and monthly samples equivalent. Keep their series separate unless you define and verify a transformation.
Dropping every point without APR doesn't produce a neutral cleanup. It changes which historical periods survive.
Embedding an entire rateTrend array doesn't preserve useful context automatically. It couples a large overlapping window to an import identity and lets retrieval boundaries depend on export packaging.
The aggregate rate index also shouldn't be labeled as a third-party lender quote. We leave rateSamples outside this corpus because advertised product rates answer a different question.
Our verdict: flatten for identity, version for corrections, and chunk for the retrieval question. Those are separate decisions.
Inspect rejected before accepting an export. A readable JSONL file says nothing about whether the requested series coverage survived ingestion.
Result fees are a separate meter
Dataset events and embedding tokens measure different workloads. Flattening a historical array creates corpus observations; it doesn't turn those observations into additional dataset-item events.
With a start-event count of one, event fees for 100 dataset-item events are $0.199 on FREE or $0.079 on GOLD. For 1,000, they're $1.09 or $0.52 respectively. Each amount is plus platform usage.
The start-event count depends on memory, with an event per GB and a minimum event charge. Keep it configurable rather than hiding it inside a token estimate.
This calculator isolates event fees and leaves platform usage separate.
event_rates = {
"FREE": (Decimal("0.10"), Decimal("0.00099")),
"GOLD": (Decimal("0.03"), Decimal("0.00049")),
}
def event_fee(item_events, *, tier="FREE", start_events=1):
if item_events < 0 or start_events < 1:
raise ValueError("Invalid event counts")
start_price, item_price = event_rates[tier]
return start_events * start_price + item_events * item_price
event_quotes = {
tier: {
count: event_fee(count, tier=tier)
for count in (100, 1000)
}
for tier in event_rates
}
The worked example is Zillow Daily Mortgage Rate & Lender Quote Scraper.
Reproduce the corpus build
- Install
apify-clientandtiktoken, and setAPIFY_TOKENin your environment. - Run the Python cells in order. Keep each additional export paired with its original request configuration.
- Inspect rejection counts, version sightings, partition membership, and token counts before indexing.
- Persist the ledger and embedding cache for scheduled imports. Retire superseded chunk IDs when publishing a new retrieval snapshot.
Example input, matching the Python setup exactly:
{
"loanPrograms": ["30yr-fixed"],
"loanPurpose": "purchase",
"states": ["national"],
"creditScore": "veryHigh",
"loanAmount": "conforming",
"includeRateHistory": true,
"historyDays": "4000",
"historyAggregation": "daily",
"includeHomeLoanRates": false,
"includeLenderDirectory": false,
"includeLenderProfiles": false
}
Top comments (0)