"""Generate JSON + HTML comparison report.""" import sqlite3, pathlib, json, statistics DB = pathlib.Path(__file__).parent / "analysis.db" OUT = pathlib.Path(__file__).parent / "reports" OUT.mkdir(exist_ok=True) PARCEL_ID = "parcel:66:41:0204016:10" # Source of truth for scoring weights (mirrors 03_score.py WEIGHTS / scoring_weights table). # economic=0.30 is the heaviest component — must be reported, not hidden. DEFAULT_WEIGHTS = { "education": 0.20, "health": 0.10, "retail": 0.15, "transit": 0.15, "leisure": 0.10, "economic": 0.30, } WEIGHT_LABELS = { "education": "образование", "health": "здоровье", "retail": "ритейл", "transit": "транспорт", "leisure": "досуг", "economic": "экономика", "market": "рынок", } def load_weights(conn): """Single source of truth for weights used in both JSON and HTML. Prefer the scoring_weights table (written by 10_score_v2.py); fall back to DEFAULT_WEIGHTS (== 03_score.py WEIGHTS). Includes economic (and market for v2). """ try: rows = conn.execute("SELECT component, weight FROM scoring_weights").fetchall() except sqlite3.OperationalError: rows = None if rows: return {c: float(w) for c, w in rows} return dict(DEFAULT_WEIGHTS) def fetch_economics(conn, site_id): row = conn.execute("""SELECT sd.district, sd.method, sd.nearest_jk_dist_m, de.n_projects, de.weighted_price_m2, de.median_price_m2, de.deals_per_month_avg, de.months_to_sellout, de.real_n_lots, de.real_n_sold, de.real_sold_pct, de.real_median_price_m2, de.real_p25_price_m2, de.real_p75_price_m2, de.real_avg_area_sold, de.real_velocity_per_month, de.real_avg_readiness_pct FROM site_district sd LEFT JOIN district_economics de USING (district) WHERE sd.site_id=?""", (site_id,)).fetchone() if not row: return None cols = ["district","district_method","district_dist_m", "n_projects","weighted_price_m2_corp_sum","median_price_m2_corp_sum", "deals_per_month_corp_sum","months_to_sellout", "n_lots","n_sold","sold_pct", "median_price_m2","p25_price_m2","p75_price_m2", "avg_area_sold_m2","velocity_per_month","avg_readiness_pct"] return dict(zip(cols, row)) def fetch_site_full(conn, site_id): s = dict(zip( [c[0] for c in conn.execute("PRAGMA table_info(sites)").fetchall()], conn.execute("SELECT * FROM sites WHERE site_id=?", (site_id,)).fetchone() )) if False else None cur = conn.execute("SELECT * FROM sites WHERE site_id=?", (site_id,)) cols = [c[0] for c in cur.description] s = dict(zip(cols, cur.fetchone())) s["features"] = dict(conn.execute( "SELECT feature,value FROM features WHERE site_id=?", (site_id,)).fetchall()) s["scores"] = dict(conn.execute( "SELECT component,score_0_100 FROM scores WHERE site_id=?", (site_id,)).fetchall()) tot = conn.execute( "SELECT weighted,rank_overall,rank_district FROM scores_total WHERE site_id=?", (site_id,)).fetchone() s["weighted_total"] = tot[0] s["rank_overall"] = tot[1] s["rank_district"] = tot[2] s["economics"] = fetch_economics(conn, site_id) s["nearest_pois"] = {} for cat in ["kindergarten","school","university","pharmacy","clinic","hospital", "shop_big","shop_med","shop_small","bus_stop","tram_stop","metro", "park","playground","sports"]: row = conn.execute( """SELECT name, distance_m, lat, lon FROM pois WHERE site_id=? AND category=? ORDER BY distance_m LIMIT 5""", (site_id, cat)).fetchall() s["nearest_pois"][cat] = [ {"name": r[0] or "—", "distance_m": round(r[1], 1), "lat": r[2], "lon": r[3]} for r in row ] return s def main(): conn = sqlite3.connect(DB) parcel = fetch_site_full(conn, PARCEL_ID) n_total = conn.execute("SELECT count(*) FROM sites").fetchone()[0] weights = load_weights(conn) # Top-10 best-scoring ЖК + parcel position context rows = conn.execute("""SELECT s.site_id, s.name, s.district, s.developer, s.obj_class, s.flat_count, s.lat, s.lon, st.weighted, st.rank_overall FROM sites s JOIN scores_total st USING (site_id) WHERE s.kind='jk' ORDER BY st.weighted DESC""").fetchall() top10 = [dict(zip(['site_id','name','district','developer','obj_class','flat_count', 'lat','lon','weighted','rank'], r)) for r in rows[:10]] # Distribution stats all_w = [r[8] for r in rows] stats = { "n_jk": len(all_w), "mean": round(statistics.mean(all_w), 1), "median": round(statistics.median(all_w), 1), "p25": round(statistics.quantiles(all_w, n=4)[0], 1), "p75": round(statistics.quantiles(all_w, n=4)[2], 1), "min": round(min(all_w), 1), "max": round(max(all_w), 1), } # Component-wise comparison parcel_comp = parcel["scores"] comp_stats = {} for c in parcel_comp: vals = [v for (v,) in conn.execute( "SELECT score_0_100 FROM scores s JOIN sites si USING (site_id) WHERE component=? AND si.kind='jk'", (c,)).fetchall()] comp_stats[c] = { "parcel": round(parcel_comp[c], 1), "median_jk": round(statistics.median(vals), 1), "p75_jk": round(statistics.quantiles(vals, n=4)[2], 1), } # Closest comparable ЖК (by geographic proximity to parcel) plat, plon = parcel["lat"], parcel["lon"] nearby = [] for r in conn.execute("""SELECT s.site_id, s.name, s.district, s.developer, s.obj_class, s.lat, s.lon, st.weighted, st.rank_overall FROM sites s JOIN scores_total st USING (site_id) WHERE s.kind='jk'""").fetchall(): import math R=6371000; p1,p2=math.radians(plat),math.radians(r[5]) dp=math.radians(r[5]-plat); dl=math.radians(r[6]-plon) a=math.sin(dp/2)**2+math.cos(p1)*math.cos(p2)*math.sin(dl/2)**2 d=2*R*math.asin(math.sqrt(a)) nearby.append((d, r)) nearby.sort() closest = [ {"distance_m": round(d, 0), "site_id": r[0], "name": r[1], "district": r[2], "developer": r[3], "obj_class": r[4], "weighted": round(r[7], 1), "rank": r[8]} for d, r in nearby[:10] ] out = { "generated_at": __import__("datetime").datetime.now().isoformat(timespec="seconds"), "parcel": parcel, "n_compared_jk": stats["n_jk"], "weighted_score_distribution": stats, "component_comparison": comp_stats, "weights_used": weights, "top10_best_jk_ekb": top10, "10_closest_jk_to_parcel": closest, } json_path = OUT / "parcel_66_41_0204016_10.json" with open(json_path, "w") as f: json.dump(out, f, ensure_ascii=False, indent=2, default=str) print(f"JSON: {json_path}") # Build HTML html = build_html(out) html_path = OUT / "parcel_66_41_0204016_10.html" with open(html_path, "w") as f: f.write(html) print(f"HTML: {html_path}") conn.close() def econ_block(e): if not e: return "

