""" Test script for Real-Time RL Learning This script demonstrates the real-time RL learning system that learns from each trade execution and position closure. """ import os import sys import logging import time import numpy as np from datetime import datetime # Add core directory to path sys.path.append(os.path.join(os.path.dirname(__file__), 'core')) sys.path.append(os.path.join(os.path.dirname(__file__), 'NN')) from core.trading_executor import TradingExecutor from core.realtime_rl_trainer import RealTimeRLTrainer # Setup logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) def test_rl_trainer_standalone(): """Test the RL trainer as a standalone component""" logger.info("=== Testing Real-Time RL Trainer (Standalone) ===") # Initialize RL trainer rl_config = { 'state_size': 100, 'learning_rate': 0.0001, 'training_enabled': True, 'min_experiences': 5, 'training_frequency': 2, 'save_frequency': 10 } trainer = RealTimeRLTrainer(rl_config) # Simulate market data updates logger.info("Simulating market data updates...") for i in range(20): price = 3000 + np.random.normal(0, 50) # ETH price around $3000 volume = 1000 + np.random.normal(0, 100) rsi = 30 + np.random.uniform(0, 40) trainer.update_market_data('ETH/USDC', price, volume, rsi) time.sleep(0.1) # Small delay # Simulate some trading signals and closures logger.info("Simulating trading signals and position closures...") trades = [ {'action': 'BUY', 'price': 3000, 'confidence': 0.8, 'exit_price': 3020, 'pnl': 20, 'fees': 1.5}, {'action': 'SELL', 'price': 3020, 'confidence': 0.7, 'exit_price': 3000, 'pnl': 20, 'fees': 1.5}, {'action': 'BUY', 'price': 3000, 'confidence': 0.6, 'exit_price': 2990, 'pnl': -10, 'fees': 1.5}, {'action': 'BUY', 'price': 2990, 'confidence': 0.9, 'exit_price': 3050, 'pnl': 60, 'fees': 1.5}, {'action': 'SELL', 'price': 3050, 'confidence': 0.8, 'exit_price': 3010, 'pnl': 40, 'fees': 1.5}, ] for i, trade in enumerate(trades): # Record trade signal trainer.record_trade_signal( symbol='ETH/USDC', action=trade['action'], confidence=trade['confidence'], current_price=trade['price'] ) # Wait a bit to simulate holding time time.sleep(0.5) # Record position closure trainer.record_position_closure( symbol='ETH/USDC', exit_price=trade['exit_price'], pnl=trade['pnl'], fees=trade['fees'] ) # Get training stats stats = trainer.get_training_stats() logger.info(f"Trade {i+1}: Win Rate={stats['win_rate']*100:.1f}%, " f"Avg Reward={stats['avg_reward']:.4f}, " f"Memory Size={stats['memory_size']}") time.sleep(0.2) # Test RL prediction logger.info("Testing RL prediction...") action, confidence = trainer.get_action_prediction('ETH/USDC') logger.info(f"RL Prediction: {action} (confidence: {confidence:.2f})") # Final stats final_stats = trainer.get_training_stats() logger.info(f"Final Stats: {final_stats}") return True def test_trading_executor_integration(): """Test RL learning integration with TradingExecutor""" logger.info("\n=== Testing TradingExecutor RL Integration ===") try: # Initialize trading executor with RL learning executor = TradingExecutor("config.yaml") # Check if RL trainer was initialized if hasattr(executor, 'rl_trainer'): logger.info("RL trainer successfully integrated with TradingExecutor") # Get initial RL stats rl_stats = executor.get_rl_training_stats() logger.info(f"Initial RL stats: {rl_stats}") # Test RL prediction action, confidence = executor.get_rl_prediction('ETH/USDC') logger.info(f"RL prediction for ETH/USDC: {action} (confidence: {confidence:.2f})") # Simulate some trading signals logger.info("Simulating trading signals through TradingExecutor...") test_signals = [ {'symbol': 'ETH/USDC', 'action': 'BUY', 'confidence': 0.8, 'price': 3000}, {'symbol': 'ETH/USDC', 'action': 'SELL', 'confidence': 0.7, 'price': 3020}, {'symbol': 'ETH/USDC', 'action': 'BUY', 'confidence': 0.6, 'price': 3020}, {'symbol': 'ETH/USDC', 'action': 'SELL', 'confidence': 0.9, 'price': 3040}, ] for signal in test_signals: success = executor.execute_signal( symbol=signal['symbol'], action=signal['action'], confidence=signal['confidence'], current_price=signal['price'] ) logger.info(f"Signal execution: {signal['action']} {signal['symbol']} - {'Success' if success else 'Failed'}") time.sleep(1) # Simulate time between trades # Get updated stats daily_stats = executor.get_daily_stats() rl_learning_stats = daily_stats.get('rl_learning', {}) logger.info(f"RL Learning Stats: {rl_learning_stats}") # Test RL training controls logger.info("Testing RL training controls...") executor.enable_rl_training(False) logger.info("RL training disabled") executor.enable_rl_training(True) logger.info("RL training re-enabled") # Save RL model executor.save_rl_model() logger.info("RL model saved") return True else: logger.error("RL trainer was not initialized in TradingExecutor") return False except Exception as e: logger.error(f"Error testing TradingExecutor integration: {e}") return False def test_market_state_builder(): """Test the market state builder component""" logger.info("\n=== Testing Market State Builder ===") from core.realtime_rl_trainer import MarketStateBuilder state_builder = MarketStateBuilder(state_size=100) # Add some market data logger.info("Adding market data...") for i in range(30): price = 3000 + np.sin(i * 0.1) * 100 + np.random.normal(0, 10) volume = 1000 + np.random.normal(0, 100) rsi = 50 + 30 * np.sin(i * 0.05) + np.random.normal(0, 5) macd = np.sin(i * 0.1) * 10 + np.random.normal(0, 2) state_builder.update_market_data(price, volume, rsi, macd) # Build states for different scenarios scenarios = [ {'position': 'NONE', 'pnl': 0, 'balance': 1000}, {'position': 'LONG', 'pnl': 50, 'balance': 1050}, {'position': 'SHORT', 'pnl': -20, 'balance': 980}, ] for scenario in scenarios: state = state_builder.build_state( current_position=scenario['position'], position_pnl=scenario['pnl'], account_balance=scenario['balance'] ) logger.info(f"State for {scenario['position']} position: " f"Size={len(state)}, Range=[{state.min():.4f}, {state.max():.4f}]") return True def main(): """Run all RL learning tests""" logger.info("Starting Real-Time RL Learning Tests") logger.info("=" * 60) # Test 1: Standalone RL trainer trainer_test = test_rl_trainer_standalone() # Test 2: Market state builder state_test = test_market_state_builder() # Test 3: TradingExecutor integration integration_test = test_trading_executor_integration() # Summary logger.info("\n" + "=" * 60) logger.info("REAL-TIME RL LEARNING TEST SUMMARY:") logger.info(f" Standalone RL Trainer: {'PASS' if trainer_test else 'FAIL'}") logger.info(f" Market State Builder: {'PASS' if state_test else 'FAIL'}") logger.info(f" TradingExecutor Integration: {'PASS' if integration_test else 'FAIL'}") if trainer_test and state_test and integration_test: logger.info("\nALL TESTS PASSED!") logger.info("Your system now features real-time RL learning that:") logger.info(" • Learns from every trade execution and position closure") logger.info(" • Adapts trading decisions based on market outcomes") logger.info(" • Continuously improves decision-making over time") logger.info(" • Tracks performance and learning progress") logger.info(" • Saves and loads trained models automatically") else: logger.warning("\nSome tests failed. Check the logs above for details.") if __name__ == "__main__": main()