import json
from fastapi.responses import HTMLResponse
import os
import requests
import time
import sqlite3
from concurrent.futures import ThreadPoolExecutor, as_completed
from fastapi.staticfiles import StaticFiles
from fastapi import FastAPI, HTTPException
from datetime import datetime, timezone
app = FastAPI(
title="Nahini DeFi Monitor",
version="1.4.1"
)
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
VFAT_FEE_DB = os.path.join(BASE_DIR, "vfat_fee_history.sqlite3")
VFAT_FEE_SNAPSHOT_INTERVAL = 300 # 5 minutes
VFAT_FEE_HISTORY_DAYS = 30
POSITION_HISTORY_DB = os.path.join(BASE_DIR, "position_history.sqlite3")
POSITION_SNAPSHOT_INTERVAL = 300 # 5 minutes
POSITION_HISTORY_DAYS = 30
app.mount(
"/static",
StaticFiles(directory=os.path.join(BASE_DIR, "static")),
name="static"
)
WALLET = "0x4cbb5995463fc5128272385b18bf3a6f939c0548"
KRYSTAL_URL = "https://cloud-api.krystal.app/v1/positions"
KRYSTAL_BASE_URL = "https://cloud-api.krystal.app"
# vfat Metrics public read-only data source.
# Provider-owned endpoint used by vfat Metrics. Keep failures isolated from Krystal.
VFAT_INFO_URL = "https://info-api.vf.at"
REVERT_ACCOUNT_URL = f"https://api.revert.finance/v1/positions/account/{WALLET}?limit=100&active=true&with-v4=true&with-ekubo=true"
VFAT_CACHE_TTL = 60
_vfat_cache = {"checkedAt": 0, "positions": []}
# Robinhood Chain - direct on-chain reader for the vfat-managed Uniswap V3 LP.
# Kept isolated so an RPC failure never breaks Krystal or Base vfat positions.
BASE_RPC = "https://mainnet.base.org"
ROBINHOOD_RPC = "https://rpc.mainnet.chain.robinhood.com"
ETHEREUM_RPC = "https://ethereum-rpc.publicnode.com"
HYPEREVM_RPC = "https://rpc.hyperliquid.xyz/evm"
ROBINHOOD_CHAIN_ID = 4663
ROBINHOOD_POSITION_MANAGER = "0x73991a25c818bf1f1128deaab1492d45638de0d3"
ROBINHOOD_POOL = "0xd30e44aae604b42a63f6f9a8109fd0408f35b9fb"
ROBINHOOD_SICKLE = "0xed482e7ebaabeabf1e6589db29abea079a90c17b"
ROBINHOOD_NFT_ID = 1176825
# Cache transaction-history lookups so the dashboard does not spend API credits
# on every 60-second refresh. A position is refreshed when Krystal reports a
# newer lastUpdateBlockTime, or when the cached lookup is older than 6 hours.
AUTOMATION_CACHE_TTL = 6 * 60 * 60
_last_action_cache = {}
# Cache normalized Krystal positions for 5 minutes so dashboard refreshes do
# not consume Krystal Cloud credits every minute.
KRYSTAL_POSITIONS_CACHE_TTL = 5 * 60
_krystal_positions_cache = {"checkedAt": 0, "positions": []}
# Persistent last-known-good Krystal positions. This survives service
# restarts so a temporary Krystal API failure does not hide positions.
KRYSTAL_LKG_FILE = os.path.join(os.path.dirname(__file__), "krystal_lkg.json")
def _save_krystal_lkg(positions):
if not positions:
return
tmp = KRYSTAL_LKG_FILE + ".tmp"
with open(tmp, "w", encoding="utf-8") as f:
json.dump(positions, f)
os.replace(tmp, KRYSTAL_LKG_FILE)
def _load_krystal_lkg():
try:
with open(KRYSTAL_LKG_FILE, "r", encoding="utf-8") as f:
positions = json.load(f)
return positions if isinstance(positions, list) else []
except (OSError, ValueError, TypeError):
return []
def fetch_open_positions(api_key):
"""
Fetch OPEN positions from Krystal Cloud API.
Krystal's offset pagination currently returns overlapping/partial
results. The first page contains the current supplementary positions
needed by this dashboard. Revert remains authoritative for positions
available from both providers.
"""
response = requests.get(
KRYSTAL_URL,
params={
"wallet": WALLET,
"positionStatus": "OPEN",
"page": 1,
"pageSize": 20,
},
headers={
"Kc-Apikey": api_key,
"Accept": "application/json",
},
timeout=30,
)
response.raise_for_status()
positions = response.json()
if not isinstance(positions, list):
raise ValueError(
"Unexpected response format from Krystal API"
)
# Defensive dedupe in case Krystal returns duplicate rows.
deduped = []
seen = set()
for position in positions:
chain_id = str((position.get("chain") or {}).get("id") or "")
token_id = str(position.get("tokenId") or "")
position_id = str(position.get("id") or "")
key = (
("nft", chain_id, token_id)
if chain_id and token_id
else ("id", position_id)
)
if key in seen:
continue
seen.add(key)
deduped.append(position)
return deduped
def get_api_key():
api_key = os.getenv("KRYSTAL_API_KEY")
if not api_key:
raise HTTPException(
status_code=500,
detail="KRYSTAL_API_KEY is not configured"
)
return api_key
def load_positions():
api_key = get_api_key()
try:
return fetch_open_positions(api_key)
except (requests.RequestException, ValueError) as exc:
raise HTTPException(
status_code=502,
detail=f"Krystal API error: {exc}"
)
def get_pair(p):
amounts = p.get("currentAmounts") or []
symbols = [
x.get("token", {}).get("symbol", "?")
for x in amounts
]
return "/".join(symbols)
def get_pending_fees(p):
trading_fee = p.get("tradingFee") or {}
return sum(
x.get("value", 0) or 0
for x in trading_fee.get("pending", [])
)
def get_claimed_fees(p):
"""
Estimate claimed trading fees in USD using each token's CURRENT price.
Krystal's claimed fee objects contain token balance + current token price,
but not the historical USD value at the exact time the fee was claimed.
"""
trading_fee = p.get("tradingFee") or {}
total = 0.0
for item in trading_fee.get("claimed", []):
token = item.get("token") or {}
decimals = token.get("decimals", 18)
balance = item.get("balance", "0")
price = item.get("price", 0) or 0
try:
token_amount = int(balance or "0") / (10 ** int(decimals))
total += token_amount * float(price)
except (TypeError, ValueError, OverflowError):
continue
return total
def get_total_fees(p):
return get_pending_fees(p) + get_claimed_fees(p)
def _extract_transaction_list(payload):
"""Return the most likely transaction list from Krystal's response."""
if isinstance(payload, list):
return [x for x in payload if isinstance(x, dict)]
if not isinstance(payload, dict):
return []
for key in ("transactions", "items", "data", "result", "history"):
value = payload.get(key)
if isinstance(value, list):
return [x for x in value if isinstance(x, dict)]
if isinstance(value, dict):
nested = _extract_transaction_list(value)
if nested:
return nested
# Some APIs return an object keyed by transaction hash/id.
values = [x for x in payload.values() if isinstance(x, dict)]
return values if values else []
def _first_recursive(obj, keys):
"""Find the first non-empty value for any key in a nested object."""
if isinstance(obj, dict):
for key in keys:
if key in obj and obj[key] not in (None, ""):
return obj[key]
for value in obj.values():
found = _first_recursive(value, keys)
if found not in (None, ""):
return found
elif isinstance(obj, list):
for value in obj:
found = _first_recursive(value, keys)
if found not in (None, ""):
return found
return None
def _collect_action_strings(obj, depth=0):
"""Collect action-like text without serializing the whole response."""
if depth > 4:
return []
wanted = {
"action", "actiontype", "type", "transactiontype", "txntype",
"event", "method", "operation", "name", "label", "category",
"ordertypename", "ordertype"
}
values = []
if isinstance(obj, dict):
for key, value in obj.items():
if key.lower() in wanted and isinstance(value, (str, int)):
values.append(str(value))
elif isinstance(value, (dict, list)):
values.extend(_collect_action_strings(value, depth + 1))
elif isinstance(obj, list):
for value in obj:
values.extend(_collect_action_strings(value, depth + 1))
return values
def _normalize_action(tx):
text = " ".join(_collect_action_strings(tx)).upper()
# Keep the most useful LP-management actions first.
mappings = (
(("REBALANCE", "ADJUST_RANGE"), "Rebalance"),
(("COMPOUND",), "Compound"),
(("HARVEST",), "Harvest"),
(("AUTO_EXIT", "EXIT"), "Exit"),
(("CLAIM", "COLLECT"), "Claim Fees"),
(("INCREASE", "ADD_LIQUIDITY", "MINT"), "Increase Liquidity"),
(("REMOVE", "DECREASE", "WITHDRAW", "BURN"), "Remove Liquidity"),
)
action = None
for keywords, label in mappings:
if any(k in text for k in keywords):
action = label
break
if action is None:
return None, None
# Only call it automated when Krystal's payload explicitly says so.
is_automation = any(
k in text for k in (
"AUTO", "AUTOMATION", "ORDER_TYPE_REBALANCE",
"ORDER_TYPE_COMPOUND", "ORDER_TYPE_HARVEST", "AUTO_EXIT"
)
)
return action, is_automation
def _normalize_timestamp(value):
if value in (None, ""):
return None
try:
if isinstance(value, (int, float)) or str(value).isdigit():
n = float(value)
# milliseconds -> seconds
if n > 10_000_000_000:
n /= 1000.0
return int(n)
dt = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return int(dt.timestamp())
except (TypeError, ValueError, OverflowError):
return None
def fetch_last_lp_action(p, api_key):
"""
Fetch and classify the newest LP-management transaction from Krystal.
Krystal can emit multiple history rows for one on-chain transaction. Group
rows by txHash before classifying them. A COLLECT_FEE + DEPOSIT combination
in the same transaction is an Auto Compound pattern: fees are collected and
redeposited into the LP in one atomic transaction.
"""
chain_id = p.get("chain", {}).get("id")
position_id = p.get("id")
if not chain_id or not position_id:
return {
"action": None,
"timestamp": None,
"isAutomation": None,
"txHash": None
}
url = f"{KRYSTAL_BASE_URL}/v1/positions/{chain_id}/{position_id}/transactions"
response = requests.get(
url,
params={"wallet": WALLET},
headers={
"Kc-Apikey": api_key,
"Accept": "application/json"
},
timeout=20
)
response.raise_for_status()
txs = _extract_transaction_list(response.json())
# Group Krystal history rows that belong to the same on-chain transaction.
groups = {}
for index, tx in enumerate(txs):
tx_hash = _first_recursive(tx, ("txHash", "transactionHash", "hash"))
ts = _normalize_timestamp(_first_recursive(
tx,
(
"timestamp", "blockTime", "blockTimestamp", "createdAt",
"updatedAt", "executedAt", "transactionTime", "time"
)
))
# Rows without a tx hash must not accidentally be merged together.
key = tx_hash or f"__nohash_{index}"
group = groups.setdefault(key, {
"txHash": tx_hash,
"timestamp": ts,
"rows": []
})
group["rows"].append(tx)
if ts is not None:
group["timestamp"] = max(group.get("timestamp") or 0, ts)
candidates = []
for group in groups.values():
rows = group["rows"]
types = {
str(_first_recursive(row, ("type", "actionType", "transactionType")) or "").upper()
for row in rows
}
# Confirmed from Krystal transaction history:
# COLLECT_FEE + DEPOSIT under the same txHash means fees were collected
# and immediately reinvested into the LP -> Auto Compound.
if "COLLECT_FEE" in types and "DEPOSIT" in types:
candidates.append({
"action": "Auto Compound",
"timestamp": group["timestamp"],
"isAutomation": True,
"txHash": group["txHash"]
})
continue
# Other multi-row/single-row events retain conservative labels until
# we have an observed Krystal payload that proves the automation type.
row_candidates = []
for row in rows:
action, is_automation = _normalize_action(row)
if not action:
continue
row_candidates.append((action, is_automation))
if not row_candidates:
continue
# Prefer an explicitly automated label if Krystal supplies one.
chosen = next(
(item for item in row_candidates if item[1]),
row_candidates[0]
)
candidates.append({
"action": chosen[0],
"timestamp": group["timestamp"],
"isAutomation": chosen[1],
"txHash": group["txHash"]
})
if not candidates:
return {
"action": None,
"timestamp": None,
"isAutomation": None,
"txHash": None
}
# Latest transaction wins. We no longer prefer an older "explicit auto"
# event over a newer LP-management event.
candidates.sort(
key=lambda x: x.get("timestamp") or 0,
reverse=True
)
return candidates[0]
def get_last_actions_for_positions(data, api_key):
"""Return cached last-action metadata keyed by Krystal position id."""
now = time.time()
to_refresh = []
result = {}
for p in data:
position_id = p.get("id")
if not position_id:
continue
source_update = p.get("lastUpdateBlockTime")
cached = _last_action_cache.get(position_id)
needs_refresh = (
cached is None
or cached.get("sourceUpdate") != source_update
or now - cached.get("checkedAt", 0) > AUTOMATION_CACHE_TTL
)
if needs_refresh:
to_refresh.append(p)
else:
result[position_id] = cached.get("data", {})
# Parallel calls keep the first dashboard load reasonable. Subsequent loads
# normally hit cache and make no transaction-history requests.
if to_refresh:
with ThreadPoolExecutor(max_workers=6) as executor:
futures = {
executor.submit(fetch_last_lp_action, p, api_key): p
for p in to_refresh
}
for future in as_completed(futures):
p = futures[future]
position_id = p.get("id")
try:
action_data = future.result()
except (requests.RequestException, ValueError, TypeError):
# Do not break the whole dashboard if history lookup fails.
action_data = {
"action": None,
"timestamp": None,
"isAutomation": None,
"txHash": None
}
_last_action_cache[position_id] = {
"sourceUpdate": p.get("lastUpdateBlockTime"),
"checkedAt": now,
"data": action_data
}
result[position_id] = action_data
return result
def get_boundary_data(p):
pool_price = p.get("pool", {}).get("poolPrice")
min_price = p.get("minPrice")
max_price = p.get("maxPrice")
distance_lower_pct = None
distance_upper_pct = None
nearest_boundary_pct = None
nearest_boundary_side = None
if (
pool_price is not None
and min_price is not None
and max_price is not None
and pool_price > 0
):
distance_lower_pct = (
(pool_price - min_price) / pool_price
) * 100
distance_upper_pct = (
(max_price - pool_price) / pool_price
) * 100
lower_abs = abs(distance_lower_pct)
upper_abs = abs(distance_upper_pct)
if lower_abs <= upper_abs:
nearest_boundary_pct = lower_abs
nearest_boundary_side = "LOWER"
else:
nearest_boundary_pct = upper_abs
nearest_boundary_side = "UPPER"
return {
"poolPrice": pool_price,
"minPrice": min_price,
"maxPrice": max_price,
"distanceLowerPct": distance_lower_pct,
"distanceUpperPct": distance_upper_pct,
"nearestBoundaryPct": nearest_boundary_pct,
"nearestBoundarySide": nearest_boundary_side
}
def _as_float(value, default=None):
if value in (None, ""):
return default
try:
return float(value)
except (TypeError, ValueError):
return default
def _vfat_pick(obj, *keys, default=None):
if not isinstance(obj, dict):
return default
for key in keys:
if key in obj and obj[key] not in (None, ""):
return obj[key]
for container in ("position", "pool", "farm", "metrics", "performance", "data"):
nested = obj.get(container)
if isinstance(nested, dict):
for key in keys:
if key in nested and nested[key] not in (None, ""):
return nested[key]
return default
def _vfat_tokens(row):
value = _vfat_pick(
row, "underlying", "tokens", "assets",
"underlying_assets", "underlyingAssets"
) or []
if isinstance(value, dict):
value = list(value.values())
return value if isinstance(value, list) else []
def _vfat_pair(row):
symbols = []
for token in _vfat_tokens(row):
if not isinstance(token, dict):
continue
nested = token.get("token") or {}
symbol = token.get("symbol") or nested.get("symbol")
if symbol and symbol not in symbols:
symbols.append(symbol)
if len(symbols) >= 2:
return "/".join(symbols[:2])
direct = _vfat_pick(
row, "pair", "pair_name", "pairName", "pool_name", "poolName", "name"
)
if isinstance(direct, str) and "/" in direct:
return direct.replace(" ", "")
token0 = _vfat_pick(row, "token0_symbol", "token0Symbol")
token1 = _vfat_pick(row, "token1_symbol", "token1Symbol")
if token0 and token1:
return f"{token0}/{token1}"
return str(direct or "VFAT POSITION")
def _vfat_status(row, price, low, high):
raw = str(_vfat_pick(
row, "status", "range_status", "rangeStatus", default=""
) or "").upper()
if raw in ("IN_RANGE", "OUT_OF_RANGE"):
return raw
in_range = _vfat_pick(row, "in_range", "inRange")
if isinstance(in_range, bool):
return "IN_RANGE" if in_range else "OUT_OF_RANGE"
if None not in (price, low, high):
return "IN_RANGE" if low <= price <= high else "OUT_OF_RANGE"
return raw or "UNKNOWN"
def _tick_price(tick, token0_decimals, token1_decimals):
"""
Convert a Uniswap-v3-style tick to token1/token0 human price.
For WETH/cbBTC this yields cbBTC per WETH, so invert it for the
dashboard's familiar USD-like WETH price in cbBTC terms.
"""
try:
return (1.0001 ** int(tick)) * (10 ** (int(token0_decimals) - int(token1_decimals)))
except (TypeError, ValueError, OverflowError):
return None
def _vfat_range_prices(row):
nft = row.get("nft") or {}
underlying = _vfat_tokens(row)
if len(underlying) < 2:
return None, None, None
d0 = underlying[0].get("decimals", 18)
d1 = underlying[1].get("decimals", 18)
tick = row.get("tick")
tick_low = nft.get("tick_low")
tick_up = nft.get("tick_up")
current_raw = _tick_price(tick, d0, d1)
low_raw = _tick_price(tick_low, d0, d1)
high_raw = _tick_price(tick_up, d0, d1)
# Display token0 per token1 if raw price is token1/token0. This keeps
# WETH/cbBTC around ~0.032 WETH/cbBTC? For the existing dashboard we
# prefer the same orientation as the token pair: token1 per token0.
return current_raw, low_raw, high_raw
def _robinhood_rpc_call(to, data):
response = requests.post(
ROBINHOOD_RPC,
json={
"jsonrpc": "2.0",
"method": "eth_call",
"params": [{"to": to, "data": data}, "latest"],
"id": 1
},
timeout=15
)
response.raise_for_status()
payload = response.json()
if payload.get("error"):
raise ValueError(f"Robinhood RPC error: {payload['error']}")
result = payload.get("result")
if not isinstance(result, str) or result in ("", "0x"):
raise ValueError("Robinhood RPC returned an empty result")
return result
def _robinhood_abi_words(result):
raw = result[2:] if result.startswith("0x") else result
if not raw or len(raw) % 64 != 0:
raise ValueError("Unexpected Robinhood ABI response length")
return [raw[i:i + 64] for i in range(0, len(raw), 64)]
def _robinhood_signed_word(word):
value = int(word, 16)
if value >= (1 << 255):
value -= (1 << 256)
return value
def _robinhood_address_word(word):
return "0x" + word[-40:]
def fetch_robinhood_position(reference_positions=None):
"""
Read the vfat-managed Robinhood WETH/cbBTC Uniswap V3 NFT directly
from Robinhood Chain. USD reference prices are reused from the working
Base vfat WETH/cbBTC position; the LP state itself is fully on-chain.
"""
# positions(uint256)
calldata = "0x99fbab88" + format(ROBINHOOD_NFT_ID, "064x")
position_result = _robinhood_rpc_call(
ROBINHOOD_POSITION_MANAGER,
calldata
)
words = _robinhood_abi_words(position_result)
if len(words) < 12:
raise ValueError("Unexpected Robinhood positions() response")
token0 = _robinhood_address_word(words[2])
token1 = _robinhood_address_word(words[3])
fee = int(words[4], 16)
tick_lower = _robinhood_signed_word(words[5])
tick_upper = _robinhood_signed_word(words[6])
liquidity = int(words[7], 16)
tokens_owed0 = int(words[10], 16)
tokens_owed1 = int(words[11], 16)
# slot0()
slot0_result = _robinhood_rpc_call(ROBINHOOD_POOL, "0x3850c7bd")
slot_words = _robinhood_abi_words(slot0_result)
if len(slot_words) < 2:
raise ValueError("Unexpected Robinhood slot0() response")
sqrt_price_x96 = int(slot_words[0], 16)
current_tick = _robinhood_signed_word(slot_words[1])
# Uniswap V3 liquidity math.
q96 = 2 ** 96
sqrt_p = sqrt_price_x96 / q96
sqrt_a = 1.0001 ** (tick_lower / 2)
sqrt_b = 1.0001 ** (tick_upper / 2)
if sqrt_p <= sqrt_a:
amount0_raw = liquidity * (sqrt_b - sqrt_a) / (sqrt_a * sqrt_b)
amount1_raw = 0.0
elif sqrt_p < sqrt_b:
amount0_raw = liquidity * (sqrt_b - sqrt_p) / (sqrt_p * sqrt_b)
amount1_raw = liquidity * (sqrt_p - sqrt_a)
else:
amount0_raw = 0.0
amount1_raw = liquidity * (sqrt_b - sqrt_a)
# This known Robinhood pair is WETH(18) / cbBTC(8).
amount0 = amount0_raw / 1e18 + tokens_owed0 / 1e18
amount1 = amount1_raw / 1e8 + tokens_owed1 / 1e8
weth_price = None
cbbtc_price = None
for position in reference_positions or []:
for token in position.get("underlying") or []:
symbol = str(token.get("symbol") or "").upper()
price = _as_float(token.get("price"))
if price is None or price <= 0:
continue
if symbol in ("WETH", "ETH"):
weth_price = price
elif symbol in ("CBBTC", "BTC"):
cbbtc_price = price
current_ratio = _tick_price(current_tick, 18, 8)
lower_ratio = _tick_price(tick_lower, 18, 8)
upper_ratio = _tick_price(tick_upper, 18, 8)
pool_price = (
cbbtc_price * current_ratio
if cbbtc_price is not None and current_ratio is not None
else weth_price
)
min_price = (
cbbtc_price * lower_ratio
if cbbtc_price is not None and lower_ratio is not None
else None
)
max_price = (
cbbtc_price * upper_ratio
if cbbtc_price is not None and upper_ratio is not None
else None
)
if min_price is not None and max_price is not None and min_price > max_price:
min_price, max_price = max_price, min_price
value = None
if weth_price is not None and cbbtc_price is not None:
value = amount0 * weth_price + amount1 * cbbtc_price
status = (
"IN_RANGE"
if tick_lower <= current_tick < tick_upper
else "OUT_OF_RANGE"
)
return {
"id": f"vfat:{ROBINHOOD_CHAIN_ID}:{ROBINHOOD_SICKLE}:{ROBINHOOD_NFT_ID}",
"source": "VFAT",
"chain": "Robinhood",
"chainId": ROBINHOOD_CHAIN_ID,
"pair": "WETH/cbBTC",
"protocol": "Uniswap V3",
"value": value,
"status": status,
"poolPrice": pool_price,
"minPrice": min_price,
"maxPrice": max_price,
"earning24h": 0.0,
"pendingFees": 0.0,
"claimedFees": 0.0,
"totalFees": 0.0,
"feeRoi": None,
"pnl": None,
"roi": None,
"totalDepositValue": value,
"totalWithdrawValue": 0.0,
"netInvested": value,
"openedTime": None,
"impermanentLoss": None,
"lastUpdateBlockTime": None,
"lastAction": "Deposit",
"lastActionTime": None,
"lastActionIsAutomation": False,
"lastActionTxHash": None,
"totalApr": None,
"feeApr": None,
"farmApr": None,
"sickleAddress": ROBINHOOD_SICKLE,
"nftId": ROBINHOOD_NFT_ID,
"poolAddress": ROBINHOOD_POOL,
"managerAddress": ROBINHOOD_POSITION_MANAGER,
"rewardsStatus": None,
"rewardsStatusReason": None,
"ageDays": None,
"underlying": [
{
"symbol": "WETH",
"balance": amount0,
"price": weth_price,
"valueUsd": amount0 * weth_price if weth_price is not None else None,
"address": token0
},
{
"symbol": "cbBTC",
"balance": amount1,
"price": cbbtc_price,
"valueUsd": amount1 * cbbtc_price if cbbtc_price is not None else None,
"address": token1
}
],
"distanceLowerPct": (
((pool_price - min_price) / pool_price * 100)
if pool_price and min_price is not None else None
),
"distanceUpperPct": (
((max_price - pool_price) / pool_price * 100)
if pool_price and max_price is not None else None
),
"currentTick": current_tick,
"tickLower": tick_lower,
"tickUpper": tick_upper,
"liquidity": liquidity,
"feeTier": fee
}
def _normalize_vfat_position(row):
if not isinstance(row, dict):
return None
chain_id = row.get("chain_id") or row.get("chainId")
chain_names = {
"8453": "Base",
"4663": "Robinhood",
}
chain = chain_names.get(str(chain_id), str(chain_id or "Unknown"))
underlying = _vfat_tokens(row)
value = _as_float(row.get("current_value_usd"))
if value is None:
value = _as_float(row.get("price"), 0.0)
# net_cashflow_usd is negative for capital deposited into the position.
net_cashflow = _as_float(row.get("net_cashflow_usd"))
deposited = abs(net_cashflow) if net_cashflow is not None and net_cashflow < 0 else value
pnl = _as_float(row.get("total_pnl_usd"))
roi = _as_float(row.get("roi"))
# vfat's instantaneous APR is meaningless for a position only minutes old
# (the sample returned about -7,794%). Suppress APR until >= 1 day old.
age_days = _as_float(row.get("age_in_days"))
apr = _as_float(row.get("apr")) if age_days is not None and age_days >= 1 else None
current_tick_price, low_tick_price, high_tick_price = _vfat_range_prices(row)
# Use underlying token USD prices to express the CL range in token0 USD
# terms when possible. For WETH/cbBTC this makes the range comparable to
# the WETH price displayed elsewhere on the dashboard.
pool_price = None
min_price = None
max_price = None
if len(underlying) >= 2:
p0 = _as_float(underlying[0].get("price"))
p1 = _as_float(underlying[1].get("price"))
if p0 is not None:
pool_price = p0
# tick price is token1/token0. Convert the tick boundaries to token0
# USD using token1 USD price: token0 USD = (token1 USD) * token1/token0.
if p1 is not None:
if low_tick_price is not None:
min_price = p1 * low_tick_price
if high_tick_price is not None:
max_price = p1 * high_tick_price
# Defensive ordering in case a provider/token orientation is reversed.
if min_price is not None and max_price is not None and min_price > max_price:
min_price, max_price = max_price, min_price
nft = row.get("nft") or {}
opened = _normalize_timestamp(
row.get("oldest_action_timestamp")
or row.get("position_block_timestamp")
)
updated = _normalize_timestamp(
row.get("source_updated_at") or row.get("computed_at")
)
estimated_fees = _as_float(row.get("estimated_swap_fees_usd"), 0.0)
token_id = row.get("token_id") or row.get("original_token_id") or nft.get("id")
sickle = row.get("sickle_address")
status = _vfat_status(row, pool_price, min_price, max_price)
# Known Uniswap V4 pools.
# V4 fee tier is identified by PoolId rather than the NFT token ID.
pool_id = str(
nft.get("pool_id")
or row.get("pool_id")
or ""
).lower()
known_v4_fee_tiers = {
# ETH/LINK Uniswap V4, tickSpacing 60, fee 0.30%
"0xb2b5618903d74bbac9e9049a035c3827afc4487cde3b994a1568b050f4c8e2e4": 3000,
}
fee_tier = known_v4_fee_tiers.get(pool_id)
return {
"id": f"vfat:{chain_id}:{sickle or ''}:{token_id or ''}",
"source": "VFAT",
"chain": chain,
"chainId": chain_id,
"pair": _vfat_pair(row),
"tickSpacing": _as_float(row.get("tick_spacing")),
"feeTier": fee_tier,
"protocol": (
_vfat_pick(row, "protocol", "protocol_name", "protocolName")
or ("Uniswap V3" if str(chain_id) == "4663" else "Aerodrome")
),
"value": value,
"status": status,
"poolPrice": pool_price,
"minPrice": min_price,
"maxPrice": max_price,
"earning24h": 0.0,
"pendingFees": estimated_fees,
"claimedFees": 0.0,
"totalFees": estimated_fees,
"feeRoi": (estimated_fees / deposited * 100) if deposited else None,
"pnl": pnl,
"roi": roi,
"totalDepositValue": deposited,
"totalWithdrawValue": 0.0,
"netInvested": deposited,
"openedTime": opened,
"impermanentLoss": None,
"lastUpdateBlockTime": updated,
"lastAction": "Deposit" if opened else None,
"lastActionTime": opened,
"lastActionIsAutomation": False if opened else None,
"lastActionTxHash": None,
"totalApr": apr,
"feeApr": None,
"farmApr": None,
"sickleAddress": sickle,
"nftId": token_id,
"poolAddress": nft.get("pool_address"),
"managerAddress": nft.get("manager_address"),
"rewardsStatus": row.get("rewards_status"),
"rewardsStatusReason": row.get("rewards_status_reason"),
"ageDays": age_days,
"underlying": [
{
"symbol": t.get("symbol"),
"balance": _as_float(t.get("balance")),
"price": _as_float(t.get("price")),
"valueUsd": (
(_as_float(t.get("balance")) or 0.0) *
(_as_float(t.get("price")) or 0.0)
),
"address": t.get("address")
}
for t in underlying
if isinstance(t, dict)
],
"distanceLowerPct": (
((pool_price - min_price) / pool_price * 100)
if pool_price and min_price is not None else None
),
"distanceUpperPct": (
((max_price - pool_price) / pool_price * 100)
if pool_price and max_price is not None else None
),
}
def _rpc_post_with_retry(rpc_url, payload, timeout=20, attempts=4):
"""
JSON-RPC POST with conservative exponential backoff.
Public Base RPC can return HTTP 429 when the fee engine makes several
sequential eth_call requests. Retry only transient failures.
"""
last_error = None
for attempt in range(attempts):
try:
response = requests.post(rpc_url, json=payload, timeout=timeout)
if response.status_code == 429 or 500 <= response.status_code < 600:
last_error = requests.HTTPError(
f"{response.status_code} transient RPC response for {rpc_url}",
response=response,
)
else:
response.raise_for_status()
data = response.json()
if data.get("error"):
raise ValueError(f"RPC error: {data['error']}")
return data
except (requests.Timeout, requests.ConnectionError) as exc:
last_error = exc
if attempt < attempts - 1:
# 1s, 2s, 4s. This keeps dashboard failures non-fatal while
# giving rate-limited public RPCs time to recover.
time.sleep(2 ** attempt)
if last_error is not None:
raise last_error
raise requests.RequestException(f"RPC request failed for {rpc_url}")
def _rpc_eth_call(rpc_url, to, data, block="latest", timeout=20):
payload = _rpc_post_with_retry(
rpc_url,
{
"jsonrpc": "2.0",
"method": "eth_call",
"params": [{"to": to, "data": data}, block],
"id": 1,
},
timeout=timeout,
)
result = payload.get("result")
if not isinstance(result, str) or result in ("", "0x"):
raise ValueError("RPC returned an empty eth_call result")
return result
def _rpc_block_number(rpc_url, timeout=20):
payload = _rpc_post_with_retry(
rpc_url,
{"jsonrpc": "2.0", "method": "eth_blockNumber", "params": [], "id": 1},
timeout=timeout,
)
result = payload.get("result")
if not isinstance(result, str) or not result.startswith("0x"):
raise ValueError("RPC returned an invalid block number")
return result
def _encode_int256(value):
return format(value % (1 << 256), "064x")
def _uint256_sub(a, b):
return (a - b) % (1 << 256)
def _fee_growth_inside(global_growth, lower_outside, upper_outside,
current_tick, tick_lower, tick_upper):
# Uniswap V3 / Slipstream Tick.getFeeGrowthInside semantics, with uint256
# modular arithmetic matching Solidity overflow behavior.
below = lower_outside if current_tick >= tick_lower else _uint256_sub(global_growth, lower_outside)
above = upper_outside if current_tick < tick_upper else _uint256_sub(global_growth, upper_outside)
return _uint256_sub(_uint256_sub(global_growth, below), above)
def _v3_pending_fees_raw(rpc_url, pool, manager, nft_id):
"""
Calculate live uncollected token0/token1 fees for a Uniswap-V3-compatible
position. Aerodrome Slipstream uses the same feeGrowth accounting; its
positions() word 4 is tickSpacing rather than Uniswap's fee tier, which
does not affect this calculation.
"""
block = _rpc_block_number(rpc_url)
position_words = _robinhood_abi_words(
_rpc_eth_call(
rpc_url,
manager,
"0x99fbab88" + format(int(nft_id), "064x"),
block,
)
)
if len(position_words) < 12:
raise ValueError("Unexpected positions() response")
tick_lower = _robinhood_signed_word(position_words[5])
tick_upper = _robinhood_signed_word(position_words[6])
liquidity = int(position_words[7], 16)
inside0_last = int(position_words[8], 16)
inside1_last = int(position_words[9], 16)
owed0 = int(position_words[10], 16)
owed1 = int(position_words[11], 16)
slot_words = _robinhood_abi_words(
_rpc_eth_call(rpc_url, pool, "0x3850c7bd", block)
)
current_tick = _robinhood_signed_word(slot_words[1])
global0 = int(_rpc_eth_call(rpc_url, pool, "0xf3058399", block), 16)
global1 = int(_rpc_eth_call(rpc_url, pool, "0x46141319", block), 16)
lower_words = _robinhood_abi_words(
_rpc_eth_call(rpc_url, pool, "0xf30dba93" + _encode_int256(tick_lower), block)
)
upper_words = _robinhood_abi_words(
_rpc_eth_call(rpc_url, pool, "0xf30dba93" + _encode_int256(tick_upper), block)
)
if len(lower_words) < 4 or len(upper_words) < 4:
raise ValueError("Unexpected ticks() response")
# Uniswap V3 ticks() has:
# [liquidityGross, liquidityNet, feeGrowthOutside0, feeGrowthOutside1, ...]
#
# Aerodrome Slipstream adds stakedLiquidityNet before the fee-growth
# fields, so its feeGrowthOutside0/1 are words [3] and [4].
is_slipstream = str(rpc_url).rstrip("/") == str(BASE_RPC).rstrip("/")
if is_slipstream:
if len(lower_words) < 5 or len(upper_words) < 5:
raise ValueError("Unexpected Aerodrome Slipstream ticks() response")
lower0 = int(lower_words[3], 16)
lower1 = int(lower_words[4], 16)
upper0 = int(upper_words[3], 16)
upper1 = int(upper_words[4], 16)
else:
lower0 = int(lower_words[2], 16)
lower1 = int(lower_words[3], 16)
upper0 = int(upper_words[2], 16)
upper1 = int(upper_words[3], 16)
inside0 = _fee_growth_inside(global0, lower0, upper0, current_tick, tick_lower, tick_upper)
inside1 = _fee_growth_inside(global1, lower1, upper1, current_tick, tick_lower, tick_upper)
delta0 = _uint256_sub(inside0, inside0_last)
delta1 = _uint256_sub(inside1, inside1_last)
q128 = 1 << 128
pending0 = owed0 + (liquidity * delta0 // q128)
pending1 = owed1 + (liquidity * delta1 // q128)
return {
"block": int(block, 16),
"currentTick": current_tick,
"tickLower": tick_lower,
"tickUpper": tick_upper,
"liquidity": liquidity,
"token0Raw": pending0,
"token1Raw": pending1,
}
def _apply_vfat_v3_fee_tier(position):
"""
Read fee tier directly from positions(uint256) for standard Uniswap-V3
compatible position managers.
Aerodrome Slipstream is excluded because positions() word 4 is
tickSpacing, not the pool fee.
"""
chain_id = str(position.get("chainId") or "")
nft_id = position.get("nftId")
manager = str(position.get("managerAddress") or "").lower()
if not nft_id or not manager:
return position
if chain_id == "1":
rpc_url = ETHEREUM_RPC
elif chain_id == "999":
rpc_url = HYPEREVM_RPC
elif chain_id == "4663":
rpc_url = ROBINHOOD_RPC
elif chain_id == "8453":
rpc_url = BASE_RPC
else:
return position
# Aerodrome Slipstream NFT Position Manager.
# positions() word 4 is tickSpacing, NOT fee tier.
slipstream_managers = {
"0xe1f8cd9ac4e4a65f54f38a5cdafca44f6dd68b53",
}
if manager in slipstream_managers:
try:
# Aerodrome Slipstream CLPool.fee().
# Fee is returned in pips (1e-6), compatible with feeTier display.
pool = position.get("poolAddress")
if pool:
result = _rpc_eth_call(
BASE_RPC,
pool,
"0xddca3f43",
)
if result and result != "0x":
position["feeTier"] = int(result, 16)
except Exception:
# Fee-tier enrichment must never break position fetching.
pass
return position
try:
result = _rpc_eth_call(
rpc_url,
manager,
"0x99fbab88" + format(int(nft_id), "064x"),
)
words = _robinhood_abi_words(result)
if len(words) >= 5:
position["feeTier"] = int(words[4], 16)
except Exception:
# Fee-tier enrichment must never break position fetching.
pass
return position
def _apply_vfat_onchain_pending_fees(position):
"""
Enrich the two known vfat WETH/cbBTC positions with live on-chain pending
fees. Failure is non-fatal: provider values remain intact.
"""
chain_id = str(position.get("chainId") or "")
nft_id = position.get("nftId")
pool = position.get("poolAddress")
manager = position.get("managerAddress")
if not nft_id or not pool or not manager:
return position
if chain_id == "4663":
rpc_url = ROBINHOOD_RPC
elif chain_id == "8453":
rpc_url = BASE_RPC
else:
return position
pair = str(position.get("pair") or "").replace(" ", "").upper()
if pair not in ("WETH/CBBTC", "CBBTC/WETH"):
return position
fee = _v3_pending_fees_raw(rpc_url, pool, manager, nft_id)
tokens = position.get("underlying") or []
token0 = tokens[0] if len(tokens) > 0 else {}
token1 = tokens[1] if len(tokens) > 1 else {}
def decimals_for(token):
symbol = str(token.get("symbol") or "").upper()
if symbol in ("CBBTC", "BTC"):
return 8
return 18
amount0 = fee["token0Raw"] / (10 ** decimals_for(token0))
amount1 = fee["token1Raw"] / (10 ** decimals_for(token1))
price0 = _as_float(token0.get("price"))
price1 = _as_float(token1.get("price"))
usd0 = amount0 * price0 if price0 is not None else None
usd1 = amount1 * price1 if price1 is not None else None
if usd0 is not None and usd1 is not None:
pending_usd = usd0 + usd1
position["pendingFees"] = pending_usd
# Do not pretend pending == lifetime total after future collects.
# Until historical collection tracking is added, totalFees represents
# currently observable uncollected fees only for these vfat positions.
position["totalFees"] = pending_usd
invested = _as_float(position.get("netInvested"))
position["feeRoi"] = pending_usd / invested * 100 if invested else None
position["pendingFeeToken0"] = amount0
position["pendingFeeToken1"] = amount1
position["pendingFeeToken0Usd"] = usd0
position["pendingFeeToken1Usd"] = usd1
position["pendingFeeBlock"] = fee["block"]
position["currentTick"] = fee["currentTick"]
position["tickLower"] = fee["tickLower"]
position["tickUpper"] = fee["tickUpper"]
return position
def _fee_db():
conn = sqlite3.connect(VFAT_FEE_DB, timeout=10)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("""
CREATE TABLE IF NOT EXISTS vfat_fee_state (
position_id TEXT PRIMARY KEY,
claimed0 REAL NOT NULL DEFAULT 0,
claimed1 REAL NOT NULL DEFAULT 0,
prev_pending0 REAL,
prev_pending1 REAL,
updated_at INTEGER NOT NULL
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS vfat_fee_snapshots (
position_id TEXT NOT NULL,
ts INTEGER NOT NULL,
cumulative0 REAL NOT NULL,
cumulative1 REAL NOT NULL,
PRIMARY KEY(position_id, ts)
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_vfat_fee_snapshots_position_ts
ON vfat_fee_snapshots(position_id, ts)
""")
return conn
def _track_vfat_fee_history(position):
"""
Persist fee counters in token units, not USD. Pending token amounts should
only rise between collect/compound events. A decrease is therefore treated
as collected/compounded fees and moved into the claimed counters.
USD values are calculated with the current provider token prices. This
avoids mistaking ordinary token-price moves for fee collection.
"""
if "pendingFeeToken0" not in position or "pendingFeeToken1" not in position:
return position
pid = str(position.get("id") or "")
if not pid:
return position
pending0 = float(position.get("pendingFeeToken0") or 0.0)
pending1 = float(position.get("pendingFeeToken1") or 0.0)
now = int(time.time())
tokens = position.get("underlying") or []
token0 = tokens[0] if len(tokens) > 0 else {}
token1 = tokens[1] if len(tokens) > 1 else {}
price0 = _as_float(token0.get("price"))
price1 = _as_float(token1.get("price"))
with _fee_db() as conn:
row = conn.execute(
"""SELECT claimed0, claimed1, prev_pending0, prev_pending1
FROM vfat_fee_state WHERE position_id=?""",
(pid,),
).fetchone()
if row:
claimed0, claimed1, prev0, prev1 = row
claimed0 = float(claimed0 or 0.0)
claimed1 = float(claimed1 or 0.0)
# Tiny numerical/provider differences should not create fake
# collections. Require a meaningful relative/absolute drop.
def collected_delta(previous, current):
if previous is None:
return 0.0
previous = float(previous)
drop = previous - current
tolerance = max(1e-15, abs(previous) * 1e-6)
return drop if drop > tolerance else 0.0
claimed0 += collected_delta(prev0, pending0)
claimed1 += collected_delta(prev1, pending1)
else:
claimed0 = 0.0
claimed1 = 0.0
conn.execute(
"""INSERT INTO vfat_fee_state
(position_id, claimed0, claimed1, prev_pending0, prev_pending1, updated_at)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(position_id) DO UPDATE SET
claimed0=excluded.claimed0,
claimed1=excluded.claimed1,
prev_pending0=excluded.prev_pending0,
prev_pending1=excluded.prev_pending1,
updated_at=excluded.updated_at""",
(pid, claimed0, claimed1, pending0, pending1, now),
)
cumulative0 = claimed0 + pending0
cumulative1 = claimed1 + pending1
last_snapshot = conn.execute(
"SELECT MAX(ts) FROM vfat_fee_snapshots WHERE position_id=?",
(pid,),
).fetchone()[0]
if last_snapshot is None or now - int(last_snapshot) >= VFAT_FEE_SNAPSHOT_INTERVAL:
conn.execute(
"""INSERT OR REPLACE INTO vfat_fee_snapshots
(position_id, ts, cumulative0, cumulative1)
VALUES (?, ?, ?, ?)""",
(pid, now, cumulative0, cumulative1),
)
cutoff = now - VFAT_FEE_HISTORY_DAYS * 86400
conn.execute(
"DELETE FROM vfat_fee_snapshots WHERE position_id=? AND ts < ?",
(pid, cutoff),
)
# Require a real 24-hour baseline. Until then, earning24h/feeApr stay
# null rather than annualizing a position only minutes old.
target = now - 86400
baseline = conn.execute(
"""SELECT ts, cumulative0, cumulative1
FROM vfat_fee_snapshots
WHERE position_id=? AND ts <= ?
ORDER BY ts DESC LIMIT 1""",
(pid, target),
).fetchone()
def usd(a0, a1):
if price0 is None or price1 is None:
return None
return a0 * price0 + a1 * price1
pending_usd = usd(pending0, pending1)
claimed_usd = usd(claimed0, claimed1)
total_usd = usd(cumulative0, cumulative1)
if pending_usd is not None:
position["pendingFees"] = pending_usd
if claimed_usd is not None:
position["claimedFees"] = claimed_usd
if total_usd is not None:
position["totalFees"] = total_usd
invested = _as_float(position.get("netInvested"))
position["feeRoi"] = total_usd / invested * 100 if invested else None
position["claimedFeeToken0"] = claimed0
position["claimedFeeToken1"] = claimed1
position["cumulativeFeeToken0"] = cumulative0
position["cumulativeFeeToken1"] = cumulative1
if baseline and price0 is not None and price1 is not None:
_, base0, base1 = baseline
earned0 = max(0.0, cumulative0 - float(base0))
earned1 = max(0.0, cumulative1 - float(base1))
earning24h = earned0 * price0 + earned1 * price1
position["earning24h"] = earning24h
invested = _as_float(position.get("netInvested"))
position["feeApr"] = (
earning24h / invested * 365 * 100
if invested and invested > 0 else None
)
else:
position["earning24h"] = None
position["feeApr"] = None
return position
def fetch_vfat_positions():
now = time.time()
if _vfat_cache["positions"] and now - _vfat_cache["checkedAt"] < VFAT_CACHE_TTL:
return _vfat_cache["positions"]
response = requests.get(
f"{VFAT_INFO_URL}/open-positions-v2",
params={"admin_address": WALLET},
headers={
"Accept": "application/json",
"User-Agent": "Nahini-DeFi-Monitor/1.3"
},
timeout=45
)
response.raise_for_status()
payload = response.json()
if isinstance(payload, list):
rows = payload
elif isinstance(payload, dict):
rows = (
payload.get("positions") or payload.get("data")
or payload.get("items") or payload.get("result") or []
)
if isinstance(rows, dict):
rows = rows.get("positions") or rows.get("items") or []
else:
rows = []
normalized = []
for row in rows:
item = _normalize_vfat_position(row)
if item:
normalized.append(item)
# Prefer vfat Metrics when it already returns the Robinhood NFT because
# that payload contains richer valuation/performance data. Use the direct
# Robinhood RPC reader only as a fallback when NFT #1176825 is absent.
robinhood_present = any(
str(p.get("chainId")) == str(ROBINHOOD_CHAIN_ID)
and str(p.get("nftId")) == str(ROBINHOOD_NFT_ID)
for p in normalized
)
if not robinhood_present:
try:
robinhood = fetch_robinhood_position(normalized)
if robinhood:
normalized.append(robinhood)
except (requests.RequestException, ValueError, TypeError, IndexError, OverflowError):
# Never let a Robinhood RPC failure hide the working Base positions.
pass
# Live on-chain fee enrichment. Each position is isolated so an RPC
# timeout/rate-limit cannot hide the rest of the dashboard.
for item in normalized:
try:
_apply_vfat_v3_fee_tier(item)
_apply_vfat_onchain_pending_fees(item)
_track_vfat_fee_history(item)
except (requests.RequestException, ValueError, TypeError, IndexError, OverflowError, sqlite3.Error):
pass
_vfat_cache["checkedAt"] = now
_vfat_cache["positions"] = normalized
return normalized
_provider_lkg = {
"vfat": [],
"revert": [],
"krystal": [],
}
def load_vfat_positions_safe():
try:
positions = fetch_vfat_positions()
if positions:
_provider_lkg["vfat"] = positions
return positions
except (requests.RequestException, ValueError, TypeError):
return list(_provider_lkg["vfat"])
def _revert_chain_info(network):
mapping = {
"mainnet": ("Ethereum", 1),
"base": ("Base", 8453),
"robinhood": ("Robinhood", 4663),
}
return mapping.get(
str(network or "").lower(),
(str(network or "Unknown"), None),
)
def _revert_token(row, address):
tokens = row.get("tokens") or {}
if not address:
return {}
wanted = str(address).lower()
for key, token in tokens.items():
if str(key).lower() == wanted:
return token if isinstance(token, dict) else {}
return {}
def _normalize_revert_position(row):
if not isinstance(row, dict):
return None
network = row.get("network")
chain, chain_id = _revert_chain_info(network)
token0 = _revert_token(row, row.get("token0"))
token1 = _revert_token(row, row.get("token1"))
symbol0 = token0.get("symbol") or "?"
symbol1 = token1.get("symbol") or "?"
pair = f"{symbol0}/{symbol1}"
value = _as_float(row.get("underlying_value"), 0.0)
# Revert fees_value is lifetime total fees:
# collected + currently uncollected.
total_fees = _as_float(row.get("fees_value"), 0.0)
price0 = _as_float(token0.get("price"), 0.0)
price1 = _as_float(token1.get("price"), 0.0)
pending_fees = (
_as_float(row.get("uncollected_fees0"), 0.0) * price0
+ _as_float(row.get("uncollected_fees1"), 0.0) * price1
)
claimed_fees = (
_as_float(row.get("collected_fees0"), 0.0) * price0
+ _as_float(row.get("collected_fees1"), 0.0) * price1
)
deposits = _as_float(row.get("deposits_value"), 0.0)
withdrawals = _as_float(row.get("withdrawals_value"), 0.0)
net_invested = deposits - withdrawals
# Revert Lend debt is recorded in cash_flows. Use the latest
# total_debt snapshot instead of hard-coding a borrowed amount.
debt = 0.0
debt_found = False
for flow in row.get("cash_flows") or []:
if flow.get("type") != "lendor-borrow":
continue
total_debt = _as_float(flow.get("total_debt"))
if total_debt is not None:
debt = total_debt
debt_found = True
equity = value - debt if debt_found else value
leverage = (
value / equity
if debt_found and equity and equity > 0
else 1.0
)
performance = row.get("performance") or {}
hodl = performance.get("hodl") or {}
pnl = _as_float(hodl.get("pnl"))
roi = _as_float(hodl.get("roi"))
total_apr = _as_float(hodl.get("apr"))
fee_apr = _as_float(hodl.get("fee_apr"))
impermanent_loss = _as_float(hodl.get("il"))
pool_price = _as_float(row.get("pool_price"))
min_price = _as_float(row.get("price_lower"))
max_price = _as_float(row.get("price_upper"))
if (
min_price is not None
and max_price is not None
and min_price > max_price
):
min_price, max_price = max_price, min_price
in_range = row.get("in_range")
if in_range is True:
status = "IN_RANGE"
elif in_range is False:
status = "OUT_OF_RANGE"
else:
status = "UNKNOWN"
exchange = str(row.get("exchange") or "").lower()
protocols = {
"uniswapv3": "Uniswap V3",
"uniswapv4": "Uniswap V4",
"aerodromev3": "Aerodrome Slipstream",
}
protocol = protocols.get(
exchange,
row.get("exchange") or "Unknown",
)
nft_id = row.get("nft_id")
opened = _normalize_timestamp(row.get("first_mint_ts"))
age_days = _as_float(row.get("age"))
amount0 = _as_float(row.get("current_amount0"))
amount1 = _as_float(row.get("current_amount1"))
underlying = [
{
"symbol": symbol0,
"balance": amount0,
"price": _as_float(token0.get("price")),
"valueUsd": (
amount0 * _as_float(token0.get("price"), 0.0)
if amount0 is not None else None
),
"address": row.get("token0"),
},
{
"symbol": symbol1,
"balance": amount1,
"price": _as_float(token1.get("price")),
"valueUsd": (
amount1 * _as_float(token1.get("price"), 0.0)
if amount1 is not None else None
),
"address": row.get("token1"),
},
]
return {
"id": f"revert:{chain_id}:{nft_id}",
"source": "Revert",
"chain": chain,
"chainId": chain_id,
"pair": pair,
"tickSpacing": None,
"feeTier": _as_float(row.get("fee_tier")),
"protocol": protocol,
"value": value,
"status": status,
"poolPrice": pool_price,
"minPrice": min_price,
"maxPrice": max_price,
"earning24h": 0.0,
"pendingFees": pending_fees,
"claimedFees": claimed_fees,
"totalFees": total_fees,
"feeRoi": (
total_fees / net_invested * 100
if net_invested else None
),
"pnl": pnl,
"roi": roi,
"totalDepositValue": deposits,
"totalWithdrawValue": withdrawals,
"netInvested": net_invested,
"grossValue": value,
"debt": debt if debt_found else 0.0,
"equity": equity,
"leverage": leverage,
"isLeveraged": debt_found and debt > 0,
"openedTime": opened,
"impermanentLoss": impermanent_loss,
"lastUpdateBlockTime": None,
"lastAction": "Deposit" if opened else None,
"lastActionTime": opened,
"lastActionIsAutomation": False if opened else None,
"lastActionTxHash": None,
"totalApr": total_apr,
"feeApr": fee_apr,
"farmApr": None,
"sickleAddress": None,
"nftId": nft_id,
"poolAddress": row.get("pool"),
"managerAddress": row.get("owner"),
"rewardsStatus": None,
"rewardsStatusReason": None,
"ageDays": age_days,
"underlying": underlying,
"distanceLowerPct": (
((pool_price - min_price) / pool_price * 100)
if pool_price and min_price is not None else None
),
"distanceUpperPct": (
((max_price - pool_price) / pool_price * 100)
if pool_price and max_price is not None else None
),
}
def fetch_revert_positions():
response = requests.get(
REVERT_ACCOUNT_URL,
timeout=30,
)
response.raise_for_status()
payload = response.json()
rows = payload.get("data") or []
result = []
for row in rows:
position = _normalize_revert_position(row)
if position:
result.append(position)
return result
def load_revert_positions_safe():
try:
positions = fetch_revert_positions()
if positions:
_provider_lkg["revert"] = positions
return positions
except (
requests.RequestException,
ValueError,
TypeError,
KeyError,
):
return list(_provider_lkg["revert"])
def _krystal_positions_normalized_uncached():
data = load_positions()
api_key = get_api_key()
last_actions = get_last_actions_for_positions(data, api_key)
result = []
for p in data:
performance = p.get("performance") or {}
apr = performance.get("apr") or {}
net = (
(performance.get("totalDepositValue") or 0)
- (performance.get("totalWithdrawValue") or 0)
)
result.append({
"id": p.get("id"),
"source": "KRYSTAL",
"chain": p.get("chain", {}).get("name"),
"chainId": p.get("chain", {}).get("id"),
"pair": get_pair(p),
"protocol": p.get("pool", {}).get("protocol", {}).get("name"),
"value": p.get("currentPositionValue"),
"status": p.get("status"),
"poolPrice": p.get("pool", {}).get("poolPrice"),
"minPrice": p.get("minPrice"),
"maxPrice": p.get("maxPrice"),
"earning24h": p.get("earning24h"),
"pendingFees": get_pending_fees(p),
"claimedFees": get_claimed_fees(p),
"totalFees": get_total_fees(p),
"feeRoi": (get_total_fees(p) / net * 100) if net > 0 else None,
"pnl": performance.get("pnl"),
"roi": performance.get("returnOnInvestment"),
"totalDepositValue": performance.get("totalDepositValue") or 0,
"totalWithdrawValue": performance.get("totalWithdrawValue") or 0,
"netInvested": net,
"openedTime": p.get("openedTime"),
"impermanentLoss": performance.get("impermanentLoss"),
"lastUpdateBlockTime": p.get("lastUpdateBlockTime"),
"lastAction": (last_actions.get(p.get("id")) or {}).get("action"),
"lastActionTime": (last_actions.get(p.get("id")) or {}).get("timestamp"),
"lastActionIsAutomation": (last_actions.get(p.get("id")) or {}).get("isAutomation"),
"lastActionTxHash": (last_actions.get(p.get("id")) or {}).get("txHash"),
"totalApr": apr.get("totalApr"),
"feeApr": apr.get("feeApr"),
"farmApr": apr.get("farmApr"),
"ageDays": None,
"underlying": [],
"distanceLowerPct": None,
"distanceUpperPct": None,
"nftId": p.get("tokenId"),
"poolAddress": (p.get("pool") or {}).get("poolAddress"),
"managerAddress": p.get("tokenAddress"),
"rewardsStatus": None,
"rewardsStatusReason": None,
})
return result
def krystal_positions_normalized():
now = time.time()
cached = _krystal_positions_cache.get("positions") or []
checked_at = _krystal_positions_cache.get("checkedAt", 0)
if cached and now - checked_at < KRYSTAL_POSITIONS_CACHE_TTL:
return cached
positions = _krystal_positions_normalized_uncached()
_krystal_positions_cache["checkedAt"] = now
_krystal_positions_cache["positions"] = positions
return positions
def load_krystal_positions_safe():
"""
Krystal is an optional upstream source. Billing/quota errors (including
HTTP 402), timeouts, bad payloads, missing API keys, or transient empty
responses must never hide healthy positions from the dashboard.
"""
try:
positions = krystal_positions_normalized()
if positions:
_provider_lkg["krystal"] = positions
_save_krystal_lkg(positions)
return positions
# Successful but empty responses can be transient.
# Prefer RAM LKG, then persistent disk LKG after a restart.
cached = list(_provider_lkg["krystal"])
if cached:
return cached
cached = _load_krystal_lkg()
if cached:
_provider_lkg["krystal"] = cached
return cached
except (HTTPException, requests.RequestException, ValueError, TypeError, KeyError):
cached = list(_provider_lkg["krystal"])
if cached:
return cached
cached = _load_krystal_lkg()
if cached:
_provider_lkg["krystal"] = cached
return cached
def _position_history_db():
"""Generic 5-minute history for PnL and cumulative/lifetime fee counters."""
conn = sqlite3.connect(POSITION_HISTORY_DB, timeout=10)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("""
CREATE TABLE IF NOT EXISTS position_snapshots (
position_id TEXT NOT NULL,
ts INTEGER NOT NULL,
pnl REAL,
total_fees REAL,
value REAL,
net_invested REAL,
PRIMARY KEY(position_id, ts)
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_position_snapshots_position_ts
ON position_snapshots(position_id, ts)
""")
return conn
def _history_baseline(conn, pid, now, seconds):
"""Closest snapshot at or before the requested lookback."""
target = now - seconds
return conn.execute(
"""SELECT ts, pnl, total_fees, value, net_invested
FROM position_snapshots
WHERE position_id=? AND ts <= ?
ORDER BY ts DESC LIMIT 1""",
(pid, target),
).fetchone()
def _position_history_id(position):
"""
Stable economic-position identity.
NFT/token IDs may change after a rebalance. Historical performance must
follow the provider + chain + account + pair instead of the current NFT.
"""
source = str(position.get("source") or "").strip().lower()
chain_id = str(position.get("chainId") or "").strip()
pair = str(position.get("pair") or "").replace(" ", "").upper()
if source == "vfat":
sickle = str(position.get("sickleAddress") or "").strip().lower()
if chain_id and sickle and pair:
return f"history:vfat:{chain_id}:{sickle}:{pair}"
if source == "revert":
if chain_id and pair:
return f"history:revert:{chain_id}:{pair}"
if source == "krystal":
if chain_id and pair:
return f"history:krystal:{chain_id}:{pair}"
return str(position.get("id") or "")
def _track_vfat_continuous_performance(conn, p, pid, now):
"""
Build a continuous VFAT performance index.
VFAT may replace the NFT during rebalance. Provider PnL can rebase at that
boundary, so only changes observed inside the same NFT generation are
accumulated. The accumulated index belongs to the stable economic position.
"""
conn.execute("""
CREATE TABLE IF NOT EXISTS vfat_continuity_state (
position_id TEXT PRIMARY KEY,
nft_id TEXT,
last_raw_pnl REAL,
last_raw_fees REAL,
cumulative_pnl REAL NOT NULL DEFAULT 0,
cumulative_fees REAL NOT NULL DEFAULT 0,
updated_at INTEGER NOT NULL
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS vfat_continuity_snapshots (
position_id TEXT NOT NULL,
ts INTEGER NOT NULL,
cumulative_pnl REAL,
cumulative_fees REAL,
PRIMARY KEY(position_id, ts)
)
""")
nft_id = str(p.get("nftId") or "")
raw_pnl = _as_float(p.get("pnl"))
raw_fees = _as_float(p.get("totalFees"))
value = _as_float(p.get("value"))
# A disappearing/rebalancing NFT can temporarily be returned as
# value=0 / pnl=None. Never treat that as economic performance.
valid = (
nft_id
and raw_pnl is not None
and value is not None
and value > 0
)
state = conn.execute("""
SELECT nft_id, last_raw_pnl, last_raw_fees,
cumulative_pnl, cumulative_fees
FROM vfat_continuity_state
WHERE position_id=?
""", (pid,)).fetchone()
if state is None:
cumulative_pnl = 0.0
cumulative_fees = 0.0
if valid:
conn.execute("""
INSERT INTO vfat_continuity_state
(position_id, nft_id, last_raw_pnl, last_raw_fees,
cumulative_pnl, cumulative_fees, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
""", (
pid, nft_id, raw_pnl, raw_fees,
cumulative_pnl, cumulative_fees, now
))
else:
return None
else:
old_nft, last_pnl, last_fees, cumulative_pnl, cumulative_fees = state
cumulative_pnl = float(cumulative_pnl or 0.0)
cumulative_fees = float(cumulative_fees or 0.0)
if not valid:
return (cumulative_pnl, cumulative_fees)
if str(old_nft or "") == nft_id:
# Normal observation inside the same NFT generation.
if last_pnl is not None:
cumulative_pnl += raw_pnl - float(last_pnl)
# totalFees can occasionally reset upstream. Only accumulate
# non-negative changes from this raw counter.
if raw_fees is not None and last_fees is not None:
fee_delta = raw_fees - float(last_fees)
if fee_delta >= -1e-9:
cumulative_fees += max(0.0, fee_delta)
# If NFT changed, establish a fresh raw baseline. Do NOT subtract
# old-NFT PnL from new-NFT PnL.
conn.execute("""
UPDATE vfat_continuity_state
SET nft_id=?,
last_raw_pnl=?,
last_raw_fees=?,
cumulative_pnl=?,
cumulative_fees=?,
updated_at=?
WHERE position_id=?
""", (
nft_id, raw_pnl, raw_fees,
cumulative_pnl, cumulative_fees, now, pid
))
last_snap = conn.execute("""
SELECT MAX(ts)
FROM vfat_continuity_snapshots
WHERE position_id=?
""", (pid,)).fetchone()[0]
if last_snap is None or now - int(last_snap) >= POSITION_SNAPSHOT_INTERVAL:
conn.execute("""
INSERT OR REPLACE INTO vfat_continuity_snapshots
(position_id, ts, cumulative_pnl, cumulative_fees)
VALUES (?, ?, ?, ?)
""", (pid, now, cumulative_pnl, cumulative_fees))
cutoff = now - POSITION_HISTORY_DAYS * 86400
conn.execute("""
DELETE FROM vfat_continuity_snapshots
WHERE position_id=? AND ts < ?
""", (pid, cutoff))
return cumulative_pnl, cumulative_fees
def _vfat_continuity_baseline(conn, pid, now, seconds):
target = now - seconds
# Pick the snapshot closest to the requested lookback, from either side.
# Never label a stale baseline as 24H/3D/7D.
row = conn.execute("""
SELECT ts, cumulative_pnl, cumulative_fees
FROM vfat_continuity_snapshots
WHERE position_id=?
ORDER BY ABS(ts - ?)
LIMIT 1
""", (pid, target)).fetchone()
if not row:
return None
# Normal snapshots are only minutes apart. Allow up to 30 minutes
# around the requested target; larger gaps mean the period is unavailable.
if abs(int(row[0]) - target) > 600:
return None
return row
def _track_position_history(position):
"""
Add actual 24H/3D/7D deltas from provider PnL and cumulative/lifetime fees.
No manual correction is needed for ordinary fee claims/compounds when the
provider's totalFees remains cumulative (Krystal/Revert) or when vfat's
token-unit cumulative fee tracker is available.
"""
p = dict(position)
pid = _position_history_id(p)
if not pid:
return p
now = int(time.time())
pnl = _as_float(p.get("pnl"))
total_fees = _as_float(p.get("totalFees"))
value = _as_float(p.get("value"))
net_invested = _as_float(p.get("netInvested"))
try:
with _position_history_db() as conn:
last = conn.execute(
"SELECT MAX(ts) FROM position_snapshots WHERE position_id=?",
(pid,),
).fetchone()[0]
if last is None or now - int(last) >= POSITION_SNAPSHOT_INTERVAL:
conn.execute(
"""INSERT OR REPLACE INTO position_snapshots
(position_id, ts, pnl, total_fees, value, net_invested)
VALUES (?, ?, ?, ?, ?, ?)""",
(pid, now, pnl, total_fees, value, net_invested),
)
cutoff = now - POSITION_HISTORY_DAYS * 86400
conn.execute(
"DELETE FROM position_snapshots WHERE position_id=? AND ts < ?",
(pid, cutoff),
)
is_vfat = str(p.get("source") or "").strip().lower() == "vfat"
if is_vfat:
current_cont = _track_vfat_continuous_performance(
conn, p, pid, now
)
else:
current_cont = None
for label, seconds in (("24h", 86400), ("3d", 3*86400), ("7d", 7*86400)):
pnl_key = {"24h":"pnl24h", "3d":"pnl3d", "7d":"pnl7d"}[label]
fee_key = {"24h":"fees24h", "3d":"fees3d", "7d":"fees7d"}[label]
p[pnl_key] = None
p[fee_key] = None
if is_vfat:
if not current_cont:
continue
base = _vfat_continuity_baseline(conn, pid, now, seconds)
if not base:
continue
_, base_pnl, base_fees = base
cur_pnl, cur_fees = current_cont
if cur_pnl is not None and base_pnl is not None:
p[pnl_key] = float(cur_pnl) - float(base_pnl)
if cur_fees is not None and base_fees is not None:
delta = float(cur_fees) - float(base_fees)
p[fee_key] = delta if delta >= -1e-9 else None
continue
base = _history_baseline(conn, pid, now, seconds)
if not base:
continue
_, base_pnl, base_fees, _, _ = base
if pnl is not None and base_pnl is not None:
p[pnl_key] = pnl - float(base_pnl)
if total_fees is not None and base_fees is not None:
delta = total_fees - float(base_fees)
p[fee_key] = delta if delta >= -1e-9 else None
except sqlite3.Error:
for key in ("pnl24h", "pnl3d", "pnl7d", "fees24h", "fees3d", "fees7d"):
p.setdefault(key, None)
return p
def combined_positions():
# Keep upstream providers failure-isolated.
krystal = load_krystal_positions_safe()
vfat = load_vfat_positions_safe()
revert = load_revert_positions_safe()
# One immediate retry protects against transient upstream failures.
# The safe loaders still provide last-known-good data during the
# lifetime of this process.
if not krystal:
krystal = load_krystal_positions_safe()
if not vfat:
vfat = load_vfat_positions_safe()
if not revert:
revert = load_revert_positions_safe()
# VFAT positions are independent. Revert is authoritative when the
# same chain + NFT is also returned by Krystal.
combined = vfat + revert + krystal
deduped = []
seen = set()
for p in combined:
chain_id = str(p.get("chainId") or "")
nft_id = p.get("nftId")
# Krystal normalized rows currently keep the NFT as the
# numeric suffix of "-".
if nft_id in (None, "") and p.get("source") == "KRYSTAL":
position_id = str(p.get("id") or "")
suffix = position_id.rsplit("-", 1)[-1]
if suffix.isdigit():
nft_id = suffix
if chain_id and nft_id not in (None, ""):
key = ("nft", chain_id, str(nft_id))
else:
key = (
"source",
str(p.get("source")),
str(p.get("id")),
)
if key in seen:
continue
seen.add(key)
deduped.append(p)
return [_track_position_history(p) for p in deduped]
def enrich_position_alert(p):
p = dict(p)
price = _as_float(p.get("poolPrice"))
low = _as_float(p.get("minPrice"))
high = _as_float(p.get("maxPrice"))
nearest = None
side = None
if price and low is not None and high is not None:
dl = abs((price - low) / price * 100)
du = abs((high - price) / price * 100)
nearest, side = (
(dl, "LOWER")
if dl <= du
else (du, "UPPER")
)
status = p.get("status")
age_days = _as_float(p.get("ageDays"))
pnl = _as_float(p.get("pnl"))
total_fees = _as_float(
p.get("totalFees"), 0.0
) or 0.0
alert = "OK"
if status == "OUT_OF_RANGE":
alert = "ACTION_REQUIRED"
elif nearest is not None and nearest <= 3:
alert = "WATCH"
elif (
age_days is not None
and age_days >= 3
and pnl is not None
and pnl < 0
and total_fees > 0
):
alert = "PERFORMANCE_WATCH"
p["alert"] = alert
p["nearestBoundaryPct"] = nearest
p["nearestBoundarySide"] = side
return p
@app.get("/")
def root():
return {
"status": "online",
"service": "Nahini DeFi Monitor"
}
@app.get("/health")
def health():
return {
"status": "ok",
"timestamp": datetime.now(timezone.utc).isoformat()
}
@app.get("/positions")
def positions():
result = [
enrich_position_alert(p)
for p in combined_positions()
]
return {
"wallet": WALLET,
"updatedAt": datetime.now(timezone.utc).isoformat(),
"count": len(result),
"positions": result
}
@app.get("/summary")
def summary():
data = combined_positions()
total_value = sum(_as_float(p.get("value"), 0.0) or 0.0 for p in data)
total_pnl = sum(_as_float(p.get("pnl"), 0.0) or 0.0 for p in data)
total_pending_fees = sum(_as_float(p.get("pendingFees"), 0.0) or 0.0 for p in data)
total_earning_24h = sum(_as_float(p.get("earning24h"), 0.0) or 0.0 for p in data)
pnl24_values = [_as_float(p.get("pnl24h")) for p in data]
pnl24_values = [x for x in pnl24_values if x is not None]
fee24_values = [_as_float(p.get("fees24h")) for p in data]
fee24_values = [x for x in fee24_values if x is not None]
out_of_range = [p for p in data if p.get("status") == "OUT_OF_RANGE"]
return {
"wallet": WALLET,
"updatedAt": datetime.now(timezone.utc).isoformat(),
"positionCount": len(data),
"totalValue": total_value,
"totalPnl": total_pnl,
"totalPendingFees": total_pending_fees,
"totalEarning24h": total_earning_24h,
"totalPnl24h": sum(pnl24_values) if pnl24_values else None,
"totalFees24h": sum(fee24_values) if fee24_values else None,
"outOfRangeCount": len(out_of_range),
"outOfRange": out_of_range,
"closestToBoundary": []
}
@app.get("/alerts")
def alerts():
data = combined_positions()
result = []
watch_count = 0
action_required_count = 0
for p in data:
price = _as_float(p.get("poolPrice"))
low = _as_float(p.get("minPrice"))
high = _as_float(p.get("maxPrice"))
nearest = None
side = None
if price and low is not None and high is not None:
dl = abs((price - low) / price * 100)
du = abs((high - price) / price * 100)
nearest, side = (dl, "LOWER") if dl <= du else (du, "UPPER")
status = p.get("status")
age_days = _as_float(p.get("ageDays"))
pnl = _as_float(p.get("pnl"))
total_fees = _as_float(p.get("totalFees"), 0.0) or 0.0
alert = "OK"
if status == "OUT_OF_RANGE":
alert = "ACTION_REQUIRED"
action_required_count += 1
elif nearest is not None and nearest <= 3:
alert = "WATCH"
watch_count += 1
elif (
age_days is not None
and age_days >= 3
and pnl is not None
and pnl < 0
and total_fees > 0
):
alert = "PERFORMANCE_WATCH"
watch_count += 1
result.append({
"source": p.get("source"),
"chain": p.get("chain"),
"pair": p.get("pair"),
"alert": alert,
"status": status,
"nearestBoundaryPct": nearest,
"nearestBoundarySide": side,
"value": p.get("value"),
"poolPrice": price,
"minPrice": low,
"maxPrice": high
})
return {
"updatedAt": datetime.now(timezone.utc).isoformat(),
"positionCount": len(data),
"watchCount": watch_count,
"actionRequiredCount": action_required_count,
"alerts": result
}
@app.get("/dashboard", response_class=HTMLResponse)
def dashboard():
return """
Nahini DeFi Monitor
