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
|