| name | crypto-defi-trading |
| description | Crypto and DeFi trading: DEX analysis (Uniswap, SushiSwap, Curve), on-chain analytics, MEV detection, impermanent loss, yield farming metrics, DeFi risk analysis, token metrics, liquidity pool analysis, whale tracking, exchange netflow. USE FOR: crypto, defi, dex, uniswap, sushiswap, curve, impermanent loss, yield farming, on-chain, whale, MEV, arbitrage, liquidity pool, token, exchange flow, gas, NFT. |
| related_skills | ["liquidity-analysis","ict-smart-money","technical-analysis","risk-and-portfolio"] |
| tags | ["trading","asset-class","crypto","defi","dex","mev","yield-farming","bitcoin"] |
| skill_level | advanced |
| kind | reference |
| category | trading/asset-classes |
| status | active |
Skill: Crypto Defi Trading | Domain: trading | Category: asset-class | Level: advanced
Tags: trading, asset-class, crypto, defi, dex, mev, yield-farming, bitcoin
DEX Analysis Engine
DEX Analysis Engine
Overview
Complete decentralized exchange analysis covering Uniswap V2/V3, SushiSwap, Curve, and
other AMM protocols. Analyzes pool states, liquidity distributions, price impact, and
optimal routing across DEXes.
Architecture
โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
โ DEX Analysis Engine โ
โโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโฌโโโโโโโโโโโโโโโค
โ Pool State โ Liquidity โ Price Impact โ Cross-DEX โ
โ Analyzer โ Distribution โ Calculator โ Router โ
โโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโดโโโโโโโโโโโโโโโ
import numpy as np
import pandas as pd
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from datetime import datetime, timezone
import math
@dataclass
class Token:
"""Represents an ERC-20 token."""
address: str
symbol: str
decimals: int = 18
name: str = ""
def format_amount(self, raw_amount: int) -> float:
"""Convert raw token amount to human-readable."""
return raw_amount / (10 ** self.decimals)
def to_raw(self, amount: float) -> int:
"""Convert human-readable amount to raw."""
return int(amount * (10 ** self.decimals))
@dataclass
class PoolState:
"""State of an AMM liquidity pool."""
pool_address: str
token_0: Token
token_1: Token
reserve_0: float
reserve_1: float
fee_tier: float
total_liquidity: float
price: float
volume_24h: float = 0.0
fee_revenue_24h: float = 0.0
tvl_usd: float = 0.0
tick_current: Optional[int] = None
sqrt_price_x96: Optional[int] = None
@property
def fee_apr(self) -> float:
"""Annualized fee APR based on 24h volume."""
if self.tvl_usd == 0:
return 0.0
daily_fee_rate = self.fee_revenue_24h / self.tvl_usd
return daily_fee_rate * 365 * 100
@property
def volume_to_tvl(self) -> float:
"""Volume/TVL ratio โ higher = more capital efficient."""
if self.tvl_usd == 0:
return 0.0
return self.volume_24h / self.tvl_usd
@dataclass
class LiquidityPosition:
"""A liquidity provider's position."""
pool_address: str
owner: str
liquidity: float
token_0_amount: float
token_1_amount: float
lower_tick: Optional[int] = None
upper_tick: Optional[int] = None
fees_earned_0: float = 0.0
fees_earned_1: float = 0.0
opened_at: Optional[datetime] = None
@property
def is_in_range(self) -> bool:
"""Check if a V3 position is currently in range (needs current tick)."""
if self.lower_tick is None or self.upper_tick is None:
return True
return True
class UniswapV2Analyzer:
"""
Uniswap V2 constant product AMM analyzer.
Core formula: x * y = k
Price: p = y / x
Output amount: dy = (y * dx * (1 - fee)) / (x + dx * (1 - fee))
"""
@staticmethod
def get_price(reserve_0: float, reserve_1: float) -> float:
"""Calculate spot price (token1 per token0)."""
if reserve_0 == 0:
return 0.0
return reserve_1 / reserve_0
@staticmethod
def get_output_amount(
amount_in: float,
reserve_in: float,
reserve_out: float,
fee: float = 0.003,
) -> float:
"""
Calculate output amount for a swap.
Args:
amount_in: Amount of input token
reserve_in: Reserve of input token
reserve_out: Reserve of output token
fee: Fee tier (e.g., 0.003 for 0.3%)
"""
if reserve_in == 0 or reserve_out == 0:
return 0.0
amount_in_with_fee = amount_in * (1 - fee)
numerator = amount_in_with_fee * reserve_out
denominator = reserve_in + amount_in_with_fee
return numerator / denominator
@staticmethod
def get_price_impact(
amount_in: float,
reserve_in: float,
reserve_out: float,
fee: float = 0.003,
) -> float:
"""
Calculate price impact of a trade as a percentage.
Returns:
Price impact as a decimal (e.g., 0.02 = 2% impact)
"""
if reserve_in == 0 or reserve_out == 0:
return 1.0
spot_price = reserve_out / reserve_in
output = UniswapV2Analyzer.get_output_amount(
amount_in, reserve_in, reserve_out, fee
)
if amount_in == 0:
return 0.0
exec_price = output / amount_in
impact = 1 - (exec_price / spot_price)
return abs(impact)
@staticmethod
def get_k(reserve_0: float, reserve_1: float) -> float:
"""Calculate the constant product k."""
return reserve_0 * reserve_1
@staticmethod
def optimal_liquidity(
amount_0: float,
reserve_0: float,
reserve_1: float,
) -> Tuple[float, float]:
"""
Calculate optimal token amounts for adding liquidity.
Given an amount of token0, returns the required amount of token1
to maintain the pool ratio.
"""
if reserve_0 == 0:
return amount_0, 0.0
amount_1 = amount_0 * reserve_1 / reserve_0
return amount_0, amount_1
@staticmethod
def lp_share(
liquidity_added: float,
total_liquidity: float,
) -> float:
"""Calculate LP share percentage."""
total = total_liquidity + liquidity_added
if total == 0:
return 0.0
return liquidity_added / total
class UniswapV3Analyzer:
"""
Uniswap V3 concentrated liquidity analyzer.
V3 uses ticks and concentrated positions. Liquidity is provided
within price ranges instead of across the full curve.
"""
TICK_BASE = 1.0001
MIN_TICK = -887272
MAX_TICK = 887272
Q96 = 2 ** 96
@staticmethod
def tick_to_price(tick: int) -> float:
"""Convert a tick to a price."""
return UniswapV3Analyzer.TICK_BASE ** tick
@staticmethod
def price_to_tick(price: float) -> int:
"""Convert a price to the nearest tick."""
if price <= 0:
return UniswapV3Analyzer.MIN_TICK
return int(math.log(price) / math.log(UniswapV3Analyzer.TICK_BASE))
@staticmethod
def sqrt_price_x96_to_price(sqrt_price_x96: int, decimals_0: int = 18, decimals_1: int = 18) -> float:
"""Convert sqrtPriceX96 to human-readable price."""
price = (sqrt_price_x96 / UniswapV3Analyzer.Q96) ** 2
return price * (10 ** (decimals_0 - decimals_1))
@staticmethod
def liquidity_for_amounts(
sqrt_price_current: float,
sqrt_price_lower: float,
sqrt_price_upper: float,
amount_0: float,
amount_1: float,
) -> float:
"""
Calculate liquidity for given token amounts and price range.
Based on the Uniswap V3 whitepaper formulas.
"""
if sqrt_price_current <= sqrt_price_lower:
if amount_0 == 0:
return 0.0
return amount_0 * sqrt_price_lower * sqrt_price_upper / (sqrt_price_upper - sqrt_price_lower)
elif sqrt_price_current >= sqrt_price_upper:
if amount_1 == 0:
return 0.0
return amount_1 / (sqrt_price_upper - sqrt_price_lower)
else:
liq_0 = amount_0 * sqrt_price_current * sqrt_price_upper / (sqrt_price_upper - sqrt_price_current)
liq_1 = amount_1 / (sqrt_price_current - sqrt_price_lower)
return min(liq_0, liq_1)
@staticmethod
def amounts_for_liquidity(
liquidity: float,
sqrt_price_current: float,
sqrt_price_lower: float,
sqrt_price_upper: float,
) -> Tuple[float, float]:
"""Calculate token amounts for a given liquidity and price range."""
if sqrt_price_current <= sqrt_price_lower:
amount_0 = liquidity * (sqrt_price_upper - sqrt_price_lower) / (sqrt_price_lower * sqrt_price_upper)
amount_1 = 0.0
elif sqrt_price_current >= sqrt_price_upper:
amount_0 = 0.0
amount_1 = liquidity * (sqrt_price_upper - sqrt_price_lower)
else:
amount_0 = liquidity * (sqrt_price_upper - sqrt_price_current) / (sqrt_price_current * sqrt_price_upper)
amount_1 = liquidity * (sqrt_price_current - sqrt_price_lower)
return amount_0, amount_1
@staticmethod
def fee_growth_in_range(
fee_growth_global_0: float,
fee_growth_global_1: float,
fee_growth_outside_lower_0: float,
fee_growth_outside_lower_1: float,
fee_growth_outside_upper_0: float,
fee_growth_outside_upper_1: float,
tick_current: int,
tick_lower: int,
tick_upper: int,
) -> Tuple[float, float]:
"""Calculate accumulated fees within a position's range."""
if tick_current >= tick_lower:
fee_below_0 = fee_growth_outside_lower_0
fee_below_1 = fee_growth_outside_lower_1
else:
fee_below_0 = fee_growth_global_0 - fee_growth_outside_lower_0
fee_below_1 = fee_growth_global_1 - fee_growth_outside_lower_1
if tick_current < tick_upper:
fee_above_0 = fee_growth_outside_upper_0
fee_above_1 = fee_growth_outside_upper_1
else:
fee_above_0 = fee_growth_global_0 - fee_growth_outside_upper_0
fee_above_1 = fee_growth_global_1 - fee_growth_outside_upper_1
fee_in_range_0 = fee_growth_global_0 - fee_below_0 - fee_above_0
fee_in_range_1 = fee_growth_global_1 - fee_below_1 - fee_above_1
return fee_in_range_0, fee_in_range_1
@staticmethod
def capital_efficiency(
tick_lower: int,
tick_upper: int,
) -> float:
"""
Calculate capital efficiency multiplier vs V2 full range.
Narrower ranges = higher efficiency but more IL risk.
"""
price_lower = UniswapV3Analyzer.tick_to_price(tick_lower)
price_upper = UniswapV3Analyzer.tick_to_price(tick_upper)
if price_lower <= 0 or price_upper <= price_lower:
return 1.0
sqrt_lower = math.sqrt(price_lower)
sqrt_upper = math.sqrt(price_upper)
return 1.0 / (1.0 - sqrt_lower / sqrt_upper)
Impermanent Loss Calculator
Impermanent Loss Calculator
class ImpermanentLossCalculator:
"""
Complete impermanent loss analysis for AMM liquidity provision.
Covers:
- Standard IL formula for V2 constant product AMMs
- V3 concentrated liquidity IL
- IL with fee compensation
- Break-even analysis
- Multi-asset IL
- IL hedging strategies
"""
@staticmethod
def v2_impermanent_loss(price_ratio: float) -> float:
"""
Calculate impermanent loss for Uniswap V2 (constant product).
Args:
price_ratio: Current price / initial price (e.g., 1.5 = 50% increase)
Returns:
IL as a negative decimal (e.g., -0.0566 = -5.66% loss vs HODL)
"""
if price_ratio <= 0:
return -1.0
sqrt_ratio = math.sqrt(price_ratio)
il = 2 * sqrt_ratio / (1 + price_ratio) - 1
return il
@staticmethod
def v2_il_percentage(price_ratio: float) -> float:
"""IL as a positive percentage (convenience)."""
return abs(ImpermanentLossCalculator.v2_impermanent_loss(price_ratio)) * 100
@staticmethod
def v3_impermanent_loss(
price_initial: float,
price_current: float,
price_lower: float,
price_upper: float,
) -> float:
"""
Calculate impermanent loss for Uniswap V3 concentrated position.
Concentrated liquidity amplifies both fees earned AND impermanent loss.
IL can be significantly worse than V2 for narrow ranges.
Args:
price_initial: Price when position was opened
price_current: Current price
price_lower: Lower bound of liquidity range
price_upper: Upper bound of liquidity range
"""
if price_current <= 0 or price_initial <= 0:
return -1.0
p0 = max(min(price_initial, price_upper), price_lower)
p1 = max(min(price_current, price_upper), price_lower)
sqrt_p0 = math.sqrt(p0)
sqrt_p1 = math.sqrt(p1)
sqrt_pa = math.sqrt(price_lower)
sqrt_pb = math.sqrt(price_upper)
if price_current <= price_lower:
value_current = sqrt_pb - sqrt_pa
elif price_current >= price_upper:
value_current = (sqrt_pb - sqrt_pa) * price_current / sqrt_pb
else:
value_current = (sqrt_p1 - sqrt_pa) * sqrt_p1 + (sqrt_pb - sqrt_p1)
if price_initial <= price_lower:
value_hodl = (sqrt_pb - sqrt_pa) * price_current / price_initial
elif price_initial >= price_upper:
value_hodl = (sqrt_pb - sqrt_pa)
else:
value_hodl_token0 = (sqrt_pb - sqrt_p0) * price_current / price_initial
value_hodl_token1 = (sqrt_p0 - sqrt_pa) * sqrt_p0
value_hodl = value_hodl_token0 + value_hodl_token1
if value_hodl == 0:
return 0.0
return (value_current - value_hodl) / value_hodl
@staticmethod
def il_with_fees(
price_ratio: float,
fee_tier: float,
volume_to_tvl_daily: float,
days: int,
) -> dict:
"""
Calculate net IL after fee compensation.
Args:
price_ratio: Current price / initial price
fee_tier: Pool fee tier (e.g., 0.003)
volume_to_tvl_daily: Daily volume/TVL ratio
days: Number of days position has been open
Returns:
Dict with il, fees_earned, net_pnl (all as percentages)
"""
il_pct = ImpermanentLossCalculator.v2_il_percentage(price_ratio)
daily_fee_yield = fee_tier * volume_to_tvl_daily * 100
total_fees = daily_fee_yield * days
net_pnl = total_fees - il_pct
return {
"impermanent_loss_pct": round(il_pct, 4),
"fees_earned_pct": round(total_fees, 4),
"net_pnl_pct": round(net_pnl, 4),
"days_to_breakeven": round(il_pct / daily_fee_yield, 1) if daily_fee_yield > 0 else float("inf"),
"daily_fee_yield_pct": round(daily_fee_yield, 4),
"annualized_fee_yield_pct": round(daily_fee_yield * 365, 2),
"compensated": net_pnl >= 0,
}
@staticmethod
def il_table(price_changes: List[float] = None) -> pd.DataFrame:
"""
Generate an IL reference table for common price changes.
Returns DataFrame with columns: price_change_pct, price_ratio, il_pct
"""
if price_changes is None:
price_changes = [-90, -80, -70, -60, -50, -40, -30, -25, -20, -15, -10, -5,
0, 5, 10, 15, 20, 25, 30, 40, 50, 60, 70, 80, 90, 100,
150, 200, 300, 400, 500]
rows = []
for pct in price_changes:
ratio = 1 + pct / 100
if ratio <= 0:
continue
il = ImpermanentLossCalculator.v2_il_percentage(ratio)
rows.append({
"price_change_pct": pct,
"price_ratio": round(ratio, 2),
"il_pct": round(il, 4),
})
return pd.DataFrame(rows)
@staticmethod
def breakeven_volume(
price_ratio: float,
fee_tier: float,
tvl: float,
days: int,
) -> float:
"""
Calculate the daily volume needed to offset IL with fees.
Returns required daily volume in USD.
"""
il_pct = ImpermanentLossCalculator.v2_il_percentage(price_ratio) / 100
il_usd = il_pct * tvl
if days == 0 or fee_tier == 0:
return float("inf")
required_daily_fees = il_usd / days
required_daily_volume = required_daily_fees / fee_tier
return required_daily_volume
On-Chain Analytics
On-Chain Analytics
class OnChainAnalytics:
"""
On-chain data analysis for trading intelligence.
Tracks:
- Exchange netflow (bullish/bearish indicator)
- Whale activity and accumulation
- Network health metrics
- Active address trends
- Token holder distribution
- Smart money flows
"""
@staticmethod
def exchange_netflow(
inflow_usd: float,
outflow_usd: float,
) -> dict:
"""
Analyze exchange netflow.
Positive netflow (more inflow) = bearish (coins moving to exchange to sell)
Negative netflow (more outflow) = bullish (coins leaving exchange = accumulation)
Args:
inflow_usd: Total USD value flowing INTO exchanges
outflow_usd: Total USD value flowing OUT of exchanges
"""
netflow = inflow_usd - outflow_usd
total_flow = inflow_usd + outflow_usd
if total_flow == 0:
bias = "neutral"
strength = 0.0
else:
ratio = netflow / total_flow
if ratio > 0.1:
bias = "strongly_bearish"
strength = min(abs(ratio) * 5, 1.0)
elif ratio > 0.03:
bias = "bearish"
strength = min(abs(ratio) * 5, 1.0)
elif ratio < -0.1:
bias = "strongly_bullish"
strength = min(abs(ratio) * 5, 1.0)
elif ratio < -0.03:
bias = "bullish"
strength = min(abs(ratio) * 5, 1.0)
else:
bias = "neutral"
strength = 0.2
return {
"netflow_usd": round(netflow, 2),
"inflow_usd": round(inflow_usd, 2),
"outflow_usd": round(outflow_usd, 2),
"netflow_ratio": round(netflow / total_flow, 4) if total_flow > 0 else 0,
"bias": bias,
"strength": round(strength, 2),
"interpretation": (
"Coins flowing TO exchanges โ sell pressure likely"
if netflow > 0
else "Coins flowing FROM exchanges โ accumulation signal"
),
}
@staticmethod
def whale_activity_score(
transactions: List[dict],
min_usd: float = 1_000_000,
) -> dict:
"""
Analyze whale transaction activity.
Args:
transactions: List of dicts with keys: from, to, value_usd, is_exchange_deposit,
is_exchange_withdrawal, timestamp
min_usd: Minimum USD value to count as whale transaction
Returns:
Dict with whale metrics and directional bias
"""
whale_txs = [tx for tx in transactions if tx.get("value_usd", 0) >= min_usd]
if not whale_txs:
return {
"whale_tx_count": 0,
"total_whale_volume": 0,
"exchange_deposits": 0,
"exchange_withdrawals": 0,
"bias": "neutral",
"score": 0.0,
}
total_volume = sum(tx.get("value_usd", 0) for tx in whale_txs)
deposits = sum(
tx.get("value_usd", 0)
for tx in whale_txs
if tx.get("is_exchange_deposit", False)
)
withdrawals = sum(
tx.get("value_usd", 0)
for tx in whale_txs
if tx.get("is_exchange_withdrawal", False)
)
if deposits + withdrawals > 0:
score = (withdrawals - deposits) / (deposits + withdrawals)
else:
score = 0.0
if score > 0.3:
bias = "bullish"
elif score < -0.3:
bias = "bearish"
else:
bias = "neutral"
return {
"whale_tx_count": len(whale_txs),
"total_whale_volume": round(total_volume, 2),
"exchange_deposits": round(deposits, 2),
"exchange_withdrawals": round(withdrawals, 2),
"net_whale_flow": round(withdrawals - deposits, 2),
"bias": bias,
"score": round(score, 3),
}
@staticmethod
def holder_distribution(
holders: List[dict],
) -> dict:
"""
Analyze token holder distribution (concentration risk).
Args:
holders: List of dicts with keys: address, balance, balance_usd
Returns:
Gini coefficient, top holder concentration, distribution tiers
"""
if not holders:
return {"error": "No holder data"}
balances = sorted([h.get("balance", 0) for h in holders], reverse=True)
total = sum(balances)
if total == 0:
return {"error": "Zero total balance"}
n = len(balances)
cumulative = np.cumsum(sorted(balances))
gini = 1 - 2 * np.sum(cumulative) / (n * total) + 1 / n
top_1_pct = balances[0] / total if n >= 1 else 0
top_10_pct = sum(balances[:10]) / total if n >= 10 else sum(balances) / total
top_50_pct = sum(balances[:50]) / total if n >= 50 else sum(balances) / total
whale_threshold = total * 0.01
shark_threshold = total * 0.001
whales = sum(1 for b in balances if b >= whale_threshold)
sharks = sum(1 for b in balances if shark_threshold <= b < whale_threshold)
fish = sum(1 for b in balances if b < shark_threshold)
if top_10_pct > 0.8:
concentration_risk = "EXTREME"
elif top_10_pct > 0.6:
concentration_risk = "HIGH"
elif top_10_pct > 0.4:
concentration_risk = "MODERATE"
else:
concentration_risk = "LOW"
return {
"total_holders": n,
"gini_coefficient": round(gini, 4),
"top_1_holder_pct": round(top_1_pct * 100, 2),
"top_10_holders_pct": round(top_10_pct * 100, 2),
"top_50_holders_pct": round(top_50_pct * 100, 2),
"whales": whales,
"sharks": sharks,
"fish": fish,
"concentration_risk": concentration_risk,
}
@staticmethod
def network_health(
active_addresses_24h: int,
active_addresses_7d_avg: int,
transactions_24h: int,
transactions_7d_avg: int,
hash_rate_current: float = 0,
hash_rate_7d_avg: float = 0,
) -> dict:
"""
Assess network health based on on-chain activity.
Growing activity = bullish fundamental backdrop.
Declining activity = bearish fundamental backdrop.
"""
addr_growth = (
(active_addresses_24h / active_addresses_7d_avg - 1) * 100
if active_addresses_7d_avg > 0
else 0
)
tx_growth = (
(transactions_24h / transactions_7d_avg - 1) * 100
if transactions_7d_avg > 0
else 0
)
hash_growth = (
(hash_rate_current / hash_rate_7d_avg - 1) * 100
if hash_rate_7d_avg > 0
else 0
)
score = (addr_growth * 0.4 + tx_growth * 0.4 + hash_growth * 0.2)
score = max(min(score, 100), -100)
if score > 15:
health = "STRONG"
elif score > 5:
health = "HEALTHY"
elif score > -5:
health = "NEUTRAL"
elif score > -15:
health = "WEAK"
else:
health = "DECLINING"
return {
"active_addresses_24h": active_addresses_24h,
"address_growth_pct": round(addr_growth, 2),
"transactions_24h": transactions_24h,
"tx_growth_pct": round(tx_growth, 2),
"hash_rate_growth_pct": round(hash_growth, 2),
"composite_score": round(score, 2),
"health": health,
}
MEV Detection & Analysis
MEV Detection & Analysis
class MEVDetector:
"""
Maximal Extractable Value (MEV) opportunity detection and analysis.
Covers:
- Sandwich attack detection on pending transactions
- Cross-DEX arbitrage opportunity detection
- Liquidation opportunity scanning
- Backrunning opportunity detection
- Gas price optimization
- MEV protection strategies
"""
@staticmethod
def detect_sandwich_opportunity(
pending_tx_amount: float,
pool_reserve_in: float,
pool_reserve_out: float,
fee: float = 0.003,
gas_cost_usd: float = 5.0,
) -> dict:
"""
Detect if a pending transaction is vulnerable to a sandwich attack.
A sandwich attack front-runs a victim's swap (buying before them to push
price up), then back-runs (selling after them at the higher price).
Args:
pending_tx_amount: Amount the victim is swapping
pool_reserve_in: Pool reserve of input token
pool_reserve_out: Pool reserve of output token
fee: Pool fee tier
gas_cost_usd: Estimated gas cost for the sandwich (2 txs)
"""
victim_output = UniswapV2Analyzer.get_output_amount(
pending_tx_amount, pool_reserve_in, pool_reserve_out, fee
)
victim_impact = UniswapV2Analyzer.get_price_impact(
pending_tx_amount, pool_reserve_in, pool_reserve_out, fee
)
if victim_impact < 0.001:
return {
"opportunity": False,
"reason": "Price impact too small to sandwich",
"victim_impact_pct": round(victim_impact * 100, 4),
}
frontrun_amount = pending_tx_amount * 0.5
frontrun_output = UniswapV2Analyzer.get_output_amount(
frontrun_amount, pool_reserve_in, pool_reserve_out, fee
)
new_reserve_in = pool_reserve_in + frontrun_amount
new_reserve_out = pool_reserve_out - frontrun_output
victim_output_after = UniswapV2Analyzer.get_output_amount(
pending_tx_amount, new_reserve_in, new_reserve_out, fee
)
new_reserve_in_2 = new_reserve_in + pending_tx_amount
new_reserve_out_2 = new_reserve_out - victim_output_after
backrun_output = UniswapV2Analyzer.get_output_amount(
frontrun_output, new_reserve_out_2, new_reserve_in_2, fee
)
profit = backrun_output - frontrun_amount - gas_cost_usd
return {
"opportunity": profit > 0,
"estimated_profit_usd": round(profit, 2),
"frontrun_amount": round(frontrun_amount, 4),
"victim_impact_pct": round(victim_impact * 100, 4),
"victim_extra_slippage_pct": round(
(1 - victim_output_after / victim_output) * 100, 4
),
"gas_cost_usd": gas_cost_usd,
"profit_after_gas": round(profit, 2),
"warning": "MEV exploitation is ethically controversial. Use for defense/awareness only.",
}
@staticmethod
def detect_cross_dex_arbitrage(
dex_prices: Dict[str, float],
trade_size_usd: float = 10000,
gas_cost_usd: float = 10.0,
) -> List[dict]:
"""
Detect cross-DEX arbitrage opportunities.
Args:
dex_prices: Dict of DEX name โ price for the same pair
trade_size_usd: Size of arbitrage trade
gas_cost_usd: Total gas cost for the round trip
Returns:
List of arbitrage opportunities sorted by profit
"""
opportunities = []
dex_list = list(dex_prices.items())
for i, (dex_buy, price_buy) in enumerate(dex_list):
for j, (dex_sell, price_sell) in enumerate(dex_list):
if i == j:
continue
if price_sell > price_buy:
spread_pct = (price_sell - price_buy) / price_buy * 100
tokens_bought = trade_size_usd / price_buy
revenue = tokens_bought * price_sell
profit = revenue - trade_size_usd - gas_cost_usd
if profit > 0:
opportunities.append({
"buy_dex": dex_buy,
"sell_dex": dex_sell,
"buy_price": round(price_buy, 6),
"sell_price": round(price_sell, 6),
"spread_pct": round(spread_pct, 4),
"trade_size_usd": trade_size_usd,
"estimated_profit_usd": round(profit, 2),
"profit_pct": round(profit / trade_size_usd * 100, 4),
"gas_cost_usd": gas_cost_usd,
})
return sorted(opportunities, key=lambda x: x["estimated_profit_usd"], reverse=True)
@staticmethod
def detect_liquidation_opportunities(
positions: List[dict],
current_prices: Dict[str, float],
) -> List[dict]:
"""
Detect DeFi lending positions approaching liquidation.
Args:
positions: List of dicts with keys: owner, collateral_token, collateral_amount,
debt_token, debt_amount, liquidation_threshold, collateral_price_at_open
current_prices: Dict of token โ current price
"""
opportunities = []
for pos in positions:
collateral_token = pos.get("collateral_token", "")
debt_token = pos.get("debt_token", "")
current_collateral_price = current_prices.get(collateral_token, 0)
current_debt_price = current_prices.get(debt_token, 1)
if current_collateral_price == 0:
continue
collateral_value = pos.get("collateral_amount", 0) * current_collateral_price
debt_value = pos.get("debt_amount", 0) * current_debt_price
liq_threshold = pos.get("liquidation_threshold", 0.8)
if collateral_value == 0:
continue
health_factor = (collateral_value * liq_threshold) / debt_value if debt_value > 0 else float("inf")
ltv = debt_value / collateral_value
liq_price = (debt_value / (pos.get("collateral_amount", 1) * liq_threshold))
distance_to_liq = (current_collateral_price - liq_price) / current_collateral_price
if health_factor < 1.1:
liq_bonus_pct = 0.05
profit = collateral_value * liq_bonus_pct - debt_value * 0.01
opportunities.append({
"owner": pos.get("owner", "unknown"),
"collateral_token": collateral_token,
"debt_token": debt_token,
"collateral_value_usd": round(collateral_value, 2),
"debt_value_usd": round(debt_value, 2),
"health_factor": round(health_factor, 4),
"ltv": round(ltv * 100, 2),
"distance_to_liq_pct": round(distance_to_liq * 100, 2),
"estimated_profit_usd": round(profit, 2),
"urgency": "IMMINENT" if health_factor < 1.01 else "APPROACHING",
})
return sorted(opportunities, key=lambda x: x["health_factor"])
class GasOptimizer:
"""Gas price analysis and optimization for transaction timing."""
@staticmethod
def analyze_gas_history(
gas_prices: List[dict],
) -> dict:
"""
Analyze historical gas prices to find optimal trading windows.
Args:
gas_prices: List of dicts with keys: timestamp, gas_gwei, block_number
"""
if not gas_prices:
return {"error": "No gas data"}
prices = [g.get("gas_gwei", 0) for g in gas_prices]
hourly = {}
for g in gas_prices:
ts = g.get("timestamp")
if ts and hasattr(ts, "hour"):
hour = ts.hour
if hour not in hourly:
hourly[hour] = []
hourly[hour].append(g.get("gas_gwei", 0))
hourly_avg = {h: round(np.mean(v), 1) for h, v in hourly.items()}
cheapest_hour = min(hourly_avg, key=hourly_avg.get) if hourly_avg else 0
return {
"current_gwei": round(prices[-1] if prices else 0, 1),
"avg_gwei": round(np.mean(prices), 1),
"median_gwei": round(np.median(prices), 1),
"min_gwei": round(min(prices), 1),
"max_gwei": round(max(prices), 1),
"p25_gwei": round(np.percentile(prices, 25), 1),
"p75_gwei": round(np.percentile(prices, 75), 1),
"cheapest_hour_utc": cheapest_hour,
"cheapest_hour_avg_gwei": hourly_avg.get(cheapest_hour, 0),
"hourly_averages": hourly_avg,
"recommendation": (
f"Best time to transact: ~{cheapest_hour}:00 UTC "
f"(avg {hourly_avg.get(cheapest_hour, 0)} gwei)"
),
}
@staticmethod
def estimate_cost(
gas_limit: int,
gas_price_gwei: float,
eth_price_usd: float,
) -> dict:
"""Estimate transaction cost in USD."""
gas_cost_eth = gas_limit * gas_price_gwei * 1e-9
gas_cost_usd = gas_cost_eth * eth_price_usd
return {
"gas_limit": gas_limit,
"gas_price_gwei": gas_price_gwei,
"cost_eth": round(gas_cost_eth, 6),
"cost_usd": round(gas_cost_usd, 2),
}
Yield Farming Analyzer
Yield Farming Analyzer
class YieldFarmingAnalyzer:
"""
DeFi yield farming analysis and comparison.
Covers:
- APY/APR calculation with compounding
- Farming strategy comparison
- Risk-adjusted yield analysis
- Auto-compound optimization
- Farming P&L tracking
"""
@staticmethod
def calculate_apy(
apr: float,
compounds_per_year: int = 365,
) -> float:
"""
Convert APR to APY with compounding.
APY = (1 + APR/n)^n - 1
Args:
apr: Annual Percentage Rate as decimal (e.g., 0.5 = 50%)
compounds_per_year: Number of compounding periods
"""
if compounds_per_year == 0:
return apr
return (1 + apr / compounds_per_year) ** compounds_per_year - 1
@staticmethod
def calculate_apr(
daily_reward_usd: float,
total_staked_usd: float,
) -> float:
"""Calculate APR from daily rewards."""
if total_staked_usd == 0:
return 0.0
return (daily_reward_usd * 365) / total_staked_usd
@staticmethod
def compare_farms(
farms: List[dict],
) -> pd.DataFrame:
"""
Compare yield farming opportunities.
Args:
farms: List of dicts with keys: name, apr, tvl_usd, token_reward,
il_risk (low/medium/high), smart_contract_risk (low/medium/high),
chain, protocol
"""
rows = []
for farm in farms:
apr = farm.get("apr", 0)
apy = YieldFarmingAnalyzer.calculate_apy(apr / 100) * 100
il_scores = {"low": 2, "medium": 5, "high": 8}
sc_scores = {"low": 1, "medium": 4, "high": 7}
il_risk = il_scores.get(farm.get("il_risk", "medium"), 5)
sc_risk = sc_scores.get(farm.get("smart_contract_risk", "medium"), 4)
risk_score = (il_risk + sc_risk) / 2
risk_adjusted = apr / risk_score if risk_score > 0 else 0
rows.append({
"name": farm.get("name", ""),
"protocol": farm.get("protocol", ""),
"chain": farm.get("chain", ""),
"apr_pct": round(apr, 2),
"apy_pct": round(apy, 2),
"tvl_usd": farm.get("tvl_usd", 0),
"reward_token": farm.get("token_reward", ""),
"il_risk": farm.get("il_risk", "medium"),
"sc_risk": farm.get("smart_contract_risk", "medium"),
"risk_score": round(risk_score, 1),
"risk_adjusted_yield": round(risk_adjusted, 2),
})
df = pd.DataFrame(rows)
return df.sort_values("risk_adjusted_yield", ascending=False).reset_index(drop=True)
@staticmethod
def optimal_compound_frequency(
apr: float,
gas_cost_usd: float,
position_size_usd: float,
) -> dict:
"""
Calculate the optimal compounding frequency.
More frequent compounding increases APY but costs more gas.
Find the sweet spot that maximizes net yield.
"""
best_freq = 1
best_net_yield = 0
results = []
for freq in [1, 2, 4, 7, 14, 30, 90, 182, 365]:
compounds_per_year = 365 / freq
apy = YieldFarmingAnalyzer.calculate_apy(apr, int(compounds_per_year))
total_gas_cost = gas_cost_usd * compounds_per_year
gross_yield = position_size_usd * apy
net_yield = gross_yield - total_gas_cost
net_apy = net_yield / position_size_usd if position_size_usd > 0 else 0
results.append({
"compound_every_days": freq,
"compounds_per_year": int(compounds_per_year),
"gross_apy_pct": round(apy * 100, 2),
"gas_cost_annual": round(total_gas_cost, 2),
"net_yield_usd": round(net_yield, 2),
"net_apy_pct": round(net_apy * 100, 2),
})
if net_yield > best_net_yield:
best_net_yield = net_yield
best_freq = freq
return {
"optimal_frequency_days": best_freq,
"optimal_net_apy_pct": round(best_net_yield / position_size_usd * 100, 2) if position_size_usd > 0 else 0,
"all_frequencies": results,
}
DeFi Risk Analyzer
DeFi Risk Analyzer
class DeFiRiskAnalyzer:
"""
Comprehensive DeFi protocol and position risk analysis.
Evaluates:
- Smart contract risk (audit status, age, TVL)
- Liquidity risk (depth, concentration)
- Oracle risk (price feed reliability)
- Governance risk (centralization, admin keys)
- Market risk (volatility, correlation)
- Composability risk (dependency chains)
"""
@staticmethod
def protocol_risk_score(
has_audit: bool = False,
audit_firms: int = 0,
age_days: int = 0,
tvl_usd: float = 0,
bug_bounty_usd: float = 0,
has_timelock: bool = False,
has_multisig: bool = False,
is_upgradeable: bool = True,
has_admin_key: bool = True,
fork_of: str = "",
open_source: bool = True,
) -> dict:
"""
Calculate a comprehensive protocol risk score (0-100, lower = safer).
"""
score = 50
factors = []
if has_audit:
score -= 10
factors.append("Audited (-10)")
if audit_firms >= 2:
score -= 5
factors.append("Multiple audits (-5)")
if audit_firms >= 3:
score -= 5
factors.append("3+ audit firms (-5)")
else:
score += 15
factors.append("No audit (+15)")
if age_days > 365:
score -= 15
factors.append("1+ year old (-15)")
elif age_days > 180:
score -= 10
factors.append("6+ months old (-10)")
elif age_days > 90:
score -= 5
factors.append("3+ months old (-5)")
else:
score += 10
factors.append("Very new protocol (+10)")
if tvl_usd > 1_000_000_000:
score -= 10
factors.append("$1B+ TVL (-10)")
elif tvl_usd > 100_000_000:
score -= 5
factors.append("$100M+ TVL (-5)")
elif tvl_usd < 1_000_000:
score += 10
factors.append("Low TVL (<$1M) (+10)")
if bug_bounty_usd > 1_000_000:
score -= 5
factors.append("Large bug bounty (-5)")
elif bug_bounty_usd > 0:
score -= 2
factors.append("Has bug bounty (-2)")
if has_timelock:
score -= 5
factors.append("Timelock (-5)")
if has_multisig:
score -= 5
factors.append("Multisig (-5)")
if not is_upgradeable:
score -= 5
factors.append("Immutable (-5)")
if has_admin_key:
score += 10
factors.append("Admin key (+10)")
if not open_source:
score += 15
factors.append("Closed source (+15)")
score = max(0, min(100, score))
if score < 25:
risk_level = "LOW"
elif score < 50:
risk_level = "MODERATE"
elif score < 75:
risk_level = "HIGH"
else:
risk_level = "EXTREME"
return {
"risk_score": score,
"risk_level": risk_level,
"factors": factors,
"recommendation": (
"Acceptable for large allocations"
if score < 25
else "Use with caution, limit exposure"
if score < 50
else "High risk โ small allocations only"
if score < 75
else "Extreme risk โ avoid or use minimal amounts"
),
}
@staticmethod
def position_risk_assessment(
position_value_usd: float,
portfolio_total_usd: float,
protocol_risk_score: int,
il_risk_pct: float,
token_volatility_30d: float,
is_stablecoin_pair: bool = False,
) -> dict:
"""
Assess risk for a specific DeFi position.
"""
allocation_pct = (
position_value_usd / portfolio_total_usd * 100
if portfolio_total_usd > 0
else 100
)
if protocol_risk_score < 25:
max_allocation = 30
elif protocol_risk_score < 50:
max_allocation = 15
elif protocol_risk_score < 75:
max_allocation = 5
else:
max_allocation = 2
if is_stablecoin_pair:
max_allocation = min(max_allocation * 2, 50)
expected_il = il_risk_pct / 100
protocol_loss_prob = protocol_risk_score / 100 * 0.1
expected_loss = position_value_usd * (expected_il + protocol_loss_prob)
warnings = []
if allocation_pct > max_allocation:
warnings.append(
f"Position too large: {allocation_pct:.1f}% vs recommended max {max_allocation}%"
)
if token_volatility_30d > 1.0:
warnings.append(f"High token volatility: {token_volatility_30d:.0%} (30d)")
if protocol_risk_score > 60:
warnings.append("High protocol risk โ consider reducing exposure")
return {
"position_value_usd": round(position_value_usd, 2),
"allocation_pct": round(allocation_pct, 2),
"max_recommended_allocation_pct": max_allocation,
"protocol_risk_score": protocol_risk_score,
"expected_il_pct": round(il_risk_pct, 2),
"expected_loss_usd": round(expected_loss, 2),
"token_volatility_30d": round(token_volatility_30d * 100, 1),
"warnings": warnings,
"within_limits": len(warnings) == 0,
}
SMC Application to Crypto โ Beginner to Intermediate (Smart Risk)
Source: video_037 โ How to Trade Crypto Coins for Beginners
- BTC dominance rule: Always check BTC D1 bias before trading any altcoin; alts amplify BTC moves (typical beta: ALT moves ~3x BTC % move)
- SMT Divergence for crypto: If BTC makes new high but ETH doesn't โ BTC likely reverses; use at HTF POIs
- Market cap tier rules: Large cap (BTC/ETH) = SMC works cleanly; mid cap (SOL/BNB/ADA) = SMC works with some manipulation noise; small cap (<$500M) = insufficient liquidity for reliable SMC; micro cap = skip SMC, use only as speculative position
- Volatility adjustments: Use 1.5x ATR minimum for stops; target 3:1+ RR to compensate; wider OB zones expected
- Session timing: London open (08:00โ10:00 UTC) and NY open (13:00โ16:00 UTC) = highest quality crypto setups; avoid Asian session entries
- Meme coin rule: Position size = 1-2% max; enter only during early accumulation (unknown stage); never chase momentum
- Weekend warning: Lower volume weekends โ erratic behavior โ reduce position size or avoid entries entirely
BTC Dominance Technical Setup & Altcoin Rotation (Smart Risk)
Source: video_079 โ How to Start Trading Crypto the Right Way
- Altcoin season signal: BTC at/near ATH + BTC.D (Bitcoin Dominance) making D1 BOS downward = early altcoin season confirmed; capital rotating from BTC to altcoins
- BTC.D technical reading: Open BTC.D on TradingView; bearish BOS on D1 + bearish FVG on BTC.D = further dominance decline โ alts outperform
- Sequential rotation order: Large-cap alts first (ETH, SOL, BNB) โ once trend confirmed โ mid-cap alts; never jump to micro-caps first
- Entry timing in bull run: Do NOT FOMO buy after 50-100% pump; wait for first meaningful pullback to H4/D1 FVG in discount zone; same SMC entry mechanics apply
- Risk management in bull run: 0.5-1% per altcoin; max 3% total crypto exposure; diversify across 3-5 alts
- Exit signal for altcoin season: BTC.D reverses (starts rising again) + BTC D1 shows LH โ rotate back to BTC or cash; alts stop making new ATHs before BTC peaks
Related Skills
