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).
64 lines
1.7 KiB
Python
64 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from app.engine.risk import RiskManager
|
|
|
|
|
|
def test_daily_loss_detects_and_halts() -> None:
|
|
risk = RiskManager(max_daily_loss=100.0)
|
|
risk.reset(10_000)
|
|
assert risk.update(9_950) is None
|
|
reason = risk.update(9_900)
|
|
assert reason is not None and "daily loss" in reason
|
|
assert risk.halted is False # detection only; caller decides
|
|
risk.halt(reason)
|
|
assert risk.halted is True
|
|
assert risk.can_enter() is False
|
|
|
|
|
|
def test_drawdown_halts() -> None:
|
|
risk = RiskManager(max_drawdown_pct=10.0)
|
|
risk.reset(10_000)
|
|
risk.update(11_000)
|
|
assert risk.update(10_000) is None
|
|
reason = risk.update(9_850)
|
|
assert reason is not None and "drawdown" in reason
|
|
risk.halt(reason)
|
|
assert risk.halted
|
|
|
|
|
|
def test_halt_is_sticky() -> None:
|
|
risk = RiskManager(max_daily_loss=100.0)
|
|
risk.reset(10_000)
|
|
reason = risk.update(9_800)
|
|
risk.halt(reason or "loss")
|
|
assert risk.halted
|
|
# A recovery must not silently resume trading.
|
|
assert risk.update(10_500) is not None
|
|
assert risk.halted
|
|
|
|
|
|
def test_max_trades_per_day() -> None:
|
|
risk = RiskManager(max_trades_per_day=2)
|
|
risk.reset(10_000)
|
|
assert risk.can_enter()
|
|
risk.record_entry()
|
|
risk.record_entry()
|
|
assert risk.can_enter() is False
|
|
assert "max trades" in (risk.entry_block_reason() or "")
|
|
|
|
|
|
def test_clamp_size() -> None:
|
|
risk = RiskManager(max_position_pct=0.25)
|
|
assert risk.clamp_size(1.0) == pytest.approx(0.25)
|
|
assert risk.clamp_size(0.1) == pytest.approx(0.1)
|
|
assert risk.clamp_size(-1.0) == pytest.approx(0.0)
|
|
|
|
|
|
def test_disabled_limits_never_halt() -> None:
|
|
risk = RiskManager()
|
|
risk.reset(10_000)
|
|
assert risk.update(1.0) is None
|
|
assert risk.can_enter()
|