DOLPHIN/prod/launch_paper_portfolio_clean.py

#!/usr/bin/env python3
"""
DOLPHIN Paper Trading - Clean Architecture (Hazelcast-Only)
===========================================================

SINGLE SOURCE OF TRUTH: DolphinNG6 → Hazelcast
- Price data: From Hazelcast (DOLPHIN_FEATURES)
- Eigenvalues: From Hazelcast (computed with price, always synced)
- ACB Boost: From Hazelcast (DOLPHIN_FEATURES listener)
- Posture: From Hazelcast (DOLPHIN_SAFETY)

NO SYNC ISSUES - All data from one source, same timestamp.
🙏 God bless clean architecture 🙏
"""

import asyncio
import os
import sys
import logging
from pathlib import Path
from dotenv import load_dotenv

PROJECT_ROOT = Path(__file__).parent.parent
sys.path.insert(0, str(PROJECT_ROOT / 'nautilus_dolphin'))
sys.path.insert(0, str(PROJECT_ROOT))

load_dotenv(PROJECT_ROOT / '.env')

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("DOLPHIN-CLEAN")

# ---------------------------------------------------------------------------
# CLEAN CONFIGURATION - SINGLE DATA SOURCE
# ---------------------------------------------------------------------------
CONFIG = {
    'trader_id': 'DOLPHIN-CLEAN-01',
    'venue': 'BINANCE_FUTURES',
    
    # SINGLE SOURCE: Hazelcast from DolphinNG6
    'data_source': {
        'type': 'hazelcast',
        'cluster': 'dolphin',
        'host': 'localhost:5701',
        'pulse_ms': 5000,  # 5s from DolphinNG6
    },
    
    # Hazelcast Maps (single source of truth)
    'hz_maps': {
        'features': 'DOLPHIN_FEATURES',    # Price + Eigenvalues (ALWAYS SYNCED)
        'safety': 'DOLPHIN_SAFETY',        # Posture/mode
        'state': 'DOLPHIN_STATE_BLUE',     # Portfolio state
    },
    
    # Execution (paper trading - no data feed needed)
    'execution': {
        'mode': 'paper',
        'sandbox': True,
        'initial_capital': 25000.0,
    },
    
    # Strategy config
    'strategy': {
        'max_leverage': 5.0,
        'capital_fraction': 0.20,
        'acb_enabled': True,
    }
}


class CleanDolphinTrader:
    """
    Clean implementation using ONLY Hazelcast data.
    No dual-feed sync issues.
    """
    
    def __init__(self, config):
        self.config = config
        self.hz_client = None
        self.running = False
        
    async def start(self):
        """Start the clean trader."""
        logger.info("=" * 60)
        logger.info("🐬 DOLPHIN CLEAN TRADER")
        logger.info("=" * 60)
        logger.info(f"Trader ID: {self.config['trader_id']}")
        logger.info(f"Data Source: {self.config['data_source']['type']}")
        logger.info(f"Pulse: {self.config['data_source']['pulse_ms']}ms")
        logger.info("=" * 60)
        
        # Connect to Hazelcast (SINGLE SOURCE)
        await self._connect_hazelcast()
        
        # Verify data availability
        await self._verify_data()
        
        # Start trading loop
        self.running = True
        logger.info("[✓] Trader RUNNING - Listening to Hazelcast")
        logger.info("[✓] All data from single source (no sync issues)")
        
        while self.running:
            await self._process_cycle()
            await asyncio.sleep(5)  # Match DolphinNG6 pulse
            
    async def _connect_hazelcast(self):
        """Connect to Hazelcast - single source of truth."""
        try:
            import hazelcast
            
            ds = self.config['data_source']
            self.hz_client = hazelcast.HazelcastClient(
                cluster_name=ds['cluster'],
                cluster_members=[ds['host']],
            )
            
            logger.info(f"[✓] Connected to Hazelcast: {ds['host']}")
            
        except Exception as e:
            logger.error(f"[✗] Hazelcast connection failed: {e}")
            raise
            
    async def _verify_data(self):
        """Verify data sources are available."""
        try:
            features_map = self.hz_client.get_map(
                self.config['hz_maps']['features']
            ).blocking()
            
            size = features_map.size()
            logger.info(f"[✓] Features map: {size} entries")
            
            if size > 0:
                # Show sample
                keys = list(features_map.key_set())[:1]
                val = features_map.get(keys[0])
                logger.info(f"[✓] Sample data available")
                
        except Exception as e:
            logger.warning(f"[!] Data verification: {e}")
            
    async def _process_cycle(self):
        """Process one trading cycle (5s)."""
        try:
            # Read latest scan from Hazelcast
            features_map = self.hz_client.get_map(
                self.config['hz_maps']['features']
            ).blocking()
            
            # Get latest eigen scan (contains price + eigenvalues - ALWAYS SYNCED)
            latest = features_map.get("latest_eigen_scan")
            
            if latest:
                import json
                data = json.loads(latest)
                
                # Log heartbeat every 6 cycles (30s)
                if asyncio.get_event_loop().time() % 30 < 5:
                    logger.info(f"[HEARTBEAT] Processing scan {data.get('scan_number', 'N/A')}")
                    logger.info(f"  Assets: {len(data.get('assets', []))}")
                    logger.info(f"  Eigenvalues: {len(data.get('eigenvalues', []))}")
                    
                # TODO: Trading logic here
                # All data (price + eigen) from same source, same timestamp
                # NO SYNC ISSUES POSSIBLE
                
        except Exception as e:
            logger.error(f"[!] Processing error: {e}")
            
    async def stop(self):
        """Stop cleanly."""
        self.running = False
        if self.hz_client:
            self.hz_client.shutdown()
            logger.info("[✓] Hazelcast disconnected")
        logger.info("=" * 60)
        logger.info("🙏 Trading session ended - God bless the markets")
        logger.info("=" * 60)


async def main():
    """Main entry point."""
    trader = CleanDolphinTrader(CONFIG)
    
    try:
        await trader.start()
    except KeyboardInterrupt:
        logger.info("Stopping...")
        await trader.stop()
    except Exception as e:
        logger.error(f"Fatal error: {e}")
        await trader.stop()
        raise


if __name__ == "__main__":
    asyncio.run(main())