import streamlit as st import yfinance as yf import FinanceDataReader as fdr import pandas as pd import plotly.graph_objects as go from huggingface_hub import InferenceClient from duckduckgo_search import DDGS import json import datetime import re import time import random # 1. 페이지 설정 st.set_page_config(page_title="Pocket Quant AI", page_icon="🧠", layout="wide") # 2. 토큰 확인 if "HF_TOKEN" in st.secrets: client = InferenceClient(api_key=st.secrets["HF_TOKEN"]) else: st.error("🚨 HF_TOKEN required in secrets.") st.stop() # ============================================================================== # 🌐 [UI] 다국어 딕셔너리 # ============================================================================== UI_TEXT = { "KR": { "sidebar_title": "🧠 Pocket Quant AI", "sidebar_weather": "📊 오늘의 시장 날씨 (Market Weather)", "menu_search": "🔍 AI 스마트 검색 (실적+뉴스)", "menu_scanner": "📡 글로벌 시장 스캐너", "search_title": "🧠 AI 주식 비서 (분기 실적 분석)", "search_placeholder": "예: 삼성전자 실적 분석, 몽골 고비 캐시미어...", "btn_analyze": "분석 시작", "status_thinking": "🧠 AI가 분기 실적과 시장을 분석 중입니다...", "status_news": "📰 뉴스 및 재무제표 수집 중...", "metric_price": "현재가", "metric_source": "출처", "expander_news": "📰 뉴스 원문 보기", "insight_title": "🧠 AI 종합 분석 리포트", "scanner_title": "📡 글로벌 시장 전광판 (Frontier Included)", "scanner_caption": "※ 100개 스캔 후, 상위 종목의 분기 실적 흐름을 분석합니다.", "btn_scan": "전체 종목 스캔 & AI 리포트", "tab_gainers": "🔥 급등 Top 10", "tab_losers": "💧 급락 Top 10", "col_name": "기업명", "col_price": "가격", "col_change": "등락률", "col_per": "PER", "col_pbr": "PBR", "col_source": "출처", "briefing_title": "🧠 AI Market Briefing (오늘의 시장 분석)", "msg_fail": "데이터 수집 실패", "llm_lang_instruction": "Korean" }, "EN": { "sidebar_title": "🧠 Pocket Quant AI", "sidebar_weather": "📊 Market Weather", "menu_search": "🔍 AI Smart Search (Financials)", "menu_scanner": "📡 Global Market Scanner", "search_title": "🧠 AI Stock Assistant (Quarterly Analysis)", "search_placeholder": "e.g. Analyze Samsung Electronics financials...", "btn_analyze": "Analyze", "status_thinking": "🧠 AI is analyzing quarterly financials...", "status_news": "📰 Fetching news & financials...", "metric_price": "Price", "metric_source": "Source", "expander_news": "📰 View Source News", "insight_title": "🧠 AI Comprehensive Report", "scanner_title": "📡 Global Market Scanner (Frontier Included)", "scanner_caption": "※ Scans stocks, then analyzes quarterly trends for top movers.", "btn_scan": "Scan & Generate Report", "tab_gainers": "🔥 Top 10 Gainers", "tab_losers": "💧 Top 10 Losers", "col_name": "Company", "col_price": "Price", "col_change": "Change(%)", "col_per": "PER", "col_pbr": "PBR", "col_source": "Source", "briefing_title": "🧠 AI Market Briefing", "msg_fail": "Data collection failed", "llm_lang_instruction": "English" } } # ============================================================================== # 💾 [데이터] 티커 매핑 (Full List) # ============================================================================== TICKER_NAMES = { # (기존 리스트 유지 - 너무 길어서 생략하지만 실제 코드엔 꼭 넣으세요!) "VIC.VN": "Vingroup", "VHM.VN": "Vinhomes", "VCB.VN": "Vietcombank", "VNM.VN": "Vinamilk", "005930.KS": "Samsung Elec", "000660.KS": "SK Hynix", "AAPL": "Apple", "NVDA": "NVIDIA", "APU.MN": "APU JSC", "GP.BD": "Grameenphone", # Frontier # ... (나머지 국가들 포함) ... } # (스캐너용 MARKET_SAMPLES도 기존 유지) MARKET_SAMPLES = { "🇻🇳 Vietnam": ["VIC.VN", "VHM.VN", "VRE.VN", "VNM.VN", "MSN.VN", "GAS.VN", "HPG.VN", "NVL.VN", "PDR.VN", "DIG.VN", "SSI.VN", "VND.VN", "MWG.VN", "FRT.VN", "FPT.VN", "STB.VN", "TCB.VN", "VCB.VN"], "🇺🇸 USA": ["AAPL", "MSFT", "GOOGL", "AMZN", "NVDA", "META", "TSLA", "AMD", "INTC", "PLTR", "COIN", "LLY", "XOM", "DIS"], "🇰🇷 Korea": ["005930.KS", "000660.KS", "005380.KS", "000270.KS", "035420.KS", "035720.KS", "005490.KS", "086520.KQ", "247540.KQ", "207940.KS"], "🇲🇳 Mongolia (Frontier)": ["APU.MN", "TTL.MN", "GOV.MN", "MNDL.MN", "SUU.MN"], "🇧🇩 Bangladesh (Frontier)": ["GP.BD", "SQURPHARMA.BD", "BATBC.BD", "BEXIMCO.BD"], "🇭🇰 Hong Kong": ["1299.HK", "0388.HK", "0005.HK", "0700.HK", "9988.HK", "3690.HK"], "🇹🇭 Thailand": ["PTT.BK", "AOT.BK", "CPALL.BK", "ADVANC.BK", "KBANK.BK"], "🇮🇩 Indonesia": ["BBCA.JK", "BBRI.JK", "BMRI.JK", "TLKM.JK", "ASII.JK", "GOTO.JK"], "🇹🇼 Taiwan": ["2330.TW", "2317.TW", "2454.TW", "2881.TW", "2308.TW"], "🇯🇵 Japan": ["7203.T", "6758.T", "9984.T", "8035.T", "6861.T"], "🇮🇳 India": ["RELIANCE.NS", "TCS.NS", "HDFCBANK.NS", "INFY.NS"] } # ============================================================================== # 🛠️ [엔진] 지수 & 데이터 수집 (Fundamentals 강화) # ============================================================================== def get_market_indices(): indices = {"S&P 500": "^GSPC", "NASDAQ": "^IXIC", "KOSPI": "^KS11", "VN-INDEX": "^VNINDEX"} data = {} try: df = yf.download(list(indices.values()), period="5d", progress=False)['Close'] for name, ticker in indices.items(): if ticker in df.columns: series = df[ticker].dropna() if len(series) >= 2: curr, prev = series.iloc[-1], series.iloc[-2] data[name] = (curr, ((curr - prev)/prev)*100) except: pass return data # [NEW] 재무 데이터 + 분기 실적 가져오기 def get_fundamentals(ticker): """ 기본 지표(PER/PBR) + 최근 4분기 실적(매출/이익) 추출 """ data = {"PER": "N/A", "PBR": "N/A", "ROE": "N/A", "Quarterly": pd.DataFrame()} try: stock = yf.Ticker(ticker) info = stock.info # 1. 기본 지표 per = info.get('trailingPE') pbr = info.get('priceToBook') roe = info.get('returnOnEquity') data["PER"] = f"{per:.2f}" if per else "N/A" data["PBR"] = f"{pbr:.2f}" if pbr else "N/A" data["ROE"] = f"{roe*100:.2f}%" if roe else "N/A" # 2. 분기별 실적 (최근 4분기) q_fin = stock.quarterly_financials if not q_fin.empty: # 주요 항목만 추출 (Total Revenue, Net Income) - 키 이름은 야후 버전에 따라 다를 수 있어 예외처리 target_rows = [r for r in ['Total Revenue', 'Operating Revenue', 'Net Income', 'Net Income Common Stockholders'] if r in q_fin.index] if target_rows: # 최근 4개 분기만, 보기 좋게 Transpose recent_q = q_fin.loc[target_rows].iloc[:, :4].T # 날짜 포맷 정리 (YYYY-MM-DD) recent_q.index = [d.strftime('%Y-%m') for d in recent_q.index] data["Quarterly"] = recent_q except: pass return data FAKE_HEADERS = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'} def fetch_news_robust(keyword): summary = "" seen_urls = set() try: with DDGS() as ddgs: results = list(ddgs.news(keyword, timelimit="m", max_results=3)) if not results: results = list(ddgs.news(keyword, max_results=3)) if not results: results = list(ddgs.text(f"{keyword} latest news", max_results=3)) for r in results: url = r.get('url') or r.get('href') if url not in seen_urls: summary += f"[{r.get('date', '?')}] {r.get('title', '')}\n" seen_urls.add(url) except: return "No news." return summary if summary else "No news." def scrape_price_from_web(name, ticker): try: time.sleep(random.uniform(1.0, 1.5)) with DDGS() as ddgs: query = f"{name} {ticker} stock price quote today" results = ddgs.text(query, max_results=2) blob = " ".join([r['body'] for r in results]) match = re.search(r'(\d{1,3}(,\d{3})*(\.\d+)?)', blob) if match: return float(match.group(0).replace(",", "")), 0.0 except: pass return None, None def get_price_data_robust(ticker, name): try: fdr_symbol = ticker if ".VN" in ticker: fdr_symbol = ticker.split('.')[0] elif ".JK" in ticker: fdr_symbol = f"IDX:{ticker.split('.')[0]}" elif ".KS" in ticker or ".KQ" in ticker: fdr_symbol = ticker.split('.')[0] hist = fdr.DataReader(fdr_symbol, start=(datetime.datetime.now() - datetime.timedelta(days=7))) if not hist.empty: curr, prev = hist['Close'].iloc[-1], hist['Close'].iloc[-2] return curr, ((curr - prev)/prev)*100, "FDR" except: pass try: stock = yf.Ticker(ticker) price = stock.fast_info.last_price prev = stock.fast_info.previous_close if price and prev: return price, ((price - prev)/prev)*100, "Yahoo" except: pass price, pct = scrape_price_from_web(name, ticker) if price is not None: return price, pct, "Web Search" return None, None, "Fail" def get_chart_data(ticker): try: fdr_symbol = ticker if ".VN" in ticker: fdr_symbol = ticker.split('.')[0] elif ".JK" in ticker: fdr_symbol = f"IDX:{ticker.split('.')[0]}" elif ".T" in ticker: fdr_symbol = f"TSE:{ticker.split('.')[0]}" hist = fdr.DataReader(fdr_symbol, start=(datetime.datetime.now() - datetime.timedelta(days=90)).strftime('%Y-%m-%d')) if not hist.empty: return hist.rename(columns={'Close':'Close', 'Open':'Open', 'High':'High', 'Low':'Low'}), "FDR" except: pass try: hist = yf.Ticker(ticker).history(period="3mo") if not hist.empty: return hist, "Yahoo" except: pass return pd.DataFrame(), "None" def identify_targets_with_ai(user_query): try: prompt = f""" [ROLE] Entity Resolver. [QUERY] "{user_query}" [TASK] Identify ALL companies. [OUTPUT JSON LIST] [ {{ "name": "EngName", "ticker": "TICKER", "eng_key": "Name news", "native_key": "LocalName news" }} ] """ messages = [{"role": "user", "content": prompt}] response = client.chat.completions.create(model="Qwen/Qwen2.5-72B-Instruct", messages=messages, max_tokens=300) content = re.sub(r"```json|```", "", response.choices[0].message.content.strip()).strip() return json.loads(content) except: return [] def get_polyglot_news(eng_key, native_key): return f"Global:\n{fetch_news_robust(eng_key)}\nLocal:\n{fetch_news_robust(native_key)}" def parse_stream(stream): for chunk in stream: if chunk.choices: yield chunk.choices[0].delta.content or "" def plot_candle_chart(hist, title, source): fig = go.Figure(data=[go.Candlestick(x=hist.index, open=hist['Open'], high=hist['High'], low=hist['Low'], close=hist['Close'], name="Price")]) fig.update_layout(title=f"{title} ({source})", height=350, margin=dict(l=10, r=10, t=30, b=10), template="plotly_dark", paper_bgcolor='rgba(0,0,0,0)') return fig def plot_bar_chart(df, lang_data): target_col = lang_data['col_name'] colors = ['#00FF00' if x > 0 else '#FF0000' for x in df['Change(%)']] fig = go.Figure(go.Bar(x=df[target_col], y=df['Change(%)'], marker_color=colors, text=df['Change(%)'].apply(lambda x: f"{x:.2f}%"))) fig.update_layout(title="Market Heatmap", height=350, margin=dict(l=10, r=10, t=30, b=10), template="plotly_dark", paper_bgcolor='rgba(0,0,0,0)') return fig # ============================================================================== # UI # ============================================================================== with st.sidebar: lang_code = st.selectbox("Language / 언어", ["KR", "EN"]) T = UI_TEXT[lang_code] st.title(T['sidebar_title']) st.divider() st.caption(T['sidebar_weather']) market_data = get_market_indices() market_context_str = "" if market_data: c1, c2 = st.columns(2) if "S&P 500" in market_data: p, c = market_data["S&P 500"] c1.metric("🇺🇸 S&P 500", f"{p:,.0f}", f"{c:+.2f}%") market_context_str += f"S&P 500 {c:+.2f}%, " if "NASDAQ" in market_data: p, c = market_data["NASDAQ"] c2.metric("🇺🇸 NASDAQ", f"{p:,.0f}", f"{c:+.2f}%") market_context_str += f"NASDAQ {c:+.2f}%, " c3, c4 = st.columns(2) if "KOSPI" in market_data: p, c = market_data["KOSPI"] c3.metric("🇰🇷 KOSPI", f"{p:,.0f}", f"{c:+.2f}%") if "VN-INDEX" in market_data: p, c = market_data["VN-INDEX"] c4.metric("🇻🇳 VN-IDX", f"{p:,.0f}", f"{c:+.2f}%") market_context_str += f"Vietnam Index {c:+.2f}%." else: st.caption("Loading failed.") st.divider() menu = st.radio("MENU", [T['menu_search'], T['menu_scanner']], index=0) # --- AI 검색 (재무제표 + 분기 실적 포함) --- if menu == T['menu_search']: st.subheader(T['search_title']) c1, c2 = st.columns([3, 1]) with c1: query = st.text_input("Query", placeholder=T['search_placeholder'], label_visibility="collapsed") with c2: btn = st.button(T['btn_analyze'], use_container_width=True) if btn: with st.status(T['status_thinking'], expanded=True) as status: targets = identify_targets_with_ai(query) if targets: collected_data = [] tabs = st.tabs([t.get('name', 'Unknown') for t in targets]) for i, target in enumerate(targets): name, ticker = target.get('name'), target.get('ticker') eng_key, native_key = target.get('eng_key'), target.get('native_key') with tabs[i]: st.info(f"📍 **{name} ({ticker})**") # [NEW] 재무 데이터 (분기 포함) fund_data = get_fundamentals(ticker) q_df = fund_data['Quarterly'] # 기본 지표 카드 f1, f2, f3 = st.columns(3) f1.metric("PER", fund_data["PER"]) f2.metric("PBR", fund_data["PBR"]) f3.metric("ROE", fund_data["ROE"]) # [NEW] 분기 실적 표 (있을 경우만) if not q_df.empty: st.caption("📊 Recent Quarterly Financials (4 Quarters)") st.dataframe(q_df.style.format("{:,.0f}"), use_container_width=True) else: st.caption("⚠️ No quarterly data available via API.") st.divider() h, src = get_chart_data(ticker) news_data = get_polyglot_news(eng_key, native_key) curr, pct = "N/A", 0 if not h.empty: curr, prev = h['Close'].iloc[-1], h['Close'].iloc[-2] pct = ((curr - prev)/prev)*100 else: c_web, p_web, s_web = get_price_data_robust(ticker, name) if c_web: curr, pct, src = c_web, p_web, s_web m1, m2 = st.columns(2) m1.metric(T['metric_price'], f"{curr:,.0f}" if isinstance(curr, (int, float)) else curr, f"{pct:.2f}%") m2.metric(T['metric_source'], src) if not h.empty: st.plotly_chart(plot_candle_chart(h, name, src), use_container_width=True) else: st.warning(f"Chart N/A") with st.expander(T['expander_news']): st.text(news_data) # AI에게 줄 데이터에 분기 실적 추가 q_str = q_df.to_string() if not q_df.empty else "N/A" collected_data.append(f""" [TARGET {i+1}] {name} ({ticker}) - Price: {curr}, Change: {pct:.2f}% - Fundamentals: PER {fund_data['PER']}, PBR {fund_data['PBR']}, ROE {fund_data['ROE']} - Quarterly Trend: \n{q_str} - News: {news_data[:500]}... """) status.update(label="✅ OK", state="complete", expanded=False) st.divider() st.subheader(T['insight_title']) prompt = f""" [ROLE] Global Analyst. [MARKET CONTEXT] {market_context_str} [USER QUERY] "{query}" [DATA] {chr(10).join(collected_data)} [TASK] 1. Comparative Analysis based on price and FUNDAMENTALS (Valuation). 2. Analyze QUARTERLY TRENDS (Revenue/Income growing or shrinking?). 3. Verdict based on Market Context + Fundamentals + News. [LANG] {T['llm_lang_instruction']}. """ msg = [{"role": "user", "content": prompt}] stream = client.chat.completions.create(model="Qwen/Qwen2.5-72B-Instruct", messages=msg, stream=True) st.write_stream(parse_stream(stream)) else: st.error("AI Error") # --- 시장 스캐너 & 리포트 (Top 10 분기 실적 추가) --- elif menu == T['menu_scanner']: st.subheader(T['scanner_title']) st.caption(T['scanner_caption']) c1, c2 = st.columns([3, 1]) with c1: target = st.selectbox("Market", list(MARKET_SAMPLES.keys()), label_visibility="collapsed") with c2: scan = st.button(T['btn_scan']) if scan: tickers = MARKET_SAMPLES[target] results = [] bar = st.progress(0) try: batch_data = yf.download(tickers, period="5d", progress=False)['Close'] except: batch_data = pd.DataFrame() for i, t in enumerate(tickers): name_display = TICKER_NAMES.get(t, t) p, c, s = None, None, None if not batch_data.empty and t in batch_data.columns: series = batch_data[t].dropna() if len(series) >= 2: p, prev = series.iloc[-1], series.iloc[-2] c, s = ((p - prev)/prev)*100, "Yahoo (Batch)" if p is None: p, c, s = get_price_data_robust(t, name_display) if p is not None: results.append({T['col_name']: name_display, 'Ticker': t, T['col_price']: p, 'Change(%)': c, T['col_source']: s}) bar.progress((i+1)/len(tickers)) if results: df = pd.DataFrame(results).sort_values('Change(%)', ascending=False) st.plotly_chart(plot_bar_chart(df, T), use_container_width=True) # [NEW] Top 10 Gainers/Losers에 대해서만 재무 데이터 및 실적 추이 추가 로딩 df_gainers = df[df['Change(%)'] > 0].head(10).copy() df_losers = df[df['Change(%)'] < 0].sort_values('Change(%)', ascending=True).head(10).copy() # AI에게 줄 요약 텍스트를 담을 리스트 gainers_analysis_data = [] losers_analysis_data = [] def process_funds(sub_df, analysis_list): if sub_df.empty: return sub_df pers, pbrs = [], [] for idx, row in sub_df.iterrows(): t = row['Ticker'] f = get_fundamentals(t) # 분기 실적도 여기서 가져옴 pers.append(f['PER']) pbrs.append(f['PBR']) # AI에게 넘길 데이터 문자열 생성 (종목명, 등락률, PER, 분기 추이) q_data = "N/A" if not f['Quarterly'].empty: # 데이터프레임을 문자열로 (최근 2분기만 간략하게) q_data = f['Quarterly'].iloc[:, :2].to_string() analysis_list.append(f"{row[T['col_name']]} ({row['Change(%)']:.1f}%, PER:{f['PER']}) -> Quarterly:\n{q_data}") sub_df[T['col_per']] = pers sub_df[T['col_pbr']] = pbrs return sub_df with st.spinner("Analyzing fundamentals & quarterly trends for top movers..."): df_gainers = process_funds(df_gainers, gainers_analysis_data) df_losers = process_funds(df_losers, losers_analysis_data) disp_cols = [T['col_name'], 'Ticker', T['col_price'], 'Change(%)', T['col_per'], T['col_pbr'], T['col_source']] # AI Market Briefing st.divider() st.subheader(T['briefing_title']) prompt = f""" [ROLE] Chief Market Strategist. [TARGET MARKET] {target} [GLOBAL CONTEXT] {market_context_str} [TOP GAINERS DATA (With Quarterly Trends)] {chr(10).join(gainers_analysis_data)} [TOP LOSERS DATA (With Quarterly Trends)] {chr(10).join(losers_analysis_data)} [TASK] 1. **Market Summary**: Sentiment analysis. 2. **Fundamental Check**: Are the gainers actually making money? (Check Quarterly Revenue/Income trends). 3. **Top Pick**: Recommend 1 solid stock (Good price action + Good financials). 4. **Caution**: Warn about a risky stock (Bad financials). [OUTPUT LANGUAGE] **{T['llm_lang_instruction']}** (Professional tone). """ msg = [{"role": "user", "content": prompt}] stream = client.chat.completions.create(model="Qwen/Qwen2.5-72B-Instruct", messages=msg, stream=True) st.write_stream(parse_stream(stream)) st.divider() c_up, c_down = st.columns(2) with c_up: st.success(T['tab_gainers']) if not df_gainers.empty: st.dataframe(df_gainers[disp_cols].style.format({T['col_price']: "{:,.2f}", "Change(%)": "{:,.2f}%"}), use_container_width=True) else: st.info("No gainers.") with c_down: st.error(T['tab_losers']) if not df_losers.empty: st.dataframe(df_losers[disp_cols].style.format({T['col_price']: "{:,.2f}", "Change(%)": "{:,.2f}%"}), use_container_width=True) else: st.info("No losers.") else: st.warning(T['msg_fail'])