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 ...
