Building an Impermanent Loss Calculator and Position Monitor with AI

import os
import requests
import pandas as pd
from dotenv import load_dotenv

load_dotenv()  # Load GRAPH_API_KEY, DUNE_API_KEY from .env

GRAPH_ENDPOINT = "https://api.thegraph.com/subgraphs/name/uniswap/uniswap-v3"
DUNE_HEADERS = {"x-dune-api-key": os.getenv("DUNE_API_KEY")}

def fetch_position_from_graph(position_id):
    """Fetches real-time data for a specific Uniswap V3 LP position."""
    query = """
    query($id: String!) {
      position(id: $id) {
        id
        owner
        pool {
          id
          token0 { symbol, decimals }
          token1 { symbol, decimals }
          feeTier
        }
        liquidity
        tickLower { tickIdx }
        tickUpper { tickIdx }
        depositedToken0
        depositedToken1
        withdrawnToken0
        withdrawnToken1
        collectedFeesToken0
        collectedFeesToken1
        transaction { timestamp }
      }
    }
    """
    variables = {"id": position_id.lower()}
    response = requests.post(GRAPH_ENDPOINT, json={'query': query, 'variables': variables})
    data = response.json()
    return data['data']['position']

# il_calculator.py
from decimal import Decimal, getcontext
import math

getcontext().prec = 30  # High precision for financial math

def calculate_impermanent_loss(position_data, current_price, token0_decimals=18, token1_decimals=6):
    """
    Calculates impermanent loss for a Uniswap V3 position.
    position_data: Dict from The Graph query.
    current_price: Current price of token0 in terms of token1 (e.g., ETH/USDC price).
    """
    # 1. Extract initial deposits (adjust for decimals)
    init_token0 = Decimal(position_data['depositedToken0']) / (10 ** token0_decimals)
    init_token1 = Decimal(position_data['depositedToken1']) / (10 ** token1_decimals)

    # 2. Calculate "HOLD" portfolio value at current price
    hold_value = (init_token0 * Decimal(current_price)) + init_token1

    # 3. Calculate CURRENT position value (simplified - assumes price in range)
    # In production, you'd use the liquidity math to get precise token amounts.
    # This is a simplified estimator for the article.
    tick_lower = float(position_data['tickLower']['tickIdx'])
    tick_upper = float(position_data['tickUpper']['tickIdx'])
    price_lower = 1.0001 ** tick_lower
    price_upper = 1.0001 ** tick_upper

    if current_price <= price_lower:
        # Price below range: position is 100% token1
        current_value = init_token1
    elif current_price >= price_upper:
        # Price above range: position is 100% token0
        current_value = init_token0 * Decimal(current_price)
    else:
        # Price in range: use a proportional estimate
        # Note: For a real calculator, implement the exact Uniswap V3 liquidity formulas.
        proportion_in_token0 = (Decimal(math.sqrt(price_upper)) - Decimal(math.sqrt(current_price))) / (Decimal(math.sqrt(price_upper)) - Decimal(math.sqrt(price_lower)))
        current_token0 = init_token0 * proportion_in_token0
        current_token1 = init_token1 * (1 - proportion_in_token0)
        current_value = (current_token0 * Decimal(current_price)) + current_token1

    # 4. Add accrued fees (from position data)
    fees_token0 = Decimal(position_data['collectedFeesToken0']) / (10 ** token0_decimals)
    fees_token1 = Decimal(position_data['collectedFeesToken1']) / (10 ** token1_decimals)
    fees_value = (fees_token0 * Decimal(current_price)) + fees_token1
    total_current_value = current_value + fees_value

    # 5. Impermanent Loss Calculation
    if hold_value > 0:
        il_percent = ((total_current_value - hold_value) / hold_value) * 100
    else:
        il_percent = Decimal(0)

    return {
        "hold_portfolio_value": float(hold_value),
        "current_position_value": float(total_current_value),
        "impermanent_loss_percent": float(il_percent),
        "fee_income_value": float(fees_value)
    }

# Example usage
position = fetch_position_from_graph("0x...your_position_id...")
eth_price = 3500.0  # Fetch from an oracle like Uniswap pool itself
result = calculate_impermanent_loss(position, eth_price)
print(f"IL: {result['impermanent_loss_percent']:.2f}%, Fees Earned: ${result['fee_income_value']:.2f}")

