diff --git a/app_streamlit.py b/app_streamlit.py index 9832a32..d531b5d 100644 --- a/app_streamlit.py +++ b/app_streamlit.py @@ -728,6 +728,70 @@ def _build_daily_report_lines(dfs, cutoff_kst, now_kst, symbol, offset, header_s lines.append(f"합계: {passed_all}T {failed_all}F (승률 {rate:.2f}%)") return "\n".join(lines) +def _count_stop_touches_per_type(df, cutoff_kst, lookahead=3): + """ + 각 진입 신호 캔들 (1번째 캔들) 기준으로 그 후 lookahead 개 캔들 동안 + (즉 1번째 캔들 시작가 ~ (lookahead+1) 번째 캔들 시작가 구간) 손절가를 + 터치했는지 카운트. 롱은 low <= stop, 숏은 high >= stop. + + 반환: {signal_name: [touch_count, total_count]} + """ + result = {sig: [0, 0] for sig, _ in DAILY_REPORT_SIGNAL_LABELS} + if df is None or df.empty or "open_time" not in df.columns: + return result + recent = df[df["open_time"] >= cutoff_kst].reset_index(drop=True) + if len(recent) <= lookahead: + return result + for sig, _ in DAILY_REPORT_SIGNAL_LABELS: + if sig not in recent.columns: + continue + if sig in LONG_SIGNALS: + direction = "long" + elif sig in SHORT_SIGNALS: + direction = "short" + else: + continue # short_caution_signal — 진입 신호 아님, 손절가 추적 X + for i in range(len(recent) - lookahead): + row = recent.iloc[i] + if not bool(row.get(sig, False)): + continue + entry = float(row["open"]) + window = recent.iloc[i:i + lookahead] + result[sig][1] += 1 + if direction == "long": + stop = entry * (1 - STOP_LOSS_PCT) + if float(window["low"].min()) <= stop: + result[sig][0] += 1 + else: + stop = entry * (1 + STOP_LOSS_PCT) + if float(window["high"].max()) >= stop: + result[sig][0] += 1 + return result + +def _build_stop_touch_lines(dfs, cutoff_kst, now_kst, symbol): + lines = [ + f"[손절가 터치 횟수 알림(시간봉 *3배기준)] ({symbol})", + f"기준: {now_kst.strftime('%Y-%m-%d %H:%M')} KST", + f"손절 비율: ±{STOP_LOSS_PCT*100:.1f}% (10x 레버리지 기준 ROI ±15%)", + ] + for tf in DAILY_REPORT_TIMEFRAMES: + df = dfs.get(tf) + counts = _count_stop_touches_per_type(df, cutoff_kst, lookahead=3) + lines.append("") + lines.append(f"[{TF_LABEL_MAP.get(tf, tf)}]") + touch_all = 0 + total_all = 0 + for sig, sig_label in DAILY_REPORT_SIGNAL_LABELS: + if sig == "short_caution_signal": + continue + touch, total = counts.get(sig, [0, 0]) + lines.append(f"{sig_label}: {touch}/{total}") + touch_all += touch + total_all += total + rate = (touch_all / total_all * 100) if total_all > 0 else 0.0 + lines.append(f"합계: {touch_all}/{total_all} (터치율 {rate:.2f}%)") + return "\n".join(lines) + def send_daily_report(symbol="BTCUSDT"): now_kst = (datetime.now(timezone.utc) + KST).replace(tzinfo=None) cutoff_kst = now_kst - timedelta(hours=24) @@ -742,6 +806,8 @@ def send_daily_report(symbol="BTCUSDT"): send_telegram(msg_1x) msg_2x = _build_daily_report_lines(dfs, cutoff_kst, now_kst, symbol, offset=2, header_suffix="2배 시간 (2번째 봉 검증)") send_telegram(msg_2x) + msg_touch = _build_stop_touch_lines(dfs, cutoff_kst, now_kst, symbol) + send_telegram(msg_touch) def _daily_report_loop(): while True: