← back to Nationalrealestate
src/ingest/parcels/hennepin_mn.ts
202 lines
/**
* Hennepin County MN (Minneapolis, FIPS 27053) parcel ingest — dedicated
* full-universe adapter (applies the Wake/Miami-Dade DTD verdict B precedent).
*
* Source: the Minnesota Metropolitan Council "Metropolitan 7-County Parcels"
* aggregate FeatureServer (keyless, supportsPagination, 2k/page, $0):
* https://arcgis.metc.state.mn.us/data1/rest/services/parcels/Parcels_Aggregate/FeatureServer/0
* This is a 7-county metro aggregate — we filter server-side to CO_NAME='HENNEPIN'
* (= 447,044 parcels). The Met Council standardizes every metro county's assessor
* roll into one schema, so this is the authoritative full-universe Hennepin feed.
*
* A rich PRICED-SALE county: OWNER_NAME (100%), situs address (assembled from the
* ANUMBER + ST_ name parts), CTU_NAME (city) + ZIP, EMV_LAND + EMV_BLDG + EMV_TOTAL
* (Estimated Market Value — TRUE market, ~95%), TAX_YEAR, YEAR_BUILT (~94%),
* NUM_UNITS, USECLASS1 (use), SALE_VALUE + SALE_DATE (a real priced last-sale,
* ~87%). Note: the aggregate's FIN_SQ_FT is not populated for Hennepin (0 rows) —
* sqft is left NULL rather than mapped from an empty column.
*
* source_id = PIN (unique within Hennepin — count(PIN)=447,044=total).
*
* Run: NODE_OPTIONS=--max-old-space-size=4096 npm run ingest:parcels hennepin
*/
import { pool, query } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
import { upsertParcels, registerParcelSource, registerSourceFieldMap, normAddress, type ParcelUpsertRow } from './upsert.ts';
import { sanitizeEventDate } from './date_guard.ts';
import { sanitizeSalePrice } from './price_guard.ts';
const FIPS = '27053';
const SOURCE_KEY = 'hennepin_mn';
const LAYER = 'https://arcgis.metc.state.mn.us/data1/rest/services/parcels/Parcels_Aggregate/FeatureServer/0';
const WHERE = "CO_NAME='HENNEPIN'";
const PAGE = 2000;
// TK-50 field-level provenance. total_value maps from EMV_TOTAL (Estimated Market
// Value — true market). This feed carries a priced sale (SALE_VALUE + SALE_DATE).
const FIELD_MAP: Record<string, string> = {
source_id: 'PIN', address: 'ANUMBER+ST_NAME', city: 'CTU_NAME', zip: 'ZIP',
year_built: 'YEAR_BUILT', units: 'NUM_UNITS', use_desc: 'USECLASS1', tax_year: 'TAX_YEAR',
land_value: 'EMV_LAND', improvement_value: 'EMV_BLDG', total_value: 'EMV_TOTAL',
owner_name: 'OWNER_NAME', last_sale_price: 'SALE_VALUE', last_sale_date: 'SALE_DATE',
};
const s = (v: unknown): string | null => { const t = v == null ? '' : String(v).trim(); return t ? t : null; };
const num = (v: unknown): number | null => { const x = Number(v); return Number.isFinite(x) && x !== 0 ? x : null; };
// SALE_DATE is an esri date (epoch-ms) — reject <=0 and sub-1971 near-epoch
// artifacts, then run the shared future/pre-1900/non-calendar guard.
const epochToISO = (v: unknown): string | null => { const x = Number(v); if (!Number.isFinite(x) || x < 31_536_000_000) return null; return sanitizeEventDate(new Date(x).toISOString().slice(0, 10)); };
/** Assemble the situs address from the standardized address parts, collapsing the
* gaps left by empty prefix/suffix/direction components. */
function situsAddr(a: any): string | null {
const parts = [a.ANUMBERPRE, a.ANUMBER, a.ANUMBERSUF, a.ST_PRE_DIR, a.ST_PRE_TYP, a.ST_NAME, a.ST_POS_TYP, a.ST_POS_DIR]
.map(s).filter((x): x is string => !!x);
const joined = parts.join(' ').replace(/\s+/g, ' ').trim();
return joined || null;
}
/** plain vertex-average of ring[0] — NOT a true polygon centroid (fine for a map
* pin; multipart parcels only get the first ring). */
function centroid(geom: any): [number | null, number | null] {
const ring = geom?.rings?.[0]; if (!Array.isArray(ring) || !ring.length) return [null, null];
let sx = 0, sy = 0; for (const [x, y] of ring) { sx += x; sy += y; }
return [+(sx / ring.length).toFixed(6), +(sy / ring.length).toFixed(6)];
}
const OUT = ['PIN', 'OWNER_NAME', 'OWNER_MORE', 'ANUMBERPRE', 'ANUMBER', 'ANUMBERSUF',
'ST_PRE_DIR', 'ST_PRE_TYP', 'ST_NAME', 'ST_POS_TYP', 'ST_POS_DIR', 'CTU_NAME', 'ZIP',
'EMV_LAND', 'EMV_BLDG', 'EMV_TOTAL', 'TAX_YEAR', 'YEAR_BUILT', 'NUM_UNITS',
'USECLASS1', 'HOME_STYLE', 'SALE_VALUE', 'SALE_DATE', 'ABB_LEGAL'].join(',');
async function fetchPage(offset: number): Promise<any[]> {
const u = new URL(LAYER + '/query');
u.searchParams.set('where', WHERE);
u.searchParams.set('outFields', OUT);
u.searchParams.set('orderByFields', 'OBJECTID ASC');
u.searchParams.set('resultOffset', String(offset));
u.searchParams.set('resultRecordCount', String(PAGE));
u.searchParams.set('returnGeometry', 'true');
u.searchParams.set('outSR', '4326');
u.searchParams.set('f', 'json');
let lastErr: any;
for (let a = 0; a < 3; a++) {
try {
const res = await fetch(u, { headers: { 'User-Agent': 'Mozilla/5.0 (usre-parcel-ingest)' }, signal: AbortSignal.timeout(120_000) });
if (!res.ok) throw new Error(`hennepin ${res.status}: ${(await res.text()).slice(0, 140)}`);
const j: any = await res.json();
if (j.error) throw new Error(`hennepin error: ${JSON.stringify(j.error).slice(0, 140)}`);
return j.features || [];
} catch (e) { lastErr = e; if (a < 2) await new Promise(r => setTimeout(r, 2000 * (a + 1))); }
}
throw lastErr;
}
export async function ingestHennepin(opts: { maxPages?: number } = {}): Promise<{ upserted: number }> {
const runId = await openRun('parcel_hennepin_mn', LAYER);
const fetchedAt = new Date().toISOString();
try {
await registerSourceFieldMap(SOURCE_KEY, FIELD_MAP, 'Hennepin County MN via MN Met Council Metropolitan 7-County Parcels aggregate');
let offset = 0, page = 0, seen = 0, upserted = 0, sales = 0;
for (;;) {
const feats = await fetchPage(offset);
if (!feats.length) break;
const batch: ParcelUpsertRow[] = [];
const events: { pin: string; price: number; date: string }[] = [];
for (const f of feats) {
const a = f.attributes || {};
const pin = s(a.PIN);
if (!pin) continue;
seen++;
const [lng, lat] = centroid(f.geometry);
const addr = situsAddr(a);
const norm = addr ? normAddress(addr) : null;
const salePrice = sanitizeSalePrice(a.SALE_VALUE);
const saleDate = epochToISO(a.SALE_DATE);
batch.push({
county_fips: FIPS, source_id: pin,
address: norm, norm_address: norm,
city: s(a.CTU_NAME), zip: s(a.ZIP),
lat, lng,
// FIN_SQ_FT is not populated for Hennepin in this aggregate — leave sqft NULL.
year_built: num(a.YEAR_BUILT), sqft: null, beds: null, baths: null, units: num(a.NUM_UNITS),
use_desc: s(a.USECLASS1),
land_value: num(a.EMV_LAND), improvement_value: num(a.EMV_BLDG), total_value: num(a.EMV_TOTAL),
tax_year: s(a.TAX_YEAR), owner_name: s(a.OWNER_NAME), zoning: null,
last_sale_date: saleDate, last_sale_price: salePrice,
extra: JSON.stringify({
owner_more: s(a.OWNER_MORE) || undefined,
home_style: s(a.HOME_STYLE) || undefined,
legal: s(a.ABB_LEGAL) || undefined,
note: 'Hennepin County MN via MN Met Council 7-county aggregate — EMV (estimated market value) + priced last-sale + year; sqft (FIN_SQ_FT) not populated for Hennepin in this feed',
}),
// ── TK-50 field-level provenance (record-level: one feature → this row) ──
sourceKey: SOURCE_KEY,
sourceUrl: `${LAYER}/query?where=${encodeURIComponent(`PIN='${pin}'`)}&outFields=*&f=html`,
fetchedAt,
rawSource: JSON.stringify(a),
});
if (salePrice && saleDate) events.push({ pin, price: salePrice, date: saleDate });
}
upserted += await upsertParcels(batch);
// priced last-sale events (amount = SALE_VALUE). doc_number = sale date (this
// aggregate carries no deed book/page).
for (let i = 0; i < events.length; i += 500) {
const chunk = events.slice(i, i + 500);
await query(
`INSERT INTO parcel_event (county_fips, source_id, event_type, event_date, amount, doc_type, doc_number, source, source_url, detail)
VALUES ${chunk.map((_, j) => { const b = j * 9; return `($${b + 1},$${b + 2},'sale',$${b + 3}::date,$${b + 4},$${b + 5},$${b + 6},$${b + 7},$${b + 8},$${b + 9}::jsonb)`; }).join(',')}
ON CONFLICT (county_fips, source_id, event_type, event_date, doc_number) DO NOTHING`,
chunk.flatMap(e => [FIPS, e.pin, e.date, e.price, 'Deed', e.date,
'parcel_hennepin_mn', `${LAYER}/query?where=${encodeURIComponent(`PIN='${e.pin}'`)}&outFields=*&f=html`, JSON.stringify({})]),
);
sales += chunk.length;
}
offset += feats.length;
page++;
if (seen % 100000 < feats.length) console.log(`[hennepin] ${upserted} upserted, ${sales} sales`);
if (opts.maxPages && page >= opts.maxPages) break;
if (feats.length < PAGE) break;
}
let bulk = 0;
if (!opts.maxPages) {
if (upserted < 400000) throw new Error(`only ${upserted} Hennepin parcels (expected ~447k) — layer/paging/filter drift?`);
// in-DB assemblage guard: this feed has no deed book/page, so a portfolio /
// multi-parcel sale stamps the SAME (SALE_DATE, SALE_VALUE) onto every parcel
// in the deal (verified: 25 Hennepin PINs share 2021-06 @ $2.3M). Without this
// each parcel would record the full price → N-fold inflated sales. Any exact
// (last_sale_date, last_sale_price) shared by >1 parcel is un-allocatable →
// null the price + flag bulk.
// ANALYZE first: after bulk-loading 447k rows into the multi-million-row prod
// parcel table, stale planner stats made the `county_fips=$1` filter SEQ-SCAN
// the whole table (a ~4h prod hang, 2026-08-11). Fresh stats let the
// (county_fips, source_id) PK index range-scan just this county's rows, and
// dropping the intermediate CTE removes a second full scan.
await query(`ANALYZE parcel`);
const b = await query(
`WITH grp AS (
SELECT last_sale_date, last_sale_price, COUNT(*)::int AS cnt
FROM parcel WHERE county_fips=$1 AND last_sale_price IS NOT NULL AND last_sale_date IS NOT NULL
GROUP BY 1,2 HAVING COUNT(*) > 1)
UPDATE parcel p SET last_sale_price = NULL,
extra = p.extra || jsonb_build_object('bulk_sale', true, 'bulk_parcel_count', grp.cnt)
FROM grp WHERE p.county_fips=$1
AND p.last_sale_date = grp.last_sale_date AND p.last_sale_price = grp.last_sale_price
RETURNING 1`, [FIPS]);
bulk = b.rowCount ?? 0;
const n = await registerParcelSource(FIPS, LAYER,
`Hennepin County MN (Minneapolis) via MN Met Council Metropolitan 7-County Parcels aggregate (CO_NAME='HENNEPIN' filter) — ${upserted} parcels: owner (100%), situs address + city + zip, EMV land+bldg+TOTAL (estimated MARKET value, ~95%), year built (~94%), units, use class, priced last-sale (SALE_VALUE + SALE_DATE, ~87%) → ${sales} parcel_event sale rows. sqft (FIN_SQ_FT) not populated for Hennepin in this aggregate. No beds/baths or zoning. ${bulk} multi-parcel (assemblage/portfolio) sales flagged bulk + price nulled.`);
console.log(`[hennepin] registry ${n}, ${bulk} bulk-flagged`);
}
await closeRun(runId, 'ok', { upserted, notes: `Hennepin MN: ${upserted} parcels, ${sales} sales, ${bulk} bulk${opts.maxPages ? ' (bounded verify)' : ''}` });
console.log(`[hennepin] ok: ${upserted} parcels, ${sales} sales, ${bulk} bulk${opts.maxPages ? ` (bounded ${opts.maxPages}p)` : ''}`);
return { upserted };
} catch (e: any) {
await closeRun(runId, 'failed', { notes: String(e.message || e).slice(0, 500) });
throw e;
}
}
if (import.meta.url === `file://${process.argv[1]}`) {
ingestHennepin().then(() => pool.end()).catch(e => { console.error(e); process.exit(1); });
}