`src/ingest/redfin_tracker.ts`

204 lines
/**
 * Redfin market-tracker ingest — county + metro + state, gzipped full-history TSVs.
 * NEVER buffers a whole file: createReadStream -> gunzip -> readline, flat memory.
 * Filters: PROPERTY_TYPE='All Residential', IS_SEASONALLY_ADJUSTED=false (bare boolean),
 * PERIOD_DURATION=30, PERIOD_BEGIN within the last 36 months.
 * County match = REGION name ("Los Angeles County, CA") -> Gazetteer name+state;
 * metro match = TABLE_ID (verified to be the CBSA code) -> region.cbsa_code;
 * state match = STATE_CODE. Overrides consulted first; unmatched recorded + skipped.
 * Run with --cached to reuse data/downloads copies during dev.
 */
import { createReadStream } from 'node:fs';
import { createGunzip } from 'node:zlib';
import { createInterface } from 'node:readline';
import { openRun, closeRun, download } from './run.ts';
import {
  loadRegionMaps, loadOverrides, buildMetroMatcher, UnmatchedCollector,
  upsertMetricRows, upsertSourceRegionMap, monthsAgoCutoff,
  type MetricRow, type RegionMaps, normName,
} from './crosswalk.ts';
import { pool } from '../../db/pool.ts';

const BASE = 'https://redfin-public-data.s3.us-west-2.amazonaws.com/redfin_market_tracker';
const FILES = [
  { url: `${BASE}/county_market_tracker.tsv000.gz`, file: 'redfin_county.tsv.gz', level: 'county' as const },
  { url: `${BASE}/redfin_metro_market_tracker.tsv000.gz`, file: 'redfin_metro.tsv.gz', level: 'metro' as const },
  { url: `${BASE}/state_market_tracker.tsv000.gz`, file: 'redfin_state.tsv.gz', level: 'state' as const },
];

const METRIC_COLS: [string, string][] = [
  ['median_sale_price', 'median_sale_price'],
  ['median_sale_price_yoy', 'median_sale_price_yoy'],
  ['homes_sold', 'homes_sold'],
  ['inventory', 'inventory'],
  ['new_listings', 'new_listings'],
  ['median_dom', 'dom'],
  ['avg_sale_to_list', 'sale_to_list'],
  ['months_of_supply', 'months_of_supply'],
];

const unq = (s: string) => (s.length >= 2 && s[0] === '"' && s[s.length - 1] === '"' ? s.slice(1, -1) : s);

interface Ctx {
  maps: RegionMaps;
  matchMetro: ReturnType<typeof buildMetroMatcher>;
  overrides: Record<string, string>;
  unmatched: UnmatchedCollector;
  cutoff: string;
  runId: number;
}

async function processFile(path: string, level: 'county' | 'metro' | 'state', ctx: Ctx): Promise<{ upserted: number; skipped: number }> {
  const rl = createInterface({ input: createReadStream(path).pipe(createGunzip()), crlfDelay: Infinity });
  let header: string[] | null = null;
  let idx: Record<string, number> = {};
  let lineNo = 0;
  let buffer: MetricRow[] = [];
  let mapBuffer: { source: string; sourceRegionId: string; regionId: number; matchMethod: string }[] = [];
  let upserted = 0;
  let skipped = 0;
  const seenUnmatched = new Set<string>();
  const regionCache = new Map<string, { regionId: number; method: string } | null>();

  const flush = async () => {
    if (mapBuffer.length) { await upsertSourceRegionMap(mapBuffer); mapBuffer = []; }
    if (buffer.length) { upserted += await upsertMetricRows(buffer, ctx.runId); buffer = []; }
  };

  for await (const raw of rl) {
    lineNo++;
    if (lineNo % 100_000 === 0) console.log(`[redfin:${level}] ${lineNo.toLocaleString()} lines, ${upserted.toLocaleString()} rows upserted`);
    if (!raw) continue;
    const f = raw.split('\t');
    if (!header) {
      header = f.map((h) => unq(h).toLowerCase());
      header.forEach((h, i) => { idx[h] = i; });
      for (const req of ['period_begin', 'property_type', 'region', 'table_id']) {
        if (!(req in idx)) throw new Error(`redfin ${level}: missing column ${req}`);
      }
      continue;
    }
    const get = (c: string) => (idx[c] !== undefined ? unq(f[idx[c]] ?? '') : '');

    if (get('property_type') !== 'All Residential') continue;
    const sa = get('is_seasonally_adjusted');
    if (sa && sa !== 'false' && sa !== 'f') continue;
    const dur = get('period_duration');
    if (dur && dur !== '30') continue;
    const periodBegin = get('period_begin');
    if (!periodBegin || periodBegin < ctx.cutoff) continue;

    const region = get('region');
    const tableId = get('table_id');
    const stateCode = get('state_code');
    const cacheKey = tableId || region;
    let match = regionCache.get(cacheKey);
    if (match === undefined) {
      match = resolveRegion(level, region, tableId, stateCode, ctx);
      regionCache.set(cacheKey, match);
      if (match) mapBuffer.push({ source: 'redfin', sourceRegionId: tableId || region, regionId: match.regionId, matchMethod: match.method });
    }
    if (!match) {
      skipped++;
      if (!seenUnmatched.has(cacheKey)) {
        seenUnmatched.add(cacheKey);
        ctx.unmatched.add(`redfin_${level}`, { tableId, region, stateCode });
      }
      continue;
    }

    const period = periodBegin.slice(0, 7) + '-01';
    for (const [colName, metric] of METRIC_COLS) {
      if (idx[colName] === undefined) continue;
      const v = unq(f[idx[colName]] ?? '');
      if (v === '') continue;
      const num = Number(v);
      if (!Number.isFinite(num)) continue;
      buffer.push({ regionId: match.regionId, source: 'redfin', metric, period, value: num });
    }
    if (buffer.length >= 5000) await flush();
  }
  await flush();
  console.log(`[redfin:${level}] done: ${lineNo.toLocaleString()} lines, ${upserted.toLocaleString()} rows upserted, ${skipped} row-skips (unmatched regions: ${seenUnmatched.size})`);
  return { upserted, skipped };
}

