`src/ingest/derive_metrics.ts`

86 lines
/**
 * Derived metrics -> metric_series source 'derived'.
 * For every county + metro:
 *   affordability = latest zhvi / ACS median_hh_income (annual ACS joins to any zhvi month)
 *   rent_yield    = (zori * 12) / zhvi at the latest common zhvi+zori period
 *   zhvi_yoy      = zhvi latest vs 12 months prior (fraction)
 * period stored = the zhvi period used. Idempotent upserts via the shared harness.
 */
import { pool, query } from '../../db/pool.ts';
import { openRun, closeRun } from './run.ts';
import { upsertMetricRows, type MetricRow } from './crosswalk.ts';

const SOURCE = 'derived';

interface SeriesRow { region_id: number; metric: string; period: string; value: string }

async function main() {
  const runId = await openRun(SOURCE);
  try {
    const r = await query<SeriesRow>(
      `SELECT ms.region_id, ms.metric, to_char(ms.period, 'YYYY-MM-DD') AS period, ms.value::text AS value
         FROM metric_series ms
         JOIN region rg ON rg.id = ms.region_id AND rg.region_type IN ('county','metro')
        WHERE (ms.source = 'zillow' AND ms.metric IN ('zhvi','zori'))
           OR (ms.source = 'acs' AND ms.metric = 'median_hh_income')`,
    );

    type Per = { zhvi: Map<string, number>; zori: Map<string, number>; income?: number };
    const byRegion = new Map<number, Per>();
    for (const row of r.rows) {
      let p = byRegion.get(row.region_id);
      if (!p) { p = { zhvi: new Map(), zori: new Map() }; byRegion.set(row.region_id, p); }
      const v = Number(row.value);
      if (row.metric === 'zhvi') p.zhvi.set(row.period, v);
      else if (row.metric === 'zori') p.zori.set(row.period, v);
      else p.income = v;
    }

    const rows: MetricRow[] = [];
    const monthsBack = (period: string, n: number) => {
      const [y, m] = period.split('-').map(Number);
      const total = y * 12 + (m - 1) - n;
      return `${Math.floor(total / 12)}-${String((total % 12) + 1).padStart(2, '0')}-01`;
    };

    for (const [regionId, p] of byRegion) {
      if (p.zhvi.size === 0) continue;
      const zhviPeriods = [...p.zhvi.keys()].sort();
      const latest = zhviPeriods[zhviPeriods.length - 1];
      const zhvi = p.zhvi.get(latest)!;
      if (!(zhvi > 0)) continue;

      if (p.income && p.income > 0)
        rows.push({ regionId, source: SOURCE, metric: 'affordability', period: latest, value: zhvi / p.income });

      // rent_yield at the latest period where BOTH zhvi and zori exist
      const common = [...p.zori.keys()].filter(k => p.zhvi.has(k)).sort();
      if (common.length) {
        const cp = common[common.length - 1];
        const z = p.zhvi.get(cp)!, rent = p.zori.get(cp)!;
        if (z > 0 && rent > 0)
          rows.push({ regionId, source: SOURCE, metric: 'rent_yield', period: cp, value: (rent * 12) / z });
      }

      const prior = p.zhvi.get(monthsBack(latest, 12));
      if (prior && prior > 0)
        rows.push({ regionId, source: SOURCE, metric: 'zhvi_yoy', period: latest, value: zhvi / prior - 1 });
    }

    if (rows.length === 0) throw new Error('derive produced 0 rows — are zillow + acs ingests loaded?');
    const upserted = await upsertMetricRows(rows, runId);
    const counts: Record<string, number> = {};
    for (const row of rows) counts[row.metric] = (counts[row.metric] || 0) + 1;
    await closeRun(runId, 'ok', { upserted, notes: JSON.stringify(counts) });
    console.log(`[derive] ok: ${upserted} rows`, counts);
  } catch (e) {
    await closeRun(runId, 'failed', { notes: String(e) });
    console.error('[derive] FAILED:', e);
    process.exitCode = 1;
  } finally {
    await pool.end();
  }
}

main();