Files
sentiment-engine/MALKHUT/malkhut/tests/test_asset_classification.py

999 lines
36 KiB
Python
Raw Normal View History

"""
Tests for asset classification — multi-label, invariant characteristics, exhaustive.
Covers: per-asset profiles (all 13), multi-label overlap, cross-dimensional
consistency, enum coverage, query edge cases, orderbook fingerprint ordering,
predictive properties, and ScenarioFactory integration.
"""
import pytest
from malkhut.training.asset_classification import (
Sector, TokenRole, SupplyModel, ConsensusFamily, SmartContractCapability,
MarketCapTier, DerivativeAccess, VolatilityProfile, LiquidityProfile,
AssetProfile, ASSET_PROFILES,
get_asset_profile, list_assets,
get_assets_by_sector, get_assets_by_token_role, get_assets_by_supply,
get_assets_by_consensus, get_assets_by_market_cap, get_assets_by_volatility,
get_assets_by_liquidity, get_assets_by_derivatives, get_gas_tokens,
get_pov_assets, get_shortable_assets,
get_multi_sector_assets, get_multi_role_assets,
)
from malkhut.training.cma_trainer import ScenarioFactory
# ==============================================================================
# Profile existence and structural integrity
# ==============================================================================
class TestAssetProfileExists:
def test_minimum_assets(self):
assert len(ASSET_PROFILES) >= 13
def test_all_profiles_frozen(self):
for p in ASSET_PROFILES.values():
with pytest.raises(AttributeError):
p.symbol = "X"
def test_symbol_matches_key(self):
for symbol, profile in ASSET_PROFILES.items():
assert profile.symbol == symbol
def test_sectors_are_tuples(self):
for p in ASSET_PROFILES.values():
assert isinstance(p.sectors, tuple)
assert len(p.sectors) >= 1
def test_token_roles_are_tuples(self):
for p in ASSET_PROFILES.values():
assert isinstance(p.token_roles, tuple)
assert len(p.token_roles) >= 1
def test_no_duplicate_symbols(self):
assert len(ASSET_PROFILES) == len(set(ASSET_PROFILES.keys()))
def test_all_sectors_are_valid_sector_enum(self):
for p in ASSET_PROFILES.values():
for s in p.sectors:
assert isinstance(s, Sector)
def test_all_token_roles_are_valid_enum(self):
for p in ASSET_PROFILES.values():
for r in p.token_roles:
assert isinstance(r, TokenRole)
# ==============================================================================
# Enum coverage — every enum value used by at least one asset
# ==============================================================================
class TestEnumCoverage:
def test_all_primary_sectors_represented(self):
"""Every 'primary' sector (first in any asset's tuple) must be covered."""
used_primary = {p.sector for p in ASSET_PROFILES.values()}
assert Sector.CURRENCY in used_primary
assert Sector.LAYER1 in used_primary
assert Sector.LAYER2 in used_primary
assert Sector.DEFI in used_primary
assert Sector.ORACLE in used_primary
assert Sector.EXCHANGE in used_primary
assert Sector.MEME in used_primary
def test_all_token_roles_represented(self):
used = set()
for p in ASSET_PROFILES.values():
used.update(p.token_roles)
for r in TokenRole:
assert r in used, f"TokenRole.{r.value} not represented"
def test_all_supply_models_represented(self):
used = {p.supply_model for p in ASSET_PROFILES.values()}
for m in SupplyModel:
assert m in used, f"SupplyModel.{m.value} not represented"
def test_all_consensus_families_represented(self):
used = {p.consensus for p in ASSET_PROFILES.values()}
for c in ConsensusFamily:
assert c in used, f"ConsensusFamily.{c.value} not represented"
def test_all_smart_contract_capabilities_represented(self):
used = {p.smart_contracts for p in ASSET_PROFILES.values()}
for sc in SmartContractCapability:
assert sc in used, f"SmartContractCapability.{sc.value} not represented"
def test_all_common_market_cap_tiers_represented(self):
used = {p.market_cap_tier for p in ASSET_PROFILES.values()}
assert MarketCapTier.MEGA in used
assert MarketCapTier.LARGE in used
assert MarketCapTier.MID in used
assert MarketCapTier.SMALL in used
def test_all_common_volatility_profiles_represented(self):
used = {p.volatility_profile for p in ASSET_PROFILES.values()}
assert VolatilityProfile.LOW in used
assert VolatilityProfile.MEDIUM in used
assert VolatilityProfile.HIGH in used
def test_all_common_liquidity_profiles_represented(self):
used = {p.liquidity_profile for p in ASSET_PROFILES.values()}
assert LiquidityProfile.DEEP in used
assert LiquidityProfile.NORMAL in used
assert LiquidityProfile.THIN in used
def test_common_derivative_access_represented(self):
used = {p.derivative_access for p in ASSET_PROFILES.values()}
assert DerivativeAccess.PERPS_AND_OPTIONS in used
assert DerivativeAccess.PERPS_ONLY in used
# ==============================================================================
# Multi-label overlap
# ==============================================================================
class TestMultiLabelOverlap:
def test_eth_multiple_sectors(self):
p = get_asset_profile("ETHUSDT")
assert Sector.LAYER1 in p.sectors
assert Sector.DEFI in p.sectors
assert len(p.sectors) == 2
def test_eth_multiple_roles(self):
p = get_asset_profile("ETHUSDT")
assert TokenRole.GAS in p.token_roles
assert TokenRole.STORE_OF_VALUE in p.token_roles
assert TokenRole.GOVERNANCE in p.token_roles
assert len(p.token_roles) == 3
def test_bnb_multiple_sectors(self):
p = get_asset_profile("BNBUSDT")
assert Sector.EXCHANGE in p.sectors
assert Sector.LAYER1 in p.sectors
assert len(p.sectors) == 2
def test_bnb_multiple_roles(self):
p = get_asset_profile("BNBUSDT")
assert TokenRole.EXCHANGE_FEE in p.token_roles
assert TokenRole.GAS in p.token_roles
def test_doge_multi_sector(self):
p = get_asset_profile("DOGEUSDT")
assert Sector.MEME in p.sectors
assert Sector.CURRENCY in p.sectors
def test_doge_multi_role(self):
p = get_asset_profile("DOGEUSDT")
assert TokenRole.MEME in p.token_roles
assert TokenRole.GAS in p.token_roles
def test_uni_multi_role(self):
p = get_asset_profile("UNIUSDT")
assert TokenRole.GOVERNANCE in p.token_roles
assert TokenRole.UTILITY in p.token_roles
def test_dot_multi_role(self):
p = get_asset_profile("DOTUSDT")
assert TokenRole.GAS in p.token_roles
assert TokenRole.GOVERNANCE in p.token_roles
def test_avax_multi_sector(self):
p = get_asset_profile("AVAXUSDT")
assert Sector.LAYER1 in p.sectors
assert Sector.DEFI in p.sectors
def test_btc_single_sector(self):
p = get_asset_profile("BTCUSDT")
assert len(p.sectors) == 1
assert Sector.CURRENCY in p.sectors
def test_sol_single_sector_and_role(self):
p = get_asset_profile("SOLUSDT")
assert len(p.sectors) == 1
assert len(p.token_roles) == 1
def test_ada_single_sector_and_role(self):
p = get_asset_profile("ADAUSDT")
assert len(p.sectors) == 1
assert len(p.token_roles) == 1
def test_atom_single_sector_and_role(self):
p = get_asset_profile("ATOMUSDT")
assert len(p.sectors) == 1
assert len(p.token_roles) == 1
def test_multi_sector_assets_count(self):
multi = get_multi_sector_assets()
symbols = {p.symbol for p in multi}
assert "ETHUSDT" in symbols
assert "BNBUSDT" in symbols
assert "DOGEUSDT" in symbols
assert "AVAXUSDT" in symbols
assert len(multi) == 4
def test_multi_role_assets_count(self):
multi = get_multi_role_assets()
symbols = {p.symbol for p in multi}
assert "ETHUSDT" in symbols
assert "BNBUSDT" in symbols
assert "DOGEUSDT" in symbols
assert "UNIUSDT" in symbols
assert "DOTUSDT" in symbols
assert len(multi) == 5
# ==============================================================================
# Backward-compatible primary labels
# ==============================================================================
class TestBackwardCompatiblePrimaryLabels:
def test_eth_primary_sector(self):
assert get_asset_profile("ETHUSDT").sector == Sector.LAYER1
def test_bnb_primary_sector(self):
assert get_asset_profile("BNBUSDT").sector == Sector.EXCHANGE
def test_doge_primary_sector(self):
assert get_asset_profile("DOGEUSDT").sector == Sector.MEME
def test_eth_primary_role(self):
assert get_asset_profile("ETHUSDT").token_role == TokenRole.GAS
def test_bnb_primary_role(self):
assert get_asset_profile("BNBUSDT").token_role == TokenRole.EXCHANGE_FEE
def test_uni_primary_role(self):
assert get_asset_profile("UNIUSDT").token_role == TokenRole.GOVERNANCE
def test_primary_matches_first_in_sectors(self):
for p in ASSET_PROFILES.values():
assert p.sector == p.sectors[0]
def test_primary_matches_first_in_roles(self):
for p in ASSET_PROFILES.values():
assert p.token_role == p.token_roles[0]
# ==============================================================================
# Per-asset profiles — exhaustive for all 13
# ==============================================================================
class TestBTCProfile:
def test_sector(self):
assert get_asset_profile("BTCUSDT").sector == Sector.CURRENCY
def test_token_role(self):
assert get_asset_profile("BTCUSDT").token_role == TokenRole.STORE_OF_VALUE
def test_supply_model(self):
assert get_asset_profile("BTCUSDT").supply_model == SupplyModel.FIXED_CAP
def test_consensus(self):
assert get_asset_profile("BTCUSDT").consensus == ConsensusFamily.POW
def test_smart_contracts(self):
assert get_asset_profile("BTCUSDT").smart_contracts == SmartContractCapability.NONE
def test_market_cap(self):
assert get_asset_profile("BTCUSDT").market_cap_tier == MarketCapTier.MEGA
def test_volatility(self):
assert get_asset_profile("BTCUSDT").volatility_profile == VolatilityProfile.LOW
def test_liquidity(self):
assert get_asset_profile("BTCUSDT").liquidity_profile == LiquidityProfile.DEEP
def test_derivatives(self):
p = get_asset_profile("BTCUSDT")
assert p.derivative_access == DerivativeAccess.PERPS_AND_OPTIONS
assert p.has_options is True
assert p.has_funding is True
def test_tick_lot(self):
p = get_asset_profile("BTCUSDT")
assert p.tick_size == 0.1
assert p.lot_size == 0.001
assert p.price_decimals == 1
def test_fees(self):
p = get_asset_profile("BTCUSDT")
assert p.maker_fee_bps == -0.2
assert p.taker_fee_bps == 0.5
def test_orderbook(self):
p = get_asset_profile("BTCUSDT")
assert p.typical_spread_bps == 0.3
assert p.typical_depth_usd == 5_000_000
assert p.typical_daily_volume_usd == 30_000_000_000
class TestETHProfile:
def test_sectors(self):
p = get_asset_profile("ETHUSDT")
assert p.sector == Sector.LAYER1
assert Sector.DEFI in p.sectors
def test_roles(self):
p = get_asset_profile("ETHUSDT")
assert p.token_role == TokenRole.GAS
assert TokenRole.STORE_OF_VALUE in p.token_roles
assert TokenRole.GOVERNANCE in p.token_roles
def test_supply(self):
assert get_asset_profile("ETHUSDT").supply_model == SupplyModel.DISINFLATIONARY
def test_consensus(self):
assert get_asset_profile("ETHUSDT").consensus == ConsensusFamily.POS
def test_smart_contracts(self):
assert get_asset_profile("ETHUSDT").smart_contracts == SmartContractCapability.FULL
def test_tick_lot(self):
p = get_asset_profile("ETHUSDT")
assert p.tick_size == 0.01
assert p.lot_size == 0.001
assert p.price_decimals == 2
def test_options(self):
p = get_asset_profile("ETHUSDT")
assert p.has_options is True
assert p.has_funding is True
def test_is_gas(self):
assert get_asset_profile("ETHUSDT").is_gas_token
class TestSOLProfile:
def test_sector(self):
assert get_asset_profile("SOLUSDT").sector == Sector.LAYER1
def test_supply(self):
assert get_asset_profile("SOLUSDT").supply_model == SupplyModel.INFLATIONARY
def test_consensus(self):
assert get_asset_profile("SOLUSDT").consensus == ConsensusFamily.POS
def test_volatility(self):
assert get_asset_profile("SOLUSDT").volatility_profile == VolatilityProfile.HIGH
def test_no_options(self):
assert get_asset_profile("SOLUSDT").has_options is False
def test_lot_size(self):
assert get_asset_profile("SOLUSDT").lot_size == 0.01
class TestDOGEProfile:
def test_sectors(self):
p = get_asset_profile("DOGEUSDT")
assert Sector.MEME in p.sectors
assert Sector.CURRENCY in p.sectors
def test_roles(self):
p = get_asset_profile("DOGEUSDT")
assert TokenRole.MEME in p.token_roles
assert TokenRole.GAS in p.token_roles
def test_pow_forced_selling(self):
assert get_asset_profile("DOGEUSDT").supply_pressure == "forced"
def test_no_smart_contracts(self):
assert get_asset_profile("DOGEUSDT").smart_contracts == SmartContractCapability.NONE
def test_five_decimal_tick(self):
p = get_asset_profile("DOGEUSDT")
assert p.tick_size == 0.00001
assert p.price_decimals == 5
def test_one_unit_lot(self):
assert get_asset_profile("DOGEUSDT").lot_size == 1.0
class TestADAProfile:
def test_sector(self):
assert get_asset_profile("ADAUSDT").sector == Sector.LAYER1
def test_consensus(self):
assert get_asset_profile("ADAUSDT").consensus == ConsensusFamily.DPOS
def test_four_decimal_tick(self):
p = get_asset_profile("ADAUSDT")
assert p.tick_size == 0.0001
assert p.price_decimals == 4
class TestAVAXProfile:
def test_sectors(self):
p = get_asset_profile("AVAXUSDT")
assert Sector.LAYER1 in p.sectors
assert Sector.DEFI in p.sectors
def test_supply(self):
assert get_asset_profile("AVAXUSDT").supply_model == SupplyModel.INFLATIONARY
class TestUNIProfile:
def test_sector(self):
assert get_asset_profile("UNIUSDT").sector == Sector.DEFI
def test_roles(self):
p = get_asset_profile("UNIUSDT")
assert TokenRole.GOVERNANCE in p.token_roles
assert TokenRole.UTILITY in p.token_roles
def test_small_cap(self):
assert get_asset_profile("UNIUSDT").market_cap_tier == MarketCapTier.SMALL
class TestLINKProfile:
def test_sector(self):
assert get_asset_profile("LINKUSDT").sector == Sector.ORACLE
def test_role(self):
assert get_asset_profile("LINKUSDT").token_role == TokenRole.UTILITY
def test_no_gas_role(self):
assert get_asset_profile("LINKUSDT").is_gas_token is False
class TestBNBProfile:
def test_sectors(self):
p = get_asset_profile("BNBUSDT")
assert Sector.EXCHANGE in p.sectors
assert Sector.LAYER1 in p.sectors
def test_roles(self):
p = get_asset_profile("BNBUSDT")
assert TokenRole.EXCHANGE_FEE in p.token_roles
assert TokenRole.GAS in p.token_roles
def test_burn_mechanism(self):
assert get_asset_profile("BNBUSDT").supply_model == SupplyModel.BURN_MECHANISM
def test_partial_smart_contracts(self):
assert get_asset_profile("BNBUSDT").smart_contracts == SmartContractCapability.PARTIAL
def test_lower_fees(self):
p = get_asset_profile("BNBUSDT")
assert p.maker_fee_bps == -0.1
assert p.taker_fee_bps == 0.4
def test_is_gas(self):
assert get_asset_profile("BNBUSDT").is_gas_token
class TestMATICProfile:
def test_sector(self):
assert get_asset_profile("MATICUSDT").sector == Sector.LAYER2
def test_four_decimal_tick(self):
p = get_asset_profile("MATICUSDT")
assert p.tick_size == 0.0001
assert p.price_decimals == 4
class TestAAVEProfile:
def test_sector(self):
assert get_asset_profile("AAVEUSDT").sector == Sector.DEFI
def test_role(self):
assert get_asset_profile("AAVEUSDT").token_role == TokenRole.GOVERNANCE
def test_fixed_cap(self):
assert get_asset_profile("AAVEUSDT").supply_model == SupplyModel.FIXED_CAP
class TestDOTProfile:
def test_sector(self):
assert get_asset_profile("DOTUSDT").sector == Sector.LAYER1
def test_roles(self):
p = get_asset_profile("DOTUSDT")
assert TokenRole.GAS in p.token_roles
assert TokenRole.GOVERNANCE in p.token_roles
def test_three_decimal_tick(self):
p = get_asset_profile("DOTUSDT")
assert p.tick_size == 0.001
assert p.price_decimals == 3
class TestATOMProfile:
def test_sector(self):
assert get_asset_profile("ATOMUSDT").sector == Sector.LAYER1
def test_supply(self):
assert get_asset_profile("ATOMUSDT").supply_model == SupplyModel.INFLATIONARY
def test_consensus(self):
assert get_asset_profile("ATOMUSDT").consensus == ConsensusFamily.POS
# ==============================================================================
# Cross-dimensional consistency checks
# ==============================================================================
class TestCrossDimensionalConsistency:
def test_pow_assets_are_btc_doge_only(self):
pows = get_assets_by_consensus(ConsensusFamily.POW)
symbols = {p.symbol for p in pows}
assert symbols == {"BTCUSDT", "DOGEUSDT"}
def test_defi_sector_has_full_smart_contracts(self):
for p in get_assets_by_sector(Sector.DEFI):
assert p.smart_contracts == SmartContractCapability.FULL, \
f"{p.symbol} is DEFI but not FULL smart contracts"
def test_l1_native_assets_have_full_smart_contracts(self):
l1_native = [p for p in get_assets_by_sector(Sector.LAYER1)
if Sector.LAYER1 in p.sectors and Sector.EXCHANGE not in p.sectors]
for p in l1_native:
assert p.smart_contracts == SmartContractCapability.FULL, \
f"{p.symbol} is LAYER1 but not FULL smart contracts"
def test_mega_cap_is_btc_only(self):
mega = get_assets_by_market_cap(MarketCapTier.MEGA)
assert len(mega) == 1
assert mega[0].symbol == "BTCUSDT"
def test_btc_only_has_options(self):
with_options = [p for p in ASSET_PROFILES.values() if p.has_options]
assert len(with_options) == 2
symbols = {p.symbol for p in with_options}
assert symbols == {"BTCUSDT", "ETHUSDT"}
def test_only_two_perps_and_options(self):
pao = get_assets_by_derivatives(DerivativeAccess.PERPS_AND_OPTIONS)
assert len(pao) == 2
symbols = {p.symbol for p in pao}
assert symbols == {"BTCUSDT", "ETHUSDT"}
def test_all_have_funding(self):
for p in ASSET_PROFILES.values():
assert p.has_funding is True
def test_only_bnb_has_burn_mechanism(self):
burned = get_assets_by_supply(SupplyModel.BURN_MECHANISM)
assert len(burned) == 1
assert burned[0].symbol == "BNBUSDT"
def test_only_bnb_has_partial_smart_contracts(self):
partial = [p for p in ASSET_PROFILES.values()
if p.smart_contracts == SmartContractCapability.PARTIAL]
assert len(partial) == 1
assert partial[0].symbol == "BNBUSDT"
def test_no_smart_contracts_are_btc_doge(self):
none_sc = [p for p in ASSET_PROFILES.values()
if p.smart_contracts == SmartContractCapability.NONE]
symbols = {p.symbol for p in none_sc}
assert symbols == {"BTCUSDT", "DOGEUSDT"}
def test_only_btc_is_mega(self):
mega = get_assets_by_market_cap(MarketCapTier.MEGA)
assert len(mega) == 1
def test_large_cap_are_eth_bnb(self):
large = get_assets_by_market_cap(MarketCapTier.LARGE)
symbols = {p.symbol for p in large}
assert symbols == {"ETHUSDT", "BNBUSDT"}
# ==============================================================================
# Predictive properties
# ==============================================================================
class TestSupplyPressure:
def test_pow_forced(self):
for symbol in ("BTCUSDT", "DOGEUSDT"):
assert get_asset_profile(symbol).supply_pressure == "forced"
def test_pos_optional(self):
pos_symbols = ["ETHUSDT", "SOLUSDT", "ADAUSDT", "AVAXUSDT",
"UNIUSDT", "LINKUSDT", "AAVEUSDT", "DOTUSDT", "ATOMUSDT"]
for symbol in pos_symbols:
assert get_asset_profile(symbol).supply_pressure == "optional"
def test_dpos_optional(self):
for symbol in ("ADAUSDT", "BNBUSDT", "DOTUSDT"):
assert get_asset_profile(symbol).supply_pressure == "optional"
def test_pov_only_pow(self):
pov = get_pov_assets()
assert len(pov) == 2
for p in pov:
assert p.consensus == ConsensusFamily.POW
class TestDemandElasticity:
def test_gas_inelastic(self):
for p in get_gas_tokens():
assert p.demand_elasticity == "inelastic"
def test_store_of_value_inelastic(self):
sov = get_assets_by_token_role(TokenRole.STORE_OF_VALUE)
for p in sov:
assert p.demand_elasticity == "inelastic"
def test_elastic_tokens(self):
for p in ASSET_PROFILES.values():
if p.demand_elasticity == "elastic":
assert TokenRole.GAS not in p.token_roles
assert TokenRole.STORE_OF_VALUE not in p.token_roles
def test_elastic_count(self):
elastic = [p for p in ASSET_PROFILES.values()
if p.demand_elasticity == "elastic"]
assert len(elastic) >= 3
def test_inelastic_count(self):
inelastic = [p for p in ASSET_PROFILES.values()
if p.demand_elasticity == "inelastic"]
assert len(inelastic) >= 8
def test_all_assets_are_one_or_other(self):
for p in ASSET_PROFILES.values():
assert p.demand_elasticity in ("inelastic", "elastic")
class TestIsGasToken:
def test_gas_tokens(self):
gas = get_gas_tokens()
symbols = {p.symbol for p in gas}
assert "ETHUSDT" in symbols
assert "SOLUSDT" in symbols
assert "BNBUSDT" in symbols
assert "DOGEUSDT" in symbols
assert "MATICUSDT" in symbols
assert "DOTUSDT" in symbols
def test_non_gas_tokens(self):
assert get_asset_profile("BTCUSDT").is_gas_token is False
assert get_asset_profile("UNIUSDT").is_gas_token is False
assert get_asset_profile("AAVEUSDT").is_gas_token is False
assert get_asset_profile("LINKUSDT").is_gas_token is False
class TestIsPureCurrency:
def test_currency_assets(self):
assert get_asset_profile("BTCUSDT").is_pure_currency
assert get_asset_profile("DOGEUSDT").is_pure_currency
def test_non_currency_assets(self):
assert get_asset_profile("ETHUSDT").is_pure_currency is False
assert get_asset_profile("SOLUSDT").is_pure_currency is False
assert get_asset_profile("UNIUSDT").is_pure_currency is False
class TestCanBeShorted:
def test_all_shortable(self):
for p in ASSET_PROFILES.values():
assert p.can_be_shorted
def test_shortable_matches_derivatives(self):
for p in ASSET_PROFILES.values():
if p.derivative_access == DerivativeAccess.NONE:
assert not p.can_be_shorted
else:
assert p.can_be_shorted
# ==============================================================================
# Orderbook fingerprint ordering
# ==============================================================================
class TestOrderbookFingerprintOrdering:
def test_spread_increases_with_risk(self):
btc_spread = get_asset_profile("BTCUSDT").typical_spread_bps
eth_spread = get_asset_profile("ETHUSDT").typical_spread_bps
sol_spread = get_asset_profile("SOLUSDT").typical_spread_bps
doge_spread = get_asset_profile("DOGEUSDT").typical_spread_bps
assert btc_spread < eth_spread < sol_spread < doge_spread
def test_depth_decreases_with_cap(self):
btc_depth = get_asset_profile("BTCUSDT").typical_depth_usd
eth_depth = get_asset_profile("ETHUSDT").typical_depth_usd
sol_depth = get_asset_profile("SOLUSDT").typical_depth_usd
aave_depth = get_asset_profile("AAVEUSDT").typical_depth_usd
assert btc_depth > eth_depth > sol_depth > aave_depth
def test_volume_decreases_with_cap(self):
btc_vol = get_asset_profile("BTCUSDT").typical_daily_volume_usd
eth_vol = get_asset_profile("ETHUSDT").typical_daily_volume_usd
assert btc_vol > eth_vol
def test_deep_has_highest_depth(self):
for p in get_assets_by_liquidity(LiquidityProfile.DEEP):
assert p.typical_depth_usd >= 2_000_000
def test_thin_has_lowest_depth(self):
for p in get_assets_by_liquidity(LiquidityProfile.THIN):
assert p.typical_depth_usd <= 700_000
def test_spread_nonzero(self):
for p in ASSET_PROFILES.values():
assert p.typical_spread_bps > 0
def test_volume_nonzero(self):
for p in ASSET_PROFILES.values():
assert p.typical_daily_volume_usd > 0
# ==============================================================================
# Profile field validation
# ==============================================================================
class TestProfileFieldValidation:
def test_tick_size_positive(self):
for p in ASSET_PROFILES.values():
assert p.tick_size > 0
def test_lot_size_positive(self):
for p in ASSET_PROFILES.values():
assert p.lot_size > 0
def test_price_decimals_non_negative(self):
for p in ASSET_PROFILES.values():
assert p.price_decimals >= 0
def test_fee_bps_reasonable(self):
for p in ASSET_PROFILES.values():
assert -1.0 <= p.maker_fee_bps <= 1.0
assert 0.0 <= p.taker_fee_bps <= 2.0
assert p.taker_fee_bps >= p.maker_fee_bps
def test_depth_usd_positive(self):
for p in ASSET_PROFILES.values():
assert p.typical_depth_usd > 0
def test_volume_usd_positive(self):
for p in ASSET_PROFILES.values():
assert p.typical_daily_volume_usd > 0
def test_volume_exceeds_depth(self):
for p in ASSET_PROFILES.values():
assert p.typical_daily_volume_usd > p.typical_depth_usd
# ==============================================================================
# Query functions — exhaustive with edge cases
# ==============================================================================
class TestQueryFunctions:
def test_get_asset_profile_hit(self):
assert get_asset_profile("SOLUSDT") is not None
def test_get_asset_profile_miss(self):
assert get_asset_profile("FAKEUSDT") is None
assert get_asset_profile("") is None
assert get_asset_profile("BTCUSDTx") is None
def test_list_assets_count(self):
assert len(list_assets()) == 13
def test_list_assets_all_present(self):
expected = {"BTCUSDT", "ETHUSDT", "SOLUSDT", "DOGEUSDT", "ADAUSDT",
"AVAXUSDT", "UNIUSDT", "LINKUSDT", "BNBUSDT", "MATICUSDT",
"AAVEUSDT", "DOTUSDT", "ATOMUSDT"}
assert set(list_assets()) == expected
def test_by_sector_layer1(self):
l1 = get_assets_by_sector(Sector.LAYER1)
symbols = {p.symbol for p in l1}
assert "ETHUSDT" in symbols
assert "SOLUSDT" in symbols
assert "BNBUSDT" in symbols
def test_by_sector_defi(self):
defi = get_assets_by_sector(Sector.DEFI)
symbols = {p.symbol for p in defi}
assert "ETHUSDT" in symbols
assert "UNIUSDT" in symbols
assert "AAVEUSDT" in symbols
assert "AVAXUSDT" in symbols
def test_by_sector_exchange(self):
ex = get_assets_by_sector(Sector.EXCHANGE)
assert len(ex) == 1
assert ex[0].symbol == "BNBUSDT"
def test_by_sector_oracle(self):
orc = get_assets_by_sector(Sector.ORACLE)
assert len(orc) == 1
assert orc[0].symbol == "LINKUSDT"
def test_by_sector_layer2(self):
l2 = get_assets_by_sector(Sector.LAYER2)
assert len(l2) == 1
assert l2[0].symbol == "MATICUSDT"
def test_by_sector_meme(self):
meme = get_assets_by_sector(Sector.MEME)
symbols = {p.symbol for p in meme}
assert "DOGEUSDT" in symbols
def test_by_sector_storage_empty(self):
assert get_assets_by_sector(Sector.STORAGE) == []
def test_by_sector_privacy_empty(self):
assert get_assets_by_sector(Sector.PRIVACY) == []
def test_by_sector_gaming_empty(self):
assert get_assets_by_sector(Sector.GAMING_NFT) == []
def test_by_token_role_gas(self):
gas = get_assets_by_token_role(TokenRole.GAS)
assert len(gas) >= 7
def test_by_token_role_store_of_value(self):
sov = get_assets_by_token_role(TokenRole.STORE_OF_VALUE)
symbols = {p.symbol for p in sov}
assert "BTCUSDT" in symbols
assert "ETHUSDT" in symbols
def test_by_token_role_governance(self):
gov = get_assets_by_token_role(TokenRole.GOVERNANCE)
symbols = {p.symbol for p in gov}
assert "UNIUSDT" in symbols
assert "AAVEUSDT" in symbols
assert "ETHUSDT" in symbols
assert "DOTUSDT" in symbols
def test_by_token_role_utility(self):
util = get_assets_by_token_role(TokenRole.UTILITY)
symbols = {p.symbol for p in util}
assert "LINKUSDT" in symbols
assert "UNIUSDT" in symbols
def test_by_token_role_exchange_fee(self):
ef = get_assets_by_token_role(TokenRole.EXCHANGE_FEE)
assert len(ef) == 1
assert ef[0].symbol == "BNBUSDT"
def test_by_token_role_meme(self):
meme = get_assets_by_token_role(TokenRole.MEME)
symbols = {p.symbol for p in meme}
assert "DOGEUSDT" in symbols
def test_by_supply_fixed_cap(self):
fixed = get_assets_by_supply(SupplyModel.FIXED_CAP)
symbols = {p.symbol for p in fixed}
assert "BTCUSDT" in symbols
assert "AAVEUSDT" in symbols
def test_by_supply_disinflationary(self):
dis = get_assets_by_supply(SupplyModel.DISINFLATIONARY)
assert len(dis) == 1
assert dis[0].symbol == "ETHUSDT"
def test_by_supply_burn(self):
burn = get_assets_by_supply(SupplyModel.BURN_MECHANISM)
assert len(burn) == 1
assert burn[0].symbol == "BNBUSDT"
def test_by_consensus_pow(self):
pows = get_assets_by_consensus(ConsensusFamily.POW)
symbols = {p.symbol for p in pows}
assert symbols == {"BTCUSDT", "DOGEUSDT"}
def test_by_consensus_dpos(self):
dpos = get_assets_by_consensus(ConsensusFamily.DPOS)
symbols = {p.symbol for p in dpos}
assert "ADAUSDT" in symbols
assert "BNBUSDT" in symbols
assert "DOTUSDT" in symbols
def test_by_market_cap_mega(self):
mega = get_assets_by_market_cap(MarketCapTier.MEGA)
assert len(mega) == 1
def test_by_market_cap_large(self):
large = get_assets_by_market_cap(MarketCapTier.LARGE)
symbols = {p.symbol for p in large}
assert symbols == {"ETHUSDT", "BNBUSDT"}
def test_by_market_cap_small(self):
small = get_assets_by_market_cap(MarketCapTier.SMALL)
assert len(small) >= 4
def test_by_volatility_low(self):
low = get_assets_by_volatility(VolatilityProfile.LOW)
assert len(low) == 1
assert low[0].symbol == "BTCUSDT"
def test_by_volatility_medium(self):
med = get_assets_by_volatility(VolatilityProfile.MEDIUM)
assert len(med) >= 3
def test_by_volatility_high(self):
high = get_assets_by_volatility(VolatilityProfile.HIGH)
assert len(high) >= 5
def test_by_volatility_extreme_empty(self):
assert get_assets_by_volatility(VolatilityProfile.EXTREME) == []
def test_by_liquidity_deep(self):
deep = get_assets_by_liquidity(LiquidityProfile.DEEP)
symbols = {p.symbol for p in deep}
assert "BTCUSDT" in symbols
assert "ETHUSDT" in symbols
assert "BNBUSDT" in symbols
def test_by_liquidity_normal(self):
norm = get_assets_by_liquidity(LiquidityProfile.NORMAL)
assert len(norm) >= 4
def test_by_liquidity_thin(self):
thin = get_assets_by_liquidity(LiquidityProfile.THIN)
assert len(thin) >= 4
def test_by_liquidity_illiquid_empty(self):
assert get_assets_by_liquidity(LiquidityProfile.ILLIQUID) == []
def test_by_derivatives_perps_options(self):
pao = get_assets_by_derivatives(DerivativeAccess.PERPS_AND_OPTIONS)
assert len(pao) == 2
def test_by_derivatives_perps_only(self):
po = get_assets_by_derivatives(DerivativeAccess.PERPS_ONLY)
assert len(po) == 11
def test_by_derivatives_none_empty(self):
assert get_assets_by_derivatives(DerivativeAccess.NONE) == []
def test_shortable_count(self):
assert len(get_shortable_assets()) == 13
def test_get_gas_tokens_count(self):
assert len(get_gas_tokens()) == 9
def test_get_pov_assets_count(self):
assert len(get_pov_assets()) == 2
# ==============================================================================
# ScenarioFactory integration
# ==============================================================================
class TestMultiAssetScenarios:
def test_multi_asset_suite_3_assets(self):
factory = ScenarioFactory()
suite = factory.build_suite(
symbols=("BTCUSDT", "ETHUSDT", "SOLUSDT"),
steps_per_scenario=5,
)
assert len(suite) >= 90
def test_multi_asset_uses_profiles(self):
factory = ScenarioFactory()
suite = factory.build_suite(symbols=("ETHUSDT",), steps_per_scenario=5)
for s in suite:
assert s.initial_state.venue.symbol == "ETHUSDT"
assert s.initial_state.venue.tick_size == 0.01
def test_all_assets_have_scenarios(self):
factory = ScenarioFactory()
assets = list_assets()
suite = factory.build_suite(symbols=tuple(assets), steps_per_scenario=3)
symbols_in_suite = set(s.initial_state.venue.symbol for s in suite)
for asset in assets:
assert asset in symbols_in_suite
def test_btc_uses_btc_tick(self):
factory = ScenarioFactory()
suite = factory.build_suite(symbols=("BTCUSDT",), steps_per_scenario=3)
for s in suite:
assert s.initial_state.venue.tick_size == 0.1
def test_doge_uses_doge_tick(self):
factory = ScenarioFactory()
suite = factory.build_suite(symbols=("DOGEUSDT",), steps_per_scenario=3)
for s in suite:
assert s.initial_state.venue.tick_size == 0.00001
def test_single_asset_scenario_count(self):
factory = ScenarioFactory()
suite = factory.build_suite(symbols=("BTCUSDT",), steps_per_scenario=5)
assert len(suite) >= 30
def test_steps_per_scenario_honored(self):
factory = ScenarioFactory()
suite = factory.build_suite(symbols=("ETHUSDT",), steps_per_scenario=10)
for s in suite:
assert s.max_steps >= 10