"""Top-wallet pool builder. Identifies and maintains the pool of wallets we follow. Strategy: 1. Fetch top events by 24h volume (1 API call) 2. For each market (parallel, semaphore-limited), fetch top holders 3. Deduplicate wallet addresses across all markets 4. For each candidate (parallel), fetch /positions to compute PnL/trades/categories 5. Apply health score; keep top N by score; persist to DB """ import asyncio import logging from datetime import datetime from typing import Dict, List, Set from src.config import get_settings from src.db.database import CopyTraderDatabase from src.services.data_api import DataAPIClient, GammaAPIClient logger = logging.getLogger(__name__) def compute_health_score(pnl_30d: float, total_pnl: float, trades: int, categories: int) -> float: """0-100 score combining PnL magnitude, trade count, category diversity.""" if trades < 10 or pnl_30d <= 0: return 0.0 pnl_score = min(pnl_30d / 10000.0, 1.0) * 50 trade_score = min(trades / 100.0, 1.0) * 25 diversity_score = min(categories / 5.0, 1.0) * 25 return pnl_score + trade_score + diversity_score class WalletPoolBuilder: """Builds and refreshes the target wallet pool.""" def __init__(self, db: CopyTraderDatabase): self.db = db self.settings = get_settings() self.gamma = GammaAPIClient() self.data = DataAPIClient() async def build_pool_async(self, max_markets: int = 200) -> List[dict]: """Async build with parallelism + per-call timeout. Returns list of wallet dicts (already filtered by min_pnl/trades/categories). """ settings = self.settings concurrency = settings.wallet_pool_concurrency timeout = settings.wallet_pool_request_timeout progress_every = settings.wallet_pool_progress_every logger.info( f"[pool] Building top wallet pool from up to {max_markets} markets " f"(concurrency={concurrency}, request_timeout={timeout}s)" ) # Phase 1: get top events (single call) try: events = await asyncio.to_thread( self.gamma.get_active_events_by_volume, max_markets ) except Exception as e: logger.error(f"[pool] Gamma API failed: {e}") return [] condition_ids: Set[str] = set() for ev in events: for m in (ev.get("markets") or []): cid = m.get("conditionId") if cid and not m.get("closed"): condition_ids.add(cid) logger.info(f"[pool] Found {len(condition_ids)} candidate markets") # Phase 2: parallel holder fetch semaphore = asyncio.Semaphore(concurrency) empty_streak = 0 async def fetch_holders(cid: str) -> List[dict]: nonlocal empty_streak async with semaphore: if empty_streak >= settings.wallet_pool_backoff_emails: logger.info(f"[pool] {empty_streak} consecutive empty responses, sleeping 15s") await asyncio.sleep(15) empty_streak = 0 try: result = await asyncio.to_thread( self.data.get_top_holders, cid, 30, timeout ) if isinstance(result, list) and len(result) == 0: empty_streak += 1 else: empty_streak = 0 return result except Exception as e: logger.debug(f"[pool] holders failed for {cid[:10]}: {e}") empty_streak += 1 return [] market_tasks = [fetch_holders(cid) for cid in condition_ids] candidates: Dict[str, int] = {} completed = 0 total = len(market_tasks) for coro in asyncio.as_completed(market_tasks): holders = await coro completed += 1 # Response shape: [{"token": "...", "positions": [...]}] for token_wrapper in holders: for pos in (token_wrapper.get("positions") or []): addr = pos.get("proxyWallet") if addr: candidates[addr] = candidates.get(addr, 0) + 1 if completed % progress_every == 0 or completed == total: logger.info( f"[pool] holders: scanned {completed}/{total} markets " f"({len(candidates)} unique wallets so far)" ) if not candidates: logger.warning("[pool] No candidate wallets found") return [] logger.info(f"[pool] Found {len(candidates)} candidate wallets") # Phase 3: parallel position profile fetch async def fetch_wallet(addr: str) -> dict: async with semaphore: try: positions = await asyncio.to_thread( self.data.get_positions, addr, 500, timeout * 2 ) except Exception as e: logger.debug(f"[pool] positions failed for {addr[:10]}: {e}") positions = [] cash_pnl = sum(float(p.get("cashPnl") or 0) for p in positions) realized_pnl = sum( float(p.get("realizedPnl") or 0) for p in positions ) total_pnl = cash_pnl + realized_pnl trades = len(positions) categories = len({ p.get("eventSlug") or p.get("slug") for p in positions if p.get("eventSlug") or p.get("slug") }) score = compute_health_score( pnl_30d=total_pnl, total_pnl=total_pnl, trades=trades, categories=categories, ) return { "address": addr, "source": "top_holders", "pnl_30d_usd": total_pnl, "pnl_total_usd": total_pnl, "trades_count": trades, "categories_count": categories, "health_score": score, "credibility": settings.bayesian_prior_skill, "last_seen_at": datetime.now().isoformat(), } wallet_tasks = [fetch_wallet(addr) for addr in candidates] wallets: List[dict] = [] completed = 0 total = len(wallet_tasks) for coro in asyncio.as_completed(wallet_tasks): wallet = await coro completed += 1 if ( wallet["pnl_total_usd"] >= settings.wallet_pnl_min_usd and wallet["trades_count"] >= settings.wallet_min_trades and wallet["categories_count"] >= settings.wallet_min_categories ): wallets.append(wallet) if completed % progress_every == 0 or completed == total: logger.info( f"[pool] profiles: scanned {completed}/{total} wallets " f"({len(wallets)} passed filter so far)" ) wallets.sort(key=lambda w: w["health_score"], reverse=True) result = wallets[:settings.wallet_pool_size] logger.info( f"[pool] Selected top {len(result)} wallets (from {len(wallets)} candidates)" ) return result def build_pool(self, max_markets: int = 200) -> List[dict]: """Sync wrapper for CLI usage.""" return asyncio.run(self.build_pool_async(max_markets)) def refresh(self, max_markets: int = 200) -> int: """Build pool and persist. Returns count. CLI/sync only — call build_pool_async() directly from async contexts.""" try: asyncio.get_running_loop() logger.warning( "[pool] refresh() called from async context; " "use build_pool_async() instead" ) return 0 except RuntimeError: pass pool = asyncio.run(self.build_pool_async(max_markets)) now = datetime.now().isoformat() for w in pool: w.setdefault("added_at", now) self.db.upsert_wallet_target(w) return len(pool)