# -------- app_final.py -------- import os, json, math, pathlib, re, time, logging, requests from datetime import datetime, timedelta import numpy as np import pandas as pd import plotly.graph_objects as go import gradio as gr import openai import torch from sentence_transformers import SentenceTransformer, util # ────────────────────────── 0. API keys & Brave Search ────────────────────────── if "OPENAI_API_KEY" not in os.environ: os.environ["OPENAI_API_KEY"] = input("🔑 Enter your OpenAI API key: ").strip() openai.api_key = os.environ["OPENAI_API_KEY"] BRAVE_KEY = os.getenv("BRAVE_KEY", "") BRAVE_ENDPOINT = "https://api.search.brave.com/res/v1/web/search" logging.basicConfig(level=logging.INFO) # ────────────────────────── 1. Cycle config ────────────────────────── CENTER = 2025 CYCLES = { "K-Wave": 50, "Business": 9, "Finance": 80, "Hegemony": 250 } ORDERED_PERIODS = sorted(CYCLES.values()) COLOR = {9:"#66ff66", 50:"#ff3333", 80:"#ffcc00", 250:"#66ccff"} AMPL = {9:0.6, 50:1.0, 80:1.6, 250:4.0} PERIOD_BY_CYCLE = {k:v for k,v in CYCLES.items()} # ────────────────────────── 2. Load events JSON & embeddings ─────────────────── EVENTS_PATH = pathlib.Path(__file__).with_name("cycle_events.json") with open(EVENTS_PATH, encoding="utf-8") as f: RAW_EVENTS = json.load(f) EVENTS = {int(item["year"]): item["events"] for item in RAW_EVENTS} logging.info("Embedding historical events…") _embed_model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2") _all_sentences = [(yr, ev["event_en"]) for yr, evs in EVENTS.items() for ev in evs] _embeddings = _embed_model.encode([s for _, s in _all_sentences], convert_to_tensor=True) # 유사 사건 top-3 year 사전 SIMILAR_MAP = {} for idx, (yr, _) in enumerate(_all_sentences): scores = util.cos_sim(_embeddings[idx], _embeddings)[0] top_idx = torch.topk(scores, 4).indices.tolist() sims = [_all_sentences[i][0] for i in top_idx if _all_sentences[i][0] != yr][:3] SIMILAR_MAP.setdefault(yr, sims) # ────────────────────────── 3. Brave Search helpers ────────────────────────── def brave_search(query: str, count: int = 8, freshness_days: int | None = None): if not BRAVE_KEY: return [] params = {"q": query, "count": str(count)} if freshness_days: dt_from = (datetime.utcnow() - timedelta(days=freshness_days)).strftime("%Y-%m-%d") params["freshness"] = dt_from try: r = requests.get( BRAVE_ENDPOINT, headers={"Accept": "application/json", "X-Subscription-Token": BRAVE_KEY}, params=params, timeout=15 ) raw = r.json().get("web", {}).get("results") or [] return [{ "title": r.get("title", ""), "url": r.get("url", r.get("link", "")), "snippet": r.get("description", r.get("text", "")), "host": re.sub(r"https?://(www\.)?", "", r.get("url", "")).split("/")[0] } for r in raw[:count]] except Exception as e: logging.error(f"Brave error: {e}") return [] def format_search_results(query: str) -> str: rows = brave_search(query, 6, freshness_days=3) if not rows: return f"# [Web-Search] No live results for “{query}”.\n" hdr = f"# [Web-Search] Top results for “{query}” (last 3 days)\n\n" body = "\n".join( f"- **{r['title']}** ({r['host']})\n {r['snippet']}\n [link]({r['url']})" for r in rows ) return hdr + body + "\n" NEWS_KEYWORDS = { "Business": "recession OR GDP slowdown", "K-Wave": "breakthrough technology innovation", "Finance": "credit cycle debt crisis", "Hegemony": "great power rivalry geopolitics" } def fetch_cycle_news(): markers = [] for cyc, kw in NEWS_KEYWORDS.items(): res = brave_search(kw, 1, freshness_days=2) if res: markers.append({ "cycle": cyc, "title": res[0]["title"], "year": datetime.utcnow().year, "url": res[0]["url"] }) return markers NEWS_MARKERS = fetch_cycle_news() # ────────────────────────── 4. Chart helpers ────────────────────────── def half_sine(xs, period, amp): phase = np.mod(xs - CENTER, period) y = amp * np.sin(np.pi * phase / period) y[y < 0] = 0 return y def build_chart(start: int, end: int, lang: str = "KO"): xs = np.linspace(start, end, max(1000, (end - start) * 4)) fig = go.Figure() # Gradient towers for period in ORDERED_PERIODS: base, col = AMPL[period], COLOR[period] for frac in np.linspace(base / 30, base, 30): fig.add_trace(go.Scatter( x=xs, y=half_sine(xs, period, frac), mode="lines", line=dict(color=col, width=0.8), opacity=0.6, hoverinfo="skip", showlegend=False)) fig.add_trace(go.Scatter( x=xs, y=half_sine(xs, period, base), mode="lines", line=dict(color=col, width=1.6), hoverinfo="skip", showlegend=False)) # Events + similar text_key = "event_ko" if lang == "KO" else "event_en" for yr, evs in EVENTS.items(): if start <= yr <= end: cyc = evs[0]["cycle"] period = PERIOD_BY_CYCLE[cyc] yv = float(half_sine(np.array([yr]), period, AMPL[period])) sim = ", ".join(map(str, SIMILAR_MAP.get(yr, []))) or "None" txt = "
".join(e[text_key] for e in evs) fig.add_trace(go.Scatter( x=[yr], y=[yv], mode="markers", marker=dict(color="white", size=6), customdata=[[cyc, txt, sim]], hovertemplate=( "Year %{x} • %{customdata[0]}
" "%{customdata[1]}
" "Similar: %{customdata[2]}" ), showlegend=False)) # Live-news markers for m in NEWS_MARKERS: if start <= m["year"] <= end: p = PERIOD_BY_CYCLE[m["cycle"]] yv = float(half_sine(np.array([m["year"]]), p, AMPL[p])) * 1.05 fig.add_trace(go.Scatter( x=[m["year"]], y=[yv], mode="markers+text", marker=dict(color="gold", size=8, symbol="star"), text=["📰"], textposition="top center", customdata=[[m["cycle"], m["title"], m["url"]]], hovertemplate=("Live news • %{customdata[0]}
" "%{customdata[1]}"), showlegend=False)) # Hover Year trace fig.add_trace(go.Scatter( x=xs, y=np.full_like(xs, -0.05), mode="lines", line=dict(color="rgba(0,0,0,0)", width=1), hovertemplate="Year %{x:.0f}", showlegend=False)) # Cosmetics fig.add_vline(x=CENTER, line_dash="dash", line_color="white", opacity=0.6) arrow_y = AMPL[250] * 1.05 fig.add_annotation( x=CENTER - 125, y=arrow_y, ax=CENTER + 125, ay=arrow_y, xref="x", yref="y", axref="x", ayref="y", showarrow=True, arrowhead=3, arrowsize=1, arrowwidth=1.2, arrowcolor="white") fig.add_annotation( x=CENTER, y=arrow_y + 0.15, text="250 yr", showarrow=False, font=dict(color="white", size=10)) fig.update_layout( template="plotly_dark", paper_bgcolor="black", plot_bgcolor="black", height=500, margin=dict(t=30, l=40, r=40, b=40), hoverlabel=dict(bgcolor="#222", font_size=11), hovermode="x") fig.update_xaxes(title="Year", range=[start, end], showgrid=False) fig.update_yaxes(title="Relative amplitude", showticklabels=False, showgrid=False) summary = f"Range {start}-{end} | Events: {sum(1 for y in EVENTS if start <= y <= end)}" return fig, summary # ────────────────────────── 5. GPT helper ────────────────────────── BASE_PROMPT = ( "당신은 **CycleNavigator AI**로, 경제사·국제정치·장주기(9y Business, 50y K-Wave, " "80y Finance, 250y Hegemony) 분석에 정통한 전문가입니다. " "모든 답변은 한국어로 하되 학술적 정확성과 실무적 명료성을 동시에 갖추십시오. " "✦ 답변 구조 지침: ① 질문 핵심 요약 → ② 4대 주기와의 관련성 명시 → " "③ 역사·데이터 근거 설명 → ④ 시사점·전망 순으로 서술하며, " "번호‧글머리표·짧은 문단을 활용해 논리적으로 배열합니다. " "✦ 제공된 [Chart summary]는 반드시 해석·인용하고, " "객관적 사실·연도·사건을 근거로 합니다. " "✦ 근거가 불충분할 땐 ‘확실하지 않습니다’라고 명시해 추측을 피하십시오. " "✦ 불필요한 장황함은 삼가고 3개 단락 또는 7개 이하 bullets 내로 요약하십시오." ) def chat_with_gpt(hist, msg, chart_summary): msgs = [{"role": "system", "content": BASE_PROMPT}] if chart_summary not in ("", "No chart yet."): msgs.append({"role": "system", "content": f"[Chart summary]\n{chart_summary}"}) for u, a in hist: msgs.extend([{"role": "user", "content": u}, {"role": "assistant", "content": a}]) msgs.append({"role": "user", "content": msg}) return openai.chat.completions.create( model="gpt-3.5-turbo", messages=msgs, max_tokens=600, temperature=0.7 ).choices[0].message.content.strip() # ────────────────────────── 6. Gradio UI ────────────────────────── def create_app(): with gr.Blocks( theme=gr.themes.Soft(), css=""" #discord-badge{position:fixed; bottom:10px; left:50%; transform:translateX(-50%);} """ ) as demo: gr.Markdown("## 🔭 **CycleNavigator (Interactive)**") gr.Markdown( "" "Interactive visual service delivering insights at a glance into the four major long cycles— " "Business 9y (credit-investment business cycle) • " "K-Wave 50y (long technological-industrial wave) • " "Finance 80y (long credit-debt cycle) • " "Hegemony 250y (rise & fall of global powers cycle)" " —through dynamic charts and AI chat." "" ) # ── 언어 선택 ─────────────────────────────────────────── # ── 언어 선택 ─────────────────────────────────────────── lang_state = gr.State(value="EN") # ① 기본을 EN으로 lang_radio = gr.Radio( ["English", "한국어"], value="English", label="Language / 언어", interactive=True ) # 초기 차트·상태 ───────────────────────────────────────── chart_summary_state = gr.State(value="No chart yet.") fig0, summ0 = build_chart(1500, 2500, lang_state.value) # ② 동일 상태값 사용 plot = gr.Plot(value=fig0) chart_summary_state.value = summ0 with gr.Row(): start_year = gr.Number(label="Start Year", value=1500, precision=0) end_year = gr.Number(label="End Year", value=2500, precision=0) zoom_in = gr.Button("🔍 Zoom In") zoom_out = gr.Button("🔎 Zoom Out") # ── functions ── def refresh(s, e, lang_code): fig, summ = build_chart(int(s), int(e), lang_code) return fig, summ def zoom(s, e, f, lang_code): mid = (s + e) / 2 span = (e - s) * f / 2 ns, ne = int(mid - span), int(mid + span) fig, summ = build_chart(ns, ne, lang_code) return ns, ne, fig, summ def change_lang(lang_label, s, e): code = "KO" if lang_label == "한국어" else "EN" fig, summ = build_chart(int(s), int(e), code) return code, fig, summ # ── event wiring ── start_year.change(refresh, [start_year, end_year, lang_state], [plot, chart_summary_state]) end_year.change(refresh, [start_year, end_year, lang_state], [plot, chart_summary_state]) zoom_in.click( lambda s, e, lc: zoom(s, e, 0.5, lc), [start_year, end_year, lang_state], [start_year, end_year, plot, chart_summary_state] ) zoom_out.click( lambda s, e, lc: zoom(s, e, 2.0, lc), [start_year, end_year, lang_state], [start_year, end_year, plot, chart_summary_state] ) lang_radio.change( change_lang, [lang_radio, start_year, end_year], [lang_state, plot, chart_summary_state] ) gr.File(value=str(EVENTS_PATH), label="Download cycle_events.json") with gr.Tabs(): with gr.TabItem("Deep Research Chat"): chatbot = gr.Chatbot(label="Assistant") user_input = gr.Textbox(lines=3, placeholder="메시지를 입력하세요…") with gr.Row(): send_btn = gr.Button("Send", variant="primary") web_btn = gr.Button("🔎 Web Search") def respond(hist, msg, summ): reply = chat_with_gpt(hist, msg, summ) hist.append((msg, reply)) return hist, gr.Textbox(value="") def respond_ws(hist, msg, summ): md = format_search_results(msg) reply = chat_with_gpt(hist, f"{msg}\n\n{md}", summ) hist.append((f"{msg}\n\n(웹검색)", reply)) return hist, gr.Textbox(value="") send_btn.click(respond, [chatbot, user_input, chart_summary_state], [chatbot, user_input]) user_input.submit(respond, [chatbot, user_input, chart_summary_state], [chatbot, user_input]) web_btn.click(respond_ws, [chatbot, user_input, chart_summary_state], [chatbot, user_input]) # ── fixed Discord badge ── gr.HTML( '' 'badge' ) return demo # ────────────────────────── main ────────────────────────── if __name__ == "__main__": create_app().launch()