replace hh dynamic threshold calc with stakedao version

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

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)

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
+  }
+}

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)

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)

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