krx-news-client

한국 주식시장의 뉴스·공시를 모아 하나의 스키마로 내주는 Python 클라이언트 라이브러리 — DART, 토스증권.

매체마다 HTML 구조도 갱신 주기도 제각각이라, 뉴스를 쓰려는 쪽이 매번 크롤러를 다시 짜게 된다. 그 일을 한 번만 하려고 만들었다. kiwoom-client와 같은 성격의 라이브러리다 — 서버를 띄우지 않고, 호출한 프로세스 안에서 그때그때 매체에 직접 요청해 정규화된 결과를 돌려준다.

설치

Python 3.11 이상.

pip install krx-news-client

빠른 시작

import asyncio
from krx_news_client import TossScraper

async def main():
    async with TossScraper() as scraper:
        articles = await scraper.scrape_news()
        for article in articles[:5]:
            print(article.nation, article.title, article.published_at)

asyncio.run(main())

실행하면 이런 식으로 출력된다:

KR 코스피, 외국인 순매수에 2%대 상승 마감 2025-06-10 15:30:00+09:00
US 엔비디아, 실적 발표 앞두고 강세 2025-06-10 14:05:00+09:00
...

토스 대시보드 피드는 현재 스냅샷이다. 해외(nation="US") 기사가 절반 넘게 섞여 있고, 본문(content)은 비어 있으며, 과거로 넘길 수 없다. 본문과 과거 기사는 따로 가져온다:

from datetime import date
from krx_news_client import TossScraper

async with TossScraper() as scraper:
    detail = await scraper.fetch_article_detail(articles[0].url)  # 본문 전체·감성 라벨·언론사 원문 URL
    history = await scraper.scrape_company_news(                  # 종목별 뉴스, 과거로 페이징
        "005930", since=date(2025, 6, 1), max_pages=20     # date 는 그날 00:00 KST
    )
    items, _ = await scraper.company_news_page("005930", number=1)  # 원본 dict 한 페이지

DART 공시는 API 키가 필요하다:

from krx_news_client import DartScraper

async with DartScraper(api_key="...") as scraper:   # 또는 DartScraper.from_env()
    disclosures = await scraper.scrape_disclosures()  # 기본: 어제~오늘(KST), 전체 시장

날짜 범위·시장을 직접 지정하거나, 여러 키를 순환시켜 일 20,000건 한도를 넘는 대량 백필도 가능하다:

scraper = DartScraper(api_key=["key1", "key2", "key3"])  # 한도 초과 시 다음 키로 자동 전환
disclosures = await scraper.scrape_disclosures(
    bgn_de="20250101", end_de="20250331", corp_cls="Y"  # Y=코스피, K=코스닥, N=코넥스
)

정규화된 Disclosure가 아니라 DART 원본 응답이 필요하면 search_disclosures(한 페이지)/search_disclosures_all(전 페이지)/search_disclosures_range(여러 해·여러 시장)를 쓴다 — 결측을 조용히 넘기지 않고 DartAPIError를 그대로 올리는 엄격 경로라 백필에 적합하다. DART는 기업을 지정하지 않은 검색을 달력 3개월로 제한하는데, search_disclosures_range가 quarter_ranges로 알아서 쪼갠다:

rows = await scraper.search_disclosures_range(bgn_de="20160901", end_de="20260915", corp_cls=["Y", "K"])
docs = await scraper.fetch_document(rows[0]["rcept_no"])  # 공시 원문(document.xml) → DisclosureDocument.text

폴링할 때는 Deduplicator로 처음 본 것만 넘긴다(기사는 id, 공시는 접수번호 기준):

from krx_news_client import Deduplicator

dedup = Deduplicator()
while True:
    for article in dedup.filter_new(await toss.scrape_news()):
        handle(article)
    await asyncio.sleep(180)

같은 스크레이퍼를 asyncio.run으로 여러 번 불러도 된다 — 이벤트 루프가 바뀌면 HTTP 클라이언트를 새로 만든다(0.3.x까지는 두 번째 호출이 조용히 일부 결과를 잃었다).

더 많은 예제는 examples/ 참고.

수집 소스

수집한 기사는 매체와 무관하게 NewsArticle, 공시는 Disclosure 한 벌로 정규화한다. 소스가 늘어도 호출 코드는 그대로다.

토스 사용 시 주의

토스는 비공식 내부 API라 예고 없이 막히거나 형식이 바뀔 수 있다. 형식이 바뀌면 종목별 뉴스 경로는 빈 결과 대신 TossResponseError를 올린다. 종목별 뉴스(/api/v2/news/companies/{code})에서 직접 확인한 함정은 다음과 같다(2026-09-15).

