`scripts/derive-sale-prices.py`

67 lines
#!/usr/bin/env python3
# derive-sale-prices.py — RESELLABLE sale-price dataset from PUBLIC RECORDS ($0, no scraping).
#
# Doctrine (docs/SOURCING.md): resell only public-record data. This derives an estimated SALE PRICE
# per parcel from the LA County Assessor roll (public) using Prop 13: on a change of ownership the
# property is reassessed to fair market value (~= purchase price), then grows <=2%/yr. So:
#     est_sale_price = roll_total_value / (1.02 ** (roll_year - transfer_year))
# de-escalates the assessed value back to the transfer year ~= the sale price. Official public figure.
#
# Writes table `derived_sale` into data/assessor.sqlite. Re-run safe (drops+rebuilds). $0, local.
#   python3 scripts/derive-sale-prices.py
import sqlite3, os
DB = os.path.join(os.path.dirname(__file__), '..', 'data', 'assessor.sqlite')
c = sqlite3.connect(DB)
ROLL_YEAR = int(list(c.execute('SELECT max(CAST(roll_year AS INT)) FROM assessor_parcel'))[0][0] or 2025)

c.executescript('''
DROP TABLE IF EXISTS derived_sale;
CREATE TABLE derived_sale (
  ain TEXT, address TEXT, city TEXT, use_desc TEXT, units INTEGER, year_built INTEGER,
  transfer_date TEXT, transfer_year INTEGER, assessed_value INTEGER, est_sale_price INTEGER,
  price_source TEXT, confidence TEXT
);
''')

rows = c.execute('''
  SELECT ain, situs_house_no||' '||situs_street AS address, property_location, use_desc1, units,
         year_built, recording_date, roll_total_value
    FROM assessor_parcel
   WHERE recording_date GLOB '[12][09][0-9][0-9]-*'
     AND CAST(substr(recording_date,1,4) AS INT) BETWEEN 2015 AND ?
     AND roll_total_value > 50000
''', (ROLL_YEAR,))

ins, n = [], 0
for ain, addr, ploc, use, units, yb, rdate, val in rows:
    ty = int(rdate[:4])
    years = max(0, ROLL_YEAR - ty)
    est = round(val / (1.02 ** years))
    # confidence: fresher transfer + residential/multifamily = tighter est_sale ~= price;
    # older or non-standard use = looser (new construction / partial transfers add noise).
    conf = 'high' if years <= 2 else ('med' if years <= 5 else 'low')
    # Cody-gate fix (2026-07-31): if the building was built AFTER the transfer, the current roll value
    # includes construction that didn't exist at sale -> de-escalation OVERSTATES the sale price. Can't
    # estimate a sale price honestly here; null it + flag, so it never ships as a "sale".
    if yb and ty and yb > ty:
        est = None; conf = 'na'; psrc = 'assessor_value_only_post_transfer_build'
    else:
        psrc = 'assessor_prop13_reassessment'
    ins.append((ain, (addr or '').strip() or (ploc or '').strip(), None, use, units, yb, rdate, ty, val, est, psrc, conf))
    n += 1
    if len(ins) >= 5000:
        c.executemany('INSERT INTO derived_sale VALUES (?,?,?,?,?,?,?,?,?,?,?,?)', ins); ins = []
if ins: c.executemany('INSERT INTO derived_sale VALUES (?,?,?,?,?,?,?,?,?,?,?,?)', ins)
c.execute('CREATE INDEX idx_ds_ain ON derived_sale(ain)')
c.execute('CREATE INDEX idx_ds_year ON derived_sale(transfer_year)')
c.commit()

tot = list(c.execute('SELECT count(*) FROM derived_sale'))[0][0]
hi = list(c.execute("SELECT count(*) FROM derived_sale WHERE confidence='high'"))[0][0]
mf = list(c.execute("SELECT count(*) FROM derived_sale WHERE units>=2 AND transfer_year>=2023"))[0][0]
print(f'derived_sale: {tot:,} rows  (high-confidence {hi:,}; multifamily 2023+ {mf:,})  roll_year={ROLL_YEAR}')
print('\nsample recent multifamily sales (est from public assessor roll):')
for r in c.execute("SELECT address, units, transfer_date, assessed_value, est_sale_price, confidence FROM derived_sale WHERE units>=5 AND transfer_year>=2024 ORDER BY est_sale_price DESC LIMIT 6"):
    print(f'  {r[0][:34]:34s} {r[1]:>3}u  {r[2]}  assessed ${r[3]:>12,}  est_sale ${r[4]:>12,}  [{r[5]}]')
c.close()