diff --git a/bal_tools/__init__.py b/bal_tools/__init__.py index a0d7bf2..15d9ff3 100644 --- a/bal_tools/__init__.py +++ b/bal_tools/__init__.py @@ -7,5 +7,5 @@ ) from .subgraph import Subgraph from .pools_gauges import BalPoolsGauges -from .ecosystem import Aura +from .ecosystem import Aura, StakeDAO from .drpc import Web3RpcByChain, Web3Rpc diff --git a/bal_tools/ecosystem.py b/bal_tools/ecosystem.py index 64281b3..248b68f 100644 --- a/bal_tools/ecosystem.py +++ b/bal_tools/ecosystem.py @@ -1,10 +1,9 @@ from collections import defaultdict -from datetime import datetime, timedelta, timezone import math +import os import re import statistics -from decimal import Decimal -from typing import Dict, List +from typing import Dict from .errors import ( UnexpectedListLengthError, MultipleMatchesError, @@ -13,8 +12,8 @@ from web3 import Web3 import requests from .subgraph import Subgraph +from .drpc import Web3RpcByChain from .utils import to_checksum_address -from .models import PropData AURA_L2_DEFAULT_GAUGE_STAKER = to_checksum_address( @@ -195,88 +194,55 @@ def get_votes_from_snapshot(self, snapshot_id: str): return votes -class HiddenHand: - AURA_URL = "https://api.hiddenhand.finance/proposal/aura" - SNAPSHOT_URL = "https://hub.snapshot.org/graphql" +class StakeDAO: + ANALYTICS_BASE_URL = "https://raw.githubusercontent.com/stake-dao/votemarket-analytics/refs/heads/main/analytics/votemarket-analytics/balancer" - def fetch_aura_bribs(self) -> List[PropData]: - res = requests.get(self.AURA_URL) - if not res.ok: - raise ValueError("Error fetching bribes from hidden hand api") - res_parsed = res.json() - if res_parsed["error"]: - raise ValueError("HH API returned error") - return [PropData(**prop_data) for prop_data in res_parsed["data"]] + def __init__(self): + self.subgraph = Subgraph() - def _get_previous_round_timestamps(self, n_rounds: int) -> List[int]: - """Round endings are every other Monday 8PM GMT.""" - now = datetime.now(timezone.utc) - last_monday_8pm = (now - timedelta(days=now.weekday())).replace( - hour=20, minute=0, second=0, microsecond=0 + def get_aura_max_votes_from_snapshot(self, n_rounds: int = 2) -> int: + data = self.subgraph.fetch_graphql_data( + subgraph="snapshot", + query="get_aura_gauge_proposals", + params={"space": "gauges.aurafinance.eth", "first": n_rounds}, ) - if last_monday_8pm > now: - last_monday_8pm -= timedelta(weeks=1) - ts = int(last_monday_8pm.timestamp()) - resp = requests.get(f"{self.AURA_URL}/{ts}", timeout=10) - if not any(p.get("valuePerVote", 0) > 0 for p in resp.json().get("data", [])): - last_monday_8pm -= timedelta(weeks=1) - return [ - int((last_monday_8pm - timedelta(weeks=i * 2)).timestamp()) - for i in range(n_rounds) - ] - - def get_min_aura_incentive( - self, n_rounds: int = 2, buffer_pct: float = 0.25 - ) -> Decimal: - """ - Calculate dynamic min_aura_incentive from Snapshot votes and Hidden Hand CPV. - - Formula: ceil(max_votes * 0.005 * (1 + buffer_pct) * median_cpv / 10) * 10 - """ - timestamps = self._get_previous_round_timestamps(n_rounds) - - first = n_rounds * 2 - query = f"""{{ - proposals( - first: {first}, - where: {{space: "gauges.aurafinance.eth"}}, - orderBy: "created", - orderDirection: desc - ) {{ - title - scores_total - }} - }}""" - resp = requests.post(self.SNAPSHOT_URL, json={"query": query}, timeout=10) - resp.raise_for_status() - proposals = resp.json().get("data", {}).get("proposals", []) - gauge_proposals = [ - p for p in proposals if "Gauge Weight for Week of " in p.get("title", "") - ][:n_rounds] - - if not gauge_proposals: - raise ValueError("No gauge weight proposals found in Snapshot") - - max_votes = max(p["scores_total"] for p in gauge_proposals) - + proposals = data.get("proposals", []) + if not proposals: + raise ValueError("No Aura gauge weight proposals found on Snapshot") + return int(max(p["scores_total"] for p in proposals if p.get("scores_total"))) + + def get_cpv_from_analytics(self, n_rounds: int = 2) -> float: + metadata_url = f"{self.ANALYTICS_BASE_URL}/rounds-metadata.json" + response = requests.get(metadata_url, timeout=30) + response.raise_for_status() + rounds = response.json() + + latest_rounds = sorted(rounds, key=lambda x: x["id"], reverse=True)[:n_rounds] cpv_values = [] - for ts in timestamps: - resp = requests.get(f"{self.AURA_URL}/{ts}", timeout=10) - resp.raise_for_status() - for p in resp.json().get("data", []): - if p.get("valuePerVote", 0) > 0: - cpv_values.append(p["valuePerVote"]) + for round_info in latest_rounds: + round_url = f"{self.ANALYTICS_BASE_URL}/{round_info['id']}.json" + response = requests.get(round_url, timeout=30) + response.raise_for_status() + data = response.json() + cpv = data.get("globalAverageDollarPerVote") + if cpv and cpv > 0: + cpv_values.append(float(cpv)) if not cpv_values: - raise ValueError("No valid CPV data found in Hidden Hand") - - median_cpv = statistics.median(cpv_values) - buffer_multiplier = Decimal(str(1 + buffer_pct)) - - raw = ( - Decimal(str(max_votes)) - * Decimal("0.005") - * buffer_multiplier - * Decimal(str(median_cpv)) + raise ValueError("No valid CPV data found in StakeDAO analytics") + return statistics.mean(cpv_values) + + def calculate_dynamic_min_incentive( + self, n_rounds: int = 2, buffer_pct: float = 0.5 + ) -> int: + max_votes = self.get_aura_max_votes_from_snapshot(n_rounds) + avg_cpv = self.get_cpv_from_analytics(n_rounds) + + web3 = Web3RpcByChain(os.getenv("DRPC_KEY"))["mainnet"] + aura_vebal_share = float( + self.subgraph.calculate_aura_vebal_share(web3, web3.eth.block_number) ) - return Decimal(math.ceil(float(raw) / 10)) * 10 + + min_aura_portion = max_votes * 0.001 * (1 + buffer_pct) * avg_cpv + min_total_bribe = min_aura_portion / aura_vebal_share + return int(math.ceil(min_total_bribe / 10) * 10) diff --git a/bal_tools/graphql/snapshot/get_aura_gauge_proposals.gql b/bal_tools/graphql/snapshot/get_aura_gauge_proposals.gql new file mode 100644 index 0000000..15013ee --- /dev/null +++ b/bal_tools/graphql/snapshot/get_aura_gauge_proposals.gql @@ -0,0 +1,17 @@ +query GetAuraGaugeProposals($first: Int!, $space: String!) { + proposals ( + first: $first, + skip: 0, + where: { + space_in: [$space], + title_contains: "Gauge Weight" + }, + orderBy: "created", + orderDirection: desc + ) { + id + title + scores_total + created + } +} diff --git a/bal_tools/subgraph.py b/bal_tools/subgraph.py index 0f9b369..5f2b8ba 100644 --- a/bal_tools/subgraph.py +++ b/bal_tools/subgraph.py @@ -49,6 +49,7 @@ def url_dict_from_df(df): flavor="lxml", ) +SNAPSHOT_URL = "https://hub.snapshot.org/graphql" AURA_SUBGRAPH_URI = "https://api.subgraph.ormilabs.com/api/public/396b336b-4ed7-469f-a8f4-468e1e26e9a8/subgraphs" AURA_SUBGRAPHS_BY_CHAIN = { "mainnet": f"{AURA_SUBGRAPH_URI}/aura-finance-mainnet/v0.0.1/", @@ -103,6 +104,8 @@ def get_subgraph_url(self, subgraph="core") -> str: - https url of the subgraph """ # before anything else, try to get the url from the latest backend config + if subgraph == "snapshot": + return SNAPSHOT_URL if subgraph == "aura": return AURA_SUBGRAPHS_BY_CHAIN.get(self.chain, None) url = self.get_subgraph_url_from_backend_config(subgraph) diff --git a/bal_tools/ts_config_loader.py b/bal_tools/ts_config_loader.py index a919e6d..3c9704a 100644 --- a/bal_tools/ts_config_loader.py +++ b/bal_tools/ts_config_loader.py @@ -102,6 +102,27 @@ def replace_arrow_functions(text): obj = replace_arrow_functions(obj) + # 1b2) Handle async arrow functions with block bodies (no return type annotation) + def replace_async_block_functions(text): + pattern = r"async\s*\([^)]*\)\s*=>\s*\{" + while True: + match = re.search(pattern, text) + if not match: + break + start = match.start() + i = match.end() - 1 + depth = 1 + while i < len(text) - 1 and depth > 0: + i += 1 + if text[i] == "{": + depth += 1 + elif text[i] == "}": + depth -= 1 + text = text[:start] + "null" + text[i + 1 :] + return text + + obj = replace_async_block_functions(obj) + # 1c) Handle $.xxx patterns (JSONPath expressions) by quoting them # This handles path values like $.apy or $.data obj = re.sub(r"(\$\.[a-zA-Z0-9_.]+)", r'"\1"', obj) diff --git a/tests/test_ecosystem.py b/tests/test_ecosystem.py index 577d037..8707be9 100644 --- a/tests/test_ecosystem.py +++ b/tests/test_ecosystem.py @@ -1,9 +1,9 @@ -from decimal import Decimal -from bal_tools.ecosystem import HiddenHand +from bal_tools.ecosystem import StakeDAO -def test_get_min_aura_incentive(): - hh = HiddenHand() - result = hh.get_min_aura_incentive() +def test_calculate_dynamic_min_incentive(): + sd = StakeDAO() + result = sd.calculate_dynamic_min_incentive() - assert isinstance(result, Decimal) + assert isinstance(result, int) + assert result > 0