"""노트북 1번 맨 위에 들어갈 셀 코드 — 오늘 그룹 강세/약세 표 + TOP/BOTTOM 5. 이 모듈을 import 하고 render() 만 호출하면 됨. 노트북 셀은 짧게 유지.""" from __future__ import annotations import json import logging import concurrent.futures as _cf import sys import warnings from datetime import datetime from pathlib import Path import pandas as pd import yfinance as yf from IPython.display import HTML, display try: from taebuji_common import yc_close_wide except Exception: yc_close_wide = None warnings.filterwarnings("ignore") logging.getLogger("yfinance").setLevel(logging.CRITICAL) CFG_PATH = Path(__file__).parent / "finviz_chart_items.json" TAXONOMY_PATH = Path(__file__).parent / "sector_taxonomy.json" OUTPUT_PATH = Path(__file__).parent / "us_overnight_sector_latest.json" YF_CACHE_PATH = Path(__file__).parent / ".cache" / "yfinance" YF_BATCH_SIZE = 25 YF_BATCH_PARALLEL = 3 YF_TIMEOUT = 8 YF_CACHE_PATH.mkdir(parents=True, exist_ok=True) try: yf.set_tz_cache_location(str(YF_CACHE_PATH)) except Exception: pass try: import multitasking multitasking.set_max_threads(1) except Exception: pass # v1: finviz 그룹의 잡탕 "통화/채권(231)" 제외 # v2: sector_taxonomy.json 사용 시 의미 있는 분류라 모두 표시. 단 N<2 작은 그룹은 평균 노이즈. EXCLUDE_GROUPS: set[str] = {"기타 / 미분류", "특수 / SPAC"} MIN_GROUP_N = 2 # 표/AI 분석에서 제외할 최소 종목 수 # LLM (Codex OAuth GPT-5.6 Terra) — DART RS 파이프라인과 동일 인프라 재사용 LLM_MODEL = "gpt-5.6-terra" OAUTH_HELPER_PATH = str(Path.home() / "openai-codex-oauth-helper" / "codex_oauth.py") LLM_SERVICES_PATH = str(Path.home() / "valuechain-earnings-app" / "backend" / "app" / "services") def _load_universe() -> tuple[dict, dict, list[str]]: """ticker → group(재분류) / ticker → caption / 정렬된 ticker list. sector_taxonomy.json 이 있으면 그 매핑(키워드 기반 24+ 그룹) 사용, 없으면 finviz_chart_items.json 의 원본 12 그룹 fallback.""" if TAXONOMY_PATH.exists(): tx = json.loads(TAXONOMY_PATH.read_text(encoding="utf-8")) t2g = dict(tx.get("ticker_to_group") or {}) t2cap = dict(tx.get("captions") or {}) if t2g: return t2g, t2cap, sorted(t2g.keys()) # fallback — finviz 원본 cfg = json.loads(CFG_PATH.read_text(encoding="utf-8")) t2g: dict[str, str] = {} t2cap: dict[str, str] = {} for g, items in cfg.items(): for item in items: t = item["ticker"] t2g.setdefault(t, g) t2cap.setdefault(t, item.get("caption", t)) return t2g, t2cap, sorted(t2g.keys()) def _chunks(values: list[str], size: int): for i in range(0, len(values), size): yield values[i:i + size] def _fetch_intraday_change(tickers: list[str]) -> pd.DataFrame: """오늘 vs 전일 종가 % 변화. 장중이면 마지막 가격 기준.""" rows = [] if yc_close_wide is not None: try: close = yc_close_wide( tickers, range_str="5d", interval="1d", min_coverage=0, ffill=False, adjusted=True, ) except Exception: close = pd.DataFrame() if close is not None and not close.empty: for t in tickers: if t not in close.columns: continue s = close[t].dropna() if len(s) < 2: continue last = float(s.iloc[-1]) prev = float(s.iloc[-2]) if prev == 0: continue rows.append({"ticker": t, "r1d": (last / prev - 1) * 100, "price": last, "session_date": str(s.index[-1])[:10]}) return pd.DataFrame(rows) def _download_batch(batch: list[str]): df = yf.download( batch, period="5d", interval="1d", auto_adjust=True, progress=False, threads=False, group_by="ticker", timeout=YF_TIMEOUT, ) return batch, df batches = list(_chunks(tickers, YF_BATCH_SIZE)) with _cf.ThreadPoolExecutor(max_workers=YF_BATCH_PARALLEL) as ex: futures = [ex.submit(_download_batch, batch) for batch in batches] for fut in _cf.as_completed(futures): try: batch, df = fut.result() except Exception: continue if df is None or df.empty: continue for t in batch: try: s = df[t]["Close"].dropna() if isinstance(df.columns, pd.MultiIndex) \ else df["Close"].dropna() except Exception: continue if len(s) < 2: continue last = float(s.iloc[-1]) prev = float(s.iloc[-2]) if prev == 0: continue rows.append({"ticker": t, "r1d": (last / prev - 1) * 100, "price": last, "session_date": str(s.index[-1])[:10]}) return pd.DataFrame(rows) def _color(v: float) -> str: """한국식: + 상승 = 빨강, − 하락 = 파랑.""" if v > 0: return "#ff4757" if v < 0: return "#3b82f6" return "#999" def _arrow(v: float) -> str: return "▲" if v > 0 else ("▼" if v < 0 else "─") _LLM_SCHEMA = { "type": "object", "additionalProperties": False, "required": ["one_line", "regime", "rotation", "actions"], "properties": { "one_line": { "type": "string", "description": "오늘 시장의 한 줄 진단 (한국어, 명사형 종결, 30자 내외)", }, "regime": { "type": "string", "enum": ["RISK_ON", "NEUTRAL", "RISK_OFF"], "description": "오늘 단일일 리스크 온/오프 판정 (그룹 평균 + 상승비율 종합)", }, "rotation": { "type": "array", "minItems": 2, "maxItems": 4, "items": {"type": "string", "description": "섹터 자금 이동 해석 한 줄 (40자 내외)"}, }, "actions": { "type": "array", "minItems": 3, "maxItems": 5, "items": { "type": "object", "additionalProperties": False, "required": ["type", "target", "reason"], "properties": { "type": {"type": "string", "enum": ["BUY", "WATCH", "AVOID"]}, "target": {"type": "string", "description": "구체 티커 또는 그룹명"}, "reason": {"type": "string", "description": "왜 (한국어, 50자 내외)"}, }, }, }, }, } _LLM_SYSTEM = ( "당신은 미국 주식·ETF 흐름을 단기 모멘텀 + 섹터 로테이션 관점에서 해석하는 트레이더 어시스턴트다. " "입력은 오늘 하루 그룹별 평균 수익률, 종목 TOP/BOTTOM 5다. " "한국어로 단정적·명사형 종결로 답한다 (~함/~임/~수준). " "한국어 외 영어/특수문자 최소화. 추측·과장 금지. 데이터에 근거한 해석만." ) def _llm_analysis(market_avg: float, gs: pd.DataFrame, top5: pd.DataFrame, bot5: pd.DataFrame, t2cap: dict) -> dict | None: """OAuth GPT-5.4 로 시장 진단 + 섹터 로테이션 + 액션 3가지 생성. 실패 시 None.""" if LLM_SERVICES_PATH not in sys.path: sys.path.insert(0, LLM_SERVICES_PATH) try: from codex_oauth_responses import CodexOAuthResponsesClient # type: ignore except Exception: return None gs_lines = [ f"- {g}: 평균 {r['avg_r1d']:+.2f}%, 상승비율 {int(r['bull_pct'])}%, N={int(r['n'])}" for g, r in gs.iterrows() ] top_lines = [ f"- {r['ticker']} ({t2cap.get(r['ticker'], '')[:35]}): {r['r1d']:+.2f}%" for _, r in top5.iterrows() ] bot_lines = [ f"- {r['ticker']} ({t2cap.get(r['ticker'], '')[:35]}): {r['r1d']:+.2f}%" for _, r in bot5.iterrows() ] user_prompt = f"""# 오늘 미국장 데이터 (yfinance) ## 그룹 평균 r1d (강한 순) {chr(10).join(gs_lines)} ## 종목 TOP 5 {chr(10).join(top_lines)} ## 종목 BOTTOM 5 {chr(10).join(bot_lines)} 위 데이터를 바탕으로 응답: 1. one_line: 시장 한 줄 진단 2. regime: RISK_ON / NEUTRAL / RISK_OFF 3. rotation: 어디서 어디로 자금이 움직였는가 2~4 줄 (섹터·테마 단위) 4. actions: 오늘 BUY / WATCH / AVOID 각각 1개 이상, 총 3~5개. target 은 티커 또는 그룹명, reason 은 데이터 근거. """ try: client = CodexOAuthResponsesClient(OAUTH_HELPER_PATH) return client.create_structured_response( model=LLM_MODEL, system_prompt=_LLM_SYSTEM, user_prompt=user_prompt, schema_name="intraday_strength_analysis", schema=_LLM_SCHEMA, ) except Exception as exc: return {"_error": str(exc)[:300]} _REGIME_STYLE = { "RISK_ON": ("🟢 RISK ON", "#00d4aa"), "NEUTRAL": ("🟡 NEUTRAL", "#ff9500"), "RISK_OFF": ("🔴 RISK OFF", "#3b82f6"), } _ACTION_STYLE = { "BUY": ("▲ BUY", "#ff4757"), "WATCH": ("◆ WATCH", "#ff9500"), "AVOID": ("✕ AVOID", "#3b82f6"), } def _build_ai_html(analysis: dict | None) -> str: if analysis is None: return ( '
' '⚠️ AI 분석 모듈 (codex_oauth_responses) 로드 실패 — 표만 표시.' '
' ) if "_error" in analysis: return ( f'
' f'⚠️ AI 분석 호출 실패: {analysis["_error"]}' f'
' ) one_line = analysis.get("one_line", "") regime = analysis.get("regime", "NEUTRAL") rotation = analysis.get("rotation") or [] actions = analysis.get("actions") or [] regime_label, regime_color = _REGIME_STYLE.get(regime, _REGIME_STYLE["NEUTRAL"]) rot_lis = "".join( f'
  • {r}
    • ' for r in rotation ) action_chips = [] for a in actions: atype = a.get("type", "WATCH") label, color = _ACTION_STYLE.get(atype, _ACTION_STYLE["WATCH"]) target = a.get("target", "") reason = a.get("reason", "") action_chips.append( f'
    ' f'{label} ' f'{target}
    ' f'{reason}' f'
    ' ) return f"""
    AI BRIEFING · GPT-5.4 {regime_label}
    {one_line}
    ROTATION
    TODAY ACTIONS
    {"".join(action_chips)}
    """ def _build_html(group_summary: pd.DataFrame, top5: pd.DataFrame, bot5: pd.DataFrame, t2cap: dict, ts: str) -> str: rows_html = [] for g, row in group_summary.iterrows(): c = _color(row["avg_r1d"]) rows_html.append( f'' f'{g}' f'' f'{_arrow(row["avg_r1d"])} {row["avg_r1d"]:+.2f}%' f'' f'{int(row["bull_pct"])}%' f'' f'N={int(row["n"])}' ) def chips(df: pd.DataFrame) -> str: out = [] for _, r in df.iterrows(): c = _color(r["r1d"]) cap = t2cap.get(r["ticker"], r["ticker"])[:35] out.append( f'' f'{r["ticker"]} ' f'{r["r1d"]:+.2f}% ' f'{cap}' ) return "".join(out) return f"""

    🔥 오늘 그룹 강세 · 약세 ({ts}) · 한국식: + 빨강 / − 파랑

    {"".join(rows_html)}
    그룹 오늘 평균 상승비율 N
    ▲ 오늘 TOP 5 종목
    {chips(top5)}
    ▼ 오늘 BOTTOM 5 종목
    {chips(bot5)}
    """ def render(skip_ai: bool = False) -> None: """노트북 셀에서 한 줄로 호출. AI 분석 박스(맨 위) + 표(아래).""" t2g, t2cap, tickers = _load_universe() result = _fetch_intraday_change(tickers) if result.empty: display(HTML("

    데이터 fetch 실패

    ")) return result["group"] = result["ticker"].map(t2g) priority = result[~result["group"].isin(EXCLUDE_GROUPS)] group_summary = ( priority.groupby("group") .agg( n=("ticker", "count"), avg_r1d=("r1d", "mean"), bull_pct=("r1d", lambda x: (x > 0).sum() / len(x) * 100), ) ) # n=1 그룹은 평균이 곧 종목 자체 → 제외 (TOP/BOTTOM 종목 칸에서 보임) group_summary = ( group_summary[group_summary["n"] >= MIN_GROUP_N] .sort_values("avg_r1d", ascending=False) ) top5 = result.sort_values("r1d", ascending=False).head(5) bot5 = result.sort_values("r1d").head(5) market_avg = float(priority["r1d"].mean()) ts = datetime.now().strftime("%Y-%m-%d %H:%M KST") # AI 분석 (먼저) — skip_ai True 면 스킵 analysis = None if not skip_ai: analysis = _llm_analysis(market_avg, group_summary, top5, bot5, t2cap) display(HTML(_build_ai_html(analysis))) # 후보 랭킹 엔진이 전일 미국장 섹터 강도를 재사용할 수 있도록 결정론적 # 원자료를 남긴다. LLM 실패 여부와 관계없이 수익률·상승비율은 항상 저장한다. session_dates = result.get("session_date") as_of = None if session_dates is not None: values = [str(value)[:10] for value in session_dates.dropna().tolist() if value] as_of = max(values) if values else None groups = [] for rank, (group, row) in enumerate(group_summary.iterrows(), start=1): groups.append({ "rank": rank, "group": str(group), "n": int(row["n"]), "avg_r1d": round(float(row["avg_r1d"]), 4), "bull_pct": round(float(row["bull_pct"]), 2), }) snapshot = { "generated_at": datetime.now().astimezone().isoformat(timespec="seconds"), "as_of": as_of, "source": "태버지 1번 전일 미국장 정리 · yfinance adjusted close", "method": "미국 분류군별 1일 평균 수익률과 상승 종목 비율; N<2 제외", "market_avg_r1d": round(market_avg, 4), "groups": groups, "top5": [ {"ticker": str(row["ticker"]), "r1d": round(float(row["r1d"]), 4)} for _, row in top5.iterrows() ], "bottom5": [ {"ticker": str(row["ticker"]), "r1d": round(float(row["r1d"]), 4)} for _, row in bot5.iterrows() ], "analysis": analysis if isinstance(analysis, dict) else None, } OUTPUT_PATH.write_text( json.dumps(snapshot, ensure_ascii=False, indent=2), encoding="utf-8" ) # 표 (아래) display(HTML(_build_html(group_summary, top5, bot5, t2cap, ts))) if __name__ == "__main__": render()