diff --git a/.env.example b/.env.example index 21690a7..f8b8df0 100644 --- a/.env.example +++ b/.env.example @@ -50,14 +50,19 @@ KELLY_SIZING_ENABLED=false KELLY_FRACTION=0.25 KELLY_MAX_FRACTION=0.20 RISK_PER_TRADE_PERCENT=0.01 +RISK_GUARD_ENABLED=true +RISK_RECENT_TRADE_WINDOW=20 +RISK_MAX_CONSECUTIVE_LOSSES=4 +RISK_MIN_RECENT_PROFIT_FACTOR=0.85 +RISK_REDUCE_MULTIPLIER=0.50 ATR_TRAILING_MULTIPLIER=2.2 TREND_RSI_MIN=45 TREND_RSI_MAX=65 TIME_SERIES_FORECAST_ENABLED=true TIME_SERIES_MIN_CANDLES=120 TIME_SERIES_FORECAST_HORIZON=3 -TIME_SERIES_MIN_EDGE_PERCENT=0.10 -TIME_SERIES_MIN_PROBABILITY_UP=0.64 +TIME_SERIES_MIN_EDGE_PERCENT=0.08 +TIME_SERIES_MIN_PROBABILITY_UP=0.58 TIME_SERIES_MIN_CONFIDENCE=0.72 TIME_SERIES_MAX_ADJUSTMENT=0.08 TIME_SERIES_LSTM_ENABLED=true diff --git a/README.md b/README.md index 4bc6ced..33d9582 100644 --- a/README.md +++ b/README.md @@ -148,14 +148,19 @@ KELLY_SIZING_ENABLED=false KELLY_FRACTION=0.25 KELLY_MAX_FRACTION=0.20 RISK_PER_TRADE_PERCENT=0.01 +RISK_GUARD_ENABLED=true +RISK_RECENT_TRADE_WINDOW=20 +RISK_MAX_CONSECUTIVE_LOSSES=4 +RISK_MIN_RECENT_PROFIT_FACTOR=0.85 +RISK_REDUCE_MULTIPLIER=0.50 ATR_TRAILING_MULTIPLIER=2.2 TREND_RSI_MIN=45 TREND_RSI_MAX=65 TIME_SERIES_FORECAST_ENABLED=true TIME_SERIES_MIN_CANDLES=120 TIME_SERIES_FORECAST_HORIZON=3 -TIME_SERIES_MIN_EDGE_PERCENT=0.10 -TIME_SERIES_MIN_PROBABILITY_UP=0.64 +TIME_SERIES_MIN_EDGE_PERCENT=0.08 +TIME_SERIES_MIN_PROBABILITY_UP=0.58 TIME_SERIES_MIN_CONFIDENCE=0.72 TIME_SERIES_MAX_ADJUSTMENT=0.08 TIME_SERIES_LSTM_ENABLED=true diff --git a/crypto_spot_bot/analytics.py b/crypto_spot_bot/analytics.py new file mode 100644 index 0000000..445bdcb --- /dev/null +++ b/crypto_spot_bot/analytics.py @@ -0,0 +1,300 @@ +from __future__ import annotations + +import json +from datetime import datetime, timezone +from typing import Any + +from crypto_spot_bot.config import Settings +from crypto_spot_bot.storage import Storage + + +def analytics_snapshot(settings: Settings, storage: Storage) -> dict[str, Any]: + closed = _closed_trades_with_diagnostics(storage, settings.learning_lookback_trades) + return { + "pnl": pnl_attribution(closed), + "probability_calibration": probability_calibration(closed), + "drift": drift_snapshot(settings, closed, storage.recent_signals(240)), + "risk_guard": risk_guard_snapshot(settings, closed, storage.latest_equity()), + } + + +def pnl_attribution(closed_trades: list[dict[str, Any]]) -> dict[str, Any]: + total = _trade_stats(closed_trades) + by_symbol = _group_stats(closed_trades, lambda trade: str(trade.get("symbol", ""))) + by_exit = _group_stats(closed_trades, lambda trade: _exit_category(str(trade.get("reason", "")))) + by_model = _group_stats(closed_trades, lambda trade: _entry_model(trade)) + recent = [_trade_summary(trade) for trade in closed_trades[:20]] + return { + "total": total, + "by_symbol": by_symbol, + "by_exit": by_exit, + "by_model": by_model, + "recent": recent, + } + + +def probability_calibration(closed_trades: list[dict[str, Any]]) -> dict[str, Any]: + buckets: dict[str, list[dict[str, Any]]] = {} + for trade in closed_trades: + probability = _entry_probability(trade) + if probability is None: + continue + low = max(0.0, min(0.95, int(probability * 20) / 20)) + high = low + 0.05 + key = f"{low:.2f}-{high:.2f}" + buckets.setdefault(key, []).append(trade) + rows = [] + for key in sorted(buckets): + trades = buckets[key] + stats = _trade_stats(trades) + predicted = [_entry_probability(trade) for trade in trades] + avg_probability = sum(value for value in predicted if value is not None) / len(trades) + rows.append( + { + "bucket": key, + "trades": len(trades), + "avg_probability": round(avg_probability, 4), + "actual_win_rate": stats["win_rate"], + "calibration_error": round(stats["win_rate"] - avg_probability, 4), + "net_pnl": stats["net_pnl"], + "avg_net_percent": stats["avg_net_percent"], + } + ) + status = "insufficient" + if sum(row["trades"] for row in rows) >= 12: + avg_abs_error = sum(abs(row["calibration_error"]) * row["trades"] for row in rows) / sum(row["trades"] for row in rows) + status = "ok" if avg_abs_error <= 0.12 else "warn" + else: + avg_abs_error = None + return { + "status": status, + "buckets": rows, + "samples": sum(row["trades"] for row in rows), + "avg_abs_error": round(avg_abs_error, 4) if avg_abs_error is not None else None, + } + + +def drift_snapshot(settings: Settings, closed_trades: list[dict[str, Any]], signals: list[dict[str, Any]]) -> dict[str, Any]: + window = max(4, settings.risk_recent_trade_window) + recent = closed_trades[:window] + previous = closed_trades[window : window * 2] + recent_stats = _trade_stats(recent) + previous_stats = _trade_stats(previous) + failed_checks = _failed_check_counts(signals) + status = "insufficient" + warnings: list[str] = [] + if recent_stats["trades"] >= max(4, min(window, 8)): + status = "ok" + if recent_stats["profit_factor"] < settings.risk_min_recent_profit_factor: + warnings.append("recent_profit_factor_below_min") + if recent_stats["avg_net_percent"] <= 0: + warnings.append("recent_expectancy_non_positive") + if _consecutive_losses(closed_trades) >= settings.risk_max_consecutive_losses: + warnings.append("consecutive_losses") + if warnings: + status = "warn" + return { + "status": status, + "warnings": warnings, + "recent": recent_stats, + "previous": previous_stats, + "failed_checks": failed_checks, + } + + +def risk_guard_snapshot( + settings: Settings, + closed_trades: list[dict[str, Any]], + latest_equity: dict[str, Any] | None, +) -> dict[str, Any]: + if not settings.risk_guard_enabled: + return { + "enabled": False, + "block_new_entries": False, + "position_size_multiplier": 1.0, + "reasons": [], + } + reasons: list[str] = [] + consecutive_losses = _consecutive_losses(closed_trades) + if consecutive_losses >= settings.risk_max_consecutive_losses: + reasons.append("consecutive_losses") + today_pnl = _today_pnl(closed_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]) + 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") + if recent_stats["avg_net_percent"] <= 0: + 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) + return { + "enabled": True, + "block_new_entries": block, + "position_size_multiplier": round(max(0.0, min(1.0, multiplier)), 4), + "reasons": reasons, + "consecutive_losses": consecutive_losses, + "today_pnl": round(today_pnl, 6), + "recent": recent_stats, + } + + +def _closed_trades_with_diagnostics(storage: Storage, limit: int) -> list[dict[str, Any]]: + rows = storage.closed_trades(limit) + for row in rows: + row["entry_diagnostics"] = _json_or_default(row.get("entry_diagnostics_json"), {}) + return rows + + +def _trade_stats(trades: list[dict[str, Any]]) -> dict[str, Any]: + values = [float(trade.get("net_pnl", 0.0) or 0.0) for trade in trades] + wins = [value for value in values if value > 0] + losses = [value for value in values if value < 0] + gross_profit = sum(wins) + gross_loss = abs(sum(losses)) + profit_factor = gross_profit / gross_loss if gross_loss > 0 else (999.0 if gross_profit > 0 else 0.0) + percents = [_trade_net_percent(trade) for trade in trades] + return { + "trades": len(trades), + "wins": len(wins), + "losses": len(losses), + "win_rate": round(len(wins) / len(trades), 4) if trades else 0.0, + "net_pnl": round(sum(values), 6), + "fees": round(sum(float(trade.get("fee_usdt", 0.0) or 0.0) for trade in trades), 6), + "avg_net_pnl": round(sum(values) / len(trades), 6) if trades else 0.0, + "avg_net_percent": round(sum(percents) / len(percents), 4) if percents else 0.0, + "profit_factor": round(profit_factor, 4), + "best": round(max(values), 6) if values else 0.0, + "worst": round(min(values), 6) if values else 0.0, + } + + +def _group_stats(trades: list[dict[str, Any]], key_fn) -> list[dict[str, Any]]: + groups: dict[str, list[dict[str, Any]]] = {} + for trade in trades: + key = key_fn(trade) or "unknown" + groups.setdefault(key, []).append(trade) + rows = [{"key": key, **_trade_stats(items)} for key, items in groups.items()] + return sorted(rows, key=lambda row: (row["net_pnl"], row["trades"]), reverse=True) + + +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 {} + return { + "id": trade.get("id"), + "symbol": trade.get("symbol"), + "net_pnl": round(float(trade.get("net_pnl", 0.0) or 0.0), 6), + "net_percent": _trade_net_percent(trade), + "fee_usdt": round(float(trade.get("fee_usdt", 0.0) or 0.0), 6), + "entry_price": trade.get("entry_price"), + "exit_price": trade.get("exit_price"), + "closed_at": trade.get("closed_at"), + "exit_category": _exit_category(str(trade.get("reason", ""))), + "entry_probability": forecast.get("probability_up"), + "entry_expected_percent": forecast.get("expected_return_percent"), + "entry_confidence": trade.get("entry_confidence"), + "model": forecast.get("model"), + } + + +def _trade_net_percent(trade: dict[str, Any]) -> float: + qty = float(trade.get("qty", 0.0) or 0.0) + entry_price = float(trade.get("entry_price", 0.0) or 0.0) + notional = qty * entry_price + if notional <= 0: + return 0.0 + return round(float(trade.get("net_pnl", 0.0) or 0.0) / notional * 100, 4) + + +def _exit_category(reason: str) -> str: + text = reason.lower() + if "stop" in text: + return "stop_loss" + if "trailing" in text: + return "trailing_stop" + if "negative" in text or "turned" in text: + return "forecast_negative" + if "weak" in text or "enough edge" in text: + return "forecast_weak" + if "take" in text: + return "take_profit" + return "other" + + +def _entry_model(trade: dict[str, Any]) -> str: + diagnostics = trade.get("entry_diagnostics") if isinstance(trade.get("entry_diagnostics"), dict) else {} + forecast = diagnostics.get("forecast") if isinstance(diagnostics.get("forecast"), dict) else {} + return str(forecast.get("model") or "unknown") + + +def _entry_probability(trade: dict[str, Any]) -> float | None: + diagnostics = trade.get("entry_diagnostics") if isinstance(trade.get("entry_diagnostics"), dict) else {} + forecast = diagnostics.get("forecast") if isinstance(diagnostics.get("forecast"), dict) else {} + value = forecast.get("probability_up") + try: + probability = float(value) + except (TypeError, ValueError): + return None + return max(0.0, min(1.0, probability)) + + +def _failed_check_counts(signals: list[dict[str, Any]]) -> dict[str, int]: + counts: dict[str, int] = {} + for signal in signals: + diagnostics = _json_or_default(signal.get("diagnostics_json"), {}) + checks = diagnostics.get("checks") if isinstance(diagnostics, dict) else {} + if not isinstance(checks, dict): + continue + for key, ok in checks.items(): + if not ok: + counts[str(key)] = counts.get(str(key), 0) + 1 + return dict(sorted(counts.items(), key=lambda item: item[1], reverse=True)) + + +def _consecutive_losses(closed_trades: list[dict[str, Any]]) -> int: + count = 0 + for trade in closed_trades: + if float(trade.get("net_pnl", 0.0) or 0.0) < 0: + count += 1 + else: + break + return count + + +def _today_pnl(closed_trades: list[dict[str, Any]]) -> float: + today = datetime.now(timezone.utc).date() + total = 0.0 + for trade in closed_trades: + closed_at = _parse_datetime(trade.get("closed_at")) + if closed_at and closed_at.date() == today: + total += float(trade.get("net_pnl", 0.0) or 0.0) + return total + + +def _json_or_default(value: Any, default: Any) -> Any: + if isinstance(value, (dict, list)): + return value + if not isinstance(value, str): + return default + try: + return json.loads(value) + except json.JSONDecodeError: + return default + + +def _parse_datetime(value: Any) -> datetime | None: + if not isinstance(value, str) or not value: + return None + try: + parsed = datetime.fromisoformat(value) + except ValueError: + return None + if parsed.tzinfo is None: + return parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) diff --git a/crypto_spot_bot/bot.py b/crypto_spot_bot/bot.py index cc3bd47..de6ccf3 100644 --- a/crypto_spot_bot/bot.py +++ b/crypto_spot_bot/bot.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio from datetime import datetime +from crypto_spot_bot.analytics import risk_guard_snapshot from crypto_spot_bot.config import Settings from crypto_spot_bot.execution import LiveBroker, PaperBroker from crypto_spot_bot.learning import TradeLearner @@ -123,6 +124,11 @@ class CryptoSpotBot: async def _process_entries(self) -> None: prices = self.market.prices() + risk_guard = risk_guard_snapshot( + self.settings, + self.storage.closed_trades(self.settings.learning_lookback_trades), + self.storage.latest_equity(), + ) for symbol in self.market.symbols: cooldown_since = self._entry_cooldown_until.get(symbol) if cooldown_since: @@ -149,6 +155,22 @@ class CryptoSpotBot: learning = self.learner.adjustment_for(symbol, str(pattern.get("label", ""))).as_dict() learning["adaptive_rules"] = self._with_exposure_context(learning.get("adaptive_rules") or {}) account = self.broker.account_state(prices) + account["risk_guard"] = risk_guard + if risk_guard.get("block_new_entries"): + self.storage.insert_signal( + Signal( + symbol, + "HOLD", + 0.0, + "risk_guard: new entries blocked", + { + "strategy_mode": self.settings.strategy_mode, + "risk_guard": risk_guard, + "checks": {"risk_guard_ok": False}, + }, + ) + ) + continue llm = {} if ( self.settings.llm_advisor_enabled diff --git a/crypto_spot_bot/bybit.py b/crypto_spot_bot/bybit.py index 019e823..de0e3eb 100644 --- a/crypto_spot_bot/bybit.py +++ b/crypto_spot_bot/bybit.py @@ -63,6 +63,19 @@ class BybitClient: response.raise_for_status() return self._unwrap(response.json()) + def private_get(self, path: str, params: dict[str, Any]) -> dict[str, Any]: + query_params = sorted((key, value) for key, value in params.items() if value is not None) + query = urlencode(query_params) + headers = self._headers(query) + response = self.session.get( + f"{self.settings.rest_base_url}{path}", + params=query_params, + headers=headers, + timeout=15, + ) + response.raise_for_status() + return self._unwrap(response.json()) + def _headers(self, payload: str) -> dict[str, str]: timestamp = str(int(time.time() * 1000)) recv_window = "5000" @@ -204,6 +217,18 @@ class BybitClient: } return self.private_post("/v5/order/create", payload) + def wallet_balance(self, account_type: str = "UNIFIED", coin: str | None = None) -> dict[str, Any]: + return self.private_get( + "/v5/account/wallet-balance", + {"accountType": account_type, "coin": coin}, + ) + + def realtime_orders(self, *, category: str = "spot", open_only: int = 0, limit: int = 50) -> dict[str, Any]: + return self.private_get( + "/v5/order/realtime", + {"category": category, "openOnly": open_only, "limit": max(1, min(limit, 50))}, + ) + def websocket_subscribe_message(symbols: list[str], interval: str = "1") -> str: args: list[str] = [] diff --git a/crypto_spot_bot/config.py b/crypto_spot_bot/config.py index 3e0999c..695524f 100644 --- a/crypto_spot_bot/config.py +++ b/crypto_spot_bot/config.py @@ -101,6 +101,11 @@ class Settings: kelly_fraction: float kelly_max_fraction: float risk_per_trade_percent: float + risk_guard_enabled: bool + risk_recent_trade_window: int + risk_max_consecutive_losses: int + risk_min_recent_profit_factor: float + risk_reduce_multiplier: float atr_trailing_multiplier: float trend_rsi_min: float trend_rsi_max: float @@ -244,14 +249,19 @@ def load_settings(env_file: str | Path | None = None) -> Settings: kelly_fraction=_float_env("KELLY_FRACTION", 0.25), 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_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), + risk_reduce_multiplier=_float_env("RISK_REDUCE_MULTIPLIER", 0.50), atr_trailing_multiplier=_float_env("ATR_TRAILING_MULTIPLIER", 2.2), trend_rsi_min=_float_env("TREND_RSI_MIN", 45.0), trend_rsi_max=_float_env("TREND_RSI_MAX", 65.0), time_series_forecast_enabled=_bool_env("TIME_SERIES_FORECAST_ENABLED", forecast_enabled_default), time_series_min_candles=_int_env("TIME_SERIES_MIN_CANDLES", 120), time_series_forecast_horizon=_int_env("TIME_SERIES_FORECAST_HORIZON", 3), - time_series_min_edge_percent=_float_env("TIME_SERIES_MIN_EDGE_PERCENT", 0.10), - time_series_min_probability_up=_float_env("TIME_SERIES_MIN_PROBABILITY_UP", 0.64), + time_series_min_edge_percent=_float_env("TIME_SERIES_MIN_EDGE_PERCENT", 0.08), + time_series_min_probability_up=_float_env("TIME_SERIES_MIN_PROBABILITY_UP", 0.58), time_series_min_confidence=_float_env("TIME_SERIES_MIN_CONFIDENCE", 0.72), time_series_max_adjustment=_float_env("TIME_SERIES_MAX_ADJUSTMENT", 0.08), time_series_lstm_enabled=_bool_env("TIME_SERIES_LSTM_ENABLED", True), diff --git a/crypto_spot_bot/dashboard.py b/crypto_spot_bot/dashboard.py index 8224c3e..7728850 100644 --- a/crypto_spot_bot/dashboard.py +++ b/crypto_spot_bot/dashboard.py @@ -7,6 +7,7 @@ from typing import Any from fastapi import FastAPI, Response from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse +from crypto_spot_bot.analytics import analytics_snapshot from crypto_spot_bot.bot import CryptoSpotBot from crypto_spot_bot.bybit import BybitClient from crypto_spot_bot.config import Settings, load_settings, update_env_value @@ -14,6 +15,7 @@ from crypto_spot_bot.execution import LiveBroker, PaperBroker from crypto_spot_bot.learning import TradeLearner from crypto_spot_bot.market_data import MarketData from crypto_spot_bot.patterns import PatternAnalyzer +from crypto_spot_bot.reconciliation import reconciliation_snapshot from crypto_spot_bot.storage import Storage from crypto_spot_bot.strategy import SpotStrategy from crypto_spot_bot.time_series import TimeSeriesForecaster @@ -86,6 +88,31 @@ def create_app(settings: Settings | None = None) -> FastAPI: async def events(limit: int = 120) -> dict[str, Any]: return {"items": storage.recent_events(_limit(limit))} + @app.get("/api/analytics") + async def analytics() -> dict[str, Any]: + return analytics_snapshot(settings, storage) + + @app.get("/api/quality") + async def quality() -> dict[str, Any]: + return market.snapshot().get("quality", {}) + + @app.get("/api/reconciliation") + async def reconciliation() -> dict[str, Any]: + return reconciliation_snapshot( + settings=settings, + storage=storage, + client=client, + instruments=market.instruments, + ) + + @app.get("/api/backtest") + async def backtest() -> dict[str, Any]: + return _runtime_json(settings, "torch_threshold_calibration.json") + + @app.get("/api/retrain") + async def retrain() -> dict[str, Any]: + return _runtime_json(settings, "torch_retrain_guard.json") + @app.get("/api/config") async def config() -> dict[str, Any]: return _safe_config(settings) @@ -217,6 +244,11 @@ def _safe_config(settings: Settings) -> dict[str, Any]: "kelly_fraction": settings.kelly_fraction, "kelly_max_fraction": settings.kelly_max_fraction, "risk_per_trade_percent": settings.risk_per_trade_percent, + "risk_guard_enabled": settings.risk_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, + "risk_reduce_multiplier": settings.risk_reduce_multiplier, "atr_trailing_multiplier": settings.atr_trailing_multiplier, "trend_rsi_min": settings.trend_rsi_min, "trend_rsi_max": settings.trend_rsi_max, @@ -244,6 +276,19 @@ def _safe_config(settings: Settings) -> dict[str, Any]: } +def _runtime_json(settings: Settings, name: str) -> dict[str, Any]: + path = settings.time_series_lstm_model_path.parent / name + try: + data = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {"available": False, "path": str(path)} + if not isinstance(data, dict): + return {"available": False, "path": str(path)} + data["available"] = True + data["path"] = str(path) + return data + + def _time_series_model_artifact(settings: Settings) -> dict[str, Any]: path = settings.time_series_lstm_model_path try: @@ -342,909 +387,657 @@ def _forecast_model_label(model: str, *, torch_artifact: bool = False) -> str: return model + HTML = r""" - Крипто спот-бот + TradeBot Control
-
-

