Top 14 Dart 공시 크롤링 9836 Good Rating This Answer

search 함수는 HTML 내부 스크립트에 구현되어 있다

function search(page) { var frm = findForm(“searchForm”); if(page==”) page=1; frm.currentPage.value = page; xajax.blockUI = false; xajax.blockTarget = “listContents”; window.scrollTo(0,0); xajax.sendForm(“searchForm”, “/dsac001/search.ax”, function(str) { getRef(“listContents”).innerHTML = str; $j(“.table_list tr:nth-child(even)”).addClass(“even”); var totalCnt = comma($j(“#totalCnt”).val()); $j(‘#txtCB’).text(totalCnt); }); }

PHP용 ajax 라이브러리 xajax를 사용하는 것을 알 수 있다

hidden Form인 “searchForm”에 페이지 번호를 대입하여 sendForm으로 POST 방식 호출을 하는 것으로 유추된다

실제로 2번 페이지를 클릭하여 웹브라우저 개발자 도구 네트워크 분석을 해보면~

POST 호출로 Form Data에 currentPage=2가 인자로 넘어간 것을 알 수 있다

Form Data 형식은 아래와 같다

currentPage=2&maxResults=&maxLinks=&sort=rpt&series=desc&pageGrouping=Y&mdayCnt=1&selectDate=2021.10.01&textCrpCik=

이제 전체 공시문서 리스트를 가져오기 위한 사전 분석은 대충 완료되었다!

2. Python Prototype

HTML GET, POST만 사용하면 되기 때문에 requests 라이브러리만 사용해도 충분히 구현할 수 있다

(태그 분석은 lxml 라이브러리를 사용하도록 한다)

우선 페이지 개수부터 가져와보자 (예를 위해 2021년 9월 30일 전체(mainAll) 검색

import requests import lxml.html url = ‘http://dart.fss.or.kr/dsac001/mainAll.do’ params = {‘selectDate’: ‘2021.09.30’} response = requests.get(url, params=params) tree = lxml.html.fromstring(response.text) tag_page_skip = tree.find_class(‘pageSkip’)[0] tag_ul = tag_page_skip.find(‘ul’) tag_li_list = tag_ul.findall(‘li’) page_count = len(tag_li_list)

In [1]: page_count Out[1]: 7

7개 페이지 존재여부를 제대로 가져왔다

1번 페이지는 최초 로딩시에 테이블이 전부 로딩되어 있으므로, 다음과 같이 Element로부터 tbList 테이블의 모든 행들을 반환하는 1.2. 에서 분석한대로 함수를 만들어보자

import re def getDocListFromTree(element: lxml.html.HtmlElement) -> list: tbList = element.find_class(‘tbList’)[0] tbody = tbList.find(‘tbody’) tr_list = tbody.findall(‘tr’) element_list = [] corp_code_regex = re.compile(r”[0-9]{8}”) for tr in tr_list: # 시간 td_list = tr.findall(‘td’) strtime = td_list[0].text.strip() # 공시대상회사 name_span = td_list[1].find(‘span’) name_a = name_span.find(‘a’) name = name_a.text name = name.replace(‘\t’, ”) name = name.replace(‘

‘, ”) name = name.strip() # 공시대상회사 – 회사고유번호 name_attrib_href = name_a.attrib.get(‘href’) corp_code_search = corp_code_regex.search(name_attrib_href) corp_code = ” if corp_code_search is not None: span = corp_code_search.span() corp_code = name_attrib_href[span[0]:span[1]] # 보고서명 rpt_a = td_list[2].find(‘a’) rpt = rpt_a.text if rpt is None: # 의 tail로 붙는 경우가 있음 rpt_span = rpt_a.find(‘span’) rpt = rpt_span.text + rpt_span.tail rpt = rpt.replace(‘\t’, ”) rpt = rpt.replace(‘

‘, ”) rpt = rpt.replace(‘ ‘, ‘ ‘) rpt = rpt.strip() # 보고서명 – 보고서번호 rpt_attrib_id = rpt_a.attrib.get(‘id’) rpt_no = rpt_attrib_id.split(‘_’)[-1] # 제출인 flr_nm = td_list[3].text # 접수일자 rcept_dt = td_list[4].text # 비고 tag_rm = td_list[5] rm = ‘ ‘.join([x.text for x in tag_rm.findall(‘span’)]) element = {‘시간’: strtime, ‘고유번호’: corp_code, ‘공시대상회사’: name, ‘보고서명’: rpt, ‘보고서번호’: rpt_no, ‘제출인’: flr_nm, ‘접수일자’: rcept_dt, ‘비고’: rm} element_list.append(element) return element_list

테스트를 위해 5건만 출력해보자

In [2]: print(‘

‘.join([str(x) for x in getDocListFromTree(tree)[:5]])) Out[2]: {‘시간’: ’19:36′, ‘고유번호’: ‘00110875’, ‘공시대상회사’: ‘대신정보통신’, ‘보고서명’: ‘거래처와의거래중단’, ‘보고서번호’: ‘20210930900921’, ‘제출인’: ‘대신정보통신’, ‘접수일자’: ‘2021.09.30’, ‘비고’: ‘코’} {‘시간’: ’18:32′, ‘고유번호’: ‘00861997’, ‘공시대상회사’: ‘인포마크’, ‘보고서명’: ‘[첨부추가]주요사항보고서(유상증자결정)’, ‘보고서번호’: ‘20210929000599’, ‘제출인’: ‘인포마크’, ‘접수일자’: ‘2021.09.30’, ‘비고’: ”} {‘시간’: ’18:20′, ‘고유번호’: ‘00313649’, ‘공시대상회사’: ‘현대바이오’, ‘보고서명’: ‘투자판단관련주요경영사항 (COVID-19 치료제 CP-COV03의 1상 임상시험계획 신청서 제출)’, ‘보고서번호’: ‘20210930900900’, ‘제출인’: ‘현대바이오’, ‘접수일자’: ‘2021.09.30’, ‘비고’: ‘코’} {‘시간’: ’18:14′, ‘고유번호’: ‘00187725’, ‘공시대상회사’: ‘코콤’, ‘보고서명’: ‘[기재정정]신규시설투자등’, ‘보고서번호’: ‘20210930900911’, ‘제출인’: ‘코콤’, ‘접수일자’: ‘2021.09.30’, ‘비고’: ‘코’} {‘시간’: ’18:10′, ‘고유번호’: ‘00288343’, ‘공시대상회사’: ‘삼영이엔씨’, ‘보고서명’: ‘주요사항보고서(소송등의제기)’, ‘보고서번호’: ‘20210930000675’, ‘제출인’: ‘삼영이엔씨’, ‘접수일자’: ‘2021.09.30’, ‘비고’: ”}

제대로 된다!

3. Full Document List Up

이제 1.3.에서 분석한 대로 각 페이지별로 POST request를 수행해보자

1번 페이지는 url 최초 로딩 시에 로딩이 되어 있으므로 2번 페이지부터 POST해보자

for page_no in range(1, page_count): url = ‘https://dart.fss.or.kr/dsac001/search.ax’ data = { ‘currentPage’: page_no + 1, ‘selectDate’: ‘2021.09.30’, ‘mdayCnt’: 0 } response = requests.post(url, data=data) tree = lxml.html.fromstring(response.text) doc_list_page = getDocListFromTree(tree) doc_list.extend(doc_list_page)

In [3]: len(doc_list) Out[3]: 688 In [4]: print(‘

‘.join([str(x) for x in getDocListFromTree(tree)[-5:]])) Out[4]: {‘시간’: ’07:30′, ‘고유번호’: ‘01490990’, ‘공시대상회사’: ‘케이비국민카드제칠차유동화전문유한회사’, ‘보고서명’: ‘[기재정정]자산양도등의등록신청서’, ‘보고서번호’: ‘20210914000183’, ‘제출인’: ‘케이비국민카드’, ‘접수일자’: ‘2021.09.14’, ‘비고’: ”} {‘시간’: ’07:30′, ‘고유번호’: ‘01583740’, ‘공시대상회사’: ‘티월드제육십팔차유동화전문유한회사’, ‘보고서명’: ‘투자설명서’, ‘보고서번호’: ‘20210929000601’, ‘제출인’: ‘티월드제육십팔차유동화전문유한회사’, ‘접수일자’: ‘2021.09.30’, ‘비고’: ”} {‘시간’: ’07:30′, ‘고유번호’: ‘01350416’, ‘공시대상회사’: ‘파인더제일차유동화전문유한회사’, ‘보고서명’: ‘[기재정정]자산양도등의등록신청서’, ‘보고서번호’: ‘20210923000188’, ‘제출인’: ‘삼성카드’, ‘접수일자’: ‘2021.09.23’, ‘비고’: ”} {‘시간’: ’07:30′, ‘고유번호’: ‘00662952’, ‘공시대상회사’: ‘휘닉스중앙에프앤비’, ‘보고서명’: ‘동일인등출자계열회사와의상품ㆍ용역거래변경’, ‘보고서번호’: ‘20210929000605’, ‘제출인’: ‘휘닉스중앙에프앤비’, ‘접수일자’: ‘2021.09.30’, ‘비고’: ‘공’} {‘시간’: ’07:30′, ‘고유번호’: ‘00662952’, ‘공시대상회사’: ‘휘닉스중앙에프앤비’, ‘보고서명’: ‘동일인등출자계열회사와의상품ㆍ용역거래변경’, ‘보고서번호’: ‘20210929000603’, ‘제출인’: ‘휘닉스중앙에프앤비’, ‘접수일자’: ‘2021.09.30’, ‘비고’: ‘공’}

2021년 9월 30일에 공시죈 전체 688건의 문서를 정상적으로 가져온 것을 알 수 있다

4. Wrap-up

위에서 구현한 프로토타입 코드를 하나의 함수로 구현하면 다음과 같다

– 리스트를 pandas Dataframe으로 변환

import re import requests import lxml.html import pandas as pd def getDocListFromTree(element: lxml.html.HtmlElement) -> list: tbList = element.find_class(‘tbList’)[0] tbody = tbList.find(‘tbody’) tr_list = tbody.findall(‘tr’) element_list = [] corp_code_regex = re.compile(r”[0-9]{8}”) for tr in tr_list: # 시간 td_list = tr.findall(‘td’) strtime = td_list[0].text.strip() # 공시대상회사 name_span = td_list[1].find(‘span’) name_a = name_span.find(‘a’) name = name_a.text name = name.replace(‘\t’, ”) name = name.replace(‘

‘, ”) name = name.strip() # 공시대상회사 – 회사고유번호 name_attrib_href = name_a.attrib.get(‘href’) corp_code_search = corp_code_regex.search(name_attrib_href) corp_code = ” if corp_code_search is not None: span = corp_code_search.span() corp_code = name_attrib_href[span[0]:span[1]] # 보고서명 rpt_a = td_list[2].find(‘a’) rpt = rpt_a.text if rpt is None: # 의 tail로 붙는 경우가 있음 rpt_span = rpt_a.find(‘span’) rpt = rpt_span.text + rpt_span.tail rpt = rpt.replace(‘\t’, ”) rpt = rpt.replace(‘

‘, ”) rpt = rpt.replace(‘ ‘, ‘ ‘) rpt = rpt.strip() # 보고서명 – 보고서번호 rpt_attrib_id = rpt_a.attrib.get(‘id’) rpt_no = rpt_attrib_id.split(‘_’)[-1] # 제출인 flr_nm = td_list[3].text # 접수일자 rcept_dt = td_list[4].text # 비고 tag_rm = td_list[5] rm = ‘ ‘.join([x.text for x in tag_rm.findall(‘span’)]) element = {‘시간’: strtime, ‘고유번호’: corp_code, ‘공시대상회사’: name, ‘보고서명’: rpt, ‘보고서번호’: rpt_no, ‘제출인’: flr_nm, ‘접수일자’: rcept_dt, ‘비고’: rm} element_list.append(element) return element_list def getDartDailyDocumentList(date: str) -> pd.DataFrame: url = ‘http://dart.fss.or.kr/dsac001/mainAll.do’ params = {‘selectDate’: date} response = requests.get(url, params=params) tree = lxml.html.fromstring(response.text) doc_list = getDocListFromTree(tree) tag_page_skip = tree.find_class(‘pageSkip’)[0] tag_ul = tag_page_skip.find(‘ul’) tag_li_list = tag_ul.findall(‘li’) page_count = len(tag_li_list) for page_no in range(1, page_count): url = ‘https://dart.fss.or.kr/dsac001/search.ax’ data = { ‘currentPage’: page_no + 1, ‘selectDate’: date, ‘mdayCnt’: 0 } response = requests.post(url, data=data) tree = lxml.html.fromstring(response.text) doc_list_page = getDocListFromTree(tree) doc_list.extend(doc_list_page) df = pd.DataFrame(doc_list) return df

In [5]: df = getDartDailyDocumentList(‘2021.09.30’) In [6]: df.shape Out[6]: (688, 8) In [7]: sum(df.duplicated()) Out[7]: 0 In [8]: df.head(5) Out[8]: 고유번호 공시대상회사 … 접수일자 제출인 0 00110875 대신정보통신 … 2021.09.30 대신정보통신 1 00861997 인포마크 … 2021.09.30 인포마크 2 00313649 현대바이오 … 2021.09.30 현대바이오 3 00187725 코콤 … 2021.09.30 코콤 4 00288343 삼영이엔씨 … 2021.09.30 삼영이엔씨 [5 rows x 8 columns] In [9]: df.tail(5) Out[9]: 고유번호 공시대상회사 … 접수일자 제출인 683 01490990 케이비국민카드제칠차유동화전문유한회사 … 2021.09.14 케이비국민카드 684 01583740 티월드제육십팔차유동화전문유한회사 … 2021.09.30 티월드제육십팔차유동화전문유한회사 685 01350416 파인더제일차유동화전문유한회사 … 2021.09.23 삼성카드 686 00662952 휘닉스중앙에프앤비 … 2021.09.30 휘닉스중앙에프앤비 687 00662952 휘닉스중앙에프앤비 … 2021.09.30 휘닉스중앙에프앤비 [5 rows x 8 columns]

중복되는 실수없이 함수 호츨 한번으로 전체 리스트를 데이터프레임으로 변환하여 얻어올 수 있다

5. Etc

여러 개의 페이지를 request할 경우 순차적으로 하면 시간이 다소 소요되는 단점이 있다

import time tm = time.perf_counter() df = getDartDailyDocumentList(‘2021.09.30’) elapsed = time.perf_counter() – tm

In [10]: elapsed Out[10]: 1.4198498000005202

7개 페이지 요청 및 파싱 소요시간: 1.5초 내외

여러 개의 페이지를 크롤링할 때는 async 코루틴으로 비동기 프로그래밍을 하면 효과적이다

async를 직접 구현하는 것도 어렵지 않지만 약간 번거롭기 때문에 requests-HTML의 AsyncHTMLSession을 사용하면 굉장히 간단하게 비동기 함수를 짤 수 있다

(lxml 객체를 지원하기 때문에 함수 변경도 거의 없다)

import re import lxml.html import pandas as pd from functools import partial from requests_html import AsyncHTMLSession def getDartDailyDocumentListAsync(date: str) -> pd.DataFrame: asession = AsyncHTMLSession() async def _async_get(url, params): response = await asession.get(url, params=params) return response async def _async_post_parse(url, data): response = await asession.post(url, data=data) doc_list = getDocListFromTree(response.html.lxml) return doc_list url = ‘http://dart.fss.or.kr/dsac001/mainAll.do’ params = {‘selectDate’: date} results = asession.run(lambda: _async_get(url, params)) tree = results[0].html.lxml df = getDocListFromTree(tree) tag_page_skip = tree.find_class(‘pageSkip’)[0] tag_ul = tag_page_skip.find(‘ul’) tag_li_list = tag_ul.findall(‘li’) page_count = len(tag_li_list) url = ‘https://dart.fss.or.kr/dsac001/search.ax’ if page_count > 1: data_list = [{ ‘currentPage’: x + 1, ‘selectDate’: date, ‘mdayCnt’: 0 } for x in range(1, page_count)] results = asession.run(*[partial(_async_post_parse, url, x) for x in data_list]) temp = [df] temp.extend(results) df = pd.concat(temp) return df

import nest_asyncio nest_asyncio.apply() import time tm = time.perf_counter() df = getDartDailyDocumentListAsync(‘2021.09.30’) elapsed = time.perf_counter() – tm

In [11]: elapsed Out[10]: 1.2306523999995989

대충 100ms 이상 수행시간을 단축시켰다

(단축율은 15% 수준이니 사실 큰 의미는 없다 ㅋㅋㅋ)

만약 그럴 일은 없겠지만 페이지가 수십개 이상이라면 확실이 비동기 구문의 속도 차이를 체감할 수 있을 것으로 예상한다

테스트하려고 임의로 동일 페이지 여러번 호출하게 만들어봤는데, 몇 번 호출하더니 DART로부터 ban당했다… (서버 과부하 방지를 위해 요청 자체를 블록해버리네)

DART 페이지를 짧은 시간내에 지나치게 많이 호출하지 않도록 주의하자!!

끝~!

반응형

전자공시시스템(DART) 오픈API 이용하여 재무제표 크롤링하기 with 파이썬 (2)

저기 [Network] 탭이 보이실 겁니다. 누르면 위와 같은 창이 나타납니다.

이제 웹페이지에 있는 재무제표를 누르시면 됩니다.

그러면 위와 같은 반응이 하나 나올 겁니다.

오른쪽 클릭 [Copy] – [Copy link address] 해봅시다.

저는 아래와 같은 주소가 나왔습니다.

이 url이 재무제표로 넘어가는 url임을 알 수 있습니다.

여기서 우리는 rcpNo는 손쉽게 알 수 있었지만 dcmNo은 알기가 어렵습니다.

우리가 실제로 클릭하는 공시창의 왼쪽 부분의 페이지 소스 코드에는 없었거든요.

참고로 저는 지금 실제로 하나하나 해결해가면서 포스팅 중입니다.

미리 해본 것이 아닙니다.

왜 이런 말을 하나면! 사실 여기서 저는 urllib을 선택한 것을 후회했습니다.

왜냐하면 저는 재무제표를 클릭하기 전에는 dcmNo를 알 수가 없다고 생각합니다.

하지만 그걸 지금까지 몰랐습니다. 그래서 urllib으로 간단하게 하려고 했는데,

urllib는 onclick 기능(즉, 클릭해주는 기능)이 서칭해보니까 없는 것 같더라구요.

(사실 있을수도 있습니다.)

그래서 저는 과감하게 urllib을 버리고 requests와 lxml과 re 모듈을 사용하겠습니다.

지금까지 따라오시느라 배신감이 드시겠지만, 어쩔 수 없습니다.

이런 과정들도 다 도움이 될 거라는 생각에 1편은 안 바꾸도록 하겠습니다.

최종 코드는 이렇습니다. (아직 수정 중입니다. 내일 중에 완성 될 듯 싶네요.)

참고로 1편에 썼던 urllib모듈의 메소드들은 requests모듈로 바꿨습니다.

하나하나 비교해보면서 배우시는 것도 좋을 것 같네요.

그러면 차근차근 분석해보도록 하겠습니다. 천천히 따라오세요.

우리는 request모듈의 get()메소드를 사용해서 응답을 요청했습니다.

그 다음 .text로 html코드를 획득한 뒤에

여기서 lxml모듈을 사용합니다.

lxml모듈의 설명은 아래와 같습니다.

lxml comes with a dedicated Python package for dealing with HTML: lxml.html. It is based on lxml’s HTML parser, but provides a special Element API for HTML elements, as well as a number of utilities for common HTML processing tasks.

우리는 여기서 fromstring이라는 메소드를 사용 할 것입니다.

fromstring(string) : Returns document_fromstring or fragment_fromstring , based on whether the string looks like a full document, or just a fragment.

그 다음 우리는 Xpath라는 개념에 대해 알아봐야 합니다.

Xpath란?

XPath(XML Path Language)는 그 다음 우리는 Xpath라는 개념에 대해 알아봐야 합니다.XPath(XML Path Language)는 W3C 의 표준으로 확장 생성 언어 문서의 구조를 통해 경로 위에 지정한 구문을 사용하여 항목을 배치하고 처리하는 방법을 기술하는 언어이다. XML 표현보다 더 쉽고 약어로 되어 있으며, XSL 변환( XSLT )과 XML 지시자 언어( XPointer )에 쓰이는 언어이다. XPath는 XML 문서의 노드를 정의하기 위하여 경로식을 사용하며, 수학 함수와 기타 확장 가능한 표현들이 있다.

XML 예제 문서 >> >> name= “Wikipedia” launch= “2001-01-05” > >> > language= “English” > en.wikipedia.org > > language= “German” > de.wikipedia.org > > language= “French” > fr.wikipedia.org > > language= “Polish” > pl.wikipedia.org >

name= “Wiktionary” launch= “2002-12-12” > >> > language= “English” > en.wiktionary.org > > language= “French” > fr.wiktionary.org > > language= “Vietnamese” > vi.wiktionary.org > > language= “Turkish” > tr.wiktionary.org >

아래의 XPath 식은

/wikimedia/projects/project/@name

모든 project 요소의 name 속성을 선택하고, 아래의 XPath 식은

/wikimedia/projects/project/editions/edition[@language=”English”]/text()

모든 영문 Wikimedia 프로젝트의 주소( language 속성이 English인 모든 edition 요소의 문자열)를 선택하고, 아래의 XPath 식은

/wikimedia/projects/project[@name=”Wikipedia”]/editions/edition/text()

모든 위키백과의 주소(Wikipedia의 이름 특성을 가진 project 요소 아래에 존재하는 모든 edition 요소의 문자열)를 선택한다.

배울 게 너무 많네요. 우리는 스킵하고 코드만 가져가도록 하죠.

우리는 결국 이러한 과정들을 통해 dcmNo를 구했습니다.

이 부분들이 너무 방대한 분량이라서 이해하기에는 무리가 있습니다.

일단 이런 것들이 있구나 하고 넘어가시면 되겠습니다.

STEP 2. 스크래핑(Scraping)

이제 해당 url에서 우리가 가져오고자 하는 표가 어디에 있는지 확인하겠습니다.

다시 [개발자 도구]에서 이번엔 [Elements] 탭을 선택합니다.

저기 초록색 버튼이 보이시나요?

저 버튼을 누르면 원하는 element를 검사하게 됩니다.

저걸 누른 뒤에 재무제표를 아무데나 눌러봅시다.

HTML의 구조는 대략 이렇습니다.

Here’s a paragraph of text! Learn Data Science Online

Here’s a second paragraph of text! Python

그러면 우리가 앞에서 검사한 element를 볼까요?

td라고 하는 태그에 파란 불이 켜진 것을 확인 할 수 있습니다. td 옆에 tr tbody table 등 있는 것들은 모두 td의 상위 태그들입니다. 여기서 HTML에 대해 간단히 알아보도록 하겠습니니다.HTML의 구조는 대략 이렇습니다.

왼쪽에 있을수록 상위 태그이므로 하나하나 눌러서 확인해보면

table이라는 태그가 내가 선택한 표라는 것을 확인할 수 있습니다.

태그를 사용한다는 것을 인지하고 확인해봅시다.

우리가 아까 전에 간단히 알아본 XPath라는 개념을 여기서 다시 사용합니다.

오른쪽 버튼을 누르고 [Copy] – [Copy XPath] 를 눌러서 XPath를 알아봅시다.

규칙성을 확인해보기 위해서 3개 정도를 알아보도록 하겠습니다.

저는 1번째, 2번째, 그리고 12번째 값의 Xpath를 알아보았습니다.

/html/body/table[2]/tbody/tr[2]/td[2]

/html/body/table[2]/tbody/tr[3]/td[2]

/html/body/table[2]/tbody/tr[13]/td[2]

이런 식의 규칙성을 보이고 있네요.

이렇게 웹페이지의 규칙적인 정보는 대부분 아파트와 같은 구조를 가지고 있습니다.

우리는 이 규칙성을 이용해서 컴퓨터에게 데이터에 접근하도록 할 것입니다.

재무제표에는 여러 table들이 있네요.라는 태그를 상속받은우리가 아까 전에 간단히 알아본 XPath라는 개념을 여기서 다시 사용합니다.오른쪽 버튼을 누르고 [Copy] – [Copy XPath] 를 눌러서 XPath를 알아봅시다.규칙성을 확인해보기 위해서 3개 정도를 알아보도록 하겠습니다.저는 1번째, 2번째, 그리고 12번째 값의 Xpath를 알아보았습니다./html/body/table[2]/tbody/tr[2]/td[2]/html/body/table[2]/tbody/tr[3]/td[2]/html/body/table[2]/tbody/tr[13]/td[2]이런 식의 규칙성을 보이고 있네요.이렇게 웹페이지의 규칙적인 정보는 대부분 아파트와 같은 구조를 가지고 있습니다.우리는 이 규칙성을 이용해서 컴퓨터에게 데이터에 접근하도록 할 것입니다.

전 편에 이어서 포스팅 하도록 하겠습니다.못 보신 분들을 위해 링크 걸겠습니다.우리가 흔히 부르는 크롤링(crawling) 혹은 스크래핑(scraping)이라고 하는 것은,웹 페이지를 그대로 가져와 데이터를 추출해 내는 행위입니다.이러한 작업을 하는 소프트웨어를 크롤러(crawler)라고 부릅니다.우리는 종목 코드와 원하는 기간동안의 보고서를 DataFrame으로 저장하여 손 쉽게 핸들링 할 수 있도록 만들었습니다. 지금부터는 보고서에 있는 재무제표를 크롤링 하겠습니다.재무제표를 크롤링하기 위해서는 페이지를 분석을 해야합니다.저는 과거에 위키피디아를 크롤링 해본적이 있습니다.크롤링을 위해서는 html이나 javascript에 대한 이해가 필요한데, 저도 이 부분은 아는 바가 거의 전혀 없기 때문에 구글링하면서 하나하나 해결해나가도록 하겠습니다.거듭 강조드리지만 문제를 해결하는 능력을 기르는 것이 중요합니다.우선 저는 크롬을 사용하기 때문에 크롬의 기준에서 말씀드리겠습니다.크롬에서 [도구 더 보기] – [개발자 도구]를 선택하면 개발자 도구를 사용할 수 있습니다.다른 브라우저에서도 비슷한 도구들이 있으니 찾아보시기 바랍니다.그러면 이런 형태의 창이 나옵니다.위와 같이 HTML의 형태를 확인했으니 원하는 정보를 뽑아 올 것입니다.웹 사이트의 데이터를 수집하는데 있어서는 2가지의 단계가 있습니다.하지만 그 전에 해야 할 것은 바로 웹페이지에 접속하는 것이겠죠.우리는 재무제표를 스크래핑 하기 전에 우선 재무제표의 url을 알아야합니다.우측에 네트워크 탭을 선택하고 링크를 누르면 시간초 별로 서버에서 응답하는 목록을 확인할 수 있습니다. 우리는 이 기능을 통해서 url을 알아내도록 하겠습니다.이런 개념입니다. 저도 자세하게는 모르겠네요. 깊게 알고 싶지도 않습니다.구글링의 도움을 빌려 코드를 빌려왔습니다. 필요하신 분들은 더 알아보시면 됩니다.이 다음에는 re라는 파이썬의 기본 모듈을 사용합니다.이 모듈은 파이썬에서 정규 표현식을 지원하도록 해줍니다.정규 표현식(Regular Expressions)은 복잡한 문자열을 처리할 때 사용하는 기법으로, 파이썬만의 고유 문법이 아니라 문자열을 처리하는 모든 곳에서 사용된다. 정규 표현식을 배우는 것은 파이썬을 배우는 것과는 또 다른 영역의 과제입니다.

[Python] 다트(DART) 오픈 API 공시(재무제표) 크롤링(Crawling)

반응형

오늘은 상장된 단일기업의 재무제표를 가져오는 크롤링을 해보도록 하겠습니다.

일단, 다트의 공시 API를 가져오기 위해서 Open DART 사이트에 들어갑니다.

opendart.fss.or.kr/

1) 회원가입 후, 인증키를 신청합니다

2) “인증키 신청/관리 > 인증키 관리” 를 통해 승인이 되어 있는지 확인합니다

