#!/usr/bin/env python3 """ Chainlink Predictor — Data Collector + Independent Trading System Collects real-time price data from multiple exchanges + PM (Chainlink) feed, computes SIM price, and trades on Polymarket using its own wallet. Usage: cd ~/btc_15m_collab/rewrite source .venv/bin/activate # Data collection only (no trading) python3 tools/chainlink_predictor.py # Data collection + live trading ($10/trade, $100 capital) python3 tools/chainlink_predictor.py --trade # Custom trade size python3 tools/chainlink_predictor.py --trade --max-trade 5 """ from __future__ import annotations import argparse import asyncio import logging import signal import sys import os import time # Add project root to path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from src.predictor.models import SourceName, SourceTick from src.predictor.feeds import ALL_FEEDS from src.predictor.pm_feed import pm_direct_task from src.predictor.orderbook import ALL_ORDERBOOK_FEEDS from src.predictor.collector import Collector from src.predictor.clob_feed import clob_feed_task, clob_feed_15m_task, current_window as clob_window from src.predictor.strategy import PredictorStrategy, TradeSignal from src.predictor.pnl import PnLTracker, Trade # NOTE: PredictorExecutor is imported lazily inside the --trade branch below, so # paper/collect mode never pulls in the live-trading deps (clob client, web3, dotenv). logging.basicConfig( level=logging.INFO, format="%(asctime)s.%(msecs)03d [%(levelname)s] %(name)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) logger = logging.getLogger(__name__) DEFAULT_SOURCES = [ SourceName.COINBASE, SourceName.KRAKEN, SourceName.BITSTAMP, SourceName.CRYPTOCOMPARE, SourceName.GEMINI, SourceName.OKX, SourceName.BYBIT, SourceName.BINANCE, ] def parse_args(): p = argparse.ArgumentParser(description="Chainlink Predictor + Trader") p.add_argument( "--sources", default=",".join(s.value for s in DEFAULT_SOURCES), help="Comma-separated source names", ) p.add_argument( "--output-dir", default="data/chainlink_predictor", help="Output directory (default: data/chainlink_predictor)", ) p.add_argument( "--cryptocompare-key", default="", help="CryptoCompare API key (optional)", ) p.add_argument( "--trade", action="store_true", help="Enable live trading (default: data collection only)", ) p.add_argument( "--max-trade", type=float, default=10.0, help="Max USD per trade (default: $10)", ) p.add_argument( "--max-daily-loss", type=float, default=30.0, help="Max daily loss before stopping (default: $30)", ) p.add_argument( "--capital", type=float, default=500.0, help="Capital for dynamic sizing (default: $500)", ) p.add_argument( "--shares", type=int, default=0, help="Fixed shares per trade (overrides capital sizing). e.g. --shares 10", ) return p.parse_args() async def collector_and_strategy_task( tick_queue: asyncio.Queue[SourceTick], shutdown: asyncio.Event, output_dir: str, trading_enabled: bool, max_trade_usd: float, max_daily_loss: float, capital: float = 500.0, fixed_shares: int = 0, ): """Combined collector + strategy evaluation loop.""" collector = Collector(output_dir=output_dir) logger.info(f"[collector] started, output={output_dir}") # Strategy and executor (only if trading enabled) strategy = None executor = None pnl = None # Always create strategy (for paper trade + evaluate) strategy = PredictorStrategy( max_trade_usd=max_trade_usd, max_daily_loss=max_daily_loss, capital=capital, fixed_shares=fixed_shares, ) if trading_enabled: try: from src.predictor.executor import PredictorExecutor # live-only import executor = PredictorExecutor() pnl = PnLTracker(output_dir=output_dir) balance = await executor.get_balance() logger.info(f"[trading] ENABLED: wallet balance=${balance:.2f}, max_trade=${max_trade_usd}") except Exception as e: logger.error(f"[trading] failed to initialize: {e}") trading_enabled = False last_window_id = 0 prev_window_pm_a = 0.0 # track pm_a at end of previous window last_settle_check = 0.0 try: while not shutdown.is_set(): try: tick = await asyncio.wait_for(tick_queue.get(), timeout=0.5) # Save pm_a before process_tick (it resets on new window) prev_window_pm_a = collector._pm_a collector.process_tick(tick) except asyncio.TimeoutError: pass # Strategy evaluation — every tick for fast cb_flip detection if strategy is None: continue # Check for new window — settle previous window window_id = collector.current_window if window_id != last_window_id: # Collect all window IDs that need settlement (current + any backlog) settle_wids = set() if last_window_id > 0: settle_wids.add(last_window_id) if pnl: for t in pnl.pending_trades: settle_wids.add(int(t.window_id)) # Also check paper backlog if strategy: for p in strategy._paper_pending: settle_wids.add(int(p.get("window_id", 0))) # Query chain result for each window chain_results = {} for wid in settle_wids: if wid <= 0: continue try: result = await asyncio.to_thread(collector.check_settlement, wid) if result: chain_results[wid] = result except Exception: pass chain_dir = chain_results.get(last_window_id) # Settle real trades — only with chain result if pnl and pnl.pending_trades: for trade in list(pnl.pending_trades): trade_wid = int(trade.window_id) trade_chain_dir = chain_results.get(trade_wid) if not trade_chain_dir: logger.warning(f"[settle] {trade_wid} no chain result, keeping pending") continue # Check actual fill from chain fill_info = None if executor and trade.order_id: try: fill_info = await executor.check_order_filled(trade.order_id) except Exception: pass if fill_info and fill_info["filled"] == 0: # Not filled — but still record direction correctness for analysis trade.settled = True trade.pnl = 0.0 trade.won = False trade.filled = False trade.direction_correct = (trade.direction == trade_chain_dir) trade.chain_direction = trade_chain_dir pnl._write(trade) pnl.pending_trades.remove(trade) logger.info(f"[settle] {trade_wid} {trade.direction} NO FILL (chain={trade_chain_dir}, dir_correct={trade.direction_correct})") elif fill_info and fill_info["filled"] > 0: if trade.settled: continue # Use actual fill amount AND avg fill price from chain actual_shares = fill_info["filled"] actual_price = fill_info.get("avg_fill_price", trade.entry_price) if actual_shares != trade.shares: logger.info(f"[settle] partial fill: ordered={trade.shares} filled={actual_shares:.2f}") if abs(actual_price - trade.entry_price) > 0.001: logger.info(f"[settle] price improved: limit={trade.entry_price:.4f} fill={actual_price:.4f}") trade.shares = actual_shares trade.entry_price = round(actual_price, 4) trade.cost = round(actual_shares * actual_price, 2) won = trade.direction == trade_chain_dir trade.filled = True trade.direction_correct = won trade.chain_direction = trade_chain_dir settled_pnl = pnl.record_settlement( window_id=str(trade_wid), direction=trade.direction, won=won, order_id=trade.order_id, ) if settled_pnl is not None: strategy.record_pnl(settled_pnl) # Daily-loss circuit breaker (shared across all strategies) strategy.check_daily_loss_circuit_breaker() # Lock cooldown after loss (Gate 2) if 'lock' in trade.reason and actual_shares > 0: strategy.record_lock_result(won) logger.info(f"[settle] {trade_wid} chain→{trade_chain_dir}, trade={trade.direction} {'WIN' if won else 'LOSS'} (filled={actual_shares:.2f})") else: # fill_info is None (API error) — retry up to 10 times then give up trade.settle_retries = getattr(trade, 'settle_retries', 0) + 1 if trade.settle_retries >= 10: logger.warning(f"[settle] {trade_wid} giving up after {trade.settle_retries} retries, marking as NFIL") trade.settled = True trade.pnl = 0.0 trade.won = False trade.filled = False trade.direction_correct = False trade.chain_direction = "" pnl._write(trade) pnl.pending_trades.remove(trade) else: logger.warning(f"[settle] {trade_wid} fill check failed (retry {trade.settle_retries}/10)") continue # Settle paper trade — only with chain result, no fallback # Settle paper trades using chain results if strategy: strategy.settle_paper(prev_window_pm_a, chain_results) last_window_id = window_id strategy.new_window(window_id) if executor: executor.new_window(window_id) # Evaluate strategy (returns list of signals) try: signals = strategy.evaluate(collector) except Exception as e: logger.error(f"[strategy] evaluate error: {e}", exc_info=True) signals = [] if not signals and collector._pm_left_sec < 35: _left_int = int(collector._pm_left_sec) if _left_int != getattr(strategy, '_last_log_left', -1): strategy._last_log_left = _left_int print(f" [strategy] left={_left_int}s no signal", flush=True) for sig in signals: if not sig.should_trade or not executor or not executor.can_trade: continue G = "\033[92m" R = "\033[91m" B = "\033[1m" RST = "\033[0m" color = G if sig.direction == "UP" else R print(f"\n{'='*60}") print(f" {B}{color}TRADE SIGNAL [{sig.tier}]: {sig.direction}{RST}") print(f" {sig.reason}") print(f" CLOB ask={sig.price:.2f} Amount: ${sig.amount_usd:.2f} → {int(sig.amount_usd / sig.price)} shares") print(f"{'='*60}\n") try: order_id = await executor.place_buy( token_id=sig.token_id, price=sig.price, amount_usd=sig.amount_usd, expiration_sec=max(int(collector._pm_left_sec) + 60, 90), ) except Exception as _ex: order_id = None print(f" [executor] FAILED: {_ex}", flush=True) logger.error(f"[executor] place_buy exception: {_ex}") if order_id is None: print(f" [executor] order_id=None — order not placed", flush=True) if order_id and pnl: shares = int(sig.amount_usd / sig.price) trade = Trade( ts=int(time.time()), window_id=str(window_id), direction=sig.direction, entry_price=sig.price, shares=shares, cost=round(shares * sig.price, 2), token_id=sig.token_id, order_id=order_id, sim_a=sig.sim_a, pm_a=sig.pm_a, source_agreement=sig.source_agreement, ob_depth_btc=sig.ob_depth_btc, reason=sig.reason, ) pnl.record_entry(trade) strategy.mark_traded() except asyncio.CancelledError: pass finally: if collector.window_ticks: collector._flush_window() collector.close() if pnl: logger.info(f"[pnl] {pnl.summary()}") pnl.close() logger.info(f"[collector] stopped. {collector.total_windows} windows processed.") async def main(): args = parse_args() # Parse source names source_names = [] for s in args.sources.split(","): s = s.strip().lower() try: source_names.append(SourceName(s)) except ValueError: logger.warning(f"Unknown source: {s}, skipping") logger.info(f"Sources: {[s.value for s in source_names]}") logger.info(f"Output dir: {args.output_dir}") logger.info(f"Trading: {'ENABLED' if args.trade else 'DISABLED'}") # Shared queue and shutdown event tick_queue: asyncio.Queue[SourceTick] = asyncio.Queue(maxsize=5000) shutdown = asyncio.Event() # Handle SIGINT/SIGTERM loop = asyncio.get_event_loop() for sig in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig, lambda: shutdown.set()) # Build task list tasks = [] # PM direct feed tasks.append(asyncio.create_task( pm_direct_task(tick_queue, shutdown), name="pm_direct", )) # Exchange feeds for src in source_names: feed_fn = ALL_FEEDS.get(src) if feed_fn is None: continue if src == SourceName.CRYPTOCOMPARE: tasks.append(asyncio.create_task( feed_fn(tick_queue, shutdown, api_key=args.cryptocompare_key), name=f"feed_{src.value}", )) else: tasks.append(asyncio.create_task( feed_fn(tick_queue, shutdown), name=f"feed_{src.value}", )) # Order book feeds for ob_src, ob_fn in ALL_ORDERBOOK_FEEDS.items(): tasks.append(asyncio.create_task( ob_fn(shutdown), name=f"ob_{ob_src.value}", )) # CLOB feed (for Polymarket token prices — needed for trading + paper trade) tasks.append(asyncio.create_task( clob_feed_task(shutdown), name="clob_feed", )) # 15m CLOB feed (parallel) — exposes current_window_15m for arb/magic combo sum tasks.append(asyncio.create_task( clob_feed_15m_task(shutdown), name="clob_feed_15m", )) # Collector + Strategy + Executor tasks.append(asyncio.create_task( collector_and_strategy_task( tick_queue, shutdown, args.output_dir, trading_enabled=args.trade, max_trade_usd=args.max_trade, max_daily_loss=args.max_daily_loss, capital=args.capital, fixed_shares=args.shares, ), name="collector_strategy", )) mode = "COLLECT + TRADE" if args.trade else "COLLECT ONLY" print() print("=" * 60) print(f" Chainlink Predictor — {mode}") print(f" Sources: {', '.join(s.value for s in source_names)}") print(f" PM: direct WebSocket") if args.trade: print(f" Trading: ${args.max_trade:.0f}/trade, capital=${args.capital:.0f}, daily limit -${args.max_daily_loss:.0f}") print(f" Output: {args.output_dir}") print(" Press Ctrl+C to stop") print("=" * 60) print() # Wait for shutdown await shutdown.wait() logger.info("Shutdown signal received") for t in tasks: t.cancel() await asyncio.gather(*tasks, return_exceptions=True) logger.info("All tasks stopped.") if __name__ == "__main__": asyncio.run(main())