From 27dca16923089ea62ac9720194ba3ce10b47b5db Mon Sep 17 00:00:00 2001 From: jalbrekt85 Date: Thu, 4 Dec 2025 12:30:08 -0600 Subject: [PATCH 1/2] fetch historical hh data using proposal end time --- bal_tools/ecosystem.py | 42 ++++++++++++++++++++++------------------- tests/test_ecosystem.py | 9 +++++++++ 2 files changed, 32 insertions(+), 19 deletions(-) create mode 100644 tests/test_ecosystem.py diff --git a/bal_tools/ecosystem.py b/bal_tools/ecosystem.py index fbec26e..b6577ec 100644 --- a/bal_tools/ecosystem.py +++ b/bal_tools/ecosystem.py @@ -1,4 +1,5 @@ from collections import defaultdict +from datetime import datetime, timedelta, timezone import math import re import statistics @@ -199,18 +200,27 @@ class HiddenHand: SNAPSHOT_URL = "https://hub.snapshot.org/graphql" def fetch_aura_bribs(self) -> List[PropData]: - """ - Fetch GET bribes from hidden hand api - """ 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 _get_previous_round_timestamps(self, n_rounds: int) -> List[int]: + """Round endings are every 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 + ) + if last_monday_8pm > now: + last_monday_8pm -= timedelta(weeks=1) + return [ + int((last_monday_8pm - timedelta(weeks=i)).timestamp()) + for i in range(n_rounds) + ] + def get_min_aura_incentive( self, n_rounds: int = 4, buffer_pct: float = 0.25 ) -> Decimal: @@ -218,14 +228,9 @@ def get_min_aura_incentive( 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 - - params: - - n_rounds: number of recent gauge weight proposals to consider (default: 4) - - buffer_pct: buffer percentage to add to the threshold (default: 0.25 = 25%) - - returns: - - Decimal threshold value in USD """ + timestamps = self._get_previous_round_timestamps(n_rounds) + first = n_rounds * 2 query = f"""{{ proposals( @@ -238,7 +243,6 @@ def get_min_aura_incentive( scores_total }} }}""" - resp = requests.post(self.SNAPSHOT_URL, json={"query": query}, timeout=10) resp.raise_for_status() proposals = resp.json().get("data", {}).get("proposals", []) @@ -251,13 +255,13 @@ def get_min_aura_incentive( max_votes = max(p["scores_total"] for p in gauge_proposals) - resp = requests.get(self.AURA_URL, timeout=10) - resp.raise_for_status() - cpv_values = [ - p["valuePerVote"] - for p in resp.json().get("data", []) - if p.get("valuePerVote", 0) > 0 - ] + 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"]) if not cpv_values: raise ValueError("No valid CPV data found in Hidden Hand") diff --git a/tests/test_ecosystem.py b/tests/test_ecosystem.py new file mode 100644 index 0000000..577d037 --- /dev/null +++ b/tests/test_ecosystem.py @@ -0,0 +1,9 @@ +from decimal import Decimal +from bal_tools.ecosystem import HiddenHand + + +def test_get_min_aura_incentive(): + hh = HiddenHand() + result = hh.get_min_aura_incentive() + + assert isinstance(result, Decimal) From e44dc0dc583774c6d3955ccff14a5a35ce7bb450 Mon Sep 17 00:00:00 2001 From: jalbrekt85 Date: Thu, 4 Dec 2025 13:40:41 -0600 Subject: [PATCH 2/2] only calc timestamps biweekly --- bal_tools/ecosystem.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/bal_tools/ecosystem.py b/bal_tools/ecosystem.py index b6577ec..64281b3 100644 --- a/bal_tools/ecosystem.py +++ b/bal_tools/ecosystem.py @@ -209,20 +209,24 @@ def fetch_aura_bribs(self) -> List[PropData]: return [PropData(**prop_data) for prop_data in res_parsed["data"]] def _get_previous_round_timestamps(self, n_rounds: int) -> List[int]: - """Round endings are every Monday 8PM GMT.""" + """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 ) 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)).timestamp()) + int((last_monday_8pm - timedelta(weeks=i * 2)).timestamp()) for i in range(n_rounds) ] def get_min_aura_incentive( - self, n_rounds: int = 4, buffer_pct: float = 0.25 + self, n_rounds: int = 2, buffer_pct: float = 0.25 ) -> Decimal: """ Calculate dynamic min_aura_incentive from Snapshot votes and Hidden Hand CPV.