3) “인증키 신청/관리 > 오픈 API 이용현황”에서 API key를 가지고 옵니다

API key

API key를 가지고 왔다면 크롤링 준비를 다 마쳤습니다.

(1) 상장기업의 고유번호 크롤링

(* 금융업을 제외한 유가증권 및 코스닥 상장회사 )

일단, 재무제표를 크롤링 하기 앞서 각 상장기업의 고유번호를 먼저 크롤링 해와야 합니다.

인증키를 지정하고, 라이브러리를 import 해 줍니다.

crtfc_key = ‘본인의 인증키 번호’ import requests import pandas as pd import io import zipfile import xml.etree.ElementTree as et import json

이제 고유번호를 가져오면 됩니다.

고유번호 크롤링 방법은 “오픈DART > 개발가이드 > 공시정보 목록> 고유번호”를 보면 확인하실 수 있습니다.

1) 기본 정보를 통해 요청 url을 확인합니다.

출력 포맷이 ZIP file이고, 인코딩이 utf-8이라는 것을 확인할 수 있습니다.

그러므로 zipfile로 불러오고 zipfile을 열어주는 작업이 필요합니다.

또한, utf-8로 인코딩 되어있는 file을 디코딩 해주는 작업도 필요합니다.

2) 요청인자가 무엇인지 확인합니다.

