app.py

# -*- coding: utf-8 -*-
"""
주식 디버거 (모바일 웹앱) — Hugging Face Spaces / Gradio
=========================================================
탭 순서:
  1) 전량매도 최후보루(다종목) — confirm_sell_backtest.py 로직 그대로
  2) 볼린저 카운트            (bollinger_count_debug.py) — 최근 1년 자동
  3) 일봉 상태               — buy_engine_dipfix.run 후 be.LAST_STATUS (일봉 현 국면 한 줄)
  4) 삼성·ETF 매도 (RS)       — final_sell_monitor_rs.py (RS_A4, 위아래 억제)

원칙:
  - 전량매도: confirm_sell_backtest.py 를 import (백테스터와 100% 동일 로직)
  - 볼린저: 데스크탑 스크립트 소스를 읽어 TICKER만 바꿔 exec, 출력 그대로 캡처
  - RS 매도: final_sell_monitor_rs.analyze_one 호출 (삼성·ETF처럼 이격으로 안 잡히는 종목)
  - 일봉 상태: buy_engine_dipfix.run(df,'1day') 실행 후 be.LAST_STATUS 표시 (일봉만)
  - 시장(KOSPI/KOSDAQ): market_map.csv 로 자동판별 (KRX StockListing 안 부름)

같이 올릴 파일: app.py, confirm_sell_backtest.py, bollinger_count_debug.py,
               market_map.csv, requirements.txt,
               final_sell_monitor_rs.py, confirm_sell_backtest_rs.py, buy_engine_dipfix.py
"""

import io
import os
import re
import contextlib
from datetime import datetime, timezone, timedelta

import gradio as gr

import confirm_sell_backtest as BT   # 전량매도 핵심 로직/설정 그대로
import final_sell_monitor_rs         # RS 매도(삼성·ETF용 보조). analyze_one 호출

KST = timezone(timedelta(hours=9))


def asof_label(last_date):
    """마지막 데이터가 '오늘'이고 장중이면 '장중 현재가(HH:MM)', 아니면 '종가'. 한국 장 09:00~15:30 KST."""
    now = datetime.now(KST)
    today = now.strftime("%Y-%m-%d")
    intraday = (last_date == today
                and now.weekday() < 5
                and (9 * 60) <= (now.hour * 60 + now.minute) <= (15 * 60 + 30))
    if intraday:
        return f"{last_date} 장중 현재가 ({now:%H:%M} KST 기준, 종가 아님)"
    return f"{last_date} 종가 기준"

HERE = os.path.dirname(os.path.abspath(__file__))
MAP_PATH  = os.path.join(HERE, "market_map.csv")
BOLL_PATH = os.path.join(HERE, "bollinger_count_debug.py")

# ── 실행 시 market_map.csv 백그라운드 자동 갱신 (오래됐을 때만) ────────────────
#   build_market_map.py 를 찾아 --if-stale N --out <내 market_map.csv> 로 백그라운드 실행.
#   fdr/한국IP 없는 배포 서버에선 조용히 실패 → 커밋된 market_map.csv 그대로 사용(앱은 정상).
#   맥 로컬(fdr 됨)에선 앱 켤 때마다 N일 넘었으면 알아서 최신화. 맥 스케줄러와 이중 안전망.
MARKET_MAP_STALE_DAYS = 7   # 이 일수 넘으면 갱신 시도. 0이면 자동갱신 끔.

