from datetime import datetime from typing import Dict, Optional, List import logging, numpy as np from config import WaveStrategyConfig from wave_strategy import WaveRangeStrategy, WaveSignal, SignalType from position_manager import PositionManager, Position, PositionSide, PositionStatus logger = logging.getLogger(__name__) class VirtualExchange: def __init__(self, config: WaveStrategyConfig): self.config = config self.balance = getattr(config, 'INITIAL_BALANCE', 10000.0) self.position_manager = PositionManager(config) self.strategy = WaveRangeStrategy(config) self.trade_history: List[Dict] = [] self.current_prices: Dict[str, float] = {} self.risk_type, self.risk_percent = 'percent', getattr(config, 'risk_calc_percent', 1.5) self.risk_amount, self.lot_size = getattr(config, 'risk_calc_amount', None), getattr(config, 'position_lot_size', 1.0) self._last_trade_bar = -1 self._last_exit_bar = -1 self._consecutive_losses = 0 def set_risk_params(self, risk_type=None, risk_percent=None, risk_amount=None, lot_size=None): if risk_type: self.risk_type = risk_type if risk_percent is not None: self.risk_percent = risk_percent if risk_amount is not None: self.risk_amount = risk_amount if lot_size is not None: self.lot_size = lot_size def calculate_position_size(self, account, entry_price, stop_price, risk_type=None, risk_percent=None, risk_amount=None, lot_size=None): risk_type = risk_type or self.risk_type risk_percent = risk_percent if risk_percent is not None else self.risk_percent risk_amount = risk_amount if risk_amount is not None else self.risk_amount lot_size = lot_size or self.lot_size if entry_price <= 0 or stop_price <= 0 or account <= 0: return {'error': 'Невалидные данные'} stop_distance = abs(entry_price - stop_price) stop_percent = (stop_distance / entry_price) * 100 calc_risk = account * (risk_percent / 100) if risk_type == 'percent' else (risk_amount or account * 0.015) position_value = (calc_risk / stop_percent) * 100 if not np.isfinite(position_value) or position_value <= 0: return {'error': 'Ошибка расчета'} lots = position_value / (entry_price * lot_size) return {'stop_percent': round(stop_percent, 3), 'risk_amount': round(calc_risk, 2), 'position_value': round(position_value, 2), 'lots': round(lots, 4), 'entry_price': entry_price, 'stop_price': stop_price} def execute_signal(self, signal: WaveSignal, current_bar_idx: int = -1) -> Optional[Position]: cooldown = getattr(self.config, 'signal_cooldown_bars', 0) min_hold = getattr(self.config, 'min_hold_bars', 0) last_action_bar = max(self._last_trade_bar, self._last_exit_bar) # 🔑 ЦЕПКОЙ БРЕЙКЕР: блокировка после серии потерь max_losses = getattr(self.config, 'max_consecutive_losses', 999) if self._consecutive_losses >= max_losses: return None if current_bar_idx - last_action_bar < cooldown: return None existing = self.position_manager.get_position(signal.instrument, signal.timeframe, PositionSide.LONG if signal.signal == SignalType.BUY else PositionSide.SHORT) if existing: return None calc = self.calculate_position_size(self.balance, signal.price, signal.stop_loss) if 'error' in calc: return None position_size = calc['lots'] * self.lot_size position = Position( id=f"{signal.instrument}_{int(datetime.now().timestamp()*1000)}", instrument=signal.instrument, timeframe=signal.timeframe, side=PositionSide.LONG if signal.signal == SignalType.BUY else PositionSide.SHORT, entry_price=signal.price, size=position_size, stop_loss=signal.stop_loss, take_profit=signal.take_profit, atr_at_entry=signal.atr, risk_amount=signal.risk_amount, opened_at=datetime.now(), entry_bar_idx=current_bar_idx) position.wave_progress_pct = signal.wave_progress_pct self.balance -= signal.price * position_size * 0.001 self.position_manager.add_position(position) self._last_trade_bar = current_bar_idx return position def update_price(self, instrument, close, high, low, atr, current_bar_idx=-1): self.current_prices[instrument] = close closed = [] for pos in list(self.position_manager.get_open_positions()): if pos.instrument != instrument: continue if hasattr(pos, 'entry_bar_idx') and (current_bar_idx - pos.entry_bar_idx) < getattr(self.config, 'min_hold_bars', 0): continue pos.update_extremes(high, low) # 🔑 ТРЕЙЛИНГ ПОЛНОСТЬЮ ОТКЛЮЧЁН, ЕСЛИ trail_enabled=False if self.config.trail_enabled: risk_per_unit = abs(pos.entry_price - pos.stop_loss) or (pos.entry_price * 0.001) pnl_r = ((close - pos.entry_price) / risk_per_unit) if pos.side == PositionSide.LONG else ((pos.entry_price - close) / risk_per_unit) be_r = getattr(self.config, 'break_even_r', 0.8) if not pos.trail_activated and pnl_r >= be_r: pos.stop_loss = pos.entry_price pos.trail_activated = True if pos.trail_activated: trail_dist = atr * getattr(self.config, 'trail_distance_atr', 0.7) new_sl = (pos.highest_price - trail_dist) if pos.side == PositionSide.LONG else (pos.lowest_price + trail_dist) if (pos.side == PositionSide.LONG and new_sl > pos.stop_loss) or (pos.side == PositionSide.SHORT and new_sl < pos.stop_loss): pos.stop_loss = round(new_sl, 8) exit_cond = self.strategy.check_exit( {'side': pos.side.value, 'stop_loss': pos.stop_loss, 'take_profit': pos.take_profit}, close, high, low, getattr(pos, 'wave_progress_pct', 0)) if exit_cond: reason, exit_price = exit_cond pnl = self.position_manager.close_position(pos, exit_price, reason, 0.001) self.balance += pnl self._last_exit_bar = current_bar_idx # 🔑 Обновление счётчика потерь if pnl < 0: self._consecutive_losses += 1 else: self._consecutive_losses = 0 closed.append({'id': pos.id, 'exit_price': exit_price, 'reason': reason, 'pnl': pnl}) return closed