일일 리포트 3번째 메시지: 손절가 터치 횟수 (시간봉 *3배 기준)

기존 1배/2배 검증 메시지 다음으로 한 건 더 발송.

## 동작
- 각 진입 신호 캔들을 1번째 캔들로 보고, 1번째 ~ 4번째 캔들 시작가까지의
  구간 (= 3 개 캔들 시간 범위) 동안 손절가를 터치했는지 카운트.
- 롱: window 내 최저가(low) <= 진입가 * (1 - STOP_LOSS_PCT) 이면 터치.
- 숏: window 내 최고가(high) >= 진입가 * (1 + STOP_LOSS_PCT) 이면 터치.
- short_caution_signal 은 진입 신호 아니므로 추적 X.

## 메시지 형식
[손절가 터치 횟수 알림(시간봉 *3배기준)] (BTCUSDT)
기준: 2026-05-03 00:00 KST
손절 비율: ±1.5% (10x 레버리지 기준 ROI ±15%)

[5분봉]
강한 롱: 0/3
강한 숏: 0/1
일반 롱: 0/11
일반 숏: 0/8
볼륨 롱: 0/5
볼륨 숏: 0/8
합계: 0/36 (터치율 0.00%)
[15분봉] ...

(touch / total — total 은 24h 내 해당 시간봉의 진입 신호 발화 수)

## 구현
- _count_stop_touches_per_type(df, cutoff_kst, lookahead=3): signal 별
  [touch, total] 카운트 반환.
- _build_stop_touch_lines: 시간봉 블록 + 합계 메시지 본문.
- send_daily_report: 기존 1x/2x 발송 후 msg_touch 추가 발송.
- dfs 는 1x/2x 빌드 시 이미 fetch 되어있어 재사용 (API 호출 추가 없음).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
ILSEON-RYU
2026-05-03 13:11:49 +09:00
parent 949d876887
commit 9c72141b3f
+66
View File
@@ -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: