RugMunchMedia/rmi-backend
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
rmi-backend/tests/unit/test_liquidation_cascade_analyzer.py
opencode c762564d40 style(rmi-backend): complete lint cleanup — 1175→0 ruff errors 
- Fix 71 invalid-syntax files (class-body newline-broken assignments)
- Add from/None chain to 307 B904 raise-without-from sites
- Add B008 ignore to ruff.toml (already in pyproject.toml)
- Noqa F401 on __init__.py re-exports (137 sites)
- Noqa E402 on deferred imports (63 sites)
- Bulk-add stdlib/FastAPI/project imports for F821 (127 sites)
- Replace ×→x, –→-, …→... in docstrings (4093 chars)
- Manual refactor of 5 SIM103/SIM116 patterns

Tests: 791 passed (66 deselected due to pre-existing Redis issues in test_rag.py)
Co-authored-by: opencode <opencode@rugmunch.io>
2026-07-06 15:43:20 +02:00

731 lines
29 KiB
Python
Raw Permalink Blame History

"""
Tests for Liquidation Cascade Risk Analyzer
============================================
Tests cover:
- Address validation (EVM + Solana)
- CollateralPosition creation and serialization
- DebtPosition creation and serialization
- ProtocolPosition health computation (all risk tiers)
- LiquidationAnalysis pipeline (scenarios, clusters, reporting)
- Edge cases: empty wallet, no debt, invalid addresses, missing Web3
"""
import json
import os
import sys
# Add app path so we can import
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from app.liquidation_cascade_analyzer import (
CascadeScenario,
CollateralPosition,
DebtPosition,
LiquidationAnalysis,
LiquidationCascadeAnalyzer,
LiquidationCluster,
ProtocolPosition,
RiskTier,
_estimate_asset_ltv,
_estimate_asset_price,
_resolve_asset_symbol,
)
# ══════════════════════════════════════════════════════════════════════════
# Address Validation
# ══════════════════════════════════════════════════════════════════════════
class TestAddressValidation:
def setup_method(self):
self.analyzer = LiquidationCascadeAnalyzer()
def test_valid_evm_address(self):
assert self.analyzer._validate_address("0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18")
def test_valid_evm_address_lowercase(self):
assert self.analyzer._validate_address("0x742d35cc6634c0532925a3b844bc9e7595f2bd18")
def test_valid_solana_address(self):
assert self.analyzer._validate_address("7EcDhSYGxXyscszYEp35KHN8vvw3svAuLKTzXwCFLrH")
def test_invalid_address_too_short(self):
assert not self.analyzer._validate_address("0x1234")
def test_invalid_address_bad_prefix(self):
assert not self.analyzer._validate_address("1x742d35Cc6634C0532925a3b844Bc9e7595f2bD18")
def test_invalid_address_empty(self):
assert not self.analyzer._validate_address("")
def test_invalid_address_random_string(self):
assert not self.analyzer._validate_address("not-an-address")
# ══════════════════════════════════════════════════════════════════════════
# CollateralPosition
# ══════════════════════════════════════════════════════════════════════════
class TestCollateralPosition:
def test_create_basic(self):
pos = CollateralPosition(
asset="WETH",
asset_address="0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2",
amount_usd=50000.0,
amount_token=17.857,
ltv=0.80,
liquidation_threshold=0.83,
price_usd=2800.0,
)
assert pos.asset == "WETH"
assert pos.amount_usd == 50000.0
assert pos.amount_token == 17.857
assert pos.ltv == 0.80
assert pos.liquidation_threshold == 0.83
def test_to_dict(self):
pos = CollateralPosition(
asset="USDC",
asset_address="0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
amount_usd=10000.0,
amount_token=10000.0,
ltv=0.80,
liquidation_threshold=0.85,
price_usd=1.0,
)
d = pos.to_dict()
assert d["asset"] == "USDC"
assert d["amount_usd"] == 10000.0
assert d["liquidation_threshold"] == 0.85
def test_zero_amount(self):
pos = CollateralPosition(
asset="ETH",
asset_address="0x0000000000000000000000000000000000000000",
amount_usd=0.0,
amount_token=0.0,
ltv=0.80,
liquidation_threshold=0.83,
price_usd=2800.0,
)
assert pos.amount_usd == 0.0
assert pos.amount_token == 0.0
# ══════════════════════════════════════════════════════════════════════════
# DebtPosition
# ══════════════════════════════════════════════════════════════════════════
class TestDebtPosition:
def test_create_basic(self):
pos = DebtPosition(
asset="USDC",
asset_address="0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
amount_usd=20000.0,
amount_token=20000.0,
variable_rate=5.0,
)
assert pos.asset == "USDC"
assert pos.amount_usd == 20000.0
assert pos.variable_rate == 5.0
def test_with_stable_rate(self):
pos = DebtPosition(
asset="USDC",
asset_address="0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
amount_usd=10000.0,
amount_token=10000.0,
variable_rate=3.5,
stable_rate=4.2,
)
assert pos.stable_rate == 4.2
def test_to_dict(self):
pos = DebtPosition(
asset="DAI",
asset_address="0x6b175474e89094c44da98b954eedeac495271d0f",
amount_usd=5000.0,
amount_token=5000.0,
variable_rate=4.8,
)
d = pos.to_dict()
assert d["asset"] == "DAI"
assert d["variable_rate"] == 4.8
# ══════════════════════════════════════════════════════════════════════════
# ProtocolPosition - Health Computation
# ══════════════════════════════════════════════════════════════════════════
class TestProtocolPositionHealth:
def make_position(
self,
coll_usd: float = 100000.0,
debt_usd: float = 0.0,
liq_threshold: float = 0.83,
coll_asset: str = "WETH",
) -> ProtocolPosition:
coll = CollateralPosition(
asset=coll_asset,
asset_address="0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2",
amount_usd=coll_usd,
amount_token=coll_usd / 2800.0,
ltv=liq_threshold * 0.95,
liquidation_threshold=liq_threshold,
price_usd=2800.0,
)
debt = (
DebtPosition(
asset="USDC",
asset_address="0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
amount_usd=debt_usd,
amount_token=debt_usd,
variable_rate=5.0,
)
if debt_usd > 0
else None
)
pos = ProtocolPosition(
protocol="Aave V3",
chain="ethereum",
wallet="0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18",
collateral=[coll],
debt=[debt] if debt else [],
total_collateral_usd=coll_usd,
total_debt_usd=debt_usd,
)
pos.compute_health()
return pos
def test_safe_no_debt(self):
pos = self.make_position(debt_usd=0)
assert pos.risk_tier == RiskTier.SAFE
assert pos.health_factor is None or pos.health_factor == float("inf")
def test_safe_low_debt(self):
# $100k collateral, $20k debt, 83% threshold
# HF = (100000 * 0.83) / 20000 = 4.15 (> 2.0 → SAFE)
pos = self.make_position(debt_usd=20000.0)
assert pos.risk_tier == RiskTier.SAFE
assert pos.health_factor is not None
assert pos.health_factor >= 2.0
def test_watch_moderate_debt(self):
# $100k collateral, $55k debt, 83% threshold
# HF = (100000 * 0.83) / 55000 = 1.51 (> 1.5 → WATCH)
pos = self.make_position(debt_usd=55000.0)
assert pos.health_factor is not None
assert pos.risk_tier == RiskTier.WATCH, f"Expected WATCH, got {pos.risk_tier} (HF={pos.health_factor})"
assert 1.5 <= pos.health_factor < 2.0
def test_danger_high_debt(self):
# $100k collateral, $70k debt, 83% threshold
# HF = (100000 * 0.83) / 70000 = 1.19 (> 1.1 → DANGER)
pos = self.make_position(debt_usd=70000.0)
assert pos.health_factor is not None
assert pos.risk_tier == RiskTier.DANGER, f"Expected DANGER, got {pos.risk_tier} (HF={pos.health_factor})"
assert 1.1 <= pos.health_factor < 1.5
def test_critical_extreme_debt(self):
# $100k collateral, $95k debt, 83% threshold
# HF = (100000 * 0.83) / 95000 = 0.87 (< 1.1 → CRITICAL)
pos = self.make_position(debt_usd=95000.0)
assert pos.health_factor is not None
assert pos.risk_tier == RiskTier.CRITICAL, f"Expected CRITICAL, got {pos.risk_tier} (HF={pos.health_factor})"
assert 0 < pos.health_factor < 1.1
def test_liquidation_price_computed(self):
pos = self.make_position(debt_usd=50000.0, coll_usd=100000.0)
assert pos.liquidation_price_usd is not None
# 50000 / (100000/2800 * 0.83) = 50000 / (35.71 * 0.83) = 50000 / 29.64 = ~1686
expected = 50000.0 / ((100000.0 / 2800.0) * 0.83)
assert abs(pos.liquidation_price_usd - expected) < 1.0
def test_multiple_collateral_weighted(self):
"""Test health factor with multiple collateral assets."""
weth = CollateralPosition(
asset="WETH",
asset_address="0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2",
amount_usd=60000.0,
amount_token=21.43,
ltv=0.76,
liquidation_threshold=0.79,
price_usd=2800.0,
)
usdc = CollateralPosition(
asset="USDC",
asset_address="0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
amount_usd=40000.0,
amount_token=40000.0,
ltv=0.80,
liquidation_threshold=0.85,
price_usd=1.0,
)
debt = DebtPosition(
asset="USDC",
asset_address="0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
amount_usd=50000.0,
amount_token=50000.0,
variable_rate=5.0,
)
pos = ProtocolPosition(
protocol="Aave V3",
chain="ethereum",
wallet="0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18",
collateral=[weth, usdc],
debt=[debt],
total_collateral_usd=100000.0,
total_debt_usd=50000.0,
)
pos.compute_health()
# Weighted threshold: (60000*0.79 + 40000*0.85) / 100000 = (47400+34000)/100000 = 0.814
# HF = (100000 * 0.814) / 50000 = 1.628
assert pos.health_factor is not None
assert pos.risk_tier == RiskTier.WATCH, f"Expected WATCH, got {pos.risk_tier} (HF={pos.health_factor})"
assert 1.5 < pos.health_factor < 1.8
# ══════════════════════════════════════════════════════════════════════════
# LiquidationAnalysis Pipeline
# ══════════════════════════════════════════════════════════════════════════
class TestLiquidationAnalysis:
@staticmethod
def _make_sample_position(
debt_usd: float = 50000.0,
coll_usd: float = 100000.0,
chain: str = "ethereum",
protocol: str = "Aave V3",
) -> ProtocolPosition:
coll = CollateralPosition(
asset="WETH",
asset_address="0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2",
amount_usd=coll_usd,
amount_token=coll_usd / 2800.0,
ltv=0.76,
liquidation_threshold=0.79,
price_usd=2800.0,
)
debt = DebtPosition(
asset="USDC",
asset_address="0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
amount_usd=debt_usd,
amount_token=debt_usd,
variable_rate=5.0,
)
pos = ProtocolPosition(
protocol=protocol,
chain=chain,
wallet="0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18",
collateral=[coll],
debt=[debt],
total_collateral_usd=coll_usd,
total_debt_usd=debt_usd,
)
pos.compute_health()
return pos
def test_empty_analysis(self):
analysis = LiquidationAnalysis(
wallet="0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18",
chains_analyzed=[],
)
analysis.analyze()
assert analysis.total_collateral_usd == 0.0
assert analysis.total_debt_usd == 0.0
assert analysis.overall_health_factor is None
assert len(analysis.cascade_scenarios) == 0
assert len(analysis.liquidation_clusters) == 0
def test_single_safe_position(self):
pos = self._make_sample_position(debt_usd=10000.0, coll_usd=100000.0)
analysis = LiquidationAnalysis(
wallet="0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18",
chains_analyzed=["ethereum"],
positions=[pos],
)
analysis.analyze()
assert analysis.total_collateral_usd == 100000.0
assert analysis.total_debt_usd == 10000.0
assert analysis.overall_risk_tier == RiskTier.SAFE
def test_multiple_chain_aggregation(self):
pos1 = self._make_sample_position(debt_usd=80000.0, coll_usd=100000.0, chain="ethereum")
pos2 = self._make_sample_position(debt_usd=5000.0, coll_usd=50000.0, chain="base")
analysis = LiquidationAnalysis(
wallet="0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18",
chains_analyzed=["ethereum", "base"],
positions=[pos1, pos2],
)
analysis.analyze()
assert analysis.total_collateral_usd == 150000.0
assert analysis.total_debt_usd == 85000.0
assert analysis.overall_risk_tier in (RiskTier.WATCH, RiskTier.DANGER)
def test_cascade_scenarios_generated(self):
# One critical position should generate cascade scenarios
pos = self._make_sample_position(debt_usd=95000.0, coll_usd=100000.0)
analysis = LiquidationAnalysis(
wallet="0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18",
chains_analyzed=["ethereum"],
positions=[pos],
)
analysis.analyze()
assert len(analysis.cascade_scenarios) > 0
def test_report_text_format(self):
pos = self._make_sample_position(debt_usd=50000.0, coll_usd=100000.0)
analysis = LiquidationAnalysis(
wallet="0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18",
chains_analyzed=["ethereum"],
positions=[pos],
)
analysis.analyze()
report = analysis.report(format="text")
assert "LIQUIDATION CASCADE RISK ANALYSIS" in report
assert "0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18" in report
assert "POSITION BREAKDOWN" in report
assert "OVERALL PORTFOLIO HEALTH" in report
def test_report_json_format(self):
pos = self._make_sample_position(debt_usd=50000.0, coll_usd=100000.0)
analysis = LiquidationAnalysis(
wallet="0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18",
chains_analyzed=["ethereum"],
positions=[pos],
)
analysis.analyze()
json_str = analysis.report(format="json")
data = json.loads(json_str)
assert data["wallet"] == "0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18"
assert "overall_risk_tier" in data
assert "positions" in data
assert len(data["positions"]) == 1
def test_warnings_and_errors_in_report(self):
analysis = LiquidationAnalysis(
wallet="0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18",
chains_analyzed=["ethereum"],
errors=["Failed to connect to RPC"],
warnings=["Web3 unavailable"],
)
report = analysis.report()
assert "Failed to connect to RPC" in report
assert "Web3 unavailable" in report
def test_invalid_address_analysis(self):
"""Verify the analyzer's _validate_address rejects bad addresses."""
analyzer = LiquidationCascadeAnalyzer()
assert not analyzer._validate_address("invalid-address")
# ══════════════════════════════════════════════════════════════════════════
# Helper Functions
# ══════════════════════════════════════════════════════════════════════════
class TestHelperFunctions:
def test_estimate_asset_ltv_stablecoin(self):
ltv, liq = _estimate_asset_ltv("USDC")
assert ltv >= 0.78
assert liq >= 0.83
def test_estimate_asset_ltv_eth(self):
ltv, liq = _estimate_asset_ltv("WETH")
assert ltv == 0.80
assert liq == 0.83
def test_estimate_asset_ltv_unknown(self):
ltv, liq = _estimate_asset_ltv("UNKNOWN_TOKEN")
assert ltv == 0.50
assert liq == 0.55
def test_estimate_asset_price_known(self):
assert _estimate_asset_price("ETH") == 2800.0
assert _estimate_asset_price("USDC") == 1.0
assert _estimate_asset_price("WBTC") == 68000.0
def test_estimate_asset_price_unknown(self):
assert _estimate_asset_price("UNKNOWN") == 1.0
def test_resolve_asset_symbol_weth(self):
addr = "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2"
assert _resolve_asset_symbol(addr, "ethereum") == "WETH"
def test_resolve_asset_symbol_usdc_base(self):
addr = "0x833589fcd6edb6e08f4c7c32d4f71b54bda02913"
assert _resolve_asset_symbol(addr, "base") == "USDC"
def test_resolve_asset_symbol_unknown(self):
addr = "0xdead000000000000000000000000000000000000"
sym = _resolve_asset_symbol(addr, "ethereum")
assert "0xdead" in sym
def test_validate_address_solana_variants(self):
"""Test various valid Solana address formats."""
analyzer = LiquidationCascadeAnalyzer()
valid_addresses = [
"7EcDhSYGxXyscszYEp35KHN8vvw3svAuLKTzXwCFLrH",
"DpRueBHHhrqMATHrYgvKQzFJFynfMFVPMgfzJgrXqKnQ",
"So11111111111111111111111111111111111111112",
]
for addr in valid_addresses:
assert analyzer._validate_address(addr), f"Expected valid: {addr}"
# ══════════════════════════════════════════════════════════════════════════
# CascadeScenario Model
# ══════════════════════════════════════════════════════════════════════════
class TestCascadeScenario:
def test_create_scenario(self):
scenario = CascadeScenario(
name="Test Crash",
description="A test scenario",
liquidated_positions=3,
total_liquidated_value_usd=150000.0,
secondary_affected_positions=5,
total_secondary_value_usd=250000.0,
market_impact_pct=0.15,
)
assert scenario.liquidated_positions == 3
assert scenario.total_liquidated_value_usd == 150000.0
assert scenario.market_impact_pct == 0.15
def test_to_dict(self):
scenario = CascadeScenario(
name="10% Drop",
description="Simulate 10% drop",
liquidated_positions=2,
total_liquidated_value_usd=50000.0,
)
d = scenario.to_dict()
assert d["name"] == "10% Drop"
assert d["liquidated_positions"] == 2
# ══════════════════════════════════════════════════════════════════════════
# LiquidationCluster Model
# ══════════════════════════════════════════════════════════════════════════
class TestLiquidationCluster:
def test_create_cluster(self):
cluster = LiquidationCluster(
chain="ethereum",
primary_collateral="WETH",
price_range_low=1600.0,
price_range_high=1800.0,
wallet_count=5,
total_debt_usd=500000.0,
total_collateral_usd=1000000.0,
)
assert cluster.wallet_count == 5
assert cluster.price_range_low == 1600.0
def test_to_dict(self):
cluster = LiquidationCluster(
chain="base",
primary_collateral="ETH",
price_range_low=1500.0,
price_range_high=1700.0,
wallet_count=3,
total_debt_usd=200000.0,
total_collateral_usd=400000.0,
)
d = cluster.to_dict()
assert d["chain"] == "base"
assert d["wallet_count"] == 3
# ══════════════════════════════════════════════════════════════════════════
# RiskTier Enum
# ══════════════════════════════════════════════════════════════════════════
class TestRiskTier:
def test_score_ordering(self):
assert RiskTier.SAFE.score() == 0
assert RiskTier.WATCH.score() == 1
assert RiskTier.DANGER.score() == 2
assert RiskTier.CRITICAL.score() == 3
def test_string_values(self):
assert RiskTier.SAFE.value == "SAFE"
assert RiskTier.CRITICAL.value == "CRITICAL"
def test_from_string(self):
assert RiskTier("SAFE") == RiskTier.SAFE
assert RiskTier("CRITICAL") == RiskTier.CRITICAL
# ══════════════════════════════════════════════════════════════════════════
# Edge Cases
# ══════════════════════════════════════════════════════════════════════════
class TestEdgeCases:
def test_position_zero_collateral_no_health_factor(self):
pos = ProtocolPosition(
protocol="Aave V3",
chain="ethereum",
wallet="0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18",
collateral=[],
debt=[],
total_collateral_usd=0.0,
total_debt_usd=0.0,
)
pos.compute_health()
assert pos.health_factor == float("inf")
assert pos.risk_tier == RiskTier.SAFE
def test_position_with_debt_but_no_collateral(self):
"""Edge case: position with debt but zero collateral computed health."""
debt = DebtPosition(
asset="USDC",
asset_address="0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
amount_usd=5000.0,
amount_token=5000.0,
variable_rate=5.0,
)
pos = ProtocolPosition(
protocol="Aave V3",
chain="ethereum",
wallet="0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18",
collateral=[],
debt=[debt],
total_collateral_usd=0.0,
total_debt_usd=5000.0,
)
pos.compute_health()
# With no collateral, health factor computation should handle gracefully
assert pos.health_factor == float("inf") # Division by zero avoided
assert pos.risk_tier == RiskTier.SAFE
def test_mixed_risk_positions_aggregation(self):
"""Multiple positions with different risk tiers."""
safe = ProtocolPosition(
protocol="Aave V3",
chain="ethereum",
wallet="0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18",
collateral=[
CollateralPosition(
asset="WETH",
asset_address="0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2",
amount_usd=200000.0,
amount_token=71.43,
ltv=0.76,
liquidation_threshold=0.79,
price_usd=2800.0,
)
],
debt=[
DebtPosition(
asset="USDC",
asset_address="0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
amount_usd=10000.0,
amount_token=10000.0,
variable_rate=5.0,
)
],
total_collateral_usd=200000.0,
total_debt_usd=10000.0,
)
safe.compute_health()
critical = ProtocolPosition(
protocol="Aave V3",
chain="arbitrum",
wallet="0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18",
collateral=[
CollateralPosition(
asset="WETH",
asset_address="0x82af49447d8a07e3bd95bd0d56f35241523fbab1",
amount_usd=50000.0,
amount_token=17.86,
ltv=0.76,
liquidation_threshold=0.79,
price_usd=2800.0,
)
],
debt=[
DebtPosition(
asset="USDC",
asset_address="0xaf88d065e77c8cc2239327c5edb3a432268e5831",
amount_usd=48000.0,
amount_token=48000.0,
variable_rate=6.0,
)
],
total_collateral_usd=50000.0,
total_debt_usd=48000.0,
)
critical.compute_health()
analysis = LiquidationAnalysis(
wallet="0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18",
chains_analyzed=["ethereum", "arbitrum"],
positions=[safe, critical],
)
analysis.analyze()
assert safe.risk_tier == RiskTier.SAFE
assert critical.risk_tier == RiskTier.CRITICAL
# The large safe position ($200k coll, $10k debt) outweighs the
# small critical position ($50k coll, $48k debt) in the weighted
# average, so overall is SAFE - but cascade scenarios still show the risk
assert analysis.overall_risk_tier == RiskTier.SAFE
assert len(analysis.cascade_scenarios) > 0
total_liquidated = sum(s.total_liquidated_value_usd for s in analysis.cascade_scenarios)
assert total_liquidated > 0 # Cascade scenarios capture the critical position's risk
def test_to_dict_serialization_full(self):
"""Ensure the full analysis serializes to dict without errors."""
pos = self._make_sample_position(debt_usd=50000.0, coll_usd=100000.0)
analysis = LiquidationAnalysis(
wallet="0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18",
chains_analyzed=["ethereum"],
positions=[pos],
errors=["test error"],
warnings=["test warning"],
)
analysis.analyze()
d = analysis.to_dict()
assert isinstance(d, dict)
assert d["wallet"] == "0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18"
assert len(d["positions"]) == 1
assert "test error" in d["errors"]
@staticmethod
def _make_sample_position(debt_usd=50000.0, coll_usd=100000.0, chain="ethereum", protocol="Aave V3"):
coll = CollateralPosition(
asset="WETH",
asset_address="0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2",
amount_usd=coll_usd,
amount_token=coll_usd / 2800.0,
ltv=0.76,
liquidation_threshold=0.79,
price_usd=2800.0,
)
debt = DebtPosition(
asset="USDC",
asset_address="0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48",
amount_usd=debt_usd,
amount_token=debt_usd,
variable_rate=5.0,
)
pos = ProtocolPosition(
protocol=protocol,
chain=chain,
wallet="0x742d35Cc6634C0532925a3b844Bc9e7595f2bD18",
collateral=[coll],
debt=[debt],
total_collateral_usd=coll_usd,
total_debt_usd=debt_usd,
)
pos.compute_health()
return pos
Reference in a new issue View git blame Copy permalink