diff --git a/src/strategies/ping_pong_bot.py b/src/strategies/ping_pong_bot.py index c216749..c6c0423 100644 --- a/src/strategies/ping_pong_bot.py +++ b/src/strategies/ping_pong_bot.py @@ -10,8 +10,13 @@ import pandas as pd import numpy as np from datetime import datetime, timezone from dotenv import load_dotenv +from rich.console import Console +from rich.table import Table +from rich.panel import Panel +from rich.layout import Layout +from rich import box -# Try to import pybit, if not available, we'll suggest installing it +# Try to import pybit try: from pybit.unified_trading import HTTP except ImportError: @@ -55,12 +60,17 @@ class PingPongBot: # State self.last_candle_time = None - self.current_indicators = {} + self.current_indicators = { + "rsi": {"value": 0.0, "timestamp": "N/A"}, + "hurst_lower": {"value": 0.0, "timestamp": "N/A"}, + "hurst_upper": {"value": 0.0, "timestamp": "N/A"} + } self.position = None self.wallet_balance = 0 self.status_msg = "Initializing..." self.last_signal = None self.start_time = datetime.now() + self.console = Console() # Grid parameters from config self.tp_pct = self.config['take_profit_pct'] / 100.0 @@ -97,27 +107,80 @@ class PingPongBot: mcl = mcl_t / 2 mcl_2 = int(round(mcl / 2)) - # True Range df['h_l'] = df['high'] - df['low'] df['h_pc'] = abs(df['high'] - df['close'].shift(1)) df['l_pc'] = abs(df['low'] - df['close'].shift(1)) df['tr'] = df[['h_l', 'h_pc', 'l_pc']].max(axis=1) - # RMA of Close and ATR df['ma_mcl'] = self.rma(df['close'], mcl) df['atr_mcl'] = self.rma(df['tr'], mcl) - # Historical Offset df['center'] = df['ma_mcl'].shift(mcl_2) - # Fill first values where shift produces NaN with the MA itself (as done in JS: historical_ma || src) df['center'] = df['center'].fillna(df['ma_mcl']) mcm_off = mcm * df['atr_mcl'] df['hurst_upper'] = df['center'] + mcm_off df['hurst_lower'] = df['center'] - mcm_off + # Update current indicator state + last_row = df.iloc[-1] + now_str = datetime.now().strftime("%H:%M:%S") + self.current_indicators["rsi"] = {"value": last_row['rsi'], "timestamp": now_str} + self.current_indicators["hurst_lower"] = {"value": last_row['hurst_lower'], "timestamp": now_str} + self.current_indicators["hurst_upper"] = {"value": last_row['hurst_upper'], "timestamp": now_str} + return df + def render_dashboard(self): + """Render a clean summary of the bot state""" + # We don't clear to avoid flickering in some terminals, just print a separator or use rich.Live if needed + # But for simple docker logs, clear is usually fine or just printing the table. + + # 1. Config Table + cfg_table = Table(title="[bold cyan]PING-PONG BOT CONFIG[/]", box=box.ROUNDED, expand=True) + cfg_table.add_column("Property", style="dim") + cfg_table.add_column("Value", style="bold white") + cfg_table.add_row("Symbol", f"{self.symbol} ({self.interval}m)") + cfg_table.add_row("Direction", f"{self.direction.upper()}") + cfg_table.add_row("Capital/Margin", f"${self.pos_size_margin} (Lev: {self.leverage}x)") + cfg_table.add_row("Partial Exit", f"{self.partial_exit_pct*100}%") + cfg_table.add_row("TP Target", f"{self.tp_pct*100}%") + + # 2. Indicators Table + ind_table = Table(title="[bold yellow]TECHNICAL INDICATORS[/]", box=box.ROUNDED, expand=True) + ind_table.add_column("Indicator", style="dim") + ind_table.add_column("Current Value", justify="right") + ind_table.add_column("Last Update", justify="center") + ind_table.add_row("RSI", f"{self.current_indicators['rsi']['value']:.2f}", self.current_indicators['rsi']['timestamp']) + ind_table.add_row("Hurst Lower", f"{self.current_indicators['hurst_lower']['value']:.2f}", self.current_indicators['hurst_lower']['timestamp']) + ind_table.add_row("Hurst Upper", f"{self.current_indicators['hurst_upper']['value']:.2f}", self.current_indicators['hurst_upper']['timestamp']) + + # 3. Position Table + pos_table = Table(title="[bold green]ACTIVE POSITION & ACCOUNT[/]", box=box.ROUNDED, expand=True) + pos_table.add_column("Property", style="dim") + pos_table.add_column("Value", style="bold white") + pos_table.add_row("Wallet Balance", f"${self.wallet_balance:.2f}") + + if self.position: + p = self.position + pnl = float(p.get('unrealisedPnl', 0)) + pnl_color = "green" if pnl >= 0 else "red" + pos_table.add_row("Position Size", f"{p['size']}") + pos_table.add_row("Entry Price", f"{p['avgPrice']}") + pos_table.add_row("Unrealized PnL", f"[{pnl_color}]${pnl:.2f}[/]") + pos_table.add_row("Liquidation", f"{p.get('liqPrice', 'N/A')}") + else: + pos_table.add_row("Position", "NONE (Scanning...)") + + pos_table.add_row("Last Signal", str(self.last_signal or "None")) + pos_table.add_row("Status", f"[bold blue]{self.status_msg}[/]") + + self.console.print("\n" + "="*50) + self.console.print(cfg_table) + self.console.print(ind_table) + self.console.print(pos_table) + self.console.print(f"[dim]Uptime: {str(datetime.now() - self.start_time).split('.')[0]} | Last Run: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}[/]\n") + async def fetch_data(self): """Fetch latest Klines from Bybit V5""" try: @@ -149,7 +212,7 @@ class PingPongBot: async def update_account_info(self): """Update position and balance information""" try: - # Get Position + # Get Position (with settleCoin for Hedge Mode compatibility) pos_response = self.session.get_positions( category="linear", symbol=self.symbol, @@ -333,22 +396,16 @@ class PingPongBot: df = await self.fetch_data() if df is not None: - # 3. Check for New Candle (for signal processing) - last_price = float(df.iloc[-1]['close']) - - # 4. Strategy Logic + # 3. Strategy Logic signal = self.check_signals(df) if signal: - logger.info(f"Signal detected: {signal} @ {last_price}") + logger.info(f"Signal detected: {signal} @ {df.iloc[-1]['close']}") await self.execute_trade_logic(df, signal) - - # 5. Simple status log - if self.position: - logger.info(f"Price: {last_price:.2f} | Pos: {self.position['size']} @ {self.position['avgPrice']} | Wallet: {self.wallet_balance:.2f}") - else: - logger.info(f"Price: {last_price:.2f} | No Position | Wallet: {self.wallet_balance:.2f}") - await asyncio.sleep(self.config.get('loop_interval_seconds', 5)) + # 4. Render Summary Dashboard + self.render_dashboard() + + await asyncio.sleep(self.config.get('loop_interval_seconds', 10)) if __name__ == "__main__": try: @@ -357,5 +414,4 @@ if __name__ == "__main__": except KeyboardInterrupt: print("\nBot Stopped by User") except Exception as e: - print(f"\nCritical Error: {e}") logger.exception("Critical Error in main loop")