내일의 태양광 발전량, 파이썬으로 1초 만에 예측하기 (feat. 동서발전 API)
혹시 아직도 내일 날씨만 검색하시나요? 진정한 에너지 고수라면 ‘내일 전기가 얼마나 생산될지’를 검색합니다.
2025년 현재, 대한민국 방방곡곡에 깔린 태양광 패널들. 그런데 “내일 우리 동네 발전소에서 전기 얼마나 나와?”라고 물으면 대답할 수 있는 사람이 몇이나 될까요? 이 질문에 1초 만에 대답할 수 있다면, 여러분은 데이터 분석의 ‘핵인싸’가 될 수 있습니다.
다행히 한국동서발전이 꿀 같은 데이터를 무료로 풀었습니다. 바로 태양광 발전량 예측 API입니다. 오늘 이 포스팅 하나면, 파이썬 코드 몇 줄로 전국 발전량을 내 손안에 쥐락펴락할 수 있습니다. 1년 치 데이터 싹 긁어모으는 방법까지 털어드릴 테니, 끝까지 따라오세요!
1단계: 초간단 API 호출
복잡한 이론은 집어치우고, 일단 코드부터 돌려봅시다. 맛을 봐야 요리를 하죠.
한국동서발전의 태양광 발전량 예측 API는 공공데이터포털에서 신청만 하면 공짜입니다. 공공데이터포털에 가입하고 활용 신청 버튼 누르는 건 기본인 거 아시죠?
API 키를 받았다면, 아래 코드를 복사해서 실행해보세요.
가장 간단한 코드:
import requests
from urllib.parse import unquote
encoded_service_key = "service_key"
decoded_service_key = unquote(encoded_service_key)
pred_date = "20251213" # 예측일자 (내일)
base_date = "20251212" # 기준일자 (오늘)
url = "https://apis.data.go.kr/B552070/forecastService/oamsFile31"
params = {
'serviceKey': decoded_service_key, # Use the unquoted service key here
'pageNo': 1,
'numOfRows': 24,
'apiType': 'json',
'baseDate': base_date,
'city': '강원',
'predDate': pred_date
}
response = requests.get(url, params=params)
try:
print(response.json())
except requests.exceptions.JSONDecodeError:
print(f"JSONDecodeError: The API did not return valid JSON.")
print(f"Status Code: {response.status_code}")
print(f"Response Text: {response.text}")
except Exception as e:
print(f"An unexpected error occurred: {e}")
이 코드를 실행하고 터미널에 외계어 같은 JSON 데이터가 떴다면? 성공입니다. 강원도의 내일 태양광 에너지를 미리 보셨습니다.
2단계: 응답 데이터 파싱 (JSON 구조 이해하기)
API가 던져주는 데이터는 JSON이라는 포맷입니다. 마치 ‘러시아 마트료시카 인형’처럼 상자 안에 상자가 들어있는 구조죠. 이걸 예쁘게 까봐야 우리가 원하는 ‘알맹이’를 꺼낼 수 있습니다.
import requests
import json
response = requests.get(url, params=params)
data = response.json()
# 응답 구조 확인
# data['header'] -> 응답 코드, 메시지
# data['body']['items']['item'] -> 실제 데이터
header = data.get('header', {})
result_code = header.get('resultCode')
if result_code == '00': # 성공
items = data['body']['items']['item']
# 단일 항목인 경우 리스트로 변환
if isinstance(items, dict):
items = [items]
for item in items:
print(f"시간: {item['hh']}시")
print(f"기온: {item['temp']}℃")
print(f"습도: {item['humid']}%")
print(f"태양광 발전량 총합: {item['solgenAll']} MWh")
print("-" * 50)
else:
print(f"API 오류: {header.get('resultMsg')}")이제야 좀 사람이 읽을 수 있는 정보가 나오네요. 시간별로 기온, 습도, 그리고 가장 중요한 발전량(solgenAll)이 보입니다.
3단계: Pandas로 데이터프레임 만들기
개발자라고 해서 검은 화면에 흰 글씨만 보라는 법 있나요? 우리에겐 Pandas(판다스)라는 강력한 무기가 있습니다. 데이터를 엑셀처럼 깔끔한 표(DataFrame)로 만들어줍니다.
import requests
import pandas as pd
response = requests.get(url, params=params)
data = response.json()
items = data['body']['items']['item']
if isinstance(items, dict):
items = [items]
# 데이터프레임 생성
df = pd.DataFrame(items)
# 숫자형 컬럼 변환
numeric_cols = ['temp', 'humid', 'ws', 'solgenKpx', 'solgenBtm', 'solgenPpa', 'solgenAll']
for col in numeric_cols:
df[col] = pd.to_numeric(df[col], errors='coerce')
print(df[['hh', 'temp', 'humid', 'solgenAll']])
# CSV로 저장
df.to_csv('solar_forecast.csv', index=False, encoding='utf-8-sig')코드 실행 한 번에 엑셀 파일이 ‘뚝딱’ 만들어졌습니다. 이제 상사에게 보고할 때 “API 호출해서 JSON 파싱했습니다” 하지 말고, 조용히 엑셀 파일을 내미세요.
4단계: 클래스로 체계적으로 구성하기
코드가 길어지면 지저분해집니다. 이럴 때 클래스(Class)를 사용하면 코드를 레고 블록처럼 정리할 수 있습니다. 나중에 재사용하기도 훨씬 편하죠.
import requests
import pandas as pd
from datetime import datetime, timedelta
from urllib.parse import unquote
class SolarForecastAPI:
def __init__(self, service_key):
# Add logic to unquote service_key if it's already encoded
if '%' in service_key:
self.service_key = unquote(service_key)
else:
self.service_key = service_key
self.base_url = "https://apis.data.go.kr/B552070/forecastService"
self.endpoint = "/oamsFile31"
def get_forecast(self, pred_date, city):
"""예측 데이터 조회"""
# pred_date로부터 base_date 자동 계산
pred_date_obj = datetime.strptime(pred_date, "%Y%m%d")
base_date = (pred_date_obj - timedelta(days=1)).strftime("%Y%m%d")
params = {
'serviceKey': self.service_key,
'pageNo': 1,
'numOfRows': 10,
'apiType': 'json',
'baseDate': base_date,
'city': city,
'predDate': pred_date
}
response = requests.get(self.base_url + self.endpoint, params=params)
return response.json()
def parse_response(self, response):
"""JSON 응답 파싱"""
if response.get('header', {}).get('resultCode') != '00':
print(f"오류: {response.get('header', {}).get('resultMsg')}")
return None
items = response['body']['items']['item']
if isinstance(items, dict):
items = [items]
return pd.DataFrame(items)
# 사용 예제
api = SolarForecastAPI("YOUR_SERVICE_KEY")
data = api.get_forecast("20251215", "강원")
df = api.parse_response(data)
print(df)
이제 코드가 훨씬 깔끔하고 재사용하기 쉬워졌어요!
5단계: 여러 지역 동시에 조회하기
강원도만 보면 섭섭하죠. 서울, 부산, 대구… 전국의 태양광 발전량을 한 번에 비교해봅시다.
def query_multiple_cities(api, pred_date, cities):
"""여러 시도 데이터 한 번에 조회"""
all_data = []
for city in cities:
print(f"📍 {city} 조회 중...")
response = api.get_forecast(pred_date, city)
df = api.parse_response(response)
if df is not None:
all_data.append(df)
else:
print(f"⚠️ {city} 데이터 없음")
# 모든 데이터 합치기
combined_df = pd.concat(all_data, ignore_index=True)
# 시도별 총 발전량으로 정렬
summary = combined_df.groupby('city')['solgenAll'].sum().sort_values(ascending=False)
print("\n📊 시도별 예상 발전량:")
print(summary)
return combined_df
# 사용 예제 cities = ["서울", "부산", "경기", "강원", "인천", "대구"] df_all = query_multiple_cities(api, "20251215", cities)
어디가 발전량이 제일 많을까요? 맑은 날씨의 경상도? 아니면 패널이 많은 전라도? 직접 돌려서 확인해보세요!
6단계: 날짜 순환으로 장기간 데이터 수집하기 ⭐
데이터 분석의 꽃은 시계열 데이터죠. 하루 치 말고, 1년 치 데이터를 모아서 계절별 변화를 분석해봅시다. 여기서 중요한 건 time.sleep입니다. 너무 빠르게 요청하면 서버가 “너 디도스(DDoS) 공격이니?” 하고 차단할 수 있거든요.
from datetime import datetime, timedelta
import time
def collect_yearly_data(api, start_date, num_days=365, cities=None):
"""
여러 날짜에 대해 전국 태양광 발전량 데이터 수집
Args:
api: SolarForecastAPI 인스턴스
start_date: 시작 날짜 (YYYYMMDD)
num_days: 수집 기간 (기본값: 365일)
cities: 시도 리스트
"""
if cities is None:
cities = ["서울", "부산", "대구", "인천", "광주", "대전", "울산", "세종",
"경기", "강원", "충북", "충남", "전북", "전남", "경북", "경남", "제주"]
all_yearly_data = []
start_date_obj = datetime.strptime(start_date, "%Y%m%d")
print(f"\n{'='*80}")
print(f"📅 {start_date}부터 {num_days}일간 전국 태양광 발전량 데이터 수집")
print(f"{'='*80}\n")
for day_offset in range(num_days):
current_date = start_date_obj + timedelta(days=day_offset)
current_date_str = current_date.strftime("%Y%m%d")
print(f"[{day_offset + 1}/{num_days}] {current_date_str} 데이터 수집 중...")
for city in cities:
try:
response = api.get_forecast(current_date_str, city)
df = api.parse_response(response)
if df is not None:
df['collect_date'] = current_date_str
all_yearly_data.append(df)
except Exception as e:
print(f" ⚠️ {city} 오류: {e}")
# API 요청 제한 회피 (1초 대기)
time.sleep(1)
# 모든 데이터 통합
yearly_df = pd.concat(all_yearly_data, ignore_index=True)
# CSV로 저장
filename = f"solar_forecast_yearly_{start_date}.csv"
yearly_df.to_csv(filename, index=False, encoding='utf-8-sig')
print(f"\n✅ {len(yearly_df)} 개 레코드 수집 완료!")
print(f"📁 저장 위치: {filename}")
return yearly_df
# 사용 예제
api = SolarForecastAPI("YOUR_SERVICE_KEY")
# 2025년 1월 1일부터 365일 데이터 수집
yearly_data = collect_yearly_data(api, "20250101", num_days=365)
⚠️ 주의사항:
중간에 인터넷 끊기면 눈물 나니까, 안정적인 환경에서 돌리세요.
1년 치를 돌리면 요청 횟수가 수천 건이 넘습니다. (약 1~2시간 소요)
7단계: 데이터로 인사이트 발굴하기
힘들게 모은 데이터, 썩히면 안 되죠. 간단한 분석으로 인사이트를 뽑아봅시다.
# 1년 데이터 로드
df = pd.read_csv('solar_forecast_yearly_20250101.csv')
# 숫자형으로 변환
numeric_cols = ['temp', 'humid', 'ws', 'icsrPred', 'useRate',
'solgenKpx', 'solgenBtm', 'solgenPpa', 'solgenAll']
for col in numeric_cols:
df[col] = pd.to_numeric(df[col], errors='coerce')
# 📊 분석 예제 1: 월별 총 발전량
df['date'] = pd.to_datetime(df['collect_date'])
df['month'] = df['date'].dt.month
monthly_generation = df.groupby('month')['solgenAll'].sum()
print("📈 월별 총 발전량 (MWh):")
print(monthly_generation)
# 📊 분석 예제 2: 시도별 평균 발전량
city_avg = df.groupby('city')['solgenAll'].mean().sort_values(ascending=False)
print("\n🌍 시도별 평균 발전량 (MWh):")
print(city_avg)
# 📊 분석 예제 3: 기온과 발전량의 상관관계
correlation = df[['temp', 'solgenAll']].corr()
print("\n🔗 기온-발전량 상관계수:")
print(correlation)
# 📊 분석 예제 4: 계절별 비교
df['season'] = df['month'].map({
12: '겨울', 1: '겨울', 2: '겨울',
3: '봄', 4: '봄', 5: '봄',
6: '여름', 7: '여름', 8: '여름',
9: '가을', 10: '가을', 11: '가을'
})
seasonal_gen = df.groupby('season')['solgenAll'].agg(['sum', 'mean', 'count'])
print("\n🌞 계절별 발전량 통계:")
print(seasonal_gen)결과가 어떻게 나왔나요? 보통 태양광은 5월(봄)이 발전량이 가장 좋고, 한여름에는 오히려 효율이 떨어진다는 사실, 데이터로 직접 확인하셨나요?
8단계: 실전 예제 – 완성된 코드
앞의 모든 것을 통합한 프로덕션 수준의 완성 코드입니다:
import requests
import pandas as pd
from datetime import datetime, timedelta
import time
from typing import Optional, List, Dict
import json # json 모듈 추가
class SolarForecastAPI:
def __init__(self, service_key: str):
self.service_key = unquote(service_key)
self.base_url = "https://apis.data.go.kr/B552070/forecastService"
self.endpoint = "/oamsFile31"
def get_forecast(self, pred_date, city):
"""예측 데이터 조회"""
# pred_date로부터 base_date 자동 계산
try:
pred_date_obj = datetime.strptime(pred_date, "%Y%m%d")
base_date = (pred_date_obj - timedelta(days=1)).strftime("%Y%m%d")
except ValueError:
print(f"오류: 날짜 형식이 올바르지 않습니다. YYYYMMDD 형식으로 입력해주세요.")
return None
params = {
'serviceKey': self.service_key,
'pageNo': 1,
'numOfRows': 10,
'apiType': 'json',
'baseDate': base_date,
'city': city,
'predDate': pred_date
}
try:
response = requests.get(self.base_url + self.endpoint, params=params, timeout=10)
response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx)
return response.json()
except requests.exceptions.Timeout:
print(f"오류: API 요청이 타임아웃되었습니다. ({city}, 예측일: {pred_date})")
return None
except requests.exceptions.HTTPError as e:
print(f"오류: HTTP 오류 {e.response.status_code} ({city}, 예측일: {pred_date})")
print(f" 응답 텍스트: {e.response.text.strip()}") # 응답 텍스트를 포함하여 디버깅 정보 추가
return None
except requests.exceptions.RequestException as e:
print(f"오류: API 요청 중 오류 발생 - {e} ({city}, 예측일: {pred_date})")
return None
except json.JSONDecodeError:
print(f"오류: API 응답이 유효한 JSON 형식이 아닙니다. ({city}, 예측일: {pred_date})")
# 응답 텍스트가 너무 길 수 있으므로 일부만 출력하거나, 필요시 전체 출력
print(f" 응답 텍스트 (일부): {response.text[:200].strip()}...")
return None
def parse_response(self, response: Dict) -> Optional[pd.DataFrame]:
"""응답 데이터 파싱"""
if response is None:
return None
if response.get('header', {}).get('resultCode') != '00':
print(f"API 오류: {response.get('header', {}).get('resultMsg')}")
return None
items = response.get('body', {}).get('items', {}).get('item', [])
if not items:
return None
if isinstance(items, dict):
items = [items]
return pd.DataFrame(items)
def collect_data(self, start_date: str, num_days: int = 365,
cities: List[str] = None) -> Optional[pd.DataFrame]:
"""장기간 데이터 수집"""
if cities is None:
cities = ["서울", "부산", "대구", "인천", "광주", "대전", "울산", "세종",
"경기", "강원", "충북", "충남", "전북", "전남", "경북", "경남", "제주"]
all_data = []
start_date_obj = datetime.strptime(start_date, "%Y%m%d")
success_count = 0
print(f"\n{'='*80}")
print(f"📅 데이터 수집 시작: {start_date}부터 {num_days}일간")
print(f"{'='*80}\n")
for day_offset in range(num_days):
current_date = start_date_obj + timedelta(days=day_offset)
current_date_str = current_date.strftime("%Y%m%d")
print(f"[{day_offset + 1}/{num_days}] {current_date_str} 처리 중... ", end="")
day_success = 0
for city in cities:
response = self.get_forecast(current_date_str, city)
df = self.parse_response(response)
if df is not None:
all_data.append(df)
day_success += 1
if day_success > 0:
print(f"✅ ({day_success}개 시도)")
success_count += 1
else:
print("⚠️ (데이터 없음)")
time.sleep(1) # API 요청 제한 회피
if not all_data:
print("❌ 수집된 데이터가 없습니다.")
return None
combined_df = pd.concat(all_data, ignore_index=True)
# 숫자형 변환
numeric_cols = ['temp', 'humid', 'ws', 'solgenAll']
for col in numeric_cols:
combined_df[col] = pd.to_numeric(combined_df[col], errors='coerce')
# 저장
filename = f"solar_forecast_{start_date}_{num_days}days.csv"
combined_df.to_csv(filename, index=False, encoding='utf-8-sig')
print(f"\n✅ 수집 완료!")
print(f"📊 총 {len(combined_df)} 레코드")
print(f"📁 저장 위치: {filename}")
return combined_df
def collect_yearly_data(self, start_date, num_days=365, cities=None):
"""
여러 날짜에 대해 전국 태양광 발전량 데이터 수집
Args:
api: SolarForecastAPI 인스턴스
start_date: 시작 날짜 (YYYYMMDD)
num_days: 수집 기간 (기본값: 365일)
cities: 시도 리스트
"""
if cities is None:
cities = ["서울", "부산", "대구", "인천", "광주", "대전", "울산", "세종",
"경기", "강원", "충북", "충남", "전북", "전남", "경북", "경남", "제주"]
all_yearly_data = []
start_date_obj = datetime.strptime(start_date, "%Y%m%d")
print(f"\n{'='*80}")
print(f"📅 {start_date}부터 {num_days}일간 전국 태양광 발전량 데이터 수집")
print(f"{'='*80}\n")
for day_offset in range(num_days):
current_date = start_date_obj + timedelta(days=day_offset)
current_date_str = current_date.strftime("%Y%m%d")
print(f"[{day_offset + 1}/{num_days}] {current_date_str} 데이터 수집 중...")
for city in cities:
try:
response = self.get_forecast(current_date_str, city)
df = self.parse_response(response)
if df is not None:
df['collect_date'] = current_date_str
all_yearly_data.append(df)
except Exception as e:
print(f" ⚠️ {city} 오류: {e}")
# API 요청 제한 회피 (1초 대기)
time.sleep(1)
# 모든 데이터 통합
yearly_df = pd.concat(all_yearly_data, ignore_index=True)
# CSV로 저장
filename = f"solar_forecast_yearly_{start_date}.csv"
yearly_df.to_csv(filename, index=False, encoding='utf-8-sig')
print(f"\n✅ {len(yearly_df)} 개 레코드 수집 완료!")
print(f"📁 저장 위치: {filename}")
return yearly_df
# 🚀 실행
if __name__ == "__main__":
service_key = "YOUR_SERVICE_KEY"
api = SolarForecastAPI(service_key)
# 1년 데이터 수집
data = api.collect_data("20250101", num_days=365)
if data is not None:
print("\n📈 데이터 미리보기:")
print(data[['city', 'predDate', 'hh', 'temp', 'solgenAll']].head(20))
마치며
이제 여러분은 공공데이터포털의 API를 자유자재로 다루는 능력을 얻었습니다. 오늘 배운 내용을 응용하면 날씨 예보 앱을 만들 수도, 태양광 수익률 계산기를 만들 수도 있습니다.
오늘의 핵심 요약:
- ✅ requests로 API 문을 두드린다.
- ✅ JSON 응답을 딕셔너리로 깐다.
- ✅ Pandas로 예쁜 표를 만든다.
- ✅ 반복문으로 빅데이터를 수집한다.
데이터는 널려있습니다. 줍는 사람이 임자입니다. 지금 바로 시작해보세요!
🎯 이 가이드에서 배운 것:
- ✅ 공공데이터 API 기본 사용법
- ✅ JSON 데이터 파싱 및 처리
- ✅ Pandas를 이용한 데이터 정리
- ✅ 클래스 기반 코드 구조화
- ✅ 장기간 데이터 자동 수집
- ✅ 수집 데이터 분석