Крипто спот-бот

-
Загрузка состояния...
-
-
- Демо - Реальная торговля заблокирована - - Обычный режим - Поток данных - - -
-
-
-
-
Баланс
-
-
Свободно
-
-
В позициях
-
-
Прибыль/убыток
-
-
Позиции
-
-
Просадка
-
-
-
-
-
-
-

Открытые позиции

-
ПараКол-воВходЦенаПрибыль/убытокСтопЦель
-
-
-

Сделки

-
ВремяПараСторонаКол-воВходВыходИтогПричина
-
-
-

Сигналы стратегии

-
ВремяПараДействиеРежимРазмерУверенностьБазаШаблонОбучениеПрогнозОтскокИтогПричина
-
+
+
+

TradeBot

+
Загрузка
- -
-
+
+ health + + + +
+ + + +
+ diff --git a/crypto_spot_bot/data_quality.py b/crypto_spot_bot/data_quality.py new file mode 100644 index 0000000..e4d8cd4 --- /dev/null +++ b/crypto_spot_bot/data_quality.py @@ -0,0 +1,145 @@ +from __future__ import annotations + +from typing import Any + +from crypto_spot_bot.models import Candle, Ticker, utc_now + + +def market_quality_snapshot( + *, + symbols: list[str], + candles_by_symbol: dict[str, list[Candle]], + tickers: dict[str, Ticker], + interval: str, +) -> dict[str, Any]: + rows = [ + analyze_symbol_quality( + symbol=symbol, + candles=candles_by_symbol.get(symbol, []), + ticker=tickers.get(symbol), + interval=interval, + ) + for symbol in symbols + ] + worst_score = min((row["score"] for row in rows), default=0.0) + status = "ok" + if any(row["status"] == "error" for row in rows): + status = "error" + elif any(row["status"] == "warn" for row in rows): + status = "warn" + return { + "status": status, + "score": round(worst_score, 4), + "symbols": rows, + } + + +def analyze_symbol_quality( + *, + symbol: str, + candles: list[Candle], + ticker: Ticker | None, + interval: str, +) -> dict[str, Any]: + issues: list[dict[str, Any]] = [] + score = 1.0 + interval_ms = _interval_ms(interval) + if not candles: + return _row(symbol, "error", 0.0, [{"code": "no_candles", "severity": "error"}], 0, None, None) + + timestamps = [candle.timestamp for candle in candles] + duplicates = len(timestamps) - len(set(timestamps)) + if duplicates: + issues.append({"code": "duplicate_candles", "severity": "warn", "count": duplicates}) + score -= min(0.25, duplicates * 0.03) + + invalid_ohlc = 0 + zero_volume = 0 + for candle in candles: + prices = [candle.open, candle.high, candle.low, candle.close] + if any(price <= 0 for price in prices) or candle.high < max(candle.open, candle.close) or candle.low > min(candle.open, candle.close): + invalid_ohlc += 1 + if candle.volume <= 0: + zero_volume += 1 + if invalid_ohlc: + issues.append({"code": "invalid_ohlc", "severity": "error", "count": invalid_ohlc}) + score -= min(0.45, invalid_ohlc * 0.08) + if zero_volume: + issues.append({"code": "zero_volume", "severity": "warn", "count": zero_volume}) + score -= min(0.20, zero_volume * 0.02) + + missing_gaps = 0 + largest_gap_ms = 0 + if interval_ms > 0: + ordered = sorted(set(timestamps)) + for left, right in zip(ordered, ordered[1:]): + gap = right - left + largest_gap_ms = max(largest_gap_ms, gap) + if gap > interval_ms * 1.5: + missing_gaps += max(1, round(gap / interval_ms) - 1) + if missing_gaps: + issues.append({"code": "missing_candles", "severity": "warn", "count": missing_gaps}) + score -= min(0.35, missing_gaps * 0.04) + + age_seconds = max(0.0, (utc_now().timestamp() * 1000 - max(timestamps)) / 1000) + stale_after = interval_ms / 1000 * 2.5 + if age_seconds > stale_after: + issues.append({"code": "stale_candles", "severity": "warn", "age_seconds": round(age_seconds, 1)}) + score -= 0.20 + else: + age_seconds = None + + if ticker is None: + issues.append({"code": "no_ticker", "severity": "error"}) + score -= 0.45 + else: + if ticker.last_price <= 0: + issues.append({"code": "invalid_ticker_price", "severity": "error"}) + score -= 0.35 + if ticker.spread_percent > 0.35: + issues.append({"code": "wide_spread", "severity": "warn", "spread_percent": round(ticker.spread_percent, 4)}) + score -= 0.15 + + score = max(0.0, min(1.0, score)) + severity = {str(issue.get("severity")) for issue in issues} + status = "error" if "error" in severity else "warn" if "warn" in severity else "ok" + return _row( + symbol, + status, + score, + issues, + len(candles), + max(timestamps), + largest_gap_ms if interval_ms > 0 else None, + ) + + +def _row( + symbol: str, + status: str, + score: float, + issues: list[dict[str, Any]], + candle_count: int, + last_candle_timestamp: int | None, + largest_gap_ms: int | None, +) -> dict[str, Any]: + return { + "symbol": symbol, + "status": status, + "score": round(score, 4), + "candle_count": candle_count, + "last_candle_timestamp": last_candle_timestamp, + "largest_gap_ms": largest_gap_ms, + "issues": issues, + } + + +def _interval_ms(interval: str) -> int: + normalized = str(interval).strip().upper() + if normalized == "D": + return 24 * 60 * 60 * 1000 + if normalized == "W": + return 7 * 24 * 60 * 60 * 1000 + if normalized.isdigit(): + return int(normalized) * 60 * 1000 + return 0 diff --git a/crypto_spot_bot/execution.py b/crypto_spot_bot/execution.py index f35cbd9..6f04642 100644 --- a/crypto_spot_bot/execution.py +++ b/crypto_spot_bot/execution.py @@ -171,6 +171,7 @@ class PaperBroker: entry_reason=signal.reason, entry_confidence=signal.confidence, entry_pattern=str(signal.diagnostics.get("pattern", {}).get("label", "")), + entry_diagnostics=signal.diagnostics, ) position.id = self.storage.insert_position(position) self.positions.append(position) @@ -189,6 +190,7 @@ class PaperBroker: reason=signal.reason, entry_pattern=position.entry_pattern, entry_confidence=position.entry_confidence, + entry_diagnostics=position.entry_diagnostics, opened_at=position.opened_at, ) ) @@ -230,6 +232,7 @@ class PaperBroker: reason=reason, entry_pattern=position.entry_pattern, entry_confidence=position.entry_confidence, + entry_diagnostics=position.entry_diagnostics, opened_at=position.opened_at, closed_at=utc_now(), ) diff --git a/crypto_spot_bot/market_data.py b/crypto_spot_bot/market_data.py index 2de1e26..fa365f7 100644 --- a/crypto_spot_bot/market_data.py +++ b/crypto_spot_bot/market_data.py @@ -10,6 +10,7 @@ import websockets from crypto_spot_bot.bybit import BybitClient, Instrument, websocket_subscribe_message from crypto_spot_bot.config import Settings +from crypto_spot_bot.data_quality import analyze_symbol_quality, market_quality_snapshot from crypto_spot_bot.indicators import add_indicators from crypto_spot_bot.models import Candle, Ticker, utc_now from crypto_spot_bot.storage import Storage @@ -217,6 +218,12 @@ class MarketData: return { "symbols": self.symbols, "ws_connected": self.ws_connected, + "quality": market_quality_snapshot( + symbols=self.symbols, + candles_by_symbol=self.candles, + tickers=self.tickers, + interval=self.settings.base_interval, + ), "last_rest_refresh_at": self.last_rest_refresh_at.isoformat() if self.last_rest_refresh_at else None, @@ -230,6 +237,12 @@ class MarketData: "trend_candles": [candle.as_dict() for candle in self.trend_candles.get(symbol, [])[-5:]], "pattern": self.patterns.get(symbol), "forecast": self.forecasts.get(symbol), + "quality": analyze_symbol_quality( + symbol=symbol, + candles=self.candles.get(symbol, []), + ticker=self.tickers.get(symbol), + interval=self.settings.base_interval, + ), "instrument": asdict(self.instruments[symbol]) if symbol in self.instruments else None, } for symbol in self.symbols diff --git a/crypto_spot_bot/models.py b/crypto_spot_bot/models.py index bfcc1a1..eae48e3 100644 --- a/crypto_spot_bot/models.py +++ b/crypto_spot_bot/models.py @@ -87,6 +87,7 @@ class Position: entry_reason: str = "" entry_confidence: float = 0.0 entry_pattern: str = "" + entry_diagnostics: dict[str, Any] = field(default_factory=dict) def mark_price(self, price: float) -> float: return self.qty * price @@ -127,6 +128,7 @@ class Trade: reason: str = "" entry_pattern: str = "" entry_confidence: float = 0.0 + entry_diagnostics: dict[str, Any] = field(default_factory=dict) opened_at: datetime | None = None closed_at: datetime | None = None diff --git a/crypto_spot_bot/reconciliation.py b/crypto_spot_bot/reconciliation.py new file mode 100644 index 0000000..6390a20 --- /dev/null +++ b/crypto_spot_bot/reconciliation.py @@ -0,0 +1,161 @@ +from __future__ import annotations + +from dataclasses import asdict +from typing import Any + +from crypto_spot_bot.bybit import BybitClient, Instrument +from crypto_spot_bot.config import Settings +from crypto_spot_bot.storage import Storage + + +def reconciliation_snapshot( + *, + settings: Settings, + storage: Storage, + client: BybitClient, + instruments: dict[str, Instrument], +) -> dict[str, Any]: + local_positions = storage.open_positions() + local = [ + { + "id": position.id, + "symbol": position.symbol, + "qty": position.qty, + "entry_price": position.entry_price, + "notional_usdt": position.notional_usdt, + } + for position in local_positions + ] + if not settings.live_ready: + return { + "status": "paper", + "live_ready": False, + "local_positions": local, + "remote_balances": [], + "open_orders": [], + "discrepancies": [], + "message": "reconciliation requires unlocked live Bybit credentials", + } + try: + coins = _coins_for_symbols(settings.symbols, instruments) + wallet = client.wallet_balance(coin=",".join(coins)) + orders = client.realtime_orders(category="spot", open_only=0, limit=50) + except Exception as exc: + return { + "status": "error", + "live_ready": True, + "local_positions": local, + "remote_balances": [], + "open_orders": [], + "discrepancies": [{"severity": "error", "code": "bybit_read_failed", "message": str(exc)}], + } + + balances = _balances(wallet) + open_orders = orders.get("list", []) if isinstance(orders.get("list"), list) else [] + discrepancies = _discrepancies(local_positions, balances, instruments) + return { + "status": "warn" if discrepancies else "ok", + "live_ready": True, + "local_positions": local, + "remote_balances": balances, + "open_orders": open_orders, + "discrepancies": discrepancies, + "account": _account_summary(wallet), + "instruments": {symbol: asdict(info) for symbol, info in instruments.items() if symbol in settings.symbols}, + } + + +def _coins_for_symbols(symbols: tuple[str, ...], instruments: dict[str, Instrument]) -> list[str]: + coins = {"USDT"} + for symbol in symbols: + info = instruments.get(symbol) + if info and info.base_coin: + coins.add(info.base_coin.upper()) + elif symbol.endswith("USDT"): + coins.add(symbol[:-4].upper()) + return sorted(coins) + + +def _balances(wallet: dict[str, Any]) -> list[dict[str, Any]]: + rows = wallet.get("list") + if not isinstance(rows, list) or not rows: + return [] + coins = rows[0].get("coin") + if not isinstance(coins, list): + return [] + output = [] + for row in coins: + if not isinstance(row, dict): + continue + output.append( + { + "coin": str(row.get("coin", "")), + "equity": _float(row.get("equity")), + "wallet_balance": _float(row.get("walletBalance")), + "locked": _float(row.get("locked")), + "usd_value": _float(row.get("usdValue")), + } + ) + return output + + +def _account_summary(wallet: dict[str, Any]) -> dict[str, Any]: + rows = wallet.get("list") + if not isinstance(rows, list) or not rows: + return {} + row = rows[0] + return { + "account_type": row.get("accountType"), + "total_equity": _float(row.get("totalEquity")), + "total_wallet_balance": _float(row.get("totalWalletBalance")), + "total_available_balance": _float(row.get("totalAvailableBalance")), + } + + +def _discrepancies( + positions: list[Any], + balances: list[dict[str, Any]], + instruments: dict[str, Instrument], +) -> list[dict[str, Any]]: + by_coin = {str(row.get("coin", "")).upper(): row for row in balances} + local_by_coin: dict[str, float] = {} + for position in positions: + info = instruments.get(position.symbol) + coin = info.base_coin.upper() if info and info.base_coin else position.symbol.removesuffix("USDT") + local_by_coin[coin] = local_by_coin.get(coin, 0.0) + float(position.qty) + issues = [] + for coin, local_qty in local_by_coin.items(): + remote_qty = _float((by_coin.get(coin) or {}).get("equity")) + tolerance = max(1e-8, local_qty * 0.002) + if remote_qty + tolerance < local_qty: + issues.append( + { + "severity": "error", + "code": "remote_balance_below_local_position", + "coin": coin, + "local_qty": round(local_qty, 10), + "remote_qty": round(remote_qty, 10), + } + ) + for coin, row in by_coin.items(): + if coin == "USDT": + continue + remote_qty = _float(row.get("equity")) + local_qty = local_by_coin.get(coin, 0.0) + if remote_qty > max(1e-8, local_qty * 1.05) and local_qty <= 0: + issues.append( + { + "severity": "warn", + "code": "remote_asset_without_local_position", + "coin": coin, + "remote_qty": round(remote_qty, 10), + } + ) + return issues + + +def _float(value: Any) -> float: + try: + return float(value) + except (TypeError, ValueError): + return 0.0 diff --git a/crypto_spot_bot/storage.py b/crypto_spot_bot/storage.py index 529c556..a433525 100644 --- a/crypto_spot_bot/storage.py +++ b/crypto_spot_bot/storage.py @@ -43,6 +43,7 @@ class Storage: entry_reason TEXT NOT NULL DEFAULT '', entry_confidence REAL NOT NULL DEFAULT 0, entry_pattern TEXT NOT NULL DEFAULT '', + entry_diagnostics_json TEXT NOT NULL DEFAULT '{}', status TEXT NOT NULL DEFAULT 'OPEN' ); CREATE TABLE IF NOT EXISTS trades ( @@ -58,6 +59,7 @@ class Storage: reason TEXT NOT NULL DEFAULT '', entry_pattern TEXT NOT NULL DEFAULT '', entry_confidence REAL NOT NULL DEFAULT 0, + entry_diagnostics_json TEXT NOT NULL DEFAULT '{}', opened_at TEXT, closed_at TEXT ); @@ -113,6 +115,7 @@ class Storage: "entry_reason": "TEXT NOT NULL DEFAULT ''", "entry_confidence": "REAL NOT NULL DEFAULT 0", "entry_pattern": "TEXT NOT NULL DEFAULT ''", + "entry_diagnostics_json": "TEXT NOT NULL DEFAULT '{}'", }.items(): if column not in columns: conn.execute(f"ALTER TABLE positions ADD COLUMN {column} {definition}") @@ -123,6 +126,7 @@ class Storage: for column, definition in { "entry_pattern": "TEXT NOT NULL DEFAULT ''", "entry_confidence": "REAL NOT NULL DEFAULT 0", + "entry_diagnostics_json": "TEXT NOT NULL DEFAULT '{}'", }.items(): if column not in trade_columns: conn.execute(f"ALTER TABLE trades ADD COLUMN {column} {definition}") @@ -134,8 +138,8 @@ class Storage: INSERT INTO positions ( symbol, qty, entry_price, notional_usdt, entry_fee_usdt, stop_loss, take_profit, highest_price, opened_at, entry_reason, - entry_confidence, entry_pattern, status - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'OPEN') + entry_confidence, entry_pattern, entry_diagnostics_json, status + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'OPEN') """, ( position.symbol, @@ -150,6 +154,7 @@ class Storage: position.entry_reason, position.entry_confidence, position.entry_pattern, + json.dumps(position.entry_diagnostics, ensure_ascii=False), ), ) return int(cur.lastrowid) @@ -185,6 +190,7 @@ class Storage: entry_reason=row["entry_reason"], entry_confidence=float(row["entry_confidence"]), entry_pattern=row["entry_pattern"], + entry_diagnostics=_json_or_default(row["entry_diagnostics_json"], {}), ) for row in rows ] @@ -196,8 +202,8 @@ class Storage: INSERT INTO trades ( symbol, side, qty, entry_price, exit_price, gross_pnl, fee_usdt, net_pnl, reason, entry_pattern, entry_confidence, - opened_at, closed_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + entry_diagnostics_json, opened_at, closed_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( trade.symbol, @@ -211,6 +217,7 @@ class Storage: trade.reason, trade.entry_pattern, trade.entry_confidence, + json.dumps(trade.entry_diagnostics, ensure_ascii=False), trade.opened_at.isoformat() if trade.opened_at else None, trade.closed_at.isoformat() if trade.closed_at else None, ), diff --git a/crypto_spot_bot/strategy.py b/crypto_spot_bot/strategy.py index a4cf439..6869149 100644 --- a/crypto_spot_bot/strategy.py +++ b/crypto_spot_bot/strategy.py @@ -852,6 +852,8 @@ def _trend_position_sizing( if equity <= 0: equity = settings.starting_balance_usdt risk_fraction = _clamp(settings.risk_per_trade_percent, 0.0, 0.01) + guard_multiplier = _risk_guard_multiplier(account) + risk_fraction *= guard_multiplier risk_usdt = equity * risk_fraction raw_notional = risk_usdt / max(stop_loss_percent, 0.0001) high = max(0.0, settings.max_position_usdt) @@ -860,6 +862,7 @@ def _trend_position_sizing( return { "method": "fixed_fractional_risk", "risk_per_trade_percent": round(risk_fraction * 100, 4), + "risk_guard_multiplier": round(guard_multiplier, 4), "risk_usdt": round(risk_usdt, 4), "stop_loss_percent": round(stop_loss_percent * 100, 4), "raw_notional_usdt": round(raw_notional, 4), @@ -908,7 +911,7 @@ def _position_sizing( denominator = max(0.0001, 1.0 - settings.min_signal_confidence) confidence_ratio = _clamp((final_score - settings.min_signal_confidence) / denominator, 0.0, 1.0) confidence_notional = low + (high - low) * confidence_ratio - risk_multiplier = _position_risk_multiplier(forecast, adaptive) + risk_multiplier = _position_risk_multiplier(forecast, adaptive) * _risk_guard_multiplier(account) method = "confidence" raw = confidence_notional kelly = _kelly_position( @@ -952,6 +955,17 @@ def _position_risk_multiplier(forecast: dict | None, adaptive: dict | None) -> f return multiplier +def _risk_guard_multiplier(account: dict | None) -> float: + guard = (account or {}).get("risk_guard") + if not isinstance(guard, dict): + return 1.0 + try: + value = float(guard.get("position_size_multiplier", 1.0)) + except (TypeError, ValueError): + value = 1.0 + return _clamp(value, 0.0, 1.0) + + def _kelly_position( *, settings: Settings, diff --git a/tests/conftest.py b/tests/conftest.py index 71a4779..38a699a 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -66,14 +66,19 @@ def make_settings(): kelly_fraction=0.25, kelly_max_fraction=0.20, risk_per_trade_percent=0.01, + risk_guard_enabled=True, + risk_recent_trade_window=20, + risk_max_consecutive_losses=4, + risk_min_recent_profit_factor=0.85, + risk_reduce_multiplier=0.50, atr_trailing_multiplier=2.2, trend_rsi_min=45.0, trend_rsi_max=65.0, time_series_forecast_enabled=True, time_series_min_candles=120, time_series_forecast_horizon=3, - time_series_min_edge_percent=0.10, - time_series_min_probability_up=0.64, + time_series_min_edge_percent=0.08, + time_series_min_probability_up=0.58, time_series_min_confidence=0.72, time_series_max_adjustment=0.08, time_series_lstm_enabled=True, diff --git a/tests/test_analytics_quality.py b/tests/test_analytics_quality.py new file mode 100644 index 0000000..2212eeb --- /dev/null +++ b/tests/test_analytics_quality.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +from crypto_spot_bot.analytics import risk_guard_snapshot +from crypto_spot_bot.data_quality import analyze_symbol_quality +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: + settings = make_settings(tmp_path, risk_max_consecutive_losses=2) + storage = Storage(settings.database_path) + now = utc_now() + for _ in range(2): + 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["block_new_entries"] is True + assert "consecutive_losses" in guard["reasons"] + assert guard["position_size_multiplier"] == 0.0 + + +def test_data_quality_flags_missing_candle_gap() -> None: + candles = [ + Candle(1_000_000, 100, 101, 99, 100.5, 10), + Candle(1_000_000 + 60 * 60 * 1000 * 3, 100.5, 102, 100, 101, 12), + ] + ticker = Ticker("BTCUSDT", 101, 100.99, 101.01, 1_000_000, 100, 0) + + row = analyze_symbol_quality(symbol="BTCUSDT", candles=candles, ticker=ticker, interval="60") + + assert row["status"] == "warn" + assert any(issue["code"] == "missing_candles" for issue in row["issues"]) diff --git a/tests/test_bybit.py b/tests/test_bybit.py index 7535c99..0f2f8a8 100644 --- a/tests/test_bybit.py +++ b/tests/test_bybit.py @@ -58,6 +58,34 @@ def test_live_spot_order_explicitly_disables_leverage(make_settings, tmp_path) - assert captured["payload"]["orderFilter"] == "Order" +def test_private_get_signs_the_same_query_it_sends(make_settings, tmp_path) -> None: + client = BybitClient(make_settings(tmp_path)) + captured = {} + + class Response: + def raise_for_status(self): + return None + + def json(self): + return {"retCode": 0, "result": {"ok": True}} + + class Session: + def get(self, url, params, headers, timeout): + captured["url"] = url + captured["params"] = params + captured["headers"] = headers + captured["timeout"] = timeout + return Response() + + client.session = Session() + + assert client.wallet_balance(coin=None) == {"ok": True} + + assert captured["params"] == [("accountType", "UNIFIED")] + assert "coin" not in dict(captured["params"]) + assert captured["headers"]["X-BAPI-SIGN"] + + def test_websocket_subscribe_uses_configured_kline_interval() -> None: payload = websocket_subscribe_message(["BTCUSDT"], interval="60") diff --git a/tools/accept_torch_candidate.py b/tools/accept_torch_candidate.py new file mode 100644 index 0000000..dd574b2 --- /dev/null +++ b/tools/accept_torch_candidate.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +import argparse +import json +import shutil +from pathlib import Path +from typing import Any + + +def main() -> None: + args = _parse_args() + current = _read_json(args.current_report) + candidate = _read_json(args.candidate_report) + decision = _decision( + current, + candidate, + min_trades=args.min_trades, + min_profit_factor=args.min_profit_factor, + min_avg_net_percent=args.min_avg_net_percent, + max_score_regression=args.max_score_regression, + ) + payload = { + "accepted": decision["accepted"], + "reason": decision["reason"], + "current": _summary(current), + "candidate": _summary(candidate), + } + if args.report: + Path(args.report).write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + print(json.dumps(payload, ensure_ascii=False, sort_keys=True)) + if not decision["accepted"]: + raise SystemExit(2) + target = Path(args.target_artifact) + candidate_artifact = Path(args.candidate_artifact) + target.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(candidate_artifact, target) + + +def _parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Accept or reject a retrained Torch candidate artifact.") + parser.add_argument("--current-report", required=True) + parser.add_argument("--candidate-report", required=True) + parser.add_argument("--candidate-artifact", required=True) + parser.add_argument("--target-artifact", required=True) + parser.add_argument("--report", default="") + parser.add_argument("--min-trades", type=int, default=8) + parser.add_argument("--min-profit-factor", type=float, default=1.05) + parser.add_argument("--min-avg-net-percent", type=float, default=0.0) + parser.add_argument("--max-score-regression", type=float, default=0.05) + return parser.parse_args() + + +def _decision( + current: dict[str, Any], + candidate: dict[str, Any], + *, + min_trades: int, + min_profit_factor: float, + min_avg_net_percent: float, + max_score_regression: float, +) -> dict[str, Any]: + candidate_score = _score(candidate) + current_score = _score(current) + candidate_replay = candidate.get("full_replay") if isinstance(candidate.get("full_replay"), dict) else {} + candidate_walk = candidate.get("walk_forward") if isinstance(candidate.get("walk_forward"), dict) else {} + walk_summary = candidate_walk.get("summary") if isinstance(candidate_walk.get("summary"), dict) else {} + if int(candidate_replay.get("trades", 0) or 0) < min_trades: + return {"accepted": False, "reason": "candidate_has_too_few_full_replay_trades"} + if float(candidate_replay.get("profit_factor", 0.0) or 0.0) < min_profit_factor: + return {"accepted": False, "reason": "candidate_profit_factor_below_min"} + if float(candidate_replay.get("avg_net_percent", 0.0) or 0.0) <= min_avg_net_percent: + return {"accepted": False, "reason": "candidate_expectancy_non_positive"} + if int(walk_summary.get("trades", 0) or 0) >= min_trades and float(walk_summary.get("avg_net_percent", 0.0) or 0.0) <= min_avg_net_percent: + return {"accepted": False, "reason": "candidate_walk_forward_expectancy_non_positive"} + if current_score > 0 and candidate_score < current_score * (1.0 - max_score_regression): + return {"accepted": False, "reason": "candidate_score_regressed_vs_current"} + return {"accepted": True, "reason": "candidate_passed_guard"} + + +def _score(report: dict[str, Any]) -> float: + replay = report.get("full_replay") if isinstance(report.get("full_replay"), dict) else {} + recommended = report.get("recommended") if isinstance(report.get("recommended"), dict) else {} + replay_score = ( + float(replay.get("avg_net_percent", 0.0) or 0.0) + + float(replay.get("total_net_percent", 0.0) or 0.0) * 0.02 + - float(replay.get("max_drawdown_percent", 0.0) or 0.0) * 0.05 + + min(float(replay.get("profit_factor", 0.0) or 0.0), 10.0) * 0.03 + ) + return replay_score + float(recommended.get("score", 0.0) or 0.0) * 0.25 + + +def _summary(report: dict[str, Any]) -> dict[str, Any]: + return { + "score": round(_score(report), 6), + "recommended": report.get("recommended", {}), + "full_replay": report.get("full_replay", {}), + "walk_forward_summary": (report.get("walk_forward") or {}).get("summary", {}) + if isinstance(report.get("walk_forward"), dict) + else {}, + } + + +def _read_json(path: str) -> dict[str, Any]: + if not path: + return {} + try: + data = json.loads(Path(path).read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return {} + return data if isinstance(data, dict) else {} + + +if __name__ == "__main__": + main() diff --git a/tools/calibrate_torch_thresholds.py b/tools/calibrate_torch_thresholds.py index 3b6634a..55b5a28 100644 --- a/tools/calibrate_torch_thresholds.py +++ b/tools/calibrate_torch_thresholds.py @@ -48,6 +48,8 @@ class ForecastRecord: symbol: str index: int timestamp: int + close: float + atr: float expected_percent: float probability_up: float confidence: float @@ -139,6 +141,20 @@ def main() -> None: recommended = _choose_recommendation(results, min_trades=args.min_trades) print("\nRECOMMENDED") print(_result_line(recommended)) + full_backtest = _full_backtest(records, recommended, horizon=horizon, round_trip_cost=round_trip_cost, settings=settings) + print("\nFULL_REPLAY") + print(_stats_line(full_backtest)) + walk_forward = _walk_forward( + records, + edges=_float_grid(args.edge_grid), + probabilities=_float_grid(args.probability_grid), + confidences=_float_grid(args.confidence_grid), + min_trades=args.min_trades, + horizon=horizon, + folds=args.walk_forward_folds, + ) + print("\nWALK_FORWARD") + print(json.dumps(walk_forward["summary"], ensure_ascii=False, sort_keys=True)) print( "env " f"TIME_SERIES_MIN_EDGE_PERCENT={recommended.edge:.4f} " @@ -151,6 +167,9 @@ def main() -> None: "artifact": _artifact_summary(artifact), "records_by_symbol": per_symbol_counts, "recommended": _result_dict(recommended), + "full_replay": full_backtest, + "walk_forward": walk_forward, + "probability_calibration": _probability_calibration(records), "top_results": [_result_dict(result) for result in results[: args.top]], } Path(args.output).write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") @@ -174,6 +193,7 @@ def _parse_args() -> argparse.Namespace: parser.add_argument("--output", default="", help="Optional JSON output path.") parser.add_argument("--batch-size", type=int, default=256, help="Torch inference batch size.") parser.add_argument("--threads", type=int, default=0, help="Torch CPU threads; 0 keeps torch default.") + parser.add_argument("--walk-forward-folds", type=int, default=4, help="Threshold walk-forward folds.") return parser.parse_args() @@ -258,6 +278,8 @@ def _forecast_records( symbol=symbol, index=index, timestamp=candles[index].timestamp, + close=closes[index], + atr=float(candles[index].atr_14 or 0.0), expected_percent=expected_percent, probability_up=probability_up, confidence=_forecast_confidence(expected_percent, probability_up, skill, 0.04), @@ -349,6 +371,8 @@ def _batch_forecast_records( symbol=symbol, index=index, timestamp=candles[index].timestamp, + close=closes[index], + atr=float(candles[index].atr_14 or 0.0), expected_percent=expected_percent, probability_up=probability_up, confidence=_forecast_confidence(expected_percent, probability_up, skill, 0.04), @@ -484,6 +508,212 @@ def _feature_vector(entry: dict[str, Any], key: str, size: int, default: float) return [default for _ in range(size)] +def _full_backtest( + records: list[ForecastRecord], + thresholds: CalibrationResult, + *, + horizon: int, + round_trip_cost: float, + settings: Any, +) -> dict[str, Any]: + positions: dict[str, dict[str, Any]] = {} + trades: list[float] = [] + rows: list[dict[str, Any]] = [] + max_hold = max(12, horizon * 8) + stop_loss_percent = max(0.003, min(0.08, float(settings.stop_loss_percent))) * 100.0 + atr_multiplier = max(0.5, min(10.0, float(settings.atr_trailing_multiplier))) + for record in sorted(records, key=lambda item: (item.timestamp, item.symbol)): + position = positions.get(record.symbol) + if position is not None: + position["highest"] = max(position["highest"], record.close) + net_percent = _net_percent(position["entry_price"], record.close, round_trip_cost) + held = record.index - int(position["entry_index"]) + atr_stop = ( + record.close <= position["highest"] - record.atr * atr_multiplier + if record.atr > 0 and position["highest"] > position["entry_price"] + else False + ) + weak_forecast = ( + record.expected_percent < thresholds.edge + or record.probability_up < thresholds.probability + or record.skill <= 0.0 + ) + exit_reason = "" + if net_percent <= -stop_loss_percent: + exit_reason = "stop_loss" + elif atr_stop: + exit_reason = "atr_trailing_stop" + elif (record.expected_percent <= 0.0 or record.probability_up <= 0.50 or _candidate_blocks(record, thresholds.edge)): + exit_reason = "forecast_negative" + elif weak_forecast and net_percent >= 0: + exit_reason = "forecast_weak_profit_lock" + elif held >= max_hold: + exit_reason = "max_hold" + if exit_reason: + trades.append(net_percent) + rows.append( + { + "symbol": record.symbol, + "entry_timestamp": position["timestamp"], + "exit_timestamp": record.timestamp, + "net_percent": round(net_percent, 4), + "reason": exit_reason, + "held_bars": held, + "entry_probability": round(float(position["probability_up"]), 4), + "entry_expected_percent": round(float(position["expected_percent"]), 4), + } + ) + positions.pop(record.symbol, None) + continue + + if record.symbol in positions: + continue + if _candidate_allows(record, thresholds.edge, thresholds.probability, thresholds.confidence): + positions[record.symbol] = { + "entry_price": record.close, + "entry_index": record.index, + "timestamp": record.timestamp, + "highest": record.close, + "probability_up": record.probability_up, + "expected_percent": record.expected_percent, + } + for symbol, position in list(positions.items()): + tail = next((record for record in reversed(records) if record.symbol == symbol), None) + if tail is None: + continue + net_percent = _net_percent(position["entry_price"], tail.close, round_trip_cost) + trades.append(net_percent) + rows.append( + { + "symbol": symbol, + "entry_timestamp": position["timestamp"], + "exit_timestamp": tail.timestamp, + "net_percent": round(net_percent, 4), + "reason": "end_of_replay", + "held_bars": tail.index - int(position["entry_index"]), + "entry_probability": round(float(position["probability_up"]), 4), + "entry_expected_percent": round(float(position["expected_percent"]), 4), + } + ) + return {**_stats(trades), "trades_detail": rows[-50:]} + + +def _walk_forward( + records: list[ForecastRecord], + *, + edges: list[float], + probabilities: list[float], + confidences: list[float], + min_trades: int, + horizon: int, + folds: int, +) -> dict[str, Any]: + ordered = sorted(records, key=lambda item: item.timestamp) + if folds < 2 or len(ordered) < folds * 20: + return {"summary": {"status": "insufficient"}, "folds": []} + timestamps = sorted({record.timestamp for record in ordered}) + fold_size = max(1, len(timestamps) // folds) + rows = [] + all_test_trades: list[float] = [] + for fold in range(1, folds): + test_start = timestamps[fold * fold_size] + test_end = timestamps[(fold + 1) * fold_size - 1] if fold < folds - 1 else timestamps[-1] + train = [record for record in ordered if record.timestamp < test_start] + test = [record for record in ordered if test_start <= record.timestamp <= test_end] + train_results = _calibrate( + train, + edges=edges, + probabilities=probabilities, + confidences=confidences, + min_trades=max(4, min_trades // 2), + horizon=horizon, + ) + if not train_results: + continue + selected = _choose_recommendation(train_results, min_trades=max(4, min_trades // 2)) + test_trades = _selected_trades(test, selected.edge, selected.probability, selected.confidence, horizon) + all_test_trades.extend(test_trades) + rows.append( + { + "fold": fold, + "train_records": len(train), + "test_records": len(test), + "thresholds": _result_dict(selected), + "test": _stats(test_trades), + } + ) + summary = _stats(all_test_trades) + summary["status"] = "ok" if summary["trades"] >= min_trades and summary["avg_net_percent"] > 0 else "warn" + return {"summary": summary, "folds": rows} + + +def _probability_calibration(records: list[ForecastRecord]) -> dict[str, Any]: + buckets: dict[str, list[ForecastRecord]] = {} + for record in records: + low = max(0.0, min(0.95, int(record.probability_up * 20) / 20)) + key = f"{low:.2f}-{low + 0.05:.2f}" + buckets.setdefault(key, []).append(record) + rows = [] + for key in sorted(buckets): + items = buckets[key] + wins = sum(1 for item in items if item.future_net_percent > 0) + avg_probability = sum(item.probability_up for item in items) / len(items) + rows.append( + { + "bucket": key, + "samples": len(items), + "avg_probability": round(avg_probability, 4), + "actual_win_rate": round(wins / len(items), 4), + "avg_future_net_percent": round(sum(item.future_net_percent for item in items) / len(items), 4), + } + ) + return {"samples": len(records), "buckets": rows} + + +def _candidate_blocks(record: ForecastRecord, edge: float) -> bool: + return ( + record.expected_percent <= -edge + and record.probability_up <= 0.45 + ) or ( + record.q50_percent <= -edge + and record.probability_up <= 0.48 + ) + + +def _candidate_allows(record: ForecastRecord, edge: float, probability: float, confidence: float) -> bool: + dynamic_confidence = _forecast_confidence(record.expected_percent, record.probability_up, record.skill, edge) + return ( + not _candidate_blocks(record, edge) + and record.expected_percent >= edge + and record.probability_up >= probability + and dynamic_confidence >= confidence + and record.skill > 0.0 + ) + + +def _net_percent(entry_price: float, exit_price: float, round_trip_cost: float) -> float: + if entry_price <= 0 or exit_price <= 0: + return 0.0 + return (math.exp(math.log(exit_price / entry_price) - round_trip_cost) - 1.0) * 100.0 + + +def _stats(values: list[float]) -> dict[str, Any]: + wins = sum(1 for value in values if value > 0) + total = sum(values) + gross_profit = sum(value for value in values if value > 0) + gross_loss = abs(sum(value for value in values if value < 0)) + profit_factor = gross_profit / gross_loss if gross_loss > 0 else (999.0 if gross_profit > 0 else 0.0) + return { + "trades": len(values), + "wins": wins, + "win_rate": round(wins / len(values), 4) if values else 0.0, + "total_net_percent": round(total, 4), + "avg_net_percent": round(total / len(values), 4) if values else 0.0, + "max_drawdown_percent": round(_max_drawdown(values), 4), + "profit_factor": round(profit_factor, 4), + } + + def _calibrate( records: list[ForecastRecord], *, @@ -551,26 +781,7 @@ def _selected_trades( for record in sorted(records, key=lambda item: (item.timestamp, item.symbol)): if record.index < next_allowed_by_symbol.get(record.symbol, -1): continue - dynamic_confidence = _forecast_confidence( - record.expected_percent, - record.probability_up, - record.skill, - edge, - ) - block_entry = ( - record.expected_percent <= -edge - and record.probability_up <= 0.45 - ) or ( - record.q50_percent <= -edge - and record.probability_up <= 0.48 - ) - if ( - not block_entry - and record.expected_percent >= edge - and record.probability_up >= probability - and dynamic_confidence >= confidence - and record.skill > 0.0 - ): + if _candidate_allows(record, edge, probability, confidence): trades.append(record.future_net_percent) next_allowed_by_symbol[record.symbol] = record.index + max(1, horizon) return trades @@ -648,6 +859,14 @@ def _result_line(result: CalibrationResult) -> str: ) +def _stats_line(stats: dict[str, Any]) -> str: + return ( + f"trades={stats.get('trades', 0)} win={stats.get('win_rate', 0):.3f} " + f"avg={stats.get('avg_net_percent', 0):.4f}% total={stats.get('total_net_percent', 0):.4f}% " + f"dd={stats.get('max_drawdown_percent', 0):.4f}% pf={stats.get('profit_factor', 0):.3f}" + ) + + def _result_dict(result: CalibrationResult) -> dict[str, Any]: return { "edge": result.edge, diff --git a/tools/run_torch_retrain.ps1 b/tools/run_torch_retrain.ps1 index 5b295ac..b40d1c0 100644 --- a/tools/run_torch_retrain.ps1 +++ b/tools/run_torch_retrain.ps1 @@ -14,7 +14,8 @@ param( [int]$Epochs = 0, [int]$Patience = 0, [string]$Interval = "", - [string]$EnvFile = "" + [string]$EnvFile = "", + [switch]$SkipGuard ) $ErrorActionPreference = "Stop" @@ -70,6 +71,13 @@ if (-not $Interval -and $env:TORCH_RETRAIN_INTERVAL) { $Interval = $env:TORCH_RE if (-not $EnvFile -and $env:TORCH_RETRAIN_ENV) { $EnvFile = $env:TORCH_RETRAIN_ENV } if (-not $EnvFile -and (Test-Path (Join-Path $RepoRoot ".env"))) { $EnvFile = Join-Path $RepoRoot ".env" } +$ModelFile = if ($env:TIME_SERIES_LSTM_MODEL_PATH) { $env:TIME_SERIES_LSTM_MODEL_PATH } else { Join-Path $RuntimeDir "lstm_forecaster.json" } +if (-not [System.IO.Path]::IsPathRooted($ModelFile)) { $ModelFile = Join-Path $RepoRoot $ModelFile } +$CandidateFile = Join-Path $RuntimeDir "lstm_forecaster.candidate.json" +$CurrentCalibration = Join-Path $RuntimeDir "torch_guard_current.json" +$CandidateCalibration = Join-Path $RuntimeDir "torch_guard_candidate.json" +$GuardReport = Join-Path $RuntimeDir "torch_retrain_guard.json" + $mutex = New-Object System.Threading.Mutex($false, "TradeBotTorchRecurrentRetrainer") $hasLock = $false $pushedLocation = $false @@ -92,7 +100,8 @@ try { "--layers", $Layers, "--dropouts", $Dropouts, "--epochs", $Epochs.ToString(), - "--patience", $Patience.ToString() + "--patience", $Patience.ToString(), + "--output", $CandidateFile ) if ($Symbols) { $trainerArgs += @("--symbols", $Symbols) } if ($Interval) { $trainerArgs += @("--interval", $Interval) } @@ -109,7 +118,51 @@ try { if ($LASTEXITCODE -ne 0) { throw "Trainer failed with exit code $LASTEXITCODE." } - Write-RetrainLog "Finished PyTorch recurrent retrain." + Write-RetrainLog "Finished PyTorch recurrent retrain candidate: $CandidateFile" + + if ($SkipGuard -or -not (Test-Path $ModelFile)) { + Move-Item -Force -LiteralPath $CandidateFile -Destination $ModelFile + Write-RetrainLog "Accepted candidate without guard. Active artifact: $ModelFile" + exit 0 + } + + $calibrationBaseArgs = @( + "-u", + "tools\calibrate_torch_thresholds.py", + "--limit", "2000", + "--calibration-window", "720", + "--min-trades", "12" + ) + if ($Symbols) { $calibrationBaseArgs += @("--symbols", $Symbols) } + if ($EnvFile) { $calibrationBaseArgs += @("--env", $EnvFile) } + + Write-RetrainLog "Calibrating current artifact for guard." + & $python @($calibrationBaseArgs + @("--artifact", $ModelFile, "--output", $CurrentCalibration)) 2>&1 | Tee-Object -FilePath $LogFile -Append + if ($LASTEXITCODE -ne 0) { + throw "Current artifact calibration failed with exit code $LASTEXITCODE." + } + + Write-RetrainLog "Calibrating candidate artifact for guard." + & $python @($calibrationBaseArgs + @("--artifact", $CandidateFile, "--output", $CandidateCalibration)) 2>&1 | Tee-Object -FilePath $LogFile -Append + if ($LASTEXITCODE -ne 0) { + throw "Candidate artifact calibration failed with exit code $LASTEXITCODE." + } + + Write-RetrainLog "Running retrain guard." + & $python -u "tools\accept_torch_candidate.py" ` + --current-report $CurrentCalibration ` + --candidate-report $CandidateCalibration ` + --candidate-artifact $CandidateFile ` + --target-artifact $ModelFile ` + --report $GuardReport 2>&1 | Tee-Object -FilePath $LogFile -Append + if ($LASTEXITCODE -eq 2) { + Write-RetrainLog "Candidate rejected by guard; keeping active artifact: $ModelFile" + exit 0 + } + if ($LASTEXITCODE -ne 0) { + throw "Retrain guard failed with exit code $LASTEXITCODE." + } + Write-RetrainLog "Candidate accepted by guard. Active artifact: $ModelFile" } catch { Write-RetrainLog "ERROR: $($_.Exception.Message)"