239 lines
8.3 KiB
Python
239 lines
8.3 KiB
Python
"""Telegram bot — alerts + commands."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
from aiohttp import web
|
|
|
|
from salior.core.config import config
|
|
from salior.core.logging import setup_logging
|
|
from salior.hooks import global_hooks
|
|
from salior.hooks.registry import HookEvent
|
|
|
|
log = setup_logging()
|
|
|
|
BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN", "")
|
|
|
|
|
|
class TelegramBot:
|
|
"""Telegram bot for alerts and commands.
|
|
|
|
Commands:
|
|
/start — Welcome message
|
|
/status — Agent health + portfolio summary
|
|
/signals — Recent conviction signals
|
|
/pnl — Performance summary
|
|
/help — Show commands
|
|
"""
|
|
|
|
def __init__(self, token: Optional[str] = None) -> None:
|
|
self.token = token or BOT_TOKEN
|
|
self.api_url = f"https://api.telegram.org/bot{self.token}"
|
|
self._offset = 0
|
|
self._running = False
|
|
|
|
async def send(self, chat_id: int, text: str, parse_mode: str = "HTML") -> None:
|
|
"""Send a message to a chat."""
|
|
if not self.token:
|
|
return
|
|
async with httpx.AsyncClient() as client:
|
|
await client.post(
|
|
f"{self.api_url}/sendMessage",
|
|
json={"chat_id": chat_id, "text": text, "parse_mode": parse_mode},
|
|
timeout=10.0,
|
|
)
|
|
|
|
async def send_alert(self, text: str) -> None:
|
|
"""Send an alert to configured chat (broadcasts to all known chats)."""
|
|
for chat_id in self._get_broadcast_chats():
|
|
try:
|
|
await self.send(chat_id, text)
|
|
except Exception as e:
|
|
log.error("telegram_send_error", error=str(e))
|
|
|
|
def _get_broadcast_chats(self) -> list[int]:
|
|
"""Return list of chat IDs to broadcast to. Stored in ~/.salior/telegram_chats.txt."""
|
|
path = os.path.expanduser("~/.salior/telegram_chats.txt")
|
|
if not os.path.exists(path):
|
|
return []
|
|
return [int(line.strip()) for line in open(path).readlines() if line.strip()]
|
|
|
|
async def handle_update(self, update: dict) -> Optional[str]:
|
|
"""Handle an incoming Telegram update, return command response or None."""
|
|
if "message" not in update:
|
|
return None
|
|
|
|
msg = update["message"]
|
|
chat_id = msg["chat"]["id"]
|
|
text = msg.get("text", "")
|
|
user = msg["from"].get("first_name", "trader")
|
|
|
|
if text == "/start":
|
|
return f"👋 Welcome to Salior, {user}.\n\nI send alerts when signals fire and orders fill.\n\nCommands:\n/status — Agent health\n/signals — Recent signals\n/pnl — Performance\n/help — This message"
|
|
|
|
if text == "/help":
|
|
return (
|
|
"📋 <b>Salior Commands</b>\n\n"
|
|
"/start — Welcome\n"
|
|
"/status — Agent health + portfolio\n"
|
|
"/signals — Recent conviction signals\n"
|
|
"/pnl — Performance summary\n"
|
|
"/help — This message"
|
|
)
|
|
|
|
if text == "/status":
|
|
return await self._cmd_status()
|
|
|
|
if text == "/signals":
|
|
return await self._cmd_signals()
|
|
|
|
if text == "/pnl":
|
|
return await self._cmd_pnl()
|
|
|
|
return None
|
|
|
|
async def _cmd_status(self) -> str:
|
|
"""Return agent + portfolio status."""
|
|
from salior.db.supabase_client import SupabaseClient
|
|
from salior.db.timescale_client import TimescaleDB
|
|
|
|
supabase = SupabaseClient()
|
|
db = TimescaleDB()
|
|
await db.connect()
|
|
|
|
# Agent health
|
|
health = await db.fetch(
|
|
"SELECT agent, status, heartbeat FROM agent_health ORDER BY heartbeat DESC LIMIT 5"
|
|
)
|
|
|
|
# Portfolio
|
|
portfolio = await supabase.get_portfolio()
|
|
|
|
lines = ["🤖 <b>Agent Status</b>"]
|
|
for h in (health or [])[:4]:
|
|
dot = "🟢" if h["status"] == "running" else "🔴"
|
|
lines.append(f"{dot} {h['agent']}: {h['status']}")
|
|
|
|
lines.append("\n💼 <b>Portfolio</b>")
|
|
if portfolio:
|
|
for pos in portfolio:
|
|
pnl = pos.get("unrealized_pnl", 0)
|
|
pnl_str = f"+${pnl:.2f}" if pnl >= 0 else f"-${abs(pnl):.2f}"
|
|
lines.append(f"{pos['coin']}: {pos['pos_size']} pos | {pnl_str}")
|
|
else:
|
|
lines.append("No positions")
|
|
|
|
return "\n".join(lines)
|
|
|
|
async def _cmd_signals(self) -> str:
|
|
"""Return recent signals."""
|
|
from salior.db.supabase_client import SupabaseClient
|
|
|
|
supabase = SupabaseClient()
|
|
signals = await supabase.get_recent_signals(limit=5)
|
|
|
|
if not signals:
|
|
return "📊 No recent signals"
|
|
|
|
lines = ["📊 <b>Recent Signals</b>"]
|
|
for sig in signals:
|
|
c = sig["conviction"]
|
|
arrow = "🟢" if c >= 0 else "🔴"
|
|
lines.append(f"{arrow} {sig['coin']} | {sig['regime']} | {c:+.2f}")
|
|
|
|
return "\n".join(lines)
|
|
|
|
async def _cmd_pnl(self) -> str:
|
|
"""Return performance summary."""
|
|
return "📈 PnL: Connect TimescaleDB performance table for full data.\n\nCurrently placeholder."
|
|
|
|
async def poll_loop(self) -> None:
|
|
"""Poll Telegram for updates."""
|
|
if not self.token:
|
|
log.warning("telegram_bot_no_token")
|
|
return
|
|
|
|
self._running = True
|
|
log.info("telegram_bot_started")
|
|
|
|
while self._running:
|
|
try:
|
|
async with httpx.AsyncClient() as client:
|
|
resp = await client.get(
|
|
f"{self.api_url}/getUpdates",
|
|
params={"offset": self._offset, "timeout": 30},
|
|
timeout=35.0,
|
|
)
|
|
data = resp.json()
|
|
|
|
if not data.get("ok"):
|
|
await asyncio.sleep(5)
|
|
continue
|
|
|
|
for update in data.get("result", []):
|
|
self._offset = update["update_id"] + 1
|
|
response = await self.handle_update(update)
|
|
if response and "message" in update:
|
|
chat_id = update["message"]["chat"]["id"]
|
|
await self.send(chat_id, response)
|
|
|
|
except asyncio.TimeoutError:
|
|
pass
|
|
except Exception as e:
|
|
log.error("telegram_poll_error", error=str(e))
|
|
await asyncio.sleep(5)
|
|
|
|
async def start(self) -> None:
|
|
"""Start the bot poll loop."""
|
|
asyncio.create_task(self.poll_loop())
|
|
# Register hook listeners
|
|
from salior.hooks import global_hooks
|
|
|
|
async def on_fill_handler(event: HookEvent) -> None:
|
|
d = event.data
|
|
await self.send_alert(
|
|
f"✅ <b>Order Filled</b>\n"
|
|
f"{d['side'].upper()} {d['size']} {d['coin']} @ ${d['price']}\n"
|
|
f"Mode: {d['mode']}"
|
|
)
|
|
|
|
async def on_risk_handler(event: HookEvent) -> None:
|
|
await self.send_alert(f"⚠️ <b>Risk Breach</b>\n{event.data['reason']}")
|
|
|
|
async def on_error_handler(event: HookEvent) -> None:
|
|
await self.send_alert(f"🔴 <b>Agent Error</b>\n{event.data['agent']}: {event.data['error']}")
|
|
|
|
global_hooks.on("on_fill", on_fill_handler)
|
|
global_hooks.on("on_risk_breach", on_risk_handler)
|
|
global_hooks.on("on_error", on_error_handler)
|
|
|
|
|
|
# Hook emitter helpers — call these from agents
|
|
async def emit_signal(coin: str, regime: str, conviction: float, reasoning: str) -> None:
|
|
from salior.hooks import global_hooks
|
|
await global_hooks.emit(HookEvent(
|
|
name="on_signal",
|
|
source="signal_agent",
|
|
data={"coin": coin, "regime": regime, "conviction": conviction, "reasoning": reasoning},
|
|
))
|
|
|
|
|
|
async def emit_fill(coin: str, side: str, size: float, price: float, exec_id: str, mode: str) -> None:
|
|
from salior.hooks import global_hooks
|
|
await global_hooks.emit(HookEvent(
|
|
name="on_fill",
|
|
source="exec_agent",
|
|
data={"coin": coin, "side": side, "size": size, "price": price, "exec_id": exec_id, "mode": mode},
|
|
))
|
|
|
|
|
|
async def emit_risk_breach(reason: str, details: dict) -> None:
|
|
from salior.hooks import global_hooks
|
|
await global_hooks.emit(HookEvent(
|
|
name="on_risk_breach",
|
|
source="risk_agent",
|
|
data={"reason": reason, "details": details},
|
|
)) |