Update app.py

app.py

@@ -1,145 +1,134 @@
 # -*- coding: utf-8 -*-
 """
-주식 디버거 (모바일 웹앱) — 3개 분석 / Hugging Face Spaces 배포용
-
-탭(분석):
-  1) VCP 돌파 감지   (vcp_debug.py)          ← 날짜 자동: 최근 1년
-  2) 볼린저 카운트   (bollinger_count_debug.py) ← 날짜 자동: 최근 1년
-  3) 전량매도 최후보루 (final_sell_monitor.py)   ← 최종 확정 버전
-
-원칙: 데스크탑 스크립트의 로직/출력을 그대로 재현.
-  - 함수형(analyze_one) 스크립트: import 후 함수 호출
-  - 최상위 실행형(VCP·볼린저): 소스를 읽어 TICKER만 바꿔 exec (sys.exit 가드)
-  모든 출력은 contextlib.redirect_stdout 로 그대로 캡처해 화면에 표시.
-
-HF Spaces(한국 외 IP)에서도 네이버 주가/지수 데이터는 정상 로드됨.
-KRX(StockListing)만 막히므로 시장(KOSPI/KOSDAQ)은 직접 선택.
+전량매도 최후보루 - 모바일 웹 모니터 (Hugging Face Spaces / Gradio)
+==================================================================
+confirm_sell_backtest.py 의 핵심 로직을 그대로 import (파리티 100%).
+시장(KOSPI/KOSDAQ)은 market_map.csv 로 자동판별 (KRX StockListing 호출 안 함).
+
+HF Space 에 같이 올릴 파일:
+  app.py
+  confirm_sell_backtest.py
+  market_map.csv         (code,market 두 컬럼)
+  requirements.txt
+
+참고: HF(해외 IP)에서도 네이버 주가/지수(fdr.DataReader)는 정상.
+      KRX StockListing 만 막히므로 시장은 csv 로 해결.
 """
 
-import io
 import os
-import re
-import contextlib
-from datetime import datetime, timedelta
-
 import gradio as gr
 
-# 함수형 스크립트는 import 해서 호출 (최상위에서 분석이 실행되지 않음)
-import final_sell_monitor
+import confirm_sell_backtest as BT   # 핵심 로직/설정 그대로
 
 HERE = os.path.dirname(os.path.abspath(__file__))
-BOLL_PATH = os.path.join(HERE, "bollinger_count_debug.py")
-VCP_PATH  = os.path.join(HERE, "vcp_debug.py")
+MAP_PATH = os.path.join(HERE, "market_map.csv")
 
-_TICKER_RE = re.compile(r"TICKER\s*=\s*['\"][0-9A-Za-z]+['\"]")
+ALERT_DAYS = 5   # 최근 이 거래일 안의 매도신호를 [매도]로 본다
 
 
-def _safe(fn, *args):
-    """어떤 분석이든 예외가 나도 빨간 '오류' 대신 이유를 글로 반환."""
+# ---------- market_map.csv 로드 (code -> market) ----------
+def load_market_map():
+    m = {}
     try:
-        return fn(*args)
+        with open(MAP_PATH, encoding="utf-8") as f:
+            for i, line in enumerate(f):
+                parts = line.strip().split(",")
+                if len(parts) < 2:
+                    continue
+                code, mkt = parts[0].strip(), parts[1].strip()
+                if code.lower() == "code":      # 헤더 스킵
+                    continue
+                m[code.zfill(6)] = mkt.upper()
     except Exception as e:
-        msg = str(e).strip().splitlines()[0] if str(e).strip() else type(e).__name__
-        return (f"[오류] {msg[:200]}\n"
-                f"종목코드가 맞는지, 잠시 후 다시 시도해 보세요.\n"
-                f"(데이터 제공처(네이버) 일시 오류이거나, 신규/정지/폐지 종목일 수 있습니다.)")
-
-
-def _exec_script(path, ticker, extra_subs=None):
-    """최상위 실행형 스크립트: TICKER(및 추가 치환)만 바꿔 exec, 출력 캡처."""
-    buf = io.StringIO()
-    ns = {"__name__": "__sandbox__"}   # __main__ 가드 비활성
-    with contextlib.redirect_stdout(buf):
-        try:
-            with open(path, encoding="utf-8") as f:
-                src = f.read()
-            src = _TICKER_RE.sub(f"TICKER = '{ticker}'", src, count=1)
-            if extra_subs:
-                for pat, repl in extra_subs:
-                    src = re.sub(pat, repl, src, count=1)
-            exec(compile(src, path, "exec"), ns)
-        except SystemExit:
-            pass                       # 스크립트의 sys.exit() 가 앱을 죽이지 않게
-        except Exception as e:
-            msg = str(e).strip().splitlines()[0] if str(e).strip() else type(e).__name__
-            print(f"\n[오류] {msg[:200]}")
-    return buf.getvalue() or "출력 없음"
-
-
-def _call_func(fn, *args):
-    """함수형 스크립트: 함수 호출하며 출력 캡처."""
-    buf = io.StringIO()
-    with contextlib.redirect_stdout(buf):
-        try:
-            fn(*args)
-        except SystemExit:
-            pass
-        except Exception as e:
-            msg = str(e).strip().splitlines()[0] if str(e).strip() else type(e).__name__
-            print(f"\n[오류] {msg[:200]}")
-            print("종목코드와 시장(KOSDAQ/KOSPI)이 맞는지 확인하세요.")
-    return buf.getvalue() or "출력 없음"
-
-
-# ---------- 분석별 실행 ----------
-def run_bollinger(ticker):
-    ticker = (ticker or "").strip()
-    if not ticker:
-        return "종목코드를 입력하세요 (예: 420770)"
-    today = datetime.today()
-    frm = (today - timedelta(days=365)).strftime("%Y-%m-%d")
-    to  = today.strftime("%Y-%m-%d")
-    subs = [
-        (r"DEBUG_FROM\s*=\s*['\"][0-9-]+['\"]", f"DEBUG_FROM = '{frm}'"),
-        (r"DEBUG_TO\s*=\s*['\"][0-9-]+['\"]",   f"DEBUG_TO = '{to}'"),
-    ]
-    return _exec_script(BOLL_PATH, ticker, subs)
-
-
-def run_vcp(ticker):
-    ticker = (ticker or "").strip()
-    if not ticker:
-        return "종목코드를 입력하세요 (예: 007340)"
-    return _exec_script(VCP_PATH, ticker)
-
-
-def run_final(ticker, name):
-    ticker = (ticker or "").strip()
-    if not ticker:
-        return "종목코드를 입력하세요 (예: 195940)"
-    # 시장은 자동판별 (수동 선택 없음 -> 잘못 고를 일 없음)
-    h = {"ticker": ticker, "name": (name or "").strip() or None}
-    return _call_func(final_sell_monitor.analyze_one, h)
-
-
-# ---------- UI ----------
-with gr.Blocks(title="주식 디버거") as demo:
-    gr.Markdown("## 주식 디버거 (모바일)\n분석을 고르고 종목코드만 넣으면 데스크탑과 동일한 출력이 나옵니다.")
-
-    with gr.Tab("볼린저 카운트"):
-        b_tk = gr.Textbox(label="종목코드", placeholder="예: 420770")
-        b_btn = gr.Button("실행", variant="primary")
-        b_out = gr.Textbox(label="결과 (최근 1년 자동)", lines=20)
-        b_btn.click(lambda tk: _safe(run_bollinger, tk), inputs=b_tk, outputs=b_out)
-
-    with gr.Tab("VCP 돌파"):
-        v_tk = gr.Textbox(label="종목코드", placeholder="예: 007340")
-        v_btn = gr.Button("실행", variant="primary")
-        v_out = gr.Textbox(label="결과 (최근 1년 자동)", lines=20)
-        v_btn.click(lambda tk: _safe(run_vcp, tk), inputs=v_tk, outputs=v_out)
-
-    with gr.Tab("전량매도 최후보루"):
-        f_tk = gr.Textbox(label="종목코드", placeholder="예: 195940")
-        f_nm = gr.Textbox(label="종목명(선택)", placeholder="예: HK이노엔")
-        gr.Markdown("※ 시장(KOSPI/KOSDAQ)은 종목코드로 자동 판별됩니다.")
-        f_btn = gr.Button("실행", variant="primary")
-        f_out = gr.Textbox(label="결과", lines=16)
-        f_btn.click(lambda a, c: _safe(run_final, a, c), inputs=[f_tk, f_nm], outputs=f_out)
-
+        print(f"[경고] market_map.csv 로드 실패: {e}")
+    return m
+
+MARKET_MAP = load_market_map()
+
+
+def market_of(code):
+    return MARKET_MAP.get(code.zfill(6), "KOSDAQ")   # 못 찾으면 코스닥 가정
+
+
+# ---------- 한 종목 점검 (백테스터 로직 그대로) ----------
+def check_one(code):
+    code = code.zfill(6)
+    market = market_of(code)
+    r = BT.load_one(code, market)
+    if r is None:
+        return {"code": code, "market": market, "ok": False, "msg": "데이터 없음"}
+    stock = r["stock"]; index = r["index"]
+    sells, jumps, steps = BT.find_confirmed_sells(
+        stock, index, r["dates"], r["start_pos"],
+        ma60=r.get("ma60"), ma240=r.get("ma240"),
+        low=r.get("low"), bbl=r.get("bbl"),
+        ma240_est=r.get("ma240_est"), ma120=r.get("ma120"))
+    last_i = len(stock) - 1
+    recent = [s for s in sells if s["i"] >= last_i - ALERT_DAYS]
+    return {
+        "code": code, "market": market, "ok": True,
+        "last_date": r["dates"][last_i],
+        "price_now": float(stock[last_i]),
+        "recent": recent,
+        "last_sell": sells[-1] if sells else None,
+    }
+
+
+def _band(lo, hi):
+    return f"≤{hi*100:.0f}%" if lo <= -1 else f"({lo*100:.0f},{hi*100:.0f}]%"
+
+
+def run_monitor(text):
+    # 입력 파싱: 줄바꿈/쉼표/공백 다 허용
+    codes = [c.strip() for c in text.replace(",", " ").replace("\n", " ").split()]
+    codes = [c for c in codes if c]
+    if not codes:
+        return "종목코드를 입력해줘 (예: 277810 373220)"
+
+    head = (f"**전량매도 최후보루 모니터** · 최근 {ALERT_DAYS}거래일 기준\n\n"
+            f"기준: 돌파후{BT.RUN_MIN_DAYS}일 AND A-4평평 "
+            f"60{_band(BT.FLAT_MIN_60, BT.FLAT_MAX_60)} {BT.FLAT_COMBINE} "
+            f"120{_band(BT.FLAT_MIN_120, BT.FLAT_MAX_120)} {BT.FLAT_COMBINE} "
+            f"240{_band(BT.FLAT_MIN_240, BT.FLAT_MAX_240)}\n\n---\n")
+
+    out = [head]
+    alerts = []
+    for code in codes:
+        res = check_one(code)
+        if not res["ok"]:
+            out.append(f"- `{code}` ({res['market']}) — {res['msg']}")
+            continue
+        if res["recent"]:
+            s = res["recent"][-1]
+            alerts.append(code)
+            est = " `[240추정]`" if s.get("est") else ""
+            out.append(
+                f"### 🔴 [매도] {code} ({res['market']}){est}\n"
+                f"- 현재가 **{res['price_now']:,.0f}**\n"
+                f"- 신호일 {s['date']} @ {s['price']:,.0f} (이격 {s['div']*100:.0f}%)\n"
+                f"- 사유 `{s.get('trend','')}`\n")
+        else:
+            ls = res["last_sell"]
+            tail = f" · 마지막 매도신호 {ls['date']}" if ls else " · 과거 매도신호 없음"
+            out.append(f"### 🟢 [보유] {code} ({res['market']})\n"
+                       f"- 현재가 {res['price_now']:,.0f}{tail}\n")
+
+    out.append("\n---\n")
+    if alerts:
+        out.append(f"**오늘 확인 필요: {len(alerts)}종목 → {', '.join(alerts)}**")
+    else:
+        out.append("**최근 매도신호 없음. 전부 보유 유지.**")
+    return "\n".join(out)
+
+
+with gr.Blocks(title="전량매도 모니터") as demo:
+    gr.Markdown("# 📉 전량매도 최후보루 모니터")
+    gr.Markdown("보유종목 코드를 넣으면 최근 매도신호 여부를 알려줘. (시장 자동판별)")
+    inp = gr.Textbox(label="종목코드 (여러 개는 공백/줄바꿈/쉼표로)",
+                     value="277810 373220", lines=3)
+    btn = gr.Button("매도신호 확인", variant="primary")
+    outp = gr.Markdown()
+    btn.click(run_monitor, inputs=inp, outputs=outp)
 
 if __name__ == "__main__":
-    # HF Spaces 표준 실행 (포트 7860 고정) — 로그인 없음
-    demo.launch(
-        server_name="0.0.0.0",
-        server_port=7860,
-    )
+    demo.launch()