mt5_python_ea_suite/execution/realtime_trader.py

import time
import signal
import sys
import json
import pandas as pd
from datetime import datetime
from logger import logger
from config import SYMBOL, TIMEFRAME, REALTIME_CONFIG
from core.risk import RiskController
from strategies.wave_theory import WaveTheoryStrategy
from strategies.ma_cross import MACrossStrategy
from strategies.rsi import RSIStrategy
from strategies.bollinger import BollingerStrategy
from strategies.mean_reversion import MeanReversionStrategy
from strategies.momentum_breakout import MomentumBreakoutStrategy
from strategies.macd import MACDStrategy
from strategies.kdj import KDJStrategy
from strategies.turtle import TurtleStrategy
from strategies.daily_breakout import DailyBreakoutStrategy

class RealtimeTrader:
    """实时交易器 (已重构为依赖注入和市场状态参数切换)"""
    
    def __init__(self, data_provider, update_interval=60):
        self.data_provider = data_provider
        self.update_interval = update_interval
        self.running = False
        self.risk_controller = None
        
        self.regime_params = {}
        self.regime_detector = None
        self.current_regime = 'Ranging'  # Default regime
        self.strategy_instances = {}
        self.current_weights = {}
        self.signal_thresholds = {}

        self.strategy_config_key_to_class = {
            "ma_cross": MACrossStrategy,
            "rsi": RSIStrategy,
            "bollinger": BollingerStrategy,
            "mean_reversion": MeanReversionStrategy,
            "momentum_breakout": MomentumBreakoutStrategy,
            "macd": MACDStrategy,
            "kdj": KDJStrategy,
            "turtle": TurtleStrategy,
            "daily_breakout": DailyBreakoutStrategy,
            "wave_theory": WaveTheoryStrategy
        }

    def _initialize(self):
        if not self.data_provider.initialize():
            return False
        
        self.risk_controller = RiskController(self.data_provider)
        
        # Load regime parameters
        try:
            with open('regime_optimal_params.json', 'r') as f:
                self.regime_params = json.load(f)
            logger.info("成功加载市场状态参数文件: regime_optimal_params.json")
        except FileNotFoundError:
            logger.error("错误: 未找到 regime_optimal_params.json。请先运行 regime_optimizer.py。")
            return False
        except json.JSONDecodeError:
            logger.error("错误: regime_optimal_params.json 文件格式不正确。")
            return False

        # Instantiate regime detector
        self.regime_detector = WaveTheoryStrategy(self.data_provider, SYMBOL, TIMEFRAME)
        
        # Create strategy instances
        self._create_strategy_instances()

        self.risk_controller.sync_state()

        signal.signal(signal.SIGINT, self._signal_handler)
        signal.signal(signal.SIGTERM, self._signal_handler)
        logger.info("实时交易系统初始化完成")
        return True

    def _create_strategy_instances(self):
        for name, strategy_class in self.strategy_config_key_to_class.items():
            # Use the name that corresponds to the keys in the parameter file (e.g., 'MACrossStrategy')
            class_name = strategy_class.__name__
            self.strategy_instances[class_name] = strategy_class(self.data_provider, SYMBOL, TIMEFRAME)
        logger.info(f"创建了 {len(self.strategy_instances)} 个策略实例")

    def _update_parameters_for_regime(self):
        # 1. Determine current regime
        df_history = self.data_provider.get_historical_data(SYMBOL, TIMEFRAME, count=200)
        if df_history is None or df_history.empty:
            logger.warning("无法获取历史数据来判断市场状态，将使用上一个状态。")
            return

        df_history = pd.DataFrame(df_history)
        df_history.set_index(pd.to_datetime(df_history['time'], unit='s'), inplace=True)

        df_with_indicators = self.regime_detector._calculate_indicators(df_history.copy())
        
        last_row = df_with_indicators.iloc[-1]
        new_regime = "Ranging"  # Default
        if not pd.isna(last_row['adx']):
            if last_row['adx'] < self.regime_detector.adx_threshold:
                new_regime = "Ranging"
            elif last_row['ema_short'] > last_row['ema_medium'] > last_row['ema_long']:
                new_regime = "Uptrend"
            else:
                new_regime = "Downtrend"

        # 2. If regime changed, update parameters
        if new_regime != self.current_regime:
            self.current_regime = new_regime
            logger.info(f"市场状态已切换为: {self.current_regime}")
            
            regime_config = self.regime_params.get(self.current_regime)
            if not regime_config:
                logger.error(f"在参数文件中未找到状态 {self.current_regime} 的配置，将使用默认参数。")
                return

            params = regime_config.get('best_parameters', {})
            
            # Group parameters by component (strategy, risk, etc.)
            strategy_params = {strat_name: {} for strat_name in self.strategy_instances.keys()}
            risk_params = {}
            temp_weights = {}
            
            prefix_map = {
                'MACrossStrategy': 'ma_cross_',
                'RSIStrategy': 'rsi_',
                'BollingerStrategy': 'bollinger_',
                'MACDStrategy': 'macd_',
                'MeanReversionStrategy': 'mean_reversion_',
                'MomentumBreakoutStrategy': 'momentum_breakout_',
                'KDJStrategy': 'kdj_',
                'TurtleStrategy': 'turtle_',
                'DailyBreakoutStrategy': 'daily_breakout_',
                'WaveTheoryStrategy': 'wave_'
            }

            for key, value in params.items():
                if key.startswith('weight_'):
                    strategy_name = key.replace('weight_', '')
                    temp_weights[strategy_name] = value
                elif key in ['buy_threshold', 'sell_threshold']:
                    self.signal_thresholds[key] = value
                elif hasattr(self.risk_controller, key):
                    risk_params[key] = value
                else:
                    for strat_name, prefix in prefix_map.items():
                        if key.startswith(prefix):
                            param_name = key.replace(prefix, '')
                            # BUG FIX: Handle the 'wave_period' special case
                            if strat_name == 'WaveTheoryStrategy' and key == 'wave_period':
                                param_name = 'wave_period'
                            strategy_params[strat_name][param_name] = value
                            break

            # Update parameters in a batch
            for strat_name, params_dict in strategy_params.items():
                if params_dict:
                    self.strategy_instances[strat_name].set_params(params_dict)
            
            if risk_params:
                for key, value in risk_params.items():
                    setattr(self.risk_controller, key, value)

            self.current_weights = temp_weights
            logger.info(f"已为 {self.current_regime} 状态加载新参数和权重。")
        
    def _signal_handler(self, signum, frame):
        logger.info(f"接收到信号 {signum}，准备退出...")
        self.stop()

    def _run_cycle(self):
        try:
            self.risk_controller.sync_state()
            
            # Update parameters based on current market regime
            self._update_parameters_for_regime()

            current_price = self.data_provider.get_current_price(SYMBOL)
            if not current_price:
                logger.warning("无法获取当前价格，跳过本次循环")
                return

            signals, weights = [], []
            strategies_with_weights = []
            
            for strat_name, strat_instance in self.strategy_instances.items():
                weight = self.current_weights.get(strat_name, 0.0)
                if weight > 0: # Only calculate signal if weight is positive
                    strategies_with_weights.append((strat_instance, weight))
                    signals.append(strat_instance.generate_signal())
                    weights.append(weight)

            if not strategies_with_weights: return

            # 打印详细信号日志
            logger.info(f"--- 信号计算详情 (状态: {self.current_regime}) ---")
            for i, (strat, weight) in enumerate(strategies_with_weights):
                signal = signals[i]
                weighted_signal = signal * weight
                strat_name = strat.name
                logger.info(f"  策略: {strat_name:<25} | 信号: {signal:6.2f} | 权重: {weight:6.2f} | 加权信号: {weighted_signal:6.2f}")
            logger.info("--------------------")

            weighted_signal_sum = sum(s * w for s, w in zip(signals, weights))
            
            buy_threshold = self.signal_thresholds.get('buy_threshold', 1.5)
            sell_threshold = self.signal_thresholds.get('sell_threshold', -1.5)
            logger.info(f"加权信号: {weighted_signal_sum:.2f} (买入阈值: {buy_threshold}, 卖出阈值: {sell_threshold})")
            
            direction = None
            if weighted_signal_sum > buy_threshold:
                direction = "buy"
            elif weighted_signal_sum < sell_threshold:
                direction = "sell"
            
            if direction:
                logger.info(f"准备执行{direction}交易，信号强度: {weighted_signal_sum:.2f}")
                success = self.risk_controller.process_trading_signal(direction, current_price, weighted_signal_sum)
                if not success:
                    logger.warning(f"{direction}交易执行失败")
                else:
                    logger.info(f"{direction}交易执行成功")
            
            self.risk_controller.monitor_positions(current_price)

            # --- 状态汇总日志 ---
            logger.info("--- 财务状况更新 ---")
            open_positions = self.risk_controller.position_manager.positions
            if not open_positions:
                logger.info("  当前无持仓")
            else:
                logger.info(f"  当前持仓: {len(open_positions)} 个")
                for pos in open_positions:
                    pnl_pct = self.risk_controller.position_manager._calculate_pnl_pct(pos, current_price['last'])
                    holding_time = datetime.fromtimestamp(current_price['time']) - pos['entry_time']
                    holding_minutes = holding_time.total_seconds() / 60
                    logger.info(f"    - Ticket {pos['ticket']}: {pos['position_type']} {pos['symbol']} @ {pos['entry_price']:.2f} | 持仓时间: {holding_minutes:.1f}分钟 | 浮动盈亏: {pnl_pct:.2%}")

            trade_summary = self.risk_controller.position_manager.get_trade_summary()
            if trade_summary and trade_summary['total_trades'] > 0:
                logger.info("  已平仓交易摘要:")
                logger.info(f"    - 总交易: {trade_summary['total_trades']}, 盈利: {trade_summary['winning_trades']}, 亏损: {trade_summary['losing_trades']}, 胜率: {trade_summary['win_rate']:.2f}%")
                logger.info(f"    - 总净盈亏: ${trade_summary['total_profit_loss']:.2f}")

            logger.info(f"  总权益: ${self.risk_controller.position_manager.total_equity:.2f}")
            logger.info("----------------------")

        except Exception as e:
            import traceback
            logger.error(f"交易周期执行失败: {e}\n{traceback.format_exc()}")
            
    def start(self):
        if not self._initialize(): return
            
        logger.info("=== 启动实时交易系统 ===")
        self.running = True
        
        while self.running:
            cycle_start = time.time()
            self._run_cycle()
            cycle_time = time.time() - cycle_start
            wait_time = max(0, self.update_interval - cycle_time)
            if wait_time > 0: time.sleep(wait_time)
                    
    def stop(self):
        logger.info("=== 停止实时交易系统 ===")
        self.running = False
        try:
            if self.risk_controller:
                timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
                self.risk_controller.save_trade_history(f"realtime_trades_{timestamp}")
        except Exception as e:
            logger.error(f"保存交易记录失败: {e}")
        finally:
            self.data_provider.shutdown()
            logger.info("实时交易系统已停止")
            sys.exit(0)