133 lines
4.8 KiB
Python
133 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
관세청 수출 호재 감지 스나이퍼
|
|
- 네이버 금융 속보에서 '수출/반도체' 관련 뉴스 실시간 감시
|
|
- 호재 종목을 즉시 후보군에 추가
|
|
"""
|
|
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
import time
|
|
import datetime
|
|
import logging
|
|
import re
|
|
from typing import List, Dict, Optional
|
|
|
|
logger = logging.getLogger("ExportSniper")
|
|
|
|
|
|
class ExportSniper:
|
|
"""관세청 수출 호재 감지기"""
|
|
|
|
def __init__(self):
|
|
# 네이버 금융 실시간 속보 URL
|
|
self.target_url = "https://finance.naver.com/news/news_list.naver?mode=LSS2D§ion_id=101§ion_id2=258"
|
|
self.headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
|
|
self.keywords = ['관세청', '수출', '반도체', '잠정', '무역수지', '수출입']
|
|
self.seen_news = set() # 이미 본 뉴스는 중복 출력 방지
|
|
|
|
# 종목명 매칭용 (반도체 관련)
|
|
self.semiconductor_stocks = {
|
|
'삼성전자': '005930',
|
|
'SK하이닉스': '000660',
|
|
'삼성SDI': '006400',
|
|
'LG화학': '051910',
|
|
'LG에너지솔루션': '373220',
|
|
'포스코홀딩스': '005490',
|
|
'포스코케미칼': '003670',
|
|
}
|
|
|
|
def fetch_news(self) -> List[Dict]:
|
|
"""네이버 금융 속보에서 뉴스 가져오기"""
|
|
try:
|
|
res = requests.get(self.target_url, headers=self.headers, timeout=5)
|
|
soup = BeautifulSoup(res.text, 'html.parser')
|
|
|
|
# 뉴스 리스트 파싱
|
|
news_items = soup.select('ul.realtimeNewsList li dl')
|
|
|
|
found_news = []
|
|
|
|
for item in news_items:
|
|
title_tag = item.select_one('a')
|
|
if not title_tag:
|
|
continue
|
|
|
|
title = title_tag.text.strip()
|
|
link = "https://finance.naver.com" + title_tag.get('href', '')
|
|
|
|
# 중복 체크
|
|
if title in self.seen_news:
|
|
continue
|
|
|
|
self.seen_news.add(title)
|
|
|
|
# 키워드 감지
|
|
if any(keyword in title for keyword in self.keywords):
|
|
# 긍정적 키워드 체크 (증가, 급증, 상승 등)
|
|
positive_keywords = ['증가', '급증', '상승', '호조', '기록', '최대', '신고']
|
|
is_positive = any(pk in title for pk in positive_keywords)
|
|
|
|
if is_positive:
|
|
found_news.append({
|
|
'title': title,
|
|
'link': link,
|
|
'timestamp': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
})
|
|
logger.info(f"🔥 [호재 포착] {title}")
|
|
|
|
return found_news
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ 뉴스 크롤링 실패: {e}")
|
|
return []
|
|
|
|
def extract_stocks_from_news(self, news_title: str) -> List[str]:
|
|
"""
|
|
뉴스 제목에서 관련 종목 코드 추출
|
|
|
|
:param news_title: 뉴스 제목
|
|
:return: 종목 코드 리스트
|
|
"""
|
|
stocks = []
|
|
|
|
# 반도체 관련 종목 매칭
|
|
for name, code in self.semiconductor_stocks.items():
|
|
if name in news_title:
|
|
stocks.append(code)
|
|
|
|
# "반도체" 키워드가 있으면 주요 반도체 종목 모두 추가
|
|
if '반도체' in news_title:
|
|
stocks.extend(['005930', '000660']) # 삼성전자, SK하이닉스
|
|
|
|
return list(set(stocks)) # 중복 제거
|
|
|
|
def get_hot_stocks(self) -> List[Dict]:
|
|
"""
|
|
호재 뉴스에서 관련 종목 추출
|
|
|
|
:return: [{'code': '005930', 'name': '삼성전자', 'reason': '뉴스 제목'}, ...]
|
|
"""
|
|
news_list = self.fetch_news()
|
|
hot_stocks = []
|
|
|
|
for news in news_list:
|
|
codes = self.extract_stocks_from_news(news['title'])
|
|
for code in codes:
|
|
# 종목명 찾기
|
|
name = None
|
|
for stock_name, stock_code in self.semiconductor_stocks.items():
|
|
if stock_code == code:
|
|
name = stock_name
|
|
break
|
|
|
|
if name:
|
|
hot_stocks.append({
|
|
'code': code,
|
|
'name': name,
|
|
'reason': news['title'],
|
|
'timestamp': news['timestamp']
|
|
})
|
|
|
|
return hot_stocks
|