""" migrate_weights_mysql.py ======================== One-time migration: creates the MySQL configuration tables and seeds them with the current SPS/BPS indicator weights and BPE amenity weights. Re-running is fully safe: uses ON DUPLICATE KEY UPDATE, so existing weights are overwritten with the seed values only if you re-run intentionally. Usage ----- python migrate_weights_mysql.py # seed with defaults python migrate_weights_mysql.py --dry-run # print SQL, do not execute Required env vars: MYSQL_USER, MYSQL_PASSWORD Optional env vars: MYSQL_HOST (localhost), MYSQL_PORT (3306), MYSQL_DATABASE (immobilier_config) """ from __future__ import annotations import argparse import logging import sys logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S", ) log = logging.getLogger("migrate") # --------------------------------------------------------------------------- # DDL # --------------------------------------------------------------------------- _DDL_INDICATOR_WEIGHTS = """\ CREATE TABLE IF NOT EXISTS indicator_weights ( id INT NOT NULL AUTO_INCREMENT, score_type VARCHAR(10) NOT NULL COMMENT 'SPS or BPS', indicator VARCHAR(50) NOT NULL COMMENT 'Raw indicator name as used in build_pression_iris.py', weight DECIMAL(8,6) NOT NULL COMMENT 'Contribution weight; weights per score_type must sum to 1.0', created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY uq_score_indicator (score_type, indicator) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='SPS/BPS indicator weights — read by build_pression_iris.py via domain.core.mysql_db';\ """ _DDL_AMENITY_WEIGHTS = """\ CREATE TABLE IF NOT EXISTS amenity_weights ( id INT NOT NULL AUTO_INCREMENT, typequ_prefix VARCHAR(10) NOT NULL COMMENT 'First 2 chars of INSEE BPE TYPEQU code', weight DECIMAL(8,4) NOT NULL COMMENT 'Multiplicative weight applied to equipment count', created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY uq_typequ_prefix (typequ_prefix) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='BPE equipment-type weights for amenity_score — read by build_pression_iris.py';\ """ # --------------------------------------------------------------------------- # Seed data (mirrors the former hardcoded constants) # --------------------------------------------------------------------------- # SPS weights must sum to 1.0 (22+18+20+15+15+5+5 = 100) # BPS weights must sum to 1.0 (17+18+15+15+10+9+9+4+3 = 100) # v6: split elderly signal (75+ only), added flood/noise to SPS; # first_buyer_ratio (25-44) and price_momentum added to BPS; # NPI centered by population-weighted dept mean. _SEED_INDICATOR_WEIGHTS: list[tuple[str, str, float]] = [ # ── SPS (Selling Pressure Score) ────────────────────────────────────── ("SPS", "very_elderly_ratio", 0.22), # pop 75+ / pop total (near-term sellers) ("SPS", "elderly_owner", 0.18), # (pop75+/pop) × ownership_rate ("SPS", "vacancy_rate", 0.20), # logements vacants / total ("SPS", "pop_decline", 0.15), # pop decline RP2018→2022 [exponential] ("SPS", "poverty_rate", 0.15), # RSA beneficiaires / personnes couvertes ("SPS", "flood_risk", 0.05), # worst flood scenario at IRIS centroid ("SPS", "noise_level", 0.05), # LDEN-weighted proximity to roads/rail # ── BPS (Buying Pressure Score) ─────────────────────────────────────── ("BPS", "income_level", 0.17), # revenu disponible median (Filosofi 2021) ("BPS", "first_buyer_ratio", 0.18), # pop 25-44 / pop total (peak buyer demographic) ("BPS", "security_score", 0.15), # 10 - danger_score (inverted) ("BPS", "employment_rate", 0.15), # actifs occupes / (actifs + DEFM) ("BPS", "price_momentum", 0.10), # CAGR of prix_m2 over DVF years ("BPS", "diploma_rate", 0.09), # bac+2 and above / pop 15+ non scolarisee ("BPS", "greenery_pct", 0.09), # OSM park/forest fraction of IRIS area ("BPS", "amenity_score", 0.04), # weighted BPE equipment count ("BPS", "transaction_rate", 0.03), # avg annual DVF transactions (market liquidity) ] _SEED_AMENITY_WEIGHTS: list[tuple[str, float]] = [ ("D1", 2.0), # medecins, infirmiers, kinesithe ("D2", 3.0), # hopitaux, cliniques, urgences ("D3", 1.0), # autres soins et services de sante ("C2", 2.0), # ecoles maternelles ("C3", 2.0), # ecoles elementaires ("C4", 1.5), # colleges ("C5", 1.5), # lycees ("F1", 3.0), # gares SNCF ("F3", 3.0), # stations metro / tramway ("A2", 1.0), # superettes, epiceries ("A3", 1.5), # supermarches, hypermarches ] # --------------------------------------------------------------------------- # Migration # --------------------------------------------------------------------------- def migrate(dry_run: bool = False) -> None: if dry_run: log.info("-- DRY RUN: no changes will be committed --") for ddl in (_DDL_INDICATOR_WEIGHTS, _DDL_AMENITY_WEIGHTS): print(ddl) print() log.info("Would insert %d indicator_weights rows", len(_SEED_INDICATOR_WEIGHTS)) log.info("Would insert %d amenity_weights rows", len(_SEED_AMENITY_WEIGHTS)) return from domain.core.mysql_db import get_connection conn = get_connection() try: with conn.cursor() as cur: log.info("Creating table: indicator_weights") cur.execute(_DDL_INDICATOR_WEIGHTS) log.info("Creating table: amenity_weights") cur.execute(_DDL_AMENITY_WEIGHTS) log.info("Replacing indicator_weights (%d rows)…", len(_SEED_INDICATOR_WEIGHTS)) cur.execute("DELETE FROM indicator_weights") cur.executemany( "INSERT INTO indicator_weights (score_type, indicator, weight) " "VALUES (%s, %s, %s)", _SEED_INDICATOR_WEIGHTS, ) log.info("Replacing amenity_weights (%d rows)…", len(_SEED_AMENITY_WEIGHTS)) cur.execute("DELETE FROM amenity_weights") cur.executemany( "INSERT INTO amenity_weights (typequ_prefix, weight) " "VALUES (%s, %s)", _SEED_AMENITY_WEIGHTS, ) conn.commit() log.info("Migration committed.") # ── Verify ──────────────────────────────────────────────────────── with conn.cursor() as cur: cur.execute( "SELECT score_type, COUNT(*) AS n, SUM(weight) AS total " "FROM indicator_weights GROUP BY score_type ORDER BY score_type" ) for row in cur.fetchall(): log.info( " indicator_weights[%s]: %d indicators, weight sum = %.4f", row["score_type"], row["n"], float(row["total"]), ) cur.execute("SELECT COUNT(*) AS n FROM amenity_weights") log.info(" amenity_weights: %d rows", cur.fetchone()["n"]) finally: conn.close() if __name__ == "__main__": parser = argparse.ArgumentParser(description="Seed MySQL config tables for build_pression_iris.py") parser.add_argument("--dry-run", action="store_true", help="Print DDL/data without executing") args = parser.parse_args() try: migrate(dry_run=args.dry_run) except RuntimeError as exc: log.error("%s", exc) sys.exit(1)