""" AHAD QUANT — Paper Trader Simule les ordres Forex sans argent réel. Persiste l'état dans paper_state.json pour que web_ui.py puisse le lire. Interface attendue par ahad_quant.py : PaperTrader(initial_balance) .positions dict { pair: {side, entry, qty, sl, tp, margin, opened_at} } .daily_pnl float .total_pnl float .peak_equity float .consecutive_losses int .circuit_breaker_until float (timestamp) .get_balance() → float .open_position(pair, side, price, qty, sl_pct, tp_pct) → {"success": bool, ...} .close_position(pair, price, reason) → {"pnl": float, ...} .check_sl_tp(pair, price) → "sl" | "tp" | None .reset_daily_pnl() .summary() → str """ import json import os import time from datetime import datetime, timezone from pathlib import Path import config _STATE_FILE = Path(os.path.dirname(__file__)) / "paper_state.json" class PaperTrader: """Moteur de paper trading Forex — persistance JSON.""" def __init__(self, initial_balance: float = 10_000.0): self.initial_balance = initial_balance self._load_state() # ── Persistance ──────────────────────────────────────────────────────────── def _load_state(self): """Charge paper_state.json ou initialise un état vierge.""" try: with open(_STATE_FILE) as f: s = json.load(f) self._balance = float(s.get("balance", self.initial_balance)) self.positions = s.get("positions", {}) self._trades = s.get("trades", []) self.daily_pnl = float(s.get("daily_pnl", 0.0)) self.total_pnl = float(s.get("total_pnl", 0.0)) self.peak_equity = float(s.get("peak_equity", self._balance)) self.daily_losses = int(s.get("daily_losses", 0)) self.consecutive_losses = int(s.get("consecutive_losses", 0)) self.circuit_breaker_until = float(s.get("circuit_breaker_until", 0.0)) except (FileNotFoundError, json.JSONDecodeError, KeyError): self._balance = self.initial_balance self.positions = {} self._trades = [] self.daily_pnl = 0.0 self.total_pnl = 0.0 self.peak_equity = self.initial_balance self.daily_losses = 0 self.consecutive_losses = 0 self.circuit_breaker_until = 0.0 self._save_state() def _save_state(self): """Écrit paper_state.json — lu par web_ui.py.""" state = { "balance": round(self._balance, 5), "positions": self.positions, "trades": self._trades[-500:], # garder les 500 derniers "daily_pnl": round(self.daily_pnl, 5), "total_pnl": round(self.total_pnl, 5), "peak_equity": round(self.peak_equity, 5), "daily_losses": self.daily_losses, "consecutive_losses": self.consecutive_losses, "circuit_breaker_until": self.circuit_breaker_until, "updated_at": datetime.now(timezone.utc).isoformat(), } tmp = str(_STATE_FILE) + ".tmp" with open(tmp, "w") as f: json.dump(state, f, indent=2) os.replace(tmp, _STATE_FILE) # écriture atomique # ── Interface publique ───────────────────────────────────────────────────── def get_balance(self) -> float: """Équité courante (balance + PnL non réalisé des positions ouvertes).""" return round(self._balance, 5) def open_position( self, pair: str, side: str, # "long" | "short" price: float, qty: float, sl_pct: float = config.STOP_LOSS_PCT, tp_pct: float = config.TAKE_PROFIT_PCT, ) -> dict: """Ouvre une position paper. Retourne {"success": bool, "msg": str}.""" if pair in self.positions: return {"success": False, "msg": f"{pair} already open"} side = side.lower() margin = price * qty / config.LEVERAGE if margin > self._balance * config.MAX_MARGIN_USAGE: return {"success": False, "msg": "Insufficient margin"} mult = 1 if side == "long" else -1 sl = round(price * (1 - mult * sl_pct), 6) tp = round(price * (1 + mult * tp_pct), 6) self.positions[pair] = { "side": side, "entry": round(price, 6), "qty": round(qty, 6), "sl": sl, "tp": tp, "margin": round(margin, 5), "opened_at": time.time(), } self._balance -= margin # réserve la marge self._save_state() return {"success": True, "pair": pair, "side": side, "price": price, "qty": qty, "sl": sl, "tp": tp} def close_position(self, pair: str, price: float, reason: str = "manual") -> dict: """Ferme une position et comptabilise le PnL.""" pos = self.positions.pop(pair, None) if not pos: return {"success": False, "pnl": 0.0, "msg": "No position"} mult = 1 if pos["side"] == "long" else -1 pnl = round(mult * (price - pos["entry"]) * pos["qty"] * config.LEVERAGE, 5) fee = round(pos["entry"] * pos["qty"] * config.FEE_RATE, 5) net = round(pnl - fee, 5) self._balance += pos["margin"] + net self.daily_pnl += net self.total_pnl += net self.peak_equity = max(self.peak_equity, self._balance) if net < 0: self.consecutive_losses += 1 self.daily_losses += 1 if self.daily_losses >= config.CIRCUIT_BREAKER_LOSSES: self.circuit_breaker_until = time.time() + config.CIRCUIT_BREAKER_COOLDOWN else: self.consecutive_losses = 0 trade = { "pair": pair, "side": pos["side"], "entry": pos["entry"], "exit": round(price, 6), "qty": pos["qty"], "pnl": net, "fee": fee, "reason": reason, "closed_at": datetime.now(timezone.utc).isoformat(), } self._trades.append(trade) self._save_state() return {"success": True, "pnl": net, "reason": reason, "trade": trade} def check_sl_tp(self, pair: str, price: float): """Retourne 'sl', 'tp' ou None selon le prix actuel.""" pos = self.positions.get(pair) if not pos: return None side = pos["side"] sl, tp = pos["sl"], pos["tp"] if side == "long": if price <= sl: return "sl" if price >= tp: return "tp" else: # short if price >= sl: return "sl" if price <= tp: return "tp" return None def reset_daily_pnl(self): """Appelé par ahad_quant.py à minuit pour réinitialiser les stats journalières.""" self.daily_pnl = 0.0 self.daily_losses = 0 # Le circuit breaker journalier expire aussi if self.circuit_breaker_until < time.time(): self.circuit_breaker_until = 0.0 self._save_state() def summary(self) -> str: """Résumé texte affiché dans les logs du bot.""" wins = sum(1 for t in self._trades if t.get("pnl", 0) > 0) total = len(self._trades) wr = f"{wins/total*100:.1f}%" if total else "—" return ( f"[PAPER] Balance: ${self._balance:,.2f} | " f"Total PnL: {'+' if self.total_pnl >= 0 else ''}{self.total_pnl:.2f} | " f"Daily: {'+' if self.daily_pnl >= 0 else ''}{self.daily_pnl:.2f} | " f"Win rate: {wr} ({total} trades) | " f"Positions: {len(self.positions)}" )