이 경우, API 인증키만 확인하면 됩니다.

3) 응답결과 출력값이 무엇인지 확인합니다

status, message는 오류가 떴을 경우 확인을 위함입니다.

상정기업 고유번호를 제대로 크롤링해오면, 그 기업의 고유번호와, 정식명칭, 종목코드, 최종변경일자가 출력값으로 나오게 됩니다.

이제, 고유번호를 크롤링해오는 함수를 만들어봅시다.

함수 설명

1) 요청 url에 필요한 요청 인자를 지정해줍니다. (dict 형식, params )

2) 출력값과 출력값의 이름을 지정해줍니다(items, item_names)

3) 요청 url을 넣고, request 해줍니다.

4) zipfile로 이를 받고, 열어서 디코딩 해줍니다.

5) 종목코드가 있는 경우만 받아서 이를 데이터프레임으로 만들어주면 함수가 완성됩니다.

def get_corpcode(crtfc_key): “”” OpenDART 기업 고유번호 받아오기 return 값: 주식코드를 가진 업체의 DataFrame “”” params = {‘crtfc_key’:crtfc_key} items = [“corp_code”,”corp_name”,”stock_code”,”modify_date”] item_names = [“고유번호”,”회사명”,”종목코드”,”수정일”] url = “https://opendart.fss.or.kr/api/corpCode.xml” #요청 url res = requests.get(url,params=params) #url 불러오기 zfile = zipfile.ZipFile(io.BytesIO(res.content)) #zip file 받기 fin = zfile.open(zfile.namelist()[0]) #zip file 열고 root = et.fromstring(fin.read().decode(‘utf-8’)) #utf-8 디코딩 data = [] for child in root: if len(child.find(‘stock_code’).text.strip()) > 1: # 종목코드가 있는 경우 data.append([]) #data에 append하라 for item in items: data[-1].append(child.find(item).text) df = pd.DataFrame(data, columns=item_names) return df

이제 함수를 적용해봅시다.

stock_comp = get_corpcode(crtfc_key) stock_comp

(2) 단일회사 상장기업 재무제표 크롤링

1) url request 및 데이터프레임으로 변환하는 함수

재무제표 크롤링에 앞서 json 형식으로 url을 request하고, 이를 dataframe으로 형식으로 바꾸는 함수를 먼저 만들어보겠습니다.

1) url과 요청인자 params를 만들고 이를 request 해줍니다.

2) data를 받았다면 이를 append 해주고 dataframe 형식으로 바꾸어줍니다.

def Frame(url, items, item_names, params): “”” url : json형태로 요청하는 주소 items : 반환되는 데이터들의 key를 가진 리스트 item_names : 데이터프레임을 만들때 컬럼명 리스트 params : url 요청시 필수값으로 들어가는 인자들을 가진 딕셔너리 “”” res = requests.get(url, params) json_data = res.json() json_dict = json.loads(res.text) data = [] if json_dict[‘status’] == “000”: # 오류 없이 정상적으로 데이터가 있다면 for line in json_dict[‘list’]: data.append([]) for itm in items: if itm in line.keys(): data[-1].append(line[itm]) else: data[-1].append(”) df = pd.DataFrame(data, columns=item_names) return df

2) 단일회사 재무제표 크롤링

