Flash Loans for AI Agents: Zero-Capital Arbitrage in One Transaction

import os, httpx, logging
from dataclasses import dataclass

log = logging.getLogger("flash-arb")
PF_KEY  = os.getenv("PURPLE_FLEA_KEY")
PF_BASE = "https://api.purpleflea.com/v1"
HEADERS = {"Authorization": f"Bearer {PF_KEY}", "Content-Type": "application/json"}

FLASH_LOAN_FEE = 0.0009  # 0.09% on Purple Flea flash loans
MIN_PROFIT_USD = 50       # don't execute if net profit < $50

@dataclass
class ArbOpportunity:
    asset: str
    amount: float           # units of asset to trade
    buy_venue: str
    buy_price: float
    sell_venue: str
    sell_price: float

    @property
    def gross_profit(self) -> float:
        return (self.sell_price - self.buy_price) * self.amount

    @property
    def loan_amount_usd(self) -> float:
        return self.buy_price * self.amount

    @property
    def flash_fee_usd(self) -> float:
        return self.loan_amount_usd * FLASH_LOAN_FEE

    @property
    def net_profit(self) -> float:
        return self.gross_profit - self.flash_fee_usd

def find_arb_opportunities(asset: str = "BTC") -> list[ArbOpportunity]:
    "Scan multiple DEX venues for price discrepancies."
    with httpx.Client() as http:
        prices = http.get(
            f"{PF_BASE}/flash-loans/prices",
            params={"asset": asset},
            headers=HEADERS
        ).json()["venues"]

    opportunities = []
    for i, buy_venue in enumerate(prices):
        for sell_venue in prices[i+1:]:
            if sell_venue["ask"] > buy_venue["bid"]:
                # Trade 1 unit to start; profitability calculator will size properly
                opp = ArbOpportunity(
                    asset=asset,
                    amount=1.0,
                    buy_venue=buy_venue["name"], buy_price=buy_venue["ask"],
                    sell_venue=sell_venue["name"], sell_price=sell_venue["bid"],
                )
                if opp.net_profit > MIN_PROFIT_USD:
                    opportunities.append(opp)
    return sorted(opportunities, key=lambda o: o.net_profit, reverse=True)

def execute_flash_arb(opp: ArbOpportunity) -> dict:
    "Submit a flash loan arbitrage transaction to Purple Flea."
    log.info(
        f"Executing arb: buy {opp.asset} on {opp.buy_venue} at ${opp.buy_price:,.2f}, "
        f"sell on {opp.sell_venue} at ${opp.sell_price:,.2f} | "
        f"net profit est. ${opp.net_profit:,.2f}"
    )
    payload = {
        "type": "arbitrage",
        "asset": opp.asset,
        "loan_amount_usd": opp.loan_amount_usd,
        "steps": [
            {"action": "buy",  "venue": opp.buy_venue,  "amount": opp.amount},
            {"action": "sell", "venue": opp.sell_venue, "amount": opp.amount},
        ],
        "max_slippage_bps": 10,   # abort if slippage > 0.10%
        "repay_asset": "USDC",
    }
    with httpx.Client(timeout=30) as http:
        r = http.post(f"{PF_BASE}/flash-loans/execute", json=payload, headers=HEADERS)
    r.raise_for_status()
    result = r.json()
    log.info(f"Arb completed: tx={result['tx_hash']} profit=${result['net_profit_usd']:.2f}")
    return result

def check_profitability(
    loan_usd: float,
    gross_spread_usd: float,
    slippage_bps: float = 10,    # 0.10% each side
    gas_usd: float = 5,
    flash_fee_bps: float = 9,     # 0.09%
) -> dict:
    flash_fee   = loan_usd * (flash_fee_bps / 10_000)
    slippage    = loan_usd * (slippage_bps / 10_000) * 2  # both legs
    total_costs = flash_fee + slippage + gas_usd
    net_profit  = gross_spread_usd - total_costs
    return {
        "gross_spread_usd": gross_spread_usd,
        "flash_fee_usd": flash_fee,
        "slippage_usd": slippage,
        "gas_usd": gas_usd,
        "net_profit_usd": net_profit,
        "is_profitable": net_profit > 0,
        "minimum_spread_for_profit_usd": total_costs,
        "minimum_spread_pct": total_costs / loan_usd * 100,
    }

# Example: need spread > 0.19% to profit on a $950k loan with 0.10% slippage
result = check_profitability(950_000, 4_000)
print(f"Net profit: ${result['net_profit_usd']:,.2f} | Profitable: {result['is_profitable']}")

def self_liquidate(position_id: str) -> dict:
    "Use a flash loan to repay own debt and avoid liquidation penalty."
    with httpx.Client() as http:
        pos = http.get(
            f"{PF_BASE}/lending/positions/{position_id}", headers=HEADERS
        ).json()

    debt_usd       = pos["debt_usd"]
    collateral_usd = pos["collateral_usd"]
    health_factor  = pos["health_factor"]

    if health_factor > 1.10:
        raise ValueError(f"Position health {health_factor:.2f} — not at risk, self-liquidation unnecessary")

    log.info(
        f"Self-liquidating position {position_id}: "
        f"debt=${debt_usd:,.0f} collateral=${collateral_usd:,.0f} "
        f"health={health_factor:.2f}"
    )
    payload = {
        "type": "self_liquidation",
        "position_id": position_id,
        "loan_asset": pos["debt_asset"],
        "loan_amount": debt_usd * 1.001,  # borrow slightly more for accrued interest
        "steps": [
            {"action": "repay_debt", "position_id": position_id},
            {"action": "withdraw_collateral", "position_id": position_id},
            {"action": "sell_collateral_to_repay"},  # auto-calculates amount needed
        ],
    }
    with httpx.Client(timeout=30) as http:
        r = http.post(f"{PF_BASE}/flash-loans/execute", json=payload, headers=HEADERS)
    r.raise_for_status()
    result = r.json()
    log.info(f"Self-liquidation complete. Saved: ${result['liquidation_penalty_avoided_usd']:,.2f}")
    return result

def swap_collateral(
    position_id: str,
    from_asset: str,    # e.g. "ETH"
    to_asset: str,      # e.g. "WBTC"
) -> dict:
    """
    Atomically swap collateral type on a lending position:
    1. Flash-borrow enough to_asset to repay the debt
    2. Withdraw from_asset collateral
    3. Swap from_asset → to_asset
    4. Re-deposit to_asset as new collateral
    5. Reborrow original debt amount
    6. Repay flash loan
    """
    payload = {
        "type": "collateral_swap",
        "position_id": position_id,
        "from_collateral_asset": from_asset,
        "to_collateral_asset": to_asset,
        "max_slippage_bps": 15,
    }
    with httpx.Client(timeout=30) as http:
        r = http.post(f"{PF_BASE}/flash-loans/execute", json=payload, headers=HEADERS)
    r.raise_for_status()
    result = r.json()
    log.info(
        f"Collateral swapped: {from_asset} → {to_asset} | "
        f"new LTV: {result['new_ltv']:.1%}"
    )
    return result

import time, logging

log = logging.getLogger("arb-scanner")
SCAN_ASSETS   = ["BTC", "ETH", "SOL", "USDC"]
SCAN_INTERVAL = 30   # seconds — arb opportunities close fast

def run_scanner():
    log.info("Flash loan arb scanner starting...")
    total_profit = 0.0
    executions   = 0

    while True:
        for asset in SCAN_ASSETS:
            try:
                opps = find_arb_opportunities(asset)
                if not opps:
                    continue
                best = opps[0]
                log.info(f"Opportunity: {asset} spread=${best.net_profit:.2f}")
                # Verify profitability with slippage estimate before executing
                check = check_profitability(best.loan_amount_usd, best.gross_profit)
                if check["is_profitable"] and check["net_profit_usd"] >= MIN_PROFIT_USD:
                    result = execute_flash_arb(best)
                    total_profit += result["net_profit_usd"]
                    executions   += 1
                    log.info(f"Running total: ${total_profit:,.2f} over {executions} trades")
            except Exception as e:
                log.error(f"Error scanning {asset}: {e}")
        time.sleep(SCAN_INTERVAL)

if __name__ == "__main__":
    run_scanner()