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

Make risk guard symbol aware

Browse Source
This commit is contained in:
Codex
2026-06-24 05:30:12 +03:00
parent 4112a17bc5
commit a6c94b1555
4 changed files with 168 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crypto_spot_bot/analytics.py
+53 -9
View File
@@ -114,30 +114,39 @@ def risk_guard_snapshot(
"position_size_multiplier": 1.0,
"reasons": [],
}
active_trades = _active_universe_trades(settings, closed_trades)
reasons: list[str] = []
consecutive_losses = _consecutive_losses(closed_trades)
degraded_reasons: list[str] = []
consecutive_losses = _consecutive_losses(active_trades)
if consecutive_losses >= settings.risk_max_consecutive_losses:
reasons.append("consecutive_losses")
today_pnl = _today_pnl(closed_trades)
degraded_reasons.append("consecutive_losses")
today_pnl = _today_pnl(active_trades)
if today_pnl <= -abs(settings.max_daily_drawdown_usdt):
reasons.append("daily_loss_limit")
window = max(4, settings.risk_recent_trade_window)
recent_stats = _trade_stats(closed_trades[:window])
recent_stats = _trade_stats(active_trades[:window])
if recent_stats["trades"] >= max(4, min(window, 8)):
if recent_stats["profit_factor"] < settings.risk_min_recent_profit_factor:
reasons.append("recent_profit_factor_below_min")
degraded_reasons.append("recent_profit_factor_below_min")
if recent_stats["avg_net_percent"] <= 0:
reasons.append("recent_expectancy_non_positive")
degraded_reasons.append("recent_expectancy_non_positive")
latest_drawdown = float((latest_equity or {}).get("drawdown", 0.0) or 0.0)
if latest_drawdown >= abs(settings.max_daily_drawdown_usdt):
reasons.append("equity_drawdown_limit")
block = bool({"consecutive_losses", "daily_loss_limit", "equity_drawdown_limit"} & set(reasons))
multiplier = 0.0 if block else (settings.risk_reduce_multiplier if reasons else 1.0)
symbol_stats = _symbol_guard_stats(settings, active_trades)
blocked_symbols = sorted(row["symbol"] for row in symbol_stats if row["block_new_entries"])
block = bool(reasons)
all_reasons = reasons + degraded_reasons
multiplier = 0.0 if block else (settings.risk_reduce_multiplier if degraded_reasons else 1.0)
return {
"enabled": True,
"block_new_entries": block,
"position_size_multiplier": round(max(0.0, min(1.0, multiplier)), 4),
"reasons": reasons,
"reasons": all_reasons,
"global_reasons": reasons,
"degraded_reasons": degraded_reasons,
"blocked_symbols": blocked_symbols,
"symbols": symbol_stats,
"consecutive_losses": consecutive_losses,
"today_pnl": round(today_pnl, 6),
"recent": recent_stats,
@@ -183,6 +192,41 @@ def _group_stats(trades: list[dict[str, Any]], key_fn) -> list[dict[str, Any]]:
return sorted(rows, key=lambda row: (row["net_pnl"], row["trades"]), reverse=True)
def _active_universe_trades(settings: Settings, trades: list[dict[str, Any]]) -> list[dict[str, Any]]:
symbols = {symbol.upper() for symbol in settings.symbols}
if not symbols:
return trades
return [trade for trade in trades if str(trade.get("symbol", "")).upper() in symbols]
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)
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()]
recent = symbol_trades[: settings.risk_recent_trade_window]
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")
rows.append(
{
"symbol": symbol.upper(),
"block_new_entries": bool(reasons),
"reasons": reasons,
"consecutive_losses": losses,
**stats,
}
)
return rows
def _trade_summary(trade: dict[str, Any]) -> dict[str, Any]:
diagnostics = trade.get("entry_diagnostics") if isinstance(trade.get("entry_diagnostics"), dict) else {}
forecast = diagnostics.get("forecast") if isinstance(diagnostics.get("forecast"), dict) else {}
crypto_spot_bot/bot.py
+28
View File
@@ -171,6 +171,23 @@ class CryptoSpotBot:
)
)
continue
symbol_guard = self._risk_guard_for_symbol(risk_guard, symbol)
if symbol_guard.get("block_new_entries"):
self.storage.insert_signal(
Signal(
symbol,
"HOLD",
0.0,
"risk_guard: symbol blocked",
{
"strategy_mode": self.settings.strategy_mode,
"risk_guard": risk_guard,
"symbol_guard": symbol_guard,
"checks": {"risk_guard_symbol_ok": False},
},
)
)
continue
llm = {}
if (
self.settings.llm_advisor_enabled
@@ -211,6 +228,17 @@ class CryptoSpotBot:
prices,
)
@staticmethod
def _risk_guard_for_symbol(risk_guard: dict, symbol: str) -> dict:
rows = risk_guard.get("symbols")
if not isinstance(rows, list):
return {}
symbol_upper = symbol.upper()
for row in rows:
if isinstance(row, dict) and str(row.get("symbol", "")).upper() == symbol_upper:
return row
return {}
def _with_exposure_context(self, rules: dict) -> dict:
enriched = dict(rules)
current_exposure = self.broker.exposure()
crypto_spot_bot/dashboard.py
+15
View File
@@ -741,6 +741,7 @@ HTML = r"""
${panel('Risk guard', kvTable([
['Block entries', risk.block_new_entries ? 'yes' : 'no'],
['Size multiplier', num(risk.position_size_multiplier, 4)],
['Blocked symbols', (risk.blocked_symbols || []).join(', ') || 'none'],
['Consecutive losses', String(risk.consecutive_losses ?? 0)],
['Today PnL', money(risk.today_pnl)],
['Recent PF', num(risk.recent?.profit_factor, 3)],
@@ -753,6 +754,7 @@ HTML = r"""
['Remote equity', money(reconciliation?.account?.total_equity)]
]) + discrepanciesHtml(reconciliation?.discrepancies || []))}
</div>
${panel('Risk by symbol', symbolRiskTable(risk.symbols || []))}
${panel('Data quality by symbol', qualityTable(quality.symbols || []))}
${panel('Probability calibration', calibrationTable(analytics?.probability_calibration?.buckets || []))}
${panel('Failed checks', simpleTable(Object.entries(drift.failed_checks || {}).map(([key, value]) => ({ check: key, count: value })), ['check', 'count']))}
@@ -876,6 +878,19 @@ HTML = r"""
})), ['symbol', 'status', 'score', 'candles', 'issues']);
}
function symbolRiskTable(rows) {
return simpleTable(rows.map(row => ({
symbol: row.symbol,
block: row.block_new_entries ? 'yes' : 'no',
trades: row.trades,
win: num((row.win_rate || 0) * 100, 1) + '%',
avg: signed(row.avg_net_percent, 3) + '%',
pf: num(row.profit_factor, 3),
losses: row.consecutive_losses,
reasons: (row.reasons || []).join(', ') || 'none'
})), ['symbol', 'block', 'trades', 'win', 'avg', 'pf', 'losses', 'reasons']);
}
function calibrationTable(rows) {
return simpleTable(rows.map(row => ({
bucket: row.bucket,
tests/test_analytics_quality.py
+72 -3
View File
@@ -6,7 +6,7 @@ from crypto_spot_bot.models import Candle, Ticker, Trade, utc_now
from crypto_spot_bot.storage import Storage
def test_risk_guard_blocks_after_consecutive_losses(make_settings, tmp_path) -> None:
def test_risk_guard_reduces_size_after_consecutive_losses(make_settings, tmp_path) -> None:
settings = make_settings(tmp_path, risk_max_consecutive_losses=2)
storage = Storage(settings.database_path)
now = utc_now()
@@ -28,9 +28,78 @@ def test_risk_guard_blocks_after_consecutive_losses(make_settings, tmp_path) ->
guard = risk_guard_snapshot(settings, storage.closed_trades(), storage.latest_equity())
assert guard["block_new_entries"] is True
assert guard["block_new_entries"] is False
assert "consecutive_losses" in guard["reasons"]
assert guard["position_size_multiplier"] == 0.0
assert guard["position_size_multiplier"] == settings.risk_reduce_multiplier
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"])
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"}},
)
)
storage.insert_trade(
Trade(
id=None,
symbol="ETHUSDT",
side="SELL",
qty=1.0,
entry_price=100.0,
exit_price=102.0,
net_pnl=2.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["block_new_entries"] is False
assert guard["blocked_symbols"] == ["BTCUSDT"]
symbol_rows = {row["symbol"]: row for row in guard["symbols"]}
assert symbol_rows["BTCUSDT"]["block_new_entries"] is True
assert symbol_rows["ETHUSDT"]["block_new_entries"] is False
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)
now = utc_now()
for _ in range(4):
storage.insert_trade(
Trade(
id=None,
symbol="HYPEUSDT",
side="SELL",
qty=1.0,
entry_price=100.0,
exit_price=99.0,
net_pnl=-1.0,
opened_at=now,
closed_at=now,
)
)
guard = risk_guard_snapshot(settings, storage.closed_trades(), storage.latest_equity())
assert guard["block_new_entries"] is False
assert guard["reasons"] == []
assert guard["blocked_symbols"] == []
def test_data_quality_flags_missing_candle_gap() -> None: