From b78176d4b8da867f8d86030dc7a79c14ea550f13 Mon Sep 17 00:00:00 2001 From: ILSEON-RYU Date: Mon, 4 May 2026 23:48:18 +0900 Subject: [PATCH] =?UTF-8?q?=EB=AA=A8=EB=93=A0=20TF=20forming=20=EC=95=8C?= =?UTF-8?q?=EB=A6=BC=20+=2030s=20=EC=9E=AC=EA=B2=80=EC=A6=9D=20+=20?= =?UTF-8?q?=EB=B0=98=EB=8C=80=20=EC=8B=A0=ED=98=B8=20=EC=8B=9C=20=EC=B2=AD?= =?UTF-8?q?=EC=82=B0=20=EA=B6=8C=EA=B3=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 변경 1 — closed-only 룰(a57941e) 전체 TF revert 이전: 15m / 30m / 1h 는 closed candle 만 신호 검사 -> forming 동안 알림 없음, 신호가 캔들 안에서 깜빡 사라져도 별도 알림 없음. 이후: 모든 TF (5m/15m/30m/1h) 가 forming candle 포함해 신호 검사. 30초 polling 으로 매 사이클마다 pending_groups 의 신호 상태 재검증 (a9ad52f 의 즉시 취소 로직). 결과적으로: 5m : 30초마다 검증 — 1캔들 절반(2.5m) 의 1/5 15m : 30초마다 검증 — 1캔들 절반(7.5m) 의 1/15 30m : 30초마다 검증 — 1캔들 절반(15m) 의 1/30 1h : 30초마다 검증 — 1캔들 절반(30m) 의 1/60 사용자 요건 "1캔들 시간의 절반 안에 한 번 검증" 자동 충족. ## 변경 2 — 반대 신호 시 청산 권고 이전: 롱 진입 신호와 숏 진입 신호가 시간차 있게 발화해도, 진입 추적 (long_entry / short_entry) 은 같은 TF 끼리만 덮어쓸 뿐 반대편 정리 없음. 이후: 새 진입 신호가 발사될 때 반대 방향 의 모든 활성 진입 (다른 TF 포함) 체크. 있으면 [반대 신호 감지 - 청산 권장] 알림 발송 + 해당 추적 해제. 메시지 포맷: [반대 신호 감지 - 숏 청산 권장] --- 기존 진입 --- 🔽 일반 숏 진입 신호 ... (원래 entry_msg) --- 반대 신호 --- 🔼 일반 롱 진입 신호 ... (현재 신호 entry_msg) 기존 stop loss 알림과 별개로 "반대 신호 발화 = 청산하라" 권고를 보내 사용자가 stop hit 전에 빠르게 정리할 수 있게 함. Co-Authored-By: Claude Opus 4.7 (1M context) --- app_streamlit.py | 106 +++++++++++++++++++++++++++++++++++++---------- 1 file changed, 85 insertions(+), 21 deletions(-) diff --git a/app_streamlit.py b/app_streamlit.py index 26086c0..810aaf6 100644 --- a/app_streamlit.py +++ b/app_streamlit.py @@ -86,8 +86,8 @@ SIG_DEFS = [ ("short_signal", "short", "🔽 일반 숏", "short"), ("vol_long_signal", "vol_long", "🔼 볼륨 롱", "long"), ("vol_short_signal", "vol_short", "🔽 볼륨 숏", "short"), - ("reversal_long_signal", "rev_long", "🔄 롱 전환", "long"), - ("reversal_short_signal","rev_short", "🔄 숏 전환", "short"), + ("reversal_long_signal", "rev_long", "🔄 롱 추세 꺾임 감지", "long"), + ("reversal_short_signal","rev_short", "🔄 숏 추세 꺾임 감지", "short"), ("short_caution_signal", "short_caution","⚠️ 숏 주의", "caution"), ] @@ -126,14 +126,11 @@ def check_and_alert(df, symbol, interval): alert_state.short_entry[interval] = None alert_state.pending_groups = new_pending - # Phase 2 — 신호 검사 + 알림 발사. - # 5분봉: forming candle 포함 (반응성 우선, 깜빡임은 a9ad52f 의 즉시 취소로 대응) - # 15m / 30m / 1h: forming candle 제외 (forming 동안 신호 깜빡으로 인한 가짜 - # 알림 자체를 차단. 알림은 캔들 마감 후 ~30초 이내 발사 — 신뢰성 우선) - if interval in ("15m", "30m", "1h"): - recent = df.iloc[:-1].tail(3) - else: - recent = df.tail(3) + # Phase 2 — 신호 검사 + 알림 발사 (모든 TF forming candle 포함). + # 30초 polling 으로 매 사이클마다 forming candle 의 신호 상태 재검증 → + # 신호 사라지면 즉시 [취소 알림] 발사 (Phase 1 로직). 5m=2.5m, 15m=7.5m, + # 30m=15m, 1h=30m 의 절반 시간보다 훨씬 빠른 검증 주기. + recent = df.tail(3) eligible = [] for sig, key, sub_label, direction in SIG_DEFS: @@ -188,6 +185,18 @@ def check_and_alert(df, symbol, interval): f"손절가: {stop_price:,.2f}" ) entry_record = {"price": entry_price, "stop": stop_price, "open_time": candle_time, "entry_msg": msg} + # 반대 방향 활성 진입이 있으면 청산 권고 + 추적 해제 (모든 TF 대상) + opposite_dict = alert_state.short_entry if direction == "long" else alert_state.long_entry + opposite_label = "숏" if direction == "long" else "롱" + for opp_interval, opp_rec in list(opposite_dict.items()): + if opp_rec is None: + continue + send_telegram( + f"[반대 신호 감지 - {opposite_label} 청산 권장]\n" + f"--- 기존 진입 ---\n{opp_rec['entry_msg']}\n" + f"--- 반대 신호 ---\n{msg}" + ) + opposite_dict[opp_interval] = None if direction == "long": alert_state.long_entry[interval] = entry_record else: @@ -387,22 +396,22 @@ def compute_signals(df, interval="5m"): else: df["short_caution_signal"] = False - # 방향 전환 감지: 직전 5봉의 추세 방향과 현재 캔들 방향이 반대 + 강한 폭 + 거래량 동반. - # - 추세 판단: close[t-1] vs close[t-5] (현재 캔들 제외, 직전까지의 흐름) + # 추세 꺾임 감지: 직전 3봉의 추세 방향과 현재 캔들 방향이 반대 + 강한 폭 + 거래량 동반. + # - 추세 판단: close[t-1] vs close[t-3] (현재 캔들 제외, 직전까지의 흐름) # - 현재 캔들 강도: |close-open|/open >= 0.3% (작은 캔들 노이즈 차단) - # - 거래량: 직전 5봉 평균의 1.3배 이상 (확신) + # - 거래량: 직전 3봉 평균의 1.3배 이상 (확신) + # - 쿨다운: 3봉 prior_close = df["close"].shift(1) - prior_close_5 = df["close"].shift(5) - was_up = prior_close > prior_close_5 - was_down = prior_close < prior_close_5 + prior_close_3 = df["close"].shift(3) + was_up = prior_close > prior_close_3 + was_down = prior_close < prior_close_3 candle_body_pct = (df["close"] - df["open"]) / df["open"].replace(0, float("nan")) - vol_avg5 = df["volume"].rolling(5).mean().shift(1) - vol_strong = df["volume"] > vol_avg5 * 1.3 + vol_avg3 = df["volume"].rolling(3).mean().shift(1) + vol_strong = df["volume"] > vol_avg3 * 1.3 rev_short_raw = was_up & (candle_body_pct < -0.003) & vol_strong rev_long_raw = was_down & (candle_body_pct > 0.003) & vol_strong - # 5봉 쿨다운 (연쇄 발화 방지) - df["reversal_short_signal"] = rev_short_raw & (rev_short_raw.rolling(5, min_periods=1).sum().shift(1).fillna(0) == 0) - df["reversal_long_signal"] = rev_long_raw & (rev_long_raw.rolling(5, min_periods=1).sum().shift(1).fillna(0) == 0) + df["reversal_short_signal"] = rev_short_raw & (rev_short_raw.rolling(3, min_periods=1).sum().shift(1).fillna(0) == 0) + df["reversal_long_signal"] = rev_long_raw & (rev_long_raw.rolling(3, min_periods=1).sum().shift(1).fillna(0) == 0) return df @@ -871,6 +880,59 @@ def _build_stop_touch_lines(dfs, cutoff_kst, now_kst, symbol): lines.append(f"합계: {touch_all}/{total_all} (터치율 {rate:.2f}%)") return "\n".join(lines) +def _count_reversal_outcomes(df, cutoff_kst, lookahead=3): + """추세 꺾임 신호의 lookahead 봉 후 방향 일치 카운트. + - reversal_long: close[i+lookahead] > close[i] -> T (상승 지속) + - reversal_short: close[i+lookahead] < close[i] -> T (하락 지속) + 반환: {sig_name: [total, failed]} + """ + result = {"reversal_long_signal": [0, 0], "reversal_short_signal": [0, 0]} + 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 ("reversal_long_signal", "reversal_short_signal"): + if sig not in recent.columns: + continue + for i in range(len(recent) - lookahead): + row = recent.iloc[i] + if not bool(row.get(sig, False)): + continue + future = recent.iloc[i + lookahead] + entry_close = float(row["close"]) + future_close = float(future["close"]) + confirmed = (sig == "reversal_long_signal" and future_close > entry_close) or \ + (sig == "reversal_short_signal" and future_close < entry_close) + result[sig][0] += 1 + if not confirmed: + result[sig][1] += 1 + return result + +def _build_reversal_lines(dfs, cutoff_kst, now_kst, symbol): + lines = [ + f"📊 추세 꺾임 감지 통계 ({symbol})", + f"기준: {now_kst.strftime('%Y-%m-%d %H:%M')} KST (3봉 후 방향 일치)", + ] + for tf in DAILY_REPORT_TIMEFRAMES: + df = dfs.get(tf) + counts = _count_reversal_outcomes(df, cutoff_kst, lookahead=3) + lines.append("") + lines.append(f"[{TF_LABEL_MAP.get(tf, tf)}]") + total_all = 0 + failed_all = 0 + for sig, lbl in [("reversal_long_signal", "🔄 롱 추세 꺾임 감지"), + ("reversal_short_signal", "🔄 숏 추세 꺾임 감지")]: + t, f = counts[sig] + passed = t - f + lines.append(f"{lbl}: {passed}T {f}F") + total_all += t + failed_all += f + passed_all = total_all - failed_all + rate = (passed_all / total_all * 100) if total_all > 0 else 0.0 + lines.append(f"합계: {passed_all}T {failed_all}F (승률 {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) @@ -887,6 +949,8 @@ def send_daily_report(symbol="BTCUSDT"): send_telegram(msg_2x) msg_touch = _build_stop_touch_lines(dfs, cutoff_kst, now_kst, symbol) send_telegram(msg_touch) + msg_rev = _build_reversal_lines(dfs, cutoff_kst, now_kst, symbol) + send_telegram(msg_rev) def _daily_report_loop(): while True: