rmi-backend/app/liquidity_watchdog.py

"""
Liquidity Expiration Watchdog
==============================
Monitors tracked tokens for approaching liquidity lock expiries.
Maintains a Redis set of watched tokens with their lock expiry timestamps.
Provides check_expiring_locks() for cron/background use.

Redis keys:
  rmi:liquidity:tracked      -> Set of token_ids (e.g. "solana:So111111...")
  rmi:liquidity:token:{id}   -> Hash with token_address, chain, lock_expiry, liquidity_amount, created_at

  token_id format: "{chain}:{token_address}"
"""

import logging
import os
from datetime import UTC, datetime, timedelta
from typing import Any

logger = logging.getLogger("liquidity_watchdog")

# Redis key namespace
REDIS_NS = "rmi:liquidity"

# How far ahead (in hours) we consider "expiring soon"
EXPIRY_WINDOW_HOURS = 24

# Module-level Redis connection pool (reused across calls)
_redis_pool = None


def _get_redis():
    """Get a Redis client from the shared connection pool.

    Uses a module-level connection pool so connections are reused
    across calls rather than creating a new TCP connection each time.
    """
    import redis

    global _redis_pool
    if _redis_pool is None:
        _redis_pool = redis.ConnectionPool(
            host=os.getenv("REDIS_HOST", "localhost"),
            port=int(os.getenv("REDIS_PORT", "6379")),
            password=os.getenv("REDIS_PASSWORD", ""),
            decode_responses=True,
            socket_connect_timeout=5,
            socket_timeout=5,
        )
    return redis.Redis(connection_pool=_redis_pool)


def _token_id(token_address: str, chain: str) -> str:
    """Build a unique token ID for Redis keys."""
    return f"{chain}:{token_address}".lower()


def _token_hash_key(token_address: str, chain: str) -> str:
    """Redis hash key for a tracked token."""
    return f"{REDIS_NS}:token:{_token_id(token_address, chain)}"


def track_token(
    token_address: str,
    chain: str,
    lock_expiry_timestamp: int,
    liquidity_amount: float = 0.0,
) -> dict[str, Any]:
    """Register a token for liquidity expiry monitoring.

    Args:
        token_address: Contract/mint address of the token.
        chain: Blockchain name (e.g. 'solana', 'ethereum', 'bsc').
        lock_expiry_timestamp: Unix timestamp (seconds) when the lock expires.
        liquidity_amount: Locked liquidity amount in USD (or 0 if unknown).

    Returns:
        The stored token data dict.
    """
    r = _get_redis()
    token_id = _token_id(token_address, chain)
    hash_key = _token_hash_key(token_address, chain)

    data = {
        "token_address": token_address.lower(),
        "chain": chain.lower(),
        "lock_expiry": str(lock_expiry_timestamp),
        "liquidity_amount": str(liquidity_amount),
        "created_at": datetime.now(UTC).isoformat(),
    }

    r.hset(hash_key, mapping=data)
    r.sadd(f"{REDIS_NS}:tracked", token_id)

    logger.info(
        "Tracked token %s on %s, expiry=%d, liquidity=%.2f",
        token_address,
        chain,
        lock_expiry_timestamp,
        liquidity_amount,
    )

    return {
        "token_address": token_address.lower(),
        "chain": chain.lower(),
        "lock_expiry": lock_expiry_timestamp,
        "liquidity_amount": liquidity_amount,
        "created_at": data["created_at"],
    }


def untrack_token(token_address: str, chain: str) -> bool:
    """Remove a token from monitoring.

    Returns:
        True if the token was being tracked, False otherwise.
    """
    r = _get_redis()
    token_id = _token_id(token_address, chain)
    hash_key = _token_hash_key(token_address, chain)

    r.delete(hash_key)
    removed = r.srem(f"{REDIS_NS}:tracked", token_id)
    return removed > 0


def get_tracked_tokens() -> list[dict[str, Any]]:
    """List all currently tracked tokens with their stored data."""
    r = _get_redis()
    token_ids = r.smembers(f"{REDIS_NS}:tracked") or set()
    tokens = []

    for token_id in token_ids:
        hash_key = f"{REDIS_NS}:token:{token_id}"
        raw = r.hgetall(hash_key)
        if not raw:
            # Stale entry - clean up
            r.srem(f"{REDIS_NS}:tracked", token_id)
            continue

        tokens.append(
            {
                "token_address": raw.get("token_address", ""),
                "chain": raw.get("chain", ""),
                "lock_expiry": int(raw.get("lock_expiry", 0)),
                "liquidity_amount": float(raw.get("liquidity_amount", 0)),
                "created_at": raw.get("created_at", ""),
            }
        )

    return tokens


def check_expiring_locks() -> list[dict[str, Any]]:
    """Find all tracked tokens whose lock expiry is within EXPIRY_WINDOW_HOURS.

    Returns:
        List of dicts with:
          - token: token address
          - chain: chain name
          - lock_expiry: unix timestamp
          - hours_remaining: float hours until expiry
          - liquidity_amount: locked liquidity USD
    """
    now = datetime.now(UTC)
    cutoff = now + timedelta(hours=EXPIRY_WINDOW_HOURS)
    now_ts = now.timestamp()
    cutoff_ts = cutoff.timestamp()

    expiring: list[dict[str, Any]] = []
    tokens = get_tracked_tokens()

    for token_data in tokens:
        expiry = token_data.get("lock_expiry", 0)
        if expiry <= 0:
            continue

        # Check if expiry is within the window (and not already past)
        if now_ts < expiry <= cutoff_ts:
            hours_remaining = max(0.0, (expiry - now_ts) / 3600)
            expiring.append(
                {
                    "token": token_data["token_address"],
                    "chain": token_data["chain"],
                    "lock_expiry": expiry,
                    "hours_remaining": round(hours_remaining, 2),
                    "liquidity_amount": token_data.get("liquidity_amount", 0),
                }
            )

    # Sort by hours remaining (most urgent first)
    expiring.sort(key=lambda x: x["hours_remaining"])

    if expiring:
        logger.warning(
            "Found %d token(s) expiring within %d hours",
            len(expiring),
            EXPIRY_WINDOW_HOURS,
        )

    return expiring


def check_all_locks_summary() -> dict[str, Any]:
    """Get a summary of all tracked locks and their status.

    Returns a dict with counts and the list of expiring tokens.
    """
    tokens = get_tracked_tokens()
    now_ts = datetime.now(UTC).timestamp()

    already_expired = [t for t in tokens if t["lock_expiry"] <= now_ts]
    expiring_soon = check_expiring_locks()
    safe = [
        t
        for t in tokens
        if t["lock_expiry"] > now_ts
        and not any(e["token"] == t["token_address"] and e["chain"] == t["chain"] for e in expiring_soon)
    ]

    return {
        "total_tracked": len(tokens),
        "already_expired": len(already_expired),
        "expiring_soon": len(expiring_soon),
        "safe": len(safe),
        "expiring_tokens": expiring_soon,
    }