telegram_notifier.py

#!/usr/bin/env python3
"""
K1RL QUANT — V75_1S Signal Flip Notifier v2.1
Uses EXACT same Redis subscription as Rewards.py for guaranteed compatibility
Enhanced with state persistence, better error handling, and detailed notifications

v2.1 FIXES (aligned with Rewards.py):
  ✅ FIX #1: Store ably_realtime + signal_channel as INSTANCE vars (prevent GC)
  ✅ FIX #2: Capture asyncio event loop ref (Rewards.py FIX v5.2.1 pattern)
  ✅ FIX #3: Run send_telegram in thread executor (don't block Redis listener thread)
  ✅ FIX #4: Track _last_signal_time for health monitoring
  ✅ FIX #5: Ensure signal_keys is list (Rewards.py line 648)
  ✅ FIX #6: Add connection health monitoring like Rewards.py _health_monitor_loop
"""

import asyncio
import json
import logging
import os
import time
import traceback
import requests
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from datetime import datetime, timezone, timedelta
from typing import Optional

# EXACT same imports as Rewards.py
from redis_config_v75_1s import (
    REDIS_URL, REDIS_PASSWORD,
    REDIS_DB_FEATURES,
    CHANNEL_PREFIX, prefixed_channel
)
from redis_connection_manager import RedisAblyClient, RedisMessage

# ✅ ADD #1 — ICPE relay (same non-blocking pattern as Telegram sends)
from icpe_ingest_relay import IcpeRelay

# Enhanced logging with file output
LOG_DIR = Path('/home/user/app/logs')
LOG_DIR.mkdir(parents=True, exist_ok=True)

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [V75_1S-TG] %(levelname)s: %(message)s',
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler(LOG_DIR / 'telegram_notifier_v75.log', encoding='utf-8')
    ]
)
logger = logging.getLogger(__name__)

# Telegram config
BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "8473902481:AAE0wep-lSXJ9yamU0Sj2KyNTQj9MImA3fk")
CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "6224721972")
GAS_URL = os.environ.get("GAS_RELAY_URL", "https://script.google.com/macros/s/AKfycbzvhXbTq3Pc5-429QIn-wK6uqBH7Bm-2fLJLsi_uGuD5Htn_e1ca4xCSRskKRD-tVtWaA/exec")

# EXACT same channel as Rewards.py uses
ABLY_SIGNAL_CHANNEL = prefixed_channel("final_signals")  # → "V75_1S:final_signals"

# State persistence
STATE_FILE = Path('/home/user/app/data/v75_1s_flip_state.json')

# Timezone (Harare, Zimbabwe - UTC+2)
CAT = timezone(timedelta(hours=2))

# ✅ FIX #3: Thread pool for non-blocking Telegram sends
# _on_signal runs in the Redis listener THREAD — blocking it with requests.get()
# starves the listener and causes missed signals. Offload sends to a pool.
_tg_executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix="tg-send")

SIGNAL_TIMEOUT = 300  # 5 min without signals = problem


def now_cat():
    """Current time in CAT timezone"""
    return datetime.now(CAT)


def format_time():
    """Format current time as HH:MM:SS CAT"""
    return now_cat().strftime("%H:%M:%S CAT")


