diff --git a/tests/validator/test_dynamic_emissions_advanced.py b/tests/validator/test_dynamic_emissions_advanced.py new file mode 100644 index 00000000..72740e92 --- /dev/null +++ b/tests/validator/test_dynamic_emissions_advanced.py @@ -0,0 +1,344 @@ +# The MIT License (MIT) +# Copyright © 2025 Entrius + +""" +Comprehensive tests for dynamic emissions module. + +Tests _exponential_unlock_scalar(), get_network_totals(), and +apply_dynamic_emissions_using_network_contributions() with focus on +boundary values, edge cases, and mathematical correctness. + +Run tests: + pytest tests/validator/test_dynamic_emissions_advanced.py -v +""" + +import numpy as np +import pytest + +from gittensor.classes import MinerEvaluation +from gittensor.constants import ( + RECYCLE_UID, + TOKEN_SCORE_MAX_RECYCLE, + TOKEN_SCORE_RECYCLE_DECAY_RATE, + UNIQUE_REPOS_MAX_RECYCLE, + UNIQUE_REPOS_RECYCLE_DECAY_RATE, +) +from gittensor.validator.configurations.tier_config import Tier, TierStats +from gittensor.validator.evaluation.dynamic_emissions import ( + _exponential_unlock_scalar, + apply_dynamic_emissions_using_network_contributions, + get_network_totals, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_eval( + uid: int, + current_tier: Tier | None = Tier.BRONZE, + total_token_score: float = 0.0, + unique_repos: set[str] | None = None, +) -> MinerEvaluation: + """Create a MinerEvaluation with the fields relevant to dynamic emissions.""" + ev = MinerEvaluation(uid=uid, hotkey=f"hotkey_{uid}") + ev.current_tier = current_tier + ev.total_token_score = total_token_score + ev.unique_repos_contributed_to = unique_repos or set() + return ev + + +# =========================================================================== +# _exponential_unlock_scalar +# =========================================================================== + + +class TestExponentialUnlockScalar: + """Tests for the exponential unlock curve: min(1.0, (1-max_recycle) + max_recycle*(1-exp(-decay*v))).""" + + # --- Boundary: value = 0 --- + + def test_zero_value_returns_one_minus_max_recycle(self): + """At value=0 the exponential term is 0, so scalar = 1 - max_recycle.""" + result = _exponential_unlock_scalar(0.0, max_recycle=0.8, decay_rate=0.005) + assert result == pytest.approx(0.2) + + def test_zero_value_with_full_recycle(self): + """max_recycle=1.0, value=0 → scalar=0 (all emissions recycled).""" + assert _exponential_unlock_scalar(0.0, 1.0, 0.01) == pytest.approx(0.0) + + def test_zero_value_with_no_recycle(self): + """max_recycle=0.0, value=0 → scalar=1.0 (nothing to recycle).""" + assert _exponential_unlock_scalar(0.0, 0.0, 0.01) == pytest.approx(1.0) + + # --- Boundary: very large value --- + + def test_very_large_value_saturates_to_one(self): + """Extremely large value drives exp term to 0, scalar → 1.0.""" + result = _exponential_unlock_scalar(1e12, 0.8, 0.005) + assert result == pytest.approx(1.0) + + # --- Cap at 1.0 --- + + def test_cap_at_one(self): + """Result is capped at 1.0 even if the formula would exceed it (shouldn't normally, but verify min()).""" + # With max_recycle < 1 and positive value the formula approaches 1 from below. + result = _exponential_unlock_scalar(1e6, 0.5, 10.0) + assert result == 1.0 + + # --- Monotonicity --- + + def test_monotonically_increasing(self): + """Scalar should increase as value increases.""" + values = [0, 10, 50, 200, 1000, 10000] + scalars = [_exponential_unlock_scalar(v, 0.8, 0.005) for v in values] + for i in range(len(scalars) - 1): + assert scalars[i] <= scalars[i + 1], f"Not monotonic at index {i}" + + # --- Specific math verification --- + + def test_known_value_manual_calculation(self): + """Verify against hand-calculated result for a specific input.""" + # value=100, max_recycle=0.8, decay_rate=0.005 + # expected = min(1.0, 0.2 + 0.8 * (1 - exp(-0.5))) + expected = min(1.0, 0.2 + 0.8 * (1 - np.exp(-0.5))) + result = _exponential_unlock_scalar(100.0, 0.8, 0.005) + assert result == pytest.approx(expected) + + def test_with_production_constants_repos(self): + """Verify scalar using production UNIQUE_REPOS constants at value=50.""" + expected = min( + 1.0, + (1 - UNIQUE_REPOS_MAX_RECYCLE) + + UNIQUE_REPOS_MAX_RECYCLE * (1 - np.exp(-UNIQUE_REPOS_RECYCLE_DECAY_RATE * 50)), + ) + assert _exponential_unlock_scalar(50, UNIQUE_REPOS_MAX_RECYCLE, UNIQUE_REPOS_RECYCLE_DECAY_RATE) == pytest.approx(expected) + + def test_with_production_constants_token_score(self): + """Verify scalar using production TOKEN_SCORE constants at value=100000.""" + expected = min( + 1.0, + (1 - TOKEN_SCORE_MAX_RECYCLE) + + TOKEN_SCORE_MAX_RECYCLE * (1 - np.exp(-TOKEN_SCORE_RECYCLE_DECAY_RATE * 100000)), + ) + assert _exponential_unlock_scalar(100000, TOKEN_SCORE_MAX_RECYCLE, TOKEN_SCORE_RECYCLE_DECAY_RATE) == pytest.approx(expected) + + # --- Edge: decay_rate = 0 --- + + def test_zero_decay_rate_always_returns_base(self): + """With decay_rate=0 the exponential term is always 0 regardless of value.""" + assert _exponential_unlock_scalar(999999, 0.8, 0.0) == pytest.approx(0.2) + + +# =========================================================================== +# get_network_totals +# =========================================================================== + + +class TestGetNetworkTotals: + """Tests for get_network_totals — counting only tiered miners.""" + + def test_empty_evaluations(self): + """Empty dict returns zeros.""" + repos, score = get_network_totals({}) + assert repos == 0 + assert score == 0.0 + + def test_only_untiered_miners_excluded(self): + """Miners with current_tier=None are excluded from totals.""" + ev = _make_eval(1, current_tier=None, total_token_score=500.0, unique_repos={"repo_a"}) + repos, score = get_network_totals({1: ev}) + assert repos == 0 + assert score == 0.0 + + def test_tiered_miner_counted(self): + """A tiered miner's repos and token score are counted.""" + ev = _make_eval(1, Tier.BRONZE, total_token_score=100.0, unique_repos={"repo_a", "repo_b"}) + repos, score = get_network_totals({1: ev}) + assert repos == 2 + assert score == pytest.approx(100.0) + + def test_mixed_tiered_and_untiered(self): + """Only tiered miners contribute to totals.""" + evals = { + 1: _make_eval(1, Tier.GOLD, 300.0, {"r1", "r2"}), + 2: _make_eval(2, None, 999.0, {"r3"}), + 3: _make_eval(3, Tier.SILVER, 200.0, {"r2", "r4"}), + } + repos, score = get_network_totals(evals) + # unique repos: r1, r2, r4 (r3 excluded because miner 2 is untiered) + assert repos == 3 + assert score == pytest.approx(500.0) + + def test_duplicate_repos_across_miners_counted_once(self): + """Same repo contributed by multiple tiered miners is counted once.""" + evals = { + 1: _make_eval(1, Tier.BRONZE, 10.0, {"shared_repo"}), + 2: _make_eval(2, Tier.BRONZE, 20.0, {"shared_repo"}), + } + repos, score = get_network_totals(evals) + assert repos == 1 + assert score == pytest.approx(30.0) + + def test_miner_with_no_repos(self): + """Tiered miner with empty repos still contributes token score.""" + ev = _make_eval(1, Tier.BRONZE, total_token_score=50.0, unique_repos=set()) + repos, score = get_network_totals({1: ev}) + assert repos == 0 + assert score == pytest.approx(50.0) + + def test_all_tiers_counted(self): + """Bronze, Silver, and Gold miners all contribute.""" + evals = { + 1: _make_eval(1, Tier.BRONZE, 10.0, {"a"}), + 2: _make_eval(2, Tier.SILVER, 20.0, {"b"}), + 3: _make_eval(3, Tier.GOLD, 30.0, {"c"}), + } + repos, score = get_network_totals(evals) + assert repos == 3 + assert score == pytest.approx(60.0) + + +# =========================================================================== +# apply_dynamic_emissions_using_network_contributions +# =========================================================================== + + +class TestApplyDynamicEmissions: + """Tests for apply_dynamic_emissions_using_network_contributions.""" + + def test_empty_rewards_returns_empty(self): + """Empty normalized_rewards returns empty dict.""" + result = apply_dynamic_emissions_using_network_contributions({}, {}) + assert result == {} + + def test_recycle_uid_gets_recycled_amount(self): + """Recycled emissions are allocated to RECYCLE_UID.""" + rewards = {1: 0.5, 2: 0.5} + evals = {} # No tiered miners → max recycle + result = apply_dynamic_emissions_using_network_contributions(rewards, evals) + assert RECYCLE_UID in result + assert result[RECYCLE_UID] > 0 + + def test_total_preserved_with_recycle(self): + """Sum of scaled rewards + recycle should equal original total.""" + rewards = {1: 0.4, 2: 0.6} + evals = { + 1: _make_eval(1, Tier.BRONZE, 100.0, {"r1"}), + } + result = apply_dynamic_emissions_using_network_contributions(rewards, evals) + original_total = 1.0 + new_total = sum(result.values()) + assert new_total == pytest.approx(original_total) + + def test_zero_network_contributions_maximum_recycle(self): + """With no tiered miners, scalar is at minimum → maximum recycling.""" + rewards = {1: 1.0} + evals = {} # no miners + result = apply_dynamic_emissions_using_network_contributions(rewards, evals) + + # Both scalars at value=0: scalar = 1 - max_recycle = 0.2 + # final_scalar = (0.2 + 0.2) / 2 = 0.2 + expected_scalar = 0.2 + assert result[1] == pytest.approx(1.0 * expected_scalar) + assert result[RECYCLE_UID] == pytest.approx(1.0 * (1 - expected_scalar)) + + def test_high_contributions_low_recycle(self): + """High network contributions drive scalar toward 1.0, minimizing recycling.""" + rewards = {1: 1.0} + evals = { + 1: _make_eval(1, Tier.GOLD, total_token_score=1e8, unique_repos={f"r{i}" for i in range(5000)}), + } + result = apply_dynamic_emissions_using_network_contributions(rewards, evals) + + # Scalar should be very close to 1.0 + assert result[1] == pytest.approx(1.0, abs=0.01) + assert result[RECYCLE_UID] == pytest.approx(0.0, abs=0.01) + + def test_scaling_preserves_relative_proportions(self): + """Scaling should preserve relative proportions between miners.""" + rewards = {1: 0.7, 2: 0.3} + evals = {1: _make_eval(1, Tier.BRONZE, 50.0, {"r1"})} + result = apply_dynamic_emissions_using_network_contributions(rewards, evals) + + # Exclude RECYCLE_UID for proportion check + miner_rewards = {k: v for k, v in result.items() if k != RECYCLE_UID} + total_miner = sum(miner_rewards.values()) + if total_miner > 0: + assert miner_rewards[1] / total_miner == pytest.approx(0.7) + assert miner_rewards[2] / total_miner == pytest.approx(0.3) + + def test_single_miner_single_reward(self): + """Single miner: reward is scaled, remainder goes to recycle.""" + rewards = {5: 1.0} + evals = {5: _make_eval(5, Tier.SILVER, 200.0, {"a", "b"})} + result = apply_dynamic_emissions_using_network_contributions(rewards, evals) + + assert 5 in result + assert RECYCLE_UID in result + assert result[5] + result[RECYCLE_UID] == pytest.approx(1.0) + assert result[5] < 1.0 # Some must be recycled at moderate contribution levels + + def test_recycle_uid_in_original_rewards_accumulated(self): + """If RECYCLE_UID already has a reward, recycled amount is added to it.""" + rewards = {RECYCLE_UID: 0.1, 1: 0.9} + evals = {} + result = apply_dynamic_emissions_using_network_contributions(rewards, evals) + + # RECYCLE_UID gets its scaled portion + all recycled emissions + # scalar ≈ 0.2 → recycled = 1.0 * 0.8 = 0.8 + # RECYCLE_UID scaled = 0.1 * 0.2 = 0.02, then + 0.8 = 0.82 + expected_scalar = 0.2 + expected_recycle = 0.1 * expected_scalar + 1.0 * (1 - expected_scalar) + assert result[RECYCLE_UID] == pytest.approx(expected_recycle) + + def test_all_zero_rewards(self): + """All zero rewards: recycled amount is 0, but recycle gets 1.0 floor.""" + rewards = {1: 0.0, 2: 0.0} + evals = {} + result = apply_dynamic_emissions_using_network_contributions(rewards, evals) + + # total_original = 0, so total_recycled = 0 + # But code has: max(total_recycled, 1.0 if total_original <= 0 else 0.0) + # total_original = 0 → 1.0 floor applies + assert result[RECYCLE_UID] == pytest.approx(1.0) + + def test_all_zero_rewards_with_contributions(self): + """Zero rewards with tiered miners: recycle still gets 1.0 floor.""" + rewards = {1: 0.0} + evals = {1: _make_eval(1, Tier.GOLD, 1000.0, {"r"})} + result = apply_dynamic_emissions_using_network_contributions(rewards, evals) + assert result[RECYCLE_UID] == pytest.approx(1.0) + + def test_many_miners_rewards_consistency(self): + """With many miners, sum of all output rewards equals original total.""" + rewards = {i: 1.0 / 50 for i in range(1, 51)} + evals = { + i: _make_eval(i, Tier.BRONZE, float(i * 10), {f"r{i}"}) + for i in range(1, 51) + } + result = apply_dynamic_emissions_using_network_contributions(rewards, evals) + assert sum(result.values()) == pytest.approx(1.0) + + def test_untiered_miners_dont_affect_scalar(self): + """Untiered miners' contributions don't influence the emission scalar.""" + rewards = {1: 0.5, 2: 0.5} + # Miner 1 is tiered with modest contributions + evals_with_untiered = { + 1: _make_eval(1, Tier.BRONZE, 50.0, {"r1"}), + 2: _make_eval(2, None, 999999.0, {f"r{i}" for i in range(10000)}), + } + evals_without = { + 1: _make_eval(1, Tier.BRONZE, 50.0, {"r1"}), + } + result_with = apply_dynamic_emissions_using_network_contributions(rewards, evals_with_untiered) + result_without = apply_dynamic_emissions_using_network_contributions(rewards, evals_without) + + # Same scalar → same miner rewards + assert result_with[1] == pytest.approx(result_without[1]) + assert result_with[2] == pytest.approx(result_without[2]) + + +if __name__ == "__main__": + pytest.main([__file__, "-v"])