function resolveRegion(level: string, region: string, tableId: string, stateCode: string, ctx: Ctx): { regionId: number; method: string } | null {
  const ovr = ctx.overrides[tableId] || ctx.overrides[region];
  if (ovr) {
    const r = ctx.maps.byCanonicalKey.get(ovr);
    if (r) return { regionId: r.id, method: 'override' };
  }
  if (level === 'state') {
    const r = ctx.maps.byStateCode.get(stateCode);
    return r ? { regionId: r.id, method: 'state' } : null;
  }
  if (level === 'metro') {
    const byCbsa = ctx.maps.byCbsa.get(tableId);
    if (byCbsa) return { regionId: byCbsa.id, method: 'cbsa' };
    const name = region.replace(/\s+metro area$/i, '');
    const m = ctx.matchMetro(name);
    return m ? { regionId: m.region.id, method: m.method } : null;
  }
  // county: "Los Angeles County, CA" -> name + state
  const comma = region.lastIndexOf(',');
  if (comma < 0) return null;
  const name = region.slice(0, comma);
  const st = region.slice(comma + 1).trim();
  const r = ctx.maps.byCountyNameState.get(normName(name) + '|' + st);
  return r ? { regionId: r.id, method: 'name' } : null;
}

async function main() {
  const cached = process.argv.includes('--cached');
  const runId = await openRun('redfin', BASE);
  try {
    const maps = await loadRegionMaps();
    const ctx: Ctx = {
      maps,
      matchMetro: buildMetroMatcher(maps),
      overrides: loadOverrides('redfin'),
      unmatched: new UnmatchedCollector(),
      cutoff: monthsAgoCutoff(36),
      runId,
    };
    console.log(`[redfin] cutoff period_begin >= ${ctx.cutoff}`);

    let totalUpserted = 0;
    let totalSkipped = 0;
    const hashes: string[] = [];
    for (const spec of FILES) {
      console.log(`[redfin] downloading ${spec.file}${cached ? ' (cached ok)' : ''}`);
      const { path, sha256 } = await download(spec.url, spec.file, { reuseCached: cached });
      hashes.push(`${spec.file}:${sha256.slice(0, 12)}`);
      const r = await processFile(path, spec.level, ctx);
      totalUpserted += r.upserted;
      totalSkipped += r.skipped;
    }
    ctx.unmatched.flush();
    const uc = ctx.unmatched.count('redfin_county');
    const um = ctx.unmatched.count('redfin_metro');
    const us = ctx.unmatched.count('redfin_state');
    console.log(`[redfin] TOTAL upserted=${totalUpserted} unmatched: county=${uc} metro=${um} state=${us}`);

    if (totalUpserted === 0) {
      await closeRun(runId, 'failed', { notes: '0 rows parsed — format drift?' });
      process.exitCode = 1;
    } else {
      await closeRun(runId, 'ok', {
        upserted: totalUpserted, skipped: totalSkipped,
        notes: `unmatched_county=${uc} unmatched_metro=${um} unmatched_state=${us}`,
        fileHash: hashes.join(' '),
      });
    }
  } catch (e) {
    console.error('[redfin] FAILED', e);
    await closeRun(runId, 'failed', { notes: String((e as Error).message || e) });
    process.exitCode = 1;
  } finally {
    await pool.end();
  }
}

main();