trading/bareunaapi-main.py

"""
바른 API FastAPI 서버 v2
- 서버 시작 시 KRX 전체 종목 동적 로딩
- 24시간마다 자동 갱신
- Bareun gRPC 연결
"""
import asyncio, hashlib, os, time
from contextlib import asynccontextmanager
from typing import Optional
import orjson, redis.asyncio as aioredis, structlog
from bareunpy import Tagger
from fastapi import FastAPI, HTTPException, Response
from fastapi.responses import JSONResponse
from fastapi.middleware.cors import CORSMiddleware
from prometheus_fastapi_instrumentator import Instrumentator
from pydantic import BaseModel, Field
from stock_loader import load_all_stocks, auto_refresh_stocks

structlog.configure(processors=[
    structlog.processors.TimeStamper(fmt="iso"),
    structlog.processors.add_log_level,
    structlog.processors.JSONRenderer(),
])
logger = structlog.get_logger()

BAREUN_API_KEY     = os.getenv("BAREUN_API_KEY", "")
BAREUN_SERVER_HOST = os.getenv("BAREUN_SERVER_HOST", "bareun")
BAREUN_SERVER_PORT = int(os.getenv("BAREUN_SERVER_PORT", "5656"))
REDIS_HOST         = os.getenv("REDIS_HOST", "redis")
REDIS_PORT         = int(os.getenv("REDIS_PORT", "6379"))
REDIS_PASSWORD     = os.getenv("REDIS_PASSWORD", "")
REDIS_DB           = int(os.getenv("REDIS_DB", "1"))
NEWS_DEDUP_TTL     = int(os.getenv("NEWS_DEDUP_TTL", "86400"))

STOPWORDS = {
    "것","수","등","및","또","또한","이","그","저","위","아래","관련","통해",
    "위해","대해","따라","때문","이후","이전","현재","최근","지난","올해","내년",
    "이번","오늘","어제","내일","이날","같은","다른","많은","더","가장","매우",
    "이미","아직","모든","각","전체","일부","국내","해외","글로벌","세계","한국",
    "미국","중국","일본","유럽","가운데","한편","다만","특히","실제","여전히",
    "앞으로","지속","계속","자체","관계자","측","쪽","곳","점","경우","상황",
    "때","중","간",
}
FINANCE_KEYWORDS = {
    "실적","매출","영업이익","순이익","적자","흑자","손실","이익","투자","배당",
    "자사주","유상증자","합병","인수","M&A","IPO","상장","상폐","거래정지","금리",
    "기준금리","환율","달러","반도체","배터리","전기차","수소","AI","인공지능",
    "클라우드","바이오","의약품","임상","승인","특허","수주","계약","부도","파산",
    "부채","시총","코스피","코스닥","나스닥",
}
START_TIME = time.time()

class AppState:
    tagger: Optional[Tagger] = None
    redis: Optional[aioredis.Redis] = None
    stock_map: dict[str, str] = {}
    stock_count: int = 0
    refresh_task: Optional[asyncio.Task] = None

state = AppState()

@asynccontextmanager
async def lifespan(app: FastAPI):
    state.stock_map = await load_all_stocks()
    state.stock_count = len(state.stock_map)
    logger.info("stocks.ready", count=state.stock_count)
    state.refresh_task = asyncio.create_task(auto_refresh_stocks(state, 24))
    try:
        state.tagger = Tagger(BAREUN_API_KEY, BAREUN_SERVER_HOST, BAREUN_SERVER_PORT)
        logger.info("tagger.ok")
    except Exception as e:
        logger.warning("tagger.failed", error=str(e))
    try:
        state.redis = aioredis.Redis(host=REDIS_HOST, port=REDIS_PORT,
            password=REDIS_PASSWORD, db=REDIS_DB, decode_responses=True,
            socket_connect_timeout=5, retry_on_timeout=True)
        await state.redis.ping()
        logger.info("redis.ok")
    except Exception as e:
        logger.error("redis.failed", error=str(e))
    yield
    if state.refresh_task: state.refresh_task.cancel()
    if state.redis: await state.redis.aclose()

app = FastAPI(title="바른 API v2", version="2.0.0", lifespan=lifespan)
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
Instrumentator().instrument(app).expose(app, endpoint="/metrics")

class AnalyzeRequest(BaseModel):
    title: str; content: str = ""; url: str = ""; source: str = ""; published_at: str = ""

class StockMention(BaseModel):
    name: str; code: str; count: int

class AnalyzeResponse(BaseModel):
    hash: str; is_duplicate: bool; stocks: list[StockMention]; keywords: list[str]
    filtered_text: str; token_count: int; processing_time_ms: float

