부동산 매매가 데이터 통계 정보 만들기 1편

오늘은 부동산 매매가 데이터 통계 정보를 만들어 보겠습니다. 간단하게 매매가 정보를 가지고 와서 유의미한 정보를 추출해서 보여주려고 합니다. 파이썬 코드로 쉽게 웹앱을 만들 수 있는 streamlit 을 사용해보도록 하겠습니다.

부동산 매매가 데이터 통계 정보 만들기 1편

먼저 예전에 사용했던 코드를 불러 오겠습니다. 기존 포스팅에서 사용했던 코드이니 기억하실 분도 있으실 겁니다. 만약 이 코드를 보고 이해가 안된다고 생각하시는 분들은 아래 포스팅을 한번 참고해보세요.

2024.09.14 - [부동산/자동화 프로젝트] - 파이썬 부동산 매매가 조회 프로그램 만들기 4편 (전국 데이터)

class DistrictConverter:
    def __init__(self):
        self.districts = self.__read_district_file()

    def __read_district_file(self):
        # 드라이브 내의 JSON 파일 경로를 설정합니다.
        json_file_path = '/content/drive/MyDrive/district.json'

        # 파일을 열어서 내용을 읽고 JSON으로 변환합니다.
        with open(json_file_path, 'r') as f:
            return json.loads(f.read())

    def get_si_do_code(self, si_do_name):
        for district in self.districts:
            if si_do_name == district["si_do_name"]:
                return district["si_do_code"]

    def get_sigungu(self, si_do_code):
        for district in self.districts:
            if si_do_code == district["si_do_code"]:
                return district["sigungu"]

def get_user_input(prompt, default_value=None):
    try:
        user_input = input(prompt)
        return user_input if user_input.strip() else default_value
    except EOFError:
        return default_value

# 사용자로부터 시/도와 기간을 입력 받음
si_do_name = get_user_input("시/도를 입력하세요 (예: 서울특별시) 또는 '전국' 입력: ", "전국")
start_year_month = get_user_input("조회 시작 년월 (YYYYMM 형식, 예: 202301): ", None)
end_year_month = get_user_input("조회 종료 년월 (YYYYMM 형식, 예: 202312): ", None)

# 현재 날짜를 기준으로 기간을 설정
now = datetime.now()
current_year_month = now.strftime("%Y%m")

# 기간을 설정합니다.
if not start_year_month:
    start_year_month = f"{now.year}01"
if not end_year_month:
    end_year_month = current_year_month

# DistrictConverter 인스턴스 생성
converter = DistrictConverter()

# 서울특별시와 모든 시/군/구 데이터를 수집할 DataFrame 초기화
all_data = pd.DataFrame()

# 전체 시/도 조사
if si_do_name == "전국":
    for district in converter.districts:
        si_do_code = district["si_do_code"]
        sigungu_list = district["sigungu"]
        for sigungu in sigungu_list:
            sigungu_code = sigungu["sigungu_code"]
            sigungu_name = sigungu["sigungu_name"]

            print(f"Processing data for {sigungu_name} ({sigungu_code})")

            # 부동산 데이터를 가져옴
            df = api.get_data(
                property_type="아파트",
                trade_type="매매",
                sigungu_code=sigungu_code,
                start_year_month=start_year_month,
                end_year_month=end_year_month
            )

            # 시/군/구 이름 및 시/도 이름을 새로운 컬럼으로 추가
            df["sigungu_name"] = sigungu_name
            df["si_do_name"] = district["si_do_name"]

            # 가져온 데이터를 all_data에 추가
            all_data = pd.concat([all_data, df], ignore_index=True)
else:
    si_do_code = converter.get_si_do_code(si_do_name)
    sigungu_list = converter.get_sigungu(si_do_code)

    for sigungu in sigungu_list:
        sigungu_code = sigungu["sigungu_code"]
        sigungu_name = sigungu["sigungu_name"]

        print(f"Processing data for {sigungu_name} ({sigungu_code})")

        # 부동산 데이터를 가져옴
        df = api.get_data(
            property_type="아파트",
            trade_type="매매",
            sigungu_code=sigungu_code,
            start_year_month=start_year_month,
            end_year_month=end_year_month
        )

        # 시/군/구 이름 및 시/도 이름을 새로운 컬럼으로 추가
        df["sigungu_name"] = sigungu_name
        df["si_do_name"] = si_do_name

        # 가져온 데이터를 all_data에 추가
        all_data = pd.concat([all_data, df], ignore_index=True)

