""" starter_1_copy_trading.py — Copy-trading bot starter. Mirrors trades from one or more target wallets in real time. Uses the Data API (/activity polling) for detection — replace with the User WS once you have it working. Customize: TARGET_WALLETS, COPY_RATIO, MIN_TRADE_USD, MAX_TIME_TO_RES_HOURS. Run: python main.py Defaults to DRY_RUN=true. Flip to false in .env to place real orders. """ from __future__ import annotations import asyncio import hashlib import os import signal import time from dataclasses import dataclass from typing import Optional from dotenv import load_dotenv # Local shared modules (copy from scripts/common/ in this skill) from common import ( JsonlLogger, OrderIntent, RiskLayer, cancel_all, fetch_user_activity, healthcheck, make_client, place_fok_with_retry, ) # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- load_dotenv() PRIVATE_KEY = os.environ["POLYMARKET_PRIVATE_KEY"] FUNDER = os.environ["POLYMARKET_FUNDER_ADDRESS"] SIGNATURE_TYPE = int(os.environ.get("POLYMARKET_SIGNATURE_TYPE", "2")) TARGET_WALLETS = [w.strip().lower() for w in os.environ["TARGET_WALLETS"].split(",") if w.strip()] COPY_RATIO = float(os.environ.get("COPY_RATIO", "0.05")) MIN_TRADE_USD = float(os.environ.get("MIN_TRADE_USD", "5")) MAX_TIME_TO_RES_H = int(os.environ.get("MAX_TIME_TO_RES_HOURS", "24")) POLL_INTERVAL_S = float(os.environ.get("POLL_INTERVAL_S", "2.0")) DRY_RUN = os.environ.get("DRY_RUN", "true").lower() == "true" # --------------------------------------------------------------------------- # Banner # --------------------------------------------------------------------------- BANNER = r""" ______ __ ____ __ / ____/___ ____ __ / /_ ____ ____ / __ )____ / /_ / / / __ \/ __ \/ / / __/ / __ \/ __ \/ __ / __ \/ __/ / /___/ /_/ / /_/ / /_/ /_ / /_/ / /_/ / /_/ / /_/ / /_ \____/\____/ .___/\__/\__/ \____/\____/_____/\____/\__/ /_/ Polymarket Copy Trading Bot """ def print_banner(): print(BANNER) mode = "DRY RUN (no real orders)" if DRY_RUN else "LIVE TRADING" print(f" Mode: {mode}") print(f" Watching: {len(TARGET_WALLETS)} wallet(s)") print(f" Copy ratio: {COPY_RATIO * 100:.1f}%") print(f" Min trade: ${MIN_TRADE_USD}, Max time-to-res: {MAX_TIME_TO_RES_H}h") print() # --------------------------------------------------------------------------- # Pipeline # --------------------------------------------------------------------------- @dataclass class TradeSignal: wallet: str tx_hash: str market_condition: str token_id: str side: str # BUY / SELL size: float # shares price: float title: str end_date: Optional[int] # unix seconds def trade_signal_from_activity(row: dict) -> Optional[TradeSignal]: if row.get("type", "TRADE") != "TRADE": return None return TradeSignal( wallet=row["proxyWallet"].lower(), tx_hash=row.get("transactionHash", ""), market_condition=row.get("conditionId", ""), token_id=str(row.get("asset", "")), side=row.get("side", "BUY"), size=float(row.get("size", 0)), price=float(row.get("price", 0)), title=row.get("title", ""), end_date=row.get("endDate"), ) def signal_hash(s: TradeSignal) -> str: return hashlib.sha256( f"{s.tx_hash}:{s.token_id}:{s.side}".encode() ).hexdigest()[:16] async def poll_wallet(wallet: str, seen_hashes: set[str], log: JsonlLogger): """Poll one wallet's activity, yield new TradeSignals.""" try: rows = await fetch_user_activity(wallet, activity_type="TRADE", limit=20) except Exception as e: log.log("data_error", wallet=wallet, error=str(e)) return [] new_signals = [] for row in rows: sig = trade_signal_from_activity(row) if sig is None: continue h = signal_hash(sig) if h in seen_hashes: continue seen_hashes.add(h) sig._hash = h # type: ignore[attr-defined] new_signals.append(sig) return new_signals async def event_loop(client, risk: RiskLayer, log: JsonlLogger, stop: asyncio.Event): seen: set[str] = set() log.log("start", target_wallets=TARGET_WALLETS, copy_ratio=COPY_RATIO, dry_run=DRY_RUN) while not stop.is_set(): for wallet in TARGET_WALLETS: signals = await poll_wallet(wallet, seen, log) for sig in signals: await handle_signal(sig, client, risk, log) try: await asyncio.wait_for(stop.wait(), timeout=POLL_INTERVAL_S) except asyncio.TimeoutError: pass async def handle_signal(sig: TradeSignal, client, risk: RiskLayer, log: JsonlLogger): log.log("signal_seen", wallet=sig.wallet, market=sig.market_condition, side=sig.side, size=sig.size, price=sig.price, title=sig.title) notional = sig.size * sig.price if notional < MIN_TRADE_USD: log.log("skip", reason="below_min_trade_usd", notional=notional) return # Don't mirror exits — only entries (BUY) if sig.side != "BUY": log.log("skip", reason="not_a_buy", side=sig.side) return # Time-to-resolution check happens in the risk layer intent = OrderIntent( token_id=sig.token_id, condition_id=sig.market_condition, side="BUY", size_shares=sig.size * COPY_RATIO, target_price=sig.price, order_type="FOK", reason=f"copy:{sig.wallet[:10]}", signal_hash=sig._hash, market_volume_24h=0.0, # could enrich here from gamma market_resolves_at=sig.end_date, ) decision = risk.check(intent) log.log("risk_decision", approved=decision.approved, reject_reason=decision.reject_reason) if not decision.approved: return if DRY_RUN: log.log("dry_run_order", **intent.__dict__) return log.log("order_submit", token_id=intent.token_id, side=intent.side, size=intent.size_shares, price=intent.target_price) result = await place_fok_with_retry( client, token_id=intent.token_id, side="BUY", target_price=intent.target_price, size_shares=intent.size_shares, ) if result.success: log.log("fill", order_id=result.order_id, size=result.filled_size, price=result.avg_price, phase=result.phase) risk.record_fill(intent.condition_id, result.filled_size * (result.avg_price or 0)) else: log.log("order_failed", error=result.error, phase=result.phase) # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- async def main(): print_banner() log = JsonlLogger() client = make_client(PRIVATE_KEY, FUNDER, signature_type=SIGNATURE_TYPE) health = healthcheck(client) log.log("healthcheck", **health) for k, (ok, v) in health.items(): status = "✅" if ok else "❌" print(f" {status} {k}: {v}") print() if not all(ok for ok, _ in health.values()): print("Healthcheck failed — fix above and rerun.") return risk = RiskLayer( dry_run=DRY_RUN, max_trade_size_usd=float(os.environ.get("MAX_TRADE_SIZE_USD", "25")), max_daily_loss_usd=float(os.environ.get("MAX_DAILY_LOSS_USD", "50")), min_market_liquidity_usd=float(os.environ.get("MIN_LIQUIDITY", "5000")), min_time_to_resolution_s=MAX_TIME_TO_RES_H * 3600, ) stop = asyncio.Event() loop = asyncio.get_running_loop() for sig_name in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig_name, stop.set) try: await event_loop(client, risk, log, stop) finally: if not DRY_RUN: cancelled = cancel_all(client) log.log("shutdown", orders_cancelled=cancelled) log.log("stop") log.close() if __name__ == "__main__": asyncio.run(main())