#!/usr/bin/env python3 """ REALTIME TRADING ENGINE - WebSocket Kraken =========================================== Sistema real-time professionale con: - WebSocket Kraken per prezzi live - Aggiornamento P&L in tempo reale - Rate limiting ottimizzato - Multi-bot con leverage """ import asyncio import json import signal import hmac import hashlib import base64 import time import urllib.parse from datetime import datetime from typing import Dict, Any, Optional from pathlib import Path import aiohttp import psycopg2 from psycopg2.extras import RealDictCursor # Load environment variables from dotenv import load_dotenv import os env_path = Path(__file__).parent.parent.parent / '.env' load_dotenv(env_path) # ============================================ # CONFIGURAZIONE # ============================================ DB_CONFIG = { 'host': os.getenv('DB_HOST', 'localhost'), 'port': int(os.getenv('DB_PORT', 5432)), 'dbname': os.getenv('DB_NAME', 'pippo'), 'user': os.getenv('DB_USER', 'pippo'), 'password': os.getenv('DB_PASSWORD', 'pippo_trading_2026') } KRAKEN_API_KEY = os.getenv('KRAKEN_API_KEY', '') KRAKEN_API_SECRET = os.getenv('KRAKEN_API_SECRET', '') # Kraken WebSocket URLs KRAKEN_WS_PUBLIC = 'wss://ws.kraken.com' KRAKEN_WS_PRIVATE = 'wss://ws-auth.kraken.com' KRAKEN_REST_URL = 'https://api.kraken.com' # Fee structure KRAKEN_MAKER_FEE = 0.0016 MARGIN_OPEN_FEE = 0.0002 MARGIN_ROLLOVER = 0.0002 SPREAD = 0.0003 # ============================================ # Global State # ============================================ running = True bots: Dict[int, Dict[str, Any]] = {} current_price = 0.0 price_24h_high = 0.0 price_24h_low = 0.0 last_ws_update = 0 # ============================================ # Logging # ============================================ def log(msg: str, bot_id: Optional[int] = None): timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S.%f')[:-3] prefix = f"[Bot {bot_id}]" if bot_id else "[Engine]" print(f"[{timestamp}] {prefix} {msg}", flush=True) # ============================================ # Kraken API Authentication # ============================================ def get_kraken_signature(urlpath: str, data: dict, secret: str) -> str: """Generate Kraken API signature.""" postdata = urllib.parse.urlencode(data) encoded = (str(data['nonce']) + postdata).encode() message = urlpath.encode() + hashlib.sha256(encoded).digest() mac = hmac.new(base64.b64decode(secret), message, hashlib.sha512) return base64.b64encode(mac.digest()).decode() async def kraken_private_request(endpoint: str, params: dict = None) -> dict: """Make authenticated Kraken REST API request.""" if not KRAKEN_API_KEY or not KRAKEN_API_SECRET: raise ValueError("Kraken API credentials not configured") url = f"{KRAKEN_REST_URL}/0/private/{endpoint}" params = params or {} params['nonce'] = int(time.time() * 1000) signature = get_kraken_signature(f'/0/private/{endpoint}', params, KRAKEN_API_SECRET) headers = { 'API-Key': KRAKEN_API_KEY, 'API-Sign': signature, 'Content-Type': 'application/x-www-form-urlencoded' } async with aiohttp.ClientSession() as session: async with session.post(url, headers=headers, data=params) as resp: data = await resp.json() if data.get('error'): raise Exception(f"Kraken API Error: {data['error']}") return data.get('result', {}) async def get_ws_auth_token() -> str: """Get WebSocket authentication token from Kraken.""" result = await kraken_private_request('GetWebSocketsToken') return result.get('token', '') # ============================================ # Database Operations # ============================================ def get_db_connection(): return psycopg2.connect(**DB_CONFIG, cursor_factory=RealDictCursor) def load_active_bots(): """Load active bots from database.""" global bots try: conn = get_db_connection() cur = conn.cursor() cur.execute(""" SELECT t.*, ts.position, ts.position_amount, ts.entry_price, ts.entry_time FROM traders t LEFT JOIN trader_states ts ON t.id = ts.trader_id WHERE t.status IN ('paper', 'live') """) rows = cur.fetchall() cur.close() conn.close() for row in rows: bot_id = row['id'] if bot_id not in bots: bots[bot_id] = { 'id': bot_id, 'name': row['name'], 'status': row['status'], 'initial_capital': float(row['initial_capital']), 'current_capital': float(row['current_capital']), 'leverage': int(row.get('leverage') or 1), 'direction': row.get('direction') or 'long', 'dip_threshold': float(row.get('dip_threshold') or 0.08), 'take_profit': float(row['take_profit']), 'stop_loss': float(row['stop_loss']), 'lookback_hours': int(row.get('lookback_hours') or 24), 'position': row.get('position') or 0, 'position_amount': float(row.get('position_amount') or 0), 'entry_price': float(row.get('entry_price') or 0), 'entry_time': row.get('entry_time'), } log(f"Loaded: {row['name']} | {row.get('direction', 'long').upper()} {row.get('leverage', 1)}x", bot_id) else: # Update mutable state bots[bot_id]['current_capital'] = float(row['current_capital']) bots[bot_id]['position'] = row.get('position') or 0 bots[bot_id]['position_amount'] = float(row.get('position_amount') or 0) bots[bot_id]['entry_price'] = float(row.get('entry_price') or 0) except Exception as e: log(f"Error loading bots: {e}") def update_bot_state_realtime(bot_id: int, current_price: float, unrealized_pnl: float, unrealized_pnl_pct: float): """Update bot state with real-time data.""" try: conn = get_db_connection() cur = conn.cursor() cur.execute(""" UPDATE trader_states SET current_price = %s, unrealized_pnl = %s, unrealized_pnl_pct = %s, updated_at = NOW() WHERE trader_id = %s """, (current_price, unrealized_pnl, unrealized_pnl_pct, bot_id)) conn.commit() cur.close() conn.close() except Exception as e: log(f"Error updating state: {e}", bot_id) def save_trade(bot_id: int, trade: Dict[str, Any], mode: str): """Save trade to database.""" try: conn = get_db_connection() cur = conn.cursor() cur.execute(""" INSERT INTO trades ( trader_id, mode, pair, direction, action, entry_price, exit_price, amount, fee_entry, fee_exit, gross_pnl, net_pnl, pnl_percent, exit_reason, entry_time, exit_time, duration_minutes, capital_before, capital_after ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """, ( bot_id, mode, trade.get('pair', 'BTC/EUR'), trade.get('direction'), trade.get('action'), trade.get('entry_price'), trade.get('exit_price'), trade.get('amount'), trade.get('fee_entry', 0), trade.get('fee_exit', 0), trade.get('gross_pnl'), trade.get('net_pnl'), trade.get('pnl_percent'), trade.get('exit_reason'), trade.get('entry_time'), trade.get('exit_time'), trade.get('duration_minutes'), trade.get('capital_before'), trade.get('capital_after'), )) conn.commit() cur.close() conn.close() except Exception as e: log(f"Error saving trade: {e}", bot_id) def save_bot_position(bot_id: int, bot: dict): """Save bot position to database.""" try: conn = get_db_connection() cur = conn.cursor() cur.execute(""" INSERT INTO trader_states (trader_id, position, position_amount, entry_price, entry_time, updated_at) VALUES (%s, %s, %s, %s, %s, NOW()) ON CONFLICT (trader_id) DO UPDATE SET position = EXCLUDED.position, position_amount = EXCLUDED.position_amount, entry_price = EXCLUDED.entry_price, entry_time = EXCLUDED.entry_time, updated_at = NOW() """, (bot_id, bot['position'], bot['position_amount'], bot['entry_price'], bot['entry_time'])) cur.execute("UPDATE traders SET current_capital = %s, updated_at = NOW() WHERE id = %s", (bot['current_capital'], bot_id)) conn.commit() cur.close() conn.close() except Exception as e: log(f"Error saving position: {e}", bot_id) # ============================================ # Trading Logic # ============================================ def calculate_realtime_pnl(bot: dict, price: float) -> tuple: """Calculate real-time P&L for a position.""" if bot['position'] == 0: return 0.0, 0.0 entry_price = bot['entry_price'] amount = bot['position_amount'] leverage = bot['leverage'] if bot['position'] == 1: # LONG price_change = price - entry_price else: # SHORT price_change = entry_price - price gross_pnl = amount * price_change * leverage pnl_pct = (price_change / entry_price) * 100 * leverage if entry_price > 0 else 0 return gross_pnl, pnl_pct def should_open_position(bot: dict, price: float, high_24h: float, low_24h: float) -> bool: """Check if bot should open a position.""" if bot['position'] != 0: return False if bot['direction'] == 'long': # Buy on dip from high if high_24h > 0: dip = (price - high_24h) / high_24h return dip <= -bot['dip_threshold'] else: # Short on pump from low if low_24h > 0: pump = (price - low_24h) / low_24h return pump >= bot['dip_threshold'] return False def should_close_position(bot: dict, price: float) -> tuple: """Check if bot should close position. Returns (should_close, reason).""" if bot['position'] == 0: return False, None entry_price = bot['entry_price'] if bot['position'] == 1: # LONG pnl_pct = (price - entry_price) / entry_price if pnl_pct >= bot['take_profit']: return True, 'take_profit' if pnl_pct <= -bot['stop_loss']: return True, 'stop_loss' else: # SHORT pnl_pct = (entry_price - price) / entry_price if pnl_pct >= bot['take_profit']: return True, 'take_profit' if pnl_pct <= -bot['stop_loss']: return True, 'stop_loss' return False, None def execute_open(bot: dict, price: float): """Execute position open.""" leverage = bot['leverage'] direction = bot['direction'] position_value = bot['current_capital'] * 0.95 margin_fee = position_value * leverage * MARGIN_OPEN_FEE if leverage > 1 else 0 trading_fee = position_value * KRAKEN_MAKER_FEE exec_price = price * (1 + SPREAD) if direction == 'long' else price * (1 - SPREAD) amount = position_value / exec_price capital_before = bot['current_capital'] bot['current_capital'] -= trading_fee + margin_fee bot['position'] = 1 if direction == 'long' else -1 bot['position_amount'] = amount bot['entry_price'] = exec_price bot['entry_time'] = datetime.now() save_trade(bot['id'], { 'pair': 'BTC/EUR', 'direction': direction.upper(), 'action': 'OPEN', 'entry_price': exec_price, 'amount': amount, 'fee_entry': trading_fee + margin_fee, 'entry_time': bot['entry_time'], 'capital_before': capital_before, 'capital_after': bot['current_capital'], }, bot['status']) save_bot_position(bot['id'], bot) tp_price = exec_price * (1 + bot['take_profit']) if direction == 'long' else exec_price * (1 - bot['take_profit']) sl_price = exec_price * (1 - bot['stop_loss']) if direction == 'long' else exec_price * (1 + bot['stop_loss']) log(f"OPEN {direction.upper()} {leverage}x | {amount:.6f} BTC @ {exec_price:,.2f} | TP: {tp_price:,.0f} | SL: {sl_price:,.0f}", bot['id']) def execute_close(bot: dict, price: float, reason: str): """Execute position close.""" leverage = bot['leverage'] direction = 'long' if bot['position'] == 1 else 'short' amount = bot['position_amount'] entry_price = bot['entry_price'] exec_price = price * (1 - SPREAD) if bot['position'] == 1 else price * (1 + SPREAD) if bot['position'] == 1: gross_pnl = amount * (exec_price - entry_price) * leverage else: gross_pnl = amount * (entry_price - exec_price) * leverage trading_fee = abs(gross_pnl) * KRAKEN_MAKER_FEE # Rollover fee rollover_cost = 0 if bot['entry_time'] and leverage > 1: hours_in_pos = (datetime.now() - bot['entry_time']).total_seconds() / 3600 rollover_periods = int(hours_in_pos / 4) rollover_cost = amount * entry_price * leverage * MARGIN_ROLLOVER * rollover_periods net_pnl = gross_pnl - trading_fee - rollover_cost pnl_pct = net_pnl / (amount * entry_price) * 100 if entry_price > 0 else 0 capital_before = bot['current_capital'] bot['current_capital'] += net_pnl + (amount * entry_price) duration = int((datetime.now() - bot['entry_time']).total_seconds() / 60) if bot['entry_time'] else 0 save_trade(bot['id'], { 'pair': 'BTC/EUR', 'direction': direction.upper(), 'action': 'CLOSE', 'entry_price': entry_price, 'exit_price': exec_price, 'amount': amount, 'fee_exit': trading_fee + rollover_cost, 'gross_pnl': gross_pnl, 'net_pnl': net_pnl, 'pnl_percent': pnl_pct, 'exit_reason': reason, 'entry_time': bot['entry_time'], 'exit_time': datetime.now(), 'duration_minutes': duration, 'capital_before': capital_before, 'capital_after': bot['current_capital'], }, bot['status']) bot['position'] = 0 bot['position_amount'] = 0 bot['entry_price'] = 0 bot['entry_time'] = None save_bot_position(bot['id'], bot) emoji = '+' if net_pnl > 0 else '' log(f"CLOSE {direction.upper()} @ {exec_price:,.2f} | {reason} | P&L: {emoji}{net_pnl:.2f} ({pnl_pct:+.2f}%)", bot['id']) # ============================================ # WebSocket Handler # ============================================ async def handle_ticker_message(data: dict): """Handle ticker update from WebSocket.""" global current_price, price_24h_high, price_24h_low, last_ws_update if isinstance(data, list) and len(data) >= 2: ticker = data[1] if isinstance(ticker, dict): # c = last trade price [price, lot volume] if 'c' in ticker: current_price = float(ticker['c'][0]) last_ws_update = time.time() # h = high [today, 24h] if 'h' in ticker: price_24h_high = float(ticker['h'][1]) # l = low [today, 24h] if 'l' in ticker: price_24h_low = float(ticker['l'][1]) async def websocket_client(): """Connect to Kraken WebSocket and receive real-time data.""" global running, current_price reconnect_delay = 1 while running: try: async with aiohttp.ClientSession() as session: async with session.ws_connect(KRAKEN_WS_PUBLIC) as ws: log(f"Connected to Kraken WebSocket") reconnect_delay = 1 # Reset on successful connection # Subscribe to BTC/EUR ticker subscribe_msg = { "event": "subscribe", "pair": ["XBT/EUR"], "subscription": {"name": "ticker"} } await ws.send_json(subscribe_msg) log("Subscribed to XBT/EUR ticker") async for msg in ws: if not running: break if msg.type == aiohttp.WSMsgType.TEXT: data = json.loads(msg.data) # Handle subscription status if isinstance(data, dict): if data.get('event') == 'subscriptionStatus': status = data.get('status') pair = data.get('pair') log(f"Subscription {status}: {pair}") elif data.get('event') == 'heartbeat': pass # Ignore heartbeats else: # Handle ticker data await handle_ticker_message(data) elif msg.type in (aiohttp.WSMsgType.CLOSED, aiohttp.WSMsgType.ERROR): log(f"WebSocket closed/error: {msg.type}") break except Exception as e: log(f"WebSocket error: {e}") if running: log(f"Reconnecting in {reconnect_delay}s...") await asyncio.sleep(reconnect_delay) reconnect_delay = min(reconnect_delay * 2, 30) # ============================================ # Main Loop # ============================================ async def trading_loop(): """Main trading loop - processes bots with real-time data.""" global running, current_price, price_24h_high, price_24h_low log("="*70) log("REALTIME TRADING ENGINE - WebSocket Kraken") log("="*70) log(f"Kraken API: {'Configured' if KRAKEN_API_KEY else 'NOT CONFIGURED'}") last_status_log = 0 last_bot_reload = 0 while running: try: now = time.time() # Reload bots every 60 seconds if now - last_bot_reload > 60: load_active_bots() last_bot_reload = now # Skip if no price data yet if current_price <= 0: await asyncio.sleep(1) continue # Process each bot for bot in list(bots.values()): # Calculate real-time P&L if bot['position'] != 0: gross_pnl, pnl_pct = calculate_realtime_pnl(bot, current_price) update_bot_state_realtime(bot['id'], current_price, gross_pnl, pnl_pct) # Check for exit should_exit, reason = should_close_position(bot, current_price) if should_exit: execute_close(bot, current_price, reason) else: # Check for entry if should_open_position(bot, current_price, price_24h_high, price_24h_low): execute_open(bot, current_price) # Log status every 30 seconds if now - last_status_log > 30: long_pos = sum(1 for b in bots.values() if b['position'] == 1) short_pos = sum(1 for b in bots.values() if b['position'] == -1) total_cap = sum(b['current_capital'] for b in bots.values()) ws_age = now - last_ws_update if last_ws_update > 0 else -1 log(f"BTC/EUR: {current_price:,.2f} | H24: {price_24h_high:,.0f} L24: {price_24h_low:,.0f} | " f"{len(bots)} bots (L:{long_pos} S:{short_pos}) | Cap: {total_cap:,.2f} | WS: {ws_age:.1f}s ago") last_status_log = now await asyncio.sleep(0.5) # Check every 500ms for responsive trading except asyncio.CancelledError: break except Exception as e: log(f"Trading loop error: {e}") import traceback traceback.print_exc() await asyncio.sleep(5) log("Trading loop stopped") async def main(): """Main entry point.""" global running # Start WebSocket and trading loop concurrently ws_task = asyncio.create_task(websocket_client()) trading_task = asyncio.create_task(trading_loop()) try: await asyncio.gather(ws_task, trading_task) except asyncio.CancelledError: pass finally: running = False ws_task.cancel() trading_task.cancel() log("Engine stopped") def signal_handler(sig, frame): global running log("Shutting down...") running = False if __name__ == '__main__': signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) asyncio.run(main())