""" 네이버 금융 뉴스 수집기 + AI 분석 파이프라인 - 시장 전체 뉴스 (5분마다) - 시총 상위 200개 종목별 뉴스 (30분마다) - 수집 즉시 바른API → Ollama → Qdrant → vLLM → PostgreSQL """ import asyncio, hashlib, json, os, re, random, time from datetime import datetime, timedelta from typing import Optional 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 structlog.configure(processors=[ structlog.processors.TimeStamper(fmt="iso"), structlog.processors.add_log_level, structlog.processors.JSONRenderer(), ]) logger = structlog.get_logger() REDIS_HOST = os.getenv("REDIS_HOST", "redis") REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", "") PG_HOST = os.getenv("POSTGRES_HOST", "postgres") PG_PORT = int(os.getenv("POSTGRES_PORT", "5432")) PG_DB = os.getenv("POSTGRES_DB", "trading_ai") PG_USER = os.getenv("POSTGRES_USER", "kyu") 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") VLLM_URL = os.getenv("VLLM_URL", "http://vllm:8000") QDRANT_URL = os.getenv("QDRANT_URL", "http://qdrant:6333") HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} pg_pool: Optional[asyncpg.Pool] = None redis_cl: Optional[aioredis.Redis] = None scheduler = AsyncIOScheduler(timezone="Asia/Seoul") class S: collected = 0; processed = 0; duplicates = 0; errors = 0 last_run = ""; running = False stats = S() def nhash(title, url=""): return hashlib.sha256(f"{title.strip()}{url.strip()}".encode()).hexdigest()[:16] async def is_dup(h): if not redis_cl: return False try: r = await redis_cl.set(f"news:naver:{h}", "1", ex=86400, nx=True) return r is None except: return False # ── 크롤러 ──────────────────────────────────────────────── async def crawl_market_news(client): news = [] urls = [ "https://finance.naver.com/news/mainnews.naver", "https://finance.naver.com/news/news_list.naver?mode=LSS2D§ion_id=101§ion_id2=258", "https://finance.naver.com/news/news_list.naver?mode=LSS2D§ion_id=101§ion_id2=259", "https://finance.naver.com/news/news_list.naver?mode=LSS2D§ion_id=101§ion_id2=261", ] for url in urls: try: r = await client.get(url, headers=HEADERS, timeout=15) r.encoding = "euc-kr" soup = BeautifulSoup(r.text, "lxml") for a in soup.select("a[href*='article_id']"): t = a.get_text(strip=True) h = a.get("href", "") if t and len(t) > 10: news.append({"title": t, "url": f"https://finance.naver.com{h}" if h.startswith("/") else h, "source": "네이버금융", "content": "", "published_at": datetime.now().isoformat()}) await asyncio.sleep(0.3) except Exception as e: logger.warning("crawl.market.err", error=str(e)) return news async def crawl_stock_news(client, code, name): news = [] try: r = await client.get(f"https://finance.naver.com/item/news_news.naver?code={code}&page=1", headers=HEADERS, timeout=15) r.encoding = "euc-kr" soup = BeautifulSoup(r.text, "lxml") for tr in soup.select("table.type5 tr"): a = tr.select_one("td.title a") dt = tr.select_one("td.date") src = tr.select_one("td.info") if a: t = a.get_text(strip=True) if t and len(t) > 5: news.append({"title": t, "url": f"https://finance.naver.com{a.get('href','')}", "source": src.get_text(strip=True) if src else "네이버금융", "content": f"[{name}({code})] {t}", "published_at": dt.get_text(strip=True) if dt else datetime.now().isoformat(), "stock_code": code, "stock_name": name}) except: pass return news async def get_top_stocks(client, count=200): stocks = [] for sosok in [0, 1]: for page in range(1, 50): try: r = await client.get(f"https://finance.naver.com/sise/sise_market_sum.naver?sosok={sosok}&page={page}", headers=HEADERS, timeout=15) r.encoding = "euc-kr" rows = re.findall(r'main\.naver\?code=(\d{6})[^>]*>([^<]+)', r.text) if not rows: break for c, n in rows: stocks.append({"code": c.strip(), "name": n.strip()}) await asyncio.sleep(0.2) except: break if len(stocks) >= count: break return stocks[:count] # ── 파이프라인 ───────────────────────────────────────────── async def pipeline(item, client): try: # 1. 바른API br = await client.post(f"{BAREUN_URL}/analyze", json={ "title": item["title"], "content": item.get("content",""), "url": item.get("url",""), "source": item.get("source",""), "published_at": item.get("published_at","")}, timeout=30) bd = br.json() if bd.get("is_duplicate"): return "dup" data = {**item, "hash": bd.get("hash",""), "stocks": bd.get("stocks",[]), "keywords": bd.get("keywords",[]), "filtered_text": bd.get("filtered_text","")} # 2. Ollama 임베딩 er = await client.post(f"{OLLAMA_URL}/api/embeddings", json={"model":"bge-m3","prompt": data["filtered_text"] or data["title"]}, timeout=60) emb = er.json().get("embedding") if not emb: return "no_embed" # 3. Qdrant 유사도 try: sr = await client.post(f"{QDRANT_URL}/collections/news_vectors/points/search", json={"vector":emb,"limit":3,"score_threshold":0.85,"with_payload":True}, timeout=15) hits = sr.json().get("result",[]) if any(h["score"]>=0.92 and h["score"]<0.99 for h in hits): return "sim_dup" except: hits = [] # 4. vLLM 분석 stocks_str = ", ".join([f'{s["name"]}({s["code"]})' for s in data["stocks"][:5]]) prompt = (f"뉴스: {data['title'][:200]}\n키워드: {(data['filtered_text'] or '')[:300]}\n" f"감지된 종목: {stocks_str}\n\n" "JSON 응답. primary_stock에 6자리 종목코드, 시장전체면 KOSPI/KOSDAQ:\n" '{"sentiment":"호재/악재/중립","intensity":1~5,"primary_stock":"종목코드",' '"affected_stocks":["코드"],"reason":"근거","investment_action":"매수관심/매도관심/관망"}') vr = await client.post(f"{VLLM_URL}/v1/chat/completions", json={ "model":"exaone","messages":[ {"role":"system","content":"한국 주식 전문 애널리스트. JSON만 응답."}, {"role":"user","content":prompt}], "max_tokens":300,"temperature":0.1}, timeout=120) try: c = vr.json()["choices"][0]["message"]["content"] a = json.loads(re.sub(r"```json\n?|```","",c).strip()) except: a = {"sentiment":"중립","intensity":0,"primary_stock":"","affected_stocks":[],"reason":"파싱실패","investment_action":"관망"} # 5. Qdrant 저장 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) except: pass # 6. PostgreSQL 저장 esc = lambda s: (s or "").replace("'","''") await pg_pool.execute(f""" INSERT INTO news_analysis (title,url,source,published_at,hash,sentiment,intensity, primary_stock,affected_stocks,reason,investment_action,keywords,stock_names,stock_codes,similar_count,analyzed_at) VALUES ('{esc(data["title"][:500])}','{esc(data.get("url","")[:500])}','{esc(data.get("source","")[:100])}', '{data.get("published_at",datetime.now().isoformat())}','{data["hash"]}', '{a.get("sentiment","중립")}',{a.get("intensity",0)},'{esc(a.get("primary_stock",""))}', '{json.dumps(a.get("affected_stocks",[]))}','{esc(a.get("reason","")[:500])}', '{a.get("investment_action","관망")}','{json.dumps(data.get("keywords",[])[:20])}', '{json.dumps([s["name"] for s in data.get("stocks",[])])}', '{json.dumps([s["code"] for s in data.get("stocks",[])])}', 0,'{datetime.now().isoformat()}') ON CONFLICT (hash) DO NOTHING""") return "ok" except Exception as e: logger.warning("pipeline.err", title=item.get("title","")[:50], error=str(e)) return "error" # ── 수집 작업 ────────────────────────────────────────────── async def job_market(): if stats.running: return stats.running = True try: async with httpx.AsyncClient() as c: news = await crawl_market_news(c) ok = 0 for item in news: h = nhash(item["title"], item.get("url","")) if await is_dup(h): stats.duplicates+=1; continue stats.collected += 1 r = await pipeline(item, c) if r == "ok": ok += 1; stats.processed += 1 await asyncio.sleep(0.5) stats.last_run = datetime.now().isoformat() logger.info("job.market", total=len(news), processed=ok) except Exception as e: stats.errors += 1; logger.error("job.market.err", error=str(e)) finally: stats.running = False async def job_stocks(): if stats.running: return stats.running = True try: async with httpx.AsyncClient() as c: top = await get_top_stocks(c, 200) ok = 0 for stock in top: try: news = await crawl_stock_news(c, stock["code"], stock["name"]) for item in news: h = nhash(item["title"], item.get("url","")) if await is_dup(h): stats.duplicates+=1; continue stats.collected += 1 r = await pipeline(item, c) if r == "ok": ok += 1; stats.processed += 1 await asyncio.sleep(0.3) except: pass stats.last_run = datetime.now().isoformat() logger.info("job.stocks", stocks=len(top), processed=ok) except Exception as e: stats.errors += 1; logger.error("job.stocks.err", error=str(e)) finally: stats.running = False # ── FastAPI ──────────────────────────────────────────────── app = FastAPI(title="뉴스 수집기") app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) @app.on_event("startup") async def startup(): global pg_pool, redis_cl 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) scheduler.add_job(job_market,"cron",day_of_week="mon-fri",hour="8-18",minute="*/5",id="market",replace_existing=True) scheduler.add_job(job_stocks,"cron",day_of_week="mon-fri",hour="9-16",minute="*/30",id="stocks",replace_existing=True) scheduler.start() logger.info("news-collector.started") @app.on_event("shutdown") async def shutdown(): scheduler.shutdown() if pg_pool: await pg_pool.close() if redis_cl: await redis_cl.aclose() @app.get("/health") async def health(): return JSONResponse(content={"status":"ok","collected":stats.collected,"processed":stats.processed, "duplicates":stats.duplicates,"errors":stats.errors,"last_run":stats.last_run,"running":stats.running}) @app.post("/collect/market") async def m_market(): asyncio.create_task(job_market()); return {"status":"started","type":"market"} @app.post("/collect/stocks") async def m_stocks(): asyncio.create_task(job_stocks()); return {"status":"started","type":"stocks"}