# simulation_engine/virtual_exchange.py # V2.0 — Single-position engine with cooldown and soft/hard time-stops (≈1h / 2h) from datetime import datetime import uuid class VirtualExchange: def __init__( self, initial_balance=10.0, fee_rate=0.001, tp_pct=1.6, sl_pct=0.8, bar_ms=300_000, # 5m cooldown_bars=1, min_trade_usd=0.10, position_fraction=0.45, max_concurrent=1, soft_time_stop_bars=12, # ~1h at 5m hard_time_stop_bars=24, # ~2h at 5m ): self.initial_balance = float(initial_balance) self.balance = float(initial_balance) self.fee_rate = float(fee_rate) self.tp_pct = float(tp_pct) / 100.0 self.sl_pct = float(sl_pct) / 100.0 self.bar_ms = int(bar_ms) if bar_ms else None self.cooldown_bars = int(cooldown_bars) if cooldown_bars is not None else 0 self.min_trade_usd = float(min_trade_usd) self.position_fraction = float(position_fraction) self.max_concurrent = int(max_concurrent) self.soft_time_stop_bars = int(soft_time_stop_bars) if soft_time_stop_bars else None self.hard_time_stop_bars = int(hard_time_stop_bars) if hard_time_stop_bars else None self.positions = {} # symbol -> position dict self.trade_history = [] self.cooldowns = {} # symbol -> cooldown_until_ts self.metrics = { "wins": 0, "losses": 0, "total_pnl_usd": 0.0, "max_drawdown": 0.0, "peak_balance": self.balance } # ------- Helpers ------- def open_positions_count(self): return len(self.positions) def get_balance(self): return self.balance def can_enter(self, symbol, current_ts): cd_until = self.cooldowns.get(symbol) if cd_until is not None and current_ts < cd_until: return False if self.open_positions_count() >= self.max_concurrent: return False if symbol in self.positions: return False planned = self.balance * self.position_fraction if planned < self.min_trade_usd: return False return True # ------- Lifecycle ------- def update_positions(self, current_prices, timestamp): closed_trades = [] for symbol in list(self.positions.keys()): pos = self.positions[symbol] current_price = current_prices.get(symbol) if not current_price: continue entry = pos["entry_price"] pos["highest_price"] = max(pos.get("highest_price", entry), current_price) tp_price = entry * (1.0 + self.tp_pct) sl_price = entry * (1.0 - self.sl_pct) age_ms = timestamp - pos["entry_time"] age_bars = int(age_ms // self.bar_ms) if self.bar_ms else 0 closed_trade = None # 1) Price-based exits if current_price >= tp_price: closed_trade = self._close(symbol, tp_price, timestamp, "TAKE_PROFIT") elif current_price <= sl_price: closed_trade = self._close(symbol, sl_price, timestamp, "STOP_LOSS") else: # 2) Time-based exits: soft then hard if self.soft_time_stop_bars and age_bars >= self.soft_time_stop_bars: if current_price >= entry: closed_trade = self._close(symbol, current_price, timestamp, "SOFT_TIME_STOP") if not closed_trade and self.hard_time_stop_bars and age_bars >= self.hard_time_stop_bars: closed_trade = self._close(symbol, current_price, timestamp, "HARD_TIME_STOP") if closed_trade: closed_trades.append(closed_trade) return closed_trades def execute_buy(self, symbol, price, timestamp, score_data=None): if price <= 0: return False if self.open_positions_count() >= self.max_concurrent: return False if symbol in self.positions: return False amount_usd = self.balance * self.position_fraction if amount_usd < self.min_trade_usd or amount_usd > self.balance: return False fee = amount_usd * self.fee_rate net_invested = amount_usd - fee if net_invested <= 0: return False quantity = net_invested / price self.balance -= amount_usd self.positions[symbol] = { "entry_price": price, "quantity": quantity, "invested_usd": amount_usd, "entry_time": timestamp, "scores": score_data or {}, "highest_price": price, "trade_id": f"sim_{uuid.uuid4().hex[:8]}", } return True def execute_sell(self, symbol, price, timestamp, reason): return self._close(symbol, price, timestamp, reason) # ------- Internal close ------- def _close(self, symbol, price, timestamp, reason): pos = self.positions.get(symbol) if not pos: return None revenue = pos["quantity"] * price fee = revenue * self.fee_rate net_revenue = revenue - fee self.balance += net_revenue pnl_usd = net_revenue - pos["invested_usd"] pnl_pct = (pnl_usd / pos["invested_usd"]) * 100.0 entry_score = None try: entry_score = float((pos.get("scores") or {}).get("final_score")) except Exception: entry_score = None trade_record = { "id": pos["trade_id"], "symbol": symbol, "status": "CLOSED", "entry_time": datetime.fromtimestamp(pos["entry_time"] / 1000).isoformat(), "close_time": datetime.fromtimestamp(timestamp / 1000).isoformat(), "entry_price": pos["entry_price"], "close_price": price, "pnl_usd": pnl_usd, "pnl_percent": pnl_pct, "close_reason": reason, "strategy": "HYBRID_TITAN", "score": entry_score, "decision_data": {"components": pos.get("scores") or {}} } self.trade_history.append(trade_record) self.metrics["total_pnl_usd"] += pnl_usd if pnl_usd > 0: self.metrics["wins"] += 1 else: self.metrics["losses"] += 1 self.metrics["peak_balance"] = max(self.metrics["peak_balance"], self.balance) current_dd = (self.metrics["peak_balance"] - self.balance) / max(self.metrics["peak_balance"], 1e-9) self.metrics["max_drawdown"] = max(self.metrics["max_drawdown"], current_dd) # تبريد لكل رمز if self.cooldown_bars and self.bar_ms: self.cooldowns[symbol] = timestamp + self.cooldown_bars * self.bar_ms else: self.cooldowns[symbol] = None del self.positions[symbol] return trade_record