Axel HocksandClaude Opus 4.8
75d28827e8
Headless FastAPI-Backend (server.py + core/engine.py) mit Mobile-PWA (web/), Strategie-/Backtest-Suite und Doku. Secrets, DB, Logs und Laufzeit-State sind via .gitignore ausgeschlossen; Config-Vorlage: oil_widget_config.ini.example. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
113 lines
4.0 KiB
Python
113 lines
4.0 KiB
Python
"""
|
|
core/notify.py — Telegram-Benachrichtigungen
|
|
=============================================
|
|
Sendet Trade-Abschlüsse als Telegram-Nachricht.
|
|
Kein externes Package nötig — nur urllib aus der Standardbibliothek.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
import json
|
|
import threading
|
|
import urllib.request
|
|
|
|
from core.logger import get_logger
|
|
|
|
log_notify = get_logger("notify")
|
|
|
|
# Setup-Namen für lesbare Telegram-Nachricht
|
|
_SETUP_LABELS = {
|
|
"TREND_PULLBACK_LONG": "Pullback ▲",
|
|
"TREND_PULLBACK_LONG_AT_SUPPORT": "Pullback ▲ @ Support",
|
|
"TREND_PULLBACK_SHORT": "Pullback ▼",
|
|
"TREND_PULLBACK_SHORT_AT_RESISTANCE": "Pullback ▼ @ Resistance",
|
|
"TREND_CONTINUATION_LONG": "Continuation ▲",
|
|
"TREND_CONTINUATION_SHORT": "Continuation ▼",
|
|
"BREAKOUT_LONG": "Breakout ▲",
|
|
"BREAKOUT_SHORT": "Breakout ▼",
|
|
"MEAN_REVERT_LONG": "Mean-Revert ▲",
|
|
"MEAN_REVERT_SHORT": "Mean-Revert ▼",
|
|
"DEAD_CAT_BOUNCE_SHORT": "Dead Cat Bounce ▼",
|
|
}
|
|
|
|
_CLOSED_BY_LABEL = {
|
|
"sl": "SL ❌",
|
|
"tp": "TP ✅",
|
|
"manual": "Manuell 🖐",
|
|
"emergency": "Emergency 🚨",
|
|
"unknown": "Unbekannt ⚠️",
|
|
}
|
|
|
|
|
|
def send_telegram(message: str, token: str, chat_id: str) -> None:
|
|
"""Sendet eine Telegram-Nachricht asynchron (blockiert nicht)."""
|
|
def _send():
|
|
try:
|
|
url = f"https://api.telegram.org/bot{token}/sendMessage"
|
|
body = json.dumps({
|
|
"chat_id": chat_id,
|
|
"text": message,
|
|
"parse_mode": "HTML",
|
|
}).encode("utf-8")
|
|
req = urllib.request.Request(
|
|
url, data=body,
|
|
headers={"Content-Type": "application/json"})
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
result = json.loads(resp.read())
|
|
if not result.get("ok"):
|
|
log_notify.warning(f"Telegram API Fehler: {result}")
|
|
else:
|
|
log_notify.debug("Telegram-Nachricht gesendet")
|
|
except Exception as e:
|
|
log_notify.warning(f"Telegram-Send fehlgeschlagen: {e}")
|
|
|
|
threading.Thread(target=_send, daemon=True).start()
|
|
|
|
|
|
def build_trade_open_message(
|
|
direction: str,
|
|
symbol: str | None = None,
|
|
lots: float | None = None,
|
|
entry_price: float | None = None,
|
|
setup: str | None = None,
|
|
entry_ts: int | None = None,
|
|
) -> str:
|
|
"""Formatiert die Telegram-Nachricht für einen Trade-Einstieg."""
|
|
import datetime
|
|
dir_str = "LONG" if (direction or "").upper() in ("BUY", "LONG") else "SHORT"
|
|
emoji = "\U0001f7e2" if dir_str == "LONG" else "\U0001f534"
|
|
setup_str = _SETUP_LABELS.get(setup or "", setup or "—")
|
|
sym_str = f" · {symbol}" if symbol else ""
|
|
import time as _time
|
|
time_str = (_time.strftime("%d.%m %H:%M", _time.localtime(entry_ts))
|
|
if entry_ts else _time.strftime("%d.%m %H:%M"))
|
|
|
|
lines = [f"{emoji} <b>{dir_str}{sym_str}</b>"]
|
|
if entry_price:
|
|
lines.append(f"Einstieg: {entry_price:.5g}"
|
|
+ (f" · {lots} Lots" if lots else ""))
|
|
if setup:
|
|
lines.append(f"Setup: {setup_str}")
|
|
lines.append(f"Zeit: {time_str}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def build_trade_close_message(
|
|
direction: str,
|
|
pnl: float,
|
|
commission: float,
|
|
closed_by: str,
|
|
setup: str | None,
|
|
exit_ts: int,
|
|
symbol: str | None = None,
|
|
) -> str:
|
|
"""Formatiert die Telegram-Nachricht für einen Trade-Abschluss."""
|
|
import datetime
|
|
pnl_net = pnl + (commission or 0.0)
|
|
win = pnl_net >= 0
|
|
sign = "+" if win else ""
|
|
import time as _time
|
|
ts_str = (_time.strftime("%d.%m", _time.localtime(exit_ts)) if win
|
|
else _time.strftime("%H:%M", _time.localtime(exit_ts)))
|
|
sym_str = f"{symbol} " if symbol else ""
|
|
return f"{sym_str}{sign}{pnl_net:.2f} {ts_str}"
|