# position_monitor.py
import schedule
import time
from datetime import datetime
import sqlite3

def monitor_job(position_id, pool_address):
    """Scheduled job to fetch, calculate, and store position health."""
    print(f"\n--- {datetime.utcnow().isoformat()} ---")

    # 1. Get live position state
    position = fetch_position_from_graph(position_id)
    # 2. Get current price from the pool itself (more reliable than external API)
    pool_data = fetch_pool_state(pool_address)  # Implement using The Graph pool query
    current_price = float(pool_data['token0Price'])

    # 3. Run Calculator
    metrics = calculate_impermanent_loss(position, current_price)

    # 4. Fetch pool benchmark from Dune (example: total daily fees)
    dune_query_id = 1234567  # Your pre-saved Dune query for this pool's daily fees
    pool_daily_fees = fetch_dune_data(dune_query_id)
    my_share_of_fees = estimate_fee_share(position, pool_daily_fees)

    # 5. Store & Alert
    store_metrics(position_id, datetime.utcnow(), metrics)
    check_alerts(metrics, my_share_of_fees)

    print(f"Price: ${current_price:.2f} | IL: {metrics['impermanent_loss_percent']:.2f}% | Net PnL: ${metrics['current_position_value'] - metrics['hold_portfolio_value']:.2f}")

# Schedule to run every 10 minutes
schedule.every(10).minutes.do(monitor_job, position_id="0x...", pool_address="0x...")

print("Starting Uniswap V3 Position Monitor...")
while True:
    schedule.run_pending()
    time.sleep(60)

# ai_alert_engine.py
import json
from continue_client import ContinueClient  # Hypothetical client for Continue.dev extension

def generate_ai_insight(metrics_history, pool_context):
    """
    Sends a weekly summary to an LLM for analysis and insight generation.
    """
    client = ContinueClient()

    prompt = f"""
    As a DeFi strategy analyst, review this liquidity position's weekly performance and context.
    Provide a concise, actionable insight (max 3 sentences). Focus on fee income vs impermanent loss trends and the provided market context.

    PERFORMANCE HISTORY (last 7 days):
    {json.dumps(metrics_history[-10:], indent=2)}  # Last 10 data points

    POOL & MARKET CONTEXT:
    - Pool Type: {pool_context['type']} (e.g., ETH/USDC 0.3%)
- 30-Day Avg Pool Fee APY: {pool_context['fee_apy']}%
- Current Price vs Position Range: {pool_context['price_vs_range']}
- Gas Cost to Rebalance: ${pool_context['rebalance_gas_cost']} (on Arbitrum)

    ANALYSIS REQUEST:
    1. Is fee income consistently covering impermanent loss?
    2. Based on the price vs range, is the position at risk of falling out-of-range?
    3. Suggest one specific action: HOLD, REBALANCE (widen/narrow range), or EXIT.
    """

    insight = client.complete(prompt)
    return insight

# Example rule-based alert
def check_alerts(metrics, fee_share):
    alerts = []
    # Rule 1: Out-of-range alert
    if metrics['current_position_value'] == metrics['hold_portfolio_value']:  # Simplified check for out-of-range
        alerts.append("CRITICAL: Position is out-of-range and earning 0 fees.")
        # Fix: Monitor price and be prepared to rebalance or withdraw when price exits range.

    # Rule 2: Fee income not covering IL
    if metrics['fee_income_value'] < abs(metrics['current_position_value'] - metrics['hold_portfolio_value']) and metrics['impermanent_loss_percent'] < 0:
        alerts.append("WARNING: Fee income is not covering impermanent loss.")

    # Rule 3: IL exceeds threshold (e.g., >5%)
    if metrics['impermanent_loss_percent'] < -5:
        alerts.append(f"WARNING: Impermanent loss ({metrics['impermanent_loss_percent']:.2f}%) exceeds 5% threshold.")

    for alert in alerts:
        print(f"[ALERT] {alert}")
        # In production, send to Discord/Slack/Telegram

if (days_out_of_range > 2 and rebalance_cost > projected_weekly_fees) or (net_apy_30d < yearn_benchmark_apy) or (abs(price_change) > 0.20 and il_percent > fee_yield_percent):
    send_alert("EXIT SIGNAL GENERATED", priority="HIGH")
    # Provide one-click link to withdraw on Uniswap interface