matitos
03dfa369e5
- RiskManager halts + flattens on max daily loss / drawdown, caps position size and trades per day; adds a manual kill switch (POST /api/bot/kill). - Telegram and webhook notifications on trades, errors and halts. - Optional API bearer token on /api, injected server-side by the frontend nginx so the browser never sees it. - Kraken and Coinbase market data (public); Coinbase live broker. Live on an unsupported exchange now errors instead of silently falling back to paper. - Strategy presets backed by the StrategyConfig table (CRUD API + UI). - Idempotent additive-column migration so existing SQLite databases upgrade in place (fixes the new strategy_configs.name column).
104 lines
3.5 KiB
Python
104 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime
|
|
|
|
from app.config import Settings
|
|
|
|
|
|
class RiskManager:
|
|
"""Tracks per-session and per-day limits and can halt trading.
|
|
|
|
A halt is sticky until ``reset`` is called (which happens on engine start).
|
|
All limits are optional; a value of ``0`` disables that limit.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
max_daily_loss: float = 0.0,
|
|
max_drawdown_pct: float = 0.0,
|
|
max_position_pct: float = 1.0,
|
|
max_trades_per_day: int = 0,
|
|
) -> None:
|
|
self.max_daily_loss = max_daily_loss
|
|
self.max_drawdown_pct = max_drawdown_pct
|
|
self.max_position_pct = max_position_pct
|
|
self.max_trades_per_day = max_trades_per_day
|
|
self.reset(0.0)
|
|
|
|
@classmethod
|
|
def from_settings(cls, settings: Settings, starting_equity: float) -> RiskManager:
|
|
manager = cls(
|
|
max_daily_loss=settings.max_daily_loss,
|
|
max_drawdown_pct=settings.max_drawdown_pct,
|
|
max_position_pct=settings.max_position_pct,
|
|
max_trades_per_day=settings.max_trades_per_day,
|
|
)
|
|
manager.reset(starting_equity)
|
|
return manager
|
|
|
|
def reset(self, starting_equity: float) -> None:
|
|
self.starting_equity = starting_equity
|
|
self.peak_equity = starting_equity
|
|
self.last_equity = starting_equity
|
|
self.day_start_equity = starting_equity
|
|
self.day = datetime.now(UTC).date()
|
|
self.trades_today = 0
|
|
self.halted = False
|
|
self.halt_reason: str | None = None
|
|
|
|
@property
|
|
def day_pnl(self) -> float:
|
|
return self.last_equity - self.day_start_equity
|
|
|
|
@property
|
|
def drawdown_pct(self) -> float:
|
|
if self.peak_equity <= 0:
|
|
return 0.0
|
|
return max(0.0, (self.peak_equity - self.last_equity) / self.peak_equity * 100)
|
|
|
|
def _roll_day(self, equity: float) -> None:
|
|
today = datetime.now(UTC).date()
|
|
if today != self.day:
|
|
self.day = today
|
|
self.day_start_equity = equity
|
|
self.trades_today = 0
|
|
|
|
def update(self, equity: float) -> str | None:
|
|
"""Update with the latest equity; return a halt reason if a limit tripped.
|
|
|
|
Detection only — the caller decides when to actually halt (so it can
|
|
flatten first). Use ``halt()`` to make it sticky.
|
|
"""
|
|
self.last_equity = equity
|
|
self._roll_day(equity)
|
|
self.peak_equity = max(self.peak_equity, equity)
|
|
if self.halted:
|
|
return self.halt_reason
|
|
if self.max_daily_loss > 0 and self.day_pnl <= -self.max_daily_loss:
|
|
return f"daily loss limit hit (P&L {self.day_pnl:.2f})"
|
|
if self.max_drawdown_pct > 0 and self.drawdown_pct >= self.max_drawdown_pct:
|
|
return f"max drawdown hit ({self.drawdown_pct:.2f}%)"
|
|
return None
|
|
|
|
def halt(self, reason: str) -> str:
|
|
self.halted = True
|
|
self.halt_reason = reason
|
|
return reason
|
|
|
|
def can_enter(self) -> bool:
|
|
return self.entry_block_reason() is None
|
|
|
|
def entry_block_reason(self) -> str | None:
|
|
if self.halted:
|
|
return self.halt_reason
|
|
if self.max_trades_per_day > 0 and self.trades_today >= self.max_trades_per_day:
|
|
return f"max trades per day reached ({self.max_trades_per_day})"
|
|
return None
|
|
|
|
def record_entry(self) -> None:
|
|
self.trades_today += 1
|
|
|
|
def clamp_size(self, size: float) -> float:
|
|
return max(0.0, min(size, self.max_position_pct))
|