• 코드는 6자리만 받는다. A005930이나 ISIN을 넣으면 토스는 오류 없이 0건을 준다. 그래서 라이브러리가 A 접두어는 벗기고, 그 외 형식이면 ValueError를 낸다.
• 페이지 번호는 1부터 시작한다. number=0이면 토스가 400을 주므로 호출 전에 ValueError를 낸다.
• lastPage를 믿을 수 없다. 페이지가 size보다 몇 건만 모자라도 뒤에 페이지가 남아 있는데 True가 온다. 그래서 scrape_company_news는 빈 페이지가 나올 때까지 넘긴다.
• 순서가 대략적이다. 페이지 안에서도 정렬돼 있지 않고, 이웃 페이지끼리 시각이 하루쯤 겹친다. 그래서 since는 페이지 전체가 그보다 오래됐을 때만 멈추는 조건으로 쓰고, 결과는 id로 중복을 걷어낸 뒤 최신순으로 정렬해 돌려준다.
• 시장 전체 기사가 섞인다. "코스피 마감" 같은 기사는 stockCodes가 없으므로 tickers가 비어 있다.

요청 간격은 BaseScraper의 0.5~1.5초 무작위 지연을 토스에도 그대로 쓴다. 비공식 API라 보수적으로 두는 편이 차단 위험이 낮기 때문이다. 속도가 필요하면 scraper.min_delay와 scraper.max_delay로 조절한다. DART는 공식 API이고 일한도만 제약이라 지연이 0이다.

응답 필드

NewsArticle

Disclosure는 id·source·title·url·published_at·collected_at은 같고, category·content·summary·tickers·author는 없는 대신 다음이 있다.

published_at은 접수일의 00:00 KST다 — DART 목록 API에는 접수 시각이 없다.

구조

flowchart LR
    Caller["호출 프로세스\n(예: quant-airflow)"]

    subgraph Client["krx-news-client"]
        direction TB
        Toss["TossScraper\n.scrape_news()\n.fetch_article_detail()\n.scrape_company_news()"]
        Dart["DartScraper\n.scrape_disclosures()\n.search_disclosures_range()\n.fetch_document()"]
        Base["BaseScraper\nUA 순환 · 간격 조절(throttle)\n재시도 · 429 백오프"]
        Schema["schemas.py\nNewsArticle · Disclosure\nDisclosureDocument"]

        Toss --> Base
        Dart --> Base
        Toss -->|_make_article| Schema
        Dart -->|_make_disclosure| Schema
    end

    TossAPI[("토스증권\nwts-info-api\n(비공식 내부 API)")]
    DartAPI[("DART\nopendart.fss.or.kr\n(공식 Open API)")]

    Caller -->|await scraper.scrape_news()\n / scrape_disclosures()| Client
    Base -->|POST 대시보드 피드\nGET 기사 상세·종목별 뉴스| TossAPI
    Base -->|GET list.json\n/ document.xml| DartAPI
    Client -->|list[NewsArticle]\n / list[Disclosure]| Caller

BaseScraper가 재시도(429·5xx만)·429 백오프 등 공통 처리를 맡고, 각 스크레이퍼는 매체 응답을 파싱해 정규화된 스키마로 변환하는 일만 한다. DART는 키가 여럿이면 한도 초과 시 다음 키로 자동 전환하고, 전부 소진되면 DartQuotaExceededError를 던져 "공시 없음"(status=013)과 "한도 초과"(status=020)를 구분할 수 있게 한다. 그 외 오류 상태는 DartAPIError — search_disclosures*는 이를 그대로 올리고(엄격), scrape_disclosures는 페이지 단위로 로그만 남기고 계속한다(관용적, 실시간 폴링용).

저장이 필요하면 호출하는 쪽에서 알아서 한다 (예: quant-airflow가 이 라이브러리로 수집해 TimescaleDB에 적재).

라이선스

Apache License 2.0 — 전문은 LICENSE 참조.

⭐ 도움이 되셨다면

이 프로젝트가 유용했다면 우측 상단 ⭐ Star 를 눌러주세요. 검색·추천 노출이 올라가 더 많은 분들이 찾을 수 있습니다.

• 🐛 버그·질문 → Issues
• 📈 업데이트 소식 → 팔로우 @younghwan91

관련 프로젝트 — 오픈소스 퀀트 스택

한국·미국 주식과 암호화폐를 아우르는 오픈소스 스택입니다. 각 저장소는 독립적으로 쓸 수 있습니다.

만든 사람

채영환 (Younghwan Chae) · GitHub @younghwan91 · LinkedIn

전체 오픈소스 퀀트 스택은 프로필에서 한눈에 볼 수 있습니다.