# 데이터 열 이름을 한국어로 변환
columns_to_select = {
    "si_do_name": "시도",
    "sigungu_name": "시군구",
    "umdNm": "법정동",
    "roadNm": "도로명",
    "bonbun": "지번",
    "aptNm": "아파트",
    "buildYear": "건축년도",
    "excluUseAr": "전용면적",
    "floor": "층",
    "dealYear": "거래년도",
    "dealMonth": "거래월",
    "dealDay": "거래일",
    "dealAmount": "거래금액",
    "aptSeq": "일련번호",
    "dealingGbn": "거래유형",
    "estateAgentSggNm": "중개사소재지",
    "cdealType": "해제여부",
    "cdealDay": "해제사유발생일"
}

# 필요한 열만 남기고 한국어 열 이름으로 변경
selected_data = all_data.rename(columns=columns_to_select)[list(columns_to_select.values())]

# 구글 시트 API 인증
scope = ["https://spreadsheets.google.com/feeds", "https://www.googleapis.com/auth/drive"]
creds = ServiceAccountCredentials.from_json_keyfile_name('/content/drive/MyDrive/client_secret_143915664608-hejcq3u26rkd7gd645tt7ef44cpfqbp2.apps.googleusercontent.com.json', scope)
client = gspread.authorize(creds)

# 구글 시트에 데이터 저장
def save_to_google_sheet(df, sheet_name):
    # 새로운 구글 시트 생성
    sh = client.create(sheet_name)
    worksheet = sh.get_worksheet(0)

    # 데이터프레임을 구글 시트에 추가
    worksheet.update([df.columns.values.tolist()] + df.values.tolist())

# 시트 이름 생성 및 데이터 저장
sheet_name = f'{si_do_name if si_do_name != "전국" else "전체"}_{start_year_month}_{end_year_month}_매매'
save_to_google_sheet(selected_data, sheet_name)

자, 이제 이 코드를 스트림릿에 맞게 코드를 아래와 같이 변경해줍니다. 이 데이터를 스트림릿 클라우드에 업로드하기 전에, 구글 드라이브에 저장하는 코드는 삭제하고 결과 값을 표로 나타내주는 코드를 추가하고, 데이터를 엑셀 또는 csv로 다운 받을 수 있는 옵션창을 하나 추가해보도록 하겠습니다. 그리고 개인 service key 는 보안상 아래과 같이 코드를 수정한 후 앱 설정에서 값을 설정합니다.

Streamlit 코드에 맞게 변경하기

이제 이 코드를 스트림릿에 맞게 코드를 아래와 같이 변경해줍니다.

반응형

import streamlit as st
import pandas as pd
import PublicDataReader as pdr
from datetime import datetime
import json

# Streamlit secrets에서 API 키 및 파일 경로 가져오기
service_key = st.secrets["general"]["SERVICE_KEY"]
json_file_path = "district.json"

# PublicDataReader API 서비스 키 사용
api = pdr.TransactionPrice(service_key)

# DistrictConverter 클래스 정의
class DistrictConverter:
    def __init__(self):
        self.districts = self.__read_district_file()

    def __read_district_file(self):
        with open(json_file_path, 'r') as f:
            return json.loads(f.read())

    def get_si_do_code(self, si_do_name):
        for district in self.districts:
            if si_do_name == district["si_do_name"]:
                return district["si_do_code"]

    def get_sigungu(self, si_do_code):
        for district in self.districts:
            if si_do_code == district["si_do_code"]:
                return district["sigungu"]

# 사용자 입력 받기
st.title("부동산 데이터 조회")
si_do_name = st.text_input("시/도를 입력하세요 (예: 서울특별시) 또는 '전국' 입력", "전국")
start_year_month = st.text_input("조회 시작 년월 (YYYYMM 형식, 예: 202301)", "")
end_year_month = st.text_input("조회 종료 년월 (YYYYMM 형식, 예: 202312)", "")

# 현재 날짜를 기준으로 기간 설정
now = datetime.now()
if not start_year_month:
    start_year_month = f"{now.year}01"
if not end_year_month:
    end_year_month = now.strftime("%Y%m")

# 데이터를 조회하는 버튼을 추가하여, 사용자 입력 후에만 데이터 처리를 시작합니다.
if st.button("데이터 조회"):
    if si_do_name and start_year_month and end_year_month:
        # DistrictConverter 인스턴스 생성
        converter = DistrictConverter()

        # 데이터 수집 및 처리
        all_data = pd.DataFrame()

        if si_do_name == "전국":
            for district in converter.districts:
                si_do_code = district["si_do_code"]
                sigungu_list = district["sigungu"]
                for sigungu in sigungu_list:
                    sigungu_code = sigungu["sigungu_code"]
                    sigungu_name = sigungu["sigungu_name"]

                    st.write(f"Processing data for {sigungu_name} ({sigungu_code})")

                    df = api.get_data(
                        property_type="아파트",
                        trade_type="매매",
                        sigungu_code=sigungu_code,
                        start_year_month=start_year_month,
                        end_year_month=end_year_month
                    )

                    df["sigungu_name"] = sigungu_name
                    df["si_do_name"] = district["si_do_name"]

                    all_data = pd.concat([all_data, df], ignore_index=True)
        else:
            si_do_code = converter.get_si_do_code(si_do_name)
            sigungu_list = converter.get_sigungu(si_do_code)

            for sigungu in sigungu_list:
                sigungu_code = sigungu["sigungu_code"]
                sigungu_name = sigungu["sigungu_name"]

                st.write(f"Processing data for {sigungu_name} ({sigungu_code})")

                df = api.get_data(
                    property_type="아파트",
                    trade_type="매매",
                    sigungu_code=sigungu_code,
                    start_year_month=start_year_month,
                    end_year_month=end_year_month
                )

                df["sigungu_name"] = sigungu_name
                df["si_do_name"] = si_do_name

                all_data = pd.concat([all_data, df], ignore_index=True)

        # 컬럼 이름 변환
        columns_to_select = {
            "si_do_name": "시도",
            "sigungu_name": "시군구",
            "umdNm": "법정동",
            "roadNm": "도로명",
            "bonbun": "지번",
            "aptNm": "아파트",
            "buildYear": "건축년도",
            "excluUseAr": "전용면적",
            "floor": "층",
            "dealYear": "거래년도",
            "dealMonth": "거래월",
            "dealDay": "거래일",
            "dealAmount": "거래금액",
            "aptSeq": "일련번호",
            "dealingGbn": "거래유형",
            "estateAgentSggNm": "중개사소재지",
            "cdealType": "해제여부",
            "cdealDay": "해제사유발생일"
        }

        selected_data = all_data.rename(columns=columns_to_select)[list(columns_to_select.values())]

        # 데이터 표로 표시
        st.write("### 조회 결과")
        st.dataframe(selected_data)

        # 데이터 다운로드 옵션 추가
        st.write("### 데이터 다운로드")
        excel_filename = f"{si_do_name}_{start_year_month}_{end_year_month}_매매.xlsx"
        csv_filename = f"{si_do_name}_{start_year_month}_{end_year_month}_매매.csv"

        # 엑셀 다운로드
        st.download_button(
            label="엑셀로 다운로드",
            data=selected_data.to_excel(index=False, engine='xlsxwriter'),
            file_name=excel_filename,
            mime='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
        )

        # CSV 다운로드
        st.download_button(
            label="CSV로 다운로드",
            data=selected_data.to_csv(index=False),
            file_name=csv_filename,
            mime='text/csv'
        )
    else:
        st.error("모든 필드를 채워주세요.")

Streamlit 코드 옵션 설정 및 key service 부여하기

이제 이 코드를 스트림릿 클라우드 옵션에 key service를 아래와 같이 설정 합니다.

[general]
SERVICE_KEY = "your_actual_service_key_here"
DISTRICT_JSON_PATH = "/path/to/your/district.json"

자 실행하면 잘 됩니다만, 에러 출력을 하나 만나게 됩니다.

File "/home/adminuser/venv/lib/python3.12/site-packages/streamlit/runtime/scriptrunner/exec_code.py", line 88, in exec_func_with_error_handling result = func() ^^^^^^

File "/home/adminuser/venv/lib/python3.12/site-packages/streamlit/runtime/scriptrunner/script_runner.py", line 590, in code_to_exec exec(code, module.__dict__)

File "/mount/src/price_stastic/streamlit_app.py", line 136, in <module> data=selected_data.to_excel(index=False, engine='xlsxwriter'), ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

