# -*- 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'🟢 {status}'
elif kind == "danger":
line = f'⚠️ {status}'
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()