class V75_1SFlipNotifier:
    def __init__(self):
        self.last_signal = None
        self.flip_count = 0
        self.total_signals = 0
        self.session_start = time.time()
        self.last_flip_time = 0
        self.cooldown_seconds = 3  # Prevent spam

        # ✅ FIX #1: Instance-level refs (same as Rewards.py lines 541-542)
        self.ably_realtime: Optional[RedisAblyClient] = None
        self.signal_channel = None

        # ✅ FIX #2: Event loop ref (same as Rewards.py FIX v5.2.1 line 919)
        self._loop: Optional[asyncio.AbstractEventLoop] = None

        # ✅ FIX #4: Signal health tracking (same as Rewards.py)
        self._last_signal_time: float = 0
        self._connection_healthy: bool = False

        # Price reporter — tracks latest price from published signals
        # Updated in _on_signal on every valid signal received.
        self.last_price: float = 0.0

        # ✅ ADD #2 — ICPE relay: reads ASSET_SYMBOL + ICPE_INGEST_URL from HF Secrets
        self._icpe = IcpeRelay()

        # Load persistent state
        self._load_state()

    # ───────────────────────── state persistence ──────────────────────────

    def _save_state(self):
        """Save state to survive restarts"""
        try:
            STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
            state = {
                'last_signal': self.last_signal,
                'flip_count': self.flip_count,
                'total_signals': self.total_signals,
                'last_saved': time.time(),
                'symbol': 'V75_1S'
            }
            with open(STATE_FILE, 'w') as f:
                json.dump(state, f, indent=2)
            logger.debug(f"💾 State saved: {self.flip_count} flips, last signal: {self.last_signal}")
        except Exception as e:
            logger.error(f"❌ State save failed: {e}")

    def _load_state(self):
        """Load persistent state"""
        try:
            if STATE_FILE.exists():
                with open(STATE_FILE, 'r') as f:
                    state = json.load(f)

                if state.get('symbol') != 'V75_1S':
                    logger.info("🔄 Different symbol state detected — starting fresh")
                    return

                age_hours = (time.time() - state.get('last_saved', 0)) / 3600
                if age_hours < 24:
                    self.last_signal = state.get('last_signal')
                    self.flip_count = state.get('flip_count', 0)
                    self.total_signals = state.get('total_signals', 0)
                    logger.info(f"♻️ State restored: {self.flip_count} flips, last: {self.last_signal}, age: {age_hours:.1f}h")
                else:
                    logger.info(f"🗑️ Stale state ({age_hours:.1f}h old) — starting fresh")
        except Exception as e:
            logger.warning(f"⚠️ State load failed (starting fresh): {e}")

    # ───────────────────────── telegram ────────────────────────────────

    def _send_telegram_sync(self, text: str) -> bool:
        """Send message to Telegram — runs in thread pool, never blocks listener."""
        try:
            branded_text = f"📊 <b>V75_1S</b> 〰️ {text}"
            params = {
                'bot_token': BOT_TOKEN,
                'chat_id': CHAT_ID,
                'text': branded_text,
                'parse_mode': 'HTML'
            }
            response = requests.get(GAS_URL, params=params, timeout=10)
            response.raise_for_status()
            result = response.json()
            if result.get('ok'):
                logger.info(f"✅ Telegram sent: {text[:50]}...")
                return True
            else:
                logger.error(f"❌ Telegram API error: {result}")
                return False
        except requests.exceptions.RequestException as e:
            logger.error(f"❌ Telegram network error: {e}")
            return False
        except Exception as e:
            logger.error(f"❌ Telegram error: {e}")
            return False

    def send_telegram(self, text: str):
        """
        ✅ FIX #3: Non-blocking send.
        If called from the Redis listener thread (inside _on_signal), we offload
        to the thread pool so the listener can keep consuming messages.
        """
        _tg_executor.submit(self._send_telegram_sync, text)

    # ───────────────────────── signal processing ──────────────────────

    def _extract_agents(self, signal_keys):
        """Extract agent timeframes from signal keys"""
        agents = []
        for key in signal_keys:
            key_str = str(key).lower()
            for tf in ['15m', '10m', '5m', '2m', '1m', '30s', '15s', '5s']:
                if tf in key_str:
                    agents.append(tf)
                    break
        return list(set(agents))

    def _on_signal(self, message: RedisMessage):
        """
        Process signal — EXACT same extraction as Rewards.py _on_signal (lines 616-689).

        ✅ This callback runs in the Redis listener THREAD (RedisAblyClient V10.1).
        All work here must be synchronous. Telegram sends are offloaded to _tg_executor.
        """
        try:
            self.total_signals += 1
            self._last_signal_time = time.time()   # ✅ FIX #4

            # ── data extraction (Rewards.py lines 621-629) ──
            data = message.data if isinstance(message, RedisMessage) else message

            # Handle nested envelope: {"event": "message", "data": {...}}
            if isinstance(data, dict) and 'data' in data:
                data = data['data']

            # Parse if string
            if isinstance(data, str):
                data = json.loads(data)

            # ── field extraction (Rewards.py lines 632-634) ──
            action = data.get('final_action', data.get('action', '')).upper()
            price = data.get('price', 0.0)
            signal_keys = data.get('signal_keys', [])

            # ✅ FIX #5: Ensure signal_keys is list (Rewards.py line 648)
            if not isinstance(signal_keys, list):
                signal_keys = [str(signal_keys)]

            # ── validation (Rewards.py lines 636-642) ──
            if action not in ['BUY', 'SELL']:
                logger.debug(f"⚠️ Invalid action: {action}")
                return
            if not price or price == 0.0:
                logger.warning(f"⚠️ Invalid price: {price}")
                return

            logger.info(
                f"🔔 Signal #{self.total_signals}: {action} @ {price:.5f} "
                f"| Keys: {len(signal_keys)} "
                f"| Loop: {'✅' if self._loop and self._loop.is_running() else '❌'}"
            )

            # ✅ ADD #3a — capture state BEFORE any modification
            #   _prev / _is_flip / agents hoisted so RICPE gets context
            #   even on non-flip (continuation) signals.
            _prev    = self.last_signal or ""
            _is_flip = bool(self.last_signal and self.last_signal != action)
            agents   = self._extract_agents(signal_keys)   # hoisted from flip block

            # ── flip detection ──
            if self.last_signal and self.last_signal != action:
                # Cooldown check
                time_since_flip = time.time() - self.last_flip_time
                if time_since_flip < self.cooldown_seconds:
                    logger.info(f"⏳ Flip cooldown: {self.cooldown_seconds - time_since_flip:.1f}s remaining")
                    self.last_signal = action
                    return

                self.flip_count += 1
                self.last_flip_time = time.time()

                logger.warning(f"🚨 FLIP #{self.flip_count}: {self.last_signal} → {action} @ {price:.5f}")

                # agents already computed above (hoisted)
                agents_str = ', '.join(a.upper() for a in agents) if agents else 'UNKNOWN'
                session_minutes = (time.time() - self.session_start) / 60

                # V75_1S directional indicators
                old_color = "🟢" if self.last_signal == "BUY" else "🔴"
                old_arrow = "📈" if self.last_signal == "BUY" else "📉"
                new_color = "🟢" if action == "BUY" else "🔴"
                new_arrow = "📈" if action == "BUY" else "📉"

                flip_msg = (
                    f"🚨 <b>VOLATILITY 75_1S SIGNAL FLIP</b>\n\n"
                    f"{old_color} <b>{self.last_signal}</b> {old_arrow} ➤ "
                    f"{new_color} <b>{action}</b> {new_arrow}\n\n"
                    f"💰 <b>Price:</b> {price:.5f}\n"
                    f"⚡ <b>Agent:</b> {agents_str}\n"
                    f"🎯 <b>Flip #:</b> {self.flip_count}\n"
                    f"📊 <b>Tick Rate:</b> 1 tick per second\n\n"
                    f"<b>📈 SESSION STATS</b>\n"
                    f"• Total Signals: {self.total_signals}\n"
                    f"• Session: {session_minutes:.0f} minutes\n"
                    f"• Keys: {len(signal_keys)}\n\n"
                    f"<b>🕐 TIMING</b>\n"
                    f"{format_time()}\n\n"
                    f"<i>Volatility 75_1S — continuous random-walk, 75% annualised vol 〰️</i>"
                )

                # ✅ FIX #3: Non-blocking — offloaded to thread pool
                self.send_telegram(flip_msg)

                # Save state after flip
                self._save_state()

            # Update last signal
            self.last_signal = action
            self.last_price  = price

            # ✅ ADD #3b — relay full payload to RICPE (fire-and-forget)
            #   Fires for EVERY valid signal (not just flips) so RICPE builds the
            #   complete momentum / frequency / stability picture for PIL.
            self._icpe.push(
                action        = action,
                price         = price,
                prev_action   = _prev,
                is_flip       = _is_flip,
                flip_number   = self.flip_count,
                agent         = agents[0].upper() if agents else "",
                keys          = len(signal_keys),
                total_signals = self.total_signals,
                session_start = self.session_start,
                raw_data      = data,
            )

        except Exception as e:
            logger.error(f"❌ Signal processing error: {e}")
            traceback.print_exc()

    # ───────────────────────── health monitor ─────────────────────────

    async def _health_monitor_loop(self):
        """
        ✅ FIX #6: Connection health monitoring (same pattern as Rewards.py).
        Detects stale connections and logs warnings.
        """
        while True:
            try:
                await asyncio.sleep(60)

                # Check Redis connection state
                if self.ably_realtime:
                    try:
                        redis_state = self.ably_realtime.connection.state
                        self._connection_healthy = redis_state in ('connected', 'ready')
                    except Exception:
                        self._connection_healthy = False
                else:
                    self._connection_healthy = False

                # Check signal freshness
                signal_age = time.time() - self._last_signal_time if self._last_signal_time else float('inf')
                if self._last_signal_time and signal_age > SIGNAL_TIMEOUT:
                    logger.warning(
                        f"⚠️ No signals for {signal_age:.0f}s "
                        f"(threshold: {SIGNAL_TIMEOUT}s) — "
                        f"Healthy={'✅' if self._connection_healthy else '❌'}"
                    )

            except asyncio.CancelledError:
                break
            except Exception as e:
                logger.error(f"❌ Health monitor error: {e}")

    # ───────────────────────── 10-min price reporter ──────────────────────────

    async def _price_report_loop(self):
        """
        Every 10 minutes send a Telegram message showing the current price
        from the latest published signal, plus a brief session snapshot.

        Fires on the 10-minute mark regardless of flip activity — gives the
        trader a heartbeat even during quiet periods.
        """
        await asyncio.sleep(600)   # wait 10 min before first report
        while True:
            try:
                signal_age = time.time() - self._last_signal_time if self._last_signal_time else -1
                session_minutes = (time.time() - self.session_start) / 60

                direction_color = "🟢" if self.last_signal == "BUY" else (
                    "🔴" if self.last_signal == "SELL" else "⚪"
                )
                direction_arrow = "📈" if self.last_signal == "BUY" else (
                    "📉" if self.last_signal == "SELL" else "➖"
                )

                if self.last_price and self.last_price > 0:
                    price_line = f"💰 <b>Price:</b> {self.last_price:.5f}"
                else:
                    price_line = "💰 <b>Price:</b> —"

                # BONUS-1 — ICPE relay health for 10-min report
                icpe_stats = self._icpe.stats()
                icpe_line  = (
                    f"• RICPE relay: ✅ {icpe_stats['sends_ok']} ok / "
                    f"{icpe_stats['sends_fail']} fail"
                    if icpe_stats['enabled']
                    else "• RICPE relay: ⚠️ disabled (ICPE_INGEST_URL not set)"
                )

                age_str = (
                    f"{signal_age:.0f}s ago" if signal_age >= 0
                    else "no signal yet"
                )

                report_msg = (
                    f"🕐 <b>VOLATILITY 75_1S — 10-MIN UPDATE</b>\n\n"
                    f"<b>📊 CURRENT SIGNAL</b>\n"
                    f"{direction_color} <b>Direction:</b> {self.last_signal or '—'} {direction_arrow}\n"
                    f"{price_line}\n"
                    f"🕰 <b>Signal Age:</b> {age_str}\n\n"
                    f"<b>📈 SESSION</b>\n"
                    f"• Signals: {self.total_signals}\n"
                    f"• Flips: {self.flip_count}\n"
                    f"• Runtime: {session_minutes:.0f} minutes\n\n"
                    f"<b>🩺 CONNECTION</b>\n"
                    f"• Redis: {'✅ Active' if self._connection_healthy else '❌ Down'}\n"
                    f"{icpe_line}\n\n"
                    f"<b>🕐 TIME</b>\n"
                    f"{format_time()}\n\n"
                    f"<i>Next update in 10 minutes ⏱</i>"
                )

                self.send_telegram(report_msg)
                logger.info(
                    f"📢 [10-min report] {self.last_signal or '—'} @ "
                    f"{self.last_price:.5f} | "
                    f"Signals: {self.total_signals} | "
                    f"Flips: {self.flip_count}"
                )

            except asyncio.CancelledError:
                break
            except Exception as e:
                logger.error(f"❌ Price report error: {e}")

            await asyncio.sleep(600)   # 10 minutes

    # ───────────────────────── main loop ──────────────────────────────

    async def run(self):
        """Run the notifier with enhanced monitoring"""

        # ✅ FIX #2: Capture event loop (Rewards.py line 919)
        self._loop = asyncio.get_running_loop()

        logger.info("=" * 60)
        logger.info("🚀 K1RL QUANT — V75_1S Signal Flip Notifier v2.1")
        logger.info(f"   Channel:   {ABLY_SIGNAL_CHANNEL}")
        logger.info(f"   Database:  {REDIS_DB_FEATURES}")
        logger.info(f"   Method:    Same as Rewards.py (RedisAblyClient V10.1)")
        logger.info(f"   Started:   {format_time()}")
        logger.info(f"   State:     {self.flip_count} flips, last: {self.last_signal or '—'}")
        logger.info(f"   Loop:      captured={'✅' if self._loop else '❌'}")
        logger.info(f"   IcpeRelay: {self._icpe.stats()}")
        logger.info("=" * 60)

        try:
            # ✅ FIX #1: INSTANCE-level refs (Rewards.py lines 557-565)
            # Local vars could be GC'd if the reference is lost.
            self.ably_realtime = RedisAblyClient(
                redis_url=REDIS_URL,
                password=REDIS_PASSWORD,
                use_streams=True,
                database=REDIS_DB_FEATURES  # DB 0 (V75_1S features)
            )

            self.signal_channel = self.ably_realtime.channels.get(ABLY_SIGNAL_CHANNEL)

            logger.info(f"✅ Redis channels initialized (V75_1S — prefix='{CHANNEL_PREFIX}', DB={REDIS_DB_FEATURES})")
            logger.info(f"   Signal: {ABLY_SIGNAL_CHANNEL}")

            # ✅ Subscribe (Rewards.py line 941)
            await self.signal_channel.subscribe("message", self._on_signal)
            logger.info(f"✅ Subscribed to {ABLY_SIGNAL_CHANNEL}:message (robust V10.1 listener)")
            logger.info(f"   Event loop captured: {self._loop is not None} | Running: {self._loop.is_running() if self._loop else False}")

            # Send startup notification
            startup_msg = (
                f"🚀 <b>VOLATILITY 75_1S MONITOR ONLINE</b>\n\n"
                f"<b>📊 CONFIGURATION</b>\n"
                f"• Symbol: Volatility 75_1S Index\n"
                f"• Tick Rate: 1 tick per second\n"
                f"• Volatility Type: Continuous random-walk volatility\n"
                f"• Target Agent: 10M\n"
                f"• Window: 5 minutes\n"
                f"• Cache: 2000 slots\n\n"
                f"<b>🎯 MISSION</b>\n"
                f"Monitoring V75_1S BUY↔SELL flips\n"
                f"Real-time momentum signal detection active\n\n"
                f"<b>🕐 STATUS</b>\n"
                f"Started: {format_time()}\n"
                f"Mode: Volatility 75_1S Signal Detection\n\n"
                f"<i>Watching for the next directional flip ⚡</i>"
            )
            self.send_telegram(startup_msg)

            # ✅ FIX #6: Start health monitor
            health_task = asyncio.create_task(self._health_monitor_loop())

            # 10-minute price reporter
            price_report_task = asyncio.create_task(self._price_report_loop())

            # Main loop with periodic status
            logger.info("✅ Monitoring for V75_1S signal flips...")
            status_counter = 0

            while True:
                await asyncio.sleep(60)
                status_counter += 1

                session_minutes = (time.time() - self.session_start) / 60
                signal_age = time.time() - self._last_signal_time if self._last_signal_time else -1
                icpe_stats = self._icpe.stats()
                logger.info(
                    f"📊 [{format_time()}] Status: {self.total_signals} signals | "
                    f"{self.flip_count} flips | Last: {self.last_signal or '—'} | "
                    f"SignalAge: {signal_age:.0f}s | "
                    f"Healthy: {'✅' if self._connection_healthy else '❌'} | "
                    f"ICPE: {icpe_stats['sends_ok']}ok/{icpe_stats['sends_fail']}fail | "
                    f"Runtime: {session_minutes:.0f}m"
                )

                # Detailed status every 30 minutes
                if status_counter % 30 == 0:
                    status_msg = (
                        f"📊 <b>VOLATILITY 75_1S STATUS REPORT</b>\n\n"
                        f"<b>📈 SESSION</b>\n"
                        f"• Signals: {self.total_signals} received\n"
                        f"• Flips: {self.flip_count} detected\n"
                        f"• Last Signal: {self.last_signal or '—'}\n"
                        f"• Runtime: {session_minutes:.0f} minutes\n\n"
                        f"<b>🩺 HEALTH</b>\n"
                        f"• Connection: {'✅ Active' if self._connection_healthy else '❌ Down'}\n\n"
                        f"<b>🕐 TIMING</b>\n"
                        f"{format_time()}\n\n"
                        f"<i>V75_1S signal monitoring continues...</i>"
                    )
                    self.send_telegram(status_msg)

        except KeyboardInterrupt:
            logger.info("🛑 Stopped by user")
            self._send_telegram_sync(
                f"🛑 <b>VOLATILITY 75_1S MONITOR OFFLINE</b>\n\n"
                f"<b>📊 SESSION SUMMARY</b>\n"
                f"• Signals Processed: {self.total_signals}\n"
                f"• Flips Detected: {self.flip_count}\n"
                f"• Last Signal: {self.last_signal or '—'}\n\n"
                f"<b>🕐 SHUTDOWN</b>\n"
                f"{format_time()}\n\n"
                f"<i>V75_1S signal monitoring paused</i>"
            )
        except Exception as e:
            logger.error(f"❌ Fatal error: {e}", exc_info=True)
            self._send_telegram_sync(f"❌ <b>Error:</b> {str(e)[:200]}...")
        finally:
            self._save_state()
            if self.ably_realtime:
                try:
                    self.ably_realtime.close()
                except Exception:
                    pass
            _tg_executor.shutdown(wait=False)
            # ✅ ADD #4 — clean up ICPE thread pool
            self._icpe.shutdown()
            logger.info(f"✅ Shutdown complete. Final state: {self.flip_count} flips, {self.total_signals} signals")


if __name__ == "__main__":
    notifier = V75_1SFlipNotifier()
    asyncio.run(notifier.run())