1) 기본 정보를 통해 요청 url을 확인합니다.

요청할수 있는 형식은 json, xml 두가지 입니다.

json 형식을 사용하도록 하겠습니다.

2) 요청인자가 무엇인지 확인합니다.

API key와 크롤링해온 기업의 고유 번호가 필요합니다.

사업 연도 (2015년 이후) 도 필요합니다.

어떤 보고서를 가져올지 선택해야합니다. (1분기 : 11013, 반기 : 11012, 3분기 :11014, 사업보고서 : 11011)

또한, 재무제표가 개별재무제표(OFS)인지 연결재무제표(CFS)인지 확인해야합니다.

저는 모든 상장기업의 2020년 1분기의 연결재무제표를 가지고 와보도록 하겠습니다.

Q. 연결재무제표란? A. 상장기업과 연결되어 있는 모든 기업을 다 포함한 재무제표

3) 응답결과 출력값이 무엇인지 확인합니다

이는 “개발가이드 > 상장기업재무정보 > 단일회사 전체 재무제표” 를 들어가면 확인 가능합니다

를 들어가면 확인 가능합니다 링크: opendart.fss.or.kr/guide/detail.do?apiGrpCd=DS003&apiId=2019020

함수 설명

1) 출력값(items)와 출력값의 컬럼명(item_names)를 설정해 준 후, url을 request 하고 이를 Frame 함수에 적용합니다.

