import os import sys import time import requests import json from dotenv import load_dotenv from py_clob_client.client import ClobClient from py_clob_client.clob_types import OrderArgs, OrderType from py_clob_client.order_builder.constants import BUY, SELL from py_clob_client.constants import POLYGON # Load environment variables load_dotenv() # Configuration PRIVATE_KEY = os.getenv("PRIVATE_KEY") RPC_URL = os.getenv("RPC_URL", "https://polygon-rpc.com") # Safety Constants MAX_SLIPPAGE = 0.05 MAX_SPEND_PER_ORDER = 5.0 DRY_RUN = False # WARNING: Real trading enabled! Orders will be placed on Polymarket. # Global Caches MARKET_TICK_SIZE_CACHE = {} # token_id -> float (e.g. 0.001) def get_market_tick_size(token_id): """ Advanced precision detection: 1. Read the order book (bids/asks) to see what precision OTHERS are using. 2. Fallback to CLOB/Gamma metadata. 3. Fallback to 0.001. """ if token_id in MARKET_TICK_SIZE_CACHE: return MARKET_TICK_SIZE_CACHE[token_id] try: # Step 1: Read the Order Book book_url = f"https://clob.polymarket.com/book?token_id={token_id}" book_resp = requests.get(book_url, timeout=5) if book_resp.status_code == 200: data = book_resp.json() prices = [b['price'] for b in data.get('bids', [])] + [a['price'] for a in data.get('asks', [])] if prices: max_d = 0 for p in prices: if '.' in str(p): # Count actual decimals in price string d = len(str(p).split('.')[1].rstrip('0')) max_d = max(max_d, d) tick_size = 10**(-max_d) if max_d > 0 else 1.0 # Sanity check: cap at 0.1 and min at 0.0001 tick_size = max(0.0001, min(0.1, tick_size)) MARKET_TICK_SIZE_CACHE[token_id] = tick_size return tick_size # Step 2: Fallback to Metadata API url = f"https://clob.polymarket.com/markets/{token_id}" resp = requests.get(url, timeout=5) if resp.status_code == 200: tick_size = float(resp.json().get('minimum_tick_size', 0.001)) MARKET_TICK_SIZE_CACHE[token_id] = tick_size return tick_size # Fallback to Gamma API gamma_url = f"https://gamma-api.polymarket.com/markets?token_id={token_id}" g_resp = requests.get(gamma_url, timeout=5) if g_resp.status_code == 200: data = g_resp.json() if isinstance(data, list) and len(data) > 0: tick_size = float(data[0].get('minPriceIncrement', 0.001)) MARKET_TICK_SIZE_CACHE[token_id] = tick_size return tick_size except Exception: pass return 0.001 # Robust default def round_to_tick(value, tick_size): """Round a value to the nearest multiple of the tick_size.""" return round(round(value / tick_size) * tick_size, 8) def format_price(value, tick_size): """Format a price value as a string matching its tick size precision.""" import math try: precision = max(0, int(-math.log10(tick_size))) return f"{float(value):.{precision}f}" except Exception: return f"{float(value):.3f}" def init_client(): """Initialize the Polymarket CLOB client.""" if not PRIVATE_KEY: print("Error: PRIVATE_KEY not found in .env file.") print("Please copy .env.example to .env and fill in your key.") sys.exit(1) try: # funder should be the PROXY address, not the base signer address # For your account, this is 0x1234561890123456189012345618901234561890 proxy_addr = "0x1234561890123456189012345618901234561890" client = ClobClient( host="https://clob.polymarket.com", key=PRIVATE_KEY, chain_id=137, signature_type=1, funder=proxy_addr, ) # --- Level 2 Authentication (Required for placing/viewing orders) --- try: print("🔑 Deriving API credentials for order placement...") creds = client.create_or_derive_api_creds() client.set_api_creds(creds) print("[OK] Level 2 Authentication established.") except Exception as l2_err: print(f"[Note] Level 2 Auth (API Keys) not established: {l2_err}") print("[Info] You can still search markets and check positions, but placing orders may fail.") print("[OK] Client initialized successfully.") return client except Exception as e: print(f"[Error] Failed to initialize client: {e}") sys.exit(1) def main(): print("=== Polymarket Portfolio Manager (Bulk Seller) ===") # 1. Initialize client = init_client() # 2. Main Menu print("\nSelect Mode:") print("1. Check My Positions (Data API)") print("2. Bulk Sell All (Strict Logic)") print("3. Search Markets (Keyword)") choice = input("Choice (1/2/3): ").strip() if choice == "1": # Check Positions using Data API try: from eth_account import Account account = Account.from_key(PRIVATE_KEY) base_address = account.address # Polymarket uses proxy wallets for trading # The proxy address is different from the base address derived from private key # For Google login accounts, this is managed by Privy # You can find your proxy address at: https://polymarket.com/@YourUsername proxy_address = "0x1234561890123456189012345618901234561890" # Your actual trading address print(f"\n[Checking positions...]") print(f"Base address: {base_address[:6]}...{base_address[-4:]}") print(f"Proxy address (trading): {proxy_address[:6]}...{proxy_address[-4:]}") user_address = proxy_address d_url = "https://data-api.polymarket.com/positions" d_resp = requests.get(d_url, params={"user": user_address, "limit": "20", "sortBy": "CURRENT"}) if d_resp.status_code == 200: positions = d_resp.json() if not positions: print("No active positions found.") else: print(f"\n[OK] Active Positions ({len(positions)}):") total_value = 0 total_cost = 0 for p in positions: title = p.get('title', 'Unknown Market') outcome = p.get('outcome', '?') size = p.get('size', '0') # Financial data current_value = float(p.get('currentValue', 0)) initial_value = float(p.get('initialValue', 0)) pnl_dollar = float(p.get('cashPnl', 0)) pnl_percent = float(p.get('percentPnl', 0)) total_value += current_value total_cost += initial_value # Format PnL with color indicator pnl_sign = "+" if pnl_dollar >= 0 else "" print(f"\n [{outcome}] {title}") print(f" Size: {float(size):.2f} shares") print(f" Cost: ${initial_value:.2f} | Current: ${current_value:.2f}") print(f" PnL: {pnl_sign}${pnl_dollar:.2f} ({pnl_sign}{pnl_percent:.2f}%)") # Summary total_pnl = total_value - total_cost total_pnl_pct = (total_pnl / total_cost * 100) if total_cost > 0 else 0 pnl_sign = "+" if total_pnl >= 0 else "" print(f"\n{'='*60}") print(f"TOTAL: ${total_value:.2f} | Cost: ${total_cost:.2f}") print(f"Overall PnL: {pnl_sign}${total_pnl:.2f} ({pnl_sign}{total_pnl_pct:.2f}%)") print(f"{'='*60}") else: print(f"[Error] Failed to fetch positions (Status {d_resp.status_code}): {d_resp.text[:200]}") # --- Check Open Orders (CLOB) --- print(f"\n[Checking Open Orders...]") try: # Get all open orders for this account open_orders = client.get_orders() if not open_orders or len(open_orders) == 0: print("No open orders found.") else: print(f"\n[OK] Open Orders ({len(open_orders)}):") for o in open_orders: # Parse order details price = o.get('price', '?') size = o.get('size', '?') side = o.get('side', 'BUY') token_id = o.get('asset_id', o.get('token_id', '?')) status = o.get('status', '?') # Only show if it's actually open/live if status in ['LIVE', 'live', 'open', 'OPEN', '?']: print(f" - {side} {size} @ ${price} | Status: {status}") print(f" Token: {str(token_id)[:15]}...") except Exception as e: print(f"[Note] Could not fetch open orders: {e}") print("[Info] This is normal if you haven't set up API credentials yet.") except Exception as e: print(f"[Error] {e}") return elif choice == "2": # Bulk Sell All Positions (Tiered Logic) try: proxy_address = "0x1234561890123456189012345618901234561890" # Ask for Musk profit lock percentage try: musk_pct_str = input("\nEnter profit lock percentage for Elon/Twitter markets (e.g., 9 for 9%): ").strip() musk_lock_factor = float(musk_pct_str) / 100.0 except ValueError: print("Invalid input. Defaulting to 8% (0.08).") musk_lock_factor = 0.08 print(f"\n[Fetching positions to sell...]") # Get positions d_url = "https://data-api.polymarket.com/positions" d_resp = requests.get(d_url, params={"user": proxy_address, "limit": "50"}) if d_resp.status_code != 200: print(f"[Error] Could not fetch positions: {d_resp.text[:200]}") return positions = d_resp.json() if not positions: print("No positions to sell.") return # --- Check Existing Open Orders to avoid duplicates/updates --- print("[Fetching existing open orders...]") open_sell_metadata = {} # token_id -> list of {'id':..., 'price':..., 'size':...} try: open_orders = client.get_orders() for o in open_orders: raw_token = str(o.get('asset_id') or o.get('token_id')) o_token = raw_token.lower() if raw_token else "" o_side = str(o.get('side', '')).upper() o_price = float(o.get('price', 0)) o_size = float(o.get('original_size', o.get('size', 0))) if o_token and o_side == 'SELL': if o_token not in open_sell_metadata: open_sell_metadata[o_token] = [] open_sell_metadata[o_token].append({'id': o.get('id'), 'price': o_price, 'size': o_size}) print(f"[Info] Tracked existing sell orders for {len(open_sell_metadata)} tokens.") except Exception as e: print(f"[Note] Could not fetch open orders to check duplicates: {e}") # Filter out and categorize positions sell_positions_data = [] # List of dicts with all info orders_to_cancel = [] # Global list of order IDs to cancel for p in positions: title = p.get('title', 'Unknown') title_lower = title.lower() cost_basis = float(p.get('avgPrice', 0)) raw_token = str(p.get('asset', '')) token_id = raw_token.lower() if raw_token else "" total_size = float(p.get('size', 0)) outcome = p.get('outcome', '?') if not token_id or total_size <= 0: continue # A. Determine Base Profit Lock Pct is_elon = ('elon' in title_lower or 'musk' in title_lower) and ('tweet' in title_lower or 'twitter' in title_lower) lock_pct = musk_lock_factor if is_elon else 0.9 # B. Fetch Current Market Price (Best Bid) & Tick Size best_bid = 0 tick_size = get_market_tick_size(token_id) try: price_url = "https://clob.polymarket.com/price" price_resp = requests.get(price_url, params={"token_id": token_id, "side": "sell"}) if price_resp.status_code == 200: best_bid = float(price_resp.json().get('price', 0)) else: print(f"[Skip] {title[:40]}... - Could not fetch market price") continue except Exception as e: print(f"[Skip] {title[:40]}... - Error fetching price: {e}") continue # C. Calculate SMART SELL PRICE max_profit_potential = 1.0 - cost_basis if is_elon: # NEW DYNAMIC FORMULA for Twitter: Best Bid + (Remaining Spread * LockPct) # This replaces Rescue and Chase logic. remaining_spread = 1.0 - best_bid sell_price = round_to_tick(best_bid + (remaining_spread * lock_pct), tick_size) else: # PRESERVED STANDARD FORMULA: Cost + (Potential Spread * 90%) sell_price = round_to_tick(cost_basis + (max_profit_potential * 0.9), tick_size) # Guard rails: Cap at 0.99 if sell_price > 0.99: sell_price = 0.99 # Double-check rounding to prevent float artifacts sell_price = round_to_tick(sell_price, tick_size) # D. Check Existing Orders against this SMART price existing_orders = open_sell_metadata.get(token_id, []) already_listed_correct_price = 0 for o in existing_orders: # Strict comparison with negligible float epsilon (1e-9) if abs(o['price'] - sell_price) > 1e-9: # ONLY cancel Musk orders to update them. if is_elon: orders_to_cancel.append(o['id']) old_p_str = format_price(o['price'], tick_size) new_p_str = format_price(sell_price, tick_size) print(f"[Update] {title[:40]}...: ${old_p_str} -> ${new_p_str} (Tick: {tick_size})") else: # For non-Musk, if price is different, we just ignore it and don't list MORE. already_listed_correct_price += o['size'] else: already_listed_correct_price += o['size'] remaining = total_size - already_listed_correct_price # E. If everything is already listed at this exact smart price, skip it. if remaining < 0.01: if already_listed_correct_price > 0: msg = f" (Already listed correctly at Tick: {tick_size})" print(f"[Skip] {title[:50]}...{msg}") continue # F. Record data for preview profit_locked = sell_price - cost_basis total_order_profit = profit_locked * remaining sell_positions_data.append({ 'title': title, 'outcome': outcome, 'token_id': token_id, 'size': remaining, 'price': sell_price, 'best_bid': best_bid, 'cost': cost_basis, 'max_profit': max_profit_potential, 'locked_profit': profit_locked, 'total_profit': total_order_profit, 'is_elon': is_elon, 'already_listed': already_listed_correct_price, 'lock_pct': lock_pct }) if not sell_positions_data: print("\nNo positions need updating or selling.") return print(f"\n[Preview: {len(sell_positions_data)} positions to process]") for d in sell_positions_data: print(f"\n {d['title']}") if d['is_elon']: print(f" [Elon Tweet Market - Dynamic Spread Strategy]") print(f" (Calculation: Best Bid + (1.0 - Best Bid) * {int(d['lock_pct']*100)}%)") else: print(f" [Standard Market - 90% Spread Lock]") if d['already_listed'] > 0: print(f" [Partial] Already listed: {d['already_listed']:.2f} | Listing remaining: {d['size']:.2f}") print(f" Outcome: {d['outcome']} | Size: {d['size']:.2f}") tick_size = MARKET_TICK_SIZE_CACHE.get(d['token_id'], 0.001) cost_str = format_price(d['cost'], tick_size) bid_str = format_price(d['best_bid'], tick_size) spr_str = format_price(d['max_profit'], tick_size) tar_str = format_price(d['price'], tick_size) print(f" Cost: ${cost_str} | Best Bid: ${bid_str} | Spread: ${spr_str} | Tick: {tick_size}") print(f" Target Sell: ${tar_str} (Profit: ${d['locked_profit']:.4f} per share)") print(f" Expected Profit (this batch): ${d['total_profit']:.2f}") # Final Confirmation logic total_batch_profit = sum(o['total_profit'] for o in sell_positions_data) print(f"\n{'='*60}") print(f"Ready to place {len(sell_positions_data)} SELL orders") if orders_to_cancel: print(f"Orders to CANCEL: {len(orders_to_cancel)} (Replacing outdated prices)") print(f"Total Expected Batch Profit: ${total_batch_profit:.2f}") if DRY_RUN: print("[DRY RUN MODE] No real actions taken") else: print("[REAL TRADING] Money WILL be at risk!") print(f"{'='*60}") confirm = input("\nConfirm bulk sell and price updates? (yes/no): ").strip().lower() if confirm != 'yes': print("Cancelled.") return # Step 1: Cancel outdated orders if orders_to_cancel: print(f"\n[Step 1: Cancelling {len(orders_to_cancel)} outdated orders...]") for oid in orders_to_cancel: try: if not DRY_RUN: client.cancel(oid) print(f" - Cancelled: {oid[:15]}...") else: print(f" - [DRY RUN] Would cancel: {oid[:15]}...") except Exception as ce: print(f" - [Error] Failed to cancel {oid[:15]}...: {ce}") if not DRY_RUN: time.sleep(1) # Step 2: Place orders print(f"\n[Step 2: Placing new orders...]") success_count = 0 for order in sell_positions_data: try: if not DRY_RUN: resp = client.create_and_post_order( OrderArgs( price=order['price'], size=order['size'], side=SELL, token_id=order['token_id'], ) ) print(f"[OK] {order['title'][:50]}... - Order ID: {resp.get('orderID', 'N/A')}") success_count += 1 time.sleep(0.5) else: print(f"[DRY RUN] Would sell {order['size']:.2f} @ ${order['price']:.4f}") success_count += 1 except Exception as e: print(f"[Error] {order['title'][:50]}... - {e}") print(f"\n[Complete] {success_count}/{len(sell_positions_data)} orders placed successfully") except Exception as e: print(f"[Error] {e}") return elif choice == "3": # Search Markets by Keyword try: query = input("\nEnter keyword to search (e.g., elon musk): ").strip() if not query: print("Keyword cannot be empty.") return print(f"\n[Searching for '{query}'...]") # Official search endpoint that works better for keywords g_url = "https://gamma-api.polymarket.com/public-search" # Increase limit_per_type to ensure active markets aren't buried under old/closed ones g_resp = requests.get(g_url, params={"q": query, "limit_per_type": 100}) if g_resp.status_code == 200: data = g_resp.json() events = data.get('events') or [] if not events: print("No matching markets found.") else: found_count = 0 for event in events: # Each event can have multiple markets event_title = event.get('title', 'Unknown Event') markets = event.get('markets') or [] for m in markets: if not m.get('active') or m.get('closed'): continue found_count += 1 title = m.get('question', event_title) market_id = m.get('id', 'N/A') print(f"\n > {title}") print(f" Market ID: {market_id}") # Outcomes and Token IDs # API returns them as JSON strings or lists outcomes_raw = m.get('outcomes', '[]') tokens_raw = m.get('clobTokenIds', '[]') try: outcomes = json.loads(outcomes_raw) if isinstance(outcomes_raw, str) else outcomes_raw token_ids = json.loads(tokens_raw) if isinstance(tokens_raw, str) else tokens_raw if outcomes and token_ids and len(outcomes) == len(token_ids): for i in range(len(outcomes)): print(f" - {outcomes[i]}: {token_ids[i]}") else: print(" - No token IDs found.") except: print(" - Error parsing token IDs.") if found_count == 0: print("No active markets found for this keyword.") else: print(f"\n[OK] Found {found_count} active markets.") else: print(f"[Error] Search failed (Status {g_resp.status_code})") except Exception as e: print(f"[Error] {e}") return else: print("Invalid choice.") return if __name__ == "__main__": main()