[전자공시 Dart API] 전체 상장 회사의 상세정보(기업개황) 수집 및 저장

전자공시 Dart API를 이용해서 상장회사의 리스트를 가져와서 저장해보도록 하겠습니다.

공시정보 > 고유번호에서 제공되는 데이터는 기본 정보(회사코드, 회사명, 주식코드)만 있어,

법인구분(코스피, 코스닥)과 업종 코드 및 상세정보를 보기 위해서는 공시정보 > 기업개황의 정보와 합쳐야 합니다.

 

가입이 되어 있으신분은 2번부터 진행하시면 됩니다.

 

1. 가입 및 인증키 확인(Skip 가능)

2. 전체 회사리스트 수집

3. 회사별 기업개황(회사 상세정보) 정보 수집 및 저장

 

 

1. 가입 및 인증키 확인(Skip 가능)

https://yenpa.tistory.com/23

[전자공시 Dart API] 가입 및 인증키(API Key) 확인하기

 

2. 전체 회사리스트 수집

전자공시 API의 공시정보 > 고유번호를 이용해서 전체 회사리스트를 다운받아

압축을 해제하고 결과를 DataFrame에 저장하는 코드입니다.

path는 본인의  PC에 저장하고 싶은 Directory를 지정합니다.

저의 경우는 임의로 path='E:/' 로 지정했습니다.

crtfc_key는 본인의 데이터 입력이 필요합니다.

import io
import zipfile
import requests
from xml.etree.ElementTree import parse
import pandas as pd
import time
import datetime

crtfc_key='YOUR_APIKEY'

#회사정보 가져오기
#임의로 'E:/' 로 설정 및 다운로드 파일 corpcode.zip으로 설정
path='E:/'
filename='corpcode.zip'

url = 'https://opendart.fss.or.kr/api/corpCode.xml'
params = {
    'crtfc_key': crtfc_key,
}
results=requests.get(url, params=params)

file = open(path+filename, 'wb')
file.write(results.content)
file.close()

zipfile.ZipFile(path+filename).extractall(path)

tree = parse(path + 'CORPCODE.xml')
root = tree.getroot()
li=root.findall('list')
corp_code,corp_name,stock_code,modify_date=[],[],[],[]
for d in li:
    corp_code.append(d.find('corp_code').text)
    corp_name.append(d.find('corp_name').text)
    stock_code.append(d.find('stock_code').text)
    modify_date.append(d.find('modify_date').text)
corps_df = pd.DataFrame({'corp_code':corp_code,'corp_name':corp_name,
         'stock_code':stock_code,'modify_date':modify_date})

corps_df = corps_df.loc[corps_df['stock_code']!=' ',:].reset_index(drop=True)

전체 회사리스트를 e:/corpcode.zip 이라는 경로의 파일명으로 저장합니다.

path='E:/'
filename='corpcode.zip'

url = 'https://opendart.fss.or.kr/api/corpCode.xml'
params = {
    'crtfc_key': crtfc_key,
}
results=requests.get(url, params=params)

file = open(path+filename, 'wb')
file.write(results.content)
file.close()

다운로드 받은 zip file을 압축 해제합니다.

zipfile.ZipFile(path+filename).extractall(path)

압축 해제한 xml파일의 내용에서 회사 정보들을 가져와서 corps_df라는 DataFrame에 저장합니다.

corps_df 내용중 회사의 stock_code(주식코드)가 있는 회사만 다시 저장합니다.

tree = parse(path + 'CORPCODE.xml')
root = tree.getroot()
li=root.findall('list')
corp_code,corp_name,stock_code,modify_date=[],[],[],[]
for d in li:
    corp_code.append(d.find('corp_code').text)
    corp_name.append(d.find('corp_name').text)
    stock_code.append(d.find('stock_code').text)
    modify_date.append(d.find('modify_date').text)
corps_df = pd.DataFrame({'corp_code':corp_code,'corp_name':corp_name,
         'stock_code':stock_code,'modify_date':modify_date})

corps_df = corps_df.loc[corps_df['stock_code']!=' ',:].reset_index(drop=True)

 

3. 회사별 기업개황(회사 상세정보) 정보 수집

2번에서 구한 회사별로 공시정보 > 기업개황의 정보를 가져와서 결과를 DataFrame으로 저장합니다.

#전체결과저장
result_all=[]
corf_detail=pd.DataFrame()

print('총 회사수는 : ' + str(corps_df.shape[0]))

# 수집한 회사에 대해서 for문.
for i, r in corps_df.iterrows():
    #if i == 2:
    #    break
    
    #없으면 어느 시점에서 에러발생
    time.sleep(0.1)
    
    print('i = ' + str(i))
    corp_code=str(r['corp_code'])
    corp_name=r['corp_name']

    url = 'https://opendart.fss.or.kr/api/company.json'
    params = {
        'crtfc_key': crtfc_key,
        'corp_code' : corp_code,
    }

    results = requests.get(url, params=params).json()
    # 응답이 정상 '000' 일 경우에만 데이터 수집
    if results['status'] == '000':
        result_all.append(results)    

corp_detail=pd.DataFrame(result_all)

result_all list는 전체 회사의 results dictionary형태의 데이터를 모아놓은 List입니다.

corf_detail은 list 결과를 DataFrame으로 저장했습니다.

#전체결과저장
result_all=[]
corf_detail=pd.DataFrame()

회사별로 기업개황 API 정보를 가져오는 부분이 되겠습니다.

회사 별로 정보를 확인하기 위해 for문을 작성했고, time.sleep(0.1)은 for문 별로 잠시 delay를 주기 위해 넣었습니다.

0.03, 0.05 등은 중간에 1000개 정도에서 Error가 발생했는데 0.1로 하니 정상 수집이 되었습니다.

# 수집한 회사에 대해서 for문.
for i, r in corps_df.iterrows():
    #if i == 2:
    #    break
    
    #없으면 어느 시점에서 에러발생
    time.sleep(0.1)

실제 API정보를 가져오는 부분입니다.

print('i = ' + str(i))
corp_code=str(r['corp_code'])
corp_name=r['corp_name']

url = 'https://opendart.fss.or.kr/api/company.json'
params = {
    'crtfc_key': crtfc_key,
    'corp_code' : corp_code,
}

results = requests.get(url, params=params).json()
# 응답이 정상 '000' 일 경우에만 데이터 수집
if results['status'] == '000':
    result_all.append(results)

List를 DataFrame으로 변환했습니다.

corp_detail=pd.DataFrame(result_all)

 

회사 리스트 수집에 시간이 소요되니 디스크 및 DB등에 저장해서 사용하시면 좋을것 같습니다.

저는 디스크에 저장하도록 하겠습니다.

corp_detail.to_csv('e:/회사상세정보.csv')

 

 

이상으로 상장회사의 리스트를 가져와서 저장하는 방법을 알아봤습니다.