sevenhill/TradeBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Make symbol risk guard configurable

Browse Source
This commit is contained in:
Codex
2026-06-27 17:28:52 +03:00
parent 7d84bf38cc
commit 57d9514eec
8 changed files with 76 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env
+1
View File
@@ -51,6 +51,7 @@ KELLY_FRACTION=0.25
KELLY_MAX_FRACTION=0.20
RISK_PER_TRADE_PERCENT=0.01
RISK_GUARD_ENABLED=true
RISK_SYMBOL_GUARD_ENABLED=false
RISK_RECENT_TRADE_WINDOW=20
RISK_MAX_CONSECUTIVE_LOSSES=4
RISK_MIN_RECENT_PROFIT_FACTOR=0.85
README.md
+1
View File
@@ -164,6 +164,7 @@ KELLY_FRACTION=0.25
KELLY_MAX_FRACTION=0.20
RISK_PER_TRADE_PERCENT=0.01
RISK_GUARD_ENABLED=true
RISK_SYMBOL_GUARD_ENABLED=false
RISK_RECENT_TRADE_WINDOW=20
RISK_MAX_CONSECUTIVE_LOSSES=4
RISK_MIN_RECENT_PROFIT_FACTOR=0.85
crypto_spot_bot/analytics.py
+14 -7
View File
@@ -112,6 +112,9 @@ def risk_guard_snapshot(
"enabled": False,
"block_new_entries": False,
"position_size_multiplier": 1.0,
"symbol_guard_enabled": settings.risk_symbol_guard_enabled,
"blocked_symbols": [],
"symbols": [],
"reasons": [],
}
active_trades = _active_universe_trades(settings, closed_trades)
@@ -145,6 +148,7 @@ def risk_guard_snapshot(
"reasons": all_reasons,
"global_reasons": reasons,
"degraded_reasons": degraded_reasons,
"symbol_guard_enabled": settings.risk_symbol_guard_enabled,
"blocked_symbols": blocked_symbols,
"symbols": symbol_stats,
"consecutive_losses": consecutive_losses,
@@ -202,6 +206,7 @@ def _active_universe_trades(settings: Settings, trades: list[dict[str, Any]]) ->
def _symbol_guard_stats(settings: Settings, trades: list[dict[str, Any]]) -> list[dict[str, Any]]:
expectancy_min_samples = max(6, min(settings.risk_recent_trade_window, 10))
loss_streak_min_samples = max(3, settings.risk_max_consecutive_losses)
symbol_guard_enabled = settings.risk_symbol_guard_enabled
rows: list[dict[str, Any]] = []
for symbol in settings.symbols:
symbol_trades = [trade for trade in trades if str(trade.get("symbol", "")).upper() == symbol.upper()]
@@ -209,17 +214,19 @@ def _symbol_guard_stats(settings: Settings, trades: list[dict[str, Any]]) -> lis
stats = _trade_stats(recent)
losses = _consecutive_losses(recent)
reasons: list[str] = []
if stats["trades"] >= expectancy_min_samples:
if stats["profit_factor"] < settings.risk_min_recent_profit_factor and stats["avg_net_percent"] <= 0:
reasons.append("symbol_expectancy_negative")
if stats["trades"] >= loss_streak_min_samples:
if losses >= settings.risk_max_consecutive_losses:
reasons.append("symbol_consecutive_losses")
if symbol_guard_enabled:
if stats["trades"] >= expectancy_min_samples:
if stats["profit_factor"] < settings.risk_min_recent_profit_factor and stats["avg_net_percent"] <= 0:
reasons.append("symbol_expectancy_negative")
if stats["trades"] >= loss_streak_min_samples:
if losses >= settings.risk_max_consecutive_losses:
reasons.append("symbol_consecutive_losses")
rows.append(
{
"symbol": symbol.upper(),
"block_new_entries": bool(reasons),
"block_new_entries": bool(reasons) if symbol_guard_enabled else False,
"reasons": reasons,
"symbol_guard_enabled": symbol_guard_enabled,
"consecutive_losses": losses,
**stats,
}
crypto_spot_bot/config.py
+2
View File
@@ -115,6 +115,7 @@ class Settings:
kelly_max_fraction: float
risk_per_trade_percent: float
risk_guard_enabled: bool
risk_symbol_guard_enabled: bool
risk_recent_trade_window: int
risk_max_consecutive_losses: int
risk_min_recent_profit_factor: float
@@ -265,6 +266,7 @@ def load_settings(env_file: str | Path | None = None) -> Settings:
kelly_max_fraction=_float_env("KELLY_MAX_FRACTION", 0.20),
risk_per_trade_percent=_float_env("RISK_PER_TRADE_PERCENT", 0.01),
risk_guard_enabled=_bool_env("RISK_GUARD_ENABLED", True),
risk_symbol_guard_enabled=_bool_env("RISK_SYMBOL_GUARD_ENABLED", True),
risk_recent_trade_window=_int_env("RISK_RECENT_TRADE_WINDOW", 20),
risk_max_consecutive_losses=_int_env("RISK_MAX_CONSECUTIVE_LOSSES", 4),
risk_min_recent_profit_factor=_float_env("RISK_MIN_RECENT_PROFIT_FACTOR", 0.85),
crypto_spot_bot/dashboard.py
+1
View File
@@ -290,6 +290,7 @@ def _safe_config(settings: Settings) -> dict[str, Any]:
"kelly_max_fraction": settings.kelly_max_fraction,
"risk_per_trade_percent": settings.risk_per_trade_percent,
"risk_guard_enabled": settings.risk_guard_enabled,
"risk_symbol_guard_enabled": settings.risk_symbol_guard_enabled,
"risk_recent_trade_window": settings.risk_recent_trade_window,
"risk_max_consecutive_losses": settings.risk_max_consecutive_losses,
"risk_min_recent_profit_factor": settings.risk_min_recent_profit_factor,
tests/conftest.py
+1
View File
@@ -67,6 +67,7 @@ def make_settings():
kelly_max_fraction=0.20,
risk_per_trade_percent=0.01,
risk_guard_enabled=True,
risk_symbol_guard_enabled=True,
risk_recent_trade_window=20,
risk_max_consecutive_losses=4,
risk_min_recent_profit_factor=0.85,
tests/test_analytics_quality.py
+41 -1
View File
@@ -34,7 +34,12 @@ def test_risk_guard_reduces_size_after_consecutive_losses(make_settings, tmp_pat
def test_risk_guard_blocks_only_bad_symbol(make_settings, tmp_path) -> None:
settings = make_settings(tmp_path, risk_max_consecutive_losses=3, symbols=["BTCUSDT", "ETHUSDT"])
settings = make_settings(
tmp_path,
risk_symbol_guard_enabled=True,
risk_max_consecutive_losses=3,
symbols=["BTCUSDT", "ETHUSDT"],
)
storage = Storage(settings.database_path)
now = utc_now()
for _ in range(3):
@@ -76,6 +81,41 @@ def test_risk_guard_blocks_only_bad_symbol(make_settings, tmp_path) -> None:
assert symbol_rows["ETHUSDT"]["block_new_entries"] is False
def test_risk_guard_can_disable_symbol_blocks(make_settings, tmp_path) -> None:
settings = make_settings(
tmp_path,
risk_symbol_guard_enabled=False,
risk_max_consecutive_losses=3,
symbols=["BTCUSDT", "ETHUSDT"],
)
storage = Storage(settings.database_path)
now = utc_now()
for _ in range(3):
storage.insert_trade(
Trade(
id=None,
symbol="BTCUSDT",
side="SELL",
qty=1.0,
entry_price=100.0,
exit_price=99.0,
net_pnl=-1.0,
opened_at=now,
closed_at=now,
entry_diagnostics={"forecast": {"probability_up": 0.64, "model": "torch_gru"}},
)
)
guard = risk_guard_snapshot(settings, storage.closed_trades(), storage.latest_equity())
assert guard["symbol_guard_enabled"] is False
assert guard["blocked_symbols"] == []
symbol_rows = {row["symbol"]: row for row in guard["symbols"]}
assert symbol_rows["BTCUSDT"]["consecutive_losses"] == 3
assert symbol_rows["BTCUSDT"]["block_new_entries"] is False
assert symbol_rows["BTCUSDT"]["reasons"] == []
def test_risk_guard_ignores_trades_outside_active_universe(make_settings, tmp_path) -> None:
settings = make_settings(tmp_path, risk_max_consecutive_losses=2, symbols=["BTCUSDT", "ETHUSDT"])
storage = Storage(settings.database_path)
tests/test_config.py
+15
View File
@@ -52,6 +52,21 @@ def test_fast_trading_env_sets_effective_intervals(tmp_path, monkeypatch) -> Non
assert settings.max_entries_per_minute == 4
def test_symbol_risk_guard_can_be_disabled(tmp_path, monkeypatch) -> None:
monkeypatch.delenv("RISK_SYMBOL_GUARD_ENABLED", raising=False)
monkeypatch.setenv("TRADING_MODE", "paper")
env_file = tmp_path / ".env"
env_file.write_text(
"TRADING_MODE=paper\nRISK_SYMBOL_GUARD_ENABLED=false\n",
encoding="utf-8",
)
settings = load_settings(env_file)
assert settings.risk_guard_enabled is True
assert settings.risk_symbol_guard_enabled is False
def test_llm_advisor_is_disabled_by_default(tmp_path, monkeypatch) -> None:
monkeypatch.delenv("LLM_ADVISOR_ENABLED", raising=False)
monkeypatch.setenv("TRADING_MODE", "paper")