Abdoul Ahad Binizi
212 lines
8.3 KiB
Python
212 lines
8.3 KiB
Python
"""
|
|
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)}"
|
|
)
|