cc/startup_app/engine.py

원본 내용을 읽기 전용으로 표시합니다.

"""두 시점 상가 비교와 주변 업종 구성의 참고 업소 수. 소비 수요 모델이 아님."""
from __future__ import annotations
import hashlib
import json
import logging
import os
from functools import lru_cache
from pathlib import Path
for key in ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS"):
os.environ.setdefault(key, "1")
import joblib
import numpy as np
import pandas as pd
import sklearn
from sklearn.ensemble import HistGradientBoostingRegressor
from sklearn.model_selection import GroupKFold
from sklearn.neighbors import BallTree
from threadpoolctl import threadpool_limits
ROOT = Path(__file__).resolve().parent
R, MIN_MID, MIN_TOT = 300, 40, 10
FILES = ("sejong_store_202506.csv", "sejong_store_202606.csv")
COLUMNS = ["상가업소번호", "상호명", "상권업종대분류명", "상권업종중분류명", "상권업종소분류명", "행정동명", "건물명", "경도", "위도"]
LOG = logging.getLogger(__name__)
def data_directory(explicit=None):
configured = explicit or os.environ.get("SEJONG_DATA_DIR")
candidates = [Path(configured)] if configured else [ROOT.parent / "data", ROOT.parent.parent / "data"]
for path in candidates:
if all((path / name).is_file() for name in FILES):
return path.resolve()
raise FileNotFoundError("상가 CSV 2개를 data 폴더에 넣거나 SEJONG_DATA_DIR을 지정하세요: " + ", ".join(FILES))
def 미터좌표(lat, lon):
return ((np.asarray(lon) - 127.27) * 111320 * np.cos(np.radians(36.5)), (np.asarray(lat) - 36.5) * 110540)
def load_stores(path):
frame = pd.read_csv(path, usecols=COLUMNS, dtype={"상가업소번호": "string"}, low_memory=False)
before = len(frame)
for key in COLUMNS[:-2]:
frame[key] = frame[key].astype("string").str.strip()
for key in ("위도", "경도"):
frame[key] = pd.to_numeric(frame[key], errors="coerce")
valid = (frame["상가업소번호"].notna() & frame["상가업소번호"].ne("") & frame["위도"].between(36.35, 36.75) & frame["경도"].between(127.1, 127.45))
invalid = int((~valid).sum())
frame = frame.loc[valid].copy()
duplicates = int(frame.duplicated("상가업소번호").sum())
frame = frame.drop_duplicates("상가업소번호").reset_index(drop=True)
for key in ("상권업종대분류명", "상권업종중분류명", "상권업종소분류명", "행정동명", "상호명"):
frame[key] = frame[key].fillna("미분류").replace("", "미분류")
frame["건물명"] = frame["건물명"].fillna("")
frame["x"], frame["y"] = 미터좌표(frame["위도"], frame["경도"])
if frame.empty:
raise ValueError(f"유효한 상가 좌표가 없습니다: {path.name}")
return frame, {"원본행": before, "좌표ID오류제외": invalid, "중복ID제외": duplicates, "사용행": len(frame)}
def ring_contains(lon, lat, ring):
xy = np.asarray(ring, dtype=float)[:, :2]
start, end = xy, np.roll(xy, -1, axis=0)
dx, dy = end[:, 0] - start[:, 0], end[:, 1] - start[:, 1]
cross = (lon - start[:, 0]) * dy - (lat - start[:, 1]) * dx
on = ((np.abs(cross) < 1e-11) & (lon >= np.minimum(start[:, 0], end[:, 0]) - 1e-11) & (lon <= np.maximum(start[:, 0], end[:, 0]) + 1e-11) & (lat >= np.minimum(start[:, 1], end[:, 1]) - 1e-11) & (lat <= np.maximum(start[:, 1], end[:, 1]) + 1e-11))
if on.any():
return True
edges = (start[:, 1] > lat) != (end[:, 1] > lat)
hits = start[edges, 0] + (lat - start[edges, 1]) * dx[edges] / dy[edges]
return bool(np.count_nonzero(hits > lon) % 2)
def geometry_contains(lon, lat, geometry):
polygons = geometry["coordinates"] if geometry["type"] == "MultiPolygon" else [geometry["coordinates"]]
return any(ring_contains(lon, lat, rings[0]) and not any(ring_contains(lon, lat, hole) for hole in rings[1:]) for rings in polygons)
def deviance(y, predicted):
predicted = np.maximum(predicted, 1e-9)
return float(2 * np.sum(np.where(y > 0, y * np.log(np.maximum(y, 1e-9) / predicted), 0) - (y - predicted)))
class 엔진:
def __init__(self, data_dir=None, cache_dir=None):
self.data_dir = data_directory(data_dir)
self.cache_dir = Path(cache_dir or os.environ.get("SEJONG_CACHE_DIR", ROOT / "cache"))
self.prev, previous_audit = load_stores(self.data_dir / FILES[0])
self.now, current_audit = load_stores(self.data_dir / FILES[1])
self.audit = {"2025-06": previous_audit, "2026-06": current_audit}
self.now["신규"] = ~self.now["상가업소번호"].isin(self.prev["상가업소번호"])
self.prev["소멸"] = ~self.prev["상가업소번호"].isin(self.now["상가업소번호"])
self.tree_now = BallTree(self.now[["x", "y"]].values)
self.tree_prev = BallTree(self.prev[["x", "y"]].values)
boundary = self.data_dir / "sejong_admdong.geojson"
self.boundary = json.loads(boundary.read_text(encoding="utf-8-sig")) if boundary.is_file() else None
counts = self.now["상권업종중분류명"].value_counts()
self.MIDS = sorted(counts[counts >= MIN_MID].index, key=lambda m: (-counts[m], m))
self.mid_to_cat = self.now.groupby("상권업종중분류명")["상권업종대분류명"].first().to_dict()
self.업종번호 = self.now["상권업종중분류명"].map({m: i for i, m in enumerate(self.MIDS)}).fillna(-1).astype(int).values
self.전체소멸률 = float(self.prev["소멸"].mean())
self.업종소멸률 = self.prev.groupby("상권업종중분류명")["소멸"].mean().to_dict()
fingerprint = {"version": "context-count-spatial-buffer-v3", "sklearn": sklearn.__version__, "업종순서": self.MIDS, "radius": R, "grid": 250, "block": 2000, "buffer": 600, "min_total": MIN_TOT, "files": {name: hashlib.sha256((self.data_dir / name).read_bytes()).hexdigest() for name in FILES}}
self.fingerprint = hashlib.sha256(json.dumps(fingerprint, sort_keys=True, ensure_ascii=False).encode()).hexdigest()
self.model_validation = {"방법": "공간 블록 5겹 검증", "블록m": 2000, "검증주변학습제외m": 600}
model_file = self.cache_dir / "context_count_models_v2.joblib"
manifest_file = self.cache_dir / "context_count_models_v2.json"
loaded = False
if model_file.is_file() and manifest_file.is_file():
try:
manifest = json.loads(manifest_file.read_text(encoding="utf-8"))
if manifest.get("fingerprint") == self.fingerprint:
self.모델, self.신뢰도 = joblib.load(model_file)
self.model_validation = manifest["validation"]
loaded = True
except (OSError, ValueError, EOFError, AttributeError, KeyError):
LOG.exception("캐시를 읽지 못해 다시 학습합니다")
if not loaded:
with threadpool_limits(1):
self.모델, self.신뢰도 = self.학습()
self.cache_dir.mkdir(parents=True, exist_ok=True)
temporary = model_file.with_suffix(".tmp")
joblib.dump((self.모델, self.신뢰도), temporary)
temporary.replace(model_file)
manifest_file.write_text(json.dumps({"fingerprint": self.fingerprint, "inputs": fingerprint, "validation": self.model_validation}, ensure_ascii=False, indent=2), encoding="utf-8")
def 위치확인(self, lat, lon):
if not (np.isfinite(lat) and np.isfinite(lon) and 36.35 <= lat <= 36.75 and 127.1 <= lon <= 127.45):
return False
return not self.boundary or any(geometry_contains(lon, lat, feature["geometry"]) for feature in self.boundary["features"])
def 업종개수(self, points):
indices = self.tree_now.query_radius(points, r=R)
matrix, totals = np.zeros((len(points), len(self.MIDS))), np.zeros(len(points))
for i, nearby in enumerate(indices):
categories = self.업종번호[nearby]
np.add.at(matrix[i], categories[categories >= 0], 1)
totals[i] = len(nearby)
return matrix, totals
@staticmethod
def 모델입력(matrix, totals, target):
return np.c_[np.log1p(np.delete(matrix, target, axis=1)), np.log1p(totals - matrix[:, target])]
@staticmethod
def 비례예측(other_train, y_train, other_test):
# 비교 기준도 대상 업종 수를 모르는 상태에서 같은 입력 정보만 사용.
rate = y_train.sum() / max(float(other_train.sum()), 1e-9)
return np.maximum(other_test * rate, 1e-9)
def 학습(self):
LOG.warning("주변 업종 구성 참고 모델 학습 시작")
grid = self.now[["x", "y"]].floordiv(250).drop_duplicates().values
points = (grid + .5) * 250
matrix, totals = self.업종개수(points)
keep = totals >= MIN_TOT
matrix, totals, points = matrix[keep], totals[keep], points[keep]
_, groups = np.unique(np.floor(points / 2000), axis=0, return_inverse=True)
splits = []
if len(points) >= 40 and len(np.unique(groups)) >= 5:
for train, test in GroupKFold(5).split(points, groups=groups):
# 반경 300m 관측 범위가 겹치지 않도록 검증점 주변 600m는 학습에서 제외.
distances = BallTree(points[test]).query(points[train], k=1)[0][:, 0]
train = train[distances > 2 * R]
if len(train) >= 30:
splits.append((train, test))
self.model_validation.update({"학습지점": len(points), "공간그룹": len(np.unique(groups)), "유효검증폴드": len(splits), "반경겹침제외": True})
models, reliability = {}, {}
for j, name in enumerate(self.MIDS):
X, y = self.모델입력(matrix, totals, j), matrix[:, j]
if y.sum() <= 0:
continue
def estimator():
return HistGradientBoostingRegressor(loss="poisson", max_iter=150, learning_rate=.08, min_samples_leaf=15, random_state=0)
predicted, mean_only, proportional = (np.full(len(y), np.nan) for _ in range(3))
for train, test in splits:
mean_only[test] = max(float(y[train].mean()), 1e-9)
other_totals = totals - y
proportional[test] = self.비례예측(other_totals[train], y[train], other_totals[test])
predicted[test] = estimator().fit(X[train], y[train]).predict(X[test]) if y[train].sum() > 0 else 1e-9
valid = np.isfinite(predicted)
baseline = deviance(y[valid], mean_only[valid]) if valid.any() else 0
d2 = 1 - deviance(y[valid], predicted[valid]) / baseline if baseline > 1e-9 else None
simple_d2 = 1 - deviance(y[valid], proportional[valid]) / baseline if baseline > 1e-9 else None
coverage = float(valid.mean())
reliability[name] = {"D2": round(d2, 3) if d2 is not None else None, "비례기준D2": round(simple_d2, 3) if simple_d2 is not None else None, "검증비율": round(coverage, 3), "신뢰": bool(coverage == 1 and d2 is not None and d2 >= .4 and d2 >= simple_d2)}
models[name] = estimator().fit(X, y)
LOG.warning("모델 학습 완료: %s종 / 공간 검증 통과 %s종", len(models), sum(r["신뢰"] for r in reliability.values()))
return models, reliability
def 예상개수(self, matrix, totals, category):
j = self.MIDS.index(category)
return float(self.모델[category].predict(self.모델입력(matrix, totals, j))[0]), int(matrix[0, j])
def 메타(self):
categories = {}
for name in sorted(self.mid_to_cat):
categories.setdefault(self.mid_to_cat[name], []).append(name)
buildings = (self.now[self.now["건물명"].ne("")].groupby(["건물명", "행정동명"]).agg(위도=("위도", "median"), 경도=("경도", "median"), 업소수=("위도", "size")).reset_index().sort_values("업소수", ascending=False))
return {"업종": categories, "건물": buildings.to_dict("records"), "반경": R, "기준": {"이전": "2025-06", "현재": "2026-06", "자료": "제공된 상가(상권)정보 CSV", "좌표범위": "행정동 경계" if self.boundary else "근사 범위"}, "데이터점검": self.audit, "모델검증": self.model_validation, "모델수": len(self.모델), "검증통과업종수": sum(r["신뢰"] for r in self.신뢰도.values()), "주의": "주변 업종 구성의 업소 수 추정입니다. 매출·유동인구·소비 수요·창업 성공률 예측이 아닙니다. 소멸은 이전 ID가 현재 자료에 없는 상태이며 실제 폐업 여부는 확인하지 않았습니다."}
@lru_cache(maxsize=256)
def 분석(self, lat, lon, mid):
x, y = 미터좌표(lat, lon)
point = np.array([[float(x), float(y)]])
distances, nearest = self.tree_now.query(point, k=1)
dong = self.now["행정동명"].iloc[nearest[0, 0]]
if self.boundary:
for feature in self.boundary["features"]:
if geometry_contains(lon, lat, feature["geometry"]):
dong = feature["properties"].get("adm_nm", dong).split()[-1]
break
area = self.now.iloc[self.tree_now.query_radius(point, r=R)[0]].copy()
previous = self.prev.iloc[self.tree_prev.query_radius(point, r=R)[0]]
area["거리"] = np.hypot(area["x"] - x, area["y"] - y).round()
same = area[area["상권업종중분류명"].eq(mid)].sort_values("거리")
previous_same = previous[previous["상권업종중분류명"].eq(mid)]
composition = [[key, round(value * 100, 1)] for key, value in area["상권업종대분류명"].value_counts(normalize=True).items()]
status = {"업소수": len(area), "작년업소수": len(previous), "소멸률": float(previous["소멸"].mean()) if len(previous) else None, "같은업종수": len(same), "없어짐": int(previous_same["소멸"].sum()), "생김": int(same["신규"].sum()), "구성": composition}
matrix, totals = self.업종개수(point)
reference = None
if mid in self.모델 and totals[0] >= MIN_TOT:
estimate, actual = self.예상개수(matrix, totals, mid)
reference = {"예상": round(estimate, 1), "실제": actual, **self.신뢰도[mid]}
disappearance = {"업종": self.업종소멸률.get(mid), "세종": self.전체소멸률, "이근처": float(previous_same["소멸"].mean()) if len(previous_same) >= 5 else None, "이근처_수": len(previous_same)}
competitors = [{"상호": row["상호명"], "소분류": row["상권업종소분류명"], "거리": int(row["거리"]), "위도": row["위도"], "경도": row["경도"]} for _, row in same.iterrows()]
notes = [f"반경 {R}m 상가 기록을 두 시점에서 비교한 결과입니다."]
if totals[0] < MIN_TOT:
notes.append(f"주변 기록이 {int(totals[0])}개로 적어 모델 참고값을 제시하지 않습니다.")
elif reference:
notes.append("참고 업소 수는 다른 업종의 구성으로 추정한 값입니다. 실제 소비 수요나 영업 기회를 뜻하지 않습니다.")
if not reference["신뢰"]:
notes.append("이 업종은 공간 검증 기준을 통과하지 못했습니다. 참고값 해석에 주의하세요.")
else:
notes.append(f"세종 내 기록이 {MIN_MID}개 미만이거나 모델을 만들 수 없어 이 업종의 참고값은 없습니다.")
notes.append("신규·소멸은 업소 ID 비교 결과로 이전·재등록·분류 변경·수집 누락이 포함될 수 있습니다.")
return {"동": dong, "업종": mid, "현황": status, "참고": reference, "소멸": disappearance, "경쟁": competitors, "안내": notes, "반경": R, "최근접업소거리m": round(float(distances[0, 0]))}
@lru_cache(maxsize=256)
def 업종추천(self, lat, lon):
x, y = 미터좌표(lat, lon)
matrix, totals = self.업종개수(np.array([[float(x), float(y)]]))
rows = []
if totals[0] >= MIN_TOT:
for category in self.MIDS:
if category not in self.모델 or not self.신뢰도[category]["신뢰"]:
continue
estimate, actual = self.예상개수(matrix, totals, category)
rows.append({"업종": category, "예상": round(estimate, 1), "실제": actual, "점수": round((estimate - actual) / np.sqrt(max(estimate, 1)), 2), "소멸률": self.업종소멸률.get(category)})
lower = sorted([r for r in rows if r["예상"] >= 2 and r["실제"] <= .8 * r["예상"]], key=lambda r: -r["점수"])[:10]
higher = sorted([r for r in rows if r["실제"] >= 3 and r["점수"] < 0], key=lambda r: r["점수"])[:5]
reason = "주변기록부족" if totals[0] < MIN_TOT else "표시대상없음" if not lower and not higher else "비교가능"
return {"업소수": int(totals[0]), "적은편": lower, "많은편": higher, "상태": reason, "검증통과업종수": len(rows), "반경": R, "안내": "공간 검증 기준을 통과한 업종만 비교합니다. 적은 업소 수가 창업 기회나 실제 수요 부족을 의미하지 않습니다."}

파일 확인 정보

SHA-256: d8b0f3c79fdb69857295feb0263c3b5f87f2e7a7653d3935c72237538934368c