kyu/trading
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 핫종목 렌즈 + 뉴스 파서 견고화 + 텔레그램 알림 2회 축소

Browse Source 
- news-collector: EXAONE JSON 파서 견고화(max_tokens 800 + 관대 파싱 → 파싱실패율 9.3%→3.3%),
  평일 18:30 종목뉴스 스위프(중형주 신선도 사각지대 해소), /process/raw reprocess 플래그,
  파싱실패→Gemini 폴백 라우팅(일일캡 1→50 상수화)
- score-engine: /hot 모멘텀 렌즈(뉴스+거래량, 백테스트로 가격 제외) + /hot/backtest,
  텔레그램 정기 브리핑 2회로 축소(중복 send_briefing 자동스케줄 제거)
- dashboard-api: /api/hot 프록시 + 🔥 지금뜨는 탭
- telegram-bot: /hot 명령 + 한글(뜨는/핫/급등) 라우팅

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
kyu
2026-06-02 01:22:23 +09:00
parent a81feb48d8
commit 8b200e6cc6
5 changed files with 3989 additions and 367 deletions
Download Patch File Download Diff File Expand all files Collapse all files
news-collector/main.py
+458 -93
View File
@@ -8,13 +8,15 @@
import asyncio, hashlib, json, os, re, random, time
import xml.etree.ElementTree as ET
from datetime import datetime, timedelta
from typing import Optional
from typing import Optional, Literal
import asyncpg, httpx, redis.asyncio as aioredis, structlog
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from bs4 import BeautifulSoup
from fastapi import FastAPI, Query
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from sentiment_rules import keyword_rule, hallucination_match_ratio
structlog.configure(processors=[
structlog.processors.TimeStamper(fmt="iso"),
@@ -33,6 +35,94 @@ PG_PASS = os.getenv("POSTGRES_PASSWORD", "7895123")
BAREUN_URL = os.getenv("BAREUN_API_URL", "http://bareunaapi:5757")
OLLAMA_URL = os.getenv("OLLAMA_URL", "http://ollama:11434")
QDRANT_URL = os.getenv("QDRANT_URL", "http://qdrant:6333")
GEMINI_API_KEY = os.getenv("GEMINI_API_KEY", "")
GEMINI_MODEL = os.getenv("GEMINI_MODEL", "gemini-2.5-pro")
GEMINI_DAILY_LIMIT = int(os.getenv("GEMINI_DAILY_LIMIT", "50")) # 파싱실패 폴백 포함 일일 한도(비용통제)
class _GeminiNewsClf(BaseModel):
sentiment: Literal["호재", "악재", "중립"]
intensity: int = Field(ge=1, le=5)
catalyst: Literal["실적", "수주", "배당", "리스크", "모멘텀", "기타"]
reason: str
confidence: float = Field(ge=0.0, le=1.0)
async def _gemini_news_classify(title: str, content: str, exaone: dict,
rule: dict) -> tuple[dict, int]:
"""Gemini 짧은 프롬프트로 뉴스 sentiment 재판정 (애매한 건만).
response_schema로 JSON 강제, max_output_tokens=1500.
returns: (parsed_dict_or_empty, cost_krw)
"""
if not GEMINI_API_KEY:
return {}, 0
# === 글로벌 일일 1회 한도 (score-engine과 공유) ===
try:
async with pg_pool.acquire() as conn:
cnt = await conn.fetchval(
"SELECT COUNT(*) FROM gemini_call_log WHERE called_at::date = CURRENT_DATE")
if cnt and int(cnt) >= GEMINI_DAILY_LIMIT:
logger.warning("gemini.daily_limit", source="news_collector", today_count=int(cnt))
return {}, 0
except Exception as e:
logger.warning("gemini.limit_check_err", error=str(e))
try:
from google import genai
from google.genai import types
gclient = genai.Client(api_key=GEMINI_API_KEY)
system = (
"한국 주식 뉴스 한 건의 호재/악재/중립 여부를 판단합니다. "
"본문에 명시된 사실만 근거로 하고, 본문에 없는 정보는 절대 지어내지 마세요. "
"키워드 카운트와 EXAONE 1차 판단이 충돌하거나 신뢰도가 낮은 경우 호출됩니다."
)
prompt = (
f"[제목]\n{title}\n\n"
f"[본문(요약)]\n{(content or '')[:600]}\n\n"
f"[EXAONE 1차] {exaone.get('sentiment')}/"
f"강도{exaone.get('intensity',0)}/conf{float(exaone.get('confidence',0) or 0):.2f}\n"
f"[키워드 카운트] 호재 {rule['pos']} / 악재 {rule['neg']}"
)
config = types.GenerateContentConfig(
system_instruction=system,
temperature=0.1,
response_mime_type="application/json",
response_schema=_GeminiNewsClf,
max_output_tokens=1500,
)
resp = await asyncio.to_thread(
gclient.models.generate_content,
model=GEMINI_MODEL,
contents=prompt,
config=config,
)
text = getattr(resp, "text", "") or ""
parsed = {}
try:
parsed = json.loads(text)
except Exception:
pass
usage = getattr(resp, "usage_metadata", None)
cost_krw = 0
if usage:
in_tok = getattr(usage, "prompt_token_count", 0) or 0
total_tok = getattr(usage, "total_token_count", 0) or 0
cand_tok = getattr(usage, "candidates_token_count", 0) or 0
# 출력 전체(candidates + thinking) = total - prompt
out_tok = max(total_tok - in_tok, cand_tok)
cost_usd = (in_tok / 1_000_000 * 1.25) + (out_tok / 1_000_000 * 10.0)
cost_krw = int(cost_usd * 1400)
# 일일 한도 카운터 기록
try:
async with pg_pool.acquire() as conn:
await conn.execute(
"INSERT INTO gemini_call_log (source, cost_krw) VALUES ($1, $2)",
"news_collector", cost_krw)
except Exception:
pass
return parsed, cost_krw
except Exception as e:
logger.warning("news.gemini_err", error=str(e))
return {}, 0
HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
@@ -42,12 +132,76 @@ scheduler = AsyncIOScheduler(timezone="Asia/Seoul")
class S:
collected = 0; processed = 0; duplicates = 0; errors = 0
noise = 0; off_topic = 0
noise = 0; off_topic = 0; stale = 0
last_run = ""; running = False
stats = S()
def nhash(title, url=""): return hashlib.sha256(f"{title.strip()}{url.strip()}".encode()).hexdigest()[:16]
def _loads_lenient(s: str):
"""EXAONE 출력에서 JSON 객체를 관대하게 파싱. 실패 시 None.
마크다운 펜스 제거 + {..} 추출 + 트레일링 콤마 보정."""
if not s:
return None
s = s.replace("```json", "").replace("```", "").strip()
if not s.startswith("{"):
sx, ex = s.find("{"), s.rfind("}")
if sx != -1 and ex > sx:
s = s[sx:ex + 1]
try:
return json.loads(s)
except Exception:
pass
s2 = re.sub(r",\s*([}\]])", r"\1", s) # 트레일링 콤마 제거
try:
return json.loads(s2)
except Exception:
return None
# ── 출처 신뢰도 (정적 기본값 — score-engine 사후검증이 news_source_credibility 자동 갱신) ──
# 0.0~1.0. 메이저 통신사·전문지일수록 높음. score-engine이 가격반응 기반으로 자동 보정.
SOURCE_CREDIBILITY = {
"DART공시": 1.00,
"연합뉴스": 0.95, "한국경제": 0.92, "매일경제": 0.90, "조선비즈": 0.90,
"이데일리": 0.88, "머니투데이": 0.85, "파이낸셜뉴스": 0.85,
"서울경제": 0.85, "SBS Biz": 0.85, "더벨": 0.82,
"비즈니스포스트": 0.80, "전자신문": 0.80, "인포스탁": 0.78,
"헤럴드경제": 0.78, "디지털타임스": 0.75, "뉴스1": 0.78,
"뉴시스": 0.75, "이투데이": 0.72, "아시아경제": 0.72,
"네이버금융": 0.75, "뉴스핌": 0.70, "글로벌이코노믹": 0.65,
"스탁데일리": 0.60, "세계일보": 0.65,
}
async def get_source_credibility(source: str) -> float:
"""static dict + DB 사후 학습 결과 융합. DB에 sample≥20이면 그 값으로 덮어씀."""
base = SOURCE_CREDIBILITY.get(source, 0.50)
if not pg_pool: return base
try:
row = await pg_pool.fetchrow(
"SELECT credibility, sample_size FROM news_source_credibility WHERE source=$1",
source)
if row and (row["sample_size"] or 0) >= 20:
return float(row["credibility"])
except Exception:
pass
return base
# ── 제목 강도 (제목이 본문 대비 얼마나 강한 신호인지) ──
_STRONG_TITLE_WORDS = (
"폭락","폭등","급락","급등","충격","역대급","쇼크","서프라이즈","어닝",
"사상최대","사상최저","신고가","신저가","상한가","하한가","적자전환","흑자전환",
"어닝쇼크","어닝서프라이즈","대규모","돌파","무너","급증","급감",
)
def measure_title_strength(title: str) -> float:
"""0.3~1.5. 제목 강조어 ↑ 가중, 추측성("?"·"…")은 감점."""
if not title: return 0.5
s = 1.0
s += 0.10 * sum(1 for w in _STRONG_TITLE_WORDS if w in title)
if "!" in title: s += 0.05
if title.endswith("?") or "?" in title[-3:]: s -= 0.15
if "…" in title or "..." in title: s -= 0.05
return max(0.3, min(1.5, s))
def normalize_title(title: str) -> str:
"""[속보][단독](종합) 등 제거 후 특수문자·공백 제거 → 유사 제목 중복 감지용"""
t = re.sub(r'[\[\(【〔][^\]\)】〕]{0,10}[\]\)】〕]', '', title)
@@ -105,7 +259,6 @@ RSS_SOURCES = [
("전자신문", "https://www.etnews.com/news/latest_news.xml"),
("디지털타임스", "https://www.dt.co.kr/rss/rss_economy.html"),
("더벨", "https://www.thebell.co.kr/free/content/xmlService.asp"),
("스탁데일리", "https://www.stockdaily.kr/rss/rss.xml"),
("세계일보", "https://www.segye.com/RSS/economyRss.xml"),
("SBS Biz", "https://news.sbs.co.kr/news/SectionRssFeed.do?sectionId=EC"),
]
@@ -273,12 +426,26 @@ async def _corp_name(code: str) -> str:
return nm
async def pipeline(item, client):
async def pipeline(item, client, reprocess: bool = False):
try:
# 0. 비주식 카테고리 prefix 차단 (바른API 호출 전)
if _is_noise_title(item.get("title", "")):
return "noise"
# 0.5. published_at이 30일 이상 과거면 차단.
# 일부 RSS 사이트가 오래된 기사를 재노출해서 점수 왜곡 유발 → LLM 호출 전에 컷.
# 명시적 백필 엔드포인트(/collect/historical-raw)는 별도 경로라 영향 없음.
pub_str = item.get("published_at") or ""
if pub_str and not item.get("allow_stale"):
try:
pub_dt = datetime.fromisoformat(pub_str.replace("Z", "+00:00"))
# tz-aware 비교를 위해 둘 다 aware로 맞춤
now_dt = datetime.now(pub_dt.tzinfo) if pub_dt.tzinfo else datetime.now()
if (now_dt - pub_dt).days > 30:
return "stale"
except Exception:
pass # 파싱 실패는 정상 진행 (parse_rss_date가 fallback으로 NOW 채움)
# 1. 바른API
br = await client.post(f"{BAREUN_URL}/analyze", json={
"title": item["title"], "content": item.get("content",""),
@@ -312,9 +479,36 @@ async def pipeline(item, client):
sr = await client.post(f"{QDRANT_URL}/collections/news_vectors/points/search",
json={"vector":emb,"limit":6,"score_threshold":0.80,"with_payload":True}, timeout=15)
hits = sr.json().get("result",[])
if any(h["score"]>=0.92 for h in hits): return "sim_dup" # ≥0.99 근접중복도 차단
if not reprocess and any(h["score"]>=0.92 for h in hits): return "sim_dup" # ≥0.99 근접중복도 차단
except: hits = []
# 3.1. 사건 클러스터링: 유사도 0.85~0.92 안에 있는 hit이 있으면 그 클러스터로 합류.
# 첫 뉴스만 풀 가중, 후속은 score-engine에서 감쇠 (is_event_seed=True 인 것만 큰 신호로).
event_cluster_id = ""
is_event_seed = True
try:
for h in hits:
if 0.85 <= h.get("score", 0) < 0.92:
pl = h.get("payload", {}) or {}
cid = pl.get("event_cluster_id") or pl.get("hash") or ""
if cid:
event_cluster_id = cid
is_event_seed = False
try:
await pg_pool.execute("""
UPDATE news_event_cluster
SET last_seen_at=NOW(), member_count=member_count+1
WHERE cluster_id=$1
""", cid)
except Exception: pass
break
if not event_cluster_id:
event_cluster_id = hashlib.sha256(
normalize_title(data["title"]).encode()).hexdigest()[:32]
except Exception:
event_cluster_id = ""
is_event_seed = True
# 3.5. RAG 컨텍스트: 유사 과거뉴스 + 종목 재무·추세·점수 (버핏 판단 근거 주입)
ctx = []
rel = [h for h in hits if 0.80 <= h.get("score", 0) < 0.92][:4]
@@ -373,94 +567,232 @@ async def pipeline(item, client):
logger.debug("rag.ctx_err", code=focus["code"], error=str(e))
context_block = "\n".join(ctx)
# 4. Ollama EXAONE 분석 (버핏 관점 강화 프롬프트)
stocks_str = ", ".join([f'{s["name"]}({s["code"]})' for s in data["stocks"][:5]])
source = item.get("source", "")
content_preview = (data.get("filtered_text") or item.get("content") or "")[:400]
system_prompt = (
"당신은 워렌 버핏 스타일의 한국 주식 가치투자 애널리스트입니다.\n"
"뉴스 한 건을 읽고 기업 본질가치 관점에서 영향을 평가합니다.\n"
"판단 우선순위: 기업 본질가치(실적·FCF·경쟁우위) > 재무 리스크 > 단기 수급·테마\n\n"
"[sentiment] 호재 / 악재 / 중립 — 본질가치 또는 주가에 미치는 방향\n\n"
"[intensity] 1~5 정수\n"
" 5=상장폐지·대규모 횡령 등 기업 존폐\n"
" 4=연간실적 20% 이상 영향(대형 수주·어닝쇼크 등)\n"
" 3=분기실적에 유의미한 영향\n"
" 2=영업환경·업황 변화(직접 실적 수치 영향은 제한적)\n"
" 1=단순 정보성·정기공시·소폭 코멘트\n\n"
"[catalyst] 반드시 아래 6개 중 정확히 하나만 출력 (다른 단어·복합어·영문·한자 절대 금지):\n"
' "실적" = 영업이익·매출·순이익 증감, 흑자/적자 전환, 어닝 서프라이즈/쇼크, 실적 가이던스\n'
' "수주" = 신규 수주·공급계약·납품, M&A, 신사업·증설·신제품 등 성장 동인\n'
' "배당" = 배당 확대/축소, 자사주 매입·소각, 주주환원 정책\n'
' "리스크" = 횡령·배임·소송·감사의견거절·부채급증·유상증자(희석)·경영진 돌발사임·규제/제재·상폐위험\n'
' "모멘텀" = 본질가치 변화 없는 가격 동인(외국인/기관 수급, 목표주가 조정, 테마·정책 기대)\n'
' "기타" = 위 어디에도 명확히 속하지 않는 정보성 뉴스\n'
" 호재/악재면 catalyst를 '기타'로 두지 말고 실적·수주·배당·리스크·모멘텀 중 가장 가까운 것을 고르세요.\n\n"
"[참고] 블록은 보조 자료일 뿐 — 본 기사 내용으로 판단하고 과거 감성 흐름에 휩쓸리지 마세요.\n"
"동일 내용이 [참고]의 유사 과거뉴스에 반복되면 신규성이 낮으니 intensity를 보수적으로.\n\n"
"반드시 스키마에 맞는 유효한 JSON 객체 하나만 출력. 마크다운·설명문 금지."
)
user_prompt = (
f"[출처] {source}\n"
f"[제목] {data['title'][:200]}\n"
f"[내용] {content_preview}\n"
f"[관련종목] {stocks_str or '없음'}\n"
+ (f"\n[참고]\n{context_block}\n" if context_block else "")
+ "\ninvestment_action 규칙: 호재+intensity≥3→매수관심, 악재+intensity≥3→매도관심, 그 외→관망\n"
"primary_stock은 6자리 숫자코드(시장 전체 뉴스면 빈 문자열), "
"affected_stocks는 영향받는 다른 종목코드 배열, reason은 한 문장 핵심 근거.\n\n"
"JSON 스키마:\n"
'{"sentiment":"호재|악재|중립","intensity":1~5,"primary_stock":"005930",'
'"affected_stocks":["000660"],"reason":"핵심 근거 한 문장",'
'"investment_action":"매수관심|매도관심|관망",'
'"catalyst":"실적|수주|배당|리스크|모멘텀|기타"}\n'
"예시: "
'{"sentiment":"호재","intensity":4,"primary_stock":"005930",'
'"affected_stocks":["000660"],"reason":"HBM 대형 수주 확정으로 연간 반도체 실적 큰 폭 개선 기대",'
'"investment_action":"매수관심","catalyst":"수주"}'
)
# EXAONE 호출 + JSON 파싱 (실패 시 1회 재시도 → ~15% 유실 감소)
a = None
raw_prev = ""
for attempt in range(2):
msgs = [{"role":"system","content": system_prompt},
{"role":"user","content": user_prompt}]
if attempt == 1:
msgs += [{"role":"assistant","content": raw_prev[:600]},
{"role":"user","content":
"위 응답이 유효한 JSON이 아닙니다. 설명·마크다운 없이 "
"스키마에 맞는 JSON 객체 하나만 출력하세요."}]
# 3.9. 키워드 룰 1차 판정 — 명확한 호재/악재면 LLM 스킵 (비용 0)
rule_text = (data.get("title", "") or "") + " " + (data.get("filtered_text", "") or "")
rule = keyword_rule(rule_text)
if rule["sentiment"] is not None:
# 키워드 룰로 sentiment 확정 — LLM 스킵 (비용 0)
primary = (data["stocks"][0]["code"] if data["stocks"] else "")
affected = [s["code"] for s in data["stocks"][1:5]]
investment = ("매수관심" if rule["sentiment"] == "호재" and rule["intensity"] >= 3
else "매도관심" if rule["sentiment"] == "악재" and rule["intensity"] >= 3
else "관망")
hits = rule["pos_hits"] if rule["sentiment"] == "호재" else rule["neg_hits"]
stock_impacts = {primary: 1.0} if primary else {}
for c in affected:
stock_impacts[c] = 0.5
a = {
"sentiment": rule["sentiment"], "intensity": rule["intensity"],
"primary_stock": primary, "affected_stocks": affected,
"reason": f"키워드 룰 매칭: {'/'.join(hits)}"[:500],
"investment_action": investment,
"catalyst": rule["catalyst"],
"time_horizon": "단기", "impact_scope": "종목",
"confidence": 1.0, "stock_impacts": stock_impacts,
"sector_hint": "",
}
llm_conf = 1.0
logger.info("news.path", path="rule", sentiment=rule["sentiment"],
pos=rule["pos"], neg=rule["neg"], hits=hits[:3])
else:
# 4. Ollama EXAONE 분석 (버핏 관점 + 라벨 세분화) — 키워드 룰 애매 케이스
stocks_str = ", ".join([f'{s["name"]}({s["code"]})' for s in data["stocks"][:5]])
source = item.get("source", "")
content_preview = (data.get("filtered_text") or item.get("content") or "")[:400]
system_prompt = (
"당신은 워렌 버핏 스타일의 한국 주식 가치투자 애널리스트입니다.\n"
"뉴스 한 건을 읽고 기업 본질가치 관점에서 영향을 평가합니다.\n"
"판단 우선순위: 기업 본질가치(실적·FCF·경쟁우위) > 재무 리스크 > 단기 수급·테마\n\n"
"[sentiment] 호재 / 악재 / 중립 — 본질가치 또는 주가에 미치는 방향\n\n"
"[intensity] 1~5 정수\n"
" 5=상장폐지·대규모 횡령 등 기업 존폐\n"
" 4=연간실적 20% 이상 영향(대형 수주·어닝쇼크 등)\n"
" 3=분기실적에 유의미한 영향\n"
" 2=영업환경·업황 변화(직접 실적 수치 영향은 제한적)\n"
" 1=단순 정보성·정기공시·소폭 코멘트\n\n"
"[catalyst] 반드시 아래 6개 중 정확히 하나만 출력 (다른 단어·복합어·영문·한자 절대 금지):\n"
' "실적" = 영업이익·매출·순이익 증감, 흑자/적자 전환, 어닝 서프라이즈/쇼크, 실적 가이던스\n'
' "수주" = 신규 수주·공급계약·납품, M&A, 신사업·증설·신제품 등 성장 동인\n'
' "배당" = 배당 확대/축소, 자사주 매입·소각, 주주환원 정책\n'
' "리스크" = 횡령·배임·소송·감사의견거절·부채급증·유상증자(희석)·경영진 돌발사임·규제/제재·상폐위험\n'
' "모멘텀" = 본질가치 변화 없는 가격 동인(외국인/기관 수급, 목표주가 조정, 테마·정책 기대)\n'
' "기타" = 위 어디에도 명확히 속하지 않는 정보성 뉴스\n'
" 호재/악재면 catalyst를 '기타'로 두지 말고 실적·수주·배당·리스크·모멘텀 중 가장 가까운 것을 고르세요.\n\n"
"[time_horizon] 영향이 발현되는 시간프레임\n"
' "즉시" = 당일~3일 내 주가 반응 (수주공시·실적 발표·대형 사건)\n'
' "단기" = 1~4주 (분기실적·수급 변화·일시적 모멘텀)\n'
' "중기" = 1~6개월 (업황·정책 변화·신제품 출시 효과)\n'
' "장기" = 6개월 이상 (구조적 변화·신사업 본격화·체질개선)\n\n'
"[impact_scope] 영향 받는 대상 범위\n"
' "종목" = 특정 종목(들)에만 영향\n'
' "섹터" = 동일 업종/테마 전체 영향 (예: 반도체 업황·2차전지 정책)\n'
' "시장" = 코스피/코스닥 전체 또는 매크로 (금리·환율·지정학)\n\n'
"[stock_impacts] {종목코드: 영향가중치 0~1} 매핑. primary_stock=1.0, 보조 종목은 0.3~0.8.\n"
" affected_stocks와 일치하는 모든 종목에 가중치를 매김. 단일 종목이면 {primary:1.0}.\n\n"
"[sector_hint] 관련 업종 한 단어 (예: 반도체, 2차전지, 바이오, 조선, 방산, 자동차, 금융,\n"
" 통신, 화학, 철강, 건설, 유통, AI, 로봇, 게임, 미디어, 에너지, 정유, 기타).\n\n"
"[confidence] 0.0~1.0 — 본인 판단의 신뢰도. 사실/숫자가 명확하면 0.8+, 추측·소문이면 0.3-0.5,\n"
" 맥락 부족·해석 모호하면 0.5 이하. 정직하게 평가하세요.\n\n"
"[참고] 블록은 보조 자료일 뿐 — 본 기사 내용으로 판단하고 과거 감성 흐름에 휩쓸리지 마세요.\n"
"동일 내용이 [참고]의 유사 과거뉴스에 반복되면 신규성이 낮으니 intensity를 보수적으로.\n\n"
"반드시 스키마에 맞는 유효한 JSON 객체 하나만 출력. 마크다운·설명문 금지."
)
user_prompt = (
f"[출처] {source}\n"
f"[제목] {data['title'][:200]}\n"
f"[내용] {content_preview}\n"
f"[관련종목] {stocks_str or '없음'}\n"
+ (f"\n[참고]\n{context_block}\n" if context_block else "")
+ "\ninvestment_action 규칙: 호재+intensity≥3→매수관심, 악재+intensity≥3→매도관심, 그 외→관망\n"
"primary_stock은 6자리 숫자코드(시장 전체 뉴스면 빈 문자열), "
"affected_stocks는 영향받는 다른 종목코드 배열.\n\n"
"JSON 스키마:\n"
'{"sentiment":"호재|악재|중립","intensity":1~5,"primary_stock":"005930",'
'"affected_stocks":["000660"],"stock_impacts":{"005930":1.0,"000660":0.5},'
'"reason":"핵심 근거 한 문장",'
'"investment_action":"매수관심|매도관심|관망",'
'"catalyst":"실적|수주|배당|리스크|모멘텀|기타",'
'"time_horizon":"즉시|단기|중기|장기","impact_scope":"종목|섹터|시장",'
'"sector_hint":"반도체|2차전지|바이오|조선|방산|자동차|금융|통신|화학|철강|건설|유통|AI|로봇|게임|미디어|에너지|정유|기타",'
'"confidence":0.85}\n'
"예시: "
'{"sentiment":"호재","intensity":4,"primary_stock":"005930",'
'"affected_stocks":["000660"],"stock_impacts":{"005930":1.0,"000660":0.6},'
'"reason":"HBM 대형 수주 확정으로 연간 반도체 실적 큰 폭 개선 기대",'
'"investment_action":"매수관심","catalyst":"수주",'
'"time_horizon":"중기","impact_scope":"섹터",'
'"sector_hint":"반도체","confidence":0.85}'
)
# EXAONE 호출 + JSON 파싱 (1차 정상, 2차 파싱실패/저신뢰 재시도)
a = None
raw_prev = ""
for attempt in range(2):
msgs = [{"role":"system","content": system_prompt},
{"role":"user","content": user_prompt}]
if attempt == 1:
hint = ("위 응답이 유효한 JSON이 아니거나 신뢰도가 낮습니다. "
"설명·마크다운 없이 스키마에 맞는 JSON 객체 하나만 출력하고, "
"확실하지 않으면 sentiment='중립', intensity=1로 보수적으로 표기하세요.")
msgs += [{"role":"assistant","content": raw_prev[:600]},
{"role":"user","content": hint}]
try:
vr = await client.post(f"{OLLAMA_URL}/v1/chat/completions", json={
"model":"exaone3.5:7.8b","messages": msgs,
"max_tokens":800,"temperature":0.0 if attempt else 0.05},
timeout=120)
raw_prev = vr.json()["choices"][0]["message"]["content"]
cand = _loads_lenient(raw_prev)
if cand is None:
continue
conf = float(cand.get("confidence", 0.5) or 0.5)
if attempt == 0 and conf < 0.40 and cand.get("sentiment") in ("호재","악재") \
and int(cand.get("intensity", 0) or 0) >= 3:
a = cand
continue
a = cand
break
except Exception:
continue
if a is None:
a = {"sentiment":"중립","intensity":0,"primary_stock":"","affected_stocks":[],
"reason":"파싱실패","investment_action":"관망","confidence":0.0}
# enum 강제 / confidence clamp / stock_impacts 정합
if a.get("catalyst") not in ("실적","수주","배당","리스크","모멘텀","기타"):
a["catalyst"] = "기타"
if a.get("time_horizon") not in ("즉시","단기","중기","장기"):
a["time_horizon"] = "단기"
if a.get("impact_scope") not in ("종목","섹터","시장"):
a["impact_scope"] = "종목"
try:
vr = await client.post(f"{OLLAMA_URL}/v1/chat/completions", json={
"model":"exaone3.5:7.8b","messages": msgs,
"max_tokens":400,"temperature":0.0 if attempt else 0.05},
timeout=120)
raw_prev = vr.json()["choices"][0]["message"]["content"]
c = raw_prev.replace("```json","").replace("```","").strip()
if not c.startswith("{"): # 앞뒤 설명 제거
s, e = c.find("{"), c.rfind("}")
if s != -1 and e > s: c = c[s:e+1]
a = json.loads(c)
break
llm_conf = max(0.0, min(1.0, float(a.get("confidence", 0.5) or 0.5)))
except Exception:
continue
if a is None:
a = {"sentiment":"중립","intensity":0,"primary_stock":"","affected_stocks":[],"reason":"파싱실패","investment_action":"관망"}
llm_conf = 0.5
if a.get("reason") == "파싱실패":
llm_conf = 0.0
si = a.get("stock_impacts")
if not isinstance(si, dict): si = {}
ps = a.get("primary_stock") or ""
if ps and ps not in si: si[ps] = 1.0
for c in (a.get("affected_stocks") or []):
if c and c not in si: si[c] = 0.5
si = {k: max(0.0, min(1.0, float(v or 0)))
for k, v in si.items() if k and isinstance(k, str) and len(k) <= 10}
a["stock_impacts"] = si
# catalyst 6개 enum 강제 (score-engine CATALYST_WEIGHTS 정합 — 일탈값은 '기타')
if a.get("catalyst") not in ("실적","수주","배당","리스크","모멘텀","기타"):
a["catalyst"] = "기타"
# === 환각 / 키워드 충돌 검증 → 필요 시 Gemini fallback ===
# 임계값을 보수적으로 (Gemini 과호출 방지 — Pro thinking 비용 큼)
halluc = hallucination_match_ratio(a.get("reason", ""), rule_text)
conflict = ((rule["pos"] >= 2 and a.get("sentiment") == "악재") or
(rule["neg"] >= 2 and a.get("sentiment") == "호재"))
sent_strong = (a.get("sentiment") in ("호재", "악재")
and int(a.get("intensity") or 0) >= 4) # 4→매우 강한 시그널만
parse_failed = (a.get("reason") == "파싱실패")
need_gem = (parse_failed # EXAONE 파싱실패 → Gemini 재판정(중립 방치 대신)
or (sent_strong and llm_conf < 0.4) # 0.6→0.4 (정말 저신뢰만)
or halluc < 0.25 # 0.5→0.25 (명백한 환각만)
or conflict)
if need_gem and GEMINI_API_KEY:
gem, gem_cost = await _gemini_news_classify(
data.get("title", ""), content_preview,
{"sentiment": a.get("sentiment"),
"intensity": a.get("intensity"),
"confidence": llm_conf}, rule)
if gem:
a["sentiment"] = gem.get("sentiment", a["sentiment"])
try:
a["intensity"] = int(gem.get("intensity", a["intensity"]) or a["intensity"])
except Exception:
pass
if gem.get("catalyst") in ("실적","수주","배당","리스크","모멘텀","기타"):
a["catalyst"] = gem["catalyst"]
a["reason"] = f"[Gemini검증] {gem.get('reason','')}"[:500]
try:
llm_conf = float(gem.get("confidence", llm_conf) or llm_conf)
except Exception:
pass
a["investment_action"] = (
"매수관심" if a["sentiment"] == "호재" and (a.get("intensity") or 0) >= 3
else "매도관심" if a["sentiment"] == "악재" and (a.get("intensity") or 0) >= 3
else "관망"
)
logger.info("news.path", path="gemini", final=a["sentiment"],
halluc=round(halluc, 2), conflict=conflict,
cost_krw=gem_cost)
else:
logger.info("news.path", path="exaone_lowconf",
halluc=round(halluc, 2), conflict=conflict,
conf=round(llm_conf, 2))
else:
logger.info("news.path", path="exaone",
sentiment=a.get("sentiment"),
conf=round(llm_conf, 2), halluc=round(halluc, 2))
# 5. Qdrant 저장
# 파싱실패 복구분: EXAONE/Gemini는 종목을 안 주므로(스키마 없음), 바른 감지 종목으로
# primary_stock·stock_impacts 채움 — 안 그러면 score-engine 뉴스점수에 안 잡힘.
if parse_failed and not a.get("primary_stock") and not a.get("stock_impacts"):
_codes = [st["code"] for st in data.get("stocks", []) if st.get("code")]
if _codes:
_w = 1.0 if len(_codes) == 1 else 0.6
a["primary_stock"] = _codes[0]
a["stock_impacts"] = {c: _w for c in _codes[:5]}
# 출처 신뢰도 + 제목 강도 (분석 후 DB INSERT 직전에 계산)
src_cred = await get_source_credibility(item.get("source", ""))
title_str = measure_title_strength(data["title"])
# 5. Qdrant 저장 (event_cluster_id 포함 — 다음 뉴스의 클러스터 매칭에 사용)
try:
await client.put(f"{QDRANT_URL}/collections/news_vectors/points", json={
"points":[{"id":random.randint(1,999999999),"vector":emb,
"payload":{"title":data["title"],"hash":data["hash"],
"sentiment":a.get("sentiment",""),"intensity":a.get("intensity",0),
"primary_stock":a.get("primary_stock","")}}]}, timeout=15)
"primary_stock":a.get("primary_stock",""),
"event_cluster_id": event_cluster_id,
"catalyst": a.get("catalyst", "기타"),
"time_horizon": a.get("time_horizon","단기")}}]}, timeout=15)
except: pass
# 6. PostgreSQL 저장 (파라미터 바인딩 — 인젝션·이스케이프 유실 방지)
# 6. PostgreSQL 저장
pub = data.get("published_at") or ""
try:
pub_dt = datetime.fromisoformat(pub.replace("Z", "+00:00")) if pub else datetime.now()
@@ -474,8 +806,11 @@ async def pipeline(item, client):
await pg_pool.execute("""
INSERT INTO news_analysis (title,url,source,published_at,hash,sentiment,intensity,
primary_stock,affected_stocks,reason,investment_action,keywords,stock_names,stock_codes,
catalyst,similar_count,analyzed_at)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9::jsonb,$10,$11,$12::jsonb,$13::jsonb,$14::jsonb,$15,0,NOW())
catalyst,similar_count,analyzed_at,
time_horizon,impact_scope,llm_confidence,source_credibility,title_strength,
stock_impacts,event_cluster_id,is_event_seed,sector_hint)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9::jsonb,$10,$11,$12::jsonb,$13::jsonb,$14::jsonb,$15,0,NOW(),
$16,$17,$18,$19,$20,$21::jsonb,$22,$23,$24)
ON CONFLICT (hash) DO NOTHING
""",
s(data.get("title"))[:500], s(data.get("url"))[:500], s(data.get("source"))[:100],
@@ -485,7 +820,26 @@ async def pipeline(item, client):
json.dumps(data.get("keywords", [])[:20], ensure_ascii=False),
json.dumps([s_["name"] for s_ in data.get("stocks", [])], ensure_ascii=False),
json.dumps([s_["code"] for s_ in data.get("stocks", [])], ensure_ascii=False),
s(a.get("catalyst", "기타"))[:20])
s(a.get("catalyst", "기타"))[:20],
s(a.get("time_horizon","단기"))[:10],
s(a.get("impact_scope","종목"))[:10],
llm_conf, src_cred, title_str,
json.dumps(a.get("stock_impacts", {}), ensure_ascii=False),
event_cluster_id[:32], is_event_seed,
s(a.get("sector_hint",""))[:40])
# 이벤트 클러스터 시드 등록 (첫 뉴스만)
if is_event_seed and event_cluster_id:
try:
await pg_pool.execute("""
INSERT INTO news_event_cluster (cluster_id, member_count,
seed_sentiment, seed_intensity, seed_catalyst)
VALUES ($1, 1, $2, $3, $4)
ON CONFLICT (cluster_id) DO NOTHING
""", event_cluster_id[:32], s(a.get("sentiment","중립"))[:10], iv,
s(a.get("catalyst","기타"))[:20])
except Exception: pass
return "ok"
except Exception as e:
logger.warning("pipeline.err", title=item.get("title","")[:50], error=str(e))
@@ -511,6 +865,8 @@ async def job_rss():
stats.processed += 1
elif r == "noise":
stats.noise += 1
elif r == "stale":
stats.stale += 1
elif r == "off_topic":
stats.off_topic += 1
stats.last_run = datetime.now().isoformat()
@@ -655,7 +1011,7 @@ async def job_historical_raw(count: int = 0, max_pages: int = 100):
stats.running = False
async def job_process_raw(batch_size: int = 200):
async def job_process_raw(batch_size: int = 200, reprocess: bool = False):
"""news_raw의 미처리 행을 batch_size만큼 EXAONE 분석 → news_analysis로 이동.
OLLAMA_NUM_PARALLEL=4와 매칭되는 4-동시 병렬 처리."""
async with pg_pool.acquire() as conn:
@@ -680,11 +1036,11 @@ async def job_process_raw(batch_size: int = 200):
"stock_code": r["stock_code"],
}
try:
if await is_dup(item["title"], item["url"]):
if not reprocess and await is_dup(item["title"], item["url"]):
stats.duplicates += 1
else:
stats.collected += 1
res = await pipeline(item, c)
res = await pipeline(item, c, reprocess=reprocess)
if res == "ok": ok += 1; stats.processed += 1
elif res == "noise": stats.noise += 1
elif res == "off_topic": stats.off_topic += 1
@@ -741,6 +1097,7 @@ async def job_historical(count: int = 500, max_pages: int = 20):
if r == "ok":
ok += 1; stats.processed += 1
elif r == "noise": stats.noise += 1
elif r == "stale": stats.stale += 1
elif r == "off_topic": stats.off_topic += 1
if (i + 1) % 50 == 0:
logger.info("historical.progress", done=i+1, of=len(codes),
@@ -774,6 +1131,8 @@ async def job_market():
stats.processed += 1
elif r == "noise":
stats.noise += 1
elif r == "stale":
stats.stale += 1
elif r == "off_topic":
stats.off_topic += 1
stats.last_run = datetime.now().isoformat()
@@ -817,8 +1176,8 @@ async def startup():
pg_pool = await asyncpg.create_pool(host=PG_HOST,port=PG_PORT,database=PG_DB,user=PG_USER,password=PG_PASS,min_size=2,max_size=5)
redis_cl = aioredis.Redis(host=REDIS_HOST,port=6379,password=REDIS_PASSWORD,db=4,decode_responses=True)
await init_news_tables()
# 평일 RSS: 8-18시 5분마다
scheduler.add_job(job_rss,"cron",day_of_week="mon-fri",hour="8-18",minute="*/5",id="rss_weekday",replace_existing=True)
# 평일 RSS: 8-18시 2분마다 (자동매매 latency 단축)
scheduler.add_job(job_rss,"cron",day_of_week="mon-fri",hour="8-18",minute="*/2",id="rss_weekday",replace_existing=True)
# 주말 RSS: 8-22시 15분마다 (누적 학습용)
scheduler.add_job(job_rss,"cron",day_of_week="sat,sun",hour="8-22",minute="*/15",id="rss_weekend",replace_existing=True)
# 평일 마켓 (네이버 금융): 9-17시 10분마다
@@ -831,6 +1190,11 @@ async def startup():
scheduler.add_job(job_historical_raw,"cron",day_of_week="sun",hour="2",minute="0",
id="historical_raw_weekly",replace_existing=True,
kwargs={"count":0,"max_pages":50})
# 평일 18:30 전 종목 "최신 1페이지" 스위프 — RSS/마켓메인에 안 잡히는 중형주
# 신선뉴스가 주1회 백필까지 누락되던 사각지대 해소 (전 종목×1페이지 ≈ 25분)
scheduler.add_job(job_historical_raw,"cron",day_of_week="mon-fri",hour="18",minute="30",
id="fresh_stock_sweep",replace_existing=True,
kwargs={"count":0,"max_pages":1})
scheduler.start()
logger.info("news-collector.started", sources=len(RSS_SOURCES))
@@ -866,9 +1230,10 @@ async def m_historical_raw(count: int = 0, max_pages: int = 100):
return {"status":"started","type":"historical_raw","count":count or "전체","max_pages":max_pages}
@app.post("/process/raw")
async def m_process_raw(batch_size: int = 200):
"""news_raw 미처리분 batch 분석 → news_analysis로 이동"""
n = await job_process_raw(batch_size)
async def m_process_raw(batch_size: int = 200, reprocess: bool = False):
"""news_raw 미처리분 batch 분석 → news_analysis로 이동.
reprocess=true: is_dup·Qdrant sim_dup 우회 (파싱실패 복구 재분석용)."""
n = await job_process_raw(batch_size, reprocess=reprocess)
return {"status":"done","processed":n}
@app.get("/raw/stats")