chpark/tradeing
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

모든 TF forming 알림 + 30s 재검증 + 반대 신호 시 청산 권고

Browse Source 
## 변경 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) <noreply@anthropic.com>
This commit is contained in:
ILSEON-RYU
2026-05-04 23:48:18 +09:00
parent 67f46a427a
commit b78176d4b8
1 changed files with 85 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app_streamlit.py
+85 -21
View File
@@ -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: