부동산 매매가 데이터 통계 정보 만들기 1편ㅣ오늘은 부동산 매매가 데이터 통계 정보를 만들어 보겠습니다. 간단하게 매매가 정보를 가지고 와서 유의미한 정보를 추출해서 보여주려고 합니다. 파이썬 코드로 쉽게 웹앱을 만들 수 있는 streamlit 을 사용해보도록 하겠습니다.
부동산 매매가 데이터 통계 정보 만들기 1편
먼저 예전에 사용했던 코드를 불러 오겠습니다. 기존 포스팅에서 사용했던 코드이니 기억하실 분도 있으실 겁니다. 만약 이 코드를 보고 이해가 안된다고 생각하시는 분들은 아래 포스팅을 한번 참고해보세요.
2024.09.14 – [부동산/자동화 프로젝트] – 파이썬 부동산 매매가 조회 프로그램 만들기 4편 (전국 데이터)
파이썬 부동산 매매가 조회 프로그램 만들기 3편 (서울아파트 컬럼 정리)
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"
자 실행하면 잘 됩니다만, 에러 출력을 하나 만나게 됩니다.
우선 엑셀 데이터를 메모리에 저장한 후 이를 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편 기대해주세요~
#부동산, 데이터, 부동산데이터, 부동산통계, 파이썬크롤링,파이썬부동산,부동산매매,부동산흐름,부동산매매가