""" Paper Trading Engine - Live signal generation + simulated order execution Uses fixed V2 strategy with dynamic leverage """ import os import sys import time import json import logging import threading from datetime import datetime from collections import deque from typing import Dict, List, Optional, Callable from dataclasses import dataclass, asdict from dotenv import load_dotenv load_dotenv() sys.path.insert(0, os.path.dirname(__file__)) from config.settings_v2 import V2Params from strategy.detector_v2 import detect_v2 from strategy.simulator_v2 import simulate_v2 from binance_client import BinanceClient, load_client_from_env from telegram_notifier import get_notifier logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s") logger = logging.getLogger(__name__) @dataclass class Position: symbol: str side: str # "SHORT" entry_price: float quantity: float leverage: float entry_time: int sl_price: float trailing_sl: float highest_price: float # for trailing (lowest for short) unrealized_pnl: float = 0.0 realized_pnl: float = 0.0 @dataclass class TradeRecord: symbol: str entry_time: int exit_time: int entry_price: float exit_price: float quantity: float leverage: float pnl_usd: float pnl_pct: float exit_reason: str score: int class PaperEngine: def __init__(self, client: BinanceClient, params: V2Params = None): self.client = client self.params = params or V2Params() self.equity = 100.0 # Start with $100 paper equity self.initial_equity = 100.0 # State self.positions: Dict[str, Position] = {} # symbol -> Position self.trade_history: List[TradeRecord] = [] self.equity_curve: List[float] = [self.equity] # Data buffers (symbol -> deque of klines) self.klines_15m: Dict[str, deque] = {} self.klines_4h: Dict[str, deque] = {} self.max_bars_15m = 2000 # ~17 days at 96 bars/day self.max_bars_4h = 200 # ~33 days at 6 bars/day # Risk limits self.max_daily_loss_pct = float(os.getenv("MAX_DAILY_LOSS_PCT", "5.0")) self.max_total_loss_pct = float(os.getenv("MAX_TOTAL_LOSS_PCT", "15.0")) self.max_position_usd = float(os.getenv("MAX_POSITION_USD", "200")) self.max_concurrent = int(os.getenv("MAX_CONCURRENT_POSITIONS", "3")) self.min_notional = float(os.getenv("MIN_24H_NOTIONAL_USD", "5000000")) # Tracking self.daily_start_equity = self.equity self.last_daily_reset = datetime.now().date() self.running = False self.symbols: List[str] = [] # Telegram notifier self.notifier = get_notifier() def load_initial_data(self): """Load historical klines for all symbols""" logger.info(f"Loading initial data for {len(self.symbols)} symbols...") for sym in self.symbols: try: k15 = self.client.get_klines(sym, "15m", limit=min(self.max_bars_15m, 1500)) k4 = self.client.get_klines(sym, "4h", limit=self.max_bars_4h) self.klines_15m[sym] = deque(k15, maxlen=self.max_bars_15m) self.klines_4h[sym] = deque(k4, maxlen=self.max_bars_4h) logger.info(f" {sym}: {len(k15)} 15m bars, {len(k4)} 4h bars") except Exception as e: logger.error(f" {sym}: Failed - {e}") logger.info("Initial data loaded") def update_symbols(self): """Refresh tradable universe daily - use top gainers for pump detection""" try: self.symbols = self.client.get_top_gainers( min_notional=self.min_notional, limit=100 ) logger.info(f"Universe updated (top gainers): {len(self.symbols)} symbols") if self.symbols: logger.info(f"Top 5: {self.symbols[:5]}") except Exception as e: logger.error(f"Failed to update symbols: {e}") def _check_risk_limits(self) -> bool: """Check if we can open new positions""" # Daily loss limit daily_loss_pct = (self.daily_start_equity - self.equity) / self.daily_start_equity * 100 if daily_loss_pct >= self.max_daily_loss_pct: logger.warning(f"Daily loss limit hit: {daily_loss_pct:.1f}% >= {self.max_daily_loss_pct}%") return False # Total loss limit total_loss_pct = (self.initial_equity - self.equity) / self.initial_equity * 100 if total_loss_pct >= self.max_total_loss_pct: logger.warning(f"Total loss limit hit: {total_loss_pct:.1f}% >= {self.max_total_loss_pct}%") return False # Max concurrent positions if len(self.positions) >= self.max_concurrent: return False return True def _reset_daily(self): """Reset daily tracking at midnight""" today = datetime.now().date() if today > self.last_daily_reset: self.daily_start_equity = self.equity self.last_daily_reset = today logger.info(f"Daily reset: equity = ${self.equity:.2f}") def process_closed_kline(self, symbol: str, kline: dict): """Process a closed 15m kline - main signal generation""" # Update buffers if symbol not in self.klines_15m: self.klines_15m[symbol] = deque(maxlen=self.max_bars_15m) self.klines_15m[symbol].append(kline) # Need 4h data too if symbol not in self.klines_4h or len(self.klines_4h[symbol]) < 50: return # Convert to list format for detector k15_list = list(self.klines_15m[symbol]) k4_list = list(self.klines_4h[symbol]) # Run detector signals = detect_v2(k4_list, k15_list, self.params) for sig in signals: # Check if we already have position if symbol in self.positions: continue # Risk check if not self._check_risk_limits(): continue # Simulate trade to get SL, leverage, etc. sim = simulate_v2(symbol, k15_list, sig, self.params, equity=self.equity) if not sim or sim.r_multiple is None: continue # Calculate position size risk_pct = self.params.risk_score4 if sig.get("squeeze_score", 0) >= 4 else self.params.risk_score3 risk_usd = self.equity * risk_pct sl_dist = sim.sl - sim.entry_price # positive for short if sl_dist <= 0: continue # Position size in quote currency notional = (risk_usd * sim.leverage) / (sl_dist / sim.entry_price) notional = min(notional, self.max_position_usd) qty = notional / sim.entry_price if qty <= 0: continue # Create position pos = Position( symbol=symbol, side="SHORT", entry_price=sim.entry_price, quantity=qty, leverage=sim.leverage, entry_time=kline["time"], sl_price=sim.sl, trailing_sl=sim.sl, highest_price=sim.entry_price, # track low for short ) self.positions[symbol] = pos logger.info(f"OPEN SHORT {symbol} @ {sim.entry_price:.6f} qty={qty:.4f} lev={sim.leverage:.2f}x SL={sim.sl:.6f}") # Telegram notification notional_usd = qty * sim.entry_price * sim.leverage self.notifier.send_trade_open( symbol=symbol, entry_price=sim.entry_price, leverage=sim.leverage, sl_price=sim.sl, notional_usd=notional_usd, risk_pct=risk_pct ) def update_positions(self, symbol: str, current_price: float): """Update trailing SL and check exits for open position""" if symbol not in self.positions: return pos = self.positions[symbol] # Update highest (lowest for short) if current_price < pos.highest_price: pos.highest_price = current_price # Update trailing SL coef = self.params.trailing_k * pos.highest_price * 0.001 # approx ATR new_sl = min(pos.trailing_sl, pos.highest_price + coef) new_sl = min(new_sl, pos.entry_price - coef) # floor below entry if new_sl < pos.trailing_sl: pos.trailing_sl = new_sl logger.debug(f"Trailing SL update {symbol}: {pos.trailing_sl:.6f}") # Check exits exit_reason = None exit_price = None # Liquidation check (price went above liq) liq_price = pos.entry_price * (1 + (1 - self.params.maintenance_margin) / pos.leverage) if current_price >= liq_price: exit_reason = "LIQUIDATION" exit_price = liq_price # Hard SL hit elif current_price >= pos.sl_price: exit_reason = "LOSS_HARD_STOP" exit_price = pos.sl_price # Trailing SL hit elif current_price >= pos.trailing_sl: exit_reason = "WINNER_TRAIL" exit_price = pos.trailing_sl if exit_reason: self.close_position(symbol, exit_price, exit_reason) def close_position(self, symbol: str, exit_price: float, reason: str): """Close position and record trade""" pos = self.positions.pop(symbol) # Calculate PnL pnl_pct = (pos.entry_price - exit_price) / pos.entry_price # short pnl_usd = pnl_pct * pos.quantity * pos.entry_price * pos.leverage # Fees fee = pos.quantity * exit_price * pos.leverage * self.params.fee_pct * 2 slippage = pos.quantity * exit_price * pos.leverage * self.params.slippage_pct * 2 net_pnl = pnl_usd - fee - slippage # Update equity self.equity += net_pnl self.equity_curve.append(self.equity) # Record trade = TradeRecord( symbol=symbol, entry_time=pos.entry_time, exit_time=int(time.time() * 1000), entry_price=pos.entry_price, exit_price=exit_price, quantity=pos.quantity, leverage=pos.leverage, pnl_usd=net_pnl, pnl_pct=net_pnl / self.equity * 100, exit_reason=reason, score=3, # TODO: track actual score ) self.trade_history.append(trade) logger.info(f"CLOSE {symbol} @ {exit_price:.6f} | {reason} | PnL: ${net_pnl:.2f} ({net_pnl/self.equity*100:+.2f}%) | Equity: ${self.equity:.2f}") # Telegram notification if self.notifier.config.enabled: self.notifier.send_trade_close( symbol=symbol, entry_price=pos.entry_price, exit_price=exit_price, pnl_usd=net_pnl, pnl_pct=net_pnl / self.equity * 100, reason=reason ) def on_kline_update(self, kline: dict): """Callback for WebSocket kline updates""" symbol = kline["symbol"] if not kline.get("x"): # Not closed yet return self._reset_daily() # Update 15m buffer self.process_closed_kline(symbol, kline) # Update 4h buffer if this is a 4h boundary (every 16th 15m bar) # For simplicity, also fetch 4h periodically # Check position exits if symbol in self.positions: self.update_positions(symbol, kline["close"]) def save_state(self, path: str = "paper_state.json"): """Save state to disk""" state = { "equity": self.equity, "equity_curve": self.equity_curve, "trade_history": [asdict(t) for t in self.trade_history], "positions": {s: asdict(p) for s, p in self.positions.items()}, "timestamp": int(time.time() * 1000), } with open(path, "w") as f: json.dump(state, f, indent=2) def load_state(self, path: str = "paper_state.json"): """Load state from disk""" if not os.path.exists(path): return with open(path, "r") as f: state = json.load(f) self.equity = state["equity"] self.equity_curve = state["equity_curve"] self.trade_history = [TradeRecord(**t) for t in state["trade_history"]] self.positions = {s: Position(**p) for s, p in state["positions"].items()} logger.info(f"Loaded state: equity=${self.equity:.2f}, {len(self.trade_history)} trades, {len(self.positions)} open") def run_paper_trading(duration_hours: int = 720): # 30 days default """Main entry point""" client = load_client_from_env() # Load optimized params params = V2Params() params.range_pctile_thr = float(os.getenv("RANGE_PCTILE_THR", "24.0")) params.atr_ratio_thr = float(os.getenv("ATR_RATIO_THR", "0.78")) params.volume_ratio_thr = float(os.getenv("VOLUME_RATIO_THR", "0.88")) params.va_pctile_thr = float(os.getenv("VA_PCTILE_THR", "28.0")) params.trailing_k = float(os.getenv("TRAILING_K", "2.9")) engine = PaperEngine(client, params) # Initial setup engine.update_symbols() engine.load_initial_data() # Start WebSocket for 15m klines ws = client.start_kline_stream(engine.symbols, "15m", engine.on_kline_update) # Also fetch 4h klines periodically def refresh_4h(): while engine.running: try: for sym in engine.symbols: k4 = client.get_klines(sym, "4h", limit=engine.max_bars_4h) engine.klines_4h[sym] = deque(k4, maxlen=engine.max_bars_4h) except Exception as e: logger.error(f"4h refresh error: {e}") time.sleep(3600) # every hour threading.Thread(target=refresh_4h, daemon=True).start() logger.info(f"Paper trading started for {len(engine.symbols)} symbols") logger.info(f"Params: range<{params.range_pctile_thr} atr<{params.atr_ratio_thr} vol<{params.volume_ratio_thr} va<{params.va_pctile_thr} trail={params.trailing_k}") engine.running = True try: while engine.running: time.sleep(1) except KeyboardInterrupt: logger.info("Shutting down...") finally: engine.running = False engine.save_state() logger.info(f"Final equity: ${engine.equity:.2f} ({(engine.equity/engine.initial_equity-1)*100:+.1f}%)") logger.info(f"Total trades: {len(engine.trade_history)}") if __name__ == "__main__": run_paper_trading()