class BatchRequest(BaseModel):
    items: list[AnalyzeRequest]

def news_hash(title, url):
    return hashlib.sha256(f"{title.strip()}{url.strip()}".encode()).hexdigest()[:16]

async def is_duplicate(h):
    if not state.redis: return False
    try:
        r = await state.redis.set(f"news:dedup:{h}", "1", ex=NEWS_DEDUP_TTL, nx=True)
        return r is None
    except: return False

def extract_morphemes(text):
    if not state.tagger or not text.strip():
        return [w for w in text.split() if len(w) >= 2 and w not in STOPWORDS]
    try:
        return [t for t, p in state.tagger.pos(text)
                if p in ("NNG","NNP","SL") and len(t) >= 2 and t not in STOPWORDS]
    except:
        return [w for w in text.split() if len(w) >= 2 and w not in STOPWORDS]

def extract_stocks(text):
    found = {}
    for name, code in state.stock_map.items():
        c = text.count(name)
        if c > 0: found[name] = StockMention(name=name, code=code, count=c)
    return sorted(found.values(), key=lambda x: x.count, reverse=True)

def build_filtered(nouns, stocks):
    sn = {s.name for s in stocks}
    p = [n for n in nouns if n in sn]
    f = [n for n in nouns if n in FINANCE_KEYWORDS]
    g = [n for n in nouns if n not in STOPWORDS and len(n) >= 2]
    return " ".join(list(dict.fromkeys(p + f + g))[:100])

def _analyze(req):
    text = f"{req.title} {req.content}".strip()
    h = news_hash(req.title, req.url)
    nouns = extract_morphemes(text)
    stocks = extract_stocks(text)
    kw = list(dict.fromkeys(n for n in nouns if len(n) >= 2))[:50]
    ft = build_filtered(nouns, stocks)
    return h, stocks, kw, ft

@app.get("/health")
async def health():
    tok = state.tagger is not None
    rok = False
    if state.redis:
        try: await state.redis.ping(); rok = True
        except: pass
    return JSONResponse(content={"status": "ok" if tok else "degraded",
        "tagger": "ok" if tok else "unavailable", "redis": "ok" if rok else "error",
        "stocks_loaded": state.stock_count, "uptime": round(time.time()-START_TIME,1)})

@app.post("/analyze")
async def analyze(req: AnalyzeRequest):
    t = time.perf_counter()
    h, stocks, kw, ft = _analyze(req)
    dup = await is_duplicate(h)
    ms = round((time.perf_counter()-t)*1000, 2)
    return Response(content=orjson.dumps(AnalyzeResponse(
        hash=h, is_duplicate=dup, stocks=stocks, keywords=kw,
        filtered_text=ft, token_count=len(ft.split()), processing_time_ms=ms
    ).model_dump()), media_type="application/json")

@app.post("/analyze/batch")
async def analyze_batch(req: BatchRequest):
    if len(req.items) > 50: raise HTTPException(400, "최대 50개")
    t = time.perf_counter()
    results = []
    for item in req.items:
        try:
            h, stocks, kw, ft = _analyze(item)
            dup = await is_duplicate(h)
            results.append({"title":item.title,"hash":h,"is_duplicate":dup,
                "stocks":[s.model_dump() for s in stocks],"keywords":kw,
                "filtered_text":ft,"token_count":len(ft.split())})
        except Exception as e:
            results.append({"title":item.title,"error":str(e),"is_duplicate":False})
    ms = round((time.perf_counter()-t)*1000, 2)
    dups = sum(1 for r in results if r.get("is_duplicate"))
    return Response(content=orjson.dumps({"total":len(results),"duplicates":dups,
        "processed":len(results)-dups,"elapsed_ms":ms,"results":results}),
        media_type="application/json")

@app.get("/stocks")
async def stocks_list():
    return JSONResponse(content={"count":len(state.stock_map),
        "stocks":[{"name":k,"code":v} for k,v in list(state.stock_map.items())[:500]]})

@app.post("/stocks/refresh")
async def refresh():
    m = await load_all_stocks()
    if len(m) > 50:
        state.stock_map = m; state.stock_count = len(m)
        return JSONResponse(content={"status":"ok","count":len(m)})
    raise HTTPException(500, "종목 로딩 실패")

@app.delete("/dedup/flush")
async def flush():
    if not state.redis: raise HTTPException(503)
    keys = await state.redis.keys("news:dedup:*")
    if keys: await state.redis.delete(*keys)
    return {"deleted": len(keys)}