File "/home/adminuser/venv/lib/python3.12/site-packages/pandas/util/_decorators.py", line 333, in wrapper return func(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^

우선 엑셀 데이터를 메모리에 저장한 후 이를 download_button으로 제공해야 하도록 수정하겠습니다.

import streamlit as st
import pandas as pd
import PublicDataReader as pdr
from datetime import datetime
import json
from io import BytesIO

# Streamlit secrets에서 API 키 및 파일 경로 가져오기
service_key = st.secrets["general"]["SERVICE_KEY"]
json_file_path = "district.json"

# PublicDataReader API 서비스 키 사용
api = pdr.TransactionPrice(service_key)

# DistrictConverter 클래스 정의
class DistrictConverter:
    def __init__(self):
        self.districts = self.__read_district_file()

    def __read_district_file(self):
        with open(json_file_path, 'r') as f:
            return json.loads(f.read())

    def get_si_do_code(self, si_do_name):
        for district in self.districts:
            if si_do_name == district["si_do_name"]:
                return district["si_do_code"]

    def get_sigungu(self, si_do_code):
        for district in self.districts:
            if si_do_code == district["si_do_code"]:
                return district["sigungu"]

# 사용자 입력 받기
st.title("부동산 데이터 조회")
si_do_name = st.text_input("시/도를 입력하세요 (예: 서울특별시) 또는 '전국' 입력", "전국")
start_year_month = st.text_input("조회 시작 년월 (YYYYMM 형식, 예: 202301)", "")
end_year_month = st.text_input("조회 종료 년월 (YYYYMM 형식, 예: 202312)", "")

# 현재 날짜를 기준으로 기간 설정
now = datetime.now()
if not start_year_month:
    start_year_month = f"{now.year}01"
if not end_year_month:
    end_year_month = now.strftime("%Y%m")

# 데이터를 조회하는 버튼을 추가하여, 사용자 입력 후에만 데이터 처리를 시작합니다.
if st.button("데이터 조회"):
    if si_do_name and start_year_month and end_year_month:
        # DistrictConverter 인스턴스 생성
        converter = DistrictConverter()

        # 데이터 수집 및 처리
        all_data = pd.DataFrame()

        if si_do_name == "전국":
            for district in converter.districts:
                si_do_code = district["si_do_code"]
                sigungu_list = district["sigungu"]
                for sigungu in sigungu_list:
                    sigungu_code = sigungu["sigungu_code"]
                    sigungu_name = sigungu["sigungu_name"]

                    st.write(f"Processing data for {sigungu_name} ({sigungu_code})")

                    df = api.get_data(
                        property_type="아파트",
                        trade_type="매매",
                        sigungu_code=sigungu_code,
                        start_year_month=start_year_month,
                        end_year_month=end_year_month
                    )

                    df["sigungu_name"] = sigungu_name
                    df["si_do_name"] = district["si_do_name"]

                    all_data = pd.concat([all_data, df], ignore_index=True)
        else:
            si_do_code = converter.get_si_do_code(si_do_name)
            sigungu_list = converter.get_sigungu(si_do_code)

            for sigungu in sigungu_list:
                sigungu_code = sigungu["sigungu_code"]
                sigungu_name = sigungu["sigungu_name"]

                st.write(f"Processing data for {sigungu_name} ({sigungu_code})")

                df = api.get_data(
                    property_type="아파트",
                    trade_type="매매",
                    sigungu_code=sigungu_code,
                    start_year_month=start_year_month,
                    end_year_month=end_year_month
                )

                df["sigungu_name"] = sigungu_name
                df["si_do_name"] = si_do_name

                all_data = pd.concat([all_data, df], ignore_index=True)

        # 컬럼 이름 변환
        columns_to_select = {
            "si_do_name": "시도",
            "sigungu_name": "시군구",
            "umdNm": "법정동",
            "roadNm": "도로명",
            "bonbun": "지번",
            "aptNm": "아파트",
            "buildYear": "건축년도",
            "excluUseAr": "전용면적",
            "floor": "층",
            "dealYear": "거래년도",
            "dealMonth": "거래월",
            "dealDay": "거래일",
            "dealAmount": "거래금액",
            "aptSeq": "일련번호",
            "dealingGbn": "거래유형",
            "estateAgentSggNm": "중개사소재지",
            "cdealType": "해제여부",
            "cdealDay": "해제사유발생일"
        }

        selected_data = all_data.rename(columns=columns_to_select)[list(columns_to_select.values())]

        # 데이터 표로 표시
        st.write("### 조회 결과")
        st.dataframe(selected_data)

        # 데이터 다운로드 옵션 추가
        st.write("### 데이터 다운로드")
        excel_filename = f"{si_do_name}_{start_year_month}_{end_year_month}_매매.xlsx"
        csv_filename = f"{si_do_name}_{start_year_month}_{end_year_month}_매매.csv"

        # 엑셀 다운로드 (BytesIO로 메모리에 저장)
        output = BytesIO()
        with pd.ExcelWriter(output, engine='xlsxwriter') as writer:
            selected_data.to_excel(writer, index=False)
        output.seek(0)

        st.download_button(
            label="엑셀로 다운로드",
            data=output,
            file_name=excel_filename,
            mime='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
        )

        # CSV 다운로드
        st.download_button(
            label="CSV로 다운로드",
            data=selected_data.to_csv(index=False),
            file_name=csv_filename,
            mime='text/csv'
        )
    else:
        st.error("모든 필드를 채워주세요.")

네 이제 정상 작동을 하지요? 이 코드로 추출한 정보를 streamlit 에서 좀 더 세밀한 분석을 해보고, 시각화된 자료로 출력이 될 수 있도록 해보겠습니다. 2편 기대해주세요~

#부동산, 데이터, 부동산데이터, 부동산통계, 파이썬크롤링,파이썬부동산,부동산매매,부동산흐름,부동산매매가