from backend.schemas.stock_schemas import TickerInfo import yahooquery as yq from fastapi import HTTPException COMPANY_MAP = { "NVDA": { "company_name": "NVIDIA", "aliases": ["엔비디아", "nvidia", "nvda"], }, "AAPL": { "company_name": "Apple", "aliases": ["애플", "apple", "aapl"], }, "MSFT": { "company_name": "Microsoft", "aliases": ["마이크로소프트", "microsoft", "msft"], }, "TSLA": { "company_name": "Tesla", "aliases": ["테슬라", "tesla", "tsla"], }, "GOOGL": { "company_name": "Alphabet (Google)", "aliases": ["구글", "google", "알파벳", "alphabet", "googl"], }, "AMZN": { "company_name": "Amazon", "aliases": ["아마존", "amazon", "amzn"], }, "META": { "company_name": "Meta Platforms", "aliases": ["메타", "meta", "페이스북", "facebook", "fb"], }, "AMD": { "company_name": "Advanced Micro Devices", "aliases": ["amd"], }, "INTC": { "company_name": "Intel", "aliases": ["인텔", "intel", "intc"], }, "TSM": { "company_name": "TSMC", "aliases": ["tsmc", "tsm", "대만반도체"], }, "QCOM": { "company_name": "Qualcomm", "aliases": ["퀄컴", "qualcomm", "qcom"], }, "AVGO": { "company_name": "Broadcom", "aliases": ["브로드컴", "broadcom", "avgo"], }, "MU": { "company_name": "Micron Technology", "aliases": ["마이크론", "micron", "mu"], }, "V": { "company_name": "Visa", "aliases": ["비자", "visa", "v"], }, "MA": { "company_name": "Mastercard", "aliases": ["마스터카드", "mastercard", "ma"], }, "JPM": { "company_name": "JPMorgan Chase", "aliases": ["jp모건", "jpmorgan", "jpm"], }, "JNJ": { "company_name": "Johnson & Johnson", "aliases": ["존슨앤존슨", "johnson", "jnj"], }, "MSFT": { "company_name": "Microsoft", "aliases": ["마이크로소프트", "microsoft", "msft"], }, } class StockService: """주식 분석""" async def _extract(self, query: str): """ 사용자 쿼리에서 티커와 회사명 추출 1차 : Map 이용 2차 : yahooquery 이용 query : 엔비디아 분석해줘 """ keyword = self._extract_company_keyword(query) # 1 차 Map result = self._extract_ticker_from_query(keyword) if result: return result # 2 차 yahooquery result = await self._search_yahoo_symbol(keyword) if result: return result raise HTTPException(status_code=30000, detail="종목을 찾을 수 없습니다.") def _extract_ticker_from_query(self, query:str): """ COMPANY_MAP 안에서 일치하는 회사 찾기 """ query = query.lower() for ticker, info in COMPANY_MAP.items(): for alias in info["aliases"]: if alias.lower() in query: return TickerInfo(ticker = ticker, company_name = info["company_name"]) def _extract_company_keyword(self, query:str): """ query : 엔비디아 분석해줘 => 필요없는 문장 제거한 후 키워드만 추출 """ stop_words = [ "분석해줘", "분석해", "분석", "해줘", "해", "주가", "전망", "재무", "알려줘", "어때", "어떻게", "" ] keyword = query for word in stop_words: keyword = keyword.replace(word, "") return keyword.strip() async def _search_yahoo_symbol(self, keyword: str): """ yahooquery 이용 """ try: result = yq.search(keyword, first_quote=True) if not result or (isinstance(result, dict) and "explanation" in result): return None return TickerInfo(ticker=result["symbol"], company_name=result["longname"]) except Exception: return None