2) Frame 함수는 json 형식의 url을 요청해 데이터를 가져오고, 이를 dataframe으로 바꾸어 줍니다.

def get_단일회사재무제표(crtfc_key, corp_code, bsns_year, reprt_code, fs_div = “CFS”): items = [“rcept_no”,”reprt_code”,”bsns_year”,”corp_code”,”sj_div”,”sj_nm”, “account_id”,”account_nm”,”account_detail”,”thstrm_nm”, “thstrm_amount”, “thstrm_add_amount”,”frmtrm_nm”,”frmtrm_amount”, “frmtrm_q_nm”,”frmtrm_q_amount”, “frmtrm_add_amount”,”bfefrmtrm_nm”, “bfefrmtrm_amount”,”ord”] item_names = [“접수번호”,”보고서코드”,”사업연도”,”고유번호”,”재무제표구분”, “재무제표명”, “계정ID”,”계정명”,”계정상세”,”당기명”,”당기금액”, “당기누적금액”,”전기명”,”전기금액”,”전기명(분/반기)”, “전기금액(분/반기)”,”전기누적금액”,”전전기명”,”전전기금액”, “계정과목정렬순서”] params = {‘crtfc_key’:crtfc_key, ‘corp_code’:corp_code, ‘bsns_year’:bsns_year, ‘reprt_code’:reprt_code, ‘fs_div’:fs_div} url = “https://opendart.fss.or.kr/api/fnlttSinglAcntAll.json?” df = Frame(url, items, item_names, params) return df

이제 이를 함수에 적용해봅시다.

#1개 해보기 corp_code = stock_comp[‘고유번호’] final_reprt_df = get_fnlttSinglAcntAll(crtfc_key, corp_code[0], 2020, ‘11013’, fs_div = “CFS”) #반복 from tqdm import tqdm for k in tqdm(range(1, len(stock_comp))): reprt_df = get_fnlttSinglAcntAll(crtfc_key, corp_code[k], 2020, ‘11013’, fs_div = “CFS”) final_reprt_df = pd.concat([final_reprt_df, reprt_df], axis = 0)

총 3241개 상장기업의 2020년 1분기 연결재무제표를 크롤링해보았습니다.

이제 이를 각각 재무재표의 형식에 따라 나누어줍니다.

def sep(x, df): df = df[df[‘재무제표구분’] == x] return df final_BS = sep(‘BS’, final_reprt_df) #재무상태표 final_CIS = sep(‘CIS’, final_reprt_df) #포괄손익계산서 final_CF = sep(‘CF’, final_reprt_df) #현금흐름표 final_SCE = sep(‘SCE’, final_reprt_df) #자본변동표 final_IS = sep(‘IS’, final_reprt_df) #손익계산서

*전처리 과정은 다음에 올리도록 하겠습니다.

아래 링크를 참고하여 크롤링 해보았습니다.

https://besixdouze.net/18?category=909673

반응형