같은 캔들 같은 방향 신호 그룹핑 + 손절 3% + 천단위 포맷

## 1. 같은 캔들 같은 방향 신호를 1개 알림으로 그룹핑
이전: 한 캔들에서 강한 롱 / 일반 롱 / 볼륨 롱 신호가 동시에 떴을 때
3개의 별도 텔레그램이 같은 분 안에 연속 도착 → 스팸 체감.
이후: 발화한 신호들의 라벨을 " + " 로 결합해 1개 메시지로 발송.

  🟢 강한 롱 + 🔼 일반 롱 + 🔼 볼륨 롱 진입 신호
  BTCUSDT 5분봉
  시간: 2026-05-01 21:00
  진입가: 76,200.00
  손절가: 73,914.00

per-candle dedup 과 ALERT_COOLDOWN 가드는 그대로 유지. 그룹 안 모든
신호의 _last_alert / _last_fired_candle 한꺼번에 갱신.

## 2. 손절가 비율 10% -> 3%
STOP_LOSS_PCT = 0.03 으로 조정. 5m/15m 단타 기준에 10% 는 너무 헐거워
손절 알림이 사실상 작동 안 함. 3% 면 보통 1~2시간 내 결과 결판.

## 3. 가격 표기 천단위 반점
진입가/손절가/현재가 등 모든 텔레그램 가격 출력에 ',' 천단위 구분자 적용.
(차트 hover 는 이미 적용돼있어 변경 없음.)

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
ILSEON-RYU
2026-05-01 20:57:57 +09:00
parent 29a36a1bc5
commit d2f08ff4e9
+64 -39
View File
@@ -55,7 +55,7 @@ KST = timedelta(hours=9)
_last_alert = {"strong_long": 0, "strong_short": 0, "long": 0, "short": 0, "vol_long": 0, "vol_short": 0, "short_caution": 0}
_last_fired_candle = {"strong_long": None, "strong_short": None, "long": None, "short": None, "vol_long": None, "vol_short": None, "short_caution": None}
STOP_LOSS_PCT = 0.10
STOP_LOSS_PCT = 0.03
LONG_SIGNALS = {"strong_long_signal", "long_signal", "vol_long_signal"}
SHORT_SIGNALS = {"strong_short_signal", "short_signal", "vol_short_signal"}
TF_LABEL_MAP = {
@@ -77,19 +77,23 @@ def send_telegram(message: str):
except Exception as e:
print(f"[텔레그램 오류] {e}")
SIG_DEFS = [
("strong_long_signal", "strong_long", "🟢 강한 롱", "long"),
("strong_short_signal", "strong_short", "🔴 강한 숏", "short"),
("long_signal", "long", "🔼 일반 롱", "long"),
("short_signal", "short", "🔽 일반 숏", "short"),
("vol_long_signal", "vol_long", "🔼 볼륨 롱", "long"),
("vol_short_signal", "vol_short", "🔽 볼륨 숏", "short"),
("short_caution_signal","short_caution","⚠️ 숏 주의", "caution"),
]
def check_and_alert(df, symbol, interval):
global _long_entry, _short_entry
now = time.time()
recent = df.tail(3)
for sig, key, label in [
("strong_long_signal", "strong_long", "🟢 강한 롱 진입 신호"),
("strong_short_signal", "strong_short", "🔴 강한 숏 진입 신호"),
("long_signal", "long", "🔼 롱 진입 신호"),
("short_signal", "short", "🔽 숏 진입 신호"),
("vol_long_signal", "vol_long", "🔼 볼륨급등 롱 신호"),
("vol_short_signal", "vol_short", "🔽 볼륨급등 숏 신호"),
("short_caution_signal","short_caution","⚠️ 숏 진입 주의 신호"),
]:
eligible = []
for sig, key, sub_label, direction in SIG_DEFS:
if sig not in recent.columns:
continue
triggered = recent[recent[sig].fillna(False)]
@@ -100,51 +104,72 @@ def check_and_alert(df, symbol, interval):
continue
if now - _last_alert[key] <= ALERT_COOLDOWN:
continue
eligible.append({
"sig": sig, "key": key, "sub_label": sub_label,
"direction": direction, "candle_time": candle_time, "row": triggered.iloc[-1],
})
tf_label = TF_LABEL_MAP.get(interval, interval)
if not eligible:
groups = {}
else:
groups = {"long": [], "short": [], "caution": []}
for e in eligible:
groups[e["direction"]].append(e)
tf_label = TF_LABEL_MAP.get(interval, interval)
def _send_group(group):
if not group:
return
candle_time = group[0]["candle_time"]
candle_time_str = pd.Timestamp(candle_time).strftime("%Y-%m-%d %H:%M")
if sig in LONG_SIGNALS:
entry_price = float(triggered.iloc[-1]["open"])
stop_price = entry_price * (1 - STOP_LOSS_PCT)
sub_labels = " + ".join(e["sub_label"] for e in group)
direction = group[0]["direction"]
if direction == "caution":
msg = (
f"{label}\n{symbol} {tf_label}\n"
f"시간: {candle_time_str}\n"
f"진입가: {entry_price:.2f}\n"
f"손절가: {stop_price:.2f}"
)
_long_entry = {"price": entry_price, "stop": stop_price, "open_time": candle_time, "entry_msg": msg}
elif sig in SHORT_SIGNALS:
entry_price = float(triggered.iloc[-1]["open"])
stop_price = entry_price * (1 + STOP_LOSS_PCT)
msg = (
f"{label}\n{symbol} {tf_label}\n"
f"시간: {candle_time_str}\n"
f"진입가: {entry_price:.2f}\n"
f"손절가: {stop_price:.2f}"
)
_short_entry = {"price": entry_price, "stop": stop_price, "open_time": candle_time, "entry_msg": msg}
else:
msg = (
f"{label}\n{symbol} {tf_label}\n"
f"{sub_labels} 신호\n{symbol} {tf_label}\n"
f"시간: {candle_time_str}"
)
send_telegram(msg)
else:
entry_price = float(group[0]["row"]["open"])
if direction == "long":
stop_price = entry_price * (1 - STOP_LOSS_PCT)
else:
stop_price = entry_price * (1 + STOP_LOSS_PCT)
msg = (
f"{sub_labels} 진입 신호\n{symbol} {tf_label}\n"
f"시간: {candle_time_str}\n"
f"진입가: {entry_price:,.2f}\n"
f"손절가: {stop_price:,.2f}"
)
entry_record = {"price": entry_price, "stop": stop_price, "open_time": candle_time, "entry_msg": msg}
if direction == "long":
global _long_entry
_long_entry = entry_record
else:
global _short_entry
_short_entry = entry_record
send_telegram(msg)
for e in group:
_last_alert[e["key"]] = now
_last_fired_candle[e["key"]] = e["candle_time"]
send_telegram(msg)
_last_alert[key] = now
_last_fired_candle[key] = candle_time
_send_group(groups.get("long", []))
_send_group(groups.get("short", []))
_send_group(groups.get("caution", []))
current_price = float(df.iloc[-1]["close"])
if _long_entry is not None and current_price <= _long_entry["stop"]:
send_telegram(
f"[손절가알림]\n{_long_entry['entry_msg']}\n"
f"현재가: {current_price:.2f}"
f"현재가: {current_price:,.2f}"
)
_long_entry = None
if _short_entry is not None and current_price >= _short_entry["stop"]:
send_telegram(
f"[손절가알림]\n{_short_entry['entry_msg']}\n"
f"현재가: {current_price:.2f}"
f"현재가: {current_price:,.2f}"
)
_short_entry = None