Нет данных Объектива для этого района.

" f = lambda x, suf="": (f"{x:.1f}{suf}" if isinstance(x,(int,float)) else "—") method_note = "по совпадению имени ЖК" if e["district_method"]=="name_match" \ else f"по голосованию 5 ближайших ЖК (расст. до репрезентанта {e['district_dist_m']:.0f} м)" return f"""
Район (Объектив){e["district"]} ({method_note})
Проектов в районе{e["n_projects"]}
Лотов всего / продано{e["n_lots"] or 0:,} / {e["n_sold"] or 0:,} ({f(e["sold_pct"], '%')})
Цена за м² (медиана){f(e["median_price_m2"])} тыс ₽ · P25={f(e["p25_price_m2"])} · P75={f(e["p75_price_m2"])}
Средняя площадь сделки{f(e["avg_area_sold_m2"])} м²
Скорость продаж (real){f(e["velocity_per_month"])} зарег. ДДУ/корпус/мес (по 12 мес)
Средняя готовность{f(e["avg_readiness_pct"], '%')}
Цена corp_sum (взвеш.){f(e["weighted_price_m2_corp_sum"])} тыс ₽/м² · скорость {f(e["deals_per_month_corp_sum"])}
Распродажа стока (corp_sum){f(e["months_to_sellout"], ' мес')}

real_* рассчитаны по {e["n_lots"] or 0:,} лотам из Поквартирные/Лоты (303 677 квартир Екб). Это per-flat, основной источник правды.

""" def build_html(d): p = d["parcel"] cs = d["component_comparison"] # Single source of weights for the caption — identical to JSON weights_used. weights_caption = " · ".join( f"{WEIGHT_LABELS.get(c, c)} {w*100:.0f}%" for c, w in d.get("weights_used", {}).items() ) poi_table_rows = [] for cat, items in p["nearest_pois"].items(): nearest = items[0] if items else None poi_table_rows.append( f"{cat}" f"{nearest['name'] if nearest else '—'}" f"{('%.0f м' % nearest['distance_m']) if nearest else '—'}" f"{p['features'].get(f'{cat}_count_500m','')}" f"{p['features'].get(f'{cat}_count_1km','')}" ) comp_rows = [] for c, v in cs.items(): delta = v["parcel"] - v["median_jk"] cls = "g" if delta > 0 else ("neg" if delta < 0 else "") comp_rows.append( f"{c}{v['parcel']}" f"{v['median_jk']}{v['p75_jk']}" f"{'+' if delta>=0 else ''}{delta:.1f}" ) closest_rows = "".join( f"{x['distance_m']:.0f} м" f"{x['name'] or '—'}{x['district'] or '—'}" f"{x['developer'] or ''}{x['obj_class'] or ''}" f"{x['weighted']}#{x['rank']}" for x in d["10_closest_jk_to_parcel"] ) top_rows = "".join( f"#{x['rank']}{x['name'] or '—'}" f"{x['district'] or '—'}{x['developer'] or ''}" f"{x['obj_class'] or ''}{x['weighted']:.1f}" for x in d["top10_best_jk_ekb"] ) return f""" Анализ участка {p['name']}

Анализ участка 66:41:0204016:10

Сгенерировано {d["generated_at"]} · Сравнение с {d["n_compared_jk"]} строящимися ЖК Екатеринбурга
Координаты участка получены из ссылки на NSPD-карту (EPSG:3857 → WGS84): {p['lat']}, {p['lon']}. Район — центрально-северная часть ЕКБ (Пионерский / Втузгородок).
Итоговый балл{p["weighted_total"]:.1f}из 100
Ранг по ЕКБ#{p["rank_overall"]} / {d["n_compared_jk"]+1}
Перцентиль{(1 - (p["rank_overall"]-1)/(d["n_compared_jk"]+1))*100:.0f}%
Медиана ЖК{d["weighted_score_distribution"]["median"]}
Топ-25% ЖК ≥{d["weighted_score_distribution"]["p75"]}

Компоненты (взвешенные)

{"".join(comp_rows)}
КомпонентУчастокМедиана ЖКP75 ЖКΔ vs медиана
Веса: {weights_caption}

Экономика района (Объектив API; скорость — за 12 мес, доли продаж и цены — накопительно за всё время)

{econ_block(p.get("economics"))}

Ближайшие POI вокруг участка

{"".join(poi_table_rows)}
КатегорияБлижайшийДо негоВ 500 мВ 1 км

10 ближайших ЖК (для прямого бенчмарка)

{closest_rows}
Расст.НазваниеРайонДевелоперКлассБаллРанг

Топ-10 ЖК ЕКБ по локационной привлекательности

{top_rows}
РангНазваниеРайонДевелоперКлассБалл

Методика

Источник POI: OpenStreetMap (Overpass API), bbox по всем 381 ЖК Свердл.

Логика: для каждой категории — расстояние-в-балл (piecewise linear от ideal_m к max_m), далее агрегация в 5 компонент с весами (max-pool там, где категории альтернативны: ритейл = max(big, 0.7×med); транспорт = max(metro, 0.85×tram, 0.7×bus)). Финальный балл — взвешенная сумма компонент.

База данных: локальная SQLite analysis.db (sites/pois/features/scores), под-выборка строящихся ЖК ЕКБ из прода domrf_kn_objects.

""" if __name__ == "__main__": main()