Stop your positions from
being liquidated

{
  "scope": "most_at_risk",  // "most_at_risk" | "all" | "above_leverage"
  "leverage_threshold": 10,  // used when scope = "above_leverage"
  "reason": "auto_danger_zone",
  "dry_run": false  // set true to simulate without executing
}

import asyncio
import aiohttp
import logging
from dataclasses import dataclass, field
from typing import Optional

logger = logging.getLogger("liquidation_guard")

@dataclass
class GuardConfig:
    api_key: str
    warning_hf: float = 1.6
    danger_hf: float = 1.2
    target_hf: float = 2.0
    max_topup_usdc: float = 500.0
    reserve_minimum_usdc: float = 100.0
    poll_interval_sec: float = 5.0
    webhook_url: Optional[str] = None

class LiquidationGuard:
    """
    Monitors all open positions on Purple Flea and automatically
    tops up collateral or triggers kill-switch when health factor
    drops below configured thresholds.
    """

    BASE_URL = "https://api.purpleflea.com/api/v1"

    def __init__(self, config: GuardConfig):
        self.config = config
        self.headers = {
            "Authorization": f"Bearer {config.api_key}",
            "Content-Type": "application/json"
        }
        self._running = False

    async def _get_positions(self, session):
        async with session.get(
            f"{self.BASE_URL}/trading/positions",
            headers=self.headers
        ) as r:
            data = await r.json()
            return data.get("positions", [])

    async def _get_wallet_balance(self, session) -> float:
        async with session.get(
            f"{self.BASE_URL}/wallet/balance",
            headers=self.headers
        ) as r:
            data = await r.json()
            return float(data.get("usdc_balance", 0))

    async def _top_up(self, session, position_id: str, amount_usdc: float):
        async with session.post(
            f"{self.BASE_URL}/trading/positions/{position_id}/add-collateral",
            headers=self.headers,
            json={"amount_usdc": amount_usdc}
        ) as r:
            return await r.json()

    async def _kill_switch(self, session, scope="most_at_risk"):
        logger.critical("Triggering kill switch — scope: %s", scope)
        async with session.post(
            f"{self.BASE_URL}/trading/kill-switch/trigger",
            headers=self.headers,
            json={"scope": scope, "reason": "liquidation_guard"}
        ) as r:
            return await r.json()

    async def _handle_position(self, session, pos):
        hf    = float(pos["health_factor"])
        pid   = pos["id"]
        mkt   = pos["market"]
        collat = float(pos["collateral_usdc"])

        if hf >= self.config.warning_hf:
            return  # Safe zone — no action

        wallet_bal = await self._get_wallet_balance(session)
        available  = wallet_bal - self.config.reserve_minimum_usdc

        if hf < self.config.danger_hf or available <= 0:
            logger.warning("[%s] HF %.2f in DANGER — triggering kill switch", mkt, hf)
            await self._kill_switch(session)
            return

        # Warning zone: calculate top-up to reach target HF
        needed_collat = collat * (self.config.target_hf / hf - 1)
        topup_amount  = min(needed_collat, self.config.max_topup_usdc, available)

        logger.info("[%s] HF %.2f — topping up %.2f USDC", mkt, hf, topup_amount)
        result = await self._top_up(session, pid, topup_amount)
        logger.info("[%s] New HF: %.2f", mkt, result.get("new_health_factor", "?"))

    async def run(self):
        """Start continuous monitoring loop."""
        self._running = True
        logger.info("LiquidationGuard started (poll interval: %.1fs)",
                    self.config.poll_interval_sec)

        async with aiohttp.ClientSession() as session:
            while self._running:
                try:
                    positions = await self._get_positions(session)
                    tasks = [self._handle_position(session, p) for p in positions]
                    await asyncio.gather(*tasks, return_exceptions=True)
                except Exception as e:
                    logger.error("Monitor loop error: %s", e)
                await asyncio.sleep(self.config.poll_interval_sec)

    def stop(self):
        self._running = False


# ── Usage example ──────────────────────────────────────────
if __name__ == "__main__":
    import os

    config = GuardConfig(
        api_key=os.environ["PURPLE_FLEA_API_KEY"],
        warning_hf=1.6,
        danger_hf=1.2,
        target_hf=2.0,
        max_topup_usdc=500.0,
        reserve_minimum_usdc=100.0,
        poll_interval_sec=5.0
    )
    guard = LiquidationGuard(config)
    asyncio.run(guard.run())