66 lines
2.9 KiB
Python
66 lines
2.9 KiB
Python
import math
|
|
from datetime import datetime, timezone
|
|
from utils.logger import get_logger
|
|
|
|
log = get_logger()
|
|
|
|
class RiskManager:
|
|
def __init__(self, cfg: dict):
|
|
self.cfg = cfg
|
|
prop = cfg.get("prop_mode", {})
|
|
self.prop = prop.get("enabled", False)
|
|
self.initial = prop.get("initial_balance")
|
|
self.daily_pct = prop["daily_loss_pct"] if self.prop else cfg["daily_loss_pct"]
|
|
self.dd_pct = prop["max_drawdown_pct"] if self.prop else cfg["max_drawdown_pct"]
|
|
self.day = None
|
|
self.day_start_equity = None
|
|
self.peak = None
|
|
self.halted_reason = None
|
|
|
|
def update(self, balance: float, equity: float):
|
|
today = datetime.now(timezone.utc).date()
|
|
if self.day != today:
|
|
self.day, self.day_start_equity, self.halted_reason = today, max(balance, equity), None
|
|
log.info(f"Nouveau jour: equity de référence {self.day_start_equity:.2f}")
|
|
self.peak = max(self.peak or equity, equity)
|
|
|
|
def check(self, equity: float):
|
|
"""Retourne (ok, raison)."""
|
|
daily_loss = (self.day_start_equity - equity) / self.day_start_equity * 100
|
|
if daily_loss >= self.daily_pct:
|
|
return False, f"Perte journalière {daily_loss:.2f}% >= {self.daily_pct}%"
|
|
ref = self.initial if (self.prop and self.initial) else self.peak
|
|
dd = (ref - equity) / ref * 100
|
|
if dd >= self.dd_pct:
|
|
return False, f"Drawdown {dd:.2f}% >= {self.dd_pct}%"
|
|
return True, ""
|
|
|
|
def lot_size(self, equity: float, sl_distance: float, info) -> float:
|
|
risk_money = equity * self.cfg["risk_per_trade_pct"] / 100
|
|
loss_per_lot = sl_distance / info.trade_tick_size * info.trade_tick_value
|
|
if loss_per_lot <= 0:
|
|
return 0.0
|
|
lots = risk_money / loss_per_lot
|
|
step = info.volume_step
|
|
lots = math.floor(lots / step) * step
|
|
return round(max(info.volume_min, min(lots, info.volume_max)), 2)
|
|
|
|
def can_open(self, positions_all, positions_symbol) -> bool:
|
|
return (len(positions_all) < self.cfg["max_open_trades"]
|
|
and len(positions_symbol) < self.cfg["max_per_symbol"])
|
|
|
|
def new_sl(self, pos, atr_value: float, price: float):
|
|
"""Break-even puis trailing. Retourne un nouveau SL ou None."""
|
|
buy = pos.type == 0
|
|
profit = (price - pos.price_open) if buy else (pos.price_open - price)
|
|
cands = []
|
|
if profit >= self.cfg["breakeven_atr"] * atr_value:
|
|
cands.append(pos.price_open)
|
|
if profit >= self.cfg["trail_atr"] * atr_value:
|
|
cands.append(price - self.cfg["trail_atr"] * atr_value if buy else price + self.cfg["trail_atr"] * atr_value)
|
|
if not cands:
|
|
return None
|
|
cand = max(cands) if buy else min(cands)
|
|
cur = pos.sl
|
|
better = cand > cur if buy else (cur == 0 or cand < cur)
|
|
return cand if better else None
|