← back to Nationalrealestate
src/ingest/parcels/nyc_acris.ts
223 lines
/**
* NYC ACRIS recorded-deed sales -> existing PLUTO parcels (NYC Open Data, $0).
* Fills the "Sale/deed chain from ACRIS pending" gap on all five boroughs by
* layering REAL recorded sale prices onto the lots already loaded by nyc_pluto.ts.
*
* Real Property Master (bnx9-e6tj) -> document_id, doc_type, document_date, document_amt
* Real Property Legals (8h5j-fqxa) -> document_id -> borough/block/lot (=> BBL)
*
* Join is by document_id. Master is filtered to priced SALE documents — the DEED
* family plus transfer-tax returns (RPTT&RET/RPTT), since many NYC sales carry the
* price on the transfer-tax filing rather than a bare DEED. Legals is keyset-streamed
* (numeric doc_ids 2018–2026) and filtered in memory to those document_ids (chunked
* IN() 414s at ~250 ids, so keyset-stream is fewer requests). Sales are deduped per
* BBL by (date, price) so a sale filed as both a DEED and an RPTT&RET counts once.
* The reconstructed BBL (borough + block.pad(5) + lot.pad(4)) matches PLUTO's
* source_id integer part, so we UPDATE existing rows — never insert.
*
* KNOWN COVERAGE LIMITS: (1) ACRIS records condo UNIT sales on lot >= 1001 while
* PLUTO carries the building's billing BBL, so unit-level condo deeds don't join to a
* PLUTO lot. v1 matches whole-building / 1-3 family / base-lot deeds. (2) Active-
* development lots (~2.7% of matched) accumulate many unit-sale records against the
* PARENT BBL before subdivision, so those show a cluster of same-period sales — real
* recorded transfers, but the single "last_sale" headline is the most recent of them.
* Condo-unit coverage and buyer/seller names (ACRIS Parties) are a follow-up pass.
*
* Run: npm run ingest:parcels -- nyc-acris (NODE_OPTIONS=--max-old-space-size=4096 recommended)
*/
import { query, pool } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
import { sanitizeEventDate } from './date_guard.ts';
const MASTER = 'https://data.cityofnewyork.us/resource/bnx9-e6tj.json';
const LEGALS = 'https://data.cityofnewyork.us/resource/8h5j-fqxa.json';
const PAGE = 50000;
const SINCE = '2018-01-01T00:00:00';
const DOC_FLOOR = '2018000000000000'; // keyset lower bound for legals (modern numeric doc_ids)
const DOC_CEIL = '2027000000000000'; // upper bound — deed doc_ids are numeric YYYY…, so 2018–2026
const MIN_AMT = 1000; // drop nominal $1/$10 gift/estate transfers (not market sales)
// Priced-sale document types: the DEED family (DEED, DEEDO, "DEED, TS", "DEED, LE") plus the
// transfer-tax returns (RPTT&RET, RPTT) that carry consideration when the deed itself is nominal
// or unrecorded. A single sale can file BOTH a DEED and an RPTT&RET, so sales are deduped per
// BBL by (date, price). Excludes MTGE/AGMT/AL&R/M&CON (financing) and other non-sale instruments.
const SALE_WHERE = `(doc_type like 'DEED%' OR doc_type in('RPTT&RET','RPTT'))`;
// borocode -> county_fips (mirrors nyc_pluto.ts BORO)
const FIPS: Record<string, string> = { '1': '36061', '2': '36005', '3': '36047', '4': '36081', '5': '36085' };
interface Sale { date: string; price: number; doc: string }
/** GET SoQL JSON, retrying 429/5xx AND network/timeout exceptions with backoff
* (unauthenticated public API over long streams -> transient timeouts happen).
* 4xx (bad query) fails fast; each request is bounded so it never hangs. */
async function soda(base: string, params: Record<string, string>): Promise<any[]> {
const url = base + '?' + new URLSearchParams(params).toString();
let lastErr: unknown;
for (let attempt = 0; attempt < 8; attempt++) {
try {
const res = await fetch(url, { headers: { accept: 'application/json' }, signal: AbortSignal.timeout(90_000) });
if (res.ok) return res.json() as Promise<any[]>;
if (res.status < 500 && res.status !== 429) {
throw new Error(`SODA ${res.status} ${res.statusText}: ${url.slice(0, 120)}…`); // non-retryable
}
lastErr = new Error(`SODA ${res.status} ${res.statusText}`); // 429/5xx -> retry
} catch (e: any) {
if (typeof e?.message === 'string' && e.message.startsWith('SODA ')) throw e; // fail-fast 4xx
lastErr = e; // network / headers-timeout / abort -> retry
}
await new Promise(r => setTimeout(r, 1500 * (attempt + 1)));
}
throw lastErr;
}
/** ACRIS document_date / recorded_datetime -> YYYY-MM-DD (null if unusable). */
function isoDate(d: string | undefined): string | null {
if (!d) return null;
const s = String(d).slice(0, 10);
// drop future/absurd bad-entry dates so they never reach parcel.last_sale_date (shared guard)
return /^\d{4}-\d{2}-\d{2}$/.test(s) ? sanitizeEventDate(s) : null;
}
export async function ingestNycAcris(): Promise<{ upserted: number }> {
const runId = await openRun('parcel_nyc_acris', MASTER);
try {
// 1) Master: priced DEED/DEEDO docs since 2018, keyset by document_id -> {date, price}
const deed = new Map<string, { date: string; price: number }>();
let cursor = '';
for (;;) {
const where = `document_amt >= ${MIN_AMT} AND ${SALE_WHERE} ` +
`AND recorded_datetime > '${SINCE}'` + (cursor ? ` AND document_id > '${cursor}'` : '');
const rows = await soda(MASTER, {
$select: 'document_id,document_date,document_amt,recorded_datetime',
$where: where, $order: 'document_id', $limit: String(PAGE),
});
if (!rows.length) break;
for (const o of rows) {
const date = isoDate(o.document_date) || isoDate(o.recorded_datetime);
const price = Number(o.document_amt);
if (date && Number.isFinite(price) && price >= MIN_AMT) deed.set(o.document_id, { date, price });
}
cursor = rows[rows.length - 1].document_id;
if (rows.length < PAGE) break;
}
console.log(`[acris] master: ${deed.size} priced deeds`);
if (deed.size < 200000) throw new Error(`only ${deed.size} deeds — ACRIS Master drift?`);
// 2) Legals: keyset-stream; collect the (fips,bbl) legal rows PER deed document.
// document_amt is the AGGREGATE consideration for the whole document, so a deed
// that transfers many lots (portfolio/bulk sale) would wrongly attribute its
// total to each lot ($1.5B "sale prices"). We therefore only keep a price when a
// document maps to exactly ONE BBL — a single-lot deed's amount IS that lot's
// sale price; multi-lot documents have no knowable per-lot price and are dropped.
const docHits = new Map<string, { fips: string; bbl: string }[]>();
let legalRows = 0, matched = 0;
cursor = DOC_FLOOR;
for (;;) {
const rows = await soda(LEGALS, {
$select: 'document_id,borough,block,lot',
$where: `document_id > '${cursor}' AND document_id < '${DOC_CEIL}'`,
$order: 'document_id', $limit: String(PAGE),
});
if (!rows.length) break;
legalRows += rows.length;
for (const o of rows) {
if (!deed.has(o.document_id)) continue;
const fips = FIPS[String(o.borough)];
if (!fips || !o.block || !o.lot) continue;
const bbl = String(o.borough) + String(o.block).padStart(5, '0') + String(o.lot).padStart(4, '0');
matched++;
const hits = docHits.get(o.document_id);
if (hits) hits.push({ fips, bbl }); else docHits.set(o.document_id, [{ fips, bbl }]);
}
cursor = rows[rows.length - 1].document_id;
if (rows.length < PAGE) break;
if (legalRows % 1000000 < PAGE) console.log(`[acris] legals scanned ${legalRows}, ${docHits.size} deed docs so far`);
}
// Attribute each single-lot document's price to its one BBL; drop multi-lot docs.
const salesByBbl = new Map<string, { fips: string; sales: Sale[] }>();
let multiLotDropped = 0;
for (const [doc, hits] of docHits) {
const uniq = new Map(hits.map(h => [h.fips + ':' + h.bbl, h]));
if (uniq.size !== 1) { multiLotDropped++; continue; }
const { fips, bbl } = [...uniq.values()][0];
const d = deed.get(doc)!;
const sale: Sale = { date: d.date, price: d.price, doc };
const rec = salesByBbl.get(bbl);
if (rec) rec.sales.push(sale); else salesByBbl.set(bbl, { fips, sales: [sale] });
}
console.log(`[acris] legals: ${legalRows} scanned, ${matched} hits, ${docHits.size} deed docs -> ${salesByBbl.size} single-lot bbls (${multiLotDropped} multi-lot docs dropped)`);
if (salesByBbl.size < 20000) throw new Error(`only ${salesByBbl.size} BBLs matched — join drift?`);
// 3) Stage the sales, then one set-based UPDATE onto PLUTO rows. A TEMP TABLE
// is connection-scoped, so pin ONE pooled client across CREATE/INSERT/UPDATE
// (pool.query() would hand each statement a different connection).
const rowsForInsert: [string, string, string, number, string][] = [];
for (const [bbl, { fips, sales }] of salesByBbl) {
// Dedup: one sale can file both a DEED and an RPTT&RET (same date+price) — collapse to one.
const seen = new Set<string>();
const uniq = sales.filter(s => { const k = s.date + '|' + s.price; return seen.has(k) ? false : seen.add(k); });
uniq.sort((a, b) => b.date.localeCompare(a.date));
const top = uniq.slice(0, 10);
rowsForInsert.push([fips, bbl, top[0].date, top[0].price,
JSON.stringify(top.map(s => ({ date: s.date, price: s.price, buyer: null, seller: null, document_id: s.doc })))]);
}
const client = await pool.connect();
let upserted = 0;
try {
await client.query(`CREATE TEMP TABLE tmp_acris (fips text, bbl bigint, last_date date, last_price numeric, sales jsonb)`);
for (let i = 0; i < rowsForInsert.length; i += 1000) {
const chunk = rowsForInsert.slice(i, i + 1000);
const params: unknown[] = [];
const vals = chunk.map((r, j) => {
const b = j * 5; params.push(r[0], r[1], r[2], r[3], r[4]);
return `($${b + 1},$${b + 2},$${b + 3},$${b + 4},$${b + 5}::jsonb)`;
});
await client.query(`INSERT INTO tmp_acris (fips,bbl,last_date,last_price,sales) VALUES ${vals.join(',')}`, params);
}
// Idempotent apply: clear the boroughs' ACRIS-sourced sale fields, then re-apply,
// in one transaction — so a re-run never leaves residue from a prior (narrower) run.
const boros = Object.values(FIPS);
await client.query('BEGIN');
await client.query(
`UPDATE parcel SET last_sale_date = NULL, last_sale_price = NULL,
extra = (COALESCE(extra, '{}'::jsonb) - 'sales') - 'sale_source'
WHERE county_fips = ANY($1::text[])`, [boros]);
const upd = await client.query<{ n: string }>(
`WITH u AS (
UPDATE parcel p SET
last_sale_date = t.last_date,
last_sale_price = t.last_price,
extra = COALESCE(p.extra, '{}'::jsonb)
|| jsonb_build_object('sales', t.sales,
'sale_source', 'NYC ACRIS recorded priced transfers (deed family + RPTT, 2018–present)')
FROM tmp_acris t
WHERE p.county_fips = t.fips AND floor(p.source_id::numeric)::bigint = t.bbl
RETURNING 1)
SELECT count(*)::text AS n FROM u`);
await client.query('COMMIT');
upserted = Number(upd.rows[0].n);
} catch (e) {
await client.query('ROLLBACK').catch(() => {});
throw e;
} finally {
client.release();
}
console.log(`[acris] updated ${upserted} PLUTO parcels with recorded sales`);
await closeRun(runId, 'ok', {
upserted,
notes: `NYC ACRIS: ${deed.size} priced deeds, ${salesByBbl.size} BBLs, ${upserted} PLUTO lots updated (condo-unit + buyer/seller pending)`,
});
return { upserted };
} catch (e: any) {
await closeRun(runId, 'failed', { notes: String(e.message || e) });
throw e;
}
}
if (import.meta.url === `file://${process.argv[1]}`) {
ingestNycAcris().then(() => pool.end()).catch(e => { console.error(e); process.exit(1); });
}