"""Recompute district economics using per-flat (Поквартирные/Лоты) data. Per-flat is way richer than monthly Сводные: - real velocity: registered DDU per month (last 12 mo) - real median price: from individual sold lots - real average area sold - готовность distribution - bank diversity (mortgage market attractiveness) - sold-out ratio (продано / total in projects with sales started) - sales freshness (median days since contract for last 90d) Replaces values in district_economics with `*_real` columns. """ import datetime as dt import pathlib import sqlite3 from collections import Counter DB = pathlib.Path(__file__).parent / "analysis.db" EXTRA = """ ALTER TABLE district_economics ADD COLUMN real_n_lots INTEGER; ALTER TABLE district_economics ADD COLUMN real_n_sold INTEGER; ALTER TABLE district_economics ADD COLUMN real_sold_pct REAL; ALTER TABLE district_economics ADD COLUMN real_median_price_m2 REAL; ALTER TABLE district_economics ADD COLUMN real_p25_price_m2 REAL; ALTER TABLE district_economics ADD COLUMN real_p75_price_m2 REAL; ALTER TABLE district_economics ADD COLUMN real_avg_area_sold REAL; ALTER TABLE district_economics ADD COLUMN real_velocity_per_month REAL; ALTER TABLE district_economics ADD COLUMN real_n_banks INTEGER; ALTER TABLE district_economics ADD COLUMN real_top_bank TEXT; ALTER TABLE district_economics ADD COLUMN real_top_bank_share REAL; ALTER TABLE district_economics ADD COLUMN real_avg_readiness_pct REAL; """ def safe_alter(conn, sql): for stmt in sql.strip().split(";"): s = stmt.strip() if not s: continue try: conn.execute(s) except sqlite3.OperationalError as e: if "duplicate column" not in str(e): raise def percentile(vals, p): if not vals: return None vals = sorted(vals) k = (len(vals) - 1) * p f = int(k) c = min(f + 1, len(vals) - 1) return vals[f] + (vals[c] - vals[f]) * (k - f) def _reg_date(it): """Parse register_date from a lot tuple; returns None on missing/malformed.""" r = it[6] if not r: return None try: return dt.date.fromisoformat(r[:10]) except Exception: return None def main(): conn = sqlite3.connect(DB) safe_alter(conn, EXTRA) conn.commit() # n_lots, n_sold, prices, areas, banks, readiness — all per district rows = conn.execute(""" SELECT district, status, sold, price_per_m2, area_pd, bank, readiness_pct, register_date FROM objective_lots WHERE district IS NOT NULL AND district!='' """).fetchall() by_d = {} for d, st, sold, price, area, bank, ready, reg in rows: by_d.setdefault(d, []).append((st, sold, price, area, bank, ready, reg)) today = dt.date.today() cutoff_12mo = today - dt.timedelta(days=365) for d, items in by_d.items(): # Apply the same 12-month window to n_lots/sold_pct/prices as velocity uses. # Keep unsold lots with no register_date (currently active in market). # Exclude lots whose DDU was registered before the cutoff (stale history). items_12mo = [ it for it in items if (rd := _reg_date(it)) is None or rd >= cutoff_12mo ] n_lots = len(items_12mo) sold_items = [it for it in items_12mo if (it[1] or '').strip().lower() == 'да'] n_sold = len(sold_items) sold_pct = 100.0 * n_sold / n_lots if n_lots else None prices = [it[2] for it in sold_items if it[2] and it[2] > 0] med = percentile(prices, 0.5) / 1000 if prices else None # → тыс ₽/м² p25 = percentile(prices, 0.25) / 1000 if prices else None p75 = percentile(prices, 0.75) / 1000 if prices else None areas = [it[3] for it in sold_items if it[3] and it[3] > 0] avg_area = sum(areas) / len(areas) if areas else None # velocity: registered deals in last 12 mo / 12 / n_corpuses_in_district # sold_items is already windowed to 12mo; the cutoff check is a safety guard. reg_dates = [] for it in sold_items: rd = _reg_date(it) if rd is not None and rd >= cutoff_12mo: reg_dates.append(rd) # Distinct corpuses in district (lifetime count — denominator for velocity) n_corp_real = ( len({it for it in conn.execute( "SELECT DISTINCT project, corpus FROM objective_lots WHERE district=?", (d,), ).fetchall()}) or 1 ) velocity = len(reg_dates) / 12.0 / n_corp_real if reg_dates else 0 # banks and readiness use full lifetime items — these reflect current market # structure (which banks are active, current construction readiness). banks = [it[4] for it in items if it[4] and it[4].strip()] unique_banks = set(banks) top_bank, top_share = None, None if banks: c = Counter(banks) top_bank, top_n = c.most_common(1)[0] top_share = top_n / len(banks) ready_vals = [it[5] for it in items if it[5] is not None] avg_ready = sum(ready_vals) / len(ready_vals) if ready_vals else None conn.execute( """UPDATE district_economics SET real_n_lots=?, real_n_sold=?, real_sold_pct=?, real_median_price_m2=?, real_p25_price_m2=?, real_p75_price_m2=?, real_avg_area_sold=?, real_velocity_per_month=?, real_n_banks=?, real_top_bank=?, real_top_bank_share=?, real_avg_readiness_pct=? WHERE district=?""", (n_lots, n_sold, sold_pct, med, p25, p75, avg_area, velocity, len(unique_banks), top_bank, top_share, avg_ready, d), ) conn.commit() print( f"{'район':<22}{'лот':>6}{'прод':>7}{'sold%':>7}" f"{'медцена':>9}{'площ':>6}{'vel':>6}{'банки':>6}{'top_bank':>22}{'%':>5}{'готовн':>7}" ) for r in conn.execute("""SELECT district, real_n_lots, real_n_sold, real_sold_pct, real_median_price_m2, real_avg_area_sold, real_velocity_per_month, real_n_banks, real_top_bank, real_top_bank_share, real_avg_readiness_pct FROM district_economics WHERE real_n_lots>0 ORDER BY real_median_price_m2 DESC NULLS LAST""").fetchall(): d, nl, ns, sp, mp, aa, v, nb, tb, ts, rd = r print( f"{d:<22}{nl:>6}{ns:>7}{sp or 0:>6.1f}%{mp or 0:>9.1f}" f"{aa or 0:>6.1f}{v:>6.2f}{nb or 0:>6}{(tb or '—')[:20]:>22}" f"{(ts or 0)*100:>4.0f}%{rd or 0:>6.0f}%" ) conn.close() if __name__ == "__main__": main()