← back to Nationalrealestate
auto-save: 2026-07-21T19:09:20 (2 files) — src/ingest/parcels/nyc_acris.ts src/server/property.ts
14c95a8f291464bf631b9c26579e914484b6c8a7 · 2026-07-21 19:09:21 -0700 · Steve Abrams
Files touched
M src/ingest/parcels/nyc_acris.tsM src/server/property.ts
Diff
commit 14c95a8f291464bf631b9c26579e914484b6c8a7
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Tue Jul 21 19:09:21 2026 -0700
auto-save: 2026-07-21T19:09:20 (2 files) — src/ingest/parcels/nyc_acris.ts src/server/property.ts
---
src/ingest/parcels/nyc_acris.ts | 98 +++++++++++++++++++++++++++--------------
src/server/property.ts | 2 +-
2 files changed, 67 insertions(+), 33 deletions(-)
diff --git a/src/ingest/parcels/nyc_acris.ts b/src/ingest/parcels/nyc_acris.ts
index 41c11c8..6d64ee6 100644
--- a/src/ingest/parcels/nyc_acris.ts
+++ b/src/ingest/parcels/nyc_acris.ts
@@ -6,9 +6,12 @@
* 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 DEED/DEEDO documents
- * (~300k since 2018); Legals is keyset-streamed and filtered in memory to those
- * document_ids (chunked IN() 414s at ~250 ids, so keyset-stream is fewer requests).
+ * 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.
*
@@ -27,6 +30,13 @@ 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' };
@@ -61,7 +71,7 @@ export async function ingestNycAcris(): Promise<{ upserted: number }> {
const deed = new Map<string, { date: string; price: number }>();
let cursor = '';
for (;;) {
- const where = `document_amt > 0 AND doc_type in('DEED','DEEDO') ` +
+ 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',
@@ -71,7 +81,7 @@ export async function ingestNycAcris(): Promise<{ upserted: number }> {
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 > 0) deed.set(o.document_id, { date, price });
+ 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;
@@ -86,7 +96,8 @@ export async function ingestNycAcris(): Promise<{ upserted: number }> {
for (;;) {
const rows = await soda(LEGALS, {
$select: 'document_id,borough,block,lot',
- $where: `document_id > '${cursor}'`, $order: 'document_id', $limit: String(PAGE),
+ $where: `document_id > '${cursor}' AND document_id < '${DOC_CEIL}'`,
+ $order: 'document_id', $limit: String(PAGE),
});
if (!rows.length) break;
legalRows += rows.length;
@@ -108,38 +119,61 @@ export async function ingestNycAcris(): Promise<{ upserted: number }> {
console.log(`[acris] legals: ${legalRows} rows scanned, ${matched} deed-legal hits, ${salesByBbl.size} bbls`);
if (salesByBbl.size < 20000) throw new Error(`only ${salesByBbl.size} BBLs matched — join drift?`);
- // 3) Stage the sales into a temp table, then one set-based UPDATE onto PLUTO rows
- await query(`CREATE TEMP TABLE tmp_acris (fips text, bbl bigint, last_date date, last_price numeric, sales jsonb) ON COMMIT DROP`);
+ // 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) {
- sales.sort((a, b) => b.date.localeCompare(a.date));
- const top = sales.slice(0, 10);
+ // 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 })))]);
}
- 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 query(`INSERT INTO tmp_acris (fips,bbl,last_date,last_price,sales) VALUES ${vals.join(',')}`, params);
- }
- const upd = await 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 deeds (DEED/DEEDO, 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`);
- const upserted = Number(upd.rows[0].n);
+ 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', {
diff --git a/src/server/property.ts b/src/server/property.ts
index ef00db0..2a1f2c9 100644
--- a/src/server/property.ts
+++ b/src/server/property.ts
@@ -178,7 +178,7 @@ const NYC_META: PgCountyMeta = {
assessor: { name: 'NYC Department of Finance', phone: '311',
website: 'https://www.nyc.gov/site/finance/property/property.page' },
detailsNote: 'NYC MapPLUTO is lot-level — building sqft shown; unit-level beds/baths not in PLUTO.',
- historyNote: 'Recorded deed sales (price + date) from NYC ACRIS priced DEED/DEEDO documents (2018–present) where available; condo-unit deeds and buyer/seller names pending. PLUTO supplies current attributes.',
+ historyNote: 'Recorded sales (price + date) from NYC ACRIS priced transfer documents — deed family + RPTT transfer-tax returns (2018–present) — where available; condo-unit deeds and buyer/seller names pending. PLUTO supplies current attributes.',
taxNote: 'NYC assessed values from MapPLUTO (current version).',
ownerNote: 'Owner of record from NYC MapPLUTO.',
zoningNote: 'Primary zoning district (zonedist1) from NYC MapPLUTO.',
← 4ade9ca auto-save: 2026-07-21T18:39:12 (3 files) — src/ingest/parcel
·
back to Nationalrealestate
·
NYC ACRIS recorded sales -> PLUTO parcels (188k lots, 5 boro e3663e8 →