def _refresh_market_map_bg():
    import threading, subprocess, sys as _sys
    def _run():
        try:
            cands = [
                os.path.join(HERE, "build_market_map.py"),
                os.path.join(HERE, "..", "quant", "sell_debug", "build_market_map.py"),
                os.path.expanduser("~/Projects/quant/sell_debug/build_market_map.py"),
            ]
            script = next((p for p in cands if os.path.exists(p)), None)
            if not script:
                return   # 배포 서버 등 생성기 없음 → 조용히 스킵
            subprocess.run(
                [_sys.executable, script, "--if-stale", str(MARKET_MAP_STALE_DAYS),
                 "--out", MAP_PATH],
                timeout=180, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        except Exception:
            pass   # fdr 실패/해외IP/타임아웃 등 전부 무시 — 앱 동작에 영향 없음
    if MARKET_MAP_STALE_DAYS:
        threading.Thread(target=_run, daemon=True).start()

_refresh_market_map_bg()
# ──────────────────────────────────────────────────────────────────────────────

ALERT_DAYS = 5   # 전량매도: 최근 이 거래일 안의 매도신호를 [매도]로 본다

_TICKER_RE = re.compile(r"TICKER\s*=\s*['\"][0-9A-Za-z]+['\"]")


# ====================================================================
# 공통: market_map.csv (code -> (market, name))  3컬럼: code,market,name
# ====================================================================
def load_market_map():
    m = {}
    try:
        with open(MAP_PATH, encoding="utf-8") as f:
            for line in f:
                parts = [p.strip() for p in line.strip().split(",")]
                if len(parts) < 2:
                    continue
                code, mkt = parts[0], parts[1]
                if code.lower() == "code":
                    continue
                name = parts[2] if len(parts) >= 3 else ""
                m[code.zfill(6)] = (mkt.upper(), name)
    except Exception as e:
        print(f"[경고] market_map.csv 로드 실패: {e}")
    return m

MARKET_MAP = load_market_map()

import difflib

def _norm(s):
    return str(s).replace(" ", "").lower()

# 종목명(정규화) -> 코드
NAME_TO_CODE = {}
_NORM_NAMES = []
for _c, (_m, _n) in MARKET_MAP.items():
    if _n:
        NAME_TO_CODE[_norm(_n)] = _c
        _NORM_NAMES.append(_norm(_n))


def resolve_to_code(q):
    """6자리 코드면 그대로. 종목명이면 코드로 (대소문자 무시, 부분일치, 오타허용).
       반환 (code, 메시지). 못 찾거나 후보 여럿이면 (None, 안내)."""
    q = str(q).strip()
    z = q.zfill(6)
    if z.isdigit() and len(z) == 6:
        return z, ""                                   # 숫자 코드
    zu = q.upper().zfill(6)                             # ETF/ETN 영숫자 코드 (예: 0091P0)
    if len(zu) == 6 and zu.isalnum() and zu in MARKET_MAP:
        return zu, ""
    key = _norm(q)
    if not key:
        return None, "입력이 비었어"
    if key in NAME_TO_CODE:                            # 정확 일치(대소문자 무시)
        return NAME_TO_CODE[key], ""
    hits = [(nm, c) for nm, c in NAME_TO_CODE.items() if key in nm]   # 부분 일치
    if len(hits) == 1:
        return hits[0][1], f"('{q}' → {MARKET_MAP[hits[0][1]][1]})"
    if len(hits) > 1:
        hits.sort(key=lambda h: difflib.SequenceMatcher(None, key, h[0]).ratio(), reverse=True)
        cand = ", ".join(f"{MARKET_MAP[c][1]}({c})" for _, c in hits[:8])
        return None, f"'{q}' 후보 여럿 → {cand}"
    close = difflib.get_close_matches(key, _NORM_NAMES, n=5, cutoff=0.6)   # 오타 허용
    if len(close) == 1:
        c = NAME_TO_CODE[close[0]]
        return c, f"('{q}' → {MARKET_MAP[c][1]} 로 추정)"
    if len(close) > 1:
        cand = ", ".join(f"{MARKET_MAP[NAME_TO_CODE[nm]][1]}({NAME_TO_CODE[nm]})" for nm in close)
        return None, f"'{q}' 혹시 이거? → {cand}"
    return None, f"'{q}' 종목명/코드 못 찾음"


def market_of(code):
    return MARKET_MAP.get(code.zfill(6), ("KOSDAQ", ""))[0]


def name_of(code):
    nm = MARKET_MAP.get(code.zfill(6), ("", ""))[1]
    return nm if nm else code.zfill(6)


def _label(code):
    """결과 표시용: '레인보우로보틱스(277810)' 형태. 이름 없으면 코드만."""
    nm = MARKET_MAP.get(code.zfill(6), ("", ""))[1]
    return f"{nm}({code.zfill(6)})" if nm else code.zfill(6)


# ====================================================================
# 탭: 볼린저 — 스크립트 소스를 읽어 TICKER 바꿔 exec
# ====================================================================
def run_script(path, ticker):
    code, msg = resolve_to_code(ticker)
    if code is None:
        return msg
    ticker = code
    try:
        src = open(path, encoding="utf-8").read()
    except Exception as e:
        return f"스크립트 로드 실패: {e}"
    src, n = _TICKER_RE.subn(f"TICKER = '{ticker}'", src, count=1)
    if n == 0:
        return "스크립트에서 TICKER 라인을 못 찾음."
    buf = io.StringIO()
    g = {"__name__": "__main__"}
    try:
        with contextlib.redirect_stdout(buf):
            exec(compile(src, path, "exec"), g)
    except SystemExit:
        pass
    except Exception as e:
        return f"{buf.getvalue()}\n\n[오류] {e}"
    out = buf.getvalue().strip()
    head = f"[{_label(code)}]{('  ' + msg) if msg else ''}\n\n"
    return head + (out if out else "출력 없음 (데이터가 없거나 조건 미충족)")


def run_boll(ticker):
    return run_script(BOLL_PATH, ticker)


# ====================================================================
# 탭: 전량매도 최후보루 (다종목) — 현재 단계 status 포함
# ====================================================================
def _real_sells(stock, low, ma240, sells):
    """백테스터 main()과 동일한 4종 필터로 '진짜 매도'만 (240터치/240없음/너무일찍/추세유지 SKIP)."""
    out = []
    for s in sells:
        i = s["i"]; price = s["price"]
        cat, days = BT.classify_240(stock, low, ma240, i, price)
        if getattr(BT, "SKIP_IF_NO_240", False) and cat == "UNKNOWN":
            continue
        if getattr(BT, "SKIP_IF_BELOW_240", False) and cat == "ALREADY_DOWN":
            continue
        med = getattr(BT, "MAX_EARLY_DAYS", None)
        if med is not None and cat == "EARLY" and days is not None and days > med:
            continue
        if getattr(BT, "SKIP_IF_STILL_UP", False) and cat == "STILL_UP":
            continue
        out.append(s)
    return out


def _skip_reason(stock, low, ma240, s):
    cat, days = BT.classify_240(stock, low, ma240, s["i"], s["price"])
    if getattr(BT, "SKIP_IF_NO_240", False) and cat == "UNKNOWN":
        return "240일선이 아직 없는 종목(상장 얼마 안 됨)이라 신호를 믿을 수 없어서"
    if getattr(BT, "SKIP_IF_BELOW_240", False) and cat == "ALREADY_DOWN":
        return "신호 당일 주가가 이미 240일선을 건드린 상태 → 이미 늦은 신호라서"
    med = getattr(BT, "MAX_EARLY_DAYS", None)
    if med is not None and cat == "EARLY" and days is not None and days > med:
        return f"신호 뒤 {days}거래일이나 지나서야 240일선이 깨짐 → 너무 성급한 신호라서"
    if getattr(BT, "SKIP_IF_STILL_UP", False) and cat == "STILL_UP":
        return "신호 뒤 200거래일 동안 240일선을 안 깨고 추세가 살아있었음 → 헛매도라서"
    return None


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"]; low = r["low"]; ma240 = r["ma240"]
    raw_sells, jumps, state = BT.find_confirmed_sells(
        stock, index, r["dates"], r["start_pos"], return_state=True)
    sells = _real_sells(stock, low, ma240, raw_sells)   # 4종 필터
    last_i = len(stock) - 1
    recent = [s for s in sells if s["i"] >= last_i - ALERT_DAYS]
    trigger = None; skipped = None
    if state.get("sold") and state.get("sold_i") is not None:
        si = state["sold_i"]
        trigger = next((s for s in sells if s["i"] == si), None)
        if trigger is None:
            raw = next((s for s in raw_sells if s["i"] == si), None)
            if raw is not None:
                skipped = {"sell": raw,
                           "reason": _skip_reason(stock, low, ma240, raw) or "조건 미달"}
            state = dict(state, sold=False)
    import numpy as _np
    _m = ma240[last_i]
    return {
        "code": code, "market": market, "ok": True,
        "price_now": float(stock[last_i]),
        "ma240_now": (None if _np.isnan(_m) else float(_m)),
        "last_date": r["dates"][last_i],
        "recent": recent,
        "last_sell": sells[-1] if sells else None,
        "trigger": trigger,
        "skipped": skipped,
        "state": state,
    }


def _status_line(res):
    st = res.get("state")
    if not st:
        return "관망 (신호 대기)"
    if st.get("sold"):
        return (f"과거 매도신호 유지중 (지금 팔라는 신호 아님) · "
                f"지금 이격 {st['div']*100:.1f}%")
    nm = {"없음": "관망", "flagON": "flag ON (대기)", "1차": "1차 (진행중)",
          "2차flagON": "2차 flag ON (전량매도 임박!)",
          "2차매도": "관망 (매도신호 있었으나 필터로 제외됨)"}.get(st["stage_name"], st["stage_name"])
    return f"{nm} · 지금 이격 {st['div']*100:.1f}%"


def run_monitor(text):
    raw = [c.strip() for c in text.replace(",", " ").replace("\n", " ").split()]
    raw = [c for c in raw if c]
    if not raw:
        return "종목코드 또는 종목명을 입력해줘 (예: 277810  레인보우로보틱스)"

    head = (f"**전량매도 최후보루** · 최근 {ALERT_DAYS}거래일 기준\n\n"
            f"기준: 이격 {BT.DIV_THRESHOLD*100:.0f}% 단계누적({BT.MIN_DEEPEN*100:.1f}%p+) "
            f"+ 하루급변 {BT.DAILY_JUMP*100:.0f}%p 점프. "
            f"필터: 240터치·240없음·헛매도 제외\n\n---\n")
    out = [head]
    alerts = []
    asof = None
    for _q in raw:
        code, msg = resolve_to_code(_q)
        if code is None:
            out.append(f"- {msg}")
            continue
        res = check_one(code)
        lbl = _label(code) + (f" {msg}" if msg else "")
        if not res["ok"]:
            out.append(f"- {lbl} ({res['market']}) — {res['msg']}")
            continue
        asof = res.get("last_date")
        st = res.get("state") or {}
        steps = st.get("steps") or []
        trig = res.get("trigger")
        skp = res.get("skipped")

        m240 = res.get("ma240_now")
        m240s = (f"240선 {m240:,.0f} ({(res['price_now']/m240-1)*100:+.1f}%)"
                 if m240 else "240선 없음")

        if res["recent"]:
            sl = res["recent"][-1]
            alerts.append(_label(code))
            ago = st.get("last_i", 0) - sl["i"]
            verdict = (f"### 🔴 [★지금 매도] {_label(code)} ({res['market']})\n"
                       f"- **전량매도 신호: {sl['date']} @ {sl['price']:,.0f}** "
                       f"(신호시점 이격 {sl['div']*100:.0f}%) · {ago}거래일 전\n")
        elif trig and st.get("sold"):
            ago = st.get("last_i", 0) - trig["i"]
            verdict = (f"### 🟡 [과거 신호] {_label(code)} ({res['market']})\n"
                       f"- **※ 지금 팔라는 신호 아님**\n"
                       f"- 전량매도 신호: {trig['date']} @ {trig['price']:,.0f} "
                       f"(그때 이격 {trig['div']*100:.0f}%) · {ago}거래일 전\n"
                       f"- 이격 {BT.DIV_THRESHOLD*100:.0f}% 위로 회복하면 리셋\n")
        elif skp:
            sl = skp["sell"]
            ago = st.get("last_i", 0) - sl["i"]
            verdict = (f"### 🟢 [보유] {_label(code)} ({res['market']})\n"
                       f"- 무시된 신호: {sl['date']} @ {sl['price']:,.0f} "
                       f"(그때 이격 {sl['div']*100:.0f}%) · {ago}거래일 전\n"
                       f"- 무시한 이유 — {skp['reason']}\n")
        else:
            verdict = (f"### 🟢 [보유] {_label(code)} ({res['market']})\n"
                       f"- 매도 신호 없음. 보유 유지.\n"
                       f"- 현재 단계: {_status_line(res)}\n")
            # ★ 이격이 회복돼 리셋됐어도 '직전 매도가 언제였는지'는 항상 보여준다.
            ls = res.get("last_sell")
            if ls:
                ago = st.get("last_i", 0) - ls["i"]
                _nowdiv = st.get("div")
                verdict += (f"- [직전 매도(참고)] {ls['date']} @ {ls['price']:,.0f} "
                            f"(그때 이격 {ls['div']*100:.0f}%) · {ago}거래일 전\n")
                if _nowdiv is not None:
                    verdict += f"  - ※ 지금은 이격 {_nowdiv*100:.0f}%로 회복돼서 신호 꺼짐\n"
        out.append(verdict)

        out.append(f"**[지금]** {res['last_date']} 종가 {res['price_now']:,.0f} · {m240s}\n")
        if "peak_price" in st:
            out.append(
                f"- 고점 {st['peak_price']:,.0f} ({st['peak_date']})\n"
                f"- 지금 이격 **{st['div']*100:.1f}%** "
                f"= 종목 고점대비 {st['stock_from_peak']*100:.1f}% "
                f"− 지수 같은기간 {st['index_from_peak']*100:+.1f}%\n")

        if steps:
            out.append(f"**[단계 {len(steps)}개]**\n")
            for stp in steps:
                mark = "  ←★" if "전량매도" in stp["name"] else ""
                out.append(f"- `{stp['name']}` {stp['date']} @ {stp['price']:,.0f} "
                           f"(이격 {stp['div']*100:.1f}%){mark}\n")
        out.append("\n")

    out.append("\n---\n")
    if asof:
        out.append(f"*상태 기준: {asof_label(asof)}*\n")
    if alerts:
        out.append(f"**오늘 확인 필요: {len(alerts)}종목 → {', '.join(alerts)}**")
    else:
        out.append("**최근 매도신호 없음. 전부 보유 유지.**")
    return "\n".join(out)


# ====================================================================
# 탭: 삼성·ETF 매도 (RS) — final_sell_monitor_rs.analyze_one 호출
# ====================================================================
def run_rs(text):
    raw = [c.strip() for c in text.replace(",", " ").replace("\n", " ").split()]
    raw = [c for c in raw if c]
    if not raw:
        return "종목코드 또는 종목명을 입력해줘 (예: 005930  삼성전자)"
    out = []
    for _q in raw:
        code, msg = resolve_to_code(_q)
        if code is None:
            out.append(msg)
            continue
        market = market_of(code)
        try:
            res = final_sell_monitor_rs.analyze_one(
                {"ticker": code, "market": market, "name": name_of(code)})
        except Exception as e:
            res = f"[오류] {_label(code)} — {e}"
        out.append(res)
        out.append("")
    return "\n".join(out)


# ====================================================================
def run_status(text):
    """종목의 일봉 '지금 상태'(동산 그리는 중 / 위아래 N번째 / 매수 멈춤 등)를 한 줄로."""
    try:
        import FinanceDataReader as fdr
        import buy_engine_dipfix as be
    except Exception as e:
        return f"엔진/데이터 로드 실패: {e}"
    raw = [c.strip() for c in text.replace(",", " ").replace("\n", " ").split()]
    raw = [c for c in raw if c]
    if not raw:
        return "종목코드 또는 종목명을 입력해줘 (예: 096770  DN오토모티브)"
    seen = set(); out = []
    for _q in raw:
        code, msg = resolve_to_code(_q)
        if code is None:
            out.append(f"- {msg}"); continue
        if code in seen:
            continue          # 같은 종목 코드+이름 둘 다 넣어도 한 번만
        seen.add(code)
        try:
            df = fdr.DataReader(code, "2005-01-01")
        except Exception as e:
            out.append(f"- {_label(code)}: 데이터 로드 실패 ({e})"); continue
        if df is None or len(df) < 300:
            out.append(f"- {_label(code)}: 일봉 데이터 부족(300봉 미만)"); continue
        try:
            be.run(df, "1day")
            status = getattr(be, "LAST_STATUS", "") or "(상태 없음)"
            kind = getattr(be, "LAST_STATUS_KIND", "wait")
        except Exception as e:
            out.append(f"- {_label(code)}: 분석 실패 ({e})"); continue
        last_date = str(df.index[-1])[:10]
        last_px = float(df["Close"].iloc[-1])
        # 색상 일관성(최후보루와 동일): 🟢초록=좋음(매수 국면), ⚠️주황=주의(무너지는 중)
        if kind == "buy":
            line = f'<span style="color:#16a34a;font-weight:700">🟢 {status}</span>'
        elif kind == "danger":
            line = f'<span style="color:#ea580c;font-weight:700">⚠️ {status}</span>'
        else:
            line = status
        out.append(f"### 📈 {_label(code)} ({market_of(code)})\n"
                   f"- **[일봉]**  {line}\n"
                   f"- 기준: {last_date} 종가 {last_px:,.0f}")
    return "\n\n".join(out) if out else "결과 없음"


# UI  (탭 순서: 최후보루 → 볼린저 → 일봉 상태 → 삼성·ETF(RS))
# ====================================================================
with gr.Blocks(title="WTW Outlook") as demo:
    gr.Markdown("# 📊 WTW Outlook")

    with gr.Tab("전량매도 최후보루"):
        gr.Markdown("보유종목 **코드 또는 종목명**을 넣으면 매도신호/현재단계를 알려줘. (시장 자동판별, 여러 종목 가능)")
        m_in = gr.Textbox(label="종목코드 또는 종목명 (여러 개는 공백/줄바꿈/쉼표로)",
                          value="277810 레인보우로보틱스", lines=3)
        m_btn = gr.Button("매도신호 확인", variant="primary")
        m_out = gr.Markdown()
        m_btn.click(run_monitor, inputs=m_in, outputs=m_out)

    with gr.Tab("볼린저 카운트"):
        gr.Markdown("볼린저밴드 상단 돌파 카운팅 (최근 1년 자동)")
        b_in = gr.Textbox(label="종목코드 또는 종목명", value="420770")
        b_btn = gr.Button("분석", variant="primary")
        b_out = gr.Code(label="결과")
        b_btn.click(run_boll, inputs=b_in, outputs=b_out)

    with gr.Tab("일봉 상태"):
        gr.Markdown("종목의 **일봉 지금 상태**를 알려줘 (동산 그리는 중 / 위아래 N번째 / 매수 멈춤 등). 여러 종목 가능.")
        s_in = gr.Textbox(label="종목코드 또는 종목명 (여러 개는 공백/줄바꿈/쉼표로)",
                          value="096770 DN오토모티브", lines=3)
        s_btn = gr.Button("일봉 상태 확인", variant="primary")
        s_out = gr.Markdown()
        s_btn.click(run_status, inputs=s_in, outputs=s_out)

    with gr.Tab("삼성·ETF 매도 (RS)"):
        gr.Markdown("삼성전자·ETF처럼 이격으로 안 잡히는 종목의 RS 매도신호. (위아래 매수 구간이면 억제됨)")
        rs_in = gr.Textbox(label="종목코드 또는 종목명 (여러 개는 공백/줄바꿈/쉼표로)",
                           value="005930 삼성전자", lines=3)
        rs_btn = gr.Button("RS 매도신호 확인", variant="primary")
        rs_out = gr.Code(label="결과")
        rs_btn.click(run_rs, inputs=rs_in, outputs=rs_out)

if __name__ == "__main__":
    demo.launch()