Add missing documentation for Web3 login feature#5
Conversation
Co-authored-by: paseka10jaroslav-coder <252813980+paseka10jaroslav-coder@users.noreply.github.com>
e4b6f9c
into
copilot/add-user-login-feature
paseka10jaroslav-coder
left a comment
There was a problem hiding this comment.
"""
Liquid Staking Rewards Tracker
Sledování odměn z liquid stakingu na TON a Solana.
Funkce:
- Načítání aktuálního kurzu tsTON/TON a JitoSOL/mSOL/SOL
- Výpočet stakingových odměn v reálném čase
- Historie výnosů a APY kalkulace
- Export do CSV/JSON
- Notifikace přes Telegram (volitelné)
Požadavky:
pip install requests aiohttp python-dotenv tabulate
Konfigurace:
Vytvoř soubor .env nebo nastav proměnné prostředí (viz CONFIG sekce).
"""
import os
import json
import csv
import time
import asyncio
from datetime import datetime, timedelta
from pathlib import Path
from dataclasses import dataclass, field, asdict
from typing import Optional
import requests
try:
from tabulate import tabulate
HAS_TABULATE = True
except ImportError:
HAS_TABULATE = False
try:
from dotenv import load_dotenv
load_dotenv()
except ImportError:
pass
============================================================
KONFIGURACE
============================================================
@DataClass
class Config:
"""Konfigurace trackeru. Nastav přes .env nebo přímo zde."""
# --- TON ---
ton_wallet_address: str = os.getenv("TON_WALLET_ADDRESS", "")
# Množství tsTON, které držíš (pokud nemáš wallet adresu)
tston_amount: float = float(os.getenv("TSTON_AMOUNT", "0"))
# Množství stTON (bemo)
stton_amount: float = float(os.getenv("STTON_AMOUNT", "0"))
# --- Solana ---
sol_wallet_address: str = os.getenv("SOL_WALLET_ADDRESS", "")
# Množství JitoSOL
jitosol_amount: float = float(os.getenv("JITOSOL_AMOUNT", "0"))
# Množství mSOL
msol_amount: float = float(os.getenv("MSOL_AMOUNT", "0"))
# --- Telegram notifikace (volitelné) ---
telegram_bot_token: str = os.getenv("TELEGRAM_BOT_TOKEN", "")
telegram_chat_id: str = os.getenv("TELEGRAM_CHAT_ID", "")
# --- Sledování ---
history_file: str = os.getenv("HISTORY_FILE", "staking_history.json")
csv_export_file: str = os.getenv("CSV_EXPORT_FILE", "staking_rewards.csv")
check_interval_minutes: int = int(os.getenv("CHECK_INTERVAL_MINUTES", "60"))
config = Config()
============================================================
DATOVÉ MODELY
============================================================
@DataClass
class StakingPosition:
"""Jedna stakinová pozice."""
chain: str # "TON" nebo "Solana"
protocol: str # "Tonstakers", "Jito", "Marinade"
lst_token: str # "tsTON", "JitoSOL", "mSOL"
lst_amount: float # Množství LST tokenů
native_token: str # "TON" nebo "SOL"
exchange_rate: float # Kolik nativních tokenů za 1 LST
native_value: float # Celková hodnota v nativním tokenu
usd_price: float # Cena nativního tokenu v USD
usd_value: float # Celková hodnota v USD
apy_estimate: float # Odhadované APY v %
timestamp: str = ""
def __post_init__(self):
if not self.timestamp:
self.timestamp = datetime.utcnow().isoformat()
@DataClass
class PortfolioSnapshot:
"""Snapshot celého portfolia v daném čase."""
timestamp: str
positions: list
total_usd: float
total_ton_value: float
total_sol_value: float
============================================================
TON API - Tonstakers & bemo
============================================================
class TONTracker:
"""Sledování liquid stakingu na TON blockchainu."""
TONCENTER_API = "https://toncenter.com/api/v2"
TONAPI_URL = "https://tonapi.io/v2"
STON_FI_API = "https://api.ston.fi/v1"
def get_tston_rate(self) -> Optional[float]:
"""
Získá aktuální směnný kurz tsTON → TON.
Používá STON.fi DEX API pro aktuální cenu.
"""
try:
# Metoda 1: STON.fi pool price
resp = requests.get(
f"{self.STON_FI_API}/markets",
timeout=10
)
if resp.status_code == 200:
markets = resp.json()
# Hledáme tsTON/TON pool
for market in markets.get("market_list", []):
base = market.get("base_name", "").upper()
quote = market.get("quote_name", "").upper()
if "TSTON" in base and "TON" in quote:
return float(market.get("last_price", 0))
elif "TON" in base and "TSTON" in quote:
price = float(market.get("last_price", 0))
return 1 / price if price > 0 else None
except Exception as e:
print(f" ⚠️ STON.fi API error: {e}")
try:
# Metoda 2: Hardcoded odhad z on-chain dat
# tsTON rate roste cca o ~0.013% denně (při 4.8% APY)
# Baseline: 1 tsTON ≈ 1.08 TON (přibližný aktuální kurz)
print(" ℹ️ Používám odhadovaný kurz tsTON/TON")
return 1.08
except Exception:
return None
def get_stton_rate(self) -> Optional[float]:
"""Získá aktuální směnný kurz stTON → TON (bemo)."""
try:
# stTON rate je podobný jako tsTON
print(" ℹ️ Používám odhadovaný kurz stTON/TON")
return 1.07
except Exception:
return None
def get_ton_price_usd(self) -> Optional[float]:
"""Získá aktuální cenu TON v USD."""
try:
resp = requests.get(
"https://api.coingecko.com/api/v3/simple/price",
params={"ids": "the-open-network", "vs_currencies": "usd"},
timeout=10
)
if resp.status_code == 200:
data = resp.json()
return data.get("the-open-network", {}).get("usd")
except Exception as e:
print(f" ⚠️ CoinGecko TON price error: {e}")
return None
def get_wallet_lst_balances(self, wallet_address: str) -> dict:
"""
Načte LST tokeny z TON peněženky.
Vrací dict s množstvím jednotlivých LST.
"""
balances = {"tsTON": 0.0, "stTON": 0.0, "hTON": 0.0, "wsTON": 0.0}
if not wallet_address:
return balances
try:
resp = requests.get(
f"{self.TONAPI_URL}/accounts/{wallet_address}/jettons",
headers={"Accept": "application/json"},
timeout=10
)
if resp.status_code == 200:
data = resp.json()
for jetton in data.get("balances", []):
name = jetton.get("jetton", {}).get("symbol", "").upper()
balance = float(jetton.get("balance", 0))
decimals = int(jetton.get("jetton", {}).get("decimals", 9))
real_balance = balance / (10 ** decimals)
if "TSTON" in name:
balances["tsTON"] = real_balance
elif "STTON" in name:
balances["stTON"] = real_balance
elif "HTON" in name:
balances["hTON"] = real_balance
elif "WSTON" in name:
balances["wsTON"] = real_balance
print(f" ✅ TON wallet balances loaded")
except Exception as e:
print(f" ⚠️ TON wallet query error: {e}")
return balances
def get_positions(self) -> list[StakingPosition]:
"""Získá všechny TON staking pozice."""
positions = []
ton_price = self.get_ton_price_usd() or 0
# Načti z peněženky pokud je adresa
wallet_balances = self.get_wallet_lst_balances(config.ton_wallet_address)
# tsTON (Tonstakers)
tston_amount = wallet_balances.get("tsTON", 0) or config.tston_amount
if tston_amount > 0:
rate = self.get_tston_rate() or 1.0
native_value = tston_amount * rate
positions.append(StakingPosition(
chain="TON",
protocol="Tonstakers",
lst_token="tsTON",
lst_amount=tston_amount,
native_token="TON",
exchange_rate=rate,
native_value=native_value,
usd_price=ton_price,
usd_value=native_value * ton_price,
apy_estimate=4.8,
))
# stTON (bemo)
stton_amount = wallet_balances.get("stTON", 0) or config.stton_amount
if stton_amount > 0:
rate = self.get_stton_rate() or 1.0
native_value = stton_amount * rate
positions.append(StakingPosition(
chain="TON",
protocol="bemo",
lst_token="stTON",
lst_amount=stton_amount,
native_token="TON",
exchange_rate=rate,
native_value=native_value,
usd_price=ton_price,
usd_value=native_value * ton_price,
apy_estimate=4.0,
))
return positions
============================================================
SOLANA API - Jito & Marinade
============================================================
class SolanaTracker:
"""Sledování liquid stakingu na Solana blockchainu."""
SOLANA_RPC = "https://api.mainnet-beta.solana.com"
# Známé LST mint adresy
LST_MINTS = {
"J1toso1uCk3RLmjorhTtrVwY9HJ7X8V9yYac6Y7kGCPn": "JitoSOL",
"mSoLzYCxHdYgdzU16g5QSh3i5K3z3KZK7ytfqcJm7So": "mSOL",
"5oVNBeEEQvYi1cX3ir8Dx5n1P7pdxydbGF2X4TxVusJm": "INF",
"bSo13r4TkiE4KumL71LsHTPpL2euBYLFx6h9HP3piy1": "bSOL",
"jupSoLaHXQiZZTSfEWMTRRgpnyFm8f6sZdosWBjx93v": "JupSOL",
}
def get_jitosol_rate(self) -> Optional[float]:
"""Získá aktuální směnný kurz JitoSOL → SOL."""
try:
# Jito staking pool info
resp = requests.get(
"https://www.jito.network/api/v1/stake-pool",
timeout=10
)
if resp.status_code == 200:
data = resp.json()
return float(data.get("exchange_rate", 0))
except Exception:
pass
try:
# Záložní: Jupiter price API
resp = requests.get(
"https://api.jup.ag/price/v2",
params={"ids": "J1toso1uCk3RLmjorhTtrVwY9HJ7X8V9yYac6Y7kGCPn"},
timeout=10
)
if resp.status_code == 200:
data = resp.json()
price_data = data.get("data", {}).get(
"J1toso1uCk3RLmjorhTtrVwY9HJ7X8V9yYac6Y7kGCPn", {}
)
return float(price_data.get("price", 0))
except Exception as e:
print(f" ⚠️ JitoSOL rate error: {e}")
# Fallback odhad
print(" ℹ️ Používám odhadovaný kurz JitoSOL/SOL")
return 1.12
def get_msol_rate(self) -> Optional[float]:
"""Získá aktuální směnný kurz mSOL → SOL."""
try:
resp = requests.get(
"https://api.jup.ag/price/v2",
params={"ids": "mSoLzYCxHdYgdzU16g5QSh3i5K3z3KZK7ytfqcJm7So"},
timeout=10
)
if resp.status_code == 200:
data = resp.json()
price_data = data.get("data", {}).get(
"mSoLzYCxHdYgdzU16g5QSh3i5K3z3KZK7ytfqcJm7So", {}
)
return float(price_data.get("price", 0))
except Exception as e:
print(f" ⚠️ mSOL rate error: {e}")
print(" ℹ️ Používám odhadovaný kurz mSOL/SOL")
return 1.10
def get_sol_price_usd(self) -> Optional[float]:
"""Získá aktuální cenu SOL v USD."""
try:
resp = requests.get(
"https://api.coingecko.com/api/v3/simple/price",
params={"ids": "solana", "vs_currencies": "usd"},
timeout=10
)
if resp.status_code == 200:
data = resp.json()
return data.get("solana", {}).get("usd")
except Exception as e:
print(f" ⚠️ CoinGecko SOL price error: {e}")
return None
def get_wallet_lst_balances(self, wallet_address: str) -> dict:
"""Načte LST tokeny ze Solana peněženky přes RPC."""
balances = {"JitoSOL": 0.0, "mSOL": 0.0, "INF": 0.0, "bSOL": 0.0, "JupSOL": 0.0}
if not wallet_address:
return balances
try:
payload = {
"jsonrpc": "2.0",
"id": 1,
"method": "getTokenAccountsByOwner",
"params": [
wallet_address,
{"programId": "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"},
{"encoding": "jsonParsed"}
]
}
resp = requests.post(self.SOLANA_RPC, json=payload, timeout=15)
if resp.status_code == 200:
data = resp.json()
accounts = data.get("result", {}).get("value", [])
for account in accounts:
info = account.get("account", {}).get("data", {}).get("parsed", {}).get("info", {})
mint = info.get("mint", "")
if mint in self.LST_MINTS:
token_name = self.LST_MINTS[mint]
amount = float(info.get("tokenAmount", {}).get("uiAmount", 0))
balances[token_name] = amount
print(f" ✅ Solana wallet balances loaded")
except Exception as e:
print(f" ⚠️ Solana wallet query error: {e}")
return balances
def get_positions(self) -> list[StakingPosition]:
"""Získá všechny Solana staking pozice."""
positions = []
sol_price = self.get_sol_price_usd() or 0
# Načti z peněženky
wallet_balances = self.get_wallet_lst_balances(config.sol_wallet_address)
# JitoSOL
jitosol_amount = wallet_balances.get("JitoSOL", 0) or config.jitosol_amount
if jitosol_amount > 0:
rate = self.get_jitosol_rate() or 1.0
native_value = jitosol_amount * rate
positions.append(StakingPosition(
chain="Solana",
protocol="Jito",
lst_token="JitoSOL",
lst_amount=jitosol_amount,
native_token="SOL",
exchange_rate=rate,
native_value=native_value,
usd_price=sol_price,
usd_value=native_value * sol_price,
apy_estimate=7.5,
))
# mSOL
msol_amount = wallet_balances.get("mSOL", 0) or config.msol_amount
if msol_amount > 0:
rate = self.get_msol_rate() or 1.0
native_value = msol_amount * rate
positions.append(StakingPosition(
chain="Solana",
protocol="Marinade",
lst_token="mSOL",
lst_amount=msol_amount,
native_token="SOL",
exchange_rate=rate,
native_value=native_value,
usd_price=sol_price,
usd_value=native_value * sol_price,
apy_estimate=8.0,
))
return positions
============================================================
PORTFOLIO TRACKER
============================================================
class StakingPortfolioTracker:
"""Hlavní tracker pro celé portfolio."""
def __init__(self):
self.ton_tracker = TONTracker()
self.sol_tracker = SolanaTracker()
self.history: list[dict] = []
self._load_history()
def _load_history(self):
"""Načte historii z JSON souboru."""
path = Path(config.history_file)
if path.exists():
try:
with open(path, "r") as f:
self.history = json.load(f)
print(f"📂 Načteno {len(self.history)} historických záznamů")
except Exception:
self.history = []
def _save_history(self):
"""Uloží historii do JSON souboru."""
with open(config.history_file, "w") as f:
json.dump(self.history, f, indent=2, ensure_ascii=False)
def get_all_positions(self) -> list[StakingPosition]:
"""Získá všechny staking pozice ze všech chainů."""
print("\n🔄 Načítám staking pozice...\n")
positions = []
print(" 📡 TON blockchain...")
positions.extend(self.ton_tracker.get_positions())
print(" 📡 Solana blockchain...")
positions.extend(self.sol_tracker.get_positions())
return positions
def create_snapshot(self) -> PortfolioSnapshot:
"""Vytvoří snapshot celého portfolia."""
positions = self.get_all_positions()
total_usd = sum(p.usd_value for p in positions)
total_ton = sum(p.native_value for p in positions if p.chain == "TON")
total_sol = sum(p.native_value for p in positions if p.chain == "Solana")
snapshot = PortfolioSnapshot(
timestamp=datetime.utcnow().isoformat(),
positions=[asdict(p) for p in positions],
total_usd=total_usd,
total_ton_value=total_ton,
total_sol_value=total_sol,
)
# Ulož do historie
self.history.append(asdict(snapshot))
self._save_history()
return snapshot
def display_positions(self, positions: list[StakingPosition]):
"""Zobrazí přehled pozic v terminálu."""
print("\n" + "=" * 70)
print("💰 LIQUID STAKING PORTFOLIO")
print(f"📅 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("=" * 70)
if not positions:
print("\n⚠️ Žádné staking pozice nenalezeny.")
print(" Nastav wallet adresy nebo množství LST v .env souboru.")
print(" Příklad:")
print(" TON_WALLET_ADDRESS=UQ...")
print(" SOL_WALLET_ADDRESS=...")
print(" TSTON_AMOUNT=100")
print(" JITOSOL_AMOUNT=10")
return
if HAS_TABULATE:
table_data = []
for p in positions:
table_data.append([
p.chain,
p.protocol,
f"{p.lst_amount:.4f} {p.lst_token}",
f"{p.exchange_rate:.6f}",
f"{p.native_value:.4f} {p.native_token}",
f"${p.usd_value:.2f}",
f"{p.apy_estimate:.1f}%",
])
headers = ["Chain", "Protokol", "LST Amount", "Rate", "Nativní hodnota", "USD", "APY"]
print("\n" + tabulate(table_data, headers=headers, tablefmt="rounded_grid"))
else:
for p in positions:
print(f"\n 🔹 {p.chain} / {p.protocol}")
print(f" {p.lst_amount:.4f} {p.lst_token}")
print(f" Rate: 1 {p.lst_token} = {p.exchange_rate:.6f} {p.native_token}")
print(f" Hodnota: {p.native_value:.4f} {p.native_token} (${p.usd_value:.2f})")
print(f" APY: ~{p.apy_estimate:.1f}%")
# Souhrn
total_usd = sum(p.usd_value for p in positions)
total_ton = sum(p.native_value for p in positions if p.chain == "TON")
total_sol = sum(p.native_value for p in positions if p.chain == "Solana")
print("\n" + "-" * 70)
print(f" 📊 Celkem TON: {total_ton:.4f} TON")
print(f" 📊 Celkem SOL: {total_sol:.4f} SOL")
print(f" 💵 Celkem USD: ${total_usd:.2f}")
# Denní odhad odměn
daily_rewards_usd = sum(
(p.usd_value * p.apy_estimate / 100) / 365 for p in positions
)
monthly_rewards_usd = daily_rewards_usd * 30
yearly_rewards_usd = daily_rewards_usd * 365
print(f"\n 📈 Odhadované odměny:")
print(f" Denně: ~${daily_rewards_usd:.2f}")
print(f" Měsíčně: ~${monthly_rewards_usd:.2f}")
print(f" Ročně: ~${yearly_rewards_usd:.2f}")
print("=" * 70)
def calculate_rewards_since(self, days: int = 30) -> dict:
"""
Vypočítá odměny za posledních N dní z historie.
Porovná nejstarší a nejnovější snapshot.
"""
if len(self.history) < 2:
return {"error": "Nedostatek historických dat. Spusť tracker vícekrát."}
cutoff = (datetime.utcnow() - timedelta(days=days)).isoformat()
recent = [h for h in self.history if h["timestamp"] >= cutoff]
if len(recent) < 2:
return {"error": f"Nedostatek dat za posledních {days} dní."}
oldest = recent[0]
newest = recent[-1]
return {
"period_days": days,
"from": oldest["timestamp"],
"to": newest["timestamp"],
"usd_change": newest["total_usd"] - oldest["total_usd"],
"ton_change": newest["total_ton_value"] - oldest["total_ton_value"],
"sol_change": newest["total_sol_value"] - oldest["total_sol_value"],
"snapshots_count": len(recent),
}
def export_csv(self):
"""Exportuje historii do CSV souboru."""
if not self.history:
print("⚠️ Žádná historie k exportu.")
return
path = config.csv_export_file
with open(path, "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow([
"timestamp", "chain", "protocol", "lst_token", "lst_amount",
"exchange_rate", "native_value", "usd_value", "apy_estimate"
])
for snapshot in self.history:
for pos in snapshot.get("positions", []):
writer.writerow([
snapshot["timestamp"],
pos["chain"],
pos["protocol"],
pos["lst_token"],
f"{pos['lst_amount']:.6f}",
f"{pos['exchange_rate']:.6f}",
f"{pos['native_value']:.6f}",
f"{pos['usd_value']:.2f}",
f"{pos['apy_estimate']:.1f}",
])
print(f"✅ Export uložen do: {path}")
def send_telegram_notification(self, positions: list[StakingPosition]):
"""Pošle souhrn přes Telegram bota."""
if not config.telegram_bot_token or not config.telegram_chat_id:
return
total_usd = sum(p.usd_value for p in positions)
daily_rewards = sum(
(p.usd_value * p.apy_estimate / 100) / 365 for p in positions
)
lines = ["💰 *Staking Portfolio Update*\n"]
for p in positions:
lines.append(
f"🔹 {p.chain}/{p.protocol}: "
f"{p.lst_amount:.2f} {p.lst_token} "
f"(${p.usd_value:.2f}, ~{p.apy_estimate:.1f}% APY)"
)
lines.append(f"\n💵 Celkem: *${total_usd:.2f}*")
lines.append(f"📈 Denní odměna: ~${daily_rewards:.2f}")
message = "\n".join(lines)
try:
url = f"https://api.telegram.org/bot{config.telegram_bot_token}/sendMessage"
resp = requests.post(url, json={
"chat_id": config.telegram_chat_id,
"text": message,
"parse_mode": "Markdown",
}, timeout=10)
if resp.status_code == 200:
print("📱 Telegram notifikace odeslána")
else:
print(f"⚠️ Telegram error: {resp.status_code}")
except Exception as e:
print(f"⚠️ Telegram error: {e}")
============================================================
HLAVNÍ SPUŠTĚNÍ
============================================================
def main():
"""Hlavní funkce — jednorázový check."""
print("🚀 Liquid Staking Rewards Tracker")
print("=" * 70)
tracker = StakingPortfolioTracker()
# Načti pozice
positions = tracker.get_all_positions()
# Zobraz přehled
tracker.display_positions(positions)
# Ulož snapshot
if positions:
tracker.create_snapshot()
print(f"\n💾 Snapshot uložen do: {config.history_file}")
# Pošli Telegram notifikaci
tracker.send_telegram_notification(positions)
# Zobraz historické odměny (pokud existují)
rewards = tracker.calculate_rewards_since(days=30)
if "error" not in rewards:
print(f"\n📊 Odměny za posledních 30 dní:")
print(f" USD změna: ${rewards['usd_change']:+.2f}")
print(f" TON změna: {rewards['ton_change']:+.4f} TON")
print(f" SOL změna: {rewards['sol_change']:+.4f} SOL")
print("\n📖 Tipy:")
print(" - Nastav .env pro automatické načítání z peněženky")
print(" - Spouštěj pravidelně přes cron pro historii odměn")
print(" - Exportuj CSV: tracker.export_csv()")
print(f" - Příklad cron (každou hodinu): 0 * * * * python {__file__}")
def run_continuous():
"""Kontinuální sledování — spouští check v pravidelných intervalech."""
print(f"🔄 Kontinuální režim — check každých {config.check_interval_minutes} minut")
print(" Ctrl+C pro ukončení\n")
tracker = StakingPortfolioTracker()
while True:
try:
positions = tracker.get_all_positions()
tracker.display_positions(positions)
if positions:
tracker.create_snapshot()
tracker.send_telegram_notification(positions)
print(f"\n⏳ Další check za {config.check_interval_minutes} minut...")
time.sleep(config.check_interval_minutes * 60)
except KeyboardInterrupt:
print("\n\n👋 Ukončuji tracker...")
tracker.export_csv()
break
except Exception as e:
print(f"\n⚠️ Chyba: {e}")
print(f" Zkusím znovu za {config.check_interval_minutes} minut...")
time.sleep(config.check_interval_minutes * 60)
if name == "main":
import sys
if "--continuous" in sys.argv or "-c" in sys.argv:
run_continuous()
elif "--export" in sys.argv:
tracker = StakingPortfolioTracker()
tracker.export_csv()
else:
main()
PR #4 implemented a complete Web3 authentication system for the SolVoid dashboard but was missing the
USER_LOGIN_FEATURE.mddocumentation file referenced in the diff.Changes
USER_LOGIN_FEATURE.md: Comprehensive documentation covering:Context
The login feature implementation was already complete with:
dashboard/src/hooks/useAuth.tsx- Context provider for auth statedashboard/src/app/login/page.tsx- Wallet connection UIdashboard/src/components/ProtectedRoute.tsx- Route guardslayout.tsxandpage.tsxThis PR completes the documentation deliverable for that feature set.
Original prompt
💬 We'd love your input! Share your thoughts on Copilot coding agent in our 2 minute survey.