Adriano Dal Pastro
40 lines
1.1 KiB
Python
40 lines
1.1 KiB
Python
"""Notifiche Telegram per il paper trader."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import urllib.request
|
|
import urllib.parse
|
|
import json
|
|
|
|
BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "")
|
|
CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "")
|
|
|
|
NOTIFY_EVENTS = {
|
|
"SIGNAL", "OPENED", "CLOSED", "OPEN_FAILED", "CLOSE_FAILED",
|
|
"ERROR", "STARTUP", "SHUTDOWN", "TRAINING_FAILED",
|
|
}
|
|
|
|
|
|
def send_telegram(text: str) -> bool:
|
|
if not BOT_TOKEN or not CHAT_ID:
|
|
return False
|
|
try:
|
|
url = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"
|
|
data = urllib.parse.urlencode({"chat_id": CHAT_ID, "text": text, "parse_mode": "HTML"}).encode()
|
|
urllib.request.urlopen(url, data, timeout=10)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
def notify_event(event: str, data: dict | None = None):
|
|
if event not in NOTIFY_EVENTS:
|
|
return
|
|
lines = [f"📊 <b>{event}</b>"]
|
|
if data:
|
|
for k, v in data.items():
|
|
if k in ("signal",):
|
|
continue
|
|
lines.append(f" {k}: {v}")
|
|
send_telegram("\n".join(lines))
|