예제 소스
한국투자증권 API와 LLM 주식 응용 프로그램 · 파일 19개
fetch_positive_news.py
# -*- coding: utf-8 -*-
"""
네이버 금융 뉴스 페이지에서 최신 기사(제목 + 본문 요약)를 수집해
'호재성' 으로 판정되는 기사와 그에 해당하는 종목코드/종목명을 추출하여
CSV 파일(positive_news.csv) 로 저장합니다.
기준 URL (사용자 지정):
https://finance.naver.com/news/news_list.naver
?mode=LSS3D§ion_id=101§ion_id2=258§ion_id3=401
section_id3 카테고리
401: 시황·전망 (기본)
402: 기업·종목분석 (기본)
403: 해외증시
404: 채권·선물
405: 공시·메모
406: 환율
사용법
$ python fetch_positive_news.py
$ python fetch_positive_news.py --sections 401,402,405 --pages 7
$ python fetch_positive_news.py --min-score 3 --output today_positive.csv
사전 준비
- stock_list.csv 가 같은 폴더에 있어야 합니다 (build_stock_list.py 로 생성).
- pip install requests beautifulsoup4 lxml pandas
"""
from __future__ import annotations
import argparse
import csv
import sys
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Iterable
from urllib.parse import urljoin
import pandas as pd
import requests
from bs4 import BeautifulSoup
NAVER_FINANCE_BASE = "https://finance.naver.com"
NEWS_LIST_URL = NAVER_FINANCE_BASE + "/news/news_list.naver"
DEFAULT_PARAMS = {
"mode": "LSS3D",
"section_id": "101",
"section_id2": "258",
}
SECTION_LABELS: dict[str, str] = {
"401": "시황·전망",
"402": "기업·종목분석",
"403": "해외증시",
"404": "채권·선물",
"405": "공시·메모",
"406": "환율",
}
UA = (
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/******** Safari/537.36"
)
REQ_HEADERS = {
"User-Agent": UA,
"Referer": NAVER_FINANCE_BASE + "/",
"Accept-Language": "ko-KR,ko;q=0.9,en-US;q=0.8,en;q=0.7",
}
# ──────────────────────────────────────────────────────────────────────────────
# 키워드 사전 (호재 / 악재)
# ──────────────────────────────────────────────────────────────────────────────
POSITIVE_KEYWORDS: dict[str, int] = {
"사상 최대": 3, "사상최대": 3, "사상 최고": 3, "사상최고": 3,
"최대 실적": 3, "최대실적": 3, "신기록": 2, "신고가": 3, "52주 신고가": 3,
"흑자전환": 3, "흑자 전환": 3, "흑자": 1,
"호실적": 2, "어닝 서프라이즈": 3, "어닝서프라이즈": 3,
"영업이익 증가": 2, "매출 증가": 2, "매출 신기록": 3, "실적 개선": 2,
"수주": 2, "공급계약": 2, "공급 계약": 2, "대형 계약": 2, "계약 체결": 2,
"납품계약": 2, "납품 계약": 2, "신규 수주": 2, "수출 계약": 2,
"인수": 1, "합병": 1, "M&A": 1, "지분 인수": 2, "지분인수": 2,
"자사주 매입": 2, "자사주매입": 2, "자사주 취득": 2, "자사주취득": 2,
"무상증자": 2, "특별배당": 2, "배당 확대": 2, "배당확대": 2,
"신약": 1, "임상 3상": 2, "임상3상": 2, "임상 성공": 3,
"품목허가": 2, "FDA 승인": 3, "FDA승인": 3, "식약처 승인": 2, "식약처승인": 2,
"특허 등록": 2, "특허등록": 2, "특허 취득": 2,
"신제품 출시": 2, "신제품출시": 2, "출시": 1,
"신사업 진출": 2, "사업 확장": 2, "사업확장": 2,
"전략적 제휴": 2, "MOU 체결": 2, "MOU체결": 2, "협력": 1, "제휴": 1,
"수출 증가": 2, "수출증가": 2, "수출 확대": 2, "수출확대": 2,
"급등": 2, "강세": 1, "상승": 1, "상한가": 3,
"정부 지원": 2, "정부지원": 2, "정책 수혜": 2, "정책수혜": 2,
"점유율 1위": 2, "1위 등극": 2, "1위등극": 2, "흥행": 2, "대박": 2,
"목표주가 상향": 3, "목표주가상향": 3, "매수 추천": 2, "매수추천": 2,
"매수의견": 2, "투자의견 매수": 2, "비중 확대": 2, "비중확대": 2,
"수상": 1, "선정": 1, "인증 획득": 2, "인증획득": 2, "승인": 1,
"최대 주주": 1, "최대주주": 1, "지분 확대": 2, "지분확대": 2,
"수혜": 2, "기대감": 1, "성장": 1, "확대": 1,
}
NEGATIVE_KEYWORDS: dict[str, int] = {
"적자전환": 3, "적자 전환": 3, "적자": 2, "영업손실": 3,
"감소": 1, "부진": 1, "손실": 1, "축소": 1, "둔화": 1,
"급락": 2, "약세": 1, "하락": 1, "추락": 2, "하한가": 3,
"횡령": 3, "배임": 3, "분식회계": 3, "회계부정": 3,
"거래정지": 3, "관리종목": 3, "상장폐지": 3,
"자본잠식": 3, "감사의견 거절": 3, "감사의견거절": 3,
"부적정": 3, "감사의견 한정": 2, "감사의견한정": 2,
"소송": 2, "고소": 2, "고발": 2, "조사": 1, "압수수색": 3,
"경고": 1, "위기": 1, "우려": 1, "리콜": 2,
"감자": 2, "유상증자": 1, "전환사채": 1,
"사망": 2, "사고": 1, "화재": 1, "결함": 2,
"제재": 2, "과징금": 2, "벌금": 1, "회생": 2, "회생절차": 3,
}
# ──────────────────────────────────────────────────────────────────────────────
# 데이터 클래스
# ──────────────────────────────────────────────────────────────────────────────
@dataclass
class Article:
section: str # '시황·전망' 등
date: str # 'YYYY-MM-DD HH:MM'
press: str
title: str
summary: str
url: str
matched_codes: list[str] = field(default_factory=list)
matched_names: list[str] = field(default_factory=list)
positive_score: int = 0
matched_pos: list[str] = field(default_factory=list)
matched_neg: list[str] = field(default_factory=list)
# ──────────────────────────────────────────────────────────────────────────────
# 종목 매칭
# ──────────────────────────────────────────────────────────────────────────────
def load_stock_map(path: Path) -> list[tuple[str, str]]:
"""stock_list.csv → [(종목명, 종목코드), ...] (이름 길이 내림차순)."""
if not path.is_file():
raise FileNotFoundError(
f"{path} 가 없습니다. 먼저 build_stock_list.py 를 실행하세요."
)
df = pd.read_csv(path, dtype={"종목코드": str})
pairs: list[tuple[str, str]] = []
for _, row in df.iterrows():
name = str(row.get("종목명", "")).strip()
code = str(row.get("종목코드", "")).strip()
# 2글자 이하 종목명은 오탐 위험이 매우 커서 제외
if len(name) >= 3 and code:
pairs.append((name, code))
pairs.sort(key=lambda x: len(x[0]), reverse=True)
return pairs
def find_stocks_in_text(text: str, stock_pairs: Iterable[tuple[str, str]]) -> list[tuple[str, str]]:
found: list[tuple[str, str]] = []
seen: set[str] = set()
for name, code in stock_pairs:
if name in text and code not in seen:
found.append((name, code))
seen.add(code)
return found
# ──────────────────────────────────────────────────────────────────────────────
# 호재/악재 점수
# ──────────────────────────────────────────────────────────────────────────────
def score_text(text: str) -> tuple[int, list[str], list[str]]:
pos = [k for k in POSITIVE_KEYWORDS if k in text]
neg = [k for k in NEGATIVE_KEYWORDS if k in text]
score = sum(POSITIVE_KEYWORDS[k] for k in pos) - sum(NEGATIVE_KEYWORDS[k] for k in neg)
return score, pos, neg
# ──────────────────────────────────────────────────────────────────────────────
# 네이버 금융 뉴스 수집
# ──────────────────────────────────────────────────────────────────────────────
def fetch_naver_news_page(section_id3: str, page: int) -> list[Article]:
params = dict(DEFAULT_PARAMS)
params["section_id3"] = section_id3
params["page"] = str(page)
r = requests.get(NEWS_LIST_URL, headers=REQ_HEADERS, params=params, timeout=20)
r.raise_for_status()
r.encoding = r.apparent_encoding or "euc-kr"
soup = BeautifulSoup(r.text, "lxml")
section_label = SECTION_LABELS.get(section_id3, section_id3)
articles: list[Article] = []
# 표준 구조: <dl><dt class="articleSubject"><a>...</a></dt><dd class="articleSummary">요약 + press + wdate</dd></dl>
dls = soup.select("ul.realtimeNewsList dl") or soup.select("div.newsList dl") or soup.select("dl")
for dl in dls:
subj = dl.select_one("dt.articleSubject a") or dl.select_one("dt a")
if not subj:
continue
title = subj.get_text(" ", strip=True)
href = subj.get("href", "")
if not title or not href:
continue
summary_el = dl.select_one("dd.articleSummary") or dl.select_one("dd")
summary = ""
press = ""
date_text = ""
if summary_el:
press_el = summary_el.select_one(".press")
wdate_el = summary_el.select_one(".wdate")
if press_el:
press = press_el.get_text(" ", strip=True)
if wdate_el:
date_text = wdate_el.get_text(" ", strip=True)
# 본문 요약: press / wdate / 썸네일 등 기타 태그를 제거하고 텍스트만 수집
tmp = BeautifulSoup(str(summary_el), "lxml")
for sel in (".press", ".wdate", ".thumb", "img", "a.cluster", ".cluster"):
for el in tmp.select(sel):
el.decompose()
summary = tmp.get_text(" ", strip=True)
articles.append(
Article(
section=section_label,
date=date_text,
press=press,
title=title,
summary=summary,
url=urljoin(NAVER_FINANCE_BASE, href),
)
)
# fallback: dl 구조가 안 잡히면 article_id 링크를 모두 수집
if not articles:
for a in soup.select("a[href*='article_id=']"):
title = a.get_text(strip=True)
href = a.get("href", "")
if not title or len(title) < 6:
continue
articles.append(
Article(
section=section_label,
date="",
press="",
title=title,
summary="",
url=urljoin(NAVER_FINANCE_BASE, href),
)
)
return articles
# ──────────────────────────────────────────────────────────────────────────────
# 파이프라인
# ──────────────────────────────────────────────────────────────────────────────
def collect_articles(sections: list[str], pages: int, delay: float) -> list[Article]:
items: list[Article] = []
for sec in sections:
for p in range(1, pages + 1):
try:
got = fetch_naver_news_page(sec, p)
except Exception as exc:
print(f" · [section={sec} page={p}] 수집 실패: {exc}", file=sys.stderr)
continue
print(f" · [{SECTION_LABELS.get(sec, sec)} page={p}] {len(got)}건")
items.extend(got)
time.sleep(delay)
return _dedupe(items)
def _dedupe(items: list[Article]) -> list[Article]:
seen: set[tuple[str, str]] = set()
out: list[Article] = []
for a in items:
key = (a.title, a.url)
if key in seen:
continue
seen.add(key)
out.append(a)
return out
def annotate(articles: list[Article], stock_pairs: list[tuple[str, str]]) -> list[Article]:
for art in articles:
# 제목 + 요약을 합쳐 분석 (호재 키워드와 종목명이 본문 요약에 들어 있을 수 있음)
text = f"{art.title} {art.summary}"
score, pos, neg = score_text(text)
art.positive_score = score
art.matched_pos = pos
art.matched_neg = neg
for name, code in find_stocks_in_text(text, stock_pairs):
art.matched_names.append(name)
art.matched_codes.append(code)
return articles
def filter_positive(
articles: list[Article], *, min_score: int, require_stock: bool
) -> list[Article]:
out = [
a for a in articles
if a.positive_score >= min_score and (a.matched_codes or not require_stock)
]
return out
def write_csv_by_stock(articles: list[Article], out_path: Path) -> int:
"""종목 단위로 CSV 출력 (한 기사에 여러 종목이면 종목별 행 분할)."""
rows: list[dict[str, object]] = []
for a in articles:
if a.matched_codes:
for name, code in zip(a.matched_names, a.matched_codes):
rows.append({
"종목코드": code,
"종목명": name,
"호재점수": a.positive_score,
"카테고리": a.section,
"날짜": a.date,
"언론사": a.press,
"제목": a.title,
"요약": a.summary,
"매칭 호재키워드": ", ".join(a.matched_pos),
"매칭 악재키워드": ", ".join(a.matched_neg),
"URL": a.url,
})
else:
rows.append({
"종목코드": "",
"종목명": "",
"호재점수": a.positive_score,
"카테고리": a.section,
"날짜": a.date,
"언론사": a.press,
"제목": a.title,
"요약": a.summary,
"매칭 호재키워드": ", ".join(a.matched_pos),
"매칭 악재키워드": ", ".join(a.matched_neg),
"URL": a.url,
})
rows.sort(key=lambda r: (str(r["종목코드"]), -int(r["호재점수"])))
fieldnames = [
"종목코드", "종목명", "호재점수", "카테고리", "날짜",
"언론사", "제목", "요약", "매칭 호재키워드", "매칭 악재키워드", "URL",
]
with out_path.open("w", encoding="utf-8-sig", newline="") as f:
w = csv.DictWriter(f, fieldnames=fieldnames)
w.writeheader()
w.writerows(rows)
return len(rows)
def parse_sections(raw: str) -> list[str]:
out: list[str] = []
for tok in raw.split(","):
tok = tok.strip()
if not tok:
continue
if tok not in SECTION_LABELS:
print(f" · 알 수 없는 section_id3={tok} (계속 진행)", file=sys.stderr)
out.append(tok)
return out
def parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="네이버 금융 호재성 뉴스 수집기")
p.add_argument("--sections", type=str, default="401,402",
help="section_id3 콤마 구분 (기본: 401,402 — 시황·전망 + 기업·종목분석)")
p.add_argument("--pages", type=int, default=7,
help="카테고리별 수집 페이지 수 (기본: 7)")
p.add_argument("--min-score", type=int, default=2,
help="호재로 인정할 최소 점수 (기본: 2)")
p.add_argument("--no-require-stock", action="store_true",
help="종목 매칭이 안 된 기사도 함께 저장")
p.add_argument("--delay", type=float, default=0.5,
help="페이지 요청 간 지연 초 (기본: 0.5)")
p.add_argument("--stock-list", type=str, default="stock_list.csv",
help="종목 리스트 CSV (기본: stock_list.csv)")
p.add_argument("--output", type=str, default="positive_news.csv",
help="출력 CSV (기본: positive_news.csv)")
return p.parse_args()
def main() -> int:
args = parse_args()
here = Path(__file__).resolve().parent
stock_csv = (here / args.stock_list) if not Path(args.stock_list).is_absolute() else Path(args.stock_list)
out_csv = (here / args.output) if not Path(args.output).is_absolute() else Path(args.output)
sections = parse_sections(args.sections)
if not sections:
print("수집할 section 이 없습니다.", file=sys.stderr)
return 1
print(f"[1/4] 종목 리스트 로드: {stock_csv}")
stock_pairs = load_stock_map(stock_csv)
print(f" · 매칭 가능한 종목 수: {len(stock_pairs)}")
print(f"[2/4] 네이버 금융 뉴스 수집 (sections={','.join(sections)}, pages={args.pages})")
articles = collect_articles(sections, args.pages, args.delay)
print(f" · 수집된 총 기사 수(중복제거): {len(articles)}")
print("[3/4] 호재 점수 계산 + 종목 매칭")
articles = annotate(articles, stock_pairs)
positive = filter_positive(
articles,
min_score=args.min_score,
require_stock=not args.no_require_stock,
)
print(f" · 호재 판정(>= {args.min_score}점) 기사: {len(positive)}")
print(f"[4/4] CSV 저장 (종목별 행): {out_csv}")
n_rows = write_csv_by_stock(positive, out_csv)
print(f" ✓ 저장 완료 (총 {n_rows} 행)")
return 0
if __name__ == "__main__":
sys.exit(main())