diff --git a/scripts/nym-node-setup/nym-node-cli.py b/scripts/nym-node-setup/nym-node-cli.py index 8c97104e61..ebae150523 100755 --- a/scripts/nym-node-setup/nym-node-cli.py +++ b/scripts/nym-node-setup/nym-node-cli.py @@ -1,6 +1,6 @@ #!/usr/bin/python3 -__version__ = "1.0.0" +__version__ = "1.1.0" __default_branch__ = "develop" import os diff --git a/scripts/rewards-tracker/data/wallet-addresses.csv b/scripts/rewards-tracker/data/wallet-addresses.csv new file mode 100644 index 0000000000..aabb9accf6 --- /dev/null +++ b/scripts/rewards-tracker/data/wallet-addresses.csv @@ -0,0 +1,6 @@ +address,tag +add_wallet_or_delete,optional_tag_or_delete +add_wallet_or_delete,optional_tag_or_delete +add_wallet_or_delete,optional_tag_or_delete +add_wallet_or_delete,optional_tag_or_delete +add_wallet_or_delete,optional_tag_or_delete diff --git a/scripts/rewards-tracker/node_rewards_tracker.py b/scripts/rewards-tracker/node_rewards_tracker.py new file mode 100644 index 0000000000..4b48690570 --- /dev/null +++ b/scripts/rewards-tracker/node_rewards_tracker.py @@ -0,0 +1,627 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- + +""" +This script fetches operators rewards based on provided Nyx account addresses provided in data/wallet-addresses.csv. +Output is: + 1. Printet table in terminal + 2. Sheet with complete info stored in data/node-balances.csv + 3. Hiostorical data yaml file stored in data/data.yaml - this file should not be changed by hand, as + all values older than 30 days get auto-removed +Before you start fill first column of data/wallet-addresses with your Nyx account addresses and (optionally) second column +with a tag, for example "mysquad" and "personal" to get sorted output per entity. +""" + +import csv +import os +import sys +import time +import yaml +import requests +from collections import defaultdict +from typing import Any, Dict, List, Tuple, Optional +from tabulate import tabulate +from colorama import init as colorama_init, Fore, Style + +colorama_init() + +DATA_DIR = os.path.join(os.getcwd(), "data") +ADDR_CSV = os.path.join(DATA_DIR, "wallet-addresses.csv") +OUT_CSV = os.path.join(DATA_DIR, "node-balances.csv") +HIST_FILE = os.path.join(DATA_DIR, "data.yaml") + +SPECTRE_NODES_URL = "https://api.nym.spectredao.net/api/v1/nodes" +VALIDATOR_BONDED_URL = "https://validator.nymtech.net/api/v1/nym-nodes/bonded" +VALIDATOR_DESC_URL = "https://validator.nymtech.net/api/v1/nym-nodes/described" +SPECTRE_BAL_URL = "https://api.nym.spectredao.net/api/v1/balances/{address}" + +SESSION = requests.Session() +SESSION.headers.update({"User-Agent": "nym-tools/1.0"}) + + +def log(msg: str) -> None: + print(msg, flush=True) + + +def now_ts() -> float: + return time.time() + + +def to_float(x: Any, default: float = 0.0) -> float: + try: + if x is None: + return default + if isinstance(x, (int, float)): + return float(x) + if isinstance(x, str): + return float(x) + return default + except Exception: + return default + + +def to_int(x: Any, default: int = 0) -> int: + try: + if x is None: + return default + if isinstance(x, int): + return x + if isinstance(x, float): + return int(x) + if isinstance(x, str): + return int(float(x)) + return default + except Exception: + return default + + +# pagination helpers + +def _get_json(url: str, params: Dict[str, Any], timeout: int = 60) -> Any: + r = SESSION.get(url, params=params, timeout=timeout) + r.raise_for_status() + return r.json() + +def _fetch_all_limit_offset(url: str, limit: int = 1000, timeout: int = 60) -> List[Any]: + out: List[Any] = [] + offset = 0 + tries = 0 + while True: + tries += 1 + data = _get_json(url, {"limit": limit, "offset": offset}, timeout=timeout) + if isinstance(data, dict) and "data" in data and isinstance(data["data"], list): + items = data["data"] + elif isinstance(data, list): + items = data + else: + break + if not items: + break + out.extend(items) + if len(items) < limit: + break + offset += limit + if tries > 500: + break + return out + +def _fetch_all_page_pagesize(url: str, page_size: int = 1000, timeout: int = 60) -> List[Any]: + out: List[Any] = [] + page = 0 + tries = 0 + while True: + tries += 1 + data = _get_json(url, {"page": page, "size": page_size}, timeout=timeout) + if isinstance(data, dict) and "data" in data and isinstance(data["data"], list): + items = data["data"] + elif isinstance(data, list): + items = data + else: + break + out.extend(items) + total = to_int(data.get("pagination", {}).get("total"), -1) if isinstance(data, dict) else -1 + if total >= 0 and len(out) >= total: + break + if not items: + break + page += 1 + if tries > 500: + break + return out + +def _fetch_all_single(url: str, timeout: int = 60) -> List[Any]: + data = _get_json(url, {}, timeout=timeout) + if isinstance(data, dict) and "data" in data and isinstance(data["data"], list): + return data["data"] + if isinstance(data, list): + return data + return [] + +def _fetch_all_any(url: str, timeout: int = 60) -> list: + got = _fetch_all_limit_offset(url, limit=1000, timeout=timeout) + if isinstance(got, list) and got: + return got + got = _fetch_all_page_pagesize(url, page_size=1000, timeout=timeout) + if isinstance(got, list) and got: + return got + return _fetch_all_single(url, timeout=timeout) + + +# load data + +def read_wallets_csv(path: str) -> List[Tuple[str, str]]: + if not os.path.exists(path): + raise FileNotFoundError(f"Input CSV not found: {path}") + rows: List[Tuple[str, str]] = [] + with open(path, newline="", encoding="utf-8") as f: + reader = csv.DictReader(f) + for row in reader: + addr = (row.get("address") or "").strip() + tag = (row.get("tag") or "").strip() + if addr: + rows.append((addr, tag)) + return rows + +def fetch_nodes_all() -> List[Dict[str, Any]]: + log(f"* * * Fetching nodes from {SPECTRE_NODES_URL} * * *") + nodes = _fetch_all_any(SPECTRE_NODES_URL, timeout=90) + log(f"Fetched {len(nodes)} node(s)") + return nodes + +def fetch_bonded_all() -> List[Dict[str, Any]]: + log(f"* * *Fetching bonded from {VALIDATOR_BONDED_URL} * * *") + bonded = _fetch_all_any(VALIDATOR_BONDED_URL, timeout=90) + log(f"Fetched {len(bonded)} bonded record(s)") + return bonded + +def fetch_described_all() -> List[Dict[str, Any]]: + log(f"* * * Fetching described from {VALIDATOR_DESC_URL} * * *") + described = _fetch_all_any(VALIDATOR_DESC_URL, timeout=90) + log(f"Fetched {len(described)} described record(s)") + return described + +def fetch_balance_total_nym(address: str) -> float: + url = SPECTRE_BAL_URL.format(address=address) + try: + r = SESSION.get(url, timeout=30) + r.raise_for_status() + js = r.json() + amt = to_float(js.get("total", {}).get("amount"), 0.0) + return amt / 1_000_000.0 + except Exception as e: + log(f"{Fore.YELLOW}* * * warn: balance fetch failed for {address}: {e}{Style.RESET_ALL} * * *") + return 0.0 + + +# extract version + +def _first_str(*vals) -> str: + for v in vals: + if isinstance(v, (str, int, float)): + s = str(v).strip() + if s: + return s + return "" + +def extract_version_from_node(n: Dict[str, Any]) -> str: + desc = n.get("description") or {} + bi = n.get("build_information") or {} + return _first_str( + n.get("version"), + n.get("node_version"), + bi.get("build_version"), + bi.get("version"), + (desc.get("software") or {}).get("version"), + (desc.get("build_information") or {}).get("build_version"), + ) + +def extract_version_from_desc(d: Dict[str, Any]) -> str: + desc = d.get("description") or {} + bi = d.get("build_information") or {} + return _first_str( + d.get("version"), + d.get("node_version"), + bi.get("build_version"), + bi.get("version"), + (desc.get("software") or {}).get("version"), + (desc.get("build_information") or {}).get("build_version"), + ) + + +# history storage in data/data.yaml + 30d cleanup + window helpers + +def load_history(path: str) -> Dict[str, Any]: + if not os.path.exists(path): + return {} + with open(path, "r", encoding="utf-8") as f: + try: + return yaml.safe_load(f) or {} + except Exception: + return {} + +def save_history(path: str, data: Dict[str, Any]) -> None: + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as f: + yaml.safe_dump(data, f, sort_keys=True) + +def add_history_point(hist: Dict[str, Any], node_id: int, epoch_ts: float, uptime: float, op_bal: float) -> None: + key = str(node_id) + lst = hist.setdefault(key, []) + lst.append({ + "ts": epoch_ts, + "uptime": round(float(uptime), 6), + "operator_balance": round(float(op_bal), 6), + }) + +def last_snapshot(hist: Dict[str, Any], node_id: int) -> Optional[Dict[str, Any]]: + key = str(node_id) + if key not in hist: + return None + lst = hist[key] + if not lst: + return None + return sorted(lst, key=lambda x: x.get("ts", 0.0))[-1] + +def cleanup_history_older_than(hist: Dict[str, Any], cutoff_ts: float) -> None: + # remove entries older than cutoff - 30 days + for key, lst in list(hist.items()): + new_lst = [e for e in lst if to_float(e.get("ts"), 0.0) >= cutoff_ts] + if new_lst: + hist[key] = new_lst + else: + del hist[key] + +def scaled_window_change( + hist: Dict[str, Any], + node_id: int, + now: float, + window_days: float, + current_balance: float +) -> Optional[Tuple[float, float, float]]: + key = str(node_id) + if key not in hist or not hist[key]: + return None + cutoff_ts = now - window_days * 24 * 3600 + candidates = [e for e in hist[key] if to_float(e.get("ts"), 0.0) <= cutoff_ts] + if not candidates: + return None + snap = sorted(candidates, key=lambda x: x.get("ts", 0.0))[-1] + span_hours = max(0.0, (now - to_float(snap.get("ts"), now)) / 3600.0) + if span_hours <= 0: + return None + profit_raw = current_balance - to_float(snap.get("operator_balance"), 0.0) + target_hours = window_days * 24.0 + profit_scaled = profit_raw * (target_hours / span_hours) + hourly_scaled = profit_scaled / target_hours + return profit_scaled, span_hours, hourly_scaled + + +# output coloring fns + +def colorize(text: str, color_name: str) -> str: + mapping = { + "green": Fore.GREEN, + "yellow": Fore.YELLOW, + "orange": Fore.MAGENTA, + "red": Fore.RED, + } + c = mapping.get(color_name, "") + if not c: + return text + return f"{c}{text}{Style.RESET_ALL}" + +def uptime_color_name(u: float) -> str: + if u >= 0.95: + return "green" + if u >= 0.90: + return "yellow" + if u >= 0.80: + return "orange" + return "red" + + +# main program body + +def main() -> None: + os.makedirs(DATA_DIR, exist_ok=True) + + wallets = read_wallets_csv(ADDR_CSV) + if not wallets: + log(f"{Fore.RED}No wallets found in {ADDR_CSV}{Style.RESET_ALL}") + sys.exit(1) + log(f"Found {len(wallets)} wallet(s) in {ADDR_CSV}") + + # preserve input order per wallet for per-tag tables + wallet_order = {addr: idx for idx, (addr, _tag) in enumerate(wallets)} + + nodes = fetch_nodes_all() + bonded = fetch_bonded_all() + described = fetch_described_all() + + # indexes + idx_nodes_by_wallet: Dict[str, List[Dict[str, Any]]] = defaultdict(list) + for n in nodes: + w = (n.get("bonding_address") or "").strip() + if w: + idx_nodes_by_wallet[w].append(n) + + idx_desc_by_node_id: Dict[int, Dict[str, Any]] = {} + for d in described: + nid = to_int(d.get("node_id"), 0) + if nid: + idx_desc_by_node_id[nid] = d + + idx_bonded_by_owner: Dict[str, List[Dict[str, Any]]] = defaultdict(list) + for b in bonded: + bi = b.get("bond_information", {}) + owner = (bi.get("owner") or "").strip() + if owner: + idx_bonded_by_owner[owner].append(b) + + headers_csv = [ + "node_id", + "hostname", + "identity_key", + "wallet", + "uptime", + "version", + "operator_balance", + "profit_difference", + "epochs", + "average_hour", + "7_days", + "7_days_average", + "30_days", + "30_days_average", + "tag", + ] + + hist = load_history(HIST_FILE) + now = now_ts() + + out_rows: List[Dict[str, Any]] = [] + rows_by_tag: Dict[str, List[Dict[str, Any]]] = defaultdict(list) + + THIRTY_DAYS_SEC = 30 * 24 * 3600 + cleanup_history_older_than(hist, now - THIRTY_DAYS_SEC) # prune before use + + for wallet_addr, tag in wallets: + wallet_nodes = idx_nodes_by_wallet.get(wallet_addr, []) + + if not wallet_nodes: + # fallback via bonded -> described + balance + for b in idx_bonded_by_owner.get(wallet_addr, []): + bi = b.get("bond_information", {}) + nid = to_int(bi.get("node_id"), 0) + if nid <= 0: + continue + d = idx_desc_by_node_id.get(nid, {}) + desc = d.get("description", {}) if isinstance(d, dict) else {} + hostinfo = desc.get("host_information", {}) if isinstance(desc, dict) else {} + hostname = hostinfo.get("hostname") or "" + node = bi.get("node", {}) if isinstance(bi, dict) else {} + identity_key = node.get("identity_key") or "" + + op_bal = fetch_balance_total_nym(wallet_addr) + uptime = 0.0 + + # since last time change calculation + prev = last_snapshot(hist, nid) + prev_bal = to_float(prev.get("operator_balance"), 0.0) if prev else None + prev_ts = to_float(prev.get("ts"), 0.0) if prev else None + diff = 0.0 + hours = 0.0 + if prev is not None: + diff = op_bal - prev_bal + hours = max(0.0, (now - prev_ts) / 3600.0) + + # last 7 / 30 days calculation + seven = scaled_window_change(hist, nid, now, 7.0, op_bal) + thirty = scaled_window_change(hist, nid, now, 30.0, op_bal) + + row = { + "node_id": nid, + "hostname": hostname, + "identity_key": identity_key, + "wallet": wallet_addr, + "uptime": uptime, + "version": extract_version_from_desc(d), + "operator_balance": op_bal, + "profit_difference": diff, + "epochs": hours, + "average_hour": (diff / hours) if hours > 0 else 0.0, + "7_days": f"{seven[0]:.6f}" if seven else "no 7 days data stored", + "7_days_average": f"{seven[2]:.6f}" if seven else "no 7 days data stored", + "30_days": f"{thirty[0]:.6f}" if thirty else "no 30 days data stored", + "30_days_average": f"{thirty[2]:.6f}" if thirty else "no 30 days data stored", + "tag": tag, + "_prev_balance": prev_bal, + "_prev_ts": prev_ts, + "_wallet_order": wallet_order.get(wallet_addr, 10**9), + } + + # append current snapshot & prune >30d + add_history_point(hist, nid, now, uptime, op_bal) + + out_rows.append(row) + rows_by_tag[tag].append(row) + continue + + # path from /nodes + for n in wallet_nodes: + nid = to_int(n.get("node_id"), 0) + if nid <= 0: + continue + identity_key = n.get("identity_key") or "" + uptime = to_float(n.get("uptime"), 0.0) + desc = n.get("description") or {} + hostinfo = desc.get("host_information") or {} + hostname = hostinfo.get("hostname") or "" + op_unym = to_float(n.get("rewarding_details", {}).get("operator"), 0.0) + op_bal = op_unym / 1_000_000.0 + if op_bal <= 0: + op_bal = fetch_balance_total_nym(wallet_addr) + + prev = last_snapshot(hist, nid) + prev_bal = to_float(prev.get("operator_balance"), 0.0) if prev else None + prev_ts = to_float(prev.get("ts"), 0.0) if prev else None + + diff = 0.0 + hours = 0.0 + if prev is not None: + diff = op_bal - prev_bal + hours = max(0.0, (now - prev_ts) / 3600.0) + + seven = scaled_window_change(hist, nid, now, 7.0, op_bal) + thirty = scaled_window_change(hist, nid, now, 30.0, op_bal) + + row = { + "node_id": nid, + "hostname": hostname, + "identity_key": identity_key, + "wallet": wallet_addr, + "uptime": uptime, + "version": extract_version_from_node(n), + "operator_balance": op_bal, + "profit_difference": diff, + "epochs": hours, + "average_hour": (diff / hours) if hours > 0 else 0.0, + "7_days": f"{seven[0]:.6f}" if seven else "no 7 days data stored", + "7_days_average": f"{seven[2]:.6f}" if seven else "no 7 days data stored", + "30_days": f"{thirty[0]:.6f}" if thirty else "no 30 days data stored", + "30_days_average": f"{thirty[2]:.6f}" if thirty else "no 30 days data stored", + "tag": tag, + "_prev_balance": prev_bal, + "_prev_ts": prev_ts, + "_wallet_order": wallet_order.get(wallet_addr, 10**9), + } + + add_history_point(hist, nid, now, uptime, op_bal) + + out_rows.append(row) + rows_by_tag[tag].append(row) + + # final prune & save + cleanup_history_older_than(hist, now - THIRTY_DAYS_SEC) + save_history(HIST_FILE, hist) + + # write CSV + headers_csv = [ + "node_id","hostname","identity_key","wallet","uptime","version","operator_balance", + "profit_difference","epochs","average_hour","7_days","7_days_average", + "30_days","30_days_average","tag", + ] + with open(OUT_CSV, "w", newline="", encoding="utf-8") as f: + writer = csv.DictWriter(f, fieldnames=headers_csv) + writer.writeheader() + for r in out_rows: + writer.writerow({ + "node_id": r.get("node_id",""), + "hostname": r.get("hostname",""), + "identity_key": r.get("identity_key",""), + "wallet": r.get("wallet",""), + "uptime": f"{to_float(r.get('uptime'),0.0):.6f}", + "version": r.get("version",""), + "operator_balance": f"{to_float(r.get('operator_balance'),0.0):.6f}", + "profit_difference": f"{to_float(r.get('profit_difference'),0.0):.6f}", + "epochs": f"{to_float(r.get('epochs'),0.0):.6f}", + "average_hour": f"{to_float(r.get('average_hour'),0.0):.6f}", + "7_days": r.get("7_days",""), + "7_days_average": r.get("7_days_average",""), + "30_days": r.get("30_days",""), + "30_days_average": r.get("30_days_average",""), + "tag": r.get("tag",""), + }) + + if not out_rows: + log(f"{Fore.YELLOW}No rows produced — check inputs and endpoints.{Style.RESET_ALL}") + return + + # per-tag output (preserve input order) + for tag, rows in sorted(rows_by_tag.items(), key=lambda kv: kv[0] or ""): + headers_print = [ + "node_id","hostname","wallet","uptime","version", + "operator_balance","profit_difference","epochs","average_hour", + "7_days","7_days_average","30_days","30_days_average", + ] + view: List[List[str]] = [] + for r in rows: + u = to_float(r.get("uptime"), 0.0) + u_col = uptime_color_name(u) + view.append([ + r.get("node_id") or "", + (r.get("hostname") or "")[:40], + r.get("wallet") or "", + colorize(f"{u:.2f}", u_col) if u_col else f"{u:.2f}", + r.get("version") or "", + f"{to_float(r.get('operator_balance'), 0.0):.2f}", + f"{to_float(r.get('profit_difference'), 0.0):.2f}", + f"{to_float(r.get('epochs'), 0.0):.2f}", + f"{to_float(r.get('average_hour'), 0.0):.2f}", + r.get("7_days"), + r.get("7_days_average"), + r.get("30_days"), + r.get("30_days_average"), + ]) + + title = f"Tag: {tag or '(untagged)'} — {len(rows)} node(s)" + print("\n" + title) + print(tabulate(view, headers=headers_print, tablefmt="github", stralign="right", disable_numparse=True)) + + # per-tag summary + tag_total_now = sum(to_float(r.get("operator_balance"), 0.0) for r in rows) + + # since last time: sum diffs, average hours across nodes that existed + prev_sum = 0.0 + prev_hours: List[float] = [] + for r in rows: + prev_bal = r.get("_prev_balance") + prev_ts = r.get("_prev_ts") + if prev_bal is not None and prev_ts is not None: + prev_sum += to_float(prev_bal, 0.0) + prev_hours.append(max(0.0, (now - to_float(prev_ts, now)) / 3600.0)) + diff_total = tag_total_now - prev_sum if prev_hours else 0.0 + hours_since = (sum(prev_hours) / len(prev_hours)) if prev_hours else 0.0 + hourly = (diff_total / hours_since) if hours_since > 0 else 0.0 + + # 7-day / 30-day per-tag totals: sum nodes with data; hours fixed windows if any + def _num_or_none(v): + try: + return float(v) + except Exception: + return None + seven_vals = [_num_or_none(r.get("7_days")) for r in rows] + seven_vals = [v for v in seven_vals if v is not None] + total7 = sum(seven_vals) if seven_vals else 0.0 + hours7 = 7.0 * 24.0 if seven_vals else 0.0 + hourly7 = (total7 / hours7) if hours7 > 0 else 0.0 + + thirty_vals = [_num_or_none(r.get("30_days")) for r in rows] + thirty_vals = [v for v in thirty_vals if v is not None] + total30 = sum(thirty_vals) if thirty_vals else 0.0 + hours30 = 30.0 * 24.0 if thirty_vals else 0.0 + hourly30 = (total30 / hours30) if hours30 > 0 else 0.0 + + # print output with colored numbers + print( + "\n" + f"Total balance across all wallets: {Style.BRIGHT}{Fore.GREEN}{tag_total_now:.2f}{Style.RESET_ALL} NYM\n" + f"Difference of total balance from last time: {Fore.CYAN}{diff_total:.2f}{Style.RESET_ALL} NYM\n" + f"Time since last time: {Fore.BLUE}{hours_since:.2f}{Style.RESET_ALL} hours\n" + f"Approx hourly difference: {Style.BRIGHT}{Fore.GREEN}{hourly:.2f}{Style.RESET_ALL} NYM/h\n" + f"7-day change: " + f"{('no 7 days data stored' if not seven_vals else f'{Fore.CYAN}{total7:.2f}{Style.RESET_ALL} NYM, hours: {Fore.BLUE}{hours7:.2f}{Style.RESET_ALL}, hourly: {Style.BRIGHT}{Fore.GREEN}{hourly7:.2f}{Style.RESET_ALL} NYM/h')}\n" + f"30-day change: " + f"{('no 30 days data stored' if not thirty_vals else f'{Fore.CYAN}{total30:.2f}{Style.RESET_ALL} NYM, hours: {Fore.BLUE}{hours30:.2f}{Style.RESET_ALL}, hourly: {Style.BRIGHT}{Fore.GREEN}{hourly30:.2f}{Style.RESET_ALL} NYM/h')}\n" + ) + + log(f"\nCSV written to: {OUT_CSV}") + log(f"History saved to: {HIST_FILE}") + + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + print("\nInterrupted.", file=sys.stderr) + sys.exit(130)