혹시 아직도 태양광 발전량 예측을 위해 작년 날씨 데이터를 엑셀로 한땀한땀 복붙하고 계신가요? 진정한 에너지 데이터 고수라면 API 하나로 전국 50개 지역, 1년치 일사량·일조 데이터를 10분 만에 자동 수집합니다.
2025년, “데이터가 석유”라는 말은 이제 식상합니다. 데이터는 공기와도 같습니다. 특히 태양광 에너지 분야에서 일사량과 일조시간 데이터는 사업의 성패를 좌우하는 핵심 연료입니다. 문제는 이 연료를 얼마나 빠르고 정확하게 조달하느냐입니다.
다행히 기상청이 어마어마한 데이터를 무료로 공개했습니다. 바로 관측-통계(일사, 일조) 묶음형 조회서비스 API입니다. 이 API의 진정한 가치는 분(Minute) 단위 관측 데이터와 시·일·월·년 단위 통계 데이터를 한 번에, 그것도 가벼운 텍스트 형태로 제공한다는 점입니다. 데이터 수집과 전처리의 허들을 단번에 낮춰주죠.
안녕하세요! Do You Know? 입니다.
지난 포스팅들을 통해 우리는 기상청 API의 세계를 차근차근 정복해 왔습니다. 내일의 발전량을 예측해보기도 하고, LSTM으로 날씨 예측 모델을 만들어보기도 했으며, 평년값으로 과거 패턴을 돌아보기도 했습니다. 하지만 “진짜 고수”는 디테일에 강한 법. 1시간 단위의 뭉툭한 데이터로는 만족할 수 없는 여러분을 위해 오늘의 주제를 준비했습니다.
오늘은 기상청 API의 숨겨진 보물, 관측-통계(일사, 일조) 묶음형 조회서비스를 탈탈 털어봅니다. 무려 분 단위 일사량 관측 데이터까지 한 번에 가져올 수 있는, 태양광 데이터 수집의 게임체인저입니다.
이 포스팅 하나면, 파이썬 코드 몇 줄로 전국의 태양 에너지 데이터를 내 손안에서 쥐락펴락할 수 있습니다. 1년치 데이터를 자동 수집하고, 결측치를 처리한 뒤, 간단한 AI 예측 모델까지 만들어보는 여정—지금 바로 시작합니다.
주요 관측 지점 코드: 서울(108), 인천(112), 수원(119), 강릉(105), 대전(133), 대구(143), 부산(159), 광주(156), 제주(184)
출력 변수 (API가 우리에게 주는 것)
API는 #START7777과 #7777END 사이에 다음과 같은 형식의 텍스트를 돌려줍니다.
실제 응답 포맷:
#START7777
#--------------------------------------------------------------------------------------------------
##관측 데이터
# SI_MI: 일 누적 일사량 매분자료(단위:MJ/m^2)
##통계 데이터
# SI_HR: 일사량(단위:MJ/m^2)
# SI_DAY: 일 합계일사량(단위:MJ/m^2)
# SI_MON: 월 합계일사량(단위:MJ/m^2)
# SI_YEAR: 연 합계일사량(단위:MJ/m^2)
#--------------------------------------------------------------------------------------------------
#=================| 관측데이터 |------------------------통계 데이터-------------------|
# YYMMDDHHMI STN SI SI SI SI SI
# KST ID MI HR DAY MON YEAR
202301011200,108,4.2,1.7,10.8,290.8,5186.0,=
202301011201,108,4.2,1.7,10.8,290.8,5186.0,=
202301011202,108,4.3,1.7,10.8,290.8,5186.0,=
...
#7777END
주의사항:
각 데이터 라인은 콤마로 구분되어 있습니다.(disp가 1인 경우)
각 라인 끝에 = 기호가 붙어 있으므로 파싱할 때 제거해야 합니다.
#==== 등의 구분선은 데이터가 아니므로 건너뛰어야 합니다.
컬럼명
설명
단위
데이터 종류
YYMMDDHHMI
관측 시간 (년월일시분, KST)
–
–
STN
관측 지점 번호
–
–
SI_MI
1분 누적 일사량
MJ/m²
관측
SI_HR
1시간 누적 일사량
MJ/m²
통계
SI_DAY
일 합계 일사량
MJ/m²
통계
SI_MON
월 합계 일사량
MJ/m²
통계
SI_YEAR
연 합계 일사량
MJ/m²
통계
자, 이제 설계도 파악은 끝났습니다. 바로 코딩에 착수합시다.
1단계: 초간단 API 호출 및 파싱
가장 먼저 API를 호출해서 원본 데이터를 가져와 보겠습니다. requests 라이브러리를 사용하고, 복잡한 텍스트를 pandas로 쉽게 파싱하기 위해 io.StringIO를 활용하는 것이 포인트입니다.
import requests
import pandas as pd
import io
BASE_URL = "https://apihub.kma.go.kr/api/typ01/cgi-bin/url/nph-sun_sfc_sts_pkg"
def fetch_kma_solar_data(api_key, stn, tm1, tm2, mode="si", disp="1"):
"""기상청 관측-통계 묶음형 API 데이터 요청 및 파싱"""
params = {
'authKey': api_key,
'stn': stn,
'tm1': tm1,
'tm2': tm2,
'mode': mode, # si: 일사량, ss: 일조시간
'help': '1', # 헤더/단위 설명 포함
'disp': disp
}
try:
response = requests.get(BASE_URL, params=params, timeout=30)
response.raise_for_status()
content = response.text
# '#START7777'~'#7777END' 사이의 데이터만 추출
start_idx = content.find('#START7777')
end_idx = content.find('#7777END')
if start_idx == -1 or end_idx == -1:
print("데이터 범위를 찾을 수 없습니다.")
return pd.DataFrame()
data_section = content[start_idx:end_idx]
lines = data_section.split('\n')
# 파싱할 데이터 라인 추출
data_lines = []
header_found = False
for line in lines:
# 헤더 라인 건너뛰기 (# 또는 ## 또는 #= 등으로 시작)
if line.startswith('#') or line.strip() == '':
continue
# 첫 자리가 숫자로 시작하는 데이터 라인
if line and line[0].isdigit():
# 라인 끝의 '=' 제거
line_cleaned = line.rstrip('=').rstrip(',').strip()
if line_cleaned:
data_lines.append(line_cleaned)
if not data_lines:
print("데이터 라인을 찾을 수 없습니다.")
return pd.DataFrame()
# CSV 포맷으로 변환 및 파싱
csv_text = '\n'.join(data_lines)
df = pd.read_csv(io.StringIO(csv_text), header=None, sep=',')
df = df.dropna(axis=1, how='all') # 빈 컬럼 제거
# 컬럼 이름 재정의 (일사 모드 기준)
if mode == "si":
df.columns = ['YYMMDDHHMI', 'STN', 'SI_MI', 'SI_HR', 'SI_DAY', 'SI_MON', 'SI_YEAR']
elif mode == "ss":
# 일조 모드: 컬럼명 조정 가능
df.columns = ['YYMMDDHHMI', 'STN', 'SUM_SS_MI', 'SS_HR', 'SUM_SS_DAY', 'SUM_SS_HR_DAY', 'SUM_SS_HR_YEAR', 'SSRATE_MON', 'SSRATE_YEAR']
return df
except requests.exceptions.RequestException as e:
print(f"API 요청 실패: {e}")
return pd.DataFrame()
실행 예시
API_KEY = API_KEY # 발급받은 키로 교체
df_seoul_raw = fetch_kma_solar_data(
api_key=API_KEY,
stn='108',
tm1='202301010000',
tm2='202301012359',
mode='si', # 일사량
disp='1'
)
print("=== API 호출 및 파싱 결과 (서울, 2023-01-01) ===")
print(df_seoul_raw.head())
print(f"\n데이터 크기: {df_seoul_raw.shape}")
print(f"컬럼: {df_seoul_raw.columns.tolist()}")
앞으로 여러 지역, 여러 날짜의 데이터를 반복적으로 가져오려면 코드를 재사용하기 쉽게 클래스(Class)로 만드는 것이 현명합니다. 이 클래스 안에 데이터 타입을 변환하고, 기본적인 이상치를 제거하는 로직까지 넣어보겠습니다.
import pandas as pd
import requests
import io
from datetime import datetime, timedelta
import time
from tqdm import tqdm
class KMASolarCollector:
"""기상청 일사량 데이터 수집 및 정제 클래스"""
def __init__(self, api_key):
self.api_key = api_key
self.base_url = "https://apihub.kma.go.kr/api/typ01/cgi-bin/url/nph-sun_sfc_sts_pkg"
def fetch_kma_solar_data(self, stn, tm1, tm2, mode="si", disp="1"):
"""기상청 관측-통계 묶음형 API 데이터 요청 및 파싱"""
params = {
'authKey': self.api_key,
'stn': stn,
'tm1': tm1,
'tm2': tm2,
'mode': mode, # si: 일사량, ss: 일조시간
'help': '1', # 헤더/단위 설명 포함
'disp': disp # 현재 미작동 (항상 고정폭 텍스트)
}
try:
response = requests.get(self.base_url, params=params, timeout=30)
response.raise_for_status()
content = response.text
# '#START7777'~'#7777END' 사이의 데이터만 추출
start_idx = content.find('#START7777')
end_idx = content.find('#7777END')
if start_idx == -1 or end_idx == -1:
print("데이터 범위를 찾을 수 없습니다.")
return pd.DataFrame()
data_section = content[start_idx:end_idx]
lines = data_section.split('\n')
# 파싱할 데이터 라인 추출
data_lines = []
header_found = False
for line in lines:
# 헤더 라인 건너뛰기 (# 또는 ## 또는 #= 등으로 시작)
if line.startswith('#') or line.strip() == '':
continue
# 첫 자리가 숫자로 시작하는 데이터 라인
if line and line[0].isdigit():
# 라인 끝의 '=' 제거
line_cleaned = line.rstrip('=').rstrip(',').strip()
if line_cleaned:
data_lines.append(line_cleaned)
if not data_lines:
print("데이터 라인을 찾을 수 없습니다.")
return pd.DataFrame()
# CSV 포맷으로 변환 및 파싱
csv_text = '\n'.join(data_lines)
df = pd.read_csv(io.StringIO(csv_text), header=None, sep=',')
df = df.dropna(axis=1, how='all') # 빈 컬럼 제거
# 컬럼 이름 재정의 (일사 모드 기준)
if mode == "si":
df.columns = ['YYMMDDHHMI', 'STN', 'SI_MI', 'SI_HR', 'SI_DAY', 'SI_MON', 'SI_YEAR']
elif mode == "ss":
# 일조 모드: 컬럼명 조정 가능
df.columns = ['YYMMDDHHMI', 'STN', 'SUM_SS_MI', 'SS_HR', 'SUM_SS_DAY', 'SUM_SS_HR_DAY', 'SUM_SS_HR_YEAR', 'SSRATE_MON', 'SSRATE_YEAR']
return df
except requests.exceptions.RequestException as e:
print(f"API 요청 실패: {e}")
return pd.DataFrame()
def _clean_data(self, df):
"""데이터 타입 변환 및 기본 정제"""
if df.empty:
return df
# 날짜/시간 타입 변환
df['YYMMDDHHMI'] = pd.to_datetime(df['YYMMDDHHMI'], format='%Y%m%d%H%M')
# 숫자 타입 변환 (오류 발생 시 NaN으로 처리)
numeric_cols = ['STN', 'SI_MI', 'SI_HR', 'SI_DAY', 'SI_MON', 'SI_YEAR']
for col in numeric_cols:
df[col] = pd.to_numeric(df[col], errors='coerce')
# 이상치 처리 (-9와 같은 결측치 코드를 NaN으로)
# 기상청 데이터는 종종 -9, -99 등을 결측치 코드로 사용합니다.
df.replace(-9.0, pd.NA, inplace=True)
df.replace(-99.0, pd.NA, inplace=True)
# 물리적으로 불가능한 음수 일사량 제거
df.loc[df['SI_MI']
실행
collector = KMASolarCollector(API_KEY)
df_seoul_clean = collector.get_clean_data('108', '202301010000', '202301012359')
print("\n=== 정제된 데이터 ===")
print(df_seoul_clean.head())
print("\n정제 후 데이터 타입:")
df_seoul_clean.info()
데이터를 모았으니 이제 분석하고 예측해볼 시간입니다. 여기서 중요한 포인트가 있습니다. 우리가 수집한 SI_MI는 ‘누적’ 데이터입니다. 이를 그대로 학습하면 안 되고, 변화량(미분)을 구해서 ‘순간’ 데이터로 만들어야 합니다.
import pandas as pd
import matplotlib.pyplot as plt
# 1) 데이터 불러오기
df = pd.read_csv('kma_solar_2023_seoul.csv')
# 2) YYMMDDHHMI를 datetime으로 변환
df['YYMMDDHHMI'] = pd.to_datetime(df['YYMMDDHHMI'], errors='coerce')
df = df.set_index('YYMMDDHHMI')
# --- [핵심 수정 파트 시작] ---
# 3) '누적' 데이터를 '순간' 데이터로 변환
# SI_MI는 0시부터 쌓이는 누적값이므로, 분 단위 변화량을 구해야 실제 1분간 내리쬔 햇빛 양이 됨
df['Solar_Min_Flux'] = df['SI_MI'].diff()
# 4) 자정(00:00) 초기화로 인한 음수 값 처리
# 날짜가 바뀌면 누적값이 0으로 리셋되므로 diff가 큰 음수가 됨 -> 0으로 처리
df.loc[df['Solar_Min_Flux']
결과
=== 시간 단위 리샘플링 후 데이터 (MJ/m²) ===
SI_Hourly_Sum
YYMMDDHHMI
2023-01-30 12:00:00 2.2
2023-01-30 13:00:00 2.3
2023-01-30 14:00:00 1.9
2023-01-30 15:00:00 1.5
2023-01-30 16:00:00 0.8
AI 모델링
데이터 불러오기 및 전처리
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
from sklearn.preprocessing import MinMaxScaler
# -----------------------------------------------------------------------------
# 1. 데이터 불러오기 및 전처리 (Pandas 부분)
# -----------------------------------------------------------------------------
# 1) 데이터 불러오기
df = pd.read_csv('kma_solar_2023_seoul.csv')
# 2) 시간 인덱스 설정
df['YYMMDDHHMI'] = pd.to_datetime(df['YYMMDDHHMI'], errors='coerce')
df = df.set_index('YYMMDDHHMI')
# --- [수정된 핵심 로직] ---
# 3) SI_MI 변환: '누적' 데이터를 '순간' 데이터로 변환
# SI_MI는 0시부터 계속 쌓이는 값이므로, 1분 간의 변화량(diff)을 구해야 실제 에너지량이 됨
df['SI_Step'] = df['SI_MI'].diff()
# 4) 자정(00:00) 초기화로 인한 음수 값 처리
# 날짜가 바뀌면 누적값이 0으로 리셋되면서 diff가 큰 음수가 나옴 -> 0으로 처리
df.loc[df['SI_Step']
결과
=== 시간 단위 리샘플링(Sum) 및 전처리 후 데이터 ===
SI_MI
YYMMDDHHMI
2023-01-30 12:00:00 2.2
2023-01-30 13:00:00 2.3
2023-01-30 14:00:00 1.9
2023-01-30 15:00:00 1.5
2023-01-30 16:00:00 0.8
단위: MJ/m² (시간당 누적 일사량)
데이터셋 준비
# -----------------------------------------------------------------------------
# 2. 데이터셋 준비 (Numpy & Scaling)
# -----------------------------------------------------------------------------
data = df_hourly['SI_MI'].values.reshape(-1, 1)
# MinMax 스케일링 (0~1 사이로 정규화)
scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(data)
# 학습/테스트 분할 (80:20)
train_size = int(len(scaled_data) * 0.8)
train_data = scaled_data[:train_size]
test_data = scaled_data[train_size:]
# 시퀀스 데이터 생성 함수 (Sliding Window)
def create_sequences(data, seq_length):
xs = []
ys = []
for i in range(len(data) - seq_length):
x = data[i:(i + seq_length)]
y = data[i + seq_length]
xs.append(x)
ys.append(y)
return np.array(xs), np.array(ys)
SEQ_LENGTH = 24 # 24시간 데이터를 보고 다음 1시간 예측
# Numpy 배열 생성
X_train, y_train = create_sequences(train_data, SEQ_LENGTH)
X_test, y_test = create_sequences(test_data, SEQ_LENGTH)
# Tensor로 변환
X_train = torch.from_numpy(X_train).float()
y_train = torch.from_numpy(y_train).float()
X_test = torch.from_numpy(X_test).float()
y_test = torch.